streampager/
buffer_cache.rs

1//! Buffer Cache.
2
3use std::borrow::Cow;
4use std::fs::File as StdFile;
5use std::io::{Read, Seek, SeekFrom};
6use std::path::{Path, PathBuf};
7
8use lru::LruCache;
9
10use crate::buffer::Buffer;
11use crate::error::Error;
12
13pub(crate) struct BufferCache {
14    path: PathBuf,
15    file: Option<StdFile>,
16    cache: LruCache<usize, Buffer>,
17    block_size: usize,
18}
19
20impl BufferCache {
21    pub(crate) fn new<P: AsRef<Path>>(path: P, block_size: usize, capacity: usize) -> Self {
22        let path = path.as_ref();
23        BufferCache {
24            path: path.to_path_buf(),
25            file: None,
26            cache: LruCache::new(capacity),
27            block_size,
28        }
29    }
30
31    pub(crate) fn clear(&mut self) {
32        self.cache.clear();
33        self.file = None;
34    }
35
36    fn open_file(&mut self) -> Result<(), Error> {
37        if self.file.is_none() {
38            self.file = Some(StdFile::open(&self.path)?);
39        }
40        Ok(())
41    }
42
43    fn get_buffer(&mut self, start: usize, end: usize) -> Result<Option<&mut Buffer>, Error> {
44        let block_index = start / self.block_size;
45        let block_offset = start % self.block_size;
46        self.open_file()?;
47        if let Some(buffer) = self.cache.get_mut(&block_index) {
48            fill_buffer(
49                self.file.as_mut().expect("file is open"),
50                buffer,
51                start,
52                block_offset,
53                end - start,
54            )?;
55        } else {
56            let mut buffer = Buffer::new(self.block_size);
57            fill_buffer(
58                self.file.as_mut().expect("file is open"),
59                &mut buffer,
60                block_index * self.block_size,
61                0,
62                self.block_size,
63            )?;
64            self.cache.put(block_index, buffer);
65        }
66        Ok(self.cache.get_mut(&block_index))
67    }
68
69    fn get_data(&mut self, start: usize, end: usize) -> Result<&[u8], Error> {
70        let block_size = self.block_size;
71        if let Some(buffer) = self.get_buffer(start, end)? {
72            let data = buffer.read();
73            let data_start = data.len().min(start % block_size);
74            let data_end = (data.len()).min((end - 1) % block_size + 1);
75            Ok(&data[data_start..data_end])
76        } else {
77            Ok(&[])
78        }
79    }
80
81    pub(crate) fn with_slice<T, F>(
82        &mut self,
83        start: usize,
84        end: usize,
85        mut call: F,
86    ) -> Result<T, Error>
87    where
88        F: FnMut(Cow<'_, [u8]>) -> T,
89    {
90        let start_block = start / self.block_size;
91        let end_block = (end - 1) / self.block_size;
92        if start_block == end_block {
93            Ok(call(Cow::Borrowed(self.get_data(start, end)?)))
94        } else {
95            // The data spans multiple buffers, so we must make a copy to make it contiguous.
96            // Ensure we fill in any gaps that might occur.
97            let mut v = Vec::with_capacity(end - start);
98            let first_end = (start_block + 1) * self.block_size;
99            let first_slice = self.get_data(start, first_end)?;
100            v.extend_from_slice(first_slice);
101            v.resize(first_end - start, 0);
102            for b in start_block + 1..end_block {
103                let block_start = b * self.block_size;
104                let block_end = block_start + self.block_size;
105                let block_slice = self.get_data(block_start, block_end)?;
106                v.extend_from_slice(block_slice);
107                v.resize(block_end - start, 0);
108            }
109            let end_start = end_block * self.block_size;
110            let end_slice = self.get_data(end_start, end)?;
111            v.extend_from_slice(end_slice);
112            v.resize(end - start, 0);
113            Ok(call(Cow::Owned(v)))
114        }
115    }
116}
117
118fn fill_buffer(
119    file: &mut StdFile,
120    buffer: &mut Buffer,
121    file_offset: usize,
122    buffer_offset: usize,
123    len: usize,
124) -> Result<(), Error> {
125    if buffer_offset + len <= buffer.available() {
126        return Ok(());
127    }
128    let mut write = buffer.write();
129    if file.seek(SeekFrom::Start(file_offset as u64)).is_err() {
130        // Ignore seek errors, treat them as though the data isn't there.
131        return Ok(());
132    }
133    loop {
134        match file.read(&mut write) {
135            Ok(0) => {
136                // We're at the end of the file.  Nothing to do.
137                break;
138            }
139            Ok(len) => {
140                // Some data has been read.
141                write.written(len);
142                break;
143            }
144            Err(ref e) if e.kind() == std::io::ErrorKind::Interrupted => {}
145            Err(e) => {
146                return Err(e.into());
147            }
148        }
149    }
150    Ok(())
151}
152
153#[cfg(test)]
154mod test {
155    use super::*;
156    use std::io::Write;
157    use tempfile::NamedTempFile;
158
159    #[test]
160    fn basic() -> Result<(), Error> {
161        let mut t = NamedTempFile::new()?;
162        write!(t, "HERE IS SOME DATA")?;
163        t.flush()?;
164        let mut c = BufferCache::new(t.path(), 4, 2);
165        let mut read_range = |start, end| c.with_slice(start, end, |data| data.into_owned());
166        assert_eq!(read_range(0, 4)?.as_slice(), b"HERE");
167        assert_eq!(read_range(5, 7)?.as_slice(), b"IS");
168        assert_eq!(read_range(3, 9)?.as_slice(), b"E IS S");
169        assert_eq!(read_range(0, 17)?.as_slice(), b"HERE IS SOME DATA");
170        assert_eq!(read_range(0, 20)?.as_slice(), b"HERE IS SOME DATA\0\0\0");
171        Ok(())
172    }
173}
streampager/buffer_cache.rs

streampager/
buffer_cache.rs
