Select links based on message Priority & Reliability

DEFAULT_CONFIG.json5

@@ -326,11 +326,20 @@
       },
     },
     link: {
-      /// An optional whitelist of protocols to be used for accepting and opening sessions.
-      /// If not configured, all the supported protocols are automatically whitelisted.
-      /// The supported protocols are: ["tcp" , "udp", "tls", "quic", "ws", "unixsock-stream", "vsock"]
-      /// For example, to only enable "tls" and "quic":
-      //   protocols: ["tls", "quic"],
+      /// An optional whitelist of protocols to be used for accepting and opening sessions. If not
+      /// configured, all the supported protocols are automatically whitelisted. The supported
+      /// protocols are: ["tcp" , "udp", "tls", "quic", "ws", "unixsock-stream", "vsock"] For
+      /// example, to only enable "tls" and "quic": protocols: ["tls", "quic"],
+      ///
+      /// ## Endpoint metadata
+      ///
+      /// **priorities**: a range bounded inclusively below and above (e.g. `2..4` signifies
+      /// priorities 2, 3 and 4). This value is used to select the link used for transmission based
+      /// on the Priority of the message in question.
+      ///
+      /// **reliability**: either "best_effort" or "reliable". This value is used to select the link
+      /// used for transmission based on the Reliability of the message in question.
+      ///
       /// Configure the zenoh TX parameters of a link
       tx: {
         /// The resolution in bits to be used for the message sequence numbers.
@@ -387,7 +396,7 @@
             enabled: true,
             /// The maximum time limit (in ms) a message should be retained for batching when back-pressure happens.
             time_limit: 1,
-          }
+          },
         },
       },
       /// Configure the zenoh RX parameters of a link

commons/zenoh-codec/src/transport/init.rs

@@ -45,6 +45,7 @@ where
             resolution,
             batch_size,
             ext_qos,
+            ext_qos_optimized,
             #[cfg(feature = "shared-memory")]
             ext_shm,
             ext_auth,
@@ -59,6 +60,7 @@ where
             header |= flag::S;
         }
         let mut n_exts = (ext_qos.is_some() as u8)
+            + (ext_qos_optimized.is_some() as u8)
             + (ext_auth.is_some() as u8)
             + (ext_mlink.is_some() as u8)
             + (ext_lowlatency.is_some() as u8)
@@ -98,6 +100,10 @@ where
             n_exts -= 1;
             self.write(&mut *writer, (qos, n_exts != 0))?;
         }
+        if let Some(qos_optimized) = ext_qos_optimized.as_ref() {
+            n_exts -= 1;
+            self.write(&mut *writer, (qos_optimized, n_exts != 0))?;
+        }
         #[cfg(feature = "shared-memory")]
         if let Some(shm) = ext_shm.as_ref() {
             n_exts -= 1;
@@ -173,6 +179,7 @@ where
 
         // Extensions
         let mut ext_qos = None;
+        let mut ext_qos_optimized = None;
         #[cfg(feature = "shared-memory")]
         let mut ext_shm = None;
         let mut ext_auth = None;
@@ -190,6 +197,11 @@ where
                     ext_qos = Some(q);
                     has_ext = ext;
                 }
+                ext::QoSOptimized::ID => {
+                    let (q, ext): (ext::QoSOptimized, bool) = eodec.read(&mut *reader)?;
+                    ext_qos_optimized = Some(q);
+                    has_ext = ext;
+                }
                 #[cfg(feature = "shared-memory")]
                 ext::Shm::ID => {
                     let (s, ext): (ext::Shm, bool) = eodec.read(&mut *reader)?;
@@ -229,6 +241,7 @@ where
             resolution,
             batch_size,
             ext_qos,
+            ext_qos_optimized,
             #[cfg(feature = "shared-memory")]
             ext_shm,
             ext_auth,
@@ -255,6 +268,7 @@ where
             batch_size,
             cookie,
             ext_qos,
+            ext_qos_optimized,
             #[cfg(feature = "shared-memory")]
             ext_shm,
             ext_auth,
@@ -269,6 +283,7 @@ where
             header |= flag::S;
         }
         let mut n_exts = (ext_qos.is_some() as u8)
+            + (ext_qos_optimized.is_some() as u8)
             + (ext_auth.is_some() as u8)
             + (ext_mlink.is_some() as u8)
             + (ext_lowlatency.is_some() as u8)
@@ -311,6 +326,10 @@ where
             n_exts -= 1;
             self.write(&mut *writer, (qos, n_exts != 0))?;
         }
