576, 585 lookup existence in aliens, separate remote & local lookup

CHANGELOG.md

@@ -4,6 +4,7 @@ Bob versions changelog
 
 ## [Unreleased]
 #### Added
+- Lookup existence in aliens, separate local and remote lookup (#576, #585)
 
 
 #### Changed

bob-apps/Cargo.toml

@@ -51,7 +51,8 @@ default-features = false
 features = ["signal"]
 
 [dependencies.pearl]
-version = "0.14.0"
+# version = "0.14.0"
+git = "https://github.com/qoollo/pearl"
 
 [build-dependencies]
 tonic-build = "0.6"
@@ -78,4 +79,4 @@ path = "bin/brt.rs"
 
 [[bin]]
 name = "bobt"
-path = "bin/bobt.rs"
+path = "bin/bobt.rs"

bob-backend/Cargo.toml

@@ -37,7 +37,8 @@ tower = "0.4"
 tower-service = "0.3"
 
 [dependencies.pearl]
-version = "0.14.0"
+# version = "0.14.0"
+git = "https://github.com/qoollo/pearl"
 
 [dependencies.tokio]
 version = "1.14"

bob-backend/src/core.rs

@@ -1,7 +1,8 @@
 use crate::prelude::*;
 use std::{
     collections::HashMap,
-    fmt::{Display, Formatter, Result as FMTResult}, hash::Hash,
+    fmt::{Display, Formatter, Result as FMTResult},
+    hash::Hash,
 };
 
 use crate::{
@@ -383,7 +384,11 @@ impl Backend {
         let mut exist = vec![false; keys.len()];
         let keys_by_id_and_path = self.group_keys_by_operations(keys, options);
         for (operation, (keys, indexes)) in keys_by_id_and_path {
-            let result = self.inner.exist(operation, &keys).await;
+            let result = if operation.is_data_alien() {
+                self.inner.exist_alien(operation, &keys).await
+            } else {
+                self.inner.exist(operation, &keys).await
+            };
             if let Ok(result) = result {
                 for (&res, ind) in result.iter().zip(indexes) {
                     exist[ind] |= res;

bob/Cargo.toml

@@ -45,7 +45,8 @@ tower-service = "0.3"
 uuid = "0.8.2"
 
 [dependencies.pearl]
-version = "0.14.0"
+# version = "0.14.0"
+git = "https://github.com/qoollo/pearl"
 
 [dependencies.tokio]
 version = "1.14"

bob/src/cluster/operations.rs

@@ -118,23 +118,6 @@ async fn call_at_least(
     (handles, errors)
 }
 
-pub(crate) fn group_keys_by_nodes(
-    mapper: &Virtual,
-    keys: &[BobKey],
-) -> HashMap<Vec<Node>, (Vec<BobKey>, Vec<usize>)> {
-    let mut keys_by_nodes: HashMap<_, (Vec<_>, Vec<_>)> = HashMap::new();
-    for (ind, &key) in keys.iter().enumerate() {
-        keys_by_nodes
-            .entry(mapper.get_target_nodes_for_key(key).to_vec())
-            .and_modify(|(keys, indexes)| {
-                keys.push(key);
-                indexes.push(ind);
-            })
-            .or_insert_with(|| (vec![key], vec![ind]));
-    }
-    keys_by_nodes
-}
-
 pub(crate) async fn lookup_local_alien(
     backend: &Backend,
     key: BobKey,
@@ -278,11 +261,46 @@ pub(crate) async fn put_local_node(
     backend.put_local(key, data, op).await
 }
 
-pub(crate) async fn delete_at_local_node(
-    backend: &Backend,
-    key: BobKey,
-) -> Result<(), Error> {
+pub(crate) async fn delete_at_local_node(backend: &Backend, key: BobKey) -> Result<(), Error> {
     debug!("local node has vdisk replica, put local");
     backend.delete(key).await?;
     Ok(())
 }
+
+pub(crate) async fn exist_on_local_node(
+    backend: &Backend,
+    keys: &[BobKey],
+) -> Result<Vec<bool>, Error> {
+    Ok(backend
+        .exist(keys, &BobOptions::new_get(Some(GetOptions::new_local())))
+        .await?)
+}
+
+pub(crate) async fn exist_on_local_alien(
+    backend: &Backend,
+    keys: &[BobKey],
+) -> Result<Vec<bool>, Error> {
+    Ok(backend
+        .exist(keys, &BobOptions::new_get(Some(GetOptions::new_alien())))
+        .await?)
+}
+
+pub(crate) async fn exist_on_remote_nodes(
+    nodes: &[Node],
+    keys: &[BobKey],
+) -> Vec<Result<NodeOutput<Vec<bool>>, NodeOutput<Error>>> {
+    LinkManager::call_nodes(nodes.iter(), |client| {
+        Box::pin(client.exist(keys.to_vec(), GetOptions::new_local()))
+    })
+    .await
+}
+
+pub(crate) async fn exist_on_remote_aliens(
+    nodes: &[Node],
+    keys: &[BobKey],
+) -> Vec<Result<NodeOutput<Vec<bool>>, NodeOutput<Error>>> {
+    LinkManager::call_nodes(nodes.iter(), |client| {
+        Box::pin(client.exist(keys.to_vec(), GetOptions::new_alien()))
+    })
+    .await
+}

bob/src/cluster/quorum.rs

@@ -2,13 +2,13 @@ use crate::prelude::*;
 
 use super::{
     operations::{
-        delete_at_nodes, delete_at_local_node, group_keys_by_nodes, lookup_local_alien, lookup_local_node,
+        delete_at_local_node, delete_at_nodes, exist_on_local_alien, exist_on_local_node,
+        exist_on_remote_aliens, exist_on_remote_nodes, lookup_local_alien, lookup_local_node,
         lookup_remote_aliens, lookup_remote_nodes, put_at_least, put_local_all, put_local_node,
         put_sup_nodes, Tasks,
     },
     Cluster,
 };
-use crate::link_manager::LinkManager;
 
 #[derive(Clone)]
 pub(crate) struct Quorum {
@@ -272,18 +272,64 @@ impl Cluster for Quorum {
     }
 
     async fn exist(&self, keys: &[BobKey]) -> Result<Vec<bool>, Error> {
-        let keys_by_nodes = group_keys_by_nodes(&self.mapper, keys);
-        debug!(
-            "EXIST Nodes for fan out: {:?}",
-            &keys_by_nodes.keys().flatten().collect::<Vec<_>>()
-        );
         let len = keys.len();
         let mut exist = vec![false; len];
-        for (nodes, (keys, indexes)) in keys_by_nodes {
-            let cluster_results = LinkManager::exist_on_nodes(&nodes, &keys).await;
-            for result in cluster_results.into_iter().flatten() {
-                for (&r, &ind) in result.inner().iter().zip(&indexes) {
-                    exist[ind] |= r;
+
+        // filter local keys
+        let (indices, local_keys) = filter_local_keys(keys, &self.mapper);
+        // end
+        debug!("local keys {:?}", local_keys);
+        debug!("local indices {:?}", indices);
+        if local_keys.len() > 0 {
+            let result = exist_on_local_node(&self.backend, &local_keys).await?;
+            for (idx, r) in indices.into_iter().zip(result.into_iter()) {
+                exist[idx] |= r;
+            }
+            debug!("exist after local node {:?}", exist);
+        }
+
+        // filter keys that were not found
+        let local_alien = filter_not_found(&exist, keys);
+        // end
+        debug!("local alien keys {:?}", local_alien);
+        if local_alien.len() > 0 {
+            let result = exist_on_local_alien(&self.backend, &local_alien).await?;
+            update_exist(&mut exist, &result);
+            debug!("exist after local alien {:?}", exist);
+        }
+
+        // filter remote not found keys by nodes
+        let remote_keys = filter_remote_not_found_by_nodes(&exist, keys, &self.mapper);
+        // end
+
+        debug!("remote keys by nodes {:?}", remote_keys);
+        if remote_keys.len() > 0 {
+            for (nodes, (keys, indices)) in remote_keys {
+                let result = exist_on_remote_nodes(&nodes, &keys).await;
+                for res in result.into_iter().flatten() {
+                    for (&r, &ind) in res.inner().iter().zip(&indices) {
+                        exist[ind] |= r;
+                    }
+                }
+            }
+            debug!("exist after remote nodes {:?}", exist);
+        }
+
+        // filter remote not found keys
+        let remote_alien = filter_not_found(&exist, keys);
+        // end
+        if remote_alien.len() > 0 {
+            // filter remote nodes
+            let remote_nodes = filter_remote_nodes(&self.mapper);
+            // end
+            debug!("remote nodes {:?}", remote_nodes);
+
+            let result = exist_on_remote_aliens(&remote_nodes, &remote_alien).await;
+            debug!("alien result {:?}", result);
+            for res in result.into_iter() {
+                if let Ok(inner) = res {
+                    debug!("inner {:?}", inner);
+                    update_exist(&mut exist, inner.inner());
                 }
             }
         }
@@ -294,3 +340,73 @@ impl Cluster for Quorum {
         self.delete_on_nodes(key).await
     }
 }
+
+fn filter_local_keys(keys: &[BobKey], mapper: &Virtual) -> (Vec<usize>, Vec<BobKey>)  {
+    let mut indices = Vec::new();
+    let local_keys = keys
+        .iter()
+        .enumerate()
+        .filter_map(|(idx, &key)| {
+            mapper.get_target_nodes_for_key(key).iter().find_map(|node| {
+              if node.name() == mapper.local_node_name() {
+                  indices.push(idx);
+                  Some(key)
+              } else {
+                  None
+              }
+            })
+        })
+        .collect::<Vec<BobKey>>();
+    (indices, local_keys)
+}
+
+fn filter_not_found(exist: &[bool], keys: &[BobKey]) -> Vec<BobKey> {
+    let mut not_found_keys = Vec::new();
+    for (idx, &r) in exist.iter().enumerate() {
+        if r == false {
+            not_found_keys.push(keys[idx]);
+        }
+    }
+    not_found_keys
+}
+
+fn filter_remote_not_found_by_nodes(exist: &[bool], keys: &[BobKey], mapper: &Virtual) -> HashMap<Vec<Node>, (Vec<BobKey>, Vec<usize>)> {
+    let mut remote_keys: HashMap<_, (Vec<_>, Vec<_>)> = HashMap::new();
+    for (idx, &r) in exist.iter().enumerate() {
+        if r == false {
+            remote_keys
+                .entry(mapper.get_target_nodes_for_key(keys[idx]).to_vec())
+                .and_modify(|(closure_keys, indices)| {
+                    closure_keys.push(keys[idx]);
+                    indices.push(idx);
+                })
+                .or_insert_with(|| (vec![keys[idx]], vec![idx]));
+        }
+    }
+    remote_keys
+}
+
+fn filter_remote_nodes(mapper: &Virtual) -> Vec<Node> {
+    let remote_nodes = mapper
+        .nodes()
+        .into_iter()
+        .filter_map(|(_, node)| {
+            if node.name() != mapper.local_node_name() {
+                Some(node.clone())
+            } else {
+                None
+            }
+        })
+        .collect::<Vec<Node>>();
+    remote_nodes
+}
+
+fn update_exist(exist: &mut [bool], result: &[bool]) {
+    let mut i = 0;
+    for r in exist.iter_mut() {
+        if *r == false {
+            *r |= result[i];
+            i += 1;
+        }
+    }
+}