#67: media files share a bounded pool of open descriptors

Problem: every FileReader owned its std::fs::File for its whole
lifetime, so each open media source pinned one descriptor. A project
with hundreds of sources hit the soft limit (256 on macOS, 1024 on
Linux) and the remaining media failed to open or showed as offline.

Fix: FileReader keeps its path and borrows the file from a shared LRU
pool of at most MAX_OPEN_FILES (128) open files. Past the cap the least
recently read file is closed and its reader reopens it on the next
read; a file that is now shorter than when it was opened is refused
instead of being read as the old one. Dropping a reader closes its file.

Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com>
This commit is contained in:
vangel76
2026-10-10 19:01:11 +02:00
co-authored by Claude Opus 5.5
parent 98ba9acbce
commit d8f22e9908
+145 -6
View File
@@ -43,24 +43,131 @@ impl ByteReader for MemReader {
}
}
/// A file on disk, read in place: the file stays open and only the requested ranges are read, so
/// opening a clip costs its index rather than its size.
/// Files [`FileReader`]s keep open at most, together (#67). Each open media source used to hold
/// its own descriptor, so a project with hundreds of sources hit the soft descriptor limit (256 on
/// macOS, 1024 on Linux) and the rest showed as failed or offline. Past this many, the least
/// recently read file is closed and its reader reopens it on demand.
#[cfg(any(unix, windows))]
pub const MAX_OPEN_FILES: usize = 128;
/// The open files of a set of [`FileReader`]s, least recently used first.
#[cfg(any(unix, windows))]
#[derive(Debug)]
pub(crate) struct FilePool {
cap: usize,
state: std::sync::Mutex<PoolState>,
}
#[cfg(any(unix, windows))]
#[derive(Debug, Default)]
struct PoolState {
next_id: u64,
open: std::collections::VecDeque<(u64, Arc<std::fs::File>)>,
}
#[cfg(any(unix, windows))]
impl FilePool {
pub(crate) fn new(cap: usize) -> Self {
Self { cap: cap.max(1), state: Default::default() }
}
fn state(&self) -> std::sync::MutexGuard<'_, PoolState> {
self.state.lock().unwrap_or_else(std::sync::PoisonError::into_inner)
}
fn next_id(&self) -> u64 {
let mut s = self.state();
s.next_id = s.next_id.wrapping_add(1);
s.next_id
}
/// The open file of reader `id`, now the most recently used.
fn get(&self, id: u64) -> Option<Arc<std::fs::File>> {
let mut s = self.state();
let at = s.open.iter().position(|(i, _)| *i == id)?;
let entry = s.open.remove(at)?;
let file = entry.1.clone();
s.open.push_back(entry);
Some(file)
}
/// Keep `file` open for reader `id`, closing the least recently used files past the cap. A
/// file another thread is reading from stays open until that read ends.
fn insert(&self, id: u64, file: std::fs::File) -> Arc<std::fs::File> {
let mut s = self.state();
if let Some((_, open)) = s.open.iter().find(|(i, _)| *i == id) {
return open.clone(); // another thread reopened it first
}
let file = Arc::new(file);
s.open.push_back((id, file.clone()));
while s.open.len() > self.cap {
s.open.pop_front();
}
file
}
fn remove(&self, id: u64) {
self.state().open.retain(|(i, _)| *i != id);
}
#[cfg(test)]
fn open_count(&self) -> usize {
self.state().open.len()
}
}
#[cfg(any(unix, windows))]
static FILE_POOL: std::sync::LazyLock<Arc<FilePool>> = std::sync::LazyLock::new(|| Arc::new(FilePool::new(MAX_OPEN_FILES)));
/// A file on disk, read in place: only the requested ranges are read, so opening a clip costs its
/// index rather than its size. The file is kept open in a shared pool of at most
/// [`MAX_OPEN_FILES`] and reopened by path when the pool closed it.
#[cfg(any(unix, windows))]
#[derive(Debug)]
pub struct FileReader {
file: std::fs::File,
path: std::path::PathBuf,
len: u64,
id: u64,
pool: Arc<FilePool>,
}
#[cfg(any(unix, windows))]
impl FileReader {
pub fn open(path: &std::path::Path) -> io::Result<Self> {
Self::open_in(path, &FILE_POOL)
}
pub(crate) fn open_in(path: &std::path::Path, pool: &Arc<FilePool>) -> io::Result<Self> {
let file = std::fs::File::open(path)?;
let m = file.metadata()?;
if !m.is_file() {
return Err(io::Error::new(io::ErrorKind::NotFound, format!("{} is not a file", path.display())));
}
Ok(Self { file, len: m.len() })
let id = pool.next_id();
pool.insert(id, file);
Ok(Self { path: path.to_path_buf(), len: m.len(), id, pool: pool.clone() })
}
/// The open file, reopened when the pool closed it. A file now shorter than when it was opened
/// (replaced or truncated) is refused rather than read as the old one; a longer one (still
/// being written) is read up to the opened length, as before.
fn file(&self) -> io::Result<Arc<std::fs::File>> {
if let Some(file) = self.pool.get(self.id) {
return Ok(file);
}
let file = std::fs::File::open(&self.path)?;
let m = file.metadata()?;
if !m.is_file() || m.len() < self.len {
return Err(io::Error::other(format!("{} changed since it was opened", self.path.display())));
}
Ok(self.pool.insert(self.id, file))
}
}
#[cfg(any(unix, windows))]
impl Drop for FileReader {
fn drop(&mut self) {
self.pool.remove(self.id);
}
}
@@ -71,15 +178,16 @@ impl ByteReader for FileReader {
}
/// Positional reads: threads decoding different parts of the file do not share a cursor.
fn read_at(&self, offset: u64, buf: &mut [u8]) -> io::Result<()> {
let file = self.file()?;
#[cfg(unix)]
{
std::os::unix::fs::FileExt::read_exact_at(&self.file, buf, offset)
std::os::unix::fs::FileExt::read_exact_at(&*file, buf, offset)
}
#[cfg(windows)]
{
let (mut done, mut at) = (0usize, offset);
while let Some(rest) = buf.get_mut(done..).filter(|r| !r.is_empty()) {
match std::os::windows::fs::FileExt::seek_read(&self.file, rest, at) {
match std::os::windows::fs::FileExt::seek_read(&*file, rest, at) {
Ok(0) => return Err(eof()),
Ok(n) => {
done = done.saturating_add(n);
@@ -169,6 +277,37 @@ mod tests {
std::fs::remove_dir_all(&dir).unwrap();
}
/// #67: more readers than the pool keeps open stay readable; past the cap the least recently
/// read file is closed and reopened on demand, and a file truncated meanwhile is refused.
#[cfg(any(unix, windows))]
#[test]
fn more_file_readers_than_the_pool_holds_stay_readable() {
let dir = std::env::temp_dir().join(format!("filmcraft-file-pool-{}", std::process::id()));
std::fs::create_dir_all(&dir).unwrap();
let pool = Arc::new(FilePool::new(2));
let paths: Vec<_> = (0..5).map(|i| dir.join(format!("{i}.bin"))).collect();
for (i, p) in paths.iter().enumerate() {
std::fs::write(p, format!("file{i}-0123456789")).unwrap();
}
let readers: Vec<_> = paths.iter().map(|p| FileReader::open_in(p, &pool).unwrap()).collect();
assert_eq!(pool.open_count(), 2);
for _ in 0..2 {
for (i, r) in readers.iter().enumerate() {
let mut b = [0u8; 5];
r.read_at(0, &mut b).unwrap();
assert_eq!(b, *format!("file{i}").as_bytes());
assert!(pool.open_count() <= 2);
}
}
// reader 0 was closed by the later reads; its file is now shorter than when opened
std::fs::write(&paths[0], b"short").unwrap();
let mut b = [0u8; 5];
assert!(readers[0].read_at(0, &mut b).is_err());
drop(readers);
assert_eq!(pool.open_count(), 0, "dropped readers close their files");
std::fs::remove_dir_all(&dir).unwrap();
}
/// A reader that counts the bytes read from it.
struct Counting(MemReader, std::sync::atomic::AtomicU64);
impl ByteReader for Counting {