+        if let Some(qos_optimized) = ext_qos_optimized.as_ref() {
+            n_exts -= 1;
+            self.write(&mut *writer, (qos_optimized, n_exts != 0))?;
+        }
         #[cfg(feature = "shared-memory")]
         if let Some(shm) = ext_shm.as_ref() {
             n_exts -= 1;
@@ -389,6 +408,7 @@ where
 
         // Extensions
         let mut ext_qos = None;
+        let mut ext_qos_optimized = None;
         #[cfg(feature = "shared-memory")]
         let mut ext_shm = None;
         let mut ext_auth = None;
@@ -406,6 +426,11 @@ where
                     ext_qos = Some(q);
                     has_ext = ext;
                 }
+                ext::QoSOptimized::ID => {
+                    let (q, ext): (ext::QoSOptimized, bool) = eodec.read(&mut *reader)?;
+                    ext_qos_optimized = Some(q);
+                    has_ext = ext;
+                }
                 #[cfg(feature = "shared-memory")]
                 ext::Shm::ID => {
                     let (s, ext): (ext::Shm, bool) = eodec.read(&mut *reader)?;
@@ -446,6 +471,7 @@ where
             batch_size,
             cookie,
             ext_qos,
+            ext_qos_optimized,
             #[cfg(feature = "shared-memory")]
             ext_shm,
             ext_auth,

commons/zenoh-protocol/src/core/endpoint.rs

@@ -187,6 +187,9 @@ impl fmt::Debug for AddressMut<'_> {
 pub struct Metadata<'a>(pub(super) &'a str);
 
 impl<'a> Metadata<'a> {
+    pub const RELIABILITY: &'static str = "reliability";
+    pub const PRIORITIES: &'static str = "priorities";
+
     pub fn as_str(&self) -> &'a str {
         self.0
     }

commons/zenoh-protocol/src/core/locator.rs

