gwseq_io/source/

local.rs

//! A local file as a [`ByteSource`].
//!
//! One `std::fs::File`, read positionally — `pread` on unix, `seek_read` on
//! Windows — so `&self` is enough and N threads share one handle, and there is
//! no pool of handles to manage or hand back.
//!
//! No `mmap`. It would be faster on a warm page cache and it would turn a
//! truncated or concurrently modified file into `SIGBUS`, which is not a
//! failure a `Result` can carry or a Python caller can catch. Every other
//! failure here comes back as an error; this one would not.

use std::fs::File;
use std::path::Path;

use bytes::Bytes;
use parking_lot::RwLock;

use crate::error::{Error, Result};
use crate::source::ByteSource;

#[derive(Debug)]
pub struct LocalSource {
    /// `None` once closed. `RwLock` rather than `Mutex`: reads take it shared,
    /// so the positioned reads stay concurrent, and only `close` takes it
    /// exclusively.
    file: RwLock<Option<File>>,
    path: String,
    /// Snapshotted at open. These are files being read, not tailed, and a
    /// length that cannot change is what lets `read_at` size its buffer to what
    /// is actually there rather than to what was asked for.
    len: u64,
}

impl LocalSource {
    pub fn open(path: impl AsRef<Path>) -> Result<Self> {
        let path = path.as_ref();
        let display = path.to_string_lossy().into_owned();
        let file = File::open(path).map_err(|e| Error::io(&display, e))?;
        let len = file.metadata().map_err(|e| Error::io(&display, e))?.len();
        Ok(Self {
            file: RwLock::new(Some(file)),
            path: display,
            len,
        })
    }
}

impl ByteSource for LocalSource {
    fn path(&self) -> &str {
        &self.path
    }

    fn len(&self) -> Result<u64> {
        Ok(self.len)
    }

    fn read_at(&self, offset: u64, len: usize) -> Result<Bytes> {
        if len == 0 || offset >= self.len {
            return Ok(Bytes::new());
        }
        // Clamp to what the file holds, so a caller asking for the rest of a
        // 2 GB file does not get a 2 GB allocation for the 4 KB that are left.
        let len = len.min((self.len - offset) as usize);

        let guard = self.file.read();
        let file = guard.as_ref().ok_or_else(|| Error::Closed {
            path: self.path.clone(),
        })?;

        let mut buf = vec![0u8; len];
        let read = pread(file, offset, &mut buf).map_err(|e| Error::io(&self.path, e))?;
        buf.truncate(read);
        Ok(Bytes::from(buf))
    }

    fn close(&self) {
        *self.file.write() = None;
    }
}

/// Positioned read, looping until the buffer is full or the file ends.
///
/// One `read_at`/`seek_read` may return short for reasons that are not EOF — a
/// signal, a large request — so the loop is not optional. `Ok(0)` is the only
/// end-of-file signal.
#[cfg(unix)]
fn pread(file: &File, offset: u64, buf: &mut [u8]) -> std::io::Result<usize> {
    use std::os::unix::fs::FileExt;
    let mut filled = 0;
    while filled < buf.len() {
        match file.read_at(&mut buf[filled..], offset + filled as u64) {
            Ok(0) => break,
            Ok(n) => filled += n,
            Err(e) if e.kind() == std::io::ErrorKind::Interrupted => {}
            Err(e) => return Err(e),
        }
    }
    Ok(filled)
}

#[cfg(windows)]
fn pread(file: &File, offset: u64, buf: &mut [u8]) -> std::io::Result<usize> {
    use std::os::windows::fs::FileExt;
    let mut filled = 0;
    while filled < buf.len() {
        match file.seek_read(&mut buf[filled..], offset + filled as u64) {
            Ok(0) => break,
            Ok(n) => filled += n,
            Err(e) if e.kind() == std::io::ErrorKind::Interrupted => {}
            Err(e) => return Err(e),
        }
    }
    Ok(filled)
}

#[cfg(test)]
mod tests {
    use super::*;
    use std::io::Write;

