get `peel()` working

4 files changed, 93 insertions, 19 deletions

diff --git a/Cargo.lock b/Cargo.lock
index 8aec840..7b26bbc 100644
--- a/Cargo.lock
+++ b/Cargo.lock
@@ -97,6 +97,7 @@ dependencies = [
  "libc",
  "lru",
  "nix",
+ "rand",
  "scan_fmt",
 ]
 
@@ -157,12 +158,59 @@ dependencies = [
 ]
 
 [[package]]
+name = "ppv-lite86"
+version = "0.2.8"
+source = "registry+https://github.com/rust-lang/crates.io-index"
+checksum = "237a5ed80e274dbc66f86bd59c1e25edc039660be53194b5fe0a482e0f2612ea"
+
+[[package]]
 name = "proc-macro-hack"
 version = "0.5.16"
 source = "registry+https://github.com/rust-lang/crates.io-index"
 checksum = "7e0456befd48169b9f13ef0f0ad46d492cf9d2dbb918bcf38e01eed4ce3ec5e4"
 
 [[package]]
+name = "rand"
+version = "0.7.3"
+source = "registry+https://github.com/rust-lang/crates.io-index"
+checksum = "6a6b1679d49b24bbfe0c803429aa1874472f50d9b363131f0e89fc356b544d03"
+dependencies = [
+ "getrandom",
+ "libc",
+ "rand_chacha",
+ "rand_core",
+ "rand_hc",
+]
+
+[[package]]
+name = "rand_chacha"
+version = "0.2.2"
+source = "registry+https://github.com/rust-lang/crates.io-index"
+checksum = "f4c8ed856279c9737206bf725bf36935d8666ead7aa69b52be55af369d193402"
+dependencies = [
+ "ppv-lite86",
+ "rand_core",
+]
+
+[[package]]
+name = "rand_core"
+version = "0.5.1"
+source = "registry+https://github.com/rust-lang/crates.io-index"
+checksum = "90bde5296fc891b0cef12a6d03ddccc162ce7b2aff54160af9338f8d40df6d19"
+dependencies = [
+ "getrandom",
+]
+
+[[package]]
+name = "rand_hc"
+version = "0.2.0"
+source = "registry+https://github.com/rust-lang/crates.io-index"
+checksum = "ca3129af7b92a17112d59ad498c6f81eaf463253766b90396d39ea7a39d6613c"
+dependencies = [
+ "rand_core",
+]
+
+[[package]]
 name = "regex"
 version = "1.3.9"
 source = "registry+https://github.com/rust-lang/crates.io-index"
diff --git a/Cargo.toml b/Cargo.toml
index 4193862..1593173 100644
--- a/Cargo.toml
+++ b/Cargo.toml
@@ -15,6 +15,7 @@ scan_fmt = "0.2.5"
 hex = "0.4.2"
 libc = "0.2.44"
 nix = "0.17.0"
+rand = "0.7.3"
 
 [lib]
 name = "growthring"
diff --git a/examples/demo1.rs b/examples/demo1.rs
index d1d488c..4c4f955 100644
--- a/examples/demo1.rs
+++ b/examples/demo1.rs
@@ -2,9 +2,10 @@ 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 libc::off_t;
 
