feat: cache fd

Author: k-jingyang

benches/value_log.rs

@@ -92,10 +92,10 @@ impl BlobCache for NoCacher {
 }
 
 impl FDCache for NoCacher {
-    fn get(&self, _: ValueLogId, _: BlobFileId) -> Option<std::io::BufReader<std::fs::File>> {
+    fn get(&self, _: ValueLogId, _: BlobFileId) -> Option<std::fs::File> {
         None
     }
-    fn insert(&self, _: ValueLogId, _: BlobFileId, _: BufReader<File>) {}
+    fn insert(&self, _: ValueLogId, _: BlobFileId, _: File) {}
 }
 
 fn prefetch(c: &mut Criterion) {

src/fd_cache.rs

@@ -8,12 +8,12 @@ use std::{fs::File, io::BufReader};
 /// The unique identifier for a value log blob file. Another name for SegmentId
 pub type BlobFileId = u64;
 
-/// File descriptor cache, to cache file descriptors after an open() syscall.
-/// Reduces the number of open() syscalls needed accessing the same blob file.
+/// File descriptor cache, to cache file descriptors after an fopen().
+/// Reduces the number of fopen() needed when accessing the same blob file.
 pub trait FDCache: Clone {
     /// Caches a file descriptor
-    fn insert(&self, vlog_id: ValueLogId, blob_file_id: BlobFileId, fd: BufReader<File>);
+    fn insert(&self, vlog_id: ValueLogId, blob_file_id: BlobFileId, fd: File);
 
     /// Retrieves a file descriptor from the cache, or `None` if it could not be found
-    fn get(&self, vlog_id: ValueLogId, blob_file_id: BlobFileId) -> Option<BufReader<File>>;
+    fn get(&self, vlog_id: ValueLogId, blob_file_id: BlobFileId) -> Option<File>;
 }

src/value_log.rs

@@ -282,8 +282,17 @@ impl<BC: BlobCache, C: Compressor + Clone, FDC: FDCache> ValueLog<BC, FDC, C> {
             return Ok(None);
         };
 
-        // TODO: this is the part with the repeated fopen() to the same file
-        let mut reader = BufReader::new(File::open(&segment.path)?);
+        let fd = match self.fd_cache.get(self.id, vhandle.segment_id) {
+            Some(fd) => fd,
+            None => {
+                let fd = File::open(&segment.path)?;
+                let fd_clone = fd.try_clone()?;
+                self.fd_cache.insert(self.id, vhandle.segment_id, fd_clone);
+                fd
+            }
+        };
+
+        let mut reader = BufReader::new(fd);
         reader.seek(std::io::SeekFrom::Start(vhandle.offset))?;
         let mut reader = SegmentReader::with_reader(vhandle.segment_id, reader)
             .use_compression(self.config.compression.clone());
@@ -315,6 +324,9 @@ impl<BC: BlobCache, C: Compressor + Clone, FDC: FDCache> ValueLog<BC, FDC, C> {
             self.blob_cache.insert(self.id, &value_handle, val);
         }
 
+        // cache the fd for future use, must ensure to always use SeekFrom::Start
+        // self.fd_cache.insert(self.id, vhandle.segment_id, reader.inner);
+
         Ok(Some(val))
     }
 

tests/common.rs

@@ -95,8 +95,8 @@ impl BlobCache for NoCacher {
 }
 
 impl FDCache for NoCacher {
-    fn get(&self, _: ValueLogId, _: BlobFileId) -> Option<std::io::BufReader<std::fs::File>> {
+    fn get(&self, _: ValueLogId, _: BlobFileId) -> Option<std::fs::File> {
         None
     }
-    fn insert(&self, _: ValueLogId, _: BlobFileId, _: BufReader<File>) {}
+    fn insert(&self, _: ValueLogId, _: BlobFileId, _: File) {}
 }