make the directory BufWriter capacity configurable (#52)

Author: rebasedming

src/directory/directory.rs

@@ -10,7 +10,9 @@ use log::Level;
 
 use crate::directory::directory_lock::Lock;
 use crate::directory::error::{DeleteError, LockError, OpenReadError, OpenWriteError};
-use crate::directory::{FileHandle, FileSlice, WatchCallback, WatchHandle, WritePtr};
+use crate::directory::{
+    FileHandle, FileSlice, TerminatingWrite, WatchCallback, WatchHandle, WritePtr,
+};
 use crate::index::SegmentMetaInventory;
 use crate::IndexMeta;
 
@@ -143,6 +145,10 @@ pub trait Directory: DirectoryClone + fmt::Debug + Send + Sync + 'static {
     /// Returns true if and only if the file exists
     fn exists(&self, path: &Path) -> Result<bool, OpenReadError>;
 
+    /// Returns a boxed `TerminatingWrite` object, to be passed into `open_write`
+    /// which wraps it in a `BufWriter`
+    fn open_write_inner(&self, path: &Path) -> Result<Box<dyn TerminatingWrite>, OpenWriteError>;
+
     /// Opens a writer for the *virtual file* associated with
     /// a [`Path`].
     ///
@@ -169,7 +175,12 @@ pub trait Directory: DirectoryClone + fmt::Debug + Send + Sync + 'static {
     /// panic! if `flush` was not called.
     ///
     /// The file may not previously exist.
-    fn open_write(&self, path: &Path) -> Result<WritePtr, OpenWriteError>;
+    fn open_write(&self, path: &Path) -> Result<WritePtr, OpenWriteError> {
+        Ok(io::BufWriter::with_capacity(
+            self.bufwriter_capacity(),
+            self.open_write_inner(path)?,
+        ))
+    }
 
     /// Reads the full content file that has been written using
     /// [`Directory::atomic_write()`].
@@ -296,6 +307,10 @@ pub trait Directory: DirectoryClone + fmt::Debug + Send + Sync + 'static {
     fn log(&self, message: &str) {
         log!(Level::Info, "{message}");
     }
+
+    fn bufwriter_capacity(&self) -> usize {
+        8192
+    }
 }
 
 /// DirectoryClone

src/directory/managed_directory.rs

@@ -12,7 +12,7 @@ use crate::directory::error::{DeleteError, LockError, OpenReadError, OpenWriteEr
 use crate::directory::footer::{Footer, FooterProxy, FOOTER_LEN};
 use crate::directory::{
     DirectoryLock, DirectoryPanicHandler, FileHandle, FileSlice, GarbageCollectionResult, Lock,
-    WatchCallback, WatchHandle, WritePtr, MANAGED_LOCK, META_LOCK,
+    TerminatingWrite, WatchCallback, WatchHandle, MANAGED_LOCK, META_LOCK,
 };
 use crate::error::DataCorruption;
 use crate::index::SegmentMetaInventory;
@@ -310,16 +310,19 @@ impl Directory for ManagedDirectory {
         Ok(reader)
     }
 
-    fn open_write(&self, path: &Path) -> result::Result<WritePtr, OpenWriteError> {
+    fn open_write_inner(
+        &self,
+        path: &Path,
+    ) -> result::Result<Box<dyn TerminatingWrite>, OpenWriteError> {
         self.register_file_as_managed(path)
             .map_err(|io_error| OpenWriteError::wrap_io_error(io_error, path.to_path_buf()))?;
-        Ok(io::BufWriter::new(Box::new(FooterProxy::new(
+        Ok(Box::new(FooterProxy::new(
             self.directory
                 .open_write(path)?
                 .into_inner()
                 .map_err(|_| ())
                 .expect("buffer should be empty"),
-        ))))
+        )))
     }
 
     fn atomic_write(&self, path: &Path, data: &[u8]) -> io::Result<()> {
@@ -380,6 +383,10 @@ impl Directory for ManagedDirectory {
     fn log(&self, message: &str) {
         self.directory.log(message);
     }
+
+    fn bufwriter_capacity(&self) -> usize {
+        self.directory.bufwriter_capacity()
+    }
 }
 
 impl Clone for ManagedDirectory {

src/directory/mmap_directory.rs

@@ -1,7 +1,7 @@
 use std::collections::HashMap;
 use std::fmt;
 use std::fs::{self, File, OpenOptions};
-use std::io::{self, BufWriter, Read, Write};
+use std::io::{self, Read, Write};
 use std::ops::Deref;
 use std::path::{Path, PathBuf};
 use std::sync::{Arc, RwLock, Weak};
@@ -21,7 +21,7 @@ use crate::directory::error::{
 use crate::directory::file_watcher::FileWatcher;
 use crate::directory::{
     AntiCallToken, Directory, DirectoryLock, FileHandle, Lock, OwnedBytes, TerminatingWrite,
-    WatchCallback, WatchHandle, WritePtr,
+    WatchCallback, WatchHandle,
 };
 
 pub type ArcBytes = Arc<dyn Deref<Target = [u8]> + Send + Sync + 'static>;
@@ -413,7 +413,7 @@ impl Directory for MmapDirectory {
             .map_err(|io_err| OpenReadError::wrap_io_error(io_err, path.to_path_buf()))
     }
 
-    fn open_write(&self, path: &Path) -> Result<WritePtr, OpenWriteError> {
+    fn open_write_inner(&self, path: &Path) -> Result<Box<dyn TerminatingWrite>, OpenWriteError> {
         debug!("Open Write {:?}", path);
         let full_path = self.resolve_path(path);
 
@@ -443,7 +443,7 @@ impl Directory for MmapDirectory {
         // sync_directory() is called.
 
         let writer = SafeFileWriter::new(file);
-        Ok(BufWriter::new(Box::new(writer)))
+        Ok(Box::new(writer))
     }
 
     fn atomic_read(&self, path: &Path) -> Result<Vec<u8>, OpenReadError> {

src/directory/ram_directory.rs

@@ -1,5 +1,5 @@
 use std::collections::HashMap;
-use std::io::{self, BufWriter, Cursor, Write};
+use std::io::{self, Cursor, Write};
 use std::path::{Path, PathBuf};
 use std::sync::{Arc, RwLock};
 use std::{fmt, result};
@@ -11,7 +11,7 @@ use crate::core::META_FILEPATH;
 use crate::directory::error::{DeleteError, OpenReadError, OpenWriteError};
 use crate::directory::{
     AntiCallToken, Directory, FileSlice, TerminatingWrite, WatchCallback, WatchCallbackList,
-    WatchHandle, WritePtr,
+    WatchHandle,
 };
 
 /// Writer associated with the [`RamDirectory`].
@@ -197,7 +197,7 @@ impl Directory for RamDirectory {
             .exists(path))
     }
 
-    fn open_write(&self, path: &Path) -> Result<WritePtr, OpenWriteError> {
+    fn open_write_inner(&self, path: &Path) -> Result<Box<dyn TerminatingWrite>, OpenWriteError> {
         let mut fs = self.fs.write().unwrap();
         let path_buf = PathBuf::from(path);
         let vec_writer = VecWriter::new(path_buf.clone(), self.clone());
@@ -206,7 +206,7 @@ impl Directory for RamDirectory {
         if exists {
             Err(OpenWriteError::FileAlreadyExists(path_buf))
         } else {
-            Ok(BufWriter::new(Box::new(vec_writer)))
+            Ok(Box::new(vec_writer))
         }
     }