feat(foundationdb): add support for complex paths in directory

Author: PierreZ

foundationdb/src/directory/directory.rs

@@ -7,6 +7,7 @@
 // copied, modified, or distributed except according to those terms.
 
 use crate::directory::node;
+use crate::directory::node::Node;
 use crate::future::FdbSlice;
 use crate::tuple::hca::HighContentionAllocator;
 use crate::tuple::{pack_into, Subspace};
@@ -59,20 +60,30 @@ impl Default for DirectoryLayer {
 }
 
 impl DirectoryLayer {
-    fn get_path(&self) -> Vec<String> {
-        self.path.to_owned()
+    pub async fn create_or_open(
+        &self,
+        txn: &Transaction,
+        paths: Vec<String>,
+    ) -> Result<Subspace, DirectoryError> {
+        self.create_or_open_internal(txn, paths, vec![], vec![], true, true)
+            .await
     }
 
-    fn get_layer(&self) -> Vec<u8> {
-        self.layer.to_owned()
+    pub async fn create(
+        &self,
+        txn: &Transaction,
+        paths: Vec<String>,
+    ) -> Result<Subspace, DirectoryError> {
+        self.create_or_open_internal(txn, paths, vec![], vec![], true, false)
+            .await
     }
 
-    pub async fn create_or_open(
+    pub async fn open(
         &self,
         txn: &Transaction,
         paths: Vec<String>,
     ) -> Result<Subspace, DirectoryError> {
-        self.create_or_open_internal(txn, paths, vec![], vec![], true, true)
+        self.create_or_open_internal(txn, paths, vec![], vec![], false, true)
             .await
     }
 
@@ -103,11 +114,15 @@ impl DirectoryLayer {
             return Err(DirectoryError::NoPathProvided);
         }
 
-        let mut node = self.find(trx, paths.to_owned()).await?;
-        node.prefetch_metadata(trx).await?;
+        let nodes = self.find_nodes(trx, paths.to_owned()).await?;
+
+        let last_node = nodes.last().expect("could not contain 0 nodes");
+
+        // if the node_subspace of the last element exists, then we do not need to create anything
+        // and we can return it directly
+        if last_node.content_subspace.is_some() {
+            let node = nodes.last().expect("could not contain 0 node");
 
-        // subspace already exists
-        if node.content_subspace.is_some() {
             if !allow_open {
                 return Err(DirectoryError::DirAlreadyExists);
             }
@@ -116,34 +131,21 @@ impl DirectoryLayer {
                 node.check_layer(layer)?;
             }
 
-            return Ok(node.content_subspace.unwrap().to_owned());
+            return Ok(node.content_subspace.clone().unwrap());
         }
 
-        // subspace does not exists
+        // at least one node does not exists, we need to create them
         if !allow_create {
             return Err(DirectoryError::DirNotExists);
         }
 
-        if prefix.len() > 0 {
-            unimplemented!("no prefix allowed yet")
-        }
+        let mut subspace = self.content_subspace.clone();
 
-        if paths.len() != 1 {
-            unimplemented!("paths too long for now")
+        for mut node in nodes {
+            let allocator = self.allocator.allocate(trx).await?;
+            subspace = node.create_subspace(&trx, allocator, &subspace).await?;
         }
-
-        let allocator = self.allocator.allocate(trx).await?;
-        let new_subspace = self.content_subspace.subspace(&allocator);
-
-        // store node in the node_subspace
-        let mut new_node_key = vec![DEFAULT_SUB_DIRS];
-        pack_into(paths.get(0).unwrap(), &mut new_node_key);
-        trx.set(
-            self.node_subspace.subspace(&new_node_key).bytes(),
-            new_subspace.bytes(),
-        );
-
-        Ok(new_subspace)
+        Ok(subspace)
     }
 
     async fn check_version(
@@ -195,42 +197,46 @@ impl DirectoryLayer {
         value.write_u32::<LittleEndian>(MINOR_VERSION).unwrap();
         value.write_u32::<LittleEndian>(PATCH_VERSION).unwrap();
         let version_subspace: &[u8] = b"version";
-        let version_key = self.node_subspace.subspace(&version_subspace);
-
-        trx.set(version_key.bytes(), &value);
+        let directory_version_key = self.node_subspace.subspace(&version_subspace);
+        trx.set(directory_version_key.bytes(), &value);
 
         Ok(())
     }
 
-    async fn find(&self, trx: &Transaction, path: Vec<String>) -> Result<node::Node, FdbError> {
-        let mut node = node::Node {
-            subspace: self.node_subspace.to_owned(),
-            path: path.to_owned(),
-            target_path: vec![],
-            layer: None,
-            content_subspace: None,
-            already_fetched_metadata: false,
-        };
-        for path_name in path.to_owned() {
+    /// walk is crawling the node_subspace and searching for the nodes
+    /// Result will hold a Vec with at least two nodes: root node and as many nodes as paths
+    async fn find_nodes(
+        &self,
+        trx: &Transaction,
+        paths: Vec<String>,
+    ) -> Result<Vec<node::Node>, FdbError> {
+        let mut nodes = vec![];
+
+        let mut subspace = self.node_subspace.to_owned();
+
+        for path_name in paths {
             let mut next_node_key = vec![DEFAULT_SUB_DIRS];
             pack_into(&path_name, &mut next_node_key);
-            let next_node_subspace = node.subspace.subspace(&next_node_key);
+            subspace = subspace.subspace(&next_node_key);
 
-            match trx.get(next_node_subspace.bytes(), false).await? {
-                None => {
-                    if !path.ends_with(&[path_name]) {
-                        unimplemented!("node not found")
-                    }
-                }
+            let mut node = Node {
+                layer: None,
+                node_subspace: subspace.to_owned(),
+                content_subspace: None,
+            };
+
+            node.retrieve_layer(&trx).await?;
+
+            match trx.get(node.node_subspace.bytes(), false).await? {
                 Some(fdb_slice) => {
-                    node.subspace = next_node_subspace;
                     node.content_subspace = Some(Subspace::from_bytes(&*fdb_slice));
-                    node.path.push(path_name.to_owned());
                 }
+                _ => {} // noop in case of a none existing node
             }
+            nodes.push(node);
         }
 
-        Ok(node)
+        Ok(nodes)
     }
 
     async fn get_version_value(&self, trx: &Transaction) -> FdbResult<Option<FdbSlice>> {

foundationdb/src/directory/node.rs

@@ -10,14 +10,10 @@ use crate::tuple::Subspace;
 use crate::{DirectoryError, FdbError, Transaction};
 
 pub(crate) struct Node {
-    pub(crate) subspace: Subspace,
-    pub(crate) path: Vec<String>,
-    pub(crate) target_path: Vec<String>,
     pub(crate) layer: Option<Vec<u8>>,
 
+    pub(crate) node_subspace: Subspace,
     pub(crate) content_subspace: Option<Subspace>,
-
-    pub(crate) already_fetched_metadata: bool,
 }
 
 impl Node {
@@ -34,23 +30,30 @@ impl Node {
         }
     }
 
-    // TODO: https://docs.rs/futures/0.3.4/futures/future/trait.FutureExt.html#method.shared
-    pub(crate) async fn prefetch_metadata(&mut self, trx: &Transaction) -> Result<(), FdbError> {
-        if !self.already_fetched_metadata {
-            self.layer(trx).await?;
-            self.already_fetched_metadata = true;
-        }
-        Ok(())
+    pub(crate) async fn create_subspace(
+        &mut self,
+        trx: &Transaction,
+        allocator: i64,
+        parent_subspace: &Subspace,
+    ) -> Result<Subspace, DirectoryError> {
+        let new_subspace = parent_subspace.subspace(&allocator);
+
+        let key = self.node_subspace.to_owned();
+        trx.set(key.bytes(), new_subspace.bytes());
+
+        self.content_subspace = Some(new_subspace.to_owned());
+
+        Ok(new_subspace.clone())
     }
 
     // retrieve the layer used for this node
-    pub(crate) async fn layer(&mut self, trx: &Transaction) -> Result<(), FdbError> {
+    pub(crate) async fn retrieve_layer(&mut self, trx: &Transaction) -> Result<(), FdbError> {
         if self.layer == None {
-            let key = self.subspace.subspace(&b"layer".to_vec());
+            let key = self.node_subspace.subspace(&b"layer".to_vec());
             self.layer = match trx.get(key.bytes(), false).await {
                 Ok(None) => Some(vec![]),
                 Err(err) => return Err(err),
-                Ok(Some(fv)) => Some(fv.to_vec()),
+                Ok(Some(fdb_slice)) => Some(fdb_slice.to_vec()),
             }
         }
         Ok(())

foundationdb/tests/directory.rs

@@ -9,27 +9,62 @@ use foundationdb::*;
 
 mod common;
 
-async fn test_create_then_open_async() -> FdbResult<()> {
-    let paths = vec![String::from("some_path")];
-    let directory = DirectoryLayer::default();
-
-    let db = common::database().await?;
+async fn test_create_then_open_async(
+    db: &Database,
+    directory: &DirectoryLayer,
+    paths: Vec<String>,
+) -> FdbResult<()> {
+    eprintln!("creating directory for {:?}", paths.to_owned());
     let trx = db.create_trx()?;
-    let create_output = directory.create_or_open(&trx, paths.to_owned()).await;
+    let create_output = directory.create(&trx, paths.to_owned()).await;
     assert!(create_output.is_ok());
 
     trx.commit().await?;
+    eprintln!("opening directory for {:?}", paths.to_owned());
 
     let trx = db.create_trx()?;
-    let get_output = directory.create_or_open(&trx, paths.to_owned()).await;
+    let get_output = directory.open(&trx, paths.to_owned()).await;
     assert!(get_output.is_ok());
+
     assert_eq!(create_output.unwrap().bytes(), get_output.unwrap().bytes());
+    Ok(())
+}
 
+async fn test_create_or_open_async(
+    db: &Database,
+    directory: &DirectoryLayer,
+    paths: Vec<String>,
+) -> FdbResult<()> {
+    let trx = db.create_trx()?;
+    let create_output = directory.create_or_open(&trx, paths.to_owned()).await;
+    assert!(create_output.is_ok());
     Ok(())
 }
 
 #[test]
-fn test_create_or_open() {
+fn test_directory() {
     let _guard = unsafe { foundationdb::boot() };
-    futures::executor::block_on(test_create_then_open_async()).expect("failed to run");
+    let db = futures::executor::block_on(common::database()).expect("cannot open fdb");
+
+    eprintln!("clearing all keys");
+    let trx = db.create_trx().expect("cannot create txn");
+    trx.clear_range(b"\x00", b"\xff");
+    futures::executor::block_on(trx.commit()).expect("could not clear keys");
+
+    eprintln!("creating directories");
+    let directory = DirectoryLayer::default();
+
+    futures::executor::block_on(test_create_or_open_async(
+        &db,
+        &directory,
+        vec![String::from("a")],
+    ))
+    .expect("failed to run");
+
+    futures::executor::block_on(test_create_then_open_async(
+        &db,
+        &directory,
+        vec![String::from("b"), String::from("a")],
+    ))
+    .expect("failed to run");
 }