cache pipeline selection

Author: wyfo

Cargo.lock

@@ -26,10 +26,10 @@ description = "Internal crate for zenoh."
 
 [features]
 shared-memory = [
-    "zenoh-protocol/shared-memory",
-    "zenoh-shm",
-    "zenoh-codec/shared-memory",
-    "zenoh-buffers/shared-memory",
+  "zenoh-protocol/shared-memory",
+  "zenoh-shm",
+  "zenoh-codec/shared-memory",
+  "zenoh-buffers/shared-memory",
 ]
 auth_pubkey = ["transport_auth", "rsa"]
 auth_usrpwd = ["transport_auth"]
@@ -44,13 +44,14 @@ transport_ws = ["zenoh-link/transport_ws"]
 transport_serial = ["zenoh-link/transport_serial"]
 transport_compression = []
 transport_unixpipe = ["zenoh-link/transport_unixpipe"]
-transport_vsock= ["zenoh-link/transport_vsock"]
+transport_vsock = ["zenoh-link/transport_vsock"]
 stats = ["zenoh-protocol/stats"]
 test = []
 unstable = []
 default = ["test", "transport_multilink"]
 
 [dependencies]
+arc-swap = { workspace = true }
 async-trait = { workspace = true }
 crossbeam-utils = { workspace = true }
 tokio = { workspace = true, features = [
@@ -63,9 +64,9 @@ tokio = { workspace = true, features = [
   "net",
 ] }
 lazy_static = { workspace = true }
-tokio-util = { workspace = true, features = ["rt"]}
+tokio-util = { workspace = true, features = ["rt"] }
 flume = { workspace = true }
-tracing = {workspace = true}
+tracing = { workspace = true }
 lz4_flex = { workspace = true }
 paste = { workspace = true }
 rand = { workspace = true, features = ["default"] }
@@ -91,7 +92,7 @@ zenoh-task = { workspace = true }
 
 [dev-dependencies]
 futures-util = { workspace = true }
-zenoh-util = {workspace = true }
+zenoh-util = { workspace = true }
 zenoh-protocol = { workspace = true, features = ["test"] }
 futures = { workspace = true }
 zenoh-link-commons = { workspace = true }

io/zenoh-transport/Cargo.toml

@@ -13,16 +13,17 @@
 //
 use std::{
     fmt::DebugStruct,
-    sync::{Arc, RwLock},
+    sync::{Arc, RwLock, RwLockReadGuard, RwLockWriteGuard},
     time::Duration,
 };
 
+use arc_swap::ArcSwap;
 use async_trait::async_trait;
 use tokio::sync::{Mutex as AsyncMutex, MutexGuard as AsyncMutexGuard};
 use zenoh_core::{zasynclock, zcondfeat, zread, zwrite};
 use zenoh_link::Link;
 use zenoh_protocol::{
-    core::{Priority, WhatAmI, ZenohIdProto},
+    core::{Locator, Priority, Reliability, WhatAmI, ZenohIdProto},
     network::NetworkMessage,
     transport::{close, Close, PrioritySn, TransportMessage, TransportSn},
 };
@@ -31,7 +32,10 @@ use zenoh_result::{bail, zerror, ZResult};
 #[cfg(feature = "stats")]
 use crate::stats::TransportStats;
 use crate::{
-    common::priority::{TransportPriorityRx, TransportPriorityTx},
+    common::{
+        pipeline::TransmissionPipelineProducer,
+        priority::{TransportPriorityRx, TransportPriorityTx},
+    },
     unicast::{
         authentication::TransportAuthId,
         link::{LinkUnicastWithOpenAck, TransportLinkUnicastDirection},
@@ -41,10 +45,54 @@ use crate::{
     },
     TransportManager, TransportPeerEventHandler,
 };
-
 /*************************************/
 /*        UNIVERSAL TRANSPORT        */
 /*************************************/
+/// Transport link storage with link selection caching functionality
+#[derive(Default)]
+pub(crate) struct TransportLinks {
+    links: RwLock<Box<[TransportLinkUnicastUniversal]>>,
+    cache: ArcSwap<[[Option<Arc<ScheduledLink>>; Priority::NUM]; 2]>,
+}
+
+impl TransportLinks {
+    fn write(&self) -> RwLockWriteGuard<'_, Box<[TransportLinkUnicastUniversal]>> {
+        // on write access to the links we clear pipeline cache
+        self.cache.store(Default::default());
+        zwrite!(self.links)
+    }
+
+    fn read(&self) -> RwLockReadGuard<'_, Box<[TransportLinkUnicastUniversal]>> {
+        zread!(self.links)
+    }
+
+    pub(super) fn get_link(
+        &self,
+        reliability: Reliability,
+        priority: Priority,
+        select: impl FnOnce(&[TransportLinkUnicastUniversal]) -> Option<usize>,
+    ) -> Option<Arc<ScheduledLink>> {
+        if let Some(link) = self.cache.load()[reliability as usize][priority as usize].clone() {
+            return Some(link);
+        }
+        let guard = self.read();
+        let transport_link = &guard[select(&guard)?];
+        let link = Arc::new(ScheduledLink {
+            pipeline: transport_link.pipeline.clone(),
+            dst: transport_link.link.link.get_dst().clone(),
+        });
+        let mut cache = self.cache.load().as_ref().clone();
+        cache[reliability as usize][priority as usize] = Some(link.clone());
+        self.cache.store(Arc::new(cache));
+        Some(link)
+    }
+}
+
+pub(super) struct ScheduledLink {
+    pub(super) pipeline: TransmissionPipelineProducer,
+    pub(super) dst: Locator,
+}
+
 #[derive(Clone)]
 pub(crate) struct TransportUnicastUniversal {
     // Transport Manager
@@ -56,7 +104,7 @@ pub(crate) struct TransportUnicastUniversal {
     // Rx priorities
     pub(super) priority_rx: Arc<[TransportPriorityRx]>,
     // The links associated to the channel
-    pub(super) links: Arc<RwLock<Box<[TransportLinkUnicastUniversal]>>>,
+    pub(super) links: Arc<TransportLinks>,
     // The callback
     pub(super) callback: Arc<RwLock<Option<Arc<dyn TransportPeerEventHandler>>>>,
     // Lock used to ensure no race in add_link method
@@ -104,7 +152,7 @@ impl TransportUnicastUniversal {
             config,
             priority_tx: priority_tx.into_boxed_slice().into(),
             priority_rx: priority_rx.into_boxed_slice().into(),
-            links: Arc::new(RwLock::new(vec![].into_boxed_slice())),
+            links: Default::default(),
             add_link_lock: Arc::new(AsyncMutex::new(())),
             callback: Arc::new(RwLock::new(None)),
             alive: Arc::new(AsyncMutex::new(false)),
@@ -136,7 +184,7 @@ impl TransportUnicastUniversal {
 
         // Close all the links
         let mut links = {
-            let mut l_guard = zwrite!(self.links);
+            let mut l_guard = self.links.write();
             let links = l_guard.to_vec();
             *l_guard = vec![].into_boxed_slice();
             links
@@ -161,7 +209,7 @@ impl TransportUnicastUniversal {
 
         // Try to remove the link
         let target = {
-            let mut guard = zwrite!(self.links);
+            let mut guard = self.links.write();
 
             if let Some(index) = guard.iter().position(|tl| {
                 // Compare LinkUnicast link to not compare TransportLinkUnicast direction
@@ -245,7 +293,7 @@ impl TransportUnicastTrait for TransportUnicastUniversal {
 
         // Check if we can add more inbound links
         {
-            let guard = zread!(self.links);
+            let guard = self.links.read();
             if let TransportLinkUnicastDirection::Inbound = link.inner_config().direction {
                 let count = guard
                     .iter()
@@ -283,7 +331,7 @@ impl TransportUnicastTrait for TransportUnicastUniversal {
             TransportLinkUnicastUniversal::new(self, link, &self.priority_tx);
 
         // Add the link to the channel
-        let mut guard = zwrite!(self.links);
+        let mut guard = self.links.write();
         let mut links = Vec::with_capacity(guard.len() + 1);
         links.extend_from_slice(&guard);
         links.push(link.clone());
@@ -357,7 +405,9 @@ impl TransportUnicastTrait for TransportUnicastUniversal {
     async fn close(&self, reason: u8) -> ZResult<()> {
         tracing::trace!("Closing transport with peer: {}", self.config.zid);
 
-        let mut pipelines = zread!(self.links)
+        let mut pipelines = self
+            .links
+            .read()
             .iter()
             .map(|sl| sl.pipeline.clone())
             .collect::<Vec<_>>();
@@ -379,13 +429,14 @@ impl TransportUnicastTrait for TransportUnicastUniversal {
     }
 
     fn get_links(&self) -> Vec<Link> {
-        zread!(self.links).iter().map(|l| l.link.link()).collect()
+        self.links.read().iter().map(|l| l.link.link()).collect()
     }
 
     fn get_auth_ids(&self) -> TransportAuthId {
         let mut transport_auth_id = TransportAuthId::default();
         // Convert LinkUnicast auth ids to AuthId
-        zread!(self.links)
+        self.links
+            .read()
             .iter()
             .for_each(|l| transport_auth_id.push_link_auth_id(l.link.link.get_auth_id().clone()));
 

io/zenoh-transport/src/unicast/universal/transport.rs

@@ -70,50 +70,39 @@ impl TransportUnicastUniversal {
     }
 
     fn schedule_on_link(&self, msg: NetworkMessage) -> ZResult<bool> {
-        let transport_links = self
+        let Some(link) = self
             .links
-            .read()
-            .expect("reading `TransportUnicastUniversal::links` should not fail");
-
-        let Some(transport_link_index) = Self::select(
-            transport_links.iter().map(|tl| {
-                (
-                    tl.link
-                        .config
-                        .reliability
-                        .unwrap_or(Reliability::from(tl.link.link.is_reliable())),
-                    tl.link.config.priorities.clone(),
+            .get_link(msg.is_reliable().into(), msg.priority(), |links| {
+                Self::select(
+                    links.iter().map(|tl| {
+                        (
+                            tl.link
+                                .config
+                                .reliability
+                                .unwrap_or(Reliability::from(tl.link.link.is_reliable())),
+                            tl.link.config.priorities.clone(),
+                        )
+                    }),
+                    Reliability::from(msg.is_reliable()),
+                    msg.priority(),
                 )
-            }),
-            Reliability::from(msg.is_reliable()),
-            msg.priority(),
-        ) else {
+            })
+        else {
+            // No Link found
             tracing::trace!(
                 "Message dropped because the transport has no links: {}",
                 msg
             );
-
-            // No Link found
             return Ok(false);
         };
-
-        let transport_link = transport_links
-            .get(transport_link_index)
-            .expect("transport link index should be valid");
-
-        let pipeline = transport_link.pipeline.clone();
         tracing::trace!(
             "Scheduled {:?} for transmission to {} ({})",
             msg,
-            transport_link.link.link.get_dst(),
+            link.dst,
             self.get_zid()
         );
-        // Drop the guard before the push_zenoh_message since
-        // the link could be congested and this operation could
-        // block for fairly long time
-        drop(transport_links);
         let droppable = msg.is_droppable();
-        let push = pipeline.push_network_message(msg)?;
+        let push = link.pipeline.push_network_message(msg)?;
         if !push && !droppable {
             tracing::error!(
                 "Unable to push non droppable network message to {}. Closing transport!",