    fn fixture(bytes: &[u8]) -> (tempdir::TempPath, LocalSource) {
        let path = tempdir::TempPath::new("local-source");
        let mut f = File::create(&path.0).unwrap();
        f.write_all(bytes).unwrap();
        f.sync_all().unwrap();
        let source = LocalSource::open(&path.0).unwrap();
        (path, source)
    }

    /// A temp file that removes itself. Not worth a dependency for one test
    /// module.
    pub mod tempdir {
        pub struct TempPath(pub std::path::PathBuf);

        impl TempPath {
            pub fn new(tag: &str) -> Self {
                let nanos = std::time::SystemTime::now()
                    .duration_since(std::time::UNIX_EPOCH)
                    .unwrap()
                    .as_nanos();
                let mut p = std::env::temp_dir();
                p.push(format!(
                    "gwseq-{tag}-{nanos}-{:?}",
                    std::thread::current().id()
                ));
                TempPath(p)
            }
        }

        impl Drop for TempPath {
            fn drop(&mut self) {
                let _ = std::fs::remove_file(&self.0);
            }
        }
    }

    #[test]
    fn reads_every_range_of_a_known_file() {
        let data: Vec<u8> = (0..=255u8).collect();
        let (_p, source) = fixture(&data);
        assert_eq!(source.len().unwrap(), 256);
        for offset in [0u64, 1, 127, 255] {
            for len in [1usize, 2, 64, 300] {
                let got = source.read_at(offset, len).unwrap();
                let end = (offset as usize + len).min(data.len());
                assert_eq!(&got[..], &data[offset as usize..end], "at {offset}+{len}");
            }
        }
    }

    #[test]
    fn a_read_past_the_end_is_short_not_an_error() {
        let (_p, source) = fixture(b"abcdef");
        assert_eq!(&source.read_at(4, 100).unwrap()[..], b"ef");
        assert!(source.read_at(6, 10).unwrap().is_empty());
        assert!(source.read_at(99, 10).unwrap().is_empty());
        assert!(source.read_at(0, 0).unwrap().is_empty());
    }

    #[test]
    fn read_exact_at_refuses_a_short_read() {
        let (_p, source) = fixture(b"abcdef");
        assert_eq!(&source.read_exact_at(0, 6).unwrap()[..], b"abcdef");
        let err = source.read_exact_at(4, 100).unwrap_err().to_string();
        assert!(err.contains("wanted 100 bytes, got 2"), "{err}");
    }

    #[test]
    fn read_to_end_takes_the_rest() {
        let (_p, source) = fixture(b"abcdef");
        assert_eq!(&source.read_to_end(2).unwrap()[..], b"cdef");
        assert!(source.read_to_end(6).unwrap().is_empty());
    }

    #[test]
    fn a_closed_source_refuses_and_names_the_file() {
        let (path, source) = fixture(b"abcdef");
        source.close();
        source.close(); // idempotent
        let err = source.read_at(0, 1).unwrap_err().to_string();
        assert!(err.contains("is closed"), "{err}");
        // The point of naming it: a reader closed by mistake in a program
        // holding several says which one.
        assert!(err.contains(path.0.to_str().unwrap()), "{err}");
    }

    #[test]
    fn concurrent_readers_share_one_handle() {
        let data: Vec<u8> = (0..64u8).cycle().take(1 << 16).collect();
        let (_p, source) = fixture(&data);
        let source = std::sync::Arc::new(source);
        let threads: Vec<_> = (0..8)
            .map(|t| {
                let source = source.clone();
                let data = data.clone();
                std::thread::spawn(move || {
                    for i in 0..500u64 {
                        let offset = (t * 977 + i * 13) % 60_000;
                        let got = source.read_at(offset, 128).unwrap();
                        let end = (offset as usize + 128).min(data.len());
                        assert_eq!(&got[..], &data[offset as usize..end]);
                    }
                })
            })
            .collect();
        for t in threads {
            t.join().unwrap();
        }
    }

    #[test]
    fn opening_a_missing_file_names_it() {
        let err = LocalSource::open("/definitely/not/here.bigwig").unwrap_err();
        assert!(matches!(err, Error::Io { .. }));
        assert!(err.to_string().contains("here.bigwig"));
    }
}