-use growthring::{WALFile, WALStore, WALPos, WALBytes, WALLoader, WALWriter};
+use growthring::{WALFile, WALStore, WALPos, WALBytes, WALLoader, WALWriter, WALRingId};
 
 struct WALFileTest {
     filename: String,
@@ -92,9 +93,9 @@ impl WALStore for WALStoreTest {
         Some(Box::new(WALFileTest::new(self.rootfd, &filename)))
     }
 
-    fn remove_file(&self, filename: &str) -> bool {
+    fn remove_file(&self, filename: &str) -> Result<(), ()> {
         println!("remove_file(filename={})", filename);
-        unlinkat(Some(self.rootfd), filename, UnlinkatFlags::NoRemoveDir).is_ok()
+        unlinkat(Some(self.rootfd), filename, UnlinkatFlags::NoRemoveDir).or_else(|_| Err(()))
     }
 
     fn enumerate_files(&self) -> Box<[String]> {
@@ -111,28 +112,49 @@ impl WALStore for WALStoreTest {
     }
 }
 
-fn test(records: Vec<String>, wal: &mut WALWriter<WALStoreTest>) {
+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);
     for ring_id in ret.iter() {
         println!("got ring id: {:?}", ring_id);
     }
+    ret
 }
 
 fn main() {
+    let mut rng = rand::thread_rng();
     let store = WALStoreTest::new("./wal_demo1", true);
     let mut wal = WALLoader::new(store, 9, 8, 1000).recover();
     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(["a".repeat(10), "b".repeat(100), "c".repeat(1000)].iter().map(|s| s.to_string()).collect::<Vec<String>>(), &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(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)
+        test(vec!["a".repeat(10), "b".repeat(100), "c".repeat(300), "d".repeat(400)], &mut wal);
     }
+
     let store = WALStoreTest::new("./wal_demo1", false);
-    let wal = WALLoader::new(store, 9, 8, 1000).recover();
+    let mut wal = WALLoader::new(store, 9, 8, 1000).recover();
+    for _ in 0..3 {
+        let mut ids = Vec::new();
+        for _ in 0..3 {
+            let mut records = Vec::new();
+            for _ in 0..100 {
+                records.push("a".repeat(rng.gen_range(1, 10)))
+            }
+            for id in test(records, &mut wal).iter() {
+                ids.push(*id)
+            }
+        }
+        ids.shuffle(&mut rng);
+        for e in ids.chunks(20) {
+            println!("peel(20)");
+            wal.peel(e);
+        }
+    }
 }
diff --git a/src/lib.rs b/src/lib.rs
index 5dcb78a..a2b88ad 100644
--- a/src/lib.rs
+++ b/src/lib.rs
@@ -44,13 +44,11 @@ impl PartialOrd for WALRingId {
 /// the state for a WAL writer
 struct WALState {
     /// the first file id of WAL
-    pub first_fid: WALFileId,
+    first_fid: WALFileId,
     /// the next position for a record, addressed in the entire WAL space
-    pub next: WALPos,
-    /// number of bits for a block
-    pub block_nbit: u64,
+    next: WALPos,
     /// number of bits for a file
-    pub file_nbit: u64,
+    file_nbit: u64,
 }
 
 pub trait WALFile {
@@ -68,7 +66,7 @@ pub trait WALStore {
     /// Open a file given the filename, create the file if not exists when `touch` is `true`.
     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;
+    fn remove_file(&self, filename: &str) -> Result<(), ()>;
     /// Enumerate all WAL files.
     fn enumerate_files(&self) -> Box<[String]>;
     /// Apply (redo) the payload during recovery.
@@ -142,7 +140,7 @@ impl<F: WALStore> WALFilePool<F> {
         }
     }
 
-    fn remove_file(&self, fid: u64) -> bool {
+    fn remove_file(&self, fid: u64) -> Result<(), ()> {
         self.store.remove_file(&Self::get_fname(fid))
     }
 
@@ -258,6 +256,8 @@ impl<F: WALStore> WALWriter<F> {
     /// 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.
     pub fn peel<T: AsRef<[WALRingId]>>(&mut self, records: T) {
+        let msize = std::mem::size_of::<WALRingBlob>() as u64;
+        let block_size = self.block_size as u64;
         for rec in records.as_ref() {
             self.io_complete.push(*rec)
         }
@@ -266,12 +266,16 @@ impl<F: WALStore> WALWriter<F> {
             if s != self.next_complete {
                 break
             }
-            let m = self.io_complete.pop().unwrap();
+            let mut m = self.io_complete.pop().unwrap();
+            let block_remain = block_size - (m.end & (block_size - 1));
+            if block_remain <= msize as u64 {
+                m.end += block_remain
+            }
             self.next_complete = m.end
         }
         let next_fid = self.next_complete >> self.state.file_nbit;
         for fid in orig_fid..next_fid {
-            self.file_pool.remove_file(fid);
+            self.file_pool.remove_file(fid).unwrap();
         }
         self.state.first_fid = next_fid;
     }
@@ -339,13 +343,12 @@ impl<F: WALStore> WALLoader<F> {
                 }
             }
             f.truncate(0).unwrap();
-            self.file_pool.remove_file(fid);
+            self.file_pool.remove_file(fid).unwrap();
         }
         self.file_pool.reset();
         WALWriter::new(WALState {
             first_fid: 0,
             next: 0,
-            block_nbit: self.file_pool.block_nbit,
             file_nbit: self.file_pool.file_nbit,
         }, self.file_pool)
     }