Compute identical routes from different source types once

Author: OlivierHecart

zenoh/src/net/routing/dispatcher/pubsub.rs

@@ -11,11 +11,14 @@
 // Contributors:
 //   ZettaScale Zenoh Team, <[email protected]>
 //
-use std::{collections::HashMap, sync::Arc};
+
+#[zenoh_macros::unstable]
+use std::collections::HashMap;
+use std::sync::Arc;
 
 use zenoh_core::zread;
 use zenoh_protocol::{
-    core::{key_expr::keyexpr, Reliability, WhatAmI, WireExpr},
+    core::{key_expr::keyexpr, Reliability, WireExpr},
     network::{
         declare::{ext, SubscriberId},
         Push,
@@ -177,57 +180,19 @@ pub(crate) fn undeclare_subscription(
     }
 }
 
-fn compute_data_routes_(tables: &Tables, routes: &mut DataRoutes, expr: &mut RoutingExpr) {
-    let indexes = tables.hat_code.get_data_routes_entries(tables);
-
-    let max_idx = indexes.routers.iter().max().unwrap();
-    routes
-        .routers
-        .resize_with((*max_idx as usize) + 1, || Arc::new(HashMap::new()));
-
-    for idx in indexes.routers {
-        routes.routers[idx as usize] =
-            tables
-                .hat_code
-                .compute_data_route(tables, expr, idx, WhatAmI::Router);
-    }
-
-    let max_idx = indexes.peers.iter().max().unwrap();
-    routes
-        .peers
-        .resize_with((*max_idx as usize) + 1, || Arc::new(HashMap::new()));
-
-    for idx in indexes.peers {
-        routes.peers[idx as usize] =
-            tables
-                .hat_code
-                .compute_data_route(tables, expr, idx, WhatAmI::Peer);
-    }
-
-    let max_idx = indexes.clients.iter().max().unwrap();
-    routes
-        .clients
-        .resize_with((*max_idx as usize) + 1, || Arc::new(HashMap::new()));
-
-    for idx in indexes.clients {
-        routes.clients[idx as usize] =
-            tables
-                .hat_code
-                .compute_data_route(tables, expr, idx, WhatAmI::Client);
-    }
-}
-
 pub(crate) fn compute_data_routes(tables: &Tables, expr: &mut RoutingExpr) -> DataRoutes {
     let mut routes = DataRoutes::default();
-    compute_data_routes_(tables, &mut routes, expr);
+    tables
+        .hat_code
+        .compute_data_routes(tables, &mut routes, expr);
     routes
 }
 
 pub(crate) fn update_data_routes(tables: &Tables, res: &mut Arc<Resource>) {
     if res.context.is_some() && !res.expr().contains('*') && res.has_subs() {
         let mut res_mut = res.clone();
         let res_mut = get_mut_unchecked(&mut res_mut);
-        compute_data_routes_(
+        tables.hat_code.compute_data_routes(
             tables,
             &mut res_mut.context_mut().data_routes,
             &mut RoutingExpr::new(res, ""),

zenoh/src/net/routing/dispatcher/queries.rs

@@ -20,7 +20,6 @@ use std::{
 use async_trait::async_trait;
 use tokio_util::sync::CancellationToken;
 use zenoh_buffers::ZBuf;
-use zenoh_config::WhatAmI;
 #[cfg(feature = "stats")]
 use zenoh_protocol::zenoh::reply::ReplyBody;
 use zenoh_protocol::{
@@ -205,57 +204,19 @@ pub(crate) fn undeclare_queryable(
     }
 }
 
-fn compute_query_routes_(tables: &Tables, routes: &mut QueryRoutes, expr: &mut RoutingExpr) {
-    let indexes = tables.hat_code.get_query_routes_entries(tables);
-
-    let max_idx = indexes.routers.iter().max().unwrap();
-    routes.routers.resize_with((*max_idx as usize) + 1, || {
-        Arc::new(QueryTargetQablSet::new())
-    });
-
-    for idx in indexes.routers {
-        routes.routers[idx as usize] =
-            tables
-                .hat_code
-                .compute_query_route(tables, expr, idx, WhatAmI::Router);
-    }
-
-    let max_idx = indexes.peers.iter().max().unwrap();
-    routes.peers.resize_with((*max_idx as usize) + 1, || {
-        Arc::new(QueryTargetQablSet::new())
-    });
-
-    for idx in indexes.peers {
-        routes.peers[idx as usize] =
-            tables
-                .hat_code
-                .compute_query_route(tables, expr, idx, WhatAmI::Peer);
-    }
-
-    let max_idx = indexes.clients.iter().max().unwrap();
-    routes.clients.resize_with((*max_idx as usize) + 1, || {
-        Arc::new(QueryTargetQablSet::new())
-    });
-
-    for idx in indexes.clients {
-        routes.clients[idx as usize] =
-            tables
-                .hat_code
-                .compute_query_route(tables, expr, idx, WhatAmI::Client);
-    }
-}
-
 pub(crate) fn compute_query_routes(tables: &Tables, res: &Arc<Resource>) -> QueryRoutes {
     let mut routes = QueryRoutes::default();
-    compute_query_routes_(tables, &mut routes, &mut RoutingExpr::new(res, ""));
+    tables
+        .hat_code
+        .compute_query_routes(tables, &mut routes, &mut RoutingExpr::new(res, ""));
     routes
 }
 
 pub(crate) fn update_query_routes(tables: &Tables, res: &Arc<Resource>) {
     if res.context.is_some() && !res.expr().contains('*') && res.has_qabls() {
         let mut res_mut = res.clone();
         let res_mut = get_mut_unchecked(&mut res_mut);
-        compute_query_routes_(
+        tables.hat_code.compute_query_routes(
             tables,
             &mut res_mut.context_mut().query_routes,
             &mut RoutingExpr::new(res, ""),

zenoh/src/net/routing/dispatcher/resource.rs

@@ -75,13 +75,6 @@ impl SessionContext {
     }
 }
 
-#[derive(Default)]
-pub(crate) struct RoutesIndexes {
-    pub(crate) routers: Vec<NodeId>,
-    pub(crate) peers: Vec<NodeId>,
-    pub(crate) clients: Vec<NodeId>,
-}
-
 #[derive(Default)]
 pub(crate) struct DataRoutes {
     pub(crate) routers: Vec<Arc<Route>>,

zenoh/src/net/routing/hat/client/mod.rs

@@ -49,7 +49,7 @@ use super::{
 use crate::net::{
     routing::{
         dispatcher::{face::Face, interests::RemoteInterest},
-        router::{compute_data_routes, compute_query_routes, RoutesIndexes},
+        router::{compute_data_routes, compute_query_routes},
     },
     runtime::Runtime,
 };
@@ -330,12 +330,3 @@ impl HatFace {
 }
 
 impl HatTrait for HatCode {}
-
-#[inline]
-fn get_routes_entries() -> RoutesIndexes {
-    RoutesIndexes {
-        routers: vec![0],
-        peers: vec![0],
-        clients: vec![0],
-    }
-}

zenoh/src/net/routing/hat/client/pubsub.rs

@@ -26,7 +26,7 @@ use zenoh_protocol::{
 };
 use zenoh_sync::get_mut_unchecked;
 
-use super::{face_hat, face_hat_mut, get_routes_entries, HatCode, HatFace};
+use super::{face_hat, face_hat_mut, HatCode, HatFace};
 use crate::{
     key_expr::KeyExpr,
     net::routing::{
@@ -36,8 +36,8 @@ use crate::{
             resource::{NodeId, Resource, SessionContext},
             tables::{Route, RoutingExpr, Tables},
         },
-        hat::{HatPubSubTrait, SendDeclare, Sources},
-        router::{update_data_routes_from, RoutesIndexes},
+        hat::{DataRoutes, HatPubSubTrait, SendDeclare, Sources},
+        router::update_data_routes_from,
         RoutingContext,
     },
 };
@@ -402,8 +402,17 @@ impl HatPubSubTrait for HatCode {
         Arc::new(route)
     }
 
-    fn get_data_routes_entries(&self, _tables: &Tables) -> RoutesIndexes {
-        get_routes_entries()
+    fn compute_data_routes(
+        &self,
+        tables: &Tables,
+        routes: &mut DataRoutes,
+        expr: &mut RoutingExpr,
+    ) {
+        let route = self.compute_data_route(tables, expr, 0, WhatAmI::Peer);
+        routes.routers.resize_with(1, || route.clone());
+        routes.peers.resize_with(1, || route.clone());
+        let route = self.compute_data_route(tables, expr, 0, WhatAmI::Client);
+        routes.clients.resize_with(1, || route.clone());
     }
 
     #[zenoh_macros::unstable]

zenoh/src/net/routing/hat/client/queries.rs

@@ -32,7 +32,7 @@ use zenoh_protocol::{
 };
 use zenoh_sync::get_mut_unchecked;
 
-use super::{face_hat, face_hat_mut, get_routes_entries, HatCode, HatFace};
+use super::{face_hat, face_hat_mut, HatCode, HatFace};
 use crate::{
     key_expr::KeyExpr,
     net::routing::{
@@ -41,8 +41,8 @@ use crate::{
             resource::{NodeId, Resource, SessionContext},
             tables::{QueryTargetQabl, QueryTargetQablSet, RoutingExpr, Tables},
         },
-        hat::{HatQueriesTrait, SendDeclare, Sources},
-        router::{update_query_routes_from, RoutesIndexes},
+        hat::{HatQueriesTrait, QueryRoutes, SendDeclare, Sources},
+        router::update_query_routes_from,
         RoutingContext,
     },
 };
@@ -427,8 +427,17 @@ impl HatQueriesTrait for HatCode {
         Arc::new(route)
     }
 
-    fn get_query_routes_entries(&self, _tables: &Tables) -> RoutesIndexes {
-        get_routes_entries()
+    fn compute_query_routes(
+        &self,
+        tables: &Tables,
+        routes: &mut QueryRoutes,
+        expr: &mut RoutingExpr,
+    ) {
+        let route = self.compute_query_route(tables, expr, 0, WhatAmI::Peer);
+        routes.routers.resize_with(1, || route.clone());
+        routes.peers.resize_with(1, || route.clone());
+        let route = self.compute_query_route(tables, expr, 0, WhatAmI::Client);
+        routes.clients.resize_with(1, || route.clone());
     }
 
     #[cfg(feature = "unstable")]

zenoh/src/net/routing/hat/linkstate_peer/mod.rs

@@ -58,7 +58,7 @@ use crate::net::{
     routing::{
         dispatcher::{face::Face, interests::RemoteInterest},
         hat::TREES_COMPUTATION_DELAY_MS,
-        router::{compute_data_routes, compute_query_routes, RoutesIndexes},
+        router::{compute_data_routes, compute_query_routes},
     },
     runtime::Runtime,
 };
@@ -554,20 +554,3 @@ fn get_peer(tables: &Tables, face: &Arc<FaceState>, nodeid: NodeId) -> Option<Ze
 }
 
 impl HatTrait for HatCode {}
-
-#[inline]
-fn get_routes_entries(tables: &Tables) -> RoutesIndexes {
-    let indexes = hat!(tables)
-        .linkstatepeers_net
-        .as_ref()
-        .unwrap()
-        .graph
-        .node_indices()
-        .map(|i| i.index() as NodeId)
-        .collect::<Vec<NodeId>>();
-    RoutesIndexes {
-        routers: indexes.clone(),
-        peers: indexes,
-        clients: vec![0],
-    }
-}

zenoh/src/net/routing/hat/linkstate_peer/pubsub.rs

@@ -31,8 +31,8 @@ use zenoh_protocol::{
 use zenoh_sync::get_mut_unchecked;
 
 use super::{
-    face_hat, face_hat_mut, get_peer, get_routes_entries, hat, hat_mut, network::Network, res_hat,
-    res_hat_mut, HatCode, HatContext, HatFace, HatTables,
+    face_hat, face_hat_mut, get_peer, hat, hat_mut, network::Network, res_hat, res_hat_mut,
+    HatCode, HatContext, HatFace, HatTables,
 };
 #[cfg(feature = "unstable")]
 use crate::key_expr::KeyExpr;
@@ -44,8 +44,7 @@ use crate::net::routing::{
         resource::{NodeId, Resource, SessionContext},
         tables::{Route, RoutingExpr, Tables},
     },
-    hat::{CurrentFutureTrait, HatPubSubTrait, SendDeclare, Sources},
-    router::RoutesIndexes,
+    hat::{CurrentFutureTrait, DataRoutes, HatPubSubTrait, SendDeclare, Sources},
     RoutingContext,
 };
 
@@ -986,8 +985,36 @@ impl HatPubSubTrait for HatCode {
         Arc::new(route)
     }
 
-    fn get_data_routes_entries(&self, tables: &Tables) -> RoutesIndexes {
-        get_routes_entries(tables)
+    fn compute_data_routes(
+        &self,
+        tables: &Tables,
+        routes: &mut DataRoutes,
+        expr: &mut RoutingExpr,
+    ) {
+        let indexes = hat!(tables)
+            .linkstatepeers_net
+            .as_ref()
+            .unwrap()
+            .graph
+            .node_indices()
+            .map(|i| i.index() as NodeId)
+            .collect::<Vec<NodeId>>();
+        let max_idx = indexes.iter().max().unwrap();
+        routes
+            .routers
+            .resize_with((*max_idx as usize) + 1, || Arc::new(HashMap::new()));
+        routes
+            .peers
+            .resize_with((*max_idx as usize) + 1, || Arc::new(HashMap::new()));
+        for idx in indexes {
+            let route = self.compute_data_route(tables, expr, idx, WhatAmI::Peer);
+            routes.routers[idx as usize] = route.clone();
+            routes.peers[idx as usize] = route;
+        }
+
+        routes.clients.resize_with(1, || {
+            self.compute_data_route(tables, expr, 0, WhatAmI::Client)
+        });
     }
 
     #[zenoh_macros::unstable]