[ZEN-547] Add support for link weights (#1914)

* add support for link weights on router

* clippy fixes

* add tests

* change zid type to ZenohId in config

* clippy fixes

* increase delay in tests

* fix test

* clippy fix

* add method to update link weights

* add tests for link weights update

* clippy fix

* clippy fix

* clippy fix

* take the average weight when have conflicting weights on different nodes

* clippy fixes

* support linkstate weights in peers network

* fix interceptor cache, network weights test interaction

* add test for linkstate peers weights

* expose information about link weights to adminspace

* clippy fixes

* default config update

* use greater of 2 weights if they are set by both source and destination

* rename link_weights to transport weights in the config;
update documentation;

* move transport_weights under linkstate field in the config

* Update conf doc

* Rename field destination_zid to dst_zid

* Code style

* Update trace

* Rename fields in LinkInfo

* fix docs

---------

Co-authored-by: OlivierHecart <olivier.hecart@adlinktech.com>

Author: DenisBiryukov91

DEFAULT_CONFIG.json5

@@ -217,11 +217,34 @@
       /// The failover brokering only works if gossip discovery is enabled
       /// and peers are configured with gossip target "router".
       peers_failover_brokering: true,
+      /// Linkstate mode configuration.
+      linkstate: {
+        /// Weights of the outgoing transports in linkstate mode.
+        /// If none of the two endpoint nodes of a transport specifies its weight, a weight of 100 is applied.
+        /// If only one of the two endpoint nodes of a transport specifies its weight, the specified weight is applied.
+        /// If both endpoint nodes of a transport specify its weight, the greater weight is applied.
+ //        transport_weights: [
+ //          { dst_zid: "1", weight: "10" },
+ //          { dst_zid: "2", weight: "200" },
+ //        ]
+      }
     },
     /// The routing strategy to use in peers and it's configuration.
     peer: {
       /// The routing strategy to use in peers. ("peer_to_peer" or "linkstate").
+      /// This option needs to be set to the same value in all peers and routers of the subsystem.
       mode: "peer_to_peer",
+        /// Linkstate mode configuration (only taken into account if mode == "linkstate").
+        linkstate: {
+        /// Weights of the outgoing transports in linkstate mode.
+        /// If none of the two endpoint nodes of a transport specifies its weight, a weight of 100 is applied.
+        /// If only one of the two endpoint nodes of a transport specifies its weight, the specified weight is applied.
+        /// If both endpoint nodes of a transport specify its weight, the greater weight is applied.
+ //        transport_weights: [
+ //          { dst_zid: "1", weight: "10" },
+ //          { dst_zid: "2", weight: "200" },
+ //        ]
+      }
     },
     /// The interests-based routing configuration.
     /// This configuration applies regardless of the mode (router, peer or client).

commons/zenoh-config/src/lib.rs

@@ -28,7 +28,8 @@ pub mod wrappers;
 use std::convert::TryFrom;
 // This is a false positive from the rust analyser
 use std::{
-    any::Any, collections::HashSet, fmt, io::Read, net::SocketAddr, ops, path::Path, sync::Weak,
+    any::Any, collections::HashSet, fmt, io::Read, net::SocketAddr, num::NonZeroU16, ops,
+    path::Path, sync::Weak,
 };
 
 use include::recursive_include;