@@ -64,6 +64,10 @@ impl Locator {
     pub fn as_str(&self) -> &str {
         self.0.as_str()
     }
+
+    pub fn to_endpoint(&self) -> EndPoint {
+        self.0.clone()
+    }
 }
 
 impl From<EndPoint> for Locator {

commons/zenoh-protocol/src/core/mod.rs

@@ -19,11 +19,13 @@ use alloc::{
 };
 use core::{
     convert::{From, TryFrom, TryInto},
-    fmt,
+    fmt::{self, Display},
     hash::Hash,
+    ops::{Deref, RangeInclusive},
     str::FromStr,
 };
 
+use serde::Serialize;
 pub use uhlc::{Timestamp, NTP64};
 use zenoh_keyexpr::OwnedKeyExpr;
 use zenoh_result::{bail, zerror};
@@ -295,7 +297,7 @@ impl EntityGlobalIdProto {
 }
 
 #[repr(u8)]
-#[derive(Debug, Default, Copy, Clone, Eq, Hash, PartialEq)]
+#[derive(Debug, Default, Copy, Clone, Eq, Hash, PartialEq, PartialOrd, Ord, Serialize)]
 pub enum Priority {
     Control = 0,
     RealTime = 1,
@@ -308,6 +310,115 @@ pub enum Priority {
     Background = 7,
 }
 
+#[derive(Debug, Clone, Eq, Hash, PartialEq, Serialize)]
+/// A [`Priority`] range bounded inclusively below and above.
+pub struct PriorityRange(RangeInclusive<Priority>);
+
+impl Deref for PriorityRange {
+    type Target = RangeInclusive<Priority>;
+
+    fn deref(&self) -> &Self::Target {
+        &self.0
+    }
+}
+
+impl PriorityRange {
+    pub fn new(range: RangeInclusive<Priority>) -> Self {
+        Self(range)
+    }
+
+    /// Returns `true` if `self` is a superset of `other`.
+    pub fn includes(&self, other: &PriorityRange) -> bool {
+        self.start() <= other.start() && other.end() <= self.end()
+    }
+
+    pub fn len(&self) -> usize {
+        *self.end() as usize - *self.start() as usize + 1
+    }
+
+    pub fn is_empty(&self) -> bool {
+        false
+    }
+
+    #[cfg(feature = "test")]
+    pub fn rand() -> Self {
+        use rand::Rng;
+        let mut rng = rand::thread_rng();
+        let start = rng.gen_range(Priority::MAX as u8..Priority::MIN as u8);
+        let end = rng.gen_range((start + 1)..=Priority::MIN as u8);
+
+        Self(Priority::try_from(start).unwrap()..=Priority::try_from(end).unwrap())
+    }
+}
+
+impl Display for PriorityRange {
+    fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result {
+        write!(f, "{}..={}", *self.start() as u8, *self.end() as u8)
+    }
+}
+
+#[derive(Debug)]
+pub enum InvalidPriorityRange {
+    InvalidSyntax { found: String },
+    InvalidBound { message: String },
+}
+
+impl Display for InvalidPriorityRange {
+    fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result {
+        match self {
+            InvalidPriorityRange::InvalidSyntax { found } => write!(f, "invalid PriorityRange string, expected an range of the form `start..=end` but found {found}"),
+            InvalidPriorityRange::InvalidBound { message } => write!(f, "invalid PriorityRange bound: {message}"),
+        }
+    }
+}
+
+#[cfg(feature = "std")]
+impl std::error::Error for InvalidPriorityRange {}
+
+impl FromStr for PriorityRange {
+    type Err = InvalidPriorityRange;
+
+    fn from_str(s: &str) -> Result<Self, Self::Err> {
+        let mut metadata = s.split("..=");
+
+        let start = metadata
+            .next()
+            .ok_or_else(|| InvalidPriorityRange::InvalidSyntax {
+                found: s.to_string(),
+            })?
+            .parse::<u8>()
+            .map(Priority::try_from)
+            .map_err(|err| InvalidPriorityRange::InvalidBound {
+                message: err.to_string(),
+            })?
+            .map_err(|err| InvalidPriorityRange::InvalidBound {
+                message: err.to_string(),
+            })?;
+
+        let end = metadata
+            .next()
+            .ok_or_else(|| InvalidPriorityRange::InvalidSyntax {
+                found: s.to_string(),
+            })?
+            .parse::<u8>()
+            .map(Priority::try_from)
+            .map_err(|err| InvalidPriorityRange::InvalidBound {
+                message: err.to_string(),
+            })?
+            .map_err(|err| InvalidPriorityRange::InvalidBound {
+                message: err.to_string(),
+            })?;
+
+        if metadata.next().is_some() {
+            return Err(InvalidPriorityRange::InvalidSyntax {
+                found: s.to_string(),
+            });
+        };
+
+        Ok(PriorityRange::new(start..=end))
+    }
+}
+
 impl Priority {
     /// Default
     pub const DEFAULT: Self = Self::Data;
@@ -342,7 +453,7 @@ impl TryFrom<u8> for Priority {
     }
 }
 
-#[derive(Debug, Default, Copy, Clone, PartialEq, Eq)]
+#[derive(Debug, Default, Copy, Clone, PartialEq, Eq, Hash, Serialize)]
 #[repr(u8)]
 pub enum Reliability {
     #[default]
@@ -353,6 +464,16 @@ pub enum Reliability {
 impl Reliability {
     pub const DEFAULT: Self = Self::Reliable;
 
+    const BEST_EFFORT_STR: &'static str = "best_effort";
+    const RELIABLE_STR: &'static str = "reliable";
+
+    pub fn as_str(&self) -> &str {
+        match self {
+            Reliability::BestEffort => Reliability::BEST_EFFORT_STR,
+            Reliability::Reliable => Reliability::RELIABLE_STR,
+        }
+    }
+
     #[cfg(feature = "test")]
     pub fn rand() -> Self {
         use rand::Rng;
@@ -367,6 +488,59 @@ impl Reliability {
     }
 }
 
+impl From<bool> for Reliability {
+    fn from(value: bool) -> Self {
+        if value {
+            Reliability::Reliable
+        } else {
+            Reliability::BestEffort
+        }
+    }
+}
+
+impl From<Reliability> for bool {
+    fn from(value: Reliability) -> Self {
+        match value {
+            Reliability::BestEffort => false,
+            Reliability::Reliable => true,
+        }
+    }
+}
+
+#[derive(Debug)]
+pub struct InvalidReliability {
+    found: String,
+}
+
+impl Display for InvalidReliability {
+    fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result {
+        write!(
+            f,
+            "invalid Reliability string, expected `{}` or `{}` but found {}",
+            Reliability::BEST_EFFORT_STR,
+            Reliability::RELIABLE_STR,
+            self.found
+        )
+    }
+}
+
+#[cfg(feature = "std")]
+impl std::error::Error for InvalidReliability {}
+
+impl FromStr for Reliability {
+    type Err = InvalidReliability;
+
+    fn from_str(s: &str) -> Result<Self, Self::Err> {
+        match s {
+            Reliability::RELIABLE_STR => Ok(Reliability::Reliable),
+            Reliability::BEST_EFFORT_STR => Ok(Reliability::BestEffort),
+            other => Err(InvalidReliability {
+                found: other.to_string(),
+            }),
+        }
+    }
+}
+
 #[derive(Debug, Copy, Clone, PartialEq, Eq, Default)]
 pub struct Channel {
     pub priority: Priority,