Router is partially reentrant on data reception (#73)

eclipse-zenoh · Mar 12, 2021 · ab4df0a · ab4df0a
1 parent b835579
commit ab4df0a
Show file tree

Hide file tree

Showing 6 changed files with 34 additions and 26 deletions.
diff --git a/zenoh/benches/tables_bench.rs b/zenoh/benches/tables_bench.rs
@@ -97,7 +97,7 @@ fn tables_bench(c: &mut Criterion) {
                 b.iter(|| {
                     task::block_on(async {
                         route_data(
-                            &mut tables,
+                            &tables,
                             &face0,
                             2 as u64,
                             "",
@@ -115,7 +115,7 @@ fn tables_bench(c: &mut Criterion) {
                 b.iter(|| {
                     task::block_on(async {
                         route_data(
-                            &mut tables,
+                            &tables,
                             &face0,
                             0 as u64,
                             "/bench/tables/*",
@@ -133,7 +133,7 @@ fn tables_bench(c: &mut Criterion) {
                 b.iter(|| {
                     task::block_on(async {
                         route_data(
-                            &mut tables,
+                            &tables,
                             &face0,
                             0 as u64,
                             "/bench/tables/A*",

diff --git a/zenoh/src/net/routing/face.rs b/zenoh/src/net/routing/face.rs
@@ -461,9 +461,9 @@ impl Face {
         routing_context: Option<RoutingContext>,
     ) {
         let (prefixid, suffix) = reskey.into();
-        let mut tables = zasyncwrite!(self.tables);
+        let tables = zasyncread!(self.tables);
         route_data(
-            &mut tables,
+            &tables,
             &self.state,
             prefixid,
             suffix,

diff --git a/zenoh/src/net/routing/pubsub.rs b/zenoh/src/net/routing/pubsub.rs
@@ -992,7 +992,7 @@ macro_rules! treat_timestamp {
 #[inline]
 #[allow(clippy::too_many_arguments)]
 pub async fn route_data(
-    tables: &mut Tables,
+    tables: &Tables,
     face: &Arc<FaceState>,
     rid: u64,
     suffix: &str,
@@ -1114,13 +1114,17 @@ pub async fn route_data(
                         }
                     }
 
-                    for context in matching_pulls.iter() {
-                        Arc::get_mut_unchecked(&mut context.clone())
-                            .last_values
-                            .insert(
-                                [&prefix.name(), suffix].concat(),
-                                (data_info.clone(), payload.clone()),
-                            );
+                    if !matching_pulls.is_empty() {
+                        let lock = zasynclock!(tables.pull_caches_lock);
+                        for context in matching_pulls.iter() {
+                            Arc::get_mut_unchecked(&mut context.clone())
+                                .last_values
+                                .insert(
+                                    [&prefix.name(), suffix].concat(),
+                                    (data_info.clone(), payload.clone()),
+                                );
+                        }
+                        drop(lock)
                     }
                 }
             }
@@ -1147,6 +1151,7 @@ pub async fn pull_data(
                 match res.contexts.get_mut(&face.id) {
                     Some(mut ctx) => match &ctx.subs {
                         Some(subinfo) => {
+                            let lock = zasynclock!(tables.pull_caches_lock);
                             for (name, (info, data)) in &ctx.last_values {
                                 let reskey =
                                     Resource::get_best_key(&tables.root_res, name, face.id);
@@ -1162,6 +1167,7 @@ pub async fn pull_data(
                                     .await;
                             }
                             Arc::get_mut_unchecked(&mut ctx).last_values.clear();
+                            drop(lock);
                         }
                         None => {
                             log::error!(

diff --git a/zenoh/src/net/routing/router.rs b/zenoh/src/net/routing/router.rs
@@ -11,7 +11,7 @@
 // Contributors:
 //   ADLINK zenoh team, <zenoh@adlink-labs.tech>
 //
-use async_std::sync::{Arc, RwLock, Weak};
+use async_std::sync::{Arc, Mutex, RwLock, Weak};
 use async_std::task::{sleep, JoinHandle};
 use std::collections::{HashMap, HashSet};
 use std::time::Duration;
@@ -45,6 +45,7 @@ pub struct Tables {
     pub(crate) hlc: Option<HLC>,
     pub(crate) root_res: Arc<Resource>,
     pub(crate) faces: HashMap<usize, Arc<FaceState>>,
+    pub(crate) pull_caches_lock: Mutex<()>,
     pub(crate) router_subs: HashSet<Arc<Resource>>,
     pub(crate) peer_subs: HashSet<Arc<Resource>>,
     pub(crate) router_qabls: HashSet<Arc<Resource>>,
@@ -64,6 +65,7 @@ impl Tables {
             hlc,
             root_res: Resource::root(),
             faces: HashMap::new(),
+            pull_caches_lock: Mutex::new(()),
             router_subs: HashSet::new(),
             peer_subs: HashSet::new(),
             router_qabls: HashSet::new(),

diff --git a/zenoh/src/net/runtime/orchestrator.rs b/zenoh/src/net/runtime/orchestrator.rs
@@ -645,10 +645,10 @@ impl OrchSession {
             }) = msg.body
             {
                 let (rid, suffix) = (&key).into();
-                let mut tables = zasyncwrite!(self.sub_event_handler.tables);
+                let tables = zasyncread!(self.sub_event_handler.tables);
                 let face = &self.sub_event_handler.demux.primitives.state;
                 route_data(
-                    &mut tables,
+                    &tables,
                     face,
                     rid,
                     suffix,

diff --git a/zenoh/tests/tables.rs b/zenoh/tests/tables.rs
@@ -614,8 +614,8 @@ fn client_test() {
         primitives1.clear_data();
         primitives2.clear_data();
         route_data(
-            &mut tables,
-            &mut face0.upgrade().unwrap(),
+            &tables,
+            &face0.upgrade().unwrap(),
             0,
             "/test/client/z1_wr1",
             CongestionControl::Block,
@@ -641,8 +641,8 @@ fn client_test() {
         primitives1.clear_data();
         primitives2.clear_data();
         route_data(
-            &mut tables,
-            &mut face0.upgrade().unwrap(),
+            &tables,
+            &face0.upgrade().unwrap(),
             11,
             "/z1_wr2",
             CongestionControl::Block,
@@ -668,8 +668,8 @@ fn client_test() {
         primitives1.clear_data();
         primitives2.clear_data();
         route_data(
-            &mut tables,
-            &mut face1.upgrade().unwrap(),
+            &tables,
+            &face1.upgrade().unwrap(),
             0,
             "/test/client/**",
             CongestionControl::Block,
@@ -695,8 +695,8 @@ fn client_test() {
         primitives1.clear_data();
         primitives2.clear_data();
         route_data(
-            &mut tables,
-            &mut face0.upgrade().unwrap(),
+            &tables,
+            &face0.upgrade().unwrap(),
             12,
             "",
             CongestionControl::Block,
@@ -722,8 +722,8 @@ fn client_test() {
         primitives1.clear_data();
         primitives2.clear_data();
         route_data(
-            &mut tables,
-            &mut face1.upgrade().unwrap(),
+            &tables,
+            &face1.upgrade().unwrap(),
             22,
             "",
             CongestionControl::Block,