Enhance subscribers, queryables and liveliness tokens propagation to improve scalability

commons/zenoh-codec/src/network/declare.rs

@@ -1075,7 +1075,7 @@ where
         if x.wire_expr.has_suffix() {
             flags |= 1;
         }
-        if let Mapping::Receiver = wire_expr.mapping {
+        if let Mapping::Sender = wire_expr.mapping {
             flags |= 1 << 1;
         }
         codec.write(&mut zriter, flags)?;
@@ -1115,9 +1115,9 @@ where
             String::new()
         };
         let mapping = if imsg::has_flag(flags, 1 << 1) {
-            Mapping::Receiver
-        } else {
             Mapping::Sender
+        } else {
+            Mapping::Receiver
         };
 
         Ok((

commons/zenoh-protocol/src/network/declare.rs

@@ -177,6 +177,19 @@ pub mod common {
     pub mod ext {
         use super::*;
 
+        /// Flags:
+        /// - N: Named          If N==1 then the key expr has name/suffix
+        /// - M: Mapping        if M==1 then key expr mapping is the one declared by the sender, else it is the one declared by the receiver
+        ///
+        ///  7 6 5 4 3 2 1 0
+        /// +-+-+-+-+-+-+-+-+
+        /// |X|X|X|X|X|X|M|N|
+        /// +-+-+-+---------+
+        /// ~ key_scope:z16 ~
+        /// +---------------+
+        /// ~  key_suffix   ~  if N==1 -- <u8;z16>
+        /// +---------------+
+        ///
         pub type WireExprExt = zextzbuf!(0x0f, true);
         #[derive(Debug, Clone, PartialEq, Eq)]
         pub struct WireExprType {

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

@@ -16,11 +16,14 @@ use super::tables::TablesLock;
 use super::{resource::*, tables};
 use crate::net::primitives::{McastMux, Mux, Primitives};
 use crate::net::routing::interceptor::{InterceptorTrait, InterceptorsChain};
+use crate::net::routing::RoutingContext;
 use crate::KeyExpr;
 use std::any::Any;
 use std::collections::HashMap;
 use std::fmt;
 use std::sync::Arc;
+use zenoh_protocol::network::declare::{FinalInterest, InterestId};
+use zenoh_protocol::network::{ext, Declare, DeclareBody};
 use zenoh_protocol::zenoh::RequestBody;
 use zenoh_protocol::{
     core::{ExprId, WhatAmI, ZenohId},
@@ -38,6 +41,7 @@ pub struct FaceState {
     #[cfg(feature = "stats")]
     pub(crate) stats: Option<Arc<TransportStats>>,
     pub(crate) primitives: Arc<dyn crate::net::primitives::EPrimitives + Send + Sync>,
+    pub(crate) remote_key_interests: HashMap<InterestId, Option<Arc<Resource>>>,
     pub(crate) local_mappings: HashMap<ExprId, Arc<Resource>>,
     pub(crate) remote_mappings: HashMap<ExprId, Arc<Resource>>,
     pub(crate) next_qid: RequestId,
@@ -66,6 +70,7 @@ impl FaceState {
             #[cfg(feature = "stats")]
             stats,
             primitives,
+            remote_key_interests: HashMap::new(),
             local_mappings: HashMap::new(),
             remote_mappings: HashMap::new(),
             next_qid: 0,
@@ -208,11 +213,75 @@ impl Primitives for Face {
                     msg.ext_nodeid.node_id,
                 );
             }
-            zenoh_protocol::network::DeclareBody::DeclareToken(_m) => todo!(),
-            zenoh_protocol::network::DeclareBody::UndeclareToken(_m) => todo!(),
-            zenoh_protocol::network::DeclareBody::DeclareInterest(_m) => todo!(),
-            zenoh_protocol::network::DeclareBody::FinalInterest(_m) => todo!(),
-            zenoh_protocol::network::DeclareBody::UndeclareInterest(_m) => todo!(),
+            zenoh_protocol::network::DeclareBody::DeclareToken(m) => {
+                log::warn!("Received unsupported {m:?}")
+            }
+            zenoh_protocol::network::DeclareBody::UndeclareToken(m) => {
+                log::warn!("Received unsupported {m:?}")
+            }
+            zenoh_protocol::network::DeclareBody::DeclareInterest(m) => {
+                if m.interest.keyexprs() && m.interest.future() {
+                    register_expr_interest(
+                        &self.tables,
+                        &mut self.state.clone(),
+                        m.id,
+                        m.wire_expr.as_ref(),
+                    );
+                }
+                if m.interest.subscribers() {
+                    declare_sub_interest(
+                        ctrl_lock.as_ref(),
+                        &self.tables,
+                        &mut self.state.clone(),
+                        m.id,
+                        m.wire_expr.as_ref(),
+                        m.interest.current(),
+                        m.interest.future(),
+                        m.interest.aggregate(),
+                    );
+                }
+                if m.interest.queryables() {
+                    declare_qabl_interest(
+                        ctrl_lock.as_ref(),
+                        &self.tables,
+                        &mut self.state.clone(),
+                        m.id,
+                        m.wire_expr.as_ref(),
+                        m.interest.current(),
+                        m.interest.future(),
+                        m.interest.aggregate(),
+                    );
+                }
+                if m.interest.current() {
+                    self.state.primitives.send_declare(RoutingContext::new_out(
+                        Declare {
+                            ext_qos: ext::QoSType::DECLARE,
+                            ext_tstamp: None,
+                            ext_nodeid: ext::NodeIdType::DEFAULT,
+                            body: DeclareBody::FinalInterest(FinalInterest { id: m.id }),
+                        },
+                        self.clone(),
+                    ));
+                }
+            }
+            zenoh_protocol::network::DeclareBody::FinalInterest(m) => {
+                log::warn!("Received unsupported {m:?}")
+            }
+            zenoh_protocol::network::DeclareBody::UndeclareInterest(m) => {
+                unregister_expr_interest(&self.tables, &mut self.state.clone(), m.id);
+                undeclare_sub_interest(
+                    ctrl_lock.as_ref(),
+                    &self.tables,
+                    &mut self.state.clone(),
+                    m.id,
+                );
+                undeclare_qabl_interest(
+                    ctrl_lock.as_ref(),
+                    &self.tables,
+                    &mut self.state.clone(),
+                    m.id,
+                );
+            }
         }
         drop(ctrl_lock);
     }

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

@@ -22,14 +22,97 @@ use std::sync::RwLock;
 use zenoh_core::zread;
 use zenoh_protocol::core::key_expr::{keyexpr, OwnedKeyExpr};
 use zenoh_protocol::network::declare::subscriber::ext::SubscriberInfo;
-use zenoh_protocol::network::declare::{Mode, SubscriberId};
+use zenoh_protocol::network::declare::{InterestId, Mode, SubscriberId};
 use zenoh_protocol::{
     core::{WhatAmI, WireExpr},
     network::{declare::ext, Push},
     zenoh::PushBody,
 };
 use zenoh_sync::get_mut_unchecked;
 
+#[allow(clippy::too_many_arguments)] // TODO refactor
+pub(crate) fn declare_sub_interest(
+    hat_code: &(dyn HatTrait + Send + Sync),
+    tables: &TablesLock,
+    face: &mut Arc<FaceState>,
+    id: InterestId,
+    expr: Option<&WireExpr>,
+    current: bool,
+    future: bool,
+    aggregate: bool,
+) {
+    if let Some(expr) = expr {
+        let rtables = zread!(tables.tables);
+        match rtables
+            .get_mapping(face, &expr.scope, expr.mapping)
+            .cloned()
+        {
+            Some(mut prefix) => {
+                log::debug!(
+                    "{} Declare sub interest {} ({}{})",
+                    face,
+                    id,
+                    prefix.expr(),
+                    expr.suffix
+                );
+                let res = Resource::get_resource(&prefix, &expr.suffix);
+                let (mut res, mut wtables) = if res
+                    .as_ref()
+                    .map(|r| r.context.is_some())
+                    .unwrap_or(false)
+                {
+                    drop(rtables);
+                    let wtables = zwrite!(tables.tables);
+                    (res.unwrap(), wtables)
+                } else {
+                    let mut fullexpr = prefix.expr();
+                    fullexpr.push_str(expr.suffix.as_ref());
+                    let mut matches = keyexpr::new(fullexpr.as_str())
+                        .map(|ke| Resource::get_matches(&rtables, ke))
+                        .unwrap_or_default();
+                    drop(rtables);
+                    let mut wtables = zwrite!(tables.tables);
+                    let mut res =
+                        Resource::make_resource(&mut wtables, &mut prefix, expr.suffix.as_ref());
+                    matches.push(Arc::downgrade(&res));
+                    Resource::match_resource(&wtables, &mut res, matches);
+                    (res, wtables)
+                };
+
+                hat_code.declare_sub_interest(
+                    &mut wtables,
+                    face,
+                    id,
+                    Some(&mut res),
+                    current,
+                    future,
+                    aggregate,
+                );
+            }
+            None => log::error!(
+                "{} Declare sub interest {} for unknown scope {}!",
+                face,
+                id,
+                expr.scope
+            ),
+        }
+    } else {
+        let mut wtables = zwrite!(tables.tables);
+        hat_code.declare_sub_interest(&mut wtables, face, id, None, current, future, aggregate);
+    }
+}
+
+pub(crate) fn undeclare_sub_interest(
+    hat_code: &(dyn HatTrait + Send + Sync),
+    tables: &TablesLock,
+    face: &mut Arc<FaceState>,
+    id: InterestId,
+) {
+    log::debug!("{} Undeclare sub interest {}", face, id,);
+    let mut wtables = zwrite!(tables.tables);
+    hat_code.undeclare_sub_interest(&mut wtables, face, id);
+}
+
 pub(crate) fn declare_subscription(
     hat_code: &(dyn HatTrait + Send + Sync),
     tables: &TablesLock,

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

@@ -21,6 +21,7 @@ use async_trait::async_trait;
 use std::collections::HashMap;
 use std::sync::{Arc, Weak};
 use zenoh_config::WhatAmI;
+use zenoh_protocol::network::declare::InterestId;
 use zenoh_protocol::{
     core::{key_expr::keyexpr, Encoding, WireExpr},
     network::{
@@ -33,6 +34,89 @@ use zenoh_protocol::{
 use zenoh_sync::get_mut_unchecked;
 use zenoh_util::Timed;
 
+#[allow(clippy::too_many_arguments)] // TODO refactor
+pub(crate) fn declare_qabl_interest(
+    hat_code: &(dyn HatTrait + Send + Sync),
+    tables: &TablesLock,
+    face: &mut Arc<FaceState>,
+    id: InterestId,
+    expr: Option<&WireExpr>,
+    current: bool,
+    future: bool,
+    aggregate: bool,
+) {
+    if let Some(expr) = expr {
+        let rtables = zread!(tables.tables);
+        match rtables
+            .get_mapping(face, &expr.scope, expr.mapping)
+            .cloned()
+        {
+            Some(mut prefix) => {
+                log::debug!(
+                    "{} Declare qabl interest {} ({}{})",
+                    face,
+                    id,
+                    prefix.expr(),
+                    expr.suffix
+                );
+                let res = Resource::get_resource(&prefix, &expr.suffix);
+                let (mut res, mut wtables) = if res
+                    .as_ref()
+                    .map(|r| r.context.is_some())
+                    .unwrap_or(false)
+                {
+                    drop(rtables);
+                    let wtables = zwrite!(tables.tables);
+                    (res.unwrap(), wtables)
+                } else {
+                    let mut fullexpr = prefix.expr();
+                    fullexpr.push_str(expr.suffix.as_ref());
+                    let mut matches = keyexpr::new(fullexpr.as_str())
+                        .map(|ke| Resource::get_matches(&rtables, ke))
+                        .unwrap_or_default();
+                    drop(rtables);
+                    let mut wtables = zwrite!(tables.tables);
+                    let mut res =
+                        Resource::make_resource(&mut wtables, &mut prefix, expr.suffix.as_ref());
+                    matches.push(Arc::downgrade(&res));
+                    Resource::match_resource(&wtables, &mut res, matches);
+                    (res, wtables)
+                };
+
+                hat_code.declare_qabl_interest(
+                    &mut wtables,
+                    face,
+                    id,
+                    Some(&mut res),
+                    current,
+                    future,
+                    aggregate,
+                );
+            }
+            None => log::error!(
+                "{} Declare qabl interest {} for unknown scope {}!",
+                face,
+                id,
+                expr.scope
+            ),
+        }
+    } else {
+        let mut wtables = zwrite!(tables.tables);
+        hat_code.declare_qabl_interest(&mut wtables, face, id, None, current, future, aggregate);
+    }
+}
+
+pub(crate) fn undeclare_qabl_interest(
+    hat_code: &(dyn HatTrait + Send + Sync),
+    tables: &TablesLock,
+    face: &mut Arc<FaceState>,
+    id: InterestId,
+) {
+    log::debug!("{} Undeclare qabl interest {}", face, id,);
+    let mut wtables = zwrite!(tables.tables);
+    hat_code.undeclare_qabl_interest(&mut wtables, face, id);
+}
+
 pub(crate) struct Query {
     src_face: Arc<FaceState>,
     src_qid: RequestId,