Automatically update the endpoint list when then network configuration change

DEFAULT_CONFIG.json5

@@ -113,6 +113,9 @@
       /// increase factor for the next timeout until next try
       period_increase_factor: 2,
     },
+    /// Interval in millisecond to check if the listening endpoints changed (i.e. when listening on 0.0.0.0 or [::]).
+    /// Also update the multicast scouting listening interfaces. Use -1 to disable.
+    endpoint_poll_interval_ms: 10000,
   },
 
   /// Configure the session open behavior.

commons/zenoh-config/src/defaults.rs

@@ -55,6 +55,8 @@ pub mod listen {
 
     pub const timeout_ms: ModeDependentValue<i64> = ModeDependentValue::Unique(0);
     pub const exit_on_failure: ModeDependentValue<bool> = ModeDependentValue::Unique(true);
+
+    pub const endpoint_poll_interval_ms: Option<i64> = Some(10_000);
 }
 
 #[allow(non_upper_case_globals)]
@@ -180,6 +182,7 @@ impl Default for ListenConfig {
                 peer: Some(vec![]),
                 client: None,
             }),
+            endpoint_poll_interval_ms: Some(10_000),
             exit_on_failure: None,
             retry: None,
         }

commons/zenoh-config/src/lib.rs

@@ -488,6 +488,9 @@ validated_struct::validator! {
             /// if connection timeout exceed, exit from application
             pub exit_on_failure: Option<ModeDependentValue<bool>>,
             pub retry: Option<connection_retry::ConnectionRetryModeDependentConf>,
+            /// Interval in millisecond to check if the listening endpoints changed (e.g. when listening on 0.0.0.0).
+            /// Also update the multicast scouting listening interfaces. Use -1 to disable.
+            pub endpoint_poll_interval_ms: Option<i64>,
         },
         /// Configure the session open behavior.
         pub open: #[derive(Default)]

commons/zenoh-util/src/net/mod.rs

@@ -12,6 +12,8 @@
 //   ZettaScale Zenoh Team, <[email protected]>
 //
 use std::net::{IpAddr, Ipv6Addr};
+#[cfg(unix)]
+use std::sync::RwLock;
 
 #[cfg(unix)]
 use lazy_static::lazy_static;
@@ -20,6 +22,8 @@ use pnet_datalink::NetworkInterface;
 use tokio::net::{TcpSocket, UdpSocket};
 use zenoh_core::zconfigurable;
 #[cfg(unix)]
+use zenoh_core::{zread, zwrite};
+#[cfg(unix)]
 use zenoh_result::zerror;
 use zenoh_result::{bail, ZResult};
 
