diff options
| -rw-r--r-- | examples/demo1.rs | 179 | ||||
| -rw-r--r-- | src/lib.rs | 3 | ||||
| -rw-r--r-- | src/wal.rs | 41 |
3 files changed, 160 insertions, 63 deletions
diff --git a/examples/demo1.rs b/examples/demo1.rs index 625f40d..47d1423 100644 --- a/examples/demo1.rs +++ b/examples/demo1.rs @@ -1,11 +1,17 @@ -use std::os::unix::io::RawFd; -use nix::unistd::{close, mkdir, unlinkat, UnlinkatFlags, ftruncate}; -use nix::fcntl::{open, openat, OFlag, fallocate, FallocateFlags}; -use nix::sys::{stat::Mode, uio::{pwrite, pread}}; -use rand::{Rng, seq::SliceRandom}; +use async_trait::async_trait; use libc::off_t; +use nix::fcntl::{fallocate, open, openat, FallocateFlags, OFlag}; +use nix::sys::{ + stat::Mode, + uio::{pread, pwrite}, +}; +use nix::unistd::{close, ftruncate, mkdir, unlinkat, UnlinkatFlags}; +use rand::{seq::SliceRandom, Rng}; +use std::os::unix::io::RawFd; -use growthring::wal::{WALFile, WALStore, WALPos, WALBytes, WALLoader, WALWriter, WALRingId}; +use growthring::wal::{ + WALBytes, WALFile, WALLoader, WALPos, WALRingId, WALStore, WALWriter, +}; struct WALFileTest { filename: String, @@ -14,15 +20,17 @@ struct WALFileTest { impl WALFileTest { fn new(rootfd: RawFd, filename: &str) -> Result<Self, ()> { - openat(rootfd, filename, + openat( + rootfd, + filename, OFlag::O_CREAT | OFlag::O_RDWR, - Mode::S_IRUSR | Mode::S_IWUSR).and_then(|fd| { + Mode::S_IRUSR | Mode::S_IWUSR, + ) + .and_then(|fd| { let filename = filename.to_string(); - Ok (WALFileTest { - filename, - fd, - }) - }).or_else(|_| Err(())) + Ok(WALFileTest { filename, fd }) + }) + .or_else(|_| Err(())) } } @@ -32,12 +40,23 @@ impl Drop for WALFileTest { } } +#[async_trait(?Send)] impl WALFile for WALFileTest { - fn allocate(&self, offset: WALPos, length: usize) -> Result<(), ()> { - println!("{}.allocate(offset=0x{:x}, end=0x{:x})", self.filename, offset, offset + length as u64); - fallocate(self.fd, - FallocateFlags::FALLOC_FL_ZERO_RANGE, - offset as off_t, length as off_t).and_then(|_| Ok(())).or_else(|_| Err(())) + async fn allocate(&self, offset: WALPos, length: usize) -> Result<(), ()> { + println!( + "{}.allocate(offset=0x{:x}, end=0x{:x})", + self.filename, + offset, + offset + length as u64 + ); + fallocate( + self.fd, + FallocateFlags::FALLOC_FL_ZERO_RANGE, + offset as off_t, + length as off_t, + ) + .and_then(|_| Ok(())) + .or_else(|_| Err(())) } fn truncate(&self, length: usize) -> Result<(), ()> { @@ -45,26 +64,47 @@ impl WALFile for WALFileTest { ftruncate(self.fd, length as off_t).or_else(|_| Err(())) } - fn write(&self, offset: WALPos, data: WALBytes) -> Result<(), ()> { - println!("{}.write(offset=0x{:x}, end=0x{:x}, data=0x{})", - self.filename, offset, offset + data.len() as u64, hex::encode(&data)); + async fn write(&self, offset: WALPos, data: WALBytes) -> Result<(), ()> { + println!( + "{}.write(offset=0x{:x}, end=0x{:x}, data=0x{})", + self.filename, + offset, + offset + data.len() as u64, + hex::encode(&data) + ); pwrite(self.fd, &*data, offset as off_t) .or_else(|_| Err(())) - .and_then(|nwrote| if nwrote == data.len() { Ok(()) } else { Err(()) }) + .and_then(|nwrote| { + if nwrote == data.len() { + Ok(()) + } else { + Err(()) + } + }) } - fn read(&self, offset: WALPos, length: usize) -> Result<Option<WALBytes>, ()> { + fn read( + &self, + offset: WALPos, + length: usize, + ) -> Result<Option<WALBytes>, ()> { let mut buff = Vec::new(); buff.resize(length, 0); pread(self.fd, &mut buff[..], offset as off_t) .or_else(|_| Err(())) - .and_then(|nread| Ok(if nread == length {Some(buff.into_boxed_slice())} else {None})) + .and_then(|nread| { + Ok(if nread == length { + Some(buff.into_boxed_slice()) + } else { + None + }) + }) } } struct WALStoreTest { rootfd: RawFd, - rootpath: String + rootpath: String, } impl WALStoreTest { @@ -74,12 +114,20 @@ impl WALStoreTest { let _ = std::fs::remove_dir_all(wal_dir); } match mkdir(wal_dir, Mode::S_IRUSR | Mode::S_IWUSR | Mode::S_IXUSR) { - Err(e) => if truncate { panic!("error while creating directory: {}", e) }, - Ok(_) => () + Err(e) => { + if truncate { + panic!("error while creating directory: {}", e) + } + } + Ok(_) => (), } - let rootfd = match open(wal_dir, OFlag::O_DIRECTORY | OFlag::O_PATH, Mode::empty()) { + let rootfd = match open( + wal_dir, + OFlag::O_DIRECTORY | OFlag::O_PATH, + Mode::empty(), + ) { Ok(fd) => fd, - Err(_) => panic!("error while opening the DB") + Err(_) => panic!("error while opening the DB"), }; WALStoreTest { rootfd, rootpath } } @@ -95,15 +143,25 @@ impl Drop for WALStoreTest { impl WALStore for WALStoreTest { type FileNameIter = std::vec::IntoIter<String>; - async fn open_file(&self, filename: &str, touch: bool) -> Result<Box<dyn WALFile>, ()> { + async fn open_file( + &self, + filename: &str, + touch: bool, + ) -> Result<Box<dyn WALFile>, ()> { println!("open_file(filename={}, touch={})", filename, touch); let filename = filename.to_string(); - WALFileTest::new(self.rootfd, &filename).and_then(|f| Ok(Box::new(f) as Box<dyn WALFile>)) + WALFileTest::new(self.rootfd, &filename) + .and_then(|f| Ok(Box::new(f) as Box<dyn WALFile>)) } async fn remove_file(&self, filename: String) -> Result<(), ()> { println!("remove_file(filename={})", filename); - unlinkat(Some(self.rootfd), &filename, UnlinkatFlags::NoRemoveDir).or_else(|_| Err(())) + unlinkat( + Some(self.rootfd), + filename.as_str(), + UnlinkatFlags::NoRemoveDir, + ) + .or_else(|_| Err(())) } fn enumerate_files(&self) -> Result<Self::FileNameIter, ()> { @@ -115,21 +173,35 @@ impl WALStore for WALStoreTest { Ok(logfiles.into_iter()) } - fn apply_payload(&self, payload: WALBytes, ringid: WALRingId) -> Result<(), ()> { - println!("apply_payload(payload={}, ringid={:?})", - std::str::from_utf8(&payload).unwrap(), - ringid); + fn apply_payload( + &self, + payload: WALBytes, + ringid: WALRingId, + ) -> Result<(), ()> { + println!( + "apply_payload(payload={}, ringid={:?})", + std::str::from_utf8(&payload).unwrap(), + ringid + ); Ok(()) } } -fn test(records: Vec<String>, wal: &mut WALWriter<WALStoreTest>) -> Box<[WALRingId]> { - let records: Vec<WALBytes> = records.into_iter().map(|s| s.into_bytes().into_boxed_slice()).collect(); - let ret = wal.grow(&records).unwrap(); - for ring_id in ret.iter() { +fn test( + records: Vec<String>, + wal: &mut WALWriter<WALStoreTest>, +) -> Vec<WALRingId> { + let records: Vec<WALBytes> = records + .into_iter() + .map(|s| s.into_bytes().into_boxed_slice()) + .collect(); + let mut res = Vec::new(); + for r in wal.grow(&records).into_iter() { + let ring_id = futures::executor::block_on(r).unwrap(); println!("got ring id: {:?}", ring_id); + res.push(ring_id); } - ret + res } fn main() { @@ -137,16 +209,33 @@ fn main() { let store = WALStoreTest::new("./wal_demo1", true); let mut wal = WALLoader::new(9, 8, 1000).recover(store).unwrap(); for _ in 0..3 { - test(["hi", "hello", "lol"].iter().map(|s| s.to_string()).collect::<Vec<String>>(), &mut wal); + test( + ["hi", "hello", "lol"] + .iter() + .map(|s| s.to_string()) + .collect::<Vec<String>>(), + &mut wal, + ); } for _ in 0..3 { - test(vec!["a".repeat(10), "b".repeat(100), "c".repeat(1000)], &mut wal); + test( + vec!["a".repeat(10), "b".repeat(100), "c".repeat(1000)], + &mut wal, + ); } let store = WALStoreTest::new("./wal_demo1", false); let mut wal = WALLoader::new(9, 8, 1000).recover(store).unwrap(); for _ in 0..3 { - test(vec!["a".repeat(10), "b".repeat(100), "c".repeat(300), "d".repeat(400)], &mut wal); + test( + vec![ + "a".repeat(10), + "b".repeat(100), + "c".repeat(300), + "d".repeat(400), + ], + &mut wal, + ); } let store = WALStoreTest::new("./wal_demo1", false); @@ -165,7 +254,7 @@ fn main() { ids.shuffle(&mut rng); for e in ids.chunks(20) { println!("peel(20)"); - wal.peel(e); + futures::executor::block_on(wal.peel(e)).unwrap(); } } } @@ -1,3 +1,2 @@ -#[macro_use] -extern crate scan_fmt; +#[macro_use] extern crate scan_fmt; pub mod wal; @@ -24,8 +24,8 @@ struct WALRingBlob { // payload follows } +type WALFileId = u64; pub type WALBytes = Box<[u8]>; -pub type WALFileId = u64; pub type WALPos = u64; #[derive(Eq, PartialEq, Copy, Clone, Debug, Hash)] @@ -77,12 +77,12 @@ struct WALState { pub trait WALFile { /// Initialize the file space in [offset, offset + length) to zero. async fn allocate(&self, offset: WALPos, length: usize) -> Result<(), ()>; - /// Write data with offset. We assume all previous `allocate/truncate` invocations are visible - /// if ordered earlier (should be guaranteed by most OS). Additionally, the final write caused + /// Write data with offset. We assume all previous `allocate`/`truncate` invocations are visible + /// if ordered earlier (should be guaranteed by most OS). Additionally, the write caused /// by each invocation of this function should be _atomic_ (the entire single write should be /// all or nothing). async fn write(&self, offset: WALPos, data: WALBytes) -> Result<(), ()>; - /// Read data with offset. Return Ok(None) when it reaches EOF. + /// Read data with offset. Return `Ok(None)` when it reaches EOF. fn read( &self, offset: WALPos, @@ -259,7 +259,11 @@ impl<F: WALStore> WALFilePool<F> { res } - fn remove_files<'a>(&'a mut self, fid_s: u64, fid_e: u64) -> impl Future<Output = Result<(), ()>> + 'a { + fn remove_files<'a>( + &'a mut self, + fid_s: u64, + fid_e: u64, + ) -> impl Future<Output = Result<(), ()>> + 'a { let last_peel = unsafe { std::mem::replace( &mut *self.last_peel.get(), @@ -279,11 +283,12 @@ impl<F: WALStore> WALFilePool<F> { r.await? } Ok(()) - }.shared(); + } + .shared(); unsafe { - (*self.last_peel.get()) = MaybeUninit::new( - std::mem::transmute( - Box::pin(p.clone()) as Pin<Box<dyn Future<Output = _> + 'a>>)) + (*self.last_peel.get()) = + MaybeUninit::new(std::mem::transmute(Box::pin(p.clone()) + as Pin<Box<dyn Future<Output = _> + 'a>>)) } p } @@ -316,9 +321,11 @@ impl<F: WALStore> WALWriter<F> { } } - /// Submit a sequence of records to WAL; WALStore/WALFile callbacks are invoked before the - /// function returns. The caller then has the knowledge of WAL writes so it should defer - /// actual data writes after WAL writes. + /// Submit a sequence of records to WAL. It returns a vector of futures, each of which + /// corresponds to one record. When a future resolves to `WALRingId`, it is guaranteed the + /// record is already logged. Then, after finalizing the changes encoded by that record to + /// the persistent storage, the caller can recycle the WAL files by invoking the given + /// `peel` with the given `WALRingId`s. pub fn grow<'a, T: AsRef<[WALBytes]>>( &'a mut self, records: T, @@ -455,8 +462,9 @@ impl<F: WALStore> WALWriter<F> { res } - /// Inform the WALWriter that data writes (specified by a slice of (offset, length) tuples) are - /// complete so that it could automatically remove obsolete WAL files. + /// Inform the `WALWriter` that some data writes are complete so that it could automatically + /// remove obsolete WAL files. The given list of `WALRingId` does not need to be ordered and + /// could be of arbitrary length. pub fn peel<'a, T: AsRef<[WALRingId]>>( &'a mut self, records: T, @@ -468,7 +476,8 @@ impl<F: WALStore> WALWriter<F> { state.io_complete.push(*rec); } let orig_fid = state.first_fid; - while let Some(s) = state.io_complete.peek().and_then(|&e| Some(e.start)) + while let Some(s) = + state.io_complete.peek().and_then(|&e| Some(e.start)) { if s != state.next_complete { break; @@ -509,7 +518,7 @@ impl WALLoader { } } - /// Recover by reading the WAL log files. + /// Recover by reading the WAL files. pub fn recover<F: WALStore>(self, store: F) -> Result<WALWriter<F>, ()> { let mut file_pool = WALFilePool::new( store, |