Rb/dont wait for notifications 8.4

mgmtd/src/app/runtime.rs

@@ -2,16 +2,21 @@ use super::*;
 use crate::ClientPulledStateNotification;
 use crate::bee_msg::dispatch_request;
 use crate::license::LicenseVerifier;
+use crate::types::SqliteEnumExt;
 use anyhow::Result;
 use protobuf::license::GetCertDataResult;
 use rusqlite::{Connection, Transaction};
 use shared::conn::msg_dispatch::{DispatchRequest, Request};
 use shared::conn::outgoing::Pool;
+use shared::peak_concurrency_tracker;
 use shared::run_state::WeakRunStateHandle;
-use sqlite::Connections;
+use sqlite::{Connections, TransactionExt, rarray_param};
+use sqlite_check::sql;
 use std::fmt::Debug;
 use std::ops::Deref;
+use std::sync::atomic::{AtomicU32, Ordering};
 use tokio::sync::mpsc;
+use tokio::time::Instant;
 
 /// A collection of Handles used for interacting and accessing the different components of the app.
 ///
@@ -124,24 +129,56 @@ impl App for RuntimeApp {
         node_types: &'static [NodeType],
         msg: &M,
     ) {
-        log::trace!("NOTIFICATION to {node_types:?}: {msg:?}");
+        if node_types.is_empty() {
+            return;
+        }
+
+        static NOTIFICATION_COUNTER: AtomicU32 = AtomicU32::new(0);
+        let trace_data = if log::log_enabled!(log::Level::Trace) {
+            let start_time = Instant::now();
+            let count = NOTIFICATION_COUNTER.fetch_add(1, Ordering::Relaxed) + 1;
+            log::trace!("NOTIFICATION #{count} to {node_types:?}: {msg:?}");
+            Some((count, start_time))
+        } else {
+            None
+        };
+
+        // We want to track the concurrent notification tasks that are going on to make it easy
+        // to catch a potential bottleneck here on big systems.
+        let concurrency_tracker = peak_concurrency_tracker!();
+        if let Some(peak) = concurrency_tracker.peaked() {
+            log::info!("Concurrent notifications peaked at {peak}",);
+        }
 
-        for t in node_types {
-            if let Err(err) = async {
-                let nodes = self
-                    .read_tx(move |tx| crate::db::node::get_with_type(tx, *t))
-                    .await?;
+        let mut node_count = 0;
+        if let Err(err) = async {
+            let nodes: Vec<Uid> = self
+                .read_tx(move |tx| {
+                    Ok(tx.query_map_collect(
+                        sql!("SELECT node_uid FROM nodes WHERE node_type IN rarray(?1)"),
+                        [rarray_param(node_types.iter().map(|e| e.sql_variant()))],
+                        |row| row.get::<_, Uid>(0),
+                    )?)
+                })
+                .await?;
 
-                self.conn
-                    .broadcast_datagram(nodes.into_iter().map(|e| e.uid), msg)
-                    .await?;
+            node_count = nodes.len();
 
-                Ok(()) as Result<_>
-            }
-            .await
-            {
-                log::error!("Notification could not be sent to all {t} nodes: {err:#}");
-            }
+            self.conn.broadcast_datagram(nodes.into_iter(), msg).await?;
+
+            Ok(()) as Result<_>
+        }
+        .await
+        {
+            log::error!("Notification could not be sent: {err:#}");
+        }
+
+        if let Some(td) = trace_data {
+            log::trace!(
+                "-> Notification #{} to {node_count} nodes completed after {:?}",
+                td.0,
+                td.1.elapsed()
+            );
         }
     }
 

mgmtd/src/bee_msg/common.rs

@@ -250,37 +250,42 @@ client version < 8.0)"
 
     let node_num_id = node.num_id();
 
