[WIP] Implement bs_load.

.circleci/scripts/run_test.sh

@@ -2,7 +2,6 @@
 
 curl https://sh.rustup.rs -sSf | sh -s -- --default-toolchain none -y
 source $HOME/.cargo/env
-dd if=/dev/zero of=/tmp/aiofile bs=2048 count=5000
 export LD_LIBRARY_PATH="${LD_LIBRARY_PATH}:/usr/local/lib"
 cd /home/circleci/project
 cargo test --all

Cargo.toml

@@ -22,5 +22,6 @@ slog-term = "=2.0.2"
 members = [
   "spdk-sys",
   "starfish-executor",
+  "starfish-file",
   "starfish-example-app"
 ]

README.md

@@ -23,10 +23,11 @@ $ ./scripts/setup.sh
 # Used for aio backed testing
 $ dd if=/dev/zero of=/tmp/aiofile bs=2048 count=5000
 
-$ sudo ldconfig /usr/local/lib
-
 # Need to run dpdk applications as root :(
 $ cargo build && sudo target/debug/starfish-example-app starfish-example-app/config/hello_blob.conf
+
+# Running tests
+$ sudo bash scripts/run_tests.sh --all
 ```
 
 ## Example apps
@@ -35,6 +36,7 @@ $ cargo build && sudo target/debug/starfish-example-app starfish-example-app/con
 * https://github.com/percona/tokudb-engine/wiki/Write-optimized-fractal-tree-storage
 * https://github.com/Tokutek
 * https://www.percona.com/blog/wp-content/uploads/2011/11/how-fractal-trees-work.pdf
+* https://crates.io/crates/sonnerie
 
 ## Futures and async/await
 

scripts/run_tests.sh

@@ -0,0 +1,5 @@
+#!/bin/bash
+source $HOME/.cargo/env
+export LD_LIBRARY_PATH="${LD_LIBRARY_PATH}:/usr/local/lib"
+export RUST_BACKTRACE=1
+cargo test $@

spdk-sys/Cargo.toml

@@ -22,4 +22,6 @@ bindgen = "= 0.41.0"
 
 [dev-dependencies]
 hamcrest2 = "*"
+tempfile = "3.0.4"
+nix = "0.11.0"
 starfish-executor = { path = "../starfish-executor" }

spdk-sys/src/blob.rs

@@ -4,16 +4,18 @@ use crate::generated::{
     spdk_blob, spdk_blob_close, spdk_blob_get_id, spdk_blob_get_num_clusters, spdk_blob_id,
     spdk_blob_io_read, spdk_blob_io_write, spdk_blob_resize, spdk_blob_store, spdk_blob_sync_md,
     spdk_bs_alloc_io_channel, spdk_bs_create_blob, spdk_bs_delete_blob, spdk_bs_free_cluster_count,
-    spdk_bs_free_io_channel, spdk_bs_get_page_size, spdk_bs_init, spdk_bs_open_blob,
-    spdk_bs_unload, spdk_io_channel,
+    spdk_bs_free_io_channel, spdk_bs_get_page_size, spdk_bs_init, spdk_bs_load, spdk_bs_open_blob,
+    spdk_bs_opts, spdk_bs_opts_init, spdk_bs_unload, spdk_io_channel,
 };
 use failure::Error;
+use futures::channel::mpsc::UnboundedSender;
 use futures::channel::oneshot;
 use futures::channel::oneshot::Sender;
 use libc::c_int;
 use libc::c_void;
 use std::fmt;
 use std::fmt::Debug;
+use std::mem::forget;
 use std::ptr;
 
 #[derive(Debug, Fail)]
@@ -154,6 +156,72 @@ pub async fn bs_init(bs_dev: &mut BlobStoreBDev) -> Result<Blobstore, Error> {
     }
 }
 
+pub async fn bs_load(
+    bs_dev: &mut BlobStoreBDev,
+    blobs_sender: UnboundedSender<Blob>,
+) -> Result<Blobstore, Error> {
+    let (sender, receiver) = oneshot::channel();
+
+    let mut bs_opts = spdk_bs_opts::default();
+
+    struct Senders {
+        completion_sender: Sender<Result<*mut spdk_blob_store, i32>>,
+        blobs_sender: UnboundedSender<Blob>,
+    }
+
+    extern "C" fn complete_callback(senders_ptr: *mut c_void, bs: *mut spdk_blob_store, bserrno: c_int) {
+        let senders = unsafe { Box::from_raw(senders_ptr as *mut Senders) };
+        let ret = if bserrno != 0 { Err(bserrno) } else { Ok(bs) };
+        senders.blobs_sender.close_channel();
+        senders
+            .completion_sender
+            .send(ret)
+            .expect("Receiver is gone");
+    }
+
+    extern "C" fn iter_callback(senders_ptr: *mut c_void, blob: *mut spdk_blob, bserrno: c_int) {
+        let senders = unsafe { Box::from_raw(senders_ptr as *mut Senders) };
+        if bserrno != 0 {
+            senders.blobs_sender.close_channel();
+            senders
+                .completion_sender
+                .send(Err(bserrno))
+                .expect("Receiver is gone");
+        } else {
+            // TODO: cleanup here (deallocate blob and whatnot) on error
+            senders.blobs_sender.unbounded_send(Blob { blob }).expect("Could not send the blob");
+            // Keep the senders allocated
+            forget(senders);
+        }
+    }
+
+    let senders_ptr = Box::into_raw(Box::new(Senders {
+        completion_sender: sender,
+        blobs_sender: blobs_sender,
+    })) as *const _ as *mut c_void;
+
+    unsafe {
+        spdk_bs_opts_init(&mut bs_opts);
+
+        bs_opts.iter_cb_fn = Some(iter_callback);
+        bs_opts.iter_cb_arg = senders_ptr;
+
+        spdk_bs_load(
+            bs_dev.bs_dev,
+            &mut bs_opts,
+            Some(complete_callback),
+            senders_ptr,
+        );
+    }
+
+    let res = await!(receiver).expect("Cancellation is not supported");
+
+    match res {
+        Ok(blob_store) => Ok(Blobstore { blob_store }),
+        Err(bserrno) => Err(BlobstoreError::InitError(bserrno))?,
+    }
+}
+
 pub async fn bs_unload(blob_store: Blobstore) -> Result<(), Error> {
     let (sender, receiver) = oneshot::channel();
     unsafe {

spdk-sys/src/env.rs

@@ -2,6 +2,7 @@ use crate::generated::{spdk_dma_free, spdk_dma_malloc};
 use libc::c_void;
 use std::ptr;
 
+#[derive(Debug)]
 pub struct Buf {
     pub(crate) len: usize,
     pub(crate) ptr: *mut c_void,

spdk-sys/src/event.rs

@@ -38,7 +38,21 @@ impl AppOpts {
     }
 
     // TODO: probably need this to properly deallocate pollers :()
-    // pub fn shutdown_cb() {
+    // pub fn shutdown_cb<F>(&mut self, f: &F)
+    // where
+    //     F: fn() -> (),
+    // {
+    //     extern "C" fn shutdown_cb<F>(closure: *mut c_void, _: *mut c_void)
+    //     where
+    //         F: Fn() -> (),
+    //     {
+    //         let opt_closure = closure as *mut F;
+    //         unsafe { (*opt_closure)() }
+    //     }
+
+    //     unsafe {
+    //        self.0.shutdown_cb = Some(shutdown_cb) 
+    //     }
     //     //spdk_app_shutdown_cb
     // }