diff options
| -rw-r--r-- | Cargo.lock | 6 | ||||
| -rw-r--r-- | Cargo.toml | 10 | ||||
| -rw-r--r-- | examples/demo1.rs | 6 | ||||
| -rw-r--r-- | src/lib.rs | 39 | ||||
| -rw-r--r-- | tests/common/mod.rs | 2 |
5 files changed, 36 insertions, 27 deletions
@@ -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", @@ -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(); @@ -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; |