...

1 files changed, 134 insertions, 45 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();
         }
     }
 }