-    // notify all nodes
-    app.send_notifications(
-        match node.node_type() {
-            NodeType::Meta => &[NodeType::Meta, NodeType::Client],
-            NodeType::Storage => &[NodeType::Meta, NodeType::Storage, NodeType::Client],
-            NodeType::Client => &[NodeType::Meta],
-            _ => &[],
-        },
-        &Heartbeat {
-            instance_version: 0,
-            nic_list_version: 0,
-            node_type: node.node_type(),
-            node_alias: String::from(node.alias).into_bytes(),
-            ack_id: "".into(),
-            node_num_id,
-            root_num_id: match meta_root {
-                MetaRoot::Unknown => 0,
-                MetaRoot::Normal(node_id, _) => node_id,
-                MetaRoot::Mirrored(group_id) => group_id.into(),
+    // Don't wait for notifications to go out because server nodes only wait for a short time for
+    // the reponse. On some systems broadcasting the notifications seems to take longer and
+    // registration fails.
+    let app = app.clone();
+    tokio::spawn(async move {
+        app.send_notifications(
+            match node.node_type() {
+                NodeType::Meta => &[NodeType::Meta, NodeType::Client],
+                NodeType::Storage => &[NodeType::Meta, NodeType::Storage, NodeType::Client],
+                NodeType::Client => &[NodeType::Meta],
+                _ => &[],
             },
-            is_root_mirrored: match meta_root {
-                MetaRoot::Unknown | MetaRoot::Normal(_, _) => 0,
-                MetaRoot::Mirrored(_) => 1,
+            &Heartbeat {
+                instance_version: 0,
+                nic_list_version: 0,
+                node_type: node.node_type(),
+                node_alias: String::from(node.alias).into_bytes(),
+                ack_id: "".into(),
+                node_num_id,
+                root_num_id: match meta_root {
+                    MetaRoot::Unknown => 0,
+                    MetaRoot::Normal(node_id, _) => node_id,
+                    MetaRoot::Mirrored(group_id) => group_id.into(),
+                },
+                is_root_mirrored: match meta_root {
+                    MetaRoot::Unknown | MetaRoot::Normal(_, _) => 0,
+                    MetaRoot::Mirrored(_) => 1,
+                },
+                port: msg.port,
+                port_tcp_unused: msg.port,
+                nic_list: nics,
+                machine_uuid: vec![], // No need for the other nodes to know machine UUIDs
             },
-            port: msg.port,
-            port_tcp_unused: msg.port,
-            nic_list: nics,
-            machine_uuid: vec![], // No need for the other nodes to know machine UUIDs
-        },
-    )
-    .await;
+        )
+        .await;
+    });
 
     Ok(node_num_id)
 }

mgmtd/src/bee_msg/map_targets.rs

@@ -35,24 +35,31 @@ impl HandleWithResponse for MapTargets {
             self.node_id
         );
 
-        app.send_notifications(
-            &[NodeType::Meta, NodeType::Storage, NodeType::Client],
-            &MapTargets {
-                target_ids: self.target_ids.clone(),
-                node_id: self.node_id,
-                ack_id: "".into(),
-            },
-        )
-        .await;
-
-        // Map targets alter pool membership, so trigger an immediate pool refresh
-        if updated > 0 {
+        // Don't wait for notifications to go out because storage only waits for a short time for
+        // the response. On some systems broadcasting the notifications seems to take longer and
+        // registration fails.
+        let slf = self.clone();
+        let app = app.clone();
+        tokio::spawn(async move {
             app.send_notifications(
-                &[NodeType::Meta, NodeType::Storage],
-                &RefreshStoragePools { ack_id: "".into() },
+                &[NodeType::Meta, NodeType::Storage, NodeType::Client],
+                &MapTargets {
+                    target_ids: slf.target_ids,
+                    node_id: slf.node_id,
+                    ack_id: "".into(),
+                },
             )
             .await;
-        }
+
+            // Map targets alter pool membership, so trigger an immediate pool refresh
+            if updated > 0 {
+                app.send_notifications(
+                    &[NodeType::Meta, NodeType::Storage],
+                    &RefreshStoragePools { ack_id: "".into() },
+                )
+                .await;
+            }
+        });
 
         // Storage server expects a separate status code for each target map requested. We, however,
         // do a all-or-nothing approach. If e.g. one target id doesn't exist (which is an

shared/src/concurrency_tracker.rs

@@ -0,0 +1,86 @@
+use std::sync::atomic::{AtomicU32, Ordering};
+
+/// Helper for tracking maximum ongoing concurrent operations
+#[must_use]
+pub struct PeakConcurrencyTracker<'a> {
+    in_flight: &'a AtomicU32,
+    peaked_at: Option<u32>,
+}
+
+impl<'a> PeakConcurrencyTracker<'a> {
+    /// Creates a peak concurrency tracker object. Caller needs to define two static atomics passed
+    /// in here that are used to determine how many instances of this object exist and store the
+    /// highest value seen before.
+    pub fn new(in_flight: &'a AtomicU32, peak: &'a AtomicU32) -> Self {
+        let count = in_flight.fetch_add(1, Ordering::Relaxed) + 1;
+        let peaked_at = if count > peak.fetch_max(count, Ordering::Relaxed) {
+            Some(count)
+        } else {
+            None
+        };
+
+        Self {
+            in_flight,
+            peaked_at,
+        }
+    }
+
+    /// If this instance exceeded a previous maximum, returns the new maximum value, otherwise
+    /// `None`.
+    pub fn peaked(&self) -> Option<u32> {
+        self.peaked_at
+    }
+}
+
+impl<'a> Drop for PeakConcurrencyTracker<'a> {
+    fn drop(&mut self) {
+        self.in_flight.fetch_sub(1, Ordering::Relaxed);
+    }
+}
+
+/// Creates a peak concurrency tracker object including the required atomics. Each expansion site
+/// gets its own tracker / atomics.
+#[macro_export]
+macro_rules! peak_concurrency_tracker {
+    () => {{
+        static IN_FLIGHT: ::std::sync::atomic::AtomicU32 = ::std::sync::atomic::AtomicU32::new(0);
+        static PEAK: ::std::sync::atomic::AtomicU32 = ::std::sync::atomic::AtomicU32::new(0);
+        $crate::concurrency_tracker::PeakConcurrencyTracker::new(&IN_FLIGHT, &PEAK)
+    }};
+}
+
+#[cfg(test)]
+mod test {
+    use super::*;
+
+    #[test]
+    fn peak() {
+        let in_flight = AtomicU32::new(0);
+        let peak = AtomicU32::new(0);
+
+        {
+            let a = PeakConcurrencyTracker::new(&in_flight, &peak);
+            assert_eq!(Some(1), a.peaked());
+
+            let b = PeakConcurrencyTracker::new(&in_flight, &peak);
+            assert_eq!(Some(2), b.peaked());
+
+            assert_eq!(2, in_flight.load(Ordering::Relaxed));
+        }
+
+        // Dropping frees up the slots again, but the peak is remembered
+        assert_eq!(0, in_flight.load(Ordering::Relaxed));
+        assert_eq!(2, peak.load(Ordering::Relaxed));
+
+        // Reaching the previous peak, but not exceeding it, doesn't report a new one
+        let c = PeakConcurrencyTracker::new(&in_flight, &peak);
+        assert_eq!(None, c.peaked());
+        let d = PeakConcurrencyTracker::new(&in_flight, &peak);
+        assert_eq!(None, d.peaked());
+
+        // Exceeding it does
+        let e = PeakConcurrencyTracker::new(&in_flight, &peak);
+        assert_eq!(Some(3), e.peaked());
+        assert_eq!(3, peak.load(Ordering::Relaxed));
+    }
+}

shared/src/lib.rs

@@ -9,6 +9,7 @@ mod impl_macros;
 
 pub mod bee_msg;
 pub mod bee_serde;
+pub mod concurrency_tracker;
 pub mod conn;
 #[cfg(feature = "grpc")]
 pub mod grpc;