diff --git a/crates/bench/src/pread_reader.rs b/crates/bench/src/pread_reader.rs index b6d43d9..6078d5a 100644 --- a/crates/bench/src/pread_reader.rs +++ b/crates/bench/src/pread_reader.rs @@ -10,7 +10,7 @@ use std::cell::RefCell; use std::fs::File; -use std::io::BufReader; +use std::io::{BufReader, Seek as _, SeekFrom}; use std::os::unix::fs::FileExt; use std::os::unix::io::AsRawFd; use std::path::Path; @@ -43,6 +43,10 @@ impl ReadCache { /// Read `len` bytes starting at `offset`. Returns a slice into the cache. /// On cache hit (range fully within cached block), no syscall needed. /// On miss, performs a `read_exact_at` syscall. + fn invalidate(&mut self) { + self.buf_len = 0; + } + fn get(&mut self, file: &File, offset: u64, len: usize) -> std::io::Result<&[u8]> { let end = offset + len as u64; if offset >= self.buf_offset && end <= self.buf_offset + self.buf_len as u64 { @@ -189,6 +193,19 @@ impl PreadReaderCore { pub fn total_lines(&self) -> usize { self.line_index.line_count() } + + pub fn refresh_index(&mut self) -> std::io::Result<()> { + let new_size = self.file.metadata()?.len(); + self.file.seek(SeekFrom::Start(0))?; + let new_index = { + let mut reader = BufReader::new(&self.file); + LineIndex::from_reader(&mut reader)? + }; + self.file_size = new_size; + self.line_index = new_index; + self.cache.get_mut().invalidate(); + Ok(()) + } } // ─── 3 Variants ─────────────────────────────────────────────────────────────── @@ -198,6 +215,12 @@ pub struct PreadReaderPlain { inner: PreadReaderCore, } +impl PreadReaderPlain { + pub fn refresh_index(&mut self) -> std::io::Result<()> { + self.inner.refresh_index() + } +} + impl FileReaderBackend for PreadReaderPlain { fn name(&self) -> &str { "pread_plain" @@ -441,6 +464,55 @@ mod tests { reader.close(); } + #[test] + fn test_refresh_index_sees_appended_lines() { + let dir = tempfile::tempdir().unwrap(); + let path = dir.path().join("refresh_test.log"); + + // Phase 1: write 3 initial lines + { + let mut f = std::fs::File::create(&path).unwrap(); + std::io::Write::write_all(&mut f, b"alpha\nbeta\ngamma\n").unwrap(); + } + + let mut reader = PreadReaderPlain::open(&path).unwrap(); + assert_eq!(reader.total_lines(), 3); + assert_eq!(reader.get_line(0), Some("alpha".to_owned())); + assert_eq!(reader.get_line(3), None, "should be out of bounds before append"); + + // Phase 2: append 2 more lines + { + use std::io::Write as _; + let mut f = std::fs::OpenOptions::new().append(true).open(&path).unwrap(); + f.write_all(b"delta\nepsilon\n").unwrap(); + } + + // Still stale — refresh needed + assert_eq!(reader.get_line(3), None, "stale index before refresh"); + + reader.refresh_index().unwrap(); + assert_eq!(reader.total_lines(), 5); + assert_eq!(reader.get_line(3), Some("delta".to_owned())); + assert_eq!(reader.get_line(4), Some("epsilon".to_owned())); + assert_eq!(reader.get_line(5), None); + + reader.close(); + } + + #[test] + fn test_refresh_index_no_change_is_noop() { + let f = create_temp_file(b"one\ntwo\n"); + let mut reader = PreadReaderPlain::open(f.path()).unwrap(); + assert_eq!(reader.total_lines(), 2); + + reader.refresh_index().unwrap(); + assert_eq!(reader.total_lines(), 2); + assert_eq!(reader.get_line(0), Some("one".to_owned())); + assert_eq!(reader.get_line(1), Some("two".to_owned())); + + reader.close(); + } + #[test] fn test_truncation_graceful_error() { // Verify that truncated file access returns None/Err instead of panicking diff --git a/crates/bench/src/suites/growth.rs b/crates/bench/src/suites/growth.rs index e206c62..49cb012 100644 --- a/crates/bench/src/suites/growth.rs +++ b/crates/bench/src/suites/growth.rs @@ -180,18 +180,40 @@ fn bench_scroll_during_append(config: &BenchConfig, dir: &std::path::Path) -> Ve } }); - let reader = PreadReaderPlain::open(&path).expect("Failed to open file"); + let mut reader = PreadReaderPlain::open(&path).expect("Failed to open file"); + let initial_lines = reader.total_lines(); let mut frame_latencies = Vec::new(); let mut current_line = 0usize; + let mut refresh_count: u64 = 0; + let mut none_count: u64 = 0; let scroll_start = std::time::Instant::now(); + let mut last_refresh = std::time::Instant::now(); + let refresh_interval = std::time::Duration::from_millis(250); while scroll_start.elapsed().as_secs() < duration_secs { - let t = std::time::Instant::now(); - let _ = reader.get_line(current_line); - frame_latencies.push(t.elapsed().as_micros() as u64); - current_line += 1; + if last_refresh.elapsed() >= refresh_interval { + reader.refresh_index().ok(); + refresh_count += 1; + last_refresh = std::time::Instant::now(); + continue; + } + + if let Some(_line) = reader.get_line(current_line) { + let t = std::time::Instant::now(); + frame_latencies.push(t.elapsed().as_micros() as u64); + current_line += 1; + } else { + none_count += 1; + std::thread::sleep(std::time::Duration::from_millis(1)); + } } + let max_line = current_line; + assert!( + max_line > initial_lines, + "benchmark never read past initial {initial_lines} lines (max={max_line})" + ); + reader.close(); bg_handle.join().ok(); @@ -202,6 +224,10 @@ fn bench_scroll_during_append(config: &BenchConfig, dir: &std::path::Path) -> Ve extra.insert("duration_secs".into(), duration_secs as f64); extra.insert("append_rate_per_sec".into(), append_rate as f64); extra.insert("frames_rendered".into(), frame_latencies.len() as f64); + extra.insert("refresh_count".into(), refresh_count as f64); + extra.insert("none_count".into(), none_count as f64); + extra.insert("initial_lines".into(), initial_lines as f64); + extra.insert("max_line_seen".into(), max_line as f64); vec![BenchmarkResult { category: "growth".into(),