allow using customized aiomgrv0.1.2

5 files changed, 36 insertions, 27 deletions

diff --git a/Cargo.lock b/Cargo.lock
index dbe9b6b..96962a3 100644
--- a/Cargo.lock
+++ b/Cargo.lock
@@ -232,7 +232,7 @@ dependencies = [
 
 [[package]]
 name = "growth-ring"
-version = "0.1.1"
+version = "0.1.2"
 dependencies = [
  "async-trait",
  "crc",
@@ -281,9 +281,9 @@ checksum = "e2abad23fbc42b3700f2f279844dc832adb2b2eb069b2df918f455c4e18cc646"
 
 [[package]]
 name = "libaio-futures"
-version = "0.1.1"
+version = "0.1.2"
 source = "registry+https://github.com/rust-lang/crates.io-index"
-checksum = "049dd6ea8958e5c868151b08c68fa771daa979304ab3de8f5bab58126ba53e3b"
+checksum = "e831955e07aa19f495c993327d647d6a49bc9e5d6856238c00fd1833a1a06eee"
 dependencies = [
  "crossbeam-channel",
  "libc",
diff --git a/Cargo.toml b/Cargo.toml
index a30512d..a8a82ef 100644
--- a/Cargo.toml
+++ b/Cargo.toml
@@ -1,6 +1,6 @@
 [package]
 name = "growth-ring"
-version = "0.1.1"
+version = "0.1.2"
 authors = ["Determinant <[email protected]>"]
 edition = "2018"
 homepage = "https://github.com/Determinant/growth-ring"
@@ -14,10 +14,10 @@ description = "Simple and modular write-ahead-logging implementation."
 crc = "1.8.1"
 lru = "0.5.1"
 scan_fmt = "0.2.5"
-regex = "1"
-async-trait = "0.1"
-futures = "0.3"
-libaio-futures = "0.1.1"
+regex = "1.3.9"
+async-trait = "0.1.35"
+futures = "0.3.5"
+libaio-futures = "0.1.2"
 nix = "0.17.0"
 libc = "0.2.71"
 
diff --git a/examples/demo1.rs b/examples/demo1.rs
index e214177..d51801d 100644
--- a/examples/demo1.rs
+++ b/examples/demo1.rs
@@ -32,7 +32,7 @@ fn main() {
     let mut loader = WALLoader::new();
     loader.file_nbit(9).block_nbit(8);
 
-    let store = WALStoreAIO::new(&wal_dir, true, recover);
+    let store = WALStoreAIO::new(&wal_dir, true, recover, None).unwrap();
     let mut wal = loader.load(store).unwrap();
     for _ in 0..3 {
         test(
@@ -50,7 +50,7 @@ fn main() {
         );
     }
 
-    let store = WALStoreAIO::new(&wal_dir, false, recover);
+    let store = WALStoreAIO::new(&wal_dir, false, recover, None).unwrap();
     let mut wal = loader.load(store).unwrap();
     for _ in 0..3 {
         test(
@@ -64,7 +64,7 @@ fn main() {
         );
     }
 
-    let store = WALStoreAIO::new(&wal_dir, false, recover);
+    let store = WALStoreAIO::new(&wal_dir, false, recover, None).unwrap();
     let mut wal = loader.load(store).unwrap();
     for _ in 0..3 {
         let mut ids = Vec::new();
diff --git a/src/lib.rs b/src/lib.rs
index 447da08..f7b3f73 100644
--- a/src/lib.rs
+++ b/src/lib.rs
@@ -10,7 +10,7 @@
 //!
 //!
 //! // Start with empty WAL (truncate = true).
-//! let store = WALStoreAIO::new("./walfiles", true, |_, _| {Ok(())});
+//! let store = WALStoreAIO::new("./walfiles", true, |_, _| {Ok(())}, None).unwrap();
 //! let mut wal = loader.load(store).unwrap();
 //! // Write a vector of records to WAL.
 //! for f in wal.grow(vec!["record1(foo)", "record2(bar)", "record3(foobar)"]).into_iter() {
@@ -26,7 +26,7 @@
 //!              std::str::from_utf8(&payload).unwrap(),
 //!              ringid);
 //!     Ok(())
-//! });
+//! }, None).unwrap();
 //! let mut wal = loader.load(store).unwrap();
 //! // We saw some log playback, even there is no failure.
 //! // Let's try to grow the WAL to create many files.
@@ -40,7 +40,7 @@
 //! let store = WALStoreAIO::new("./walfiles", false, |payload, _| {
 //!     println!("payload.len() = {}", payload.len());
 //!     Ok(())
-//! });
+//! }, None).unwrap();
 //! let wal = loader.load(store).unwrap();
 //! // After each recovery, the ./walfiles is empty.
 //! ```
@@ -62,14 +62,14 @@ use wal::{WALBytes, WALFile, WALPos, WALRingId, WALStore};
 
 pub struct WALFileAIO {
     fd: RawFd,
-    aiomgr: Rc<RefCell<AIOManager<AIOBatchSchedulerIn>>>,
+    aiomgr: Rc<AIOManager<AIOBatchSchedulerIn>>,
 }
 
 impl WALFileAIO {
     pub fn new(
         rootfd: RawFd,
         filename: &str,
-        aiomgr: Rc<RefCell<AIOManager<AIOBatchSchedulerIn>>>,
+        aiomgr: Rc<AIOManager<AIOBatchSchedulerIn>>,
     ) -> Result<Self, ()> {
         openat(
             rootfd,
@@ -108,7 +108,6 @@ impl WALFile for WALFileAIO {
 
     async fn write(&self, offset: WALPos, data: WALBytes) -> Result<(), ()> {
         self.aiomgr
-            .borrow_mut()
             .write(self.fd, offset, data, None)
             .await
             .or_else(|_| Err(()))
@@ -126,7 +125,7 @@ impl WALFile for WALFileAIO {
         offset: WALPos,
         length: usize,
     ) -> Result<Option<WALBytes>, ()> {
-        block_on(self.aiomgr.borrow_mut().read(self.fd, offset, length, None))
+        block_on(self.aiomgr.read(self.fd, offset, length, None))
             .or_else(|_| Err(()))
             .and_then(|(nread, data)| {
                 Ok(if nread == length { Some(data) } else { None })
@@ -138,12 +137,25 @@ pub struct WALStoreAIO<F: FnMut(WALBytes, WALRingId) -> Result<(), ()>> {
     rootfd: RawFd,
     rootpath: String,
     recover_func: RefCell<F>,
+    aiomgr: Rc<AIOManager<AIOBatchSchedulerIn>>,
 }
 
 impl<F: FnMut(WALBytes, WALRingId) -> Result<(), ()>> WALStoreAIO<F> {
-    pub fn new(wal_dir: &str, truncate: bool, recover_func: F) -> Self {
+    pub fn new(
+        wal_dir: &str,
+        truncate: bool,
+        recover_func: F,
+        aiomgr: Option<AIOManager<AIOBatchSchedulerIn>>,
+    ) -> Result<Self, ()> {
         let recover_func = RefCell::new(recover_func);
         let rootpath = wal_dir.to_string();
+        let aiomgr = Rc::new(aiomgr.ok_or(Err(())).or_else(
+            |_: Result<AIOManager<AIOBatchSchedulerIn>, ()>| {
+                AIOManager::new(new_batch_scheduler(None), 128, None, None)
+                    .or(Err(()))
+            },
+        )?);
+
         if truncate {
             let _ = std::fs::remove_dir_all(wal_dir);
         }
@@ -163,11 +175,12 @@ impl<F: FnMut(WALBytes, WALRingId) -> Result<(), ()>> WALStoreAIO<F> {
             Ok(fd) => fd,
             Err(_) => panic!("error while opening the WAL directory"),
         };
-        WALStoreAIO {
+        Ok(WALStoreAIO {
             rootfd,
             rootpath,
             recover_func,
-        }
+            aiomgr,
+        })
     }
 }
 
@@ -183,11 +196,7 @@ impl<F: FnMut(WALBytes, WALRingId) -> Result<(), ()>> WALStore
         _touch: bool,
     ) -> Result<Box<dyn WALFile>, ()> {
         let filename = filename.to_string();
-        let aiomgr = Rc::new(RefCell::new(
-            AIOManager::new(new_batch_scheduler(None), 10, None, None)
-                .or(Err(()))?,
-        ));
-        WALFileAIO::new(self.rootfd, &filename, aiomgr.clone())
+        WALFileAIO::new(self.rootfd, &filename, self.aiomgr.clone())
             .and_then(|f| Ok(Box::new(f) as Box<dyn WALFile>))
     }
 
diff --git a/tests/common/mod.rs b/tests/common/mod.rs
index 247b2ee..769f422 100644
--- a/tests/common/mod.rs
+++ b/tests/common/mod.rs
@@ -3,7 +3,7 @@
 #[allow(dead_code)]
 use async_trait::async_trait;
 use growthring::wal::{
-    WALBytes, WALFile, WALLoader, WALPos, WALRingId, WALStore, RecoverPolicy
+    WALBytes, WALFile, WALLoader, WALPos, WALRingId, WALStore
 };
 use indexmap::{map::Entry, IndexMap};
 use rand::Rng;