@@ -82,6 +83,14 @@ impl Zeroize for SecretString {
 
 pub type SecretValue = Secret<SecretString>;
 
+#[derive(Debug, Deserialize, Serialize, Clone)]
+pub struct TransportWeight {
+    /// A zid of destination node.
+    pub dst_zid: ZenohId,
+    /// A weight of link from this node to the destination.
+    pub weight: NonZeroU16,
+}
+
 #[derive(Debug, Deserialize, Serialize, Clone, Copy, Eq, PartialEq)]
 #[serde(rename_all = "lowercase")]
 pub enum InterceptorFlow {
@@ -458,12 +467,24 @@ validated_struct::validator! {
                 /// connected to each other.
                 /// The failover brokering only works if gossip discovery is enabled.
                 peers_failover_brokering: Option<bool>,
+                /// Linkstate mode configuration.
+                pub linkstate: #[derive(Default)]
+                LinkstateConf {
+                    /// Weights of the outgoing links in linkstate mode.
+                    /// If none of the two endpoint nodes of a transport specifies its weight, a weight of 100 is applied.
+                    /// If only one of the two endpoint nodes of a transport specifies its weight, the specified weight is applied.
+                    /// If both endpoint nodes of a transport specify its weight, the greater weight is applied.
+                    pub transport_weights: Vec<TransportWeight>,
+                },
             },
             /// The routing strategy to use in peers and it's configuration.
             pub peer: #[derive(Default)]
             PeerRoutingConf {
                 /// The routing strategy to use in peers. ("peer_to_peer" or "linkstate").
+                /// This option needs to be set to the same value in all peers and routers of the subsystem.
                 mode: Option<String>,
+                /// Linkstate mode configuration (only taken into account if mode == "linkstate").
+                pub linkstate: LinkstateConf,
             },
             /// The interests-based routing configuration.
             /// This configuration applies regardless of the mode (router, peer or client).
@@ -735,7 +756,7 @@ validated_struct::validator! {
         /// Configuration of the downsampling.
         downsampling: Vec<DownsamplingItemConf>,
 
-        ///Configuration of the access control (ACL)
+        /// Configuration of the access control (ACL)
         pub access_control: AclConfig {
             pub enabled: bool,
             pub default_permission: Permission,

zenoh/src/net/codec/linkstate.rs

@@ -49,6 +49,9 @@ where
         if x.locators.is_some() {
             options |= linkstate::LOC;
         }
+        if x.link_weights.is_some() {
+            options |= linkstate::WGT;
+        }
         codec.write(&mut *writer, options)?;
 
         // Body
@@ -68,6 +71,12 @@ where
         for l in x.links.iter() {
             codec.write(&mut *writer, *l)?;
         }
+        if let Some(link_weights) = x.link_weights.as_ref() {
+            // do not write len since it is the same as that of links
+            for w in link_weights.iter() {
+                codec.write(&mut *writer, w)?;
+            }
+        }
 
         Ok(())
     }
@@ -102,20 +111,33 @@ where
         } else {
             None
         };
-        let len: usize = codec.read(&mut *reader)?;
-        let mut links: Vec<u64> = Vec::with_capacity(len);
-        for _ in 0..len {
+        let links_len: usize = codec.read(&mut *reader)?;
+        let mut links: Vec<u64> = Vec::with_capacity(links_len);
+        for _ in 0..links_len {
             let l: u64 = codec.read(&mut *reader)?;
             links.push(l);
         }
 
+        let link_weights = if imsg::has_option(options, linkstate::WGT) {
+            // number of weights is the same as number of links
+            let mut weights: Vec<u16> = Vec::with_capacity(links_len);
+            for _ in 0..links_len {
+                let w: u16 = codec.read(&mut *reader)?;
+                weights.push(w);
+            }
+            Some(weights)
+        } else {
+            None
+        };
+
         Ok(LinkState {
             psid,
             sn,
             zid,
             whatami,
             locators,
             links,
+            link_weights,
         })
     }
 }

zenoh/src/net/protocol/linkstate.rs

@@ -11,15 +11,20 @@
 // Contributors:
 //   ZettaScale Zenoh Team, <zenoh@zettascale.tech>
 //
+use std::{collections::HashMap, num::NonZeroU16};
+
+use zenoh_config::TransportWeight;
 use zenoh_protocol::core::{Locator, WhatAmI, ZenohIdProto};
+use zenoh_result::ZResult;
 
 pub const PID: u64 = 1; // 0x01
 pub const WAI: u64 = 1 << 1; // 0x02
 pub const LOC: u64 = 1 << 2; // 0x04
+pub const WGT: u64 = 1 << 3; // 0x08
 
 //  7 6 5 4 3 2 1 0
 // +-+-+-+-+-+-+-+-+
-// ~X|X|X|X|X|L|W|P~
+// ~X|X|X|X|H|L|W|P~
 // +-+-+-+-+-+-+-+-+
 // ~     psid      ~
 // +---------------+
@@ -33,6 +38,8 @@ pub const LOC: u64 = 1 << 2; // 0x04
 // +---------------+
 // ~    [links]    ~
 // +---------------+
+// ~    [weights]  ~ if H = 1
+// +---------------+
 #[derive(Debug, Clone, PartialEq, Eq)]
 pub(crate) struct LinkState {
     pub(crate) psid: u64,
@@ -41,6 +48,49 @@ pub(crate) struct LinkState {
     pub(crate) whatami: Option<WhatAmI>,
     pub(crate) locators: Option<Vec<Locator>>,
     pub(crate) links: Vec<u64>,
+    pub(crate) link_weights: Option<Vec<u16>>,
+}
+
+#[derive(Default, Copy, Debug, Clone, PartialEq, Eq)]
+pub(crate) struct LinkEdgeWeight(pub(crate) Option<NonZeroU16>);
+
+impl LinkEdgeWeight {
+    const DEFAULT_LINK_WEIGHT: u16 = 100;
+
+    pub(crate) fn new(val: NonZeroU16) -> Self {
+        LinkEdgeWeight(Some(val))
+    }
+
+    pub(crate) fn from_raw(val: u16) -> Self {
+        LinkEdgeWeight(NonZeroU16::new(val))
+    }
+
+    pub(crate) fn value(&self) -> u16 {
+        match self.0 {
+            Some(v) => v.get(),
+            None => Self::DEFAULT_LINK_WEIGHT,
+        }
+    }
+
+    pub(crate) fn as_raw(&self) -> u16 {
+        match self.0 {
+            Some(v) => v.get(),
+            None => 0,
+        }
+    }
+
+    pub(crate) fn is_set(&self) -> bool {
+        self.0.is_some()
+    }
+}
+
+#[derive(Debug, Clone, PartialEq, Eq)]
+pub(crate) struct LocalLinkState {
+    pub(crate) sn: u64,
+    pub(crate) zid: ZenohIdProto,
+    pub(crate) whatami: WhatAmI,
+    pub(crate) locators: Option<Vec<Locator>>,
+    pub(crate) links: HashMap<ZenohIdProto, LinkEdgeWeight>,
 }
 
 impl LinkState {
@@ -115,3 +165,36 @@ impl LinkStateList {
         Self { link_states }
     }
 }
+
+pub(crate) fn link_weights_from_config(
+    link_weights: Vec<TransportWeight>,
+    network_name: &str,
+) -> ZResult<HashMap<ZenohIdProto, LinkEdgeWeight>> {
+    let mut link_weights_by_zid = HashMap::new();
+    for lw in link_weights {
+        if link_weights_by_zid
+            .insert(lw.dst_zid.into(), LinkEdgeWeight::new(lw.weight))
+            .is_some()
+        {
+            bail!(
+                "{} config contains a duplicate zid value for transport weight: {}",
+                network_name,
+                lw.dst_zid
+            );
+        }
+    }
+    Ok(link_weights_by_zid)
+}
+
+impl From<LinkEdgeWeight> for Option<u16> {
+    fn from(value: LinkEdgeWeight) -> Self {
+        value.is_set().then_some(value.value())
+    }
+}
+
+#[derive(PartialEq, Debug, serde::Serialize)]
+pub(crate) struct LinkInfo {
+    pub(crate) src_weight: Option<u16>,
+    pub(crate) dst_weight: Option<u16>,
+    pub(crate) actual_weight: u16,
+}

zenoh/src/net/protocol/mod.rs

@@ -12,3 +12,7 @@
 //   ZettaScale Zenoh Team, <zenoh@zettascale.tech>
 //
 pub(crate) mod linkstate;
+pub(crate) mod network;
+
+pub(crate) const ROUTERS_NET_NAME: &str = "[Routers Network]";
+pub(crate) const PEERS_NET_NAME: &str = "[Peers Network]";