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

CHANGELOG.md

@@ -6,6 +6,7 @@ Bob versions changelog
 #### Added
 - Add clusterwide delete operation (#364)
 - TLS support, TLS for grpc or rest can be enabled via cluster & node config (#303)
+- Lookup existence in aliens, separate local and remote lookup (#576, #585)
 
 #### Changed
 - Update rust edition to 2021 (#484)

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/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,21 +272,130 @@ 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 mut indices = Vec::new();
+        let local_keys = keys
+            .iter()
+            .enumerate()
+            .filter_map(|(idx, &key)| {
+                if self
+                    .mapper
+                    .get_target_nodes_for_key(key)
+                    .iter()
+                    .find(|node| node.name() == self.mapper.local_node_name())
+                    .is_some()
+                {
+                    indices.push(idx);
+                    Some(key)
+                } else {
+                    None
                 }
+            })
+            .collect::<Vec<BobKey>>();
+        // 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 mut local_alien = Vec::new();
+        for (idx, &r) in exist.iter().enumerate() {
+            if r == false {
+                local_alien.push(keys[idx]);
+            }
+        }
+        // end
+        debug!("local alien keys {:?}", local_alien);
+        if local_alien.len() > 0 {
+            let result = exist_on_local_alien(&self.backend, &local_alien).await?;
+            let mut i = 0;
+            for r in exist.iter_mut() {
+                if *r == false {
+                    *r |= result[i];
+                    i += 1;
+                }
+            }
+            debug!("exist after local alien {:?}", exist);
+        }
+
+        // filter remote not found keys by nodes
+        let mut remote_keys: HashMap<_, (Vec<_>, Vec<_>)> = HashMap::new();
+        for (idx, &r) in exist.iter().enumerate() {
+            if r == false {
+                remote_keys
+                    .entry(self.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]));
             }
         }
+        // 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 mut remote_alien = Vec::new();
+        for (idx, &r) in exist.iter().enumerate() {
+            if r == false {
+                remote_alien.push(keys[idx]);
+            }
+        }
+        // end
+        if remote_alien.len() > 0 {
+            // filter remote nodes
+            let remote_nodes = self
+                .mapper
+                .nodes()
+                .into_iter()
+                .filter_map(|(_, node)| {
+                    if node.name() != self.mapper.local_node_name() {
+                        Some(node.clone())
+                    } else {
+                        None
+                    }
+                })
+                .collect::<Vec<Node>>();
+            // 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);
+                    let mut i = 0;
+                    for r in exist.iter_mut() {
+                        if *r == false {
+                            *r |= inner.inner()[i];
+                            i += 1;
+                        }
+                    }
+                }
+            }
+        }
+
         Ok(exist)
     }