use bytes::Bytes;
use memmap2::Mmap;
use parking_lot::RwLock;
use std::fs::File;
use std::path::{Path, PathBuf};
use std::sync::Arc;
use crate::{Error, Result};
#[derive(Clone)]
pub struct MmapReader {
inner: Arc<MmapReaderInner>,
}
struct MmapReaderInner {
path: PathBuf,
mmap: RwLock<Option<Mmap>>,
}
impl MmapReader {
pub fn open(path: impl AsRef<Path>) -> Result<Self> {
let path = path.as_ref().to_path_buf();
let file = File::open(&path)?;
let mmap = unsafe { Mmap::map(&file)? };
Ok(Self {
inner: Arc::new(MmapReaderInner {
path,
mmap: RwLock::new(Some(mmap)),
}),
})
}
pub fn read(&self, offset: u64, len: usize) -> Result<Bytes> {
let mmap_guard = self.inner.mmap.read();
let mmap = mmap_guard.as_ref()
.ok_or_else(|| Error::Internal("Mmap has been closed".to_string()))?;
let offset = offset as usize;
let end = offset.checked_add(len)
.ok_or_else(|| Error::InvalidArgument("Read would overflow".to_string()))?;
if end > mmap.len() {
return Err(Error::InvalidArgument(format!(
"Read beyond file bounds: {} + {} > {}",
offset, len, mmap.len()
)));
}
Ok(Bytes::copy_from_slice(&mmap[offset..end]))
}
pub fn read_block(&self, offset: u64) -> Result<Bytes> {
self.read(offset, crate::layout::BLOCK_SIZE)
}
pub fn len(&self) -> usize {
let mmap_guard = self.inner.mmap.read();
mmap_guard.as_ref().map(|m| m.len()).unwrap_or(0)
}
pub fn is_empty(&self) -> bool {
self.len() == 0
}
pub fn path(&self) -> &Path {
&self.inner.path
}
pub fn close(&self) {
let mut mmap_guard = self.inner.mmap.write();
*mmap_guard = None;
}
}
pub struct MmapPool {
readers: RwLock<std::collections::HashMap<PathBuf, MmapReader>>,
}
impl MmapPool {
pub fn new() -> Self {
Self {
readers: RwLock::new(std::collections::HashMap::new()),
}
}
pub fn get_or_open(&self, path: impl AsRef<Path>) -> Result<MmapReader> {
let path = path.as_ref();
{
let readers = self.readers.read();
if let Some(reader) = readers.get(path) {
return Ok(reader.clone());
}
}
let reader = MmapReader::open(path)?;
let mut readers = self.readers.write();
readers.insert(path.to_path_buf(), reader.clone());
Ok(reader)
}
pub fn remove(&self, path: impl AsRef<Path>) {
let mut readers = self.readers.write();
if let Some(reader) = readers.remove(path.as_ref()) {
reader.close();
}
}
pub fn clear(&self) {
let mut readers = self.readers.write();
for (_, reader) in readers.drain() {
reader.close();
}
}
pub fn len(&self) -> usize {
let readers = self.readers.read();
readers.len()
}
pub fn is_empty(&self) -> bool {
self.len() == 0
}
}
impl Default for MmapPool {
fn default() -> Self {
Self::new()
}
}
#[cfg(test)]
mod tests {
use super::*;
use tempfile::NamedTempFile;
use std::io::Write;
#[test]
fn test_mmap_reader_basic() {
let mut tmp = NamedTempFile::new().unwrap();
tmp.write_all(b"hello world").unwrap();
tmp.flush().unwrap();
let reader = MmapReader::open(tmp.path()).unwrap();
let data = reader.read(0, 5).unwrap();
assert_eq!(&data[..], b"hello");
let data = reader.read(6, 5).unwrap();
assert_eq!(&data[..], b"world");
assert_eq!(reader.len(), 11);
assert!(!reader.is_empty());
}
#[test]
fn test_mmap_reader_out_of_bounds() {
let mut tmp = NamedTempFile::new().unwrap();
tmp.write_all(b"test").unwrap();
tmp.flush().unwrap();
let reader = MmapReader::open(tmp.path()).unwrap();
let result = reader.read(0, 100);
assert!(matches!(result, Err(Error::InvalidArgument(_))));
let result = reader.read(100, 1);
assert!(matches!(result, Err(Error::InvalidArgument(_))));
}
#[test]
fn test_mmap_reader_close() {
let mut tmp = NamedTempFile::new().unwrap();
tmp.write_all(b"data").unwrap();
tmp.flush().unwrap();
let reader = MmapReader::open(tmp.path()).unwrap();
let data = reader.read(0, 4).unwrap();
assert_eq!(&data[..], b"data");
reader.close();
let result = reader.read(0, 1);
assert!(matches!(result, Err(Error::Internal(_))));
}
#[test]
fn test_mmap_pool_basic() {
let mut tmp = NamedTempFile::new().unwrap();
tmp.write_all(b"pool test").unwrap();
tmp.flush().unwrap();
let pool = MmapPool::new();
assert!(pool.is_empty());
let reader1 = pool.get_or_open(tmp.path()).unwrap();
assert_eq!(pool.len(), 1);
let reader2 = pool.get_or_open(tmp.path()).unwrap();
assert_eq!(pool.len(), 1);
let data1 = reader1.read(0, 4).unwrap();
let data2 = reader2.read(0, 4).unwrap();
assert_eq!(data1, data2);
}
#[test]
fn test_mmap_pool_remove() {
let mut tmp = NamedTempFile::new().unwrap();
tmp.write_all(b"test").unwrap();
tmp.flush().unwrap();
let pool = MmapPool::new();
let _reader = pool.get_or_open(tmp.path()).unwrap();
assert_eq!(pool.len(), 1);
pool.remove(tmp.path());
assert_eq!(pool.len(), 0);
}
#[test]
fn test_mmap_pool_clear() {
let mut tmp1 = NamedTempFile::new().unwrap();
tmp1.write_all(b"file1").unwrap();
tmp1.flush().unwrap();
let mut tmp2 = NamedTempFile::new().unwrap();
tmp2.write_all(b"file2").unwrap();
tmp2.flush().unwrap();
let pool = MmapPool::new();
pool.get_or_open(tmp1.path()).unwrap();
pool.get_or_open(tmp2.path()).unwrap();
assert_eq!(pool.len(), 2);
pool.clear();
assert_eq!(pool.len(), 0);
}
#[test]
fn test_mmap_read_block() {
let mut tmp = NamedTempFile::new().unwrap();
let data = vec![0xAB; crate::layout::BLOCK_SIZE];
tmp.write_all(&data).unwrap();
tmp.flush().unwrap();
let reader = MmapReader::open(tmp.path()).unwrap();
let block = reader.read_block(0).unwrap();
assert_eq!(block.len(), crate::layout::BLOCK_SIZE);
assert_eq!(&block[..], &data[..]);
}
}