diff options
| -rw-r--r-- | examples/demo1.rs | 57 | ||||
| -rw-r--r-- | src/lib.rs | 70 |
2 files changed, 77 insertions, 50 deletions
diff --git a/examples/demo1.rs b/examples/demo1.rs index 536aba2..d1d488c 100644 --- a/examples/demo1.rs +++ b/examples/demo1.rs @@ -1,13 +1,11 @@ -use growthring::{WALFile, WALStore, WALPos, WALBytes, WALLoader, WALWriter}; - use std::os::unix::io::RawFd; -use nix::Error::Sys; -use nix::errno::Errno; -use nix::unistd::{close, mkdir, sysconf, SysconfVar}; +use nix::unistd::{close, mkdir, unlinkat, UnlinkatFlags, ftruncate}; use nix::fcntl::{open, openat, OFlag, fallocate, FallocateFlags}; -use nix::sys::{stat::Mode, uio::pwrite}; +use nix::sys::{stat::Mode, uio::{pwrite, pread}}; use libc::off_t; +use growthring::{WALFile, WALStore, WALPos, WALBytes, WALLoader, WALWriter}; + struct WALFileTest { filename: String, fd: RawFd, @@ -33,26 +31,39 @@ impl Drop for WALFileTest { } impl WALFile for WALFileTest { - fn allocate(&self, offset: WALPos, length: usize) { + 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); + 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<(), ()> { + println!("{}.truncate(length={})", self.filename, length); + ftruncate(self.fd, length as off_t).or_else(|_| Err(())) + } + fn write(&self, offset: WALPos, data: WALBytes) { 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); + pwrite(self.fd, &*data, offset as off_t).unwrap(); } fn read(&self, offset: WALPos, length: usize) -> WALBytes { - unreachable!() + let mut buff = Vec::new(); + buff.resize(length, 0); + pread(self.fd, &mut buff[..], offset as off_t).unwrap(); + buff.into_boxed_slice() } } struct WALStoreTest { - rootfd: RawFd + rootfd: RawFd, + rootpath: String } impl WALStoreTest { fn new(wal_dir: &str, truncate: bool) -> Self { + let rootpath = wal_dir.to_string(); if truncate { let _ = std::fs::remove_dir_all(wal_dir); } @@ -64,7 +75,7 @@ impl WALStoreTest { Ok(fd) => fd, Err(_) => panic!("error while opening the DB") }; - WALStoreTest { rootfd } + WALStoreTest { rootfd, rootpath } } } @@ -76,20 +87,27 @@ impl Drop for WALStoreTest { impl WALStore for WALStoreTest { fn open_file(&self, filename: &str, touch: bool) -> Option<Box<dyn WALFile>> { - println!("open_file(filename={}, touch={}", filename, touch); + println!("open_file(filename={}, touch={})", filename, touch); let filename = filename.to_string(); Some(Box::new(WALFileTest::new(self.rootfd, &filename))) } + fn remove_file(&self, filename: &str) -> bool { println!("remove_file(filename={})", filename); - true + unlinkat(Some(self.rootfd), filename, UnlinkatFlags::NoRemoveDir).is_ok() } + fn enumerate_files(&self) -> Box<[String]> { println!("enumerate_files()"); - Vec::new().into_boxed_slice() + let mut logfiles = Vec::new(); + for fname in std::fs::read_dir(&self.rootpath).unwrap() { + logfiles.push(fname.unwrap().file_name().into_string().unwrap()) + } + logfiles.into_boxed_slice() } + fn apply_payload(&self, payload: WALBytes) { - println!("apply_payload(payload=0x{})", hex::encode(payload)) + println!("apply_payload(payload={})", std::str::from_utf8(&payload).unwrap()) } } @@ -110,4 +128,11 @@ fn main() { for _ in 0..3 { test(["a".repeat(10), "b".repeat(100), "c".repeat(1000)].iter().map(|s| s.to_string()).collect::<Vec<String>>(), &mut wal) } + let store = WALStoreTest::new("./wal_demo1", false); + let mut wal = WALLoader::new(store, 9, 8, 1000).recover(); + for _ in 0..3 { + test(["a".repeat(10), "b".repeat(100), "c".repeat(300), "d".repeat(400)].iter().map(|s| s.to_string()).collect::<Vec<String>>(), &mut wal) + } + let store = WALStoreTest::new("./wal_demo1", false); + let wal = WALLoader::new(store, 9, 8, 1000).recover(); } @@ -55,7 +55,9 @@ struct WALState { pub trait WALFile { /// Initialize the file space in [offset, offset + length) to zero. - fn allocate(&self, offset: WALPos, length: usize); + fn allocate(&self, offset: WALPos, length: usize) -> Result<(), ()>; + /// Truncate a file to a specified length. + fn truncate(&self, length: usize) -> Result<(), ()>; /// Write data with offset. fn write(&self, offset: WALPos, data: WALBytes); /// Read data with offset. @@ -67,7 +69,7 @@ pub trait WALStore { fn open_file(&self, filename: &str, touch: bool) -> Option<Box<dyn WALFile>>; /// Unlink a file given the filename. fn remove_file(&self, filename: &str) -> bool; - /// Enumerate all WAL files, ordered by their filenames. + /// Enumerate all WAL files. fn enumerate_files(&self) -> Box<[String]>; /// Apply (redo) the payload during recovery. fn apply_payload(&self, payload: WALBytes); @@ -112,7 +114,7 @@ impl<F: WALStore> WALFilePool<F> { } fn get_fid(&mut self, fname: &str) -> WALFileId { - scan_fmt!(fname, "{:x}.log", [hex WALFileId]).unwrap() + scan_fmt!(fname, "{x}.log", [hex WALFileId]).unwrap() } // TODO: evict stale handles @@ -125,7 +127,7 @@ impl<F: WALStore> WALFilePool<F> { for (off, w) in &writes[1..] { let next_fid = off >> self.file_nbit; if next_fid != fid { - h.allocate(alloc_start, (alloc_end - alloc_start) as usize); + h.allocate(alloc_start, (alloc_end - alloc_start) as usize).unwrap(); h = self.get_file(next_fid, true); alloc_start = 0; alloc_end = alloc_start + w.len() as u64; @@ -134,7 +136,7 @@ impl<F: WALStore> WALFilePool<F> { alloc_end += w.len() as u64; } } - h.allocate(alloc_start, (alloc_end - alloc_start) as usize); + h.allocate(alloc_start, (alloc_end - alloc_start) as usize).unwrap(); for (off, w) in writes.into_iter() { self.get_file(off >> self.file_nbit, true).write(off & (self.file_size - 1), w); } @@ -143,6 +145,8 @@ impl<F: WALStore> WALFilePool<F> { fn remove_file(&self, fid: u64) -> bool { self.store.remove_file(&Self::get_fname(fid)) } + + fn reset(&mut self) { self.handles.clear() } } pub struct WALWriter<F: WALStore> { @@ -201,7 +205,6 @@ impl<F: WALStore> WALWriter<F> { if d >= rsize { // the remaining rec fits in the block let payload = rec; - println!("rsize {} {}", rsize, d); blob.crc32 = crc::crc32::checksum_ieee(payload); blob.rsize = rsize; blob.rtype = if started {WALRingType::Last} else {WALRingType::Full}; @@ -285,61 +288,60 @@ impl<F: WALStore> WALLoader<F> { } pub fn recover(mut self) -> WALWriter<F> { + let block_size = 1 << self.file_pool.block_nbit; let msize = std::mem::size_of::<WALRingBlob>() as u32; - let logfiles = self.file_pool.store.enumerate_files(); + let mut logfiles = self.file_pool.store.enumerate_files(); + // TODO: use regex to filter out invalid files + // TODO: check for missing logfiles + logfiles.sort(); + let mut chunks = None; for fname in logfiles.iter() { let fid = self.file_pool.get_fid(fname); let f = self.file_pool.get_file(fid, false); let mut off = 0; - let mut end = false; - while !end { + while block_size - (off & (block_size - 1)) > msize as u64 { let header_raw = f.read(off, msize as usize); + off += msize as u64; let header = unsafe { std::mem::transmute::<*const u8, &WALRingBlob>(header_raw.as_ptr())}; let rsize = header.rsize; - off += msize as u64; match header.rtype { WALRingType::Full => { + assert!(chunks.is_none()); let payload = f.read(off, rsize as usize); - self.file_pool.store.apply_payload(payload); off += rsize as u64; + self.file_pool.store.apply_payload(payload); }, WALRingType::First => { - let mut chunks = vec![f.read(off, rsize as usize)]; + assert!(chunks.is_none()); + chunks = Some(vec![f.read(off, rsize as usize)]); off += rsize as u64; - loop { - let header_raw = f.read(off, msize as usize); - let header = unsafe { - std::mem::transmute::<*const u8, &WALRingBlob>(header_raw.as_ptr())}; - if let WALRingType::Null = header.rtype { - end = true; - break - } - let rsize = header.rsize; - let payload = f.read(off, rsize as usize); - off += msize as u64; - chunks.push(payload); - match header.rtype { - WALRingType::Middle => (), - WALRingType::Last => break, - _ => unreachable!() - } - } + }, + WALRingType::Middle => { + chunks.as_mut().unwrap().push(f.read(off, rsize as usize)); + off += rsize as u64; + }, + WALRingType::Last => { + chunks.as_mut().unwrap().push(f.read(off, rsize as usize)); + off += rsize as u64; + + let _chunks = chunks.take().unwrap(); let mut payload = Vec::new(); - payload.resize(chunks.iter().fold(0, |acc, v| acc + v.len()), 0); + payload.resize(_chunks.iter().fold(0, |acc, v| acc + v.len()), 0); let mut ps = &mut payload[..]; - for c in chunks { + for c in _chunks { ps[..c.len()].copy_from_slice(&*c); ps = &mut ps[c.len()..]; } self.file_pool.store.apply_payload(payload.into_boxed_slice()); }, - WALRingType::Null => end = true, - _ => unreachable!() + WALRingType::Null => break, } } + f.truncate(0).unwrap(); self.file_pool.remove_file(fid); } + self.file_pool.reset(); WALWriter::new(WALState { first_fid: 0, next: 0, |