Implement async API for random-access-storage (#26)

This commit moves the API from a sync call into async functions using async_trait.

We would like to use async operations on hypercore, and it would be nice to have the underlying storage using async-io when possible.
As discussed on datrs/hypercore#97, we could change the API to use async functions on the 'ram' trait.

Async Trait methods are still unstable, but possible using the 'async-trait' crate through a macro.

This PR moves the API to use async-fns. Tests are passing.

Author: bltavares

Cargo.toml

@@ -11,8 +11,11 @@ edition = "2018"
 
 [dependencies]
 anyhow = "1.0.26"
-random-access-storage = "3.0.0"
+random-access-storage = "4.0.0"
+futures = "0.3.4"
+async-trait = "0.1.24"
 
 [dev-dependencies]
 quickcheck = "0.9.2"
 rand = "0.7.3"
+async-std = { version = "1.5.0", features = ["attributes"] }

README.md

@@ -9,12 +9,12 @@ Continuously read,write to memory using random offsets and lengths.
 
 ## Usage
 ```rust
-extern crate random_access_memory as ram;
+use random_access_memory as ram;
 
 let mut file = ram::Sync::default();
-file.write(0, b"hello").unwrap();
-file.write(5, b" world").unwrap();
-let text = file.read(0, 11).unwrap();
+file.write(0, b"hello").await.unwrap();
+file.write(5, b" world").await.unwrap();
+let text = file.read(0, 11).await.unwrap();
 assert_eq!(text, b"hello world");
 ```
 

benches/sync.rs

@@ -1,39 +1,46 @@
 #![feature(test)]
 
 mod sync {
-  extern crate random_access_memory as ram;
-  extern crate random_access_storage;
   extern crate test;
 
-  use self::random_access_storage::RandomAccess;
-  use self::test::Bencher;
+  use random_access_memory as ram;
+  use random_access_storage::RandomAccess;
+  use test::Bencher;
 
   #[bench]
   fn write_hello_world(b: &mut Bencher) {
     b.iter(|| {
-      let mut file = ram::RandomAccessMemory::default();
-      file.write(0, b"hello").unwrap();
-      file.write(5, b" world").unwrap();
+      async_std::task::block_on(async {
+        let mut file = ram::RandomAccessMemory::default();
+        file.write(0, b"hello").await.unwrap();
+        file.write(5, b" world").await.unwrap();
+      })
     });
   }
 
   #[bench]
   fn read_hello_world(b: &mut Bencher) {
-    let mut file = ram::RandomAccessMemory::default();
-    file.write(0, b"hello").unwrap();
-    file.write(5, b" world").unwrap();
-    b.iter(|| {
-      let _text = file.read(0, 11).unwrap();
+    async_std::task::block_on(async {
+      let mut file = ram::RandomAccessMemory::default();
+      file.write(0, b"hello").await.unwrap();
+      file.write(5, b" world").await.unwrap();
+      b.iter(|| {
+        async_std::task::block_on(async {
+          let _text = file.read(0, 11).await.unwrap();
+        })
+      });
     });
   }
 
   #[bench]
   fn read_write_hello_world(b: &mut Bencher) {
     b.iter(|| {
-      let mut file = ram::RandomAccessMemory::default();
-      file.write(0, b"hello").unwrap();
-      file.write(5, b" world").unwrap();
-      let _text = file.read(0, 11).unwrap();
+      async_std::task::block_on(async {
+        let mut file = ram::RandomAccessMemory::default();
+        file.write(0, b"hello").await.unwrap();
+        file.write(5, b" world").await.unwrap();
+        let _text = file.read(0, 11).await.unwrap();
+      })
     });
   }
 }

src/lib.rs

@@ -6,7 +6,6 @@
 use anyhow::anyhow;
 use random_access_storage::RandomAccess;
 use std::cmp;
-use std::io;
 
 /// Main constructor.
 #[derive(Debug)]
@@ -52,10 +51,15 @@ impl RandomAccessMemory {
   }
 }
 
+#[async_trait::async_trait]
 impl RandomAccess for RandomAccessMemory {
   type Error = Box<dyn std::error::Error + Send + Sync>;
 
-  fn write(&mut self, offset: u64, data: &[u8]) -> Result<(), Self::Error> {
+  async fn write(
+    &mut self,
+    offset: u64,
+    data: &[u8],
+  ) -> Result<(), Self::Error> {
     let new_len = offset + data.len() as u64;
     if new_len > self.length {
       self.length = new_len;
@@ -101,11 +105,15 @@ impl RandomAccess for RandomAccessMemory {
     Ok(())
   }
 
-  fn sync_all(&mut self) -> Result<(), Self::Error> {
+  async fn sync_all(&mut self) -> Result<(), Self::Error> {
     Ok(())
   }
 
-  fn read(&mut self, offset: u64, length: u64) -> Result<Vec<u8>, Self::Error> {
+  async fn read(
+    &mut self,
+    offset: u64,
+    length: u64,
+  ) -> Result<Vec<u8>, Self::Error> {
     if (offset + length) > self.length {
       return Err(
         anyhow!(
@@ -156,16 +164,16 @@ impl RandomAccess for RandomAccessMemory {
     Ok(res_buf)
   }
 
-  fn read_to_writer(
+  async fn read_to_writer(
     &mut self,
     _offset: u64,
     _length: u64,
-    _buf: &mut impl io::Write,
+    _buf: &mut (impl futures::io::AsyncWrite + Send),
   ) -> Result<(), Self::Error> {
     unimplemented!()
   }
 
-  fn del(&mut self, offset: u64, length: u64) -> Result<(), Self::Error> {
+  async fn del(&mut self, offset: u64, length: u64) -> Result<(), Self::Error> {
     let overflow = offset % self.page_size as u64;
     let inc = match overflow {
       0 => 0,
@@ -190,15 +198,15 @@ impl RandomAccess for RandomAccessMemory {
     Ok(())
   }
 
-  fn truncate(&mut self, _length: u64) -> Result<(), Self::Error> {
+  async fn truncate(&mut self, _length: u64) -> Result<(), Self::Error> {
     unimplemented!()
   }
 
-  fn len(&self) -> Result<u64, Self::Error> {
+  async fn len(&self) -> Result<u64, Self::Error> {
     Ok(self.length)
   }
 
-  fn is_empty(&mut self) -> Result<bool, Self::Error> {
+  async fn is_empty(&mut self) -> Result<bool, Self::Error> {
     Ok(self.length == 0)
   }
 }

tests/model.rs

@@ -32,32 +32,34 @@ impl Arbitrary for Op {
 
 quickcheck! {
   fn implementation_matches_model(ops: Vec<Op>) -> bool {
-    let mut implementation = ram::RandomAccessMemory::new(10);
-    let mut model = vec![];
+    async_std::task::block_on(async {
+      let mut implementation = ram::RandomAccessMemory::new(10);
+      let mut model = vec![];
 
-    for op in ops {
-      match op {
-        Read { offset, length } => {
-          let end = offset + length;
-          if model.len() >= end as usize {
-            assert_eq!(
-              &*implementation.read(offset, length).expect("Reads should be successful."),
-              &model[offset as usize..end as usize]
-            );
-          } else {
-            assert!(implementation.read(offset, length).is_err());
-          }
-        },
-        Write { offset, ref data } => {
-          let end = offset + data.len() as u64;
-          if model.len() < end as usize {
-            model.resize(end as usize, 0);
-          }
-          implementation.write(offset, &*data).expect("Writes should be successful.");
-          model[offset as usize..end as usize].copy_from_slice(data);
-        },
+      for op in ops {
+        match op {
+          Read { offset, length } => {
+            let end = offset + length;
+            if model.len() >= end as usize {
+              assert_eq!(
+                &*implementation.read(offset, length).await.expect("Reads should be successful."),
+                &model[offset as usize..end as usize]
+              );
+            } else {
+              assert!(implementation.read(offset, length).await.is_err());
+            }
+          },
+          Write { offset, ref data } => {
+            let end = offset + data.len() as u64;
+            if model.len() < end as usize {
+              model.resize(end as usize, 0);
+            }
+            implementation.write(offset, &*data).await.expect("Writes should be successful.");
+            model[offset as usize..end as usize].copy_from_slice(data);
+          },
+        }
       }
-    }
-    true
+      true
+    })
   }
 }

tests/regression.rs

@@ -1,47 +1,47 @@
 use random_access_memory as ram;
 use random_access_storage::RandomAccess;
 
-#[test]
+#[async_std::test]
 // Postmortem: looks like we were reading out of bounds by accidentally
 // adding an offset to the index while reading.
-fn regress_1() {
+async fn regress_1() {
   let mut file = ram::RandomAccessMemory::new(50);
-  file.write(30, &[30]).unwrap();
-  file.read(15, 15).unwrap();
+  file.write(30, &[30]).await.unwrap();
+  file.read(15, 15).await.unwrap();
 }
 
-#[test]
+#[async_std::test]
 // Postmortem: our buffers weren't zero-filled. Intead we were relying on
 // uninitialized (but claimed!) memory, which caused all sorts of weirdness.
-fn regress_2() {
+async fn regress_2() {
   let mut file = ram::RandomAccessMemory::new(50);
-  file.write(22, &[22, 22, 22, 22]).unwrap();
-  let buf = file.read(1, 4).unwrap();
+  file.write(22, &[22, 22, 22, 22]).await.unwrap();
+  let buf = file.read(1, 4).await.unwrap();
   assert_eq!(buf, vec![0, 0, 0, 0]);
 
   let mut file = ram::RandomAccessMemory::new(50);
-  file.write(48, &[48, 48, 48, 48]).unwrap();
-  let buf = file.read(39, 9).unwrap();
+  file.write(48, &[48, 48, 48, 48]).await.unwrap();
+  let buf = file.read(39, 9).await.unwrap();
   assert_eq!(buf, vec![0, 0, 0, 0, 0, 0, 0, 0, 0]);
 }
 
-#[test]
+#[async_std::test]
 // Postmortem: the way we were reading was off. We were messing up both reading
 // and writing. We now keep two cursors, and compute the bounds of every loop
 // ahead of time. Also simplified our allocation logic.
-fn regress_3() {
+async fn regress_3() {
   let mut file = ram::RandomAccessMemory::new(50);
-  file.write(45, &[56, 46, 14, 93, 15, 54, 2]).unwrap();
-  let buf = file.read(42, 10).unwrap();
+  file.write(45, &[56, 46, 14, 93, 15, 54, 2]).await.unwrap();
+  let buf = file.read(42, 10).await.unwrap();
   assert_eq!(buf, vec![0, 0, 0, 56, 46, 14, 93, 15, 54, 2]);
 }
 
-#[test]
+#[async_std::test]
 // Postmortem: we were having trouble when we were reading with an index that's
 // larger than the page size. Turned out we weren't doing some math properly,
 // which caused a cursor to jump.
-fn regress_4() {
+async fn regress_4() {
   let mut file = ram::RandomAccessMemory::new(10);
-  file.write(44, &[54, 59]).unwrap();
-  file.read(13, 3).unwrap();
+  file.write(44, &[54, 59]).await.unwrap();
+  file.read(13, 3).await.unwrap();
 }

tests/test.rs

@@ -1,48 +1,48 @@
 use random_access_memory as ram;
 use random_access_storage::RandomAccess;
 
-#[test]
-fn can_call_new() {
+#[async_std::test]
+async fn can_call_new() {
   let _file = ram::RandomAccessMemory::default();
 }
 
-#[test]
-fn can_open_buffer() {
+#[async_std::test]
+async fn can_open_buffer() {
   let mut file = ram::RandomAccessMemory::default();
-  file.write(0, b"hello").unwrap();
+  file.write(0, b"hello").await.unwrap();
 }
 
-#[test]
-fn can_write() {
+#[async_std::test]
+async fn can_write() {
   let mut file = ram::RandomAccessMemory::default();
-  file.write(0, b"hello").unwrap();
-  file.write(5, b" world").unwrap();
+  file.write(0, b"hello").await.unwrap();
+  file.write(5, b" world").await.unwrap();
 }
 
-#[test]
-fn can_read() {
+#[async_std::test]
+async fn can_read() {
   let mut file = ram::RandomAccessMemory::default();
-  file.write(0, b"hello").unwrap();
-  file.write(5, b" world").unwrap();
-  let text = file.read(0, 11).unwrap();
+  file.write(0, b"hello").await.unwrap();
+  file.write(5, b" world").await.unwrap();
+  let text = file.read(0, 11).await.unwrap();
   let text = String::from_utf8(text.to_vec()).unwrap();
   assert_eq!(text, "hello world");
 }
 
-#[test]
-fn can_len() {
+#[async_std::test]
+async fn can_len() {
   let mut file = ram::RandomAccessMemory::default();
-  assert_eq!(file.len().unwrap(), 0);
-  file.write(0, b"hello").unwrap();
-  assert_eq!(file.len().unwrap(), 5);
-  file.write(5, b" world").unwrap();
-  assert_eq!(file.len().unwrap(), 11);
+  assert_eq!(file.len().await.unwrap(), 0);
+  file.write(0, b"hello").await.unwrap();
+  assert_eq!(file.len().await.unwrap(), 5);
+  file.write(5, b" world").await.unwrap();
+  assert_eq!(file.len().await.unwrap(), 11);
 }
 
-#[test]
-fn can_is_empty() {
+#[async_std::test]
+async fn can_is_empty() {
   let mut file = ram::RandomAccessMemory::default();
-  assert_eq!(file.is_empty().unwrap(), true);
-  file.write(0, b"hello").unwrap();
-  assert_eq!(file.is_empty().unwrap(), false);
+  assert_eq!(file.is_empty().await.unwrap(), true);
+  file.write(0, b"hello").await.unwrap();
+  assert_eq!(file.is_empty().await.unwrap(), false);
 }