refactor: refactor

ZettaScaleLabs · Dec 3, 2024 · 1753654 · 1753654
1 parent 5455ecc
commit 1753654
Show file tree

Hide file tree

Showing 5 changed files with 23 additions and 19 deletions.
diff --git a/Cargo.lock b/Cargo.lock
diff --git a/Cargo.toml b/Cargo.toml
@@ -151,6 +151,7 @@ rustls-pemfile = "2.1.3"
 rustls-webpki = "0.102.8"
 rustls-pki-types = "1.8.0"
 schemars = { version = "0.8.21", features = ["either"] }
+scopeguard = "1.2.0"
 secrecy = { version = "0.8.0", features = ["serde", "alloc"] }
 serde = { version = "1.0.210", default-features = false, features = [
   "derive",

diff --git a/zenoh/Cargo.toml b/zenoh/Cargo.toml
@@ -84,6 +84,7 @@ paste = { workspace = true }
 petgraph = { workspace = true }
 phf = { workspace = true }
 rand = { workspace = true, features = ["default"] }
+scopeguard = { workspace = true }
 serde = { workspace = true, features = ["default"] }
 serde_json = { workspace = true }
 socket2 = { workspace = true }

diff --git a/zenoh/src/api/builders/publisher.rs b/zenoh/src/api/builders/publisher.rs
@@ -388,8 +388,7 @@ impl Wait for PublisherBuilder<'_, '_> {
             .declare_publisher_inner(key_expr.clone(), self.destination)?;
         Ok(Publisher {
             session: self.session.downgrade(),
-            // TODO use constants here
-            cache: AtomicU64::new(0b11),
+            cache: AtomicU64::new(0),
             id,
             key_expr,
             encoding: self.encoding,

diff --git a/zenoh/src/api/session.rs b/zenoh/src/api/session.rs
@@ -2171,8 +2171,8 @@ impl SessionInner {
         #[cfg(feature = "unstable")] source_info: SourceInfo,
         attachment: Option<ZBytes>,
     ) -> ZResult<()> {
-        const REMOTE_TAG: u64 = 0b01;
-        const LOCAL_TAG: u64 = 0b10;
+        const NO_REMOTE_FLAG: u64 = 0b01;
+        const NO_LOCAL_FLAG: u64 = 0b10;
         const VERSION_SHIFT: u64 = 2;
         trace!("write({:?}, [...])", key_expr);
         let state = zread!(self.state);
@@ -2181,21 +2181,26 @@ impl SessionInner {
             .as_ref()
             .cloned()
             .ok_or(SessionClosedError)?;
-        let mut cached = REMOTE_TAG | LOCAL_TAG;
-        let mut to_cache = REMOTE_TAG | LOCAL_TAG;
+        let mut cached = 0;
+        let mut update_cache = None;
         if let Some(cache) = cache {
-            cached = cache.load(Ordering::Relaxed);
-            let version = cached >> VERSION_SHIFT;
+            let c = cache.load(Ordering::Relaxed);
+            let version = c >> VERSION_SHIFT;
             if version == state.subscription_version {
-                to_cache = cached;
+                cached = c;
             } else {
-                to_cache = (state.subscription_version << VERSION_SHIFT) | REMOTE_TAG | LOCAL_TAG;
+                cached = (state.subscription_version << VERSION_SHIFT);
             }
+            update_cache = Some(scopeguard::guard((), |_| {
+                if cached != c {
+                    let _ = cache.compare_exchange(c, cached, Ordering::Relaxed, Ordering::Relaxed);
+                }
+            }));
         }
         drop(state);
         let timestamp = timestamp.or_else(|| self.runtime.new_timestamp());
         let wire_expr = key_expr.to_wire(self);
-        if (to_cache & REMOTE_TAG) != 0 && destination != Locality::SessionLocal {
+        if (cached & NO_REMOTE_FLAG) == 0 && destination != Locality::SessionLocal {
             let remote = primitives.route_data(
                 Push {
                     wire_expr: wire_expr.to_owned(),
@@ -2237,10 +2242,10 @@ impl SessionInner {
                 Reliability::DEFAULT,
             );
             if !remote {
-                to_cache &= !REMOTE_TAG;
+                cached |= NO_REMOTE_FLAG
             }
         }
-        if (to_cache & LOCAL_TAG) != 0 && destination != Locality::Remote {
+        if (cached & NO_LOCAL_FLAG) == 0 && destination != Locality::Remote {
             let data_info = DataInfo {
                 kind,
                 encoding: Some(encoding),
@@ -2265,12 +2270,9 @@ impl SessionInner {
                 attachment,
             );
             if !local {
-                to_cache &= !LOCAL_TAG;
+                cached |= NO_LOCAL_FLAG;
             }
         }
-        if let Some(cache) = cache.filter(|_| to_cache != cached) {
-            let _ = cache.compare_exchange(cached, to_cache, Ordering::Relaxed, Ordering::Relaxed);
-        }
         Ok(())
     }
 
@@ -2581,10 +2583,10 @@ impl Primitives for WeakSession {
             }
             zenoh_protocol::network::DeclareBody::DeclareSubscriber(m) => {
                 trace!("recv DeclareSubscriber {} {:?}", m.id, m.wire_expr);
+                let mut state = zwrite!(self.state);
+                state.subscription_version += 1;
                 #[cfg(feature = "unstable")]
                 {
-                    let mut state = zwrite!(self.state);
-                    state.subscription_version += 1;
                     if state.primitives.is_none() {
                         return; // Session closing or closed
                     }