@@ -30,7 +34,7 @@ zconfigurable! {
 
 #[cfg(unix)]
 lazy_static! {
-    static ref IFACES: Vec<NetworkInterface> = pnet_datalink::interfaces();
+    static ref IFACES: RwLock<Vec<NetworkInterface>> = RwLock::new(pnet_datalink::interfaces());
 }
 
 #[cfg(windows)]
@@ -68,7 +72,7 @@ unsafe fn get_adapters_addresses(af_spec: i32) -> ZResult<Vec<u8>> {
 pub fn get_interface(name: &str) -> ZResult<Option<IpAddr>> {
     #[cfg(unix)]
     {
-        for iface in IFACES.iter() {
+        for iface in zread!(IFACES).iter() {
             if iface.name == name {
                 for ifaddr in &iface.ips {
                     if ifaddr.is_ipv4() {
@@ -131,7 +135,7 @@ pub fn get_interface(name: &str) -> ZResult<Option<IpAddr>> {
 pub fn get_multicast_interfaces() -> Vec<IpAddr> {
     #[cfg(unix)]
     {
-        IFACES
+        zread!(IFACES)
             .iter()
             .filter_map(|iface| {
                 if iface.is_up() && iface.is_running() && iface.is_multicast() {
@@ -155,7 +159,7 @@ pub fn get_multicast_interfaces() -> Vec<IpAddr> {
 pub fn get_local_addresses(interface: Option<&str>) -> ZResult<Vec<IpAddr>> {
     #[cfg(unix)]
     {
-        Ok(IFACES
+        Ok(zread!(IFACES)
             .iter()
             .filter(|iface| {
                 if let Some(interface) = interface.as_ref() {
@@ -205,7 +209,7 @@ pub fn get_local_addresses(interface: Option<&str>) -> ZResult<Vec<IpAddr>> {
 pub fn get_unicast_addresses_of_multicast_interfaces() -> Vec<IpAddr> {
     #[cfg(unix)]
     {
-        IFACES
+        zread!(IFACES)
             .iter()
             .filter(|iface| iface.is_up() && iface.is_running() && iface.is_multicast())
             .flat_map(|iface| {
@@ -228,7 +232,7 @@ pub fn get_unicast_addresses_of_multicast_interfaces() -> Vec<IpAddr> {
 pub fn get_unicast_addresses_of_interface(name: &str) -> ZResult<Vec<IpAddr>> {
     #[cfg(unix)]
     {
-        match IFACES.iter().find(|iface| iface.name == name) {
+        match zread!(IFACES).iter().find(|iface| iface.name == name) {
             Some(iface) => {
                 if !iface.is_up() {
                     bail!("Interface {name} is not up");
@@ -282,7 +286,7 @@ pub fn get_unicast_addresses_of_interface(name: &str) -> ZResult<Vec<IpAddr>> {
 pub fn get_index_of_interface(addr: IpAddr) -> ZResult<u32> {
     #[cfg(unix)]
     {
-        IFACES
+        zread!(IFACES)
             .iter()
             .find(|iface| iface.ips.iter().any(|ipnet| ipnet.ip() == addr))
             .map(|iface| iface.index)
@@ -319,12 +323,12 @@ pub fn get_interface_names_by_addr(addr: IpAddr) -> ZResult<Vec<String>> {
     #[cfg(unix)]
     {
         if addr.is_unspecified() {
-            Ok(IFACES
+            Ok(zread!(IFACES)
                 .iter()
                 .map(|iface| iface.name.clone())
                 .collect::<Vec<String>>())
         } else {
-            Ok(IFACES
+            Ok(zread!(IFACES)
                 .iter()
                 .filter(|iface| iface.ips.iter().any(|ipnet| ipnet.ip() == addr))
                 .map(|iface| iface.name.clone())
@@ -435,6 +439,12 @@ pub fn get_ipv6_ipaddrs(interface: Option<&str>) -> Vec<IpAddr> {
         .collect()
 }
 
+#[cfg(unix)]
+pub fn update_iface_cache() {
+    let mut interfaces = zwrite!(IFACES);
+    *interfaces = pnet_datalink::interfaces();
+}
+
 #[cfg(any(target_os = "linux", target_os = "android"))]
 pub fn set_bind_to_device_tcp_socket(socket: &TcpSocket, iface: &str) -> ZResult<()> {
     socket.bind_device(Some(iface.as_bytes()))?;

zenoh/src/api/scouting.rs

@@ -24,7 +24,7 @@ use crate::{
         builders::scouting::ScoutBuilder,
         handlers::{Callback, CallbackParameter, DefaultHandler},
     },
-    net::runtime::{orchestrator::Loop, Runtime},
+    net::runtime::{orchestrator::Loop, Runtime, Scouting},
     Config,
 };
 
@@ -181,7 +181,7 @@ pub(crate) fn _scout(
             let task = TerminatableTask::spawn(
                 zenoh_runtime::ZRuntime::Acceptor,
                 async move {
-                    let scout = Runtime::scout(&sockets, what, &addr, move |hello| {
+                    let scout = Scouting::scout(&sockets, what, &addr, move |hello| {
                         let callback = callback.clone();
                         async move {
                             callback.call(hello.into());

zenoh/src/net/protocol/network.rs

@@ -987,6 +987,21 @@ impl Network {
         }
     }
 
+    pub(crate) fn update_locators(&mut self) {
+        self.graph[self.idx].sn += 1;
+        self.send_on_links(
+            vec![(
+                self.idx,
+                Details {
+                    zid: false,
+                    locators: true,
+                    links: self.full_linkstate || self.router_peers_failover_brokering,
+                },
+            )],
+            |link| link.transport.get_whatami().unwrap_or(WhatAmI::Peer) == WhatAmI::Router,
+        );
+    }
+
     fn remove_detached_nodes(&mut self) -> Vec<(NodeIndex, Node)> {
         let mut dfs_stack = vec![self.idx];
         let mut visit_map = self.graph.visit_map();

zenoh/src/net/routing/hat/linkstate_peer/mod.rs

@@ -277,6 +277,12 @@ impl HatBaseTrait for HatCode {
         Ok(())
     }
 
+    fn update_self_locators(&self, tables: &mut Tables) {
+        if let Some(net) = hat_mut!(tables).linkstatepeers_net.as_mut() {
+            net.update_locators();
+        }
+    }
+
     fn close_face(
         &self,
         tables: &TablesLock,

zenoh/src/net/routing/hat/mod.rs

@@ -106,6 +106,8 @@ pub(crate) trait HatBaseTrait {
         send_declare: &mut SendDeclare,
     ) -> ZResult<()>;
 
+    fn update_self_locators(&self, _tables: &mut Tables) {}
+
     fn handle_oam(
         &self,
         tables: &mut Tables,

zenoh/src/net/routing/hat/p2p_peer/gossip.rs

@@ -605,4 +605,19 @@ impl Network {
         }
         vec![]
     }
+
+    pub(super) fn update_locators(&mut self) {
+        self.graph[self.idx].sn += 1;
+        self.send_on_links(
+            vec![(
+                self.idx,
+                Details {
+                    zid: false,
+                    locators: true,
+                    links: self.router_peers_failover_brokering,
+                },
+            )],
+            |link| link.transport.get_whatami().unwrap_or(WhatAmI::Peer) == WhatAmI::Router,
+        );
+    }
 }

zenoh/src/net/routing/hat/p2p_peer/mod.rs

@@ -216,6 +216,12 @@ impl HatBaseTrait for HatCode {
         Ok(())
     }
 
+    fn update_self_locators(&self, tables: &mut Tables) {
+        if let Some(net) = hat_mut!(tables).gossip.as_mut() {
+            net.update_locators();
+        }
+    }
+
     fn close_face(
         &self,
         tables: &TablesLock,

zenoh/src/net/routing/hat/router/mod.rs

@@ -460,6 +460,15 @@ impl HatBaseTrait for HatCode {
         Ok(())
     }
 
+    fn update_self_locators(&self, tables: &mut Tables) {
+        if let Some(net) = hat_mut!(tables).routers_net.as_mut() {
+            net.update_locators();
+        }
+        if let Some(net) = hat_mut!(tables).linkstatepeers_net.as_mut() {
+            net.update_locators();
+        }
+    }
+
     fn close_face(
         &self,
         tables: &TablesLock,

zenoh/src/net/runtime/mod.rs

@@ -19,6 +19,7 @@
 //! [Click here for Zenoh's documentation](https://docs.rs/zenoh/latest/zenoh)
 mod adminspace;
 pub mod orchestrator;
+mod scouting;
 
 #[cfg(feature = "plugins")]
 use std::sync::{Mutex, MutexGuard};
@@ -29,11 +30,13 @@ use std::{
         atomic::{AtomicU32, Ordering},
         Arc, Weak,
     },
+    time::Duration,
 };
 
 pub use adminspace::AdminSpace;
 use async_trait::async_trait;
 use futures::{stream::StreamExt, Future};
+pub use scouting::Scouting;
 use tokio::task::JoinHandle;
 use tokio_util::sync::CancellationToken;
 use uhlc::{HLCBuilder, HLC};
@@ -55,6 +58,8 @@ use zenoh_transport::{
     multicast::TransportMulticast, unicast::TransportUnicast, TransportEventHandler,
     TransportManager, TransportMulticastEventHandler, TransportPeer, TransportPeerEventHandler,
 };
+#[cfg(unix)]
+use zenoh_util::net::update_iface_cache;
 
 use self::orchestrator::StartConditions;
 use super::{primitives::DeMux, routing, routing::router::Router};
@@ -87,6 +92,7 @@ pub(crate) struct RuntimeState {
     plugins_manager: Mutex<PluginsManager>,
     start_conditions: Arc<StartConditions>,
     pending_connections: tokio::sync::Mutex<HashSet<ZenohIdProto>>,
+    scouting: tokio::sync::Mutex<Option<Scouting>>,
 }
 
 pub struct WeakRuntime {
@@ -175,6 +181,7 @@ impl RuntimeBuilder {
         // SHM lazy init flag
         #[cfg(feature = "shared-memory")]
         let shm_init_mode = *config.transport.shared_memory.mode();
+        let endpoint_poll_interval = config.listen.endpoint_poll_interval_ms().unwrap_or(10_000);
 
         let config = Notifier::new(crate::config::Config(config));
         let runtime = Runtime {
@@ -193,6 +200,7 @@ impl RuntimeBuilder {
                 plugins_manager: Mutex::new(plugins_manager),
                 start_conditions: Arc::new(StartConditions::default()),
                 pending_connections: tokio::sync::Mutex::new(HashSet::new()),
+                scouting: tokio::sync::Mutex::new(None),
             }),
         };
         *handler.runtime.write().unwrap() = Runtime::downgrade(&runtime);
@@ -240,6 +248,14 @@ impl RuntimeBuilder {
             zenoh_config::ShmInitMode::Lazy => {}
         };
 
+        if endpoint_poll_interval > 0 {
+            let poll_interval = Duration::from_millis(endpoint_poll_interval as u64);
+            runtime.spawn({
+                let runtime2 = runtime.clone();
+                async move { runtime2.monitor_available_addrs(poll_interval).await }
+            });
+        }
+
         Ok(runtime)
     }
 }
@@ -366,6 +382,33 @@ impl Runtime {
     pub(crate) async fn remove_pending_connection(&self, zid: &ZenohIdProto) -> bool {
         self.state.pending_connections.lock().await.remove(zid)
     }
+
+    async fn monitor_available_addrs(&self, poll_interval: Duration) {
+        let token = self.get_cancellation_token();
+        loop {
+            tokio::select! {
+                _ = tokio::time::sleep(poll_interval) => self.update_available_addrs().await,
+                _ = token.cancelled() => return,
+            }
+        }
+    }
+
+    async fn update_available_addrs(&self) {
+        #[cfg(unix)]
+        update_iface_cache();
+
+        if self.update_locators() {
+            let tables_lock = &self.state.router.tables;
+            let _ctrl_lock = zlock!(tables_lock.ctrl_lock);
+            let mut tables = zwrite!(tables_lock.tables);
+            tables_lock.hat_code.update_self_locators(&mut tables);
+        }
+
+        let scouting = self.state.scouting.lock().await;
+        if let Some(scouting) = scouting.as_ref() {
+            scouting.update_addrs_if_needed().await;
+        }
+    }
 }
 
 struct RuntimeTransportEventHandler {
@@ -547,6 +590,7 @@ impl Closee for Arc<RuntimeState> {
         self.manager.close().await;
         // clean up to break cyclic reference of self.state to itself
         self.transport_handlers.write().unwrap().clear();
+        zasynclock!(self.scouting).take();
         // TODO: the call below is needed to prevent intermittent leak
         // due to not freed resource Arc, that apparently happens because
         // the task responsible for resource clean up was aborted earlier than expected.