Enhance subscribers, queryables and liveliness tokens propagation to improve scalability

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

@@ -24,8 +24,8 @@ use zenoh_protocol::{
     core::WireExpr,
     network::{
         declare, id,
-        interest::{self, InterestMode, InterestOptions},
-        Interest, Mapping,
+        interest::{self, Interest, InterestMode, InterestOptions},
+        Mapping,
     },
 };
 

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

@@ -120,7 +120,7 @@ pub mod flag {
 pub type DeclareRequestId = u32;
 pub type AtomicDeclareRequestId = AtomicU32;
 
-#[derive(Debug, Clone, PartialEq, Eq)]
+#[derive(Debug, Clone, Copy, PartialEq, Eq)]
 pub enum InterestMode {
     Final,
     Current,

zenoh/src/net/primitives/demux.rs

@@ -67,7 +67,7 @@ impl TransportPeerEventHandler for DeMux {
         match msg.body {
             NetworkBody::Push(m) => self.face.send_push(m),
             NetworkBody::Declare(m) => self.face.send_declare(m),
-            NetworkBody::Interest(_) => todo!(),
+            NetworkBody::Interest(m) => self.face.send_interest(m),
             NetworkBody::Request(m) => self.face.send_request(m),
             NetworkBody::Response(m) => self.face.send_response(m),
             NetworkBody::ResponseFinal(m) => self.face.send_response_final(m),

zenoh/src/net/primitives/mod.rs

@@ -43,6 +43,8 @@ pub trait Primitives: Send + Sync {
 pub(crate) trait EPrimitives: Send + Sync {
     fn as_any(&self) -> &dyn Any;
 
+    fn send_interest(&self, ctx: RoutingContext<Interest>);
+
     fn send_declare(&self, ctx: RoutingContext<Declare>);
 
     fn send_push(&self, msg: Push);
@@ -76,6 +78,8 @@ impl Primitives for DummyPrimitives {
 }
 
 impl EPrimitives for DummyPrimitives {
+    fn send_interest(&self, _ctx: RoutingContext<Interest>) {}
+
     fn send_declare(&self, _ctx: RoutingContext<Declare>) {}
 
     fn send_push(&self, _msg: Push) {}

zenoh/src/net/primitives/mux.rs

@@ -195,6 +195,31 @@ impl Primitives for Mux {
 }
 
 impl EPrimitives for Mux {
+    fn send_interest(&self, ctx: RoutingContext<Interest>) {
+        let ctx = RoutingContext {
+            msg: NetworkMessage {
+                body: NetworkBody::Interest(ctx.msg),
+                #[cfg(feature = "stats")]
+                size: None,
+            },
+            inface: ctx.inface,
+            outface: ctx.outface,
+            prefix: ctx.prefix,
+            full_expr: ctx.full_expr,
+        };
+        let prefix = ctx
+            .wire_expr()
+            .and_then(|we| (!we.has_suffix()).then(|| ctx.prefix()))
+            .flatten()
+            .cloned();
+        let cache = prefix
+            .as_ref()
+            .and_then(|p| p.get_egress_cache(ctx.outface.get().unwrap()));
+        if let Some(ctx) = self.interceptor.intercept(ctx, cache) {
+            let _ = self.handler.schedule(ctx.msg);
+        }
+    }
+
     fn send_declare(&self, ctx: RoutingContext<Declare>) {
         let ctx = RoutingContext {
             msg: NetworkMessage {
@@ -495,6 +520,31 @@ impl Primitives for McastMux {
 }
 
 impl EPrimitives for McastMux {
+    fn send_interest(&self, ctx: RoutingContext<Interest>) {
+        let ctx = RoutingContext {
+            msg: NetworkMessage {
+                body: NetworkBody::Interest(ctx.msg),
+                #[cfg(feature = "stats")]
+                size: None,
+            },
+            inface: ctx.inface,
+            outface: ctx.outface,
+            prefix: ctx.prefix,
+            full_expr: ctx.full_expr,
+        };
+        let prefix = ctx
+            .wire_expr()
+            .and_then(|we| (!we.has_suffix()).then(|| ctx.prefix()))
+            .flatten()
+            .cloned();
+        let cache = prefix
+            .as_ref()
+            .and_then(|p| p.get_egress_cache(ctx.outface.get().unwrap()));
+        if let Some(ctx) = self.interceptor.intercept(ctx, cache) {
+            let _ = self.handler.schedule(ctx.msg);
+        }
+    }
+
     fn send_declare(&self, ctx: RoutingContext<Declare>) {
         let ctx = RoutingContext {
             msg: NetworkMessage {

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

@@ -21,9 +21,10 @@ use crate::KeyExpr;
 use std::any::Any;
 use std::collections::HashMap;
 use std::fmt;
-use std::sync::Arc;
-use zenoh_protocol::network::declare::{FinalInterest, Interest, InterestId};
-use zenoh_protocol::network::{ext, Declare, DeclareBody};
+use std::sync::{Arc, Weak};
+use tokio_util::sync::CancellationToken;
+use zenoh_protocol::network::interest::{InterestId, InterestMode, InterestOptions};
+use zenoh_protocol::network::{ext, Declare, DeclareBody, DeclareFinal};
 use zenoh_protocol::zenoh::RequestBody;
 use zenoh_protocol::{
     core::{ExprId, WhatAmI, ZenohId},
@@ -35,14 +36,20 @@ use zenoh_transport::multicast::TransportMulticast;
 #[cfg(feature = "stats")]
 use zenoh_transport::stats::TransportStats;
 
+pub(crate) struct InterestState {
+    pub(crate) options: InterestOptions,
+    pub(crate) res: Option<Arc<Resource>>,
+    pub(crate) finalized: bool,
+}
+
 pub struct FaceState {
     pub(crate) id: usize,
     pub(crate) zid: ZenohId,
     pub(crate) whatami: WhatAmI,
     #[cfg(feature = "stats")]
     pub(crate) stats: Option<Arc<TransportStats>>,
     pub(crate) primitives: Arc<dyn crate::net::primitives::EPrimitives + Send + Sync>,
-    pub(crate) local_interests: HashMap<InterestId, (Interest, Option<Arc<Resource>>, bool)>,
+    pub(crate) local_interests: HashMap<InterestId, InterestState>,
     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>>,
@@ -191,8 +198,67 @@ impl Face {
 }
 
 impl Primitives for Face {
-    fn send_interest(&self, _msg: zenoh_protocol::network::Interest) {
-        todo!()
+    fn send_interest(&self, msg: zenoh_protocol::network::Interest) {
+        let ctrl_lock = zlock!(self.tables.ctrl_lock);
+        if msg.mode != InterestMode::Final {
+            if msg.options.keyexprs() && msg.mode != InterestMode::Current {
+                register_expr_interest(
+                    &self.tables,
+                    &mut self.state.clone(),
+                    msg.id,
+                    msg.wire_expr.as_ref(),
+                );
+            }
+            if msg.options.subscribers() {
+                declare_sub_interest(
+                    ctrl_lock.as_ref(),
+                    &self.tables,
+                    &mut self.state.clone(),
+                    msg.id,
+                    msg.wire_expr.as_ref(),
+                    msg.mode,
+                    msg.options.aggregate(),
+                );
+            }
+            if msg.options.queryables() {
+                declare_qabl_interest(
+                    ctrl_lock.as_ref(),
+                    &self.tables,
+                    &mut self.state.clone(),
+                    msg.id,
+                    msg.wire_expr.as_ref(),
+                    msg.mode,
+                    msg.options.aggregate(),
+                );
+            }
+            if msg.mode != InterestMode::Future {
+                self.state.primitives.send_declare(RoutingContext::new_out(
+                    Declare {
+                        interest_id: Some(msg.id),
+                        ext_qos: ext::QoSType::DECLARE,
+                        ext_tstamp: None,
+                        ext_nodeid: ext::NodeIdType::DEFAULT,
+                        body: DeclareBody::DeclareFinal(DeclareFinal),
+                    },
+                    self.clone(),
+                ));
+            }
+        } else {
+            unregister_expr_interest(&self.tables, &mut self.state.clone(), msg.id);
+            undeclare_sub_interest(
+                ctrl_lock.as_ref(),
+                &self.tables,
+                &mut self.state.clone(),
+                msg.id,
+            );
+            undeclare_qabl_interest(
+                ctrl_lock.as_ref(),
+                &self.tables,
+                &mut self.state.clone(),
+                msg.id,
+            );
+        }
+        drop(ctrl_lock);
     }
 
     fn send_declare(&self, msg: zenoh_protocol::network::Declare) {
@@ -247,76 +313,18 @@ impl Primitives for Face {
                 );
             }
             zenoh_protocol::network::DeclareBody::DeclareToken(m) => {
-                log::warn!("Received unsupported {m:?}")
+                tracing::warn!("Received unsupported {m:?}")
             }
             zenoh_protocol::network::DeclareBody::UndeclareToken(m) => {
-                log::warn!("Received unsupported {m:?}")
+                tracing::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(),
-                    );
+            zenoh_protocol::network::DeclareBody::DeclareFinal(_) => {
+                if let Some(id) = msg.interest_id {
+                    get_mut_unchecked(&mut self.state.clone())
+                        .local_interests
+                        .entry(id)
+                        .and_modify(|interest| interest.finalized = true);
                 }
-                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) => {
-                get_mut_unchecked(&mut self.state.clone())
-                    .local_interests
-                    .entry(m.id)
-                    .and_modify(|interest| interest.2 = true);
-            }
-            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

@@ -20,23 +20,22 @@ use std::sync::Arc;
 use zenoh_core::zread;
 use zenoh_protocol::core::key_expr::keyexpr;
 use zenoh_protocol::network::declare::subscriber::ext::SubscriberInfo;
-use zenoh_protocol::network::declare::{InterestId, SubscriberId};
+use zenoh_protocol::network::declare::SubscriberId;
+use zenoh_protocol::network::interest::{InterestId, InterestMode};
 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,
+    mode: InterestMode,
     aggregate: bool,
 ) {
     if let Some(expr) = expr {
@@ -46,7 +45,7 @@ pub(crate) fn declare_sub_interest(
             .cloned()
         {
             Some(mut prefix) => {
-                log::debug!(
+                tracing::debug!(
                     "{} Declare sub interest {} ({}{})",
                     face,
                     id,
@@ -82,12 +81,11 @@ pub(crate) fn declare_sub_interest(
                     face,
                     id,
                     Some(&mut res),
-                    current,
-                    future,
+                    mode,
                     aggregate,
                 );
             }
-            None => log::error!(
+            None => tracing::error!(
                 "{} Declare sub interest {} for unknown scope {}!",
                 face,
                 id,
@@ -96,7 +94,7 @@ pub(crate) fn declare_sub_interest(
         }
     } else {
         let mut wtables = zwrite!(tables.tables);
-        hat_code.declare_sub_interest(&mut wtables, face, id, None, current, future, aggregate);
+        hat_code.declare_sub_interest(&mut wtables, face, id, None, mode, aggregate);
     }
 }
 
@@ -106,7 +104,7 @@ pub(crate) fn undeclare_sub_interest(
     face: &mut Arc<FaceState>,
     id: InterestId,
 ) {
-    log::debug!("{} Undeclare sub interest {}", face, id,);
+    tracing::debug!("{} Undeclare sub interest {}", face, id,);
     let mut wtables = zwrite!(tables.tables);
     hat_code.undeclare_sub_interest(&mut wtables, face, id);
 }