add assertion for empty record payload; make `load` async funcv0.1.6

6 files changed, 155 insertions, 140 deletions

diff --git a/Cargo.lock b/Cargo.lock
index 4a278a9..88ae519 100644
--- a/Cargo.lock
+++ b/Cargo.lock
@@ -2,12 +2,9 @@
 # It is not intended for manual editing.
 [[package]]
 name = "ahash"
-version = "0.2.18"
+version = "0.3.8"
 source = "registry+https://github.com/rust-lang/crates.io-index"
-checksum = "6f33b5018f120946c1dcf279194f238a9f146725593ead1c08fa47ff22b0b5d3"
-dependencies = [
- "const-random",
-]
+checksum = "e8fd72866655d1904d6b0997d0b07ba561047d070fbe29de039031c641b61217"
 
 [[package]]
 name = "aho-corasick"
@@ -20,9 +17,9 @@ dependencies = [
 
 [[package]]
 name = "async-trait"
-version = "0.1.35"
+version = "0.1.41"
 source = "registry+https://github.com/rust-lang/crates.io-index"
-checksum = "89cb5d814ab2a47fd66d3266e9efccb53ca4c740b7451043b8ffcf9a6208f3f8"
+checksum = "b246867b8b3b6ae56035f1eb1ed557c1d8eae97f0d53696138a50fa0e3a3b8c0"
 dependencies = [
  "proc-macro2",
  "quote",
@@ -31,12 +28,6 @@ dependencies = [
 
 [[package]]
 name = "autocfg"
-version = "0.1.7"
-source = "registry+https://github.com/rust-lang/crates.io-index"
-checksum = "1d49d90015b3c36167a20fe2810c5cd875ad504b39cff3d4eae7977e6b7c1cb2"
-
-[[package]]
-name = "autocfg"
 version = "1.0.0"
 source = "registry+https://github.com/rust-lang/crates.io-index"
 checksum = "f8aac770f1885fd7e387acedd76065302551364496e46b3dd00860b2f8359b9d"
@@ -75,26 +66,6 @@ dependencies = [
 ]
 
 [[package]]
-name = "const-random"
-version = "0.1.8"
-source = "registry+https://github.com/rust-lang/crates.io-index"
-checksum = "2f1af9ac737b2dd2d577701e59fd09ba34822f6f2ebdb30a7647405d9e55e16a"
-dependencies = [
- "const-random-macro",
- "proc-macro-hack",
-]
-
-[[package]]
-name = "const-random-macro"
-version = "0.1.8"
-source = "registry+https://github.com/rust-lang/crates.io-index"
-checksum = "25e4c606eb459dd29f7c57b2e0879f2b6f14ee130918c2b78ccb58a9624e6c7a"
-dependencies = [
- "getrandom",
- "proc-macro-hack",
-]
-
-[[package]]
 name = "crc"
 version = "1.8.1"
 source = "registry+https://github.com/rust-lang/crates.io-index"
@@ -119,16 +90,16 @@ version = "0.7.2"
 source = "registry+https://github.com/rust-lang/crates.io-index"
 checksum = "c3c7c73a2d1e9fc0886a08b93e98eb643461230d5f1925e4036204d5f2e261a8"
 dependencies = [
- "autocfg 1.0.0",
+ "autocfg",
  "cfg-if",
  "lazy_static",
 ]
 
 [[package]]
 name = "futures"
-version = "0.3.5"
+version = "0.3.6"
 source = "registry+https://github.com/rust-lang/crates.io-index"
-checksum = "1e05b85ec287aac0dc34db7d4a569323df697f9c55b99b15d6b4ef8cde49f613"
+checksum = "5d8e3078b7b2a8a671cb7a3d17b4760e4181ea243227776ba83fd043b4ca034e"
 dependencies = [
  "futures-channel",
  "futures-core",
@@ -141,9 +112,9 @@ dependencies = [
 
 [[package]]
 name = "futures-channel"
-version = "0.3.5"
+version = "0.3.6"
 source = "registry+https://github.com/rust-lang/crates.io-index"
-checksum = "f366ad74c28cca6ba456d95e6422883cfb4b252a83bed929c83abfdbbf2967d5"
+checksum = "a7a4d35f7401e948629c9c3d6638fb9bf94e0b2121e96c3b428cc4e631f3eb74"
 dependencies = [
  "futures-core",
  "futures-sink",
@@ -151,15 +122,15 @@ dependencies = [
 
 [[package]]
 name = "futures-core"
-version = "0.3.5"
+version = "0.3.6"
 source = "registry+https://github.com/rust-lang/crates.io-index"
-checksum = "59f5fff90fd5d971f936ad674802482ba441b6f09ba5e15fd8b39145582ca399"
+checksum = "d674eaa0056896d5ada519900dbf97ead2e46a7b6621e8160d79e2f2e1e2784b"
 
 [[package]]
 name = "futures-executor"
-version = "0.3.5"
+version = "0.3.6"
 source = "registry+https://github.com/rust-lang/crates.io-index"
-checksum = "10d6bb888be1153d3abeb9006b11b02cf5e9b209fda28693c31ae1e4e012e314"
+checksum = "cc709ca1da6f66143b8c9bec8e6260181869893714e9b5a490b169b0414144ab"
 dependencies = [
  "futures-core",
  "futures-task",
@@ -168,15 +139,15 @@ dependencies = [
 
 [[package]]
 name = "futures-io"
-version = "0.3.5"
+version = "0.3.6"
 source = "registry+https://github.com/rust-lang/crates.io-index"
-checksum = "de27142b013a8e869c14957e6d2edeef89e97c289e69d042ee3a49acd8b51789"
+checksum = "5fc94b64bb39543b4e432f1790b6bf18e3ee3b74653c5449f63310e9a74b123c"
 
 [[package]]
 name = "futures-macro"
-version = "0.3.5"
+version = "0.3.6"
 source = "registry+https://github.com/rust-lang/crates.io-index"
-checksum = "d0b5a30a4328ab5473878237c447333c093297bded83a4983d10f4deea240d39"
+checksum = "f57ed14da4603b2554682e9f2ff3c65d7567b53188db96cb71538217fc64581b"
 dependencies = [
  "proc-macro-hack",
  "proc-macro2",
@@ -186,24 +157,24 @@ dependencies = [
 
 [[package]]
 name = "futures-sink"
-version = "0.3.5"
+version = "0.3.6"
 source = "registry+https://github.com/rust-lang/crates.io-index"
-checksum = "3f2032893cb734c7a05d85ce0cc8b8c4075278e93b24b66f9de99d6eb0fa8acc"
+checksum = "0d8764258ed64ebc5d9ed185cf86a95db5cac810269c5d20ececb32e0088abbd"
 
 [[package]]
 name = "futures-task"
-version = "0.3.5"
+version = "0.3.6"
 source = "registry+https://github.com/rust-lang/crates.io-index"
-checksum = "bdb66b5f09e22019b1ab0830f7785bcea8e7a42148683f99214f73f8ec21a626"
+checksum = "4dd26820a9f3637f1302da8bceba3ff33adbe53464b54ca24d4e2d4f1db30f94"
 dependencies = [
  "once_cell",
 ]
 
 [[package]]
 name = "futures-util"
-version = "0.3.5"
+version = "0.3.6"
 source = "registry+https://github.com/rust-lang/crates.io-index"
-checksum = "8764574ff08b701a084482c3c7031349104b07ac897393010494beaa18ce32c6"
+checksum = "8a894a0acddba51a2d49a6f4263b1e64b8c579ece8af50fa86503d52cd1eea34"
 dependencies = [
  "futures-channel",
  "futures-core",
@@ -232,7 +203,7 @@ dependencies = [
 
 [[package]]
 name = "growth-ring"
-version = "0.1.5"
+version = "0.1.6"
 dependencies = [
  "async-trait",
  "crc",
@@ -250,15 +221,21 @@ dependencies = [
 
 [[package]]
 name = "hashbrown"
-version = "0.6.3"
+version = "0.8.2"
 source = "registry+https://github.com/rust-lang/crates.io-index"
-checksum = "8e6073d0ca812575946eb5f35ff68dbe519907b25c42530389ff946dc84c6ead"
+checksum = "e91b62f79061a0bc2e046024cb7ba44b08419ed238ecbd9adbd787434b9e8c25"
 dependencies = [
  "ahash",
- "autocfg 0.1.7",
+ "autocfg",
 ]
 
 [[package]]
+name = "hashbrown"
+version = "0.9.1"
+source = "registry+https://github.com/rust-lang/crates.io-index"
+checksum = "d7afe4a420e3fe79967a00898cc1f4db7c8a49a9333a29f8a4bd76a253d5cd04"
+
+[[package]]
 name = "hex"
 version = "0.4.2"
 source = "registry+https://github.com/rust-lang/crates.io-index"
@@ -266,11 +243,12 @@ checksum = "644f9158b2f133fd50f5fb3242878846d9eb792e445c893805ff0e3824006e35"
 
 [[package]]
 name = "indexmap"
-version = "1.4.0"
+version = "1.6.0"
 source = "registry+https://github.com/rust-lang/crates.io-index"
-checksum = "c398b2b113b55809ceb9ee3e753fcbac793f1956663f3c36549c1346015c2afe"
+checksum = "55e2e4c765aa53a0424761bf9f41aa7a6ac1efa87238f59560640e27fca028f2"
 dependencies = [
- "autocfg 1.0.0",
+ "autocfg",
+ "hashbrown 0.9.1",
 ]
 
 [[package]]
@@ -292,9 +270,9 @@ dependencies = [
 
 [[package]]
 name = "libc"
-version = "0.2.71"
+version = "0.2.79"
 source = "registry+https://github.com/rust-lang/crates.io-index"
-checksum = "9457b06509d27052635f90d6466700c65095fdf75409b3fbdd903e988b886f49"
+checksum = "2448f6066e80e3bfc792e9c98bf705b4b0fc6e8ef5b43e5889aff0eaa9c58743"
 
 [[package]]
 name = "lock_api"
@@ -307,11 +285,11 @@ dependencies = [
 
 [[package]]
 name = "lru"
-version = "0.5.2"
+version = "0.6.0"
 source = "registry+https://github.com/rust-lang/crates.io-index"
-checksum = "297efb9401445cf7b6986a583d7ac194023334b46b294ff7da0d36662c1251c2"
+checksum = "111b945ac72ec09eb7bc62a0fbdc3cc6e80555a7245f52a69d3921a75b53b153"
 dependencies = [
- "hashbrown",
+ "hashbrown 0.8.2",
 ]
 
 [[package]]
@@ -328,15 +306,14 @@ checksum = "3728d817d99e5ac407411fa471ff9800a778d88a24685968b36824eaf4bee400"
 
 [[package]]
 name = "nix"
-version = "0.17.0"
+version = "0.18.0"
 source = "registry+https://github.com/rust-lang/crates.io-index"
-checksum = "50e4785f2c3b7589a0d0c1dd60285e1188adac4006e8abd6dd578e1567027363"
+checksum = "83450fe6a6142ddd95fb064b746083fc4ef1705fe81f64a64e1d4b39f54a1055"
 dependencies = [
  "bitflags",
  "cc",
  "cfg-if",
  "libc",
- "void",
 ]
 
 [[package]]
@@ -480,9 +457,9 @@ checksum = "2439c63f3f6139d1b57529d16bc3b8bb855230c8efcc5d3a896c8bea7c3b1e84"
 
 [[package]]
 name = "regex"
-version = "1.3.9"
+version = "1.4.0"
 source = "registry+https://github.com/rust-lang/crates.io-index"
-checksum = "9c3780fcf44b193bc4d09f36d2a3c87b251da4a046c87795a0d35f4f927ad8e6"
+checksum = "36f45b719a674bf4b828ff318906d6c133264c793eff7a41e30074a45b5099e2"
 dependencies = [
  "aho-corasick",
  "memchr",
@@ -492,9 +469,9 @@ dependencies = [
 
 [[package]]
 name = "regex-syntax"
-version = "0.6.18"
+version = "0.6.19"
 source = "registry+https://github.com/rust-lang/crates.io-index"
-checksum = "26412eb97c6b088a6997e05f69403a802a92d520de2f8e63c2b65f9e0f47c4e8"
+checksum = "c17be88d9eaa858870aa5e48cc406c206e4600e983fc4f06bbe5750d93d09761"
 
 [[package]]
 name = "scan_fmt"
@@ -550,12 +527,6 @@ source = "registry+https://github.com/rust-lang/crates.io-index"
 checksum = "826e7639553986605ec5979c7dd957c7895e93eabed50ab2ffa7f6128a75097c"
 
 [[package]]
-name = "void"
-version = "1.0.2"
-source = "registry+https://github.com/rust-lang/crates.io-index"
-checksum = "6a02e4885ed3bc0f2de90ea6dd45ebcbb66dacffe03547fadbb0eeae2770887d"
-
-[[package]]
 name = "wasi"
 version = "0.9.0+wasi-snapshot-preview1"
 source = "registry+https://github.com/rust-lang/crates.io-index"
diff --git a/Cargo.toml b/Cargo.toml
index 22b3232..d5fa31c 100644
--- a/Cargo.toml
+++ b/Cargo.toml
@@ -1,6 +1,6 @@
 [package]
 name = "growth-ring"
-version = "0.1.5"
+version = "0.1.6"
 authors = ["Determinant <[email protected]>"]
 edition = "2018"
 homepage = "https://github.com/Determinant/growth-ring"
@@ -12,19 +12,19 @@ description = "Simple and modular write-ahead-logging implementation."
 
 [dependencies]
 crc = "1.8.1"
-lru = "0.5.2"
+lru = "0.6.0"
 scan_fmt = "0.2.5"
-regex = "1.3.9"
-async-trait = "0.1.35"
-futures = "0.3.5"
+regex = "1.4.0"
+async-trait = "0.1.41"
+futures = "0.3.6"
 libaio-futures = "0.1.7"
-nix = "0.17.0"
-libc = "0.2.71"
+nix = "0.18.0"
+libc = "0.2.79"
 
 [dev-dependencies]
 hex = "0.4.2"
 rand = "0.7.3"
-indexmap = "1.4.0"
+indexmap = "1.6.0"
 
 [lib]
 name = "growthring"
diff --git a/examples/demo1.rs b/examples/demo1.rs
index 6f029ab..31a9b7d 100644
--- a/examples/demo1.rs
+++ b/examples/demo1.rs
@@ -1,3 +1,4 @@
+use futures::executor::block_on;
 use growthring::{
     wal::{WALBytes, WALRingId, WALLoader, WALWriter},
     WALStoreAIO,
@@ -32,8 +33,8 @@ fn main() {
     let mut loader = WALLoader::new();
     loader.file_nbit(9).block_nbit(8);
 
-    let store = WALStoreAIO::new(&wal_dir, true, None).unwrap();
-    let mut wal = loader.load(store, recover).unwrap();
+    let store = WALStoreAIO::new(&wal_dir, true, None, None).unwrap();
+    let mut wal = block_on(loader.load(store, recover)).unwrap();
     for _ in 0..3 {
         test(
             ["hi", "hello", "lol"]
@@ -50,8 +51,8 @@ fn main() {
         );
     }
 
-    let store = WALStoreAIO::new(&wal_dir, false, None).unwrap();
-    let mut wal = loader.load(store, recover).unwrap();
+    let store = WALStoreAIO::new(&wal_dir, false, None, None).unwrap();
+    let mut wal = block_on(loader.load(store, recover)).unwrap();
     for _ in 0..3 {
         test(
             vec![
@@ -64,8 +65,8 @@ fn main() {
         );
     }
 
-    let store = WALStoreAIO::new(&wal_dir, false, None).unwrap();
-    let mut wal = loader.load(store, recover).unwrap();
+    let store = WALStoreAIO::new(&wal_dir, false, None, None).unwrap();
+    let mut wal = block_on(loader.load(store, recover)).unwrap();
     for _ in 0..3 {
         let mut ids = Vec::new();
         for _ in 0..3 {
diff --git a/src/lib.rs b/src/lib.rs
index 2a67cd3..7674adc 100644
--- a/src/lib.rs
+++ b/src/lib.rs
@@ -10,8 +10,8 @@
 //!
 //!
 //! // Start with empty WAL (truncate = true).
-//! let store = WALStoreAIO::new("./walfiles", true, None).unwrap();
-//! let mut wal = loader.load(store, |_, _| {Ok(())}).unwrap();
+//! let store = WALStoreAIO::new("./walfiles", true, None, None).unwrap();
+//! let mut wal = block_on(loader.load(store, |_, _| {Ok(())})).unwrap();
 //! // Write a vector of records to WAL.
 //! for f in wal.grow(vec!["record1(foo)", "record2(bar)", "record3(foobar)"]).into_iter() {
 //!     let ring_id = block_on(f).unwrap().1;
@@ -20,28 +20,28 @@
 //!
 //!
 //! // Load from WAL (truncate = false).
-//! let store = WALStoreAIO::new("./walfiles", false, None).unwrap();
-//! let mut wal = loader.load(store, |payload, ringid| {
+//! let store = WALStoreAIO::new("./walfiles", false, None, None).unwrap();
+//! let mut wal = block_on(loader.load(store, |payload, ringid| {
 //!     // redo the operations in your application
 //!     println!("recover(payload={}, ringid={:?})",
 //!              std::str::from_utf8(&payload).unwrap(),
 //!              ringid);
 //!     Ok(())
-//! }).unwrap();
+//! })).unwrap();
 //! // We saw some log playback, even there is no failure.
 //! // Let's try to grow the WAL to create many files.
-//! let ring_ids = wal.grow((0..100).into_iter().map(|i| "a".repeat(i)).collect::<Vec<_>>())
+//! let ring_ids = wal.grow((1..100).into_iter().map(|i| "a".repeat(i)).collect::<Vec<_>>())
 //!                   .into_iter().map(|f| block_on(f).unwrap().1).collect::<Vec<_>>();
 //! // Then assume all these records are not longer needed. We can tell WALWriter by the `peel`
 //! // method.
 //! block_on(wal.peel(ring_ids)).unwrap();
 //! // There will only be one remaining file in ./walfiles.
 //!
-//! let store = WALStoreAIO::new("./walfiles", false, None).unwrap();
-//! let wal = loader.load(store, |payload, _| {
+//! let store = WALStoreAIO::new("./walfiles", false, None, None).unwrap();
+//! let wal = block_on(loader.load(store, |payload, _| {
 //!     println!("payload.len() = {}", payload.len());
 //!     Ok(())
-//! }).unwrap();
+//! })).unwrap();
 //! // After each recovery, the ./walfiles is empty.
 //! ```
 
@@ -49,7 +49,6 @@
 pub mod wal;
 
 use async_trait::async_trait;
-use futures::executor::block_on;
 use libaiofut::{AIOBuilder, AIOManager};
 use libc::off_t;
 use nix::fcntl::{fallocate, open, openat, FallocateFlags, OFlag};
@@ -116,13 +115,12 @@ impl WALFile for WALFileAIO {
         })
     }
 
-    fn read(
+    async fn read(
         &self,
         offset: WALPos,
         length: usize,
     ) -> Result<Option<WALBytes>, ()> {
-        let (res, data) =
-            block_on(self.aiomgr.read(self.fd, offset, length, None));
+        let (res, data) = self.aiomgr.read(self.fd, offset, length, None).await;
         res.or_else(|_| Err(())).and_then(|nread| {
             Ok(if nread == length { Some(data) } else { None })
         })
@@ -131,7 +129,6 @@ impl WALFile for WALFileAIO {
 
 pub struct WALStoreAIO {
     rootfd: RawFd,
-    rootpath: String,
     aiomgr: Rc<AIOManager>,
 }
 
@@ -139,9 +136,9 @@ impl WALStoreAIO {
     pub fn new(
         wal_dir: &str,
         truncate: bool,
+        rootfd: Option<RawFd>,
         aiomgr: Option<AIOManager>,
     ) -> Result<Self, ()> {
-        let rootpath = wal_dir.to_string();
         let aiomgr = Rc::new(aiomgr.ok_or(Err(())).or_else(
             |_: Result<AIOManager, ()>| {
                 AIOBuilder::default().build().or(Err(()))
@@ -151,25 +148,55 @@ impl WALStoreAIO {
         if truncate {
             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)
+        let walfd;
+        match rootfd {
+            None => {
+                match mkdir(
+                    wal_dir,
+                    Mode::S_IRUSR | Mode::S_IWUSR | Mode::S_IXUSR,
+                ) {
+                    Err(e) => {
+                        if truncate {
+                            panic!("error while creating directory: {}", e)
+                        }
+                    }
+                    Ok(_) => (),
+                }
+                walfd = match open(
+                    wal_dir,
+                    OFlag::O_DIRECTORY | OFlag::O_PATH,
+                    Mode::empty(),
+                ) {
+                    Ok(fd) => fd,
+                    Err(_) => panic!("error while opening the WAL directory"),
+                }
+            }
+            Some(fd) => {
+                let ret = unsafe {
+                    libc::mkdirat(
+                        fd,
+                        std::ffi::CString::new(wal_dir).unwrap().as_ptr(),
+                        libc::S_IRUSR | libc::S_IWUSR | libc::S_IXUSR,
+                    )
+                };
+                if ret != 0 {
+                    if truncate {
+                        panic!("error while creating directory")
+                    }
+                }
+                walfd = match nix::fcntl::openat(
+                    fd,
+                    wal_dir,
+                    OFlag::O_DIRECTORY | OFlag::O_PATH,
+                    Mode::empty(),
+                ) {
+                    Ok(fd) => fd,
+                    Err(_) => panic!("error while opening the WAL directory"),
                 }
             }
-            Ok(_) => (),
         }
-        let rootfd = match open(
-            wal_dir,
-            OFlag::O_DIRECTORY | OFlag::O_PATH,
-            Mode::empty(),
-        ) {
-            Ok(fd) => fd,
-            Err(_) => panic!("error while opening the WAL directory"),
-        };
         Ok(WALStoreAIO {
-            rootfd,
-            rootpath,
+            rootfd: walfd,
             aiomgr,
         })
     }
@@ -200,8 +227,17 @@ impl WALStore for WALStoreAIO {
 
     fn enumerate_files(&self) -> Result<Self::FileNameIter, ()> {
         let mut logfiles = Vec::new();
-        for fname in std::fs::read_dir(&self.rootpath).unwrap() {
-            logfiles.push(fname.unwrap().file_name().into_string().unwrap())
+        for ent in nix::dir::Dir::openat(
+            self.rootfd,
+            "./",
+            OFlag::empty(),
+            Mode::empty(),
+        )
+        .unwrap()
+        .iter()
+        {
+            logfiles
+                .push(ent.unwrap().file_name().to_str().unwrap().to_string())
         }
         Ok(logfiles.into_iter())
     }
diff --git a/src/wal.rs b/src/wal.rs
index deae98d..7975f38 100644
--- a/src/wal.rs
+++ b/src/wal.rs
@@ -1,5 +1,4 @@
 use async_trait::async_trait;
-use futures::executor::block_on;
 use futures::future::{self, FutureExt, TryFutureExt};
 use std::cell::{RefCell, UnsafeCell};
 use std::collections::{hash_map, BinaryHeap, HashMap};
@@ -106,7 +105,7 @@ pub trait WALFile {
     /// all or nothing).
     async fn write(&self, offset: WALPos, data: WALBytes) -> Result<(), ()>;
     /// Read data with offset. Return `Ok(None)` when it reaches EOF.
-    fn read(
+    async fn read(
         &self,
         offset: WALPos,
         length: usize,
@@ -408,7 +407,8 @@ impl<F: WALStore> WALWriter<F> {
     /// 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.
+    /// `peel` with the given `WALRingId`s. Note: each serialized record should contain at least 1
+    /// byte (empty record payload will result in assertion failure).
     pub fn grow<'a, R: Record + 'a>(
         &'a mut self,
         records: Vec<R>,
@@ -427,6 +427,7 @@ impl<F: WALStore> WALWriter<F> {
             let mut rec = &bytes[..];
             let mut rsize = rec.len() as u32;
             let mut ring_start = None;
+            assert!(rsize > 0);
             while rsize > 0 {
                 let remain = self.block_size - bbuff_cur;
                 if remain > msize {
@@ -647,7 +648,10 @@ impl WALLoader {
     }
 
     /// Recover by reading the WAL files.
-    pub fn load<S: WALStore, F: FnMut(WALBytes, WALRingId) -> Result<(), ()>>(
+    pub async fn load<
+        S: WALStore,
+        F: FnMut(WALBytes, WALRingId) -> Result<(), ()>,
+    >(
         &self,
         store: S,
         mut recover_func: F,
@@ -674,14 +678,14 @@ impl WALLoader {
         let mut skip = false;
         for fname in logfiles.into_iter() {
             let fid = file_pool.get_fid(&fname);
-            let f = block_on(file_pool.get_file(fid, false))?;
+            let f = file_pool.get_file(fid, false).await?;
             let mut off = 0;
             if skip {
                 f.truncate(0)?;
-                block_on(file_pool.store.remove_file(fname))?;
+                file_pool.store.remove_file(fname).await?;
                 continue;
             }
-            while let Some(header_raw) = f.read(off, msize as usize)? {
+            while let Some(header_raw) = f.read(off, msize as usize).await? {
                 let ringid_start = (fid << file_pool.file_nbit) + off;
                 off += msize as u64;
                 let header = unsafe {
@@ -693,7 +697,8 @@ impl WALLoader {
                 match header.rtype {
                     WALRingType::Full => {
                         assert!(chunks.is_none());
-                        let payload = f.read(off, rsize as usize)?.ok_or(())?;
+                        let payload =
+                            f.read(off, rsize as usize).await?.ok_or(())?;
                         // TODO: improve the behavior when CRC32 fails
                         if !self.verify_checksum(&payload, header.crc32)? {
                             skip = true;
@@ -710,7 +715,8 @@ impl WALLoader {
                     }
                     WALRingType::First => {
                         assert!(chunks.is_none());
-                        let chunk = f.read(off, rsize as usize)?.ok_or(())?;
+                        let chunk =
+                            f.read(off, rsize as usize).await?.ok_or(())?;
                         if !self.verify_checksum(&chunk, header.crc32)? {
                             skip = true;
                             break;
@@ -721,7 +727,7 @@ impl WALLoader {
                     WALRingType::Middle => {
                         if let Some((chunks, _)) = &mut chunks {
                             let chunk =
-                                f.read(off, rsize as usize)?.ok_or(())?;
+                                f.read(off, rsize as usize).await?.ok_or(())?;
                             if !self.verify_checksum(&chunk, header.crc32)? {
                                 skip = true;
                                 break;
@@ -734,7 +740,7 @@ impl WALLoader {
                         if let Some((mut chunks, ringid_start)) = chunks.take()
                         {
                             let chunk =
-                                f.read(off, rsize as usize)?.ok_or(())?;
+                                f.read(off, rsize as usize).await?.ok_or(())?;
                             off += rsize as u64;
                             if !self.verify_checksum(&chunk, header.crc32)? {
                                 skip = true;
@@ -772,7 +778,7 @@ impl WALLoader {
                 }
             }
             f.truncate(0)?;
-            block_on(file_pool.store.remove_file(fname))?;
+            file_pool.store.remove_file(fname).await?;
         }
         file_pool.reset();
         Ok(WALWriter::new(
diff --git a/tests/common/mod.rs b/tests/common/mod.rs
index 9ec9e8e..fa6c343 100644
--- a/tests/common/mod.rs
+++ b/tests/common/mod.rs
@@ -1,6 +1,7 @@
 #[cfg(test)]
 #[allow(dead_code)]
 use async_trait::async_trait;
+use futures::executor::block_on;
 use growthring::wal::{
     WALBytes, WALFile, WALLoader, WALPos, WALRingId, WALStore,
 };
@@ -71,7 +72,7 @@ impl<G: FailGen> WALFile for WALFileEmul<G> {
         Ok(())
     }
 
-    fn read(
+    async fn read(
         &self,
         offset: WALPos,
         length: usize,
@@ -513,13 +514,13 @@ impl PaintingSim {
         let mut rng =
             <rand::rngs::StdRng as rand::SeedableRng>::seed_from_u64(self.seed);
         let mut wal =
-            loader.load(WALStoreEmul::new(state, fgen.clone()), |_, _| {
+            block_on(loader.load(WALStoreEmul::new(state, fgen.clone()), |_, _| {
                 if fgen.next_fail() {
                     Err(())
                 } else {
                     Ok(())
                 }
-            })?;
+            }))?;
         for _ in 0..self.n {
             let pss = (0..self.m)
                 .map(|_| {
@@ -616,7 +617,7 @@ impl PaintingSim {
         let mut last_idx = 0;
         let mut napplied = 0;
         canvas.clear_queued();
-        wal.load(
+        block_on(wal.load(
             WALStoreEmul::new(state, Rc::new(ZeroFailGen)),
             |payload, ringid| {
                 let s = PaintStrokes::from_bytes(&payload);
@@ -625,7 +626,7 @@ impl PaintingSim {
                 napplied += 1;
                 Ok(())
             },
-        )
+        ))
         .unwrap();
         println!("last = {}/{}, applied = {}", last_idx, ops.len(), napplied);
         canvas.paint_all();