diff --git a/mgmtd/src/app/runtime.rs b/mgmtd/src/app/runtime.rs index ed6da0e2..a465be1a 100644 --- a/mgmtd/src/app/runtime.rs +++ b/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 = 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() + ); } } diff --git a/mgmtd/src/bee_msg/common.rs b/mgmtd/src/bee_msg/common.rs index 1f01675d..d3d9808a 100644 --- a/mgmtd/src/bee_msg/common.rs +++ b/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) } diff --git a/mgmtd/src/bee_msg/map_targets.rs b/mgmtd/src/bee_msg/map_targets.rs index bb7ee7d6..e30493ea 100644 --- a/mgmtd/src/bee_msg/map_targets.rs +++ b/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 diff --git a/shared/src/concurrency_tracker.rs b/shared/src/concurrency_tracker.rs new file mode 100644 index 00000000..af9f5f9d --- /dev/null +++ b/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, +} + +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 { + 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)); + } +} diff --git a/shared/src/lib.rs b/shared/src/lib.rs index 47a679a3..37b1306b 100644 --- a/shared/src/lib.rs +++ b/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;