diff --git a/benches/benchmark.rs b/benches/benchmark.rs index 55ad7e8..c1387db 100644 --- a/benches/benchmark.rs +++ b/benches/benchmark.rs @@ -4,6 +4,8 @@ use std::fs::OpenOptions; use std::hint::black_box; use std::io::Cursor; use std::io::Read; +use std::io::Seek; +use std::io::SeekFrom; use std::io::Write; use tempfile::NamedTempFile; @@ -119,5 +121,45 @@ fn criterion_benchmark(c: &mut Criterion) { read_group.finish(); } -criterion_group!(benches, criterion_benchmark); +/// Reads one large stream the way a caller that streams it out would: in +/// pieces, so the stream's buffer is refilled many times. +fn large_stream_benchmark(c: &mut Criterion) { + let size = 256 * 1024 * 1024usize; + let buff = write_many_streams(1, size); + + let mut group = c.benchmark_group("read_large_stream_memory"); + group.sample_size(10); + group.throughput(Throughput::Bytes(size as u64)); + group.bench_function("copy 256MiB", |b| { + b.iter(|| { + let mut comp = CompoundFile::open(Cursor::new(&buff)).unwrap(); + let mut stream = comp.open_stream("test0").unwrap(); + let n = std::io::copy(&mut stream, &mut std::io::sink()).unwrap(); + black_box(n); + }) + }); + group.finish(); + + let reads = 1000u64; + let mut group = c.benchmark_group("seek_large_stream_memory"); + group.sample_size(10); + group.throughput(Throughput::Elements(reads)); + group.bench_function("1000 x seek+read 16B in 256MiB", |b| { + b.iter(|| { + let mut comp = CompoundFile::open(Cursor::new(&buff)).unwrap(); + let mut stream = comp.open_stream("test0").unwrap(); + let len = stream.len(); + let mut chunk = [0u8; 16]; + for i in 0..reads { + let offset = (i * 7919 * 4099) % (len - 16); + stream.seek(SeekFrom::Start(offset)).unwrap(); + stream.read_exact(&mut chunk).unwrap(); + black_box(chunk); + } + }) + }); + group.finish(); +} + +criterion_group!(benches, criterion_benchmark, large_stream_benchmark); criterion_main!(benches); diff --git a/src/internal/alloc.rs b/src/internal/alloc.rs index eaea12d..34b8d75 100644 --- a/src/internal/alloc.rs +++ b/src/internal/alloc.rs @@ -25,6 +25,13 @@ pub struct Allocator { difat: Vec, fat: Vec, free_sectors: Vec, + /// The sector IDs of the most recently used chain, keyed by its start + /// sector. A `Stream` opens its chain anew for every buffer refill, + /// and walking a long FAT chain from the start each time made reading + /// or writing a stream of `n` sectors cost `O(n^2)`. The list is + /// handed to the next `Chain` opened for the same start sector, which + /// gives it back when dropped; any FAT change discards it. + chain_cache: Option<(u32, Vec)>, } impl Allocator { @@ -41,6 +48,7 @@ impl Allocator { difat, fat, free_sectors: Vec::new(), + chain_cache: None, }; alloc.validate(validation)?; Ok(alloc) @@ -86,7 +94,26 @@ impl Allocator { start_sector_id: u32, init: SectorInit, ) -> io::Result> { - Chain::new(self, start_sector_id, init) + match self.chain_cache.take() { + Some((cached_start, sector_ids)) + if cached_start == start_sector_id => + { + Ok(Chain::from_sector_ids(self, sector_ids, init)) + } + _ => Chain::new(self, start_sector_id, init), + } + } + + /// Remembers the sector IDs of the chain starting at `start_sector_id` + /// so that the next `open_chain` for it need not walk the FAT again. + /// The IDs must reflect the current FAT. + pub(crate) fn cache_chain( + &mut self, + start_sector_id: u32, + sector_ids: Vec, + ) { + debug_assert_eq!(sector_ids.first(), Some(&start_sector_id)); + self.chain_cache = Some((start_sector_id, sector_ids)); } fn validate(&mut self, validation: Validation) -> io::Result<()> { @@ -380,6 +407,7 @@ impl Allocator { /// Sets `self.fat[index] = value`, and also writes that change to the /// underlying file. The `index` must be <= `self.fat.len()`. fn set_fat(&mut self, index: u32, value: u32) -> io::Result<()> { + self.chain_cache = None; let index = index as usize; debug_assert!(index <= self.fat.len()); let fat_entries_per_sector = @@ -409,7 +437,9 @@ impl Allocator { #[cfg(test)] mod tests { use super::Allocator; - use crate::internal::{consts, Sectors, Validation, Version}; + use crate::internal::{ + consts, Chain, SectorInit, Sectors, Validation, Version, + }; use std::io::Cursor; fn make_sectors( @@ -545,6 +575,57 @@ mod tests { make_allocator(difat, fat, Validation::Permissive); } + #[test] + fn chain_cache_follows_the_fat() { + let difat = vec![0]; + let fat = vec![consts::FAT_SECTOR, 2, 3, consts::END_OF_CHAIN]; + let mut alloc = make_allocator(difat, fat, Validation::Strict); + assert_eq!(alloc.chain_cache, None); + + // Dropping a chain leaves its sector list behind for the next open. + let chain = alloc.open_chain(1, SectorInit::Zero).unwrap(); + assert_eq!(chain.sector_ids(), &[1, 2, 3]); + drop(chain); + assert_eq!(alloc.chain_cache, Some((1, vec![1, 2, 3]))); + + // A chain that grew hands back the grown list. + let mut chain = alloc.open_chain(1, SectorInit::Zero).unwrap(); + chain.set_len(4 * 512).unwrap(); + assert_eq!(chain.sector_ids(), &[1, 2, 3, 4]); + drop(chain); + assert_eq!(alloc.chain_cache, Some((1, vec![1, 2, 3, 4]))); + let chain = Chain::new(&mut alloc, 1, SectorInit::Zero).unwrap(); + assert_eq!(chain.sector_ids(), &[1, 2, 3, 4]); + drop(chain); + + // A chain that shrank hands back the shortened list. + let mut chain = alloc.open_chain(1, SectorInit::Zero).unwrap(); + chain.set_len(2 * 512).unwrap(); + assert_eq!(chain.sector_ids(), &[1, 2]); + drop(chain); + assert_eq!(alloc.chain_cache, Some((1, vec![1, 2]))); + let chain = Chain::new(&mut alloc, 1, SectorInit::Zero).unwrap(); + assert_eq!(chain.sector_ids(), &[1, 2]); + drop(chain); + + // Opening a different chain does not use the cached one. + let start = alloc.begin_chain(SectorInit::Zero).unwrap(); + let chain = alloc.open_chain(start, SectorInit::Zero).unwrap(); + assert_eq!(chain.sector_ids(), &[start]); + drop(chain); + assert_eq!(alloc.chain_cache, Some((start, vec![start]))); + + // Any FAT change from outside a chain discards the cache. + alloc.free_chain(1).unwrap(); + assert_eq!(alloc.chain_cache, None); + + // A freed chain leaves nothing behind. + let chain = alloc.open_chain(start, SectorInit::Zero).unwrap(); + chain.free().unwrap(); + assert_eq!(alloc.chain_cache, None); + assert_eq!(alloc.fat[start as usize], consts::FREE_SECTOR); + } + #[test] #[should_panic(expected = "Malformed FAT (sector 3 pointed to twice)")] fn double_pointee() { diff --git a/src/internal/chain.rs b/src/internal/chain.rs index 7d639c1..d83ca83 100644 --- a/src/internal/chain.rs +++ b/src/internal/chain.rs @@ -1,6 +1,7 @@ use crate::internal::{consts, Allocator, SectorInit}; use std::cmp; use std::io::{self, Read, Seek, SeekFrom, Write}; +use std::mem; //===========================================================================// @@ -33,6 +34,15 @@ impl<'a, F> Chain<'a, F> { Ok(Chain { allocator, init, sector_ids, offset_from_start: 0 }) } + /// Creates a chain from sector IDs that are known to match the FAT. + pub(crate) fn from_sector_ids( + allocator: &'a mut Allocator, + sector_ids: Vec, + init: SectorInit, + ) -> Chain<'a, F> { + Chain { allocator, init, sector_ids, offset_from_start: 0 } + } + pub fn start_sector_id(&self) -> u32 { self.sector_ids.first().copied().unwrap_or(consts::END_OF_CHAIN) } @@ -98,11 +108,13 @@ impl<'a, F: Write + Seek> Chain<'a, F> { if new_num_sectors == 0 { if let Some(&start_sector) = self.sector_ids.first() { self.allocator.free_chain(start_sector)?; + self.sector_ids.clear(); } } else if new_num_sectors <= self.sector_ids.len() { if new_num_sectors < self.sector_ids.len() { self.allocator .free_chain_after(self.sector_ids[new_num_sectors - 1])?; + self.sector_ids.truncate(new_num_sectors); } self.zero_tail(new_len)?; } else { @@ -128,8 +140,23 @@ impl<'a, F: Write + Seek> Chain<'a, F> { .write_all(&zeros) } - pub fn free(self) -> io::Result<()> { - self.allocator.free_chain(self.start_sector_id()) + pub fn free(mut self) -> io::Result<()> { + let start_sector_id = self.start_sector_id(); + self.sector_ids.clear(); + self.allocator.free_chain(start_sector_id) + } +} + +impl<'a, F> Drop for Chain<'a, F> { + fn drop(&mut self) { + // The list is kept in step with every change this chain made to the + // FAT, so the next chain opened for the same start sector can reuse + // it instead of walking the FAT again. + if !self.sector_ids.is_empty() { + let start_sector_id = self.sector_ids[0]; + self.allocator + .cache_chain(start_sector_id, mem::take(&mut self.sector_ids)); + } } } diff --git a/tests/chain_cache.rs b/tests/chain_cache.rs new file mode 100644 index 0000000..8d7d6a7 --- /dev/null +++ b/tests/chain_cache.rs @@ -0,0 +1,191 @@ +use cfb::{CompoundFile, OpenOptions}; +use std::io::{Cursor, Read, Seek, SeekFrom, Write}; + +//===========================================================================// + +// Every test uses a small stream buffer so that a stream is refilled many +// times, which is what makes the allocator hand a chain's cached sector list +// from one refill to the next. + +const BUFFER_SIZE: usize = 1024; +const SECTOR: usize = 4096; + +fn create_data(len: usize, seed: u8) -> Vec { + (0..len).map(|i| (i as u8).wrapping_mul(31).wrapping_add(seed)).collect() +} + +fn create_comp() -> CompoundFile>> { + OpenOptions::new() + .max_buffer_size(BUFFER_SIZE) + .create_with(Cursor::new(Vec::new())) + .unwrap() +} + +fn reopen( + comp: CompoundFile>>, +) -> CompoundFile>> { + OpenOptions::new() + .max_buffer_size(BUFFER_SIZE) + .strict() + .open_with(comp.into_inner()) + .unwrap() +} + +fn read_all(comp: &mut CompoundFile>>, path: &str) -> Vec { + let mut data = Vec::new(); + comp.open_stream(path).unwrap().read_to_end(&mut data).unwrap(); + data +} + +//===========================================================================// + +#[test] +fn sequential_reads_reuse_the_chain() { + let data = create_data(40 * SECTOR + 123, 1); + let mut comp = create_comp(); + comp.create_stream("/big").unwrap().write_all(&data).unwrap(); + let mut comp = reopen(comp); + let mut stream = comp.open_stream("/big").unwrap(); + let mut actual = vec![0u8; 100]; + let mut offset = 0; + while offset < data.len() { + let n = stream.read(&mut actual).unwrap(); + assert!(n > 0, "short read at {}", offset); + assert_eq!(actual[..n], data[offset..offset + n], "at {}", offset); + offset += n; + } + assert_eq!(stream.read(&mut actual).unwrap(), 0); +} + +#[test] +fn scattered_reads_reuse_the_chain() { + let data = create_data(64 * SECTOR, 2); + let mut comp = create_comp(); + comp.create_stream("/big").unwrap().write_all(&data).unwrap(); + let mut comp = reopen(comp); + let mut stream = comp.open_stream("/big").unwrap(); + let mut actual = [0u8; 16]; + for i in 0..200u64 { + let offset = (i * 7919 * 13) % (data.len() as u64 - 16); + stream.seek(SeekFrom::Start(offset)).unwrap(); + stream.read_exact(&mut actual).unwrap(); + let offset = offset as usize; + assert_eq!(actual, data[offset..offset + 16], "at {}", offset); + } +} + +#[test] +fn interleaved_writes_to_two_streams() { + let data_a = create_data(30 * SECTOR + 7, 3); + let data_b = create_data(25 * SECTOR + 9, 4); + let mut comp = create_comp(); + comp.create_stream("/a").unwrap().write_all(&data_a[..SECTOR]).unwrap(); + comp.create_stream("/b").unwrap().write_all(&data_b[..SECTOR]).unwrap(); + // Each append opens the chain anew after the other stream changed the + // FAT, so a stale cached chain would corrupt one of the streams. + let mut off_a = SECTOR; + let mut off_b = SECTOR; + while off_a < data_a.len() || off_b < data_b.len() { + if off_a < data_a.len() { + let end = (off_a + 3000).min(data_a.len()); + let mut stream = comp.open_stream("/a").unwrap(); + stream.seek(SeekFrom::End(0)).unwrap(); + stream.write_all(&data_a[off_a..end]).unwrap(); + off_a = end; + } + if off_b < data_b.len() { + let end = (off_b + 5000).min(data_b.len()); + let mut stream = comp.open_stream("/b").unwrap(); + stream.seek(SeekFrom::End(0)).unwrap(); + stream.write_all(&data_b[off_b..end]).unwrap(); + off_b = end; + } + } + assert_eq!(read_all(&mut comp, "/a"), data_a); + assert_eq!(read_all(&mut comp, "/b"), data_b); + let mut comp = reopen(comp); + assert_eq!(read_all(&mut comp, "/a"), data_a); + assert_eq!(read_all(&mut comp, "/b"), data_b); +} + +#[test] +fn shrink_then_append() { + let data = create_data(20 * SECTOR, 5); + let mut comp = create_comp(); + comp.create_stream("/big").unwrap().write_all(&data).unwrap(); + { + let mut stream = comp.open_stream("/big").unwrap(); + stream.set_len((8 * SECTOR + 100) as u64).unwrap(); + } + // The freed sectors must not linger in the cached chain, or the + // appended data would land in sectors that are no longer part of it. + let tail = create_data(5 * SECTOR, 7); + { + let mut stream = comp.open_stream("/big").unwrap(); + stream.seek(SeekFrom::End(0)).unwrap(); + stream.write_all(&tail).unwrap(); + } + let other = create_data(12 * SECTOR, 6); + comp.create_stream("/other").unwrap().write_all(&other).unwrap(); + let mut expected = data[..8 * SECTOR + 100].to_vec(); + expected.extend_from_slice(&tail); + let mut comp = reopen(comp); + assert_eq!(read_all(&mut comp, "/big"), expected); + assert_eq!(read_all(&mut comp, "/other"), other); +} + +#[test] +fn shrink_into_mini_stream_then_grow_again() { + let data = create_data(20 * SECTOR, 8); + let mut comp = create_comp(); + comp.create_stream("/big").unwrap().write_all(&data).unwrap(); + { + let mut stream = comp.open_stream("/big").unwrap(); + stream.set_len(1000).unwrap(); + } + let tail = create_data(10 * SECTOR, 10); + { + let mut stream = comp.open_stream("/big").unwrap(); + stream.seek(SeekFrom::End(0)).unwrap(); + stream.write_all(&tail).unwrap(); + } + let other = create_data(12 * SECTOR, 9); + comp.create_stream("/other").unwrap().write_all(&other).unwrap(); + let mut expected = data[..1000].to_vec(); + expected.extend_from_slice(&tail); + let mut comp = reopen(comp); + assert_eq!(read_all(&mut comp, "/big"), expected); + assert_eq!(read_all(&mut comp, "/other"), other); +} + +#[test] +fn truncate_to_zero_then_rewrite() { + let data = create_data(20 * SECTOR, 11); + let mut comp = create_comp(); + comp.create_stream("/big").unwrap().write_all(&data).unwrap(); + comp.open_stream("/big").unwrap().set_len(0).unwrap(); + let other = create_data(12 * SECTOR, 12); + comp.create_stream("/other").unwrap().write_all(&other).unwrap(); + let again = create_data(15 * SECTOR + 1, 13); + comp.open_stream("/big").unwrap().write_all(&again).unwrap(); + let mut comp = reopen(comp); + assert_eq!(read_all(&mut comp, "/big"), again); + assert_eq!(read_all(&mut comp, "/other"), other); +} + +#[test] +fn remove_then_recreate_stream() { + let data = create_data(20 * SECTOR, 14); + let mut comp = create_comp(); + comp.create_stream("/big").unwrap().write_all(&data).unwrap(); + comp.remove_stream("/big").unwrap(); + let other = create_data(12 * SECTOR, 15); + comp.create_stream("/other").unwrap().write_all(&other).unwrap(); + let again = create_data(15 * SECTOR + 1, 16); + comp.create_stream("/big").unwrap().write_all(&again).unwrap(); + let mut comp = reopen(comp); + assert_eq!(read_all(&mut comp, "/big"), again); + assert_eq!(read_all(&mut comp, "/other"), other); +} + +//===========================================================================//