Improve interceptors performances (#2053)

* Improve interceptors performances

* Fix bugs$

* Fix reentrant deadlock

* Uncomment tests

Author: OlivierHecart

zenoh/src/net/primitives/demux.rs

@@ -11,21 +11,22 @@
 // Contributors:
 //   ZettaScale Zenoh Team, <[email protected]>
 //
-use std::{any::Any, sync::Arc};
+use std::{any::Any, cell::OnceCell, sync::Arc};
 
 use arc_swap::ArcSwap;
 use zenoh_link::Link;
 use zenoh_protocol::network::{
-    ext, response, Declare, DeclareBody, DeclareFinal, NetworkBodyMut, NetworkMessageMut,
-    ResponseFinal,
+    ext, response, Declare, DeclareBody, DeclareFinal, NetworkBodyMut, NetworkMessageExt as _,
+    NetworkMessageMut, ResponseFinal,
 };
 use zenoh_result::ZResult;
 use zenoh_transport::{unicast::TransportUnicast, TransportPeerEventHandler};
 
 use super::Primitives;
 use crate::net::routing::{
     dispatcher::face::Face,
-    interceptor::{InterceptorTrait, InterceptorsChain},
+    interceptor::{InterceptorContext, InterceptorTrait, InterceptorsChain},
+    router::{InterceptorCacheValueType, Resource},
     RoutingContext,
 };
 
@@ -49,26 +50,73 @@ impl DeMux {
     }
 }
 
+struct DeMuxContext<'a> {
+    demux: &'a DeMux,
+    cache: OnceCell<InterceptorCacheValueType>,
+    expr: OnceCell<String>,
+}
+
+impl DeMuxContext<'_> {
+    fn prefix(&self, msg: &NetworkMessageMut) -> Option<Arc<Resource>> {
+        if let Some(wire_expr) = msg.wire_expr() {
+            let wire_expr = wire_expr.to_owned();
+            if let Some(prefix) = zread!(self.demux.face.tables.tables)
+                .get_mapping(&self.demux.face.state, &wire_expr.scope, wire_expr.mapping)
+                .cloned()
+            {
+                return Some(prefix);
+            }
+        }
+        None
+    }
+}
+
+impl InterceptorContext for DeMuxContext<'_> {
+    fn face(&self) -> Option<Face> {
+        Some(self.demux.face.clone())
+    }
+
+    fn full_expr(&self, msg: &NetworkMessageMut) -> Option<&str> {
+        if self.expr.get().is_none() {
+            if let Some(wire_expr) = msg.wire_expr() {
+                if let Some(prefix) = self.prefix(msg) {
+                    self.expr
+                        .set(prefix.expr().to_string() + wire_expr.suffix.as_ref())
+                        .ok();
+                }
+            }
+        }
+        self.expr.get().map(|x| x.as_str())
+    }
+    fn get_cache(&self, msg: &NetworkMessageMut) -> Option<&Box<dyn Any + Send + Sync>> {
+        if self.cache.get().is_none() && msg.wire_expr().is_some_and(|we| !we.has_suffix()) {
+            if let Some(prefix) = self.prefix(msg) {
+                if let Some(cache) =
+                    prefix.get_ingress_cache(&self.demux.face, &self.demux.interceptor.load())
+                {
+                    self.cache.set(cache).ok();
+                }
+            }
+        }
+        self.cache.get().and_then(|c| c.get_ref().as_ref())
+    }
+}
+
 impl TransportPeerEventHandler for DeMux {
     #[inline]
     fn handle_message(&self, mut msg: NetworkMessageMut) -> ZResult<()> {
         let interceptor = self.interceptor.load();
         if !interceptor.interceptors.is_empty() {
-            let mut ctx = RoutingContext::new_in(msg.as_mut(), self.face.clone());
-            let prefix = ctx
-                .wire_expr()
-                .and_then(|we| (!we.has_suffix()).then(|| ctx.prefix()))
-                .flatten()
-                .cloned();
-            let cache_guard = prefix
-                .as_ref()
-                .and_then(|p| p.get_ingress_cache(&self.face, &interceptor));
-            let cache = cache_guard.as_ref().and_then(|c| c.get_ref().as_ref());
-
-            match &ctx.msg.body {
+            let mut ctx = DeMuxContext {
+                demux: self,
+                cache: OnceCell::new(),
+                expr: OnceCell::new(),
+            };
+
+            match &msg.body {
                 NetworkBodyMut::Request(request) => {
                     let request_id = request.id;
-                    if !interceptor.intercept(&mut ctx, cache) {
+                    if !interceptor.intercept(&mut msg, &mut ctx as &mut dyn InterceptorContext) {
                         // request was blocked by an interceptor, we need to send response final to avoid timeout error
                         self.face
                             .state
@@ -83,26 +131,22 @@ impl TransportPeerEventHandler for DeMux {
                 }
                 NetworkBodyMut::Interest(interest) => {
                     let interest_id = interest.id;
-                    if !interceptor.intercept(&mut ctx, cache) {
+                    if !interceptor.intercept(&mut msg, &mut ctx as &mut dyn InterceptorContext) {
                         // request was blocked by an interceptor, we need to send declare final to avoid timeout error
-                        self.face
-                            .state
-                            .primitives
-                            .send_declare(RoutingContext::new_in(
-                                &mut Declare {
-                                    interest_id: Some(interest_id),
-                                    ext_qos: ext::QoSType::DECLARE,
-                                    ext_tstamp: None,
-                                    ext_nodeid: ext::NodeIdType::DEFAULT,
-                                    body: DeclareBody::DeclareFinal(DeclareFinal),
-                                },
-                                self.face.clone(),
-                            ));
+                        self.face.state.primitives.send_declare(RoutingContext::new(
+                            &mut Declare {
+                                interest_id: Some(interest_id),
+                                ext_qos: ext::QoSType::DECLARE,
+                                ext_tstamp: None,
+                                ext_nodeid: ext::NodeIdType::DEFAULT,
+                                body: DeclareBody::DeclareFinal(DeclareFinal),
+                            },
+                        ));
                         return Ok(());
                     }
                 }
                 _ => {
-                    if !interceptor.intercept(&mut ctx, cache) {
+                    if !interceptor.intercept(&mut msg, &mut ctx as &mut dyn InterceptorContext) {
                         return Ok(());
                     }
                 }