IntenseWebs/LibreQoS
3
0 Fork 0
mirror of https://github.com/LibreQoE/LibreQoS.git synced 2026-10-24 22:07:40 -05:00
Code Issues Packages Projects Releases Wiki Activity

Fix: prevent circuit bus handler deadlocks

Browse Source
This commit is contained in:
Herbert Wolverson committed 2026-07-17 14:21:45 -05:00
1 parent ee324cbdf6
commit ba939dcd8c
5 files changed
+863 -263
Download Patch File Download Diff File Expand all files Collapse all files

No files matched your search

src/rust/lqos_bus/src/bus/unix_socket_server.rs
+251 -4
View File
@@ -16,6 +16,7 @@ use thiserror::Error;
use tokio::{
io::{AsyncReadExt, AsyncWriteExt},
net::UnixListener,
sync::{OwnedSemaphorePermit, Semaphore},
task::spawn_blocking,
time::timeout,
};
@@ -25,6 +26,30 @@ use super::BUS_SOCKET_DIRECTORY;
use super::protocol::{decode_session_cbor, encode_reply_cbor, read_frame, write_frame};
const BUS_HANDLER_TIMEOUT: Duration = Duration::from_secs(30);
const MAX_CONCURRENT_BUS_HANDLERS: usize = 16;
#[derive(Clone)]
struct BusHandlerLimiter {
permits: std::sync::Arc<Semaphore>,
limit: usize,
}
impl BusHandlerLimiter {
fn new(limit: usize) -> Self {
Self {
permits: std::sync::Arc::new(Semaphore::new(limit)),
limit,
}
}
fn try_acquire(&self) -> Option<OwnedSemaphorePermit> {
self.permits.clone().try_acquire_owned().ok()
}
fn active_handlers(&self) -> usize {
self.limit.saturating_sub(self.permits.available_permits())
}
}
fn dropped_reply_response_count(reply: &BusReply) -> usize {
reply.responses.len()
@@ -38,6 +63,14 @@ fn timeout_reply(request_count: usize) -> BusReply {
}
}
fn busy_reply(request_count: usize) -> BusReply {
BusReply {
responses: (0..request_count)
.map(|_| BusResponse::Fail("Bus request handler busy".to_string()))
.collect(),
}
}
fn request_kind_summary(request_kinds: &[&'static str]) -> String {
if request_kinds.is_empty() {
return "<none>".to_string();
@@ -62,27 +95,60 @@ async fn handle_requests_with_deadline(
handle_bus_requests: fn(&[BusRequest], &mut Vec<BusResponse>),
requests: Vec<BusRequest>,
request_source: &'static str,
limiter: &BusHandlerLimiter,
) -> BusReply {
handle_requests_with_deadline_for_duration(
handle_requests_with_limiter_for_duration(
handle_bus_requests,
requests,
request_source,
BUS_HANDLER_TIMEOUT,
limiter,
)
.await
}
#[cfg(test)]
async fn handle_requests_with_deadline_for_duration(
handle_bus_requests: fn(&[BusRequest], &mut Vec<BusResponse>),
requests: Vec<BusRequest>,
request_source: &'static str,
timeout_duration: Duration,
) -> BusReply {
let limiter = BusHandlerLimiter::new(MAX_CONCURRENT_BUS_HANDLERS);
handle_requests_with_limiter_for_duration(
handle_bus_requests,
requests,
request_source,
timeout_duration,
&limiter,
)
.await
}
async fn handle_requests_with_limiter_for_duration(
handle_bus_requests: fn(&[BusRequest], &mut Vec<BusResponse>),
requests: Vec<BusRequest>,
request_source: &'static str,
timeout_duration: Duration,
limiter: &BusHandlerLimiter,
) -> BusReply {
let request_count = requests.len();
let request_kinds = requests.iter().map(BusRequest::kind).collect::<Vec<_>>();
let can_fail_fast = requests.iter().all(BusRequest::can_fail_fast_on_timeout);
let Some(permit) = limiter.try_acquire() else {
warn!(
source = request_source,
request_count,
request_kinds = %request_kind_summary(&request_kinds),
active_handlers = limiter.active_handlers(),
handler_limit = limiter.limit,
"Bus request handler capacity exhausted"
);
return busy_reply(request_count);
};
let start = Instant::now();
let mut handler = spawn_blocking(move || {
let _permit = permit;
let mut response = BusReply {
responses: Vec::with_capacity(request_count),
};
@@ -171,7 +237,9 @@ fn handle_bus_result(
/// Implements a Tokio-friendly server using Unix Sockets and the bus protocol.
/// Requests are handled and then forwarded to the handler.
pub struct UnixSocketServer {}
pub struct UnixSocketServer {
handler_limiter: BusHandlerLimiter,
}
impl UnixSocketServer {
/// Creates a new `UnixSocketServer`. Will delete any pre-existing
@@ -180,7 +248,9 @@ impl UnixSocketServer {
Self::delete_local_socket()?;
Self::check_directory()?;
Self::path_permissions()?;
Ok(Self {})
Ok(Self {
handler_limiter: BusHandlerLimiter::new(MAX_CONCURRENT_BUS_HANDLERS),
})
}
/// We can't guaranty that Drop will be called on a process exit
@@ -266,6 +336,7 @@ impl UnixSocketServer {
handle_bus_requests,
vec![msg],
"internal_channel",
&self.handler_limiter,
).await;
if let Err(reply) = reply_channel.send(response) {
warn!(
@@ -284,6 +355,7 @@ impl UnixSocketServer {
}
return Err(UnixSocketServerError::ListenFail);
};
let handler_limiter = self.handler_limiter.clone();
tokio::spawn(async move {
// Listen for the magic number
let mut magic_buf = [0; 4];
@@ -338,6 +410,7 @@ impl UnixSocketServer {
handle_bus_requests,
request.requests,
"unix_socket",
&handler_limiter,
)
.await;
@@ -393,12 +466,59 @@ pub enum UnixSocketServerError {
#[cfg(test)]
mod tests {
use super::{
dropped_reply_response_count, handle_requests_with_deadline_for_duration,
BusHandlerLimiter, MAX_CONCURRENT_BUS_HANDLERS, dropped_reply_response_count,
handle_requests_with_deadline_for_duration, handle_requests_with_limiter_for_duration,
request_kind_summary,
};
use crate::{BusReply, BusRequest, BusResponse};
use std::sync::{
Barrier, OnceLock,
atomic::{AtomicUsize, Ordering},
};
use std::time::Duration;
static TIMEOUT_HANDLER_INVOCATIONS: AtomicUsize = AtomicUsize::new(0);
static BURST_HANDLER_INVOCATIONS: AtomicUsize = AtomicUsize::new(0);
static BURST_ACTIVE_HANDLERS: AtomicUsize = AtomicUsize::new(0);
static BURST_MAX_ACTIVE_HANDLERS: AtomicUsize = AtomicUsize::new(0);
fn timeout_handler_started() -> &'static Barrier {
static BARRIER: OnceLock<Barrier> = OnceLock::new();
BARRIER.get_or_init(|| Barrier::new(2))
}
fn timeout_handler_release() -> &'static Barrier {
static BARRIER: OnceLock<Barrier> = OnceLock::new();
BARRIER.get_or_init(|| Barrier::new(2))
}
fn burst_handlers_started() -> &'static Barrier {
static BARRIER: OnceLock<Barrier> = OnceLock::new();
BARRIER.get_or_init(|| Barrier::new(MAX_CONCURRENT_BUS_HANDLERS + 1))
}
fn burst_handlers_release() -> &'static Barrier {
static BARRIER: OnceLock<Barrier> = OnceLock::new();
BARRIER.get_or_init(|| Barrier::new(MAX_CONCURRENT_BUS_HANDLERS + 1))
}
fn coordinated_timeout_handler(_requests: &[BusRequest], responses: &mut Vec<BusResponse>) {
TIMEOUT_HANDLER_INVOCATIONS.fetch_add(1, Ordering::SeqCst);
timeout_handler_started().wait();
timeout_handler_release().wait();
responses.push(BusResponse::Ack);
}
fn coordinated_burst_handler(_requests: &[BusRequest], responses: &mut Vec<BusResponse>) {
BURST_HANDLER_INVOCATIONS.fetch_add(1, Ordering::SeqCst);
let active = BURST_ACTIVE_HANDLERS.fetch_add(1, Ordering::SeqCst) + 1;
BURST_MAX_ACTIVE_HANDLERS.fetch_max(active, Ordering::SeqCst);
burst_handlers_started().wait();
burst_handlers_release().wait();
BURST_ACTIVE_HANDLERS.fetch_sub(1, Ordering::SeqCst);
responses.push(BusResponse::Ack);
}
#[test]
fn dropped_reply_summary_only_counts_responses() {
let reply = BusReply {
@@ -464,6 +584,133 @@ mod tests {
);
}
#[tokio::test]
async fn timed_out_handler_retains_capacity_until_it_finishes() {
fn fast_handler(_requests: &[BusRequest], responses: &mut Vec<BusResponse>) {
responses.push(BusResponse::Ack);
}
TIMEOUT_HANDLER_INVOCATIONS.store(0, Ordering::SeqCst);
let limiter = BusHandlerLimiter::new(1);
let timed_limiter = limiter.clone();
let timed_handler = tokio::spawn(async move {
handle_requests_with_limiter_for_duration(
coordinated_timeout_handler,
vec![BusRequest::Ping],
"test",
Duration::from_millis(10),
&timed_limiter,
)
.await
});
tokio::task::spawn_blocking(|| timeout_handler_started().wait())
.await
.expect("test coordinator should join");
let timed_out = timed_handler.await.expect("timed handler should join");
assert_eq!(
timed_out.responses,
vec![BusResponse::Fail(
"Bus request handler timed out".to_string()
)]
);
let rejected = handle_requests_with_limiter_for_duration(
fast_handler,
vec![BusRequest::Ping],
"test",
Duration::from_millis(5),
&limiter,
)
.await;
assert_eq!(
rejected.responses,
vec![BusResponse::Fail("Bus request handler busy".to_string())]
);
assert_eq!(TIMEOUT_HANDLER_INVOCATIONS.load(Ordering::SeqCst), 1);
tokio::task::spawn_blocking(|| timeout_handler_release().wait())
.await
.expect("test coordinator should join");
tokio::time::timeout(Duration::from_secs(1), async {
while limiter.active_handlers() != 0 {
tokio::task::yield_now().await;
}
})
.await
.expect("timed-out handler should eventually release capacity");
let recovered = handle_requests_with_limiter_for_duration(
fast_handler,
vec![BusRequest::Ping],
"test",
Duration::from_millis(20),
&limiter,
)
.await;
assert_eq!(recovered.responses, vec![BusResponse::Ack]);
}
#[tokio::test]
async fn bus_handler_limiter_caps_running_handlers_at_configured_limit() {
BURST_HANDLER_INVOCATIONS.store(0, Ordering::SeqCst);
BURST_ACTIVE_HANDLERS.store(0, Ordering::SeqCst);
BURST_MAX_ACTIVE_HANDLERS.store(0, Ordering::SeqCst);
let limiter = BusHandlerLimiter::new(MAX_CONCURRENT_BUS_HANDLERS);
let mut handlers = Vec::with_capacity(MAX_CONCURRENT_BUS_HANDLERS);
for _ in 0..MAX_CONCURRENT_BUS_HANDLERS {
let handler_limiter = limiter.clone();
handlers.push(tokio::spawn(async move {
handle_requests_with_limiter_for_duration(
coordinated_burst_handler,
vec![BusRequest::Ping],
"test",
Duration::from_secs(5),
&handler_limiter,
)
.await
}));
}
tokio::task::spawn_blocking(|| burst_handlers_started().wait())
.await
.expect("test coordinator should join");
assert_eq!(limiter.active_handlers(), MAX_CONCURRENT_BUS_HANDLERS);
assert_eq!(
BURST_HANDLER_INVOCATIONS.load(Ordering::SeqCst),
MAX_CONCURRENT_BUS_HANDLERS
);
let rejected = handle_requests_with_limiter_for_duration(
coordinated_burst_handler,
vec![BusRequest::Ping],
"test",
Duration::from_millis(10),
&limiter,
)
.await;
assert_eq!(
rejected.responses,
vec![BusResponse::Fail("Bus request handler busy".to_string())]
);
assert_eq!(
BURST_HANDLER_INVOCATIONS.load(Ordering::SeqCst),
MAX_CONCURRENT_BUS_HANDLERS
);
tokio::task::spawn_blocking(|| burst_handlers_release().wait())
.await
.expect("test coordinator should join");
for handler in handlers {
let response = handler.await.expect("burst handler should join");
assert_eq!(response.responses, vec![BusResponse::Ack]);
}
assert_eq!(limiter.active_handlers(), 0);
assert_eq!(BURST_ACTIVE_HANDLERS.load(Ordering::SeqCst), 0);
assert_eq!(
BURST_MAX_ACTIVE_HANDLERS.load(Ordering::SeqCst),
MAX_CONCURRENT_BUS_HANDLERS
);
}
#[tokio::test]
async fn side_effecting_request_waits_for_handler_after_deadline() {
fn slow_mutating_handler(_requests: &[BusRequest], responses: &mut Vec<BusResponse>) {
src/rust/lqos_network_devices/src/lib.rs
+4 -1
View File
@@ -31,7 +31,10 @@ pub use catalog::{CircuitRateCaps, ShapedDevicesCatalog};
pub use combined_catalog::NetworkDevicesCatalog;
pub use dynamic::{CircuitObservation, DynamicCircuit};
pub use hash_cache::ShapedDeviceHashCache;
pub use topology::{ResolvedParentNode, resolve_parent_node, resolve_parent_node_reference};
pub use topology::{
ParentNodeLookup, ResolvedParentNode, resolve_parent_node, resolve_parent_node_reference,
resolve_parent_node_reference_in_nodes,
};
/// Provenance for caller-provided shaped-device snapshots.
#[derive(Clone, Copy, Debug, Eq, PartialEq)]
src/rust/lqos_network_devices/src/tests.rs
+24 -3
View File
@@ -1,7 +1,7 @@
use crate::{
DynamicCircuit, ShapedDevicesCatalog, load_shaped_devices_for_config,
load_source_shaped_devices_for_config, resolve_parent_node_reference, runtime_inputs,
with_network_json_write,
DynamicCircuit, ParentNodeLookup, ShapedDevicesCatalog, load_shaped_devices_for_config,
load_source_shaped_devices_for_config, resolve_parent_node_reference,
resolve_parent_node_reference_in_nodes, runtime_inputs, with_network_json_write,
};
use lqos_config::{
CircuitAnchorsFile, Config, ConfigShapedDevices, NetworkJsonNode, ShapedDevice,
@@ -162,6 +162,27 @@ fn resolve_parent_node_reference_prefers_id_then_name_then_alias() {
assert_eq!(by_alias.name, "Site B");
}
#[test]
fn borrowed_parent_node_resolver_uses_the_same_precedence() {
let nodes = vec![
make_node("Root", Some("root"), None),
make_node("Site A", Some("node-a"), Some("shared")),
make_node("shared", Some("node-b"), None),
];
let by_id = resolve_parent_node_reference_in_nodes(&nodes, "shared", Some("node-a"))
.expect("node id should take precedence");
assert_eq!(by_id.name, "Site A");
let by_name = resolve_parent_node_reference_in_nodes(&nodes, "shared", None)
.expect("canonical name should take precedence over an alias");
assert_eq!(by_name.id.as_deref(), Some("node-b"));
let lookup = ParentNodeLookup::from_nodes(&nodes);
assert_eq!(lookup.resolve("shared", Some("node-a")), Some(by_id));
assert_eq!(lookup.resolve("shared", None), Some(by_name));
}
#[test]
fn dynamic_circuit_last_seen_updates_for_seen_hashes() {
let _guard = TEST_LOCK.lock();
src/rust/lqos_network_devices/src/topology.rs
+97 -33
View File
@@ -1,5 +1,6 @@
use crate::state;
use lqos_config::NetworkJsonNode;
use std::collections::HashMap;
/// Canonical parent-node metadata resolved from `network.json`.
#[derive(Clone, Debug, PartialEq, Eq)]
@@ -10,49 +11,86 @@ pub struct ResolvedParentNode {
pub id: Option<String>,
}
/// Borrowed parent-node indexes for resolving a batch against one node snapshot.
#[derive(Clone, Debug, Default)]
pub struct ParentNodeLookup<'a> {
by_id: HashMap<&'a str, &'a NetworkJsonNode>,
by_name: HashMap<&'a str, &'a NetworkJsonNode>,
by_alias: HashMap<&'a str, &'a NetworkJsonNode>,
}
impl<'a> ParentNodeLookup<'a> {
/// Builds ID, name, and active-attachment alias indexes from a borrowed node slice.
pub fn from_nodes(nodes: &'a [NetworkJsonNode]) -> Self {
let mut lookup = Self::default();
for node in nodes {
if let Some(id) = node.id.as_deref() {
lookup.by_id.entry(id).or_insert(node);
}
lookup.by_name.entry(node.name.as_str()).or_insert(node);
if let Some(alias) = node.active_attachment_name.as_deref() {
lookup.by_alias.entry(alias.trim()).or_insert(node);
}
}
lookup
}
/// Resolves a parent reference with ID, canonical-name, then alias precedence.
pub fn resolve(
&self,
parent_node: &str,
parent_node_id: Option<&str>,
) -> Option<ResolvedParentNode> {
resolve_parent_node_with(
parent_node,
parent_node_id,
|id| self.by_id.get(id).copied(),
|name| self.by_name.get(name).copied(),
|alias| self.by_alias.get(alias).copied(),
)
}
}
/// Resolve a shaped-device parent reference into canonical `network.json`
/// parent metadata, preferring a stable node ID when one is available.
pub fn resolve_parent_node_reference(
parent_node: &str,
parent_node_id: Option<&str>,
) -> Option<ResolvedParentNode> {
let trimmed_id = parent_node_id.map(str::trim).filter(|id| !id.is_empty());
let trimmed = parent_node.trim();
if trimmed.is_empty() && trimmed_id.is_none() {
return None;
}
state::with_network_json_read(|net_json| {
let nodes = net_json.get_nodes_when_ready();
if let Some(parent_node_id) = trimmed_id
&& let Some(node) = find_node_by_id(nodes, parent_node_id)
{
return Some(ResolvedParentNode {
name: node.name.clone(),
id: node.id.clone(),
});
}
if let Some(node) = nodes.iter().find(|node| node.name == trimmed) {
return Some(ResolvedParentNode {
name: node.name.clone(),
id: node.id.clone(),
});
}
nodes.iter().find_map(|node| {
node.active_attachment_name
.as_deref()
.filter(|alias| alias.trim() == trimmed)
.map(|_| ResolvedParentNode {
name: node.name.clone(),
id: node.id.clone(),
})
})
resolve_parent_node_reference_in_nodes(
net_json.get_nodes_when_ready(),
parent_node,
parent_node_id,
)
})
}
/// Resolve a parent reference against an already-borrowed `network.json` node slice.
///
/// This preserves the canonical ID, name, and active-attachment alias precedence without
/// acquiring the shared `network.json` lock. Callers that already hold a network snapshot can
/// therefore resolve a batch of references under one read lock.
pub fn resolve_parent_node_reference_in_nodes(
nodes: &[NetworkJsonNode],
parent_node: &str,
parent_node_id: Option<&str>,
) -> Option<ResolvedParentNode> {
resolve_parent_node_with(
parent_node,
parent_node_id,
|id| find_node_by_id(nodes, id),
|name| nodes.iter().find(|node| node.name == name),
|alias| {
nodes.iter().find(|node| {
node.active_attachment_name
.as_deref()
.is_some_and(|candidate| candidate.trim() == alias)
})
},
)
}
/// Resolve a shaped-device parent node or active attachment alias into canonical `network.json`
/// parent metadata.
pub fn resolve_parent_node(parent_node: &str) -> Option<ResolvedParentNode> {
@@ -62,3 +100,29 @@ pub fn resolve_parent_node(parent_node: &str) -> Option<ResolvedParentNode> {
fn find_node_by_id<'a>(nodes: &'a [NetworkJsonNode], id: &str) -> Option<&'a NetworkJsonNode> {
nodes.iter().find(|node| node.id.as_deref() == Some(id))
}
fn resolved_parent_node(node: &NetworkJsonNode) -> ResolvedParentNode {
ResolvedParentNode {
name: node.name.clone(),
id: node.id.clone(),
}
}
fn resolve_parent_node_with<'a>(
parent_node: &str,
parent_node_id: Option<&str>,
resolve_id: impl FnOnce(&str) -> Option<&'a NetworkJsonNode>,
resolve_name: impl FnOnce(&str) -> Option<&'a NetworkJsonNode>,
resolve_alias: impl FnOnce(&str) -> Option<&'a NetworkJsonNode>,
) -> Option<ResolvedParentNode> {
let trimmed_id = parent_node_id.map(str::trim).filter(|id| !id.is_empty());
let trimmed_name = parent_node.trim();
if trimmed_name.is_empty() && trimmed_id.is_none() {
return None;
}
trimmed_id
.and_then(resolve_id)
.or_else(|| resolve_name(trimmed_name))
.or_else(|| resolve_alias(trimmed_name))
.map(resolved_parent_node)
}
src/rust/lqosd/src/shaped_devices_tracker/mod.rs
+487 -222
View File
@@ -43,6 +43,10 @@ pub static EFFECTIVE_CIRCUIT_PARENTS: Lazy<ArcSwap<FxHashMap<String, RuntimeCirc
Lazy::new(|| ArcSwap::new(Arc::new(FxHashMap::default())));
static LAST_TOPOLOGY_STATUS_IDENTITY: Lazy<Mutex<Option<TopologyRuntimeShapingPayloadIdentity>>> =
Lazy::new(|| Mutex::new(None));
#[cfg(test)]
static CIRCUIT_SNAPSHOT_TEST_HOOK: Lazy<
Mutex<Option<(std::thread::ThreadId, std::sync::mpsc::Sender<()>)>>,
> = Lazy::new(|| Mutex::new(None));
#[derive(Clone, Debug, PartialEq, Eq)]
pub struct RuntimeCircuitParent {
@@ -684,10 +688,6 @@ pub fn map_node_names(nodes: &[usize]) -> BusResponse {
})
}
pub fn resolve_parent_node_alias(parent_node: &str) -> Option<String> {
lqos_network_devices::resolve_parent_node(parent_node).map(|resolved| resolved.name)
}
pub fn get_funnel(circuit_id: &str) -> BusResponse {
lqos_network_devices::with_network_json_read(|net_json| {
if let Some(index) = net_json.get_index_for_name(circuit_id) {
@@ -711,34 +711,49 @@ pub fn get_funnel(circuit_id: &str) -> BusResponse {
})
}
pub fn get_all_circuits() -> BusResponse {
if let Ok(kernel_now) = time_since_boot() {
let catalog = lqos_network_devices::network_devices_catalog();
let data = THROUGHPUT_TRACKER
.raw_data
.lock()
.iter()
.map(|(k, v)| {
let last_seen_nanos = if v.last_seen > 0 {
let last_seen_nanos = v.last_seen as u128;
let since_boot = Duration::from(kernel_now).as_nanos();
//println!("since_boot: {:?}, last_seen: {:?}", since_boot, last_seen_nanos);
since_boot.saturating_sub(last_seen_nanos) as u64
} else {
u64::MAX
};
struct PendingCircuitParent {
circuit: Circuit,
configured_parent: Option<String>,
}
// Map to circuit et al
let mut circuit_id = v.circuit_id.clone();
fn snapshot_circuits(desired_circuit_id: Option<&str>) -> Vec<PendingCircuitParent> {
let Ok(kernel_now) = time_since_boot() else {
return Vec::new();
};
let since_boot_nanos = Duration::from(kernel_now).as_nanos();
let desired_hash = desired_circuit_id.map(hash_to_i64);
let catalog = lqos_network_devices::network_devices_catalog();
let pending = {
let raw_data = THROUGHPUT_TRACKER.raw_data.lock();
raw_data
.iter()
.filter_map(|(ip, entry)| {
let device = catalog
.device_by_hashes(entry.device_hash, entry.circuit_hash)
.or_else(|| {
catalog
.device_longest_match_for_ip(ip)
.map(|(_, device)| device)
});
if let Some(desired_circuit_id) = desired_circuit_id {
let desired_hash = desired_hash.expect("desired hash accompanies desired id");
let matches_desired = entry.circuit_hash == Some(desired_hash)
|| entry.circuit_id.as_deref() == Some(desired_circuit_id)
|| device.is_some_and(|device| device.circuit_hash == desired_hash)
|| device.is_some_and(|device| device.circuit_id == desired_circuit_id);
if !matches_desired {
return None;
}
}
let mut circuit_id = entry.circuit_id.clone();
let mut circuit_name = None;
let mut device_id = None;
let mut device_name = None;
let mut parent_node = None;
// Plan is expressed in Mbps as f32
let mut plan: DownUpOrder<f32> = DownUpOrder { down: 0.0, up: 0.0 };
let device = catalog
.device_by_hashes(v.device_hash, v.circuit_hash)
.or_else(|| catalog.device_longest_match_for_ip(k).map(|(_, dev)| dev));
let mut configured_parent = None;
let mut plan = DownUpOrder { down: 0.0, up: 0.0 };
if let Some(device) = device {
if circuit_id.as_deref().unwrap_or_default().is_empty() {
circuit_id = Some(device.circuit_id.clone());
@@ -746,218 +761,225 @@ pub fn get_all_circuits() -> BusResponse {
circuit_name = Some(device.circuit_name.clone());
device_id = Some(device.device_id.clone());
device_name = Some(device.device_name.clone());
parent_node = Some(
effective_parent_for_circuit(&device.circuit_id)
.map(|parent| parent.name)
.or_else(|| resolve_parent_node_alias(&device.parent_node))
.unwrap_or_else(|| device.parent_node.clone()),
);
if let Some(effective_parent) = effective_parent_for_circuit(&device.circuit_id)
{
parent_node = Some(effective_parent.name);
} else {
configured_parent = Some(device.parent_node.clone());
}
plan.down = device.download_max_mbps.round();
plan.up = device.upload_max_mbps.round();
}
Circuit {
ip: k.as_ip(),
bytes_per_second: v.bytes_per_second,
actual_bytes_per_second: v.actual_bytes_per_second,
median_latency: v.median_latency(),
rtt_current_p50_nanos: DownUpOrder {
down: v
.rtt_buffer
.percentile(RttBucket::Current, FlowbeeEffectiveDirection::Download, 50)
.map(|rtt| rtt.as_nanos()),
up: v
.rtt_buffer
.percentile(RttBucket::Current, FlowbeeEffectiveDirection::Upload, 50)
.map(|rtt| rtt.as_nanos()),
},
rtt_current_p95_nanos: DownUpOrder {
down: v
.rtt_buffer
.percentile(RttBucket::Current, FlowbeeEffectiveDirection::Download, 95)
.map(|rtt| rtt.as_nanos()),
up: v
.rtt_buffer
.percentile(RttBucket::Current, FlowbeeEffectiveDirection::Upload, 95)
.map(|rtt| rtt.as_nanos()),
},
rtt_total_p50_nanos: DownUpOrder {
down: v
.rtt_buffer
.percentile(RttBucket::Total, FlowbeeEffectiveDirection::Download, 50)
.map(|rtt| rtt.as_nanos()),
up: v
.rtt_buffer
.percentile(RttBucket::Total, FlowbeeEffectiveDirection::Upload, 50)
.map(|rtt| rtt.as_nanos()),
},
rtt_total_p95_nanos: DownUpOrder {
down: v
.rtt_buffer
.percentile(RttBucket::Total, FlowbeeEffectiveDirection::Download, 95)
.map(|rtt| rtt.as_nanos()),
up: v
.rtt_buffer
.percentile(RttBucket::Total, FlowbeeEffectiveDirection::Upload, 95)
.map(|rtt| rtt.as_nanos()),
},
qoo: DownUpOrder {
down: v.qoq.download_total_f32(),
up: v.qoq.upload_total_f32(),
},
tcp_retransmit_sample: down_up_retransmit_sample(
v.tcp_retransmits,
v.tcp_retransmit_packets,
),
circuit_id,
device_id,
circuit_name,
device_name,
parent_node,
plan,
last_seen_nanos,
if circuit_id.is_none() {
circuit_id = desired_circuit_id.map(str::to_string);
}
let last_seen_nanos = if entry.last_seen > 0 {
since_boot_nanos.saturating_sub(entry.last_seen as u128) as u64
} else {
u64::MAX
};
let percentile = |bucket, direction, percentile| {
entry
.rtt_buffer
.percentile(bucket, direction, percentile)
.map(|rtt| rtt.as_nanos())
};
Some(PendingCircuitParent {
circuit: Circuit {
ip: ip.as_ip(),
bytes_per_second: entry.bytes_per_second,
actual_bytes_per_second: entry.actual_bytes_per_second,
median_latency: entry.median_latency(),
rtt_current_p50_nanos: DownUpOrder {
down: percentile(
RttBucket::Current,
FlowbeeEffectiveDirection::Download,
50,
),
up: percentile(
RttBucket::Current,
FlowbeeEffectiveDirection::Upload,
50,
),
},
rtt_current_p95_nanos: DownUpOrder {
down: percentile(
RttBucket::Current,
FlowbeeEffectiveDirection::Download,
95,
),
up: percentile(
RttBucket::Current,
FlowbeeEffectiveDirection::Upload,
95,
),
},
rtt_total_p50_nanos: DownUpOrder {
down: percentile(
RttBucket::Total,
FlowbeeEffectiveDirection::Download,
50,
),
up: percentile(RttBucket::Total, FlowbeeEffectiveDirection::Upload, 50),
},
rtt_total_p95_nanos: DownUpOrder {
down: percentile(
RttBucket::Total,
FlowbeeEffectiveDirection::Download,
95,
),
up: percentile(RttBucket::Total, FlowbeeEffectiveDirection::Upload, 95),
},
qoo: DownUpOrder {
down: entry.qoq.download_total_f32(),
up: entry.qoq.upload_total_f32(),
},
tcp_retransmit_sample: down_up_retransmit_sample(
entry.tcp_retransmits,
entry.tcp_retransmit_packets,
),
circuit_id,
device_id,
circuit_name,
device_name,
parent_node,
plan,
last_seen_nanos,
},
configured_parent,
})
})
.collect();
BusResponse::CircuitData(data)
} else {
BusResponse::CircuitData(Vec::new())
.collect()
};
#[cfg(test)]
{
let mut hook = CIRCUIT_SNAPSHOT_TEST_HOOK.lock();
if hook
.as_ref()
.is_some_and(|(thread_id, _)| *thread_id == std::thread::current().id())
&& let Some((_, sender)) = hook.take()
{
let _ = sender.send(());
}
}
pending
}
fn resolve_pending_circuit_parents_with(
pending: Vec<PendingCircuitParent>,
mut resolve_parent: impl FnMut(&str) -> Option<String>,
) -> Vec<Circuit> {
let mut resolved_parents: FxHashMap<String, String> = FxHashMap::default();
pending
.into_iter()
.map(|mut pending| {
if let Some(configured_parent) = pending.configured_parent {
let resolved_parent =
if let Some(resolved) = resolved_parents.get(&configured_parent) {
resolved.clone()
} else {
let resolved = resolve_parent(&configured_parent)
.unwrap_or_else(|| configured_parent.clone());
resolved_parents.insert(configured_parent, resolved.clone());
resolved
};
pending.circuit.parent_node = Some(resolved_parent);
}
pending.circuit
})
.collect()
}
fn resolve_pending_circuit_parents(pending: Vec<PendingCircuitParent>) -> Vec<Circuit> {
if pending
.iter()
.all(|pending| pending.configured_parent.is_none())
{
return pending.into_iter().map(|pending| pending.circuit).collect();
}
let configured_parents = pending
.iter()
.filter_map(|pending| pending.configured_parent.as_deref())
.collect::<FxHashSet<_>>();
let resolved_parents = lqos_network_devices::with_network_json_read(|network_json| {
let lookup =
lqos_network_devices::ParentNodeLookup::from_nodes(network_json.get_nodes_when_ready());
configured_parents
.into_iter()
.filter_map(|parent| {
lookup
.resolve(parent, None)
.map(|resolved| (parent.to_string(), resolved.name))
})
.collect::<FxHashMap<_, _>>()
});
resolve_pending_circuit_parents_with(pending, |parent| resolved_parents.get(parent).cloned())
}
pub fn get_all_circuits() -> BusResponse {
BusResponse::CircuitData(resolve_pending_circuit_parents(snapshot_circuits(None)))
}
pub fn get_circuit_by_id(desired_circuit_id: String) -> BusResponse {
if let Ok(kernel_now) = time_since_boot() {
let desired_hash = hash_to_i64(&desired_circuit_id);
let catalog = lqos_network_devices::network_devices_catalog();
let data = THROUGHPUT_TRACKER
.raw_data
.lock()
.iter()
.filter_map(|(k, v)| {
let device = catalog
.device_by_hashes(v.device_hash, v.circuit_hash)
.or_else(|| catalog.device_longest_match_for_ip(k).map(|(_, dev)| dev));
let matches_desired = v.circuit_hash == Some(desired_hash)
|| v.circuit_id.as_deref() == Some(desired_circuit_id.as_str())
|| device.is_some_and(|device| device.circuit_hash == desired_hash)
|| device.is_some_and(|device| device.circuit_id == desired_circuit_id);
if !matches_desired {
return None;
}
let last_seen_nanos = if v.last_seen > 0 {
let last_seen_nanos = v.last_seen as u128;
let since_boot = Duration::from(kernel_now).as_nanos();
//println!("since_boot: {:?}, last_seen: {:?}", since_boot, last_seen_nanos);
since_boot.saturating_sub(last_seen_nanos) as u64
} else {
u64::MAX
};
// Map to circuit et al
let mut circuit_id = v.circuit_id.clone();
let mut circuit_name = None;
let mut device_id = None;
let mut device_name = None;
let mut parent_node = None;
// Plan is expressed in Mbps as f32
let mut plan: DownUpOrder<f32> = DownUpOrder { down: 0.0, up: 0.0 };
if let Some(device) = device {
if circuit_id.as_deref().unwrap_or_default().is_empty() {
circuit_id = Some(device.circuit_id.clone());
}
circuit_name = Some(device.circuit_name.clone());
device_id = Some(device.device_id.clone());
device_name = Some(device.device_name.clone());
parent_node = Some(
effective_parent_for_circuit(&device.circuit_id)
.map(|parent| parent.name)
.or_else(|| resolve_parent_node_alias(&device.parent_node))
.unwrap_or_else(|| device.parent_node.clone()),
);
plan.down = device.download_max_mbps.round();
plan.up = device.upload_max_mbps.round();
}
let circuit_id = Some(circuit_id.unwrap_or_else(|| desired_circuit_id.clone()));
Some(Circuit {
ip: k.as_ip(),
bytes_per_second: v.bytes_per_second,
actual_bytes_per_second: v.actual_bytes_per_second,
median_latency: v.median_latency(),
rtt_current_p50_nanos: DownUpOrder {
down: v
.rtt_buffer
.percentile(RttBucket::Current, FlowbeeEffectiveDirection::Download, 50)
.map(|rtt| rtt.as_nanos()),
up: v
.rtt_buffer
.percentile(RttBucket::Current, FlowbeeEffectiveDirection::Upload, 50)
.map(|rtt| rtt.as_nanos()),
},
rtt_current_p95_nanos: DownUpOrder {
down: v
.rtt_buffer
.percentile(RttBucket::Current, FlowbeeEffectiveDirection::Download, 95)
.map(|rtt| rtt.as_nanos()),
up: v
.rtt_buffer
.percentile(RttBucket::Current, FlowbeeEffectiveDirection::Upload, 95)
.map(|rtt| rtt.as_nanos()),
},
rtt_total_p50_nanos: DownUpOrder {
down: v
.rtt_buffer
.percentile(RttBucket::Total, FlowbeeEffectiveDirection::Download, 50)
.map(|rtt| rtt.as_nanos()),
up: v
.rtt_buffer
.percentile(RttBucket::Total, FlowbeeEffectiveDirection::Upload, 50)
.map(|rtt| rtt.as_nanos()),
},
rtt_total_p95_nanos: DownUpOrder {
down: v
.rtt_buffer
.percentile(RttBucket::Total, FlowbeeEffectiveDirection::Download, 95)
.map(|rtt| rtt.as_nanos()),
up: v
.rtt_buffer
.percentile(RttBucket::Total, FlowbeeEffectiveDirection::Upload, 95)
.map(|rtt| rtt.as_nanos()),
},
qoo: DownUpOrder {
down: v.qoq.download_total_f32(),
up: v.qoq.upload_total_f32(),
},
tcp_retransmit_sample: down_up_retransmit_sample(
v.tcp_retransmits,
v.tcp_retransmit_packets,
),
circuit_id,
device_id,
circuit_name,
device_name,
parent_node,
plan,
last_seen_nanos,
})
})
.collect();
BusResponse::CircuitData(data)
} else {
BusResponse::CircuitData(Vec::new())
}
BusResponse::CircuitData(resolve_pending_circuit_parents(snapshot_circuits(Some(
&desired_circuit_id,
))))
}
#[cfg(test)]
mod tests {
use super::*;
use lqos_config::{
Config, TOPOLOGY_RUNTIME_STATUS_FILENAME, TopologyShapingCircuitInput,
TopologyShapingDeviceInput, TopologyShapingInputsFile,
use crate::test_support::runtime_config_test_lock;
use crate::throughput_tracker::{
RawThroughputTestEntry, RttBuffer, replace_raw_throughput_for_test,
};
use lqos_config::{
Config, ConfigShapedDevices, ShapedDevice, TOPOLOGY_RUNTIME_STATUS_FILENAME,
TopologyShapingCircuitInput, TopologyShapingDeviceInput, TopologyShapingInputsFile,
};
use lqos_utils::XdpIpAddress;
use std::cell::Cell;
use std::fs;
use std::path::PathBuf;
use std::time::{SystemTime, UNIX_EPOCH};
use std::sync::{Arc, mpsc};
use std::thread;
use std::time::{Duration, SystemTime, UNIX_EPOCH};
struct CircuitLookupStateGuard {
old_shaped_devices: Option<Arc<ConfigShapedDevices>>,
old_network_nodes: Option<Vec<NetworkJsonNode>>,
old_effective_parents: Option<Arc<FxHashMap<String, RuntimeCircuitParent>>>,
}
struct CircuitSnapshotTestHookGuard;
impl Drop for CircuitSnapshotTestHookGuard {
fn drop(&mut self) {
CIRCUIT_SNAPSHOT_TEST_HOOK.lock().take();
}
}
impl Drop for CircuitLookupStateGuard {
fn drop(&mut self) {
if let Some(nodes) = self.old_network_nodes.take() {
lqos_network_devices::with_network_json_write(|network_json| {
network_json.nodes = nodes;
});
}
if let Some(shaped_devices) = self.old_shaped_devices.take() {
lqos_network_devices::swap_shaped_devices_snapshot(
"circuit-lookup-test-restore",
shaped_devices,
);
}
if let Some(effective_parents) = self.old_effective_parents.take() {
EFFECTIVE_CIRCUIT_PARENTS.store(effective_parents);
}
}
}
fn unique_temp_dir(prefix: &str) -> PathBuf {
let unique = SystemTime::now()
@@ -991,6 +1013,249 @@ mod tests {
.expect("status should write");
}
fn pending_circuit(
parent_node: Option<&str>,
configured_parent: Option<&str>,
) -> PendingCircuitParent {
PendingCircuitParent {
circuit: Circuit {
ip: "192.0.2.1".parse().expect("test IP should parse"),
bytes_per_second: DownUpOrder::default(),
actual_bytes_per_second: DownUpOrder::default(),
median_latency: None,
rtt_current_p50_nanos: DownUpOrder::default(),
rtt_current_p95_nanos: DownUpOrder::default(),
rtt_total_p50_nanos: DownUpOrder::default(),
rtt_total_p95_nanos: DownUpOrder::default(),
qoo: DownUpOrder::default(),
tcp_retransmit_sample: DownUpOrder::default(),
circuit_id: Some("test-circuit".to_string()),
device_id: Some("test-device".to_string()),
parent_node: parent_node.map(str::to_string),
circuit_name: Some("Test Circuit".to_string()),
device_name: Some("Test Device".to_string()),
plan: DownUpOrder::default(),
last_seen_nanos: 0,
},
configured_parent: configured_parent.map(str::to_string),
}
}
#[test]
fn pending_circuit_parent_resolution_is_cached_and_fallback_safe() {
let pending = vec![
pending_circuit(None, Some("Tower Alias")),
pending_circuit(None, Some("Tower Alias")),
pending_circuit(None, Some("Unknown Parent")),
pending_circuit(Some("Effective Parent"), None),
];
let resolution_calls = Cell::new(0usize);
let circuits = resolve_pending_circuit_parents_with(pending, |parent| {
resolution_calls.set(resolution_calls.get() + 1);
(parent == "Tower Alias").then(|| "Canonical Tower".to_string())
});
assert_eq!(resolution_calls.get(), 2);
assert_eq!(circuits[0].parent_node.as_deref(), Some("Canonical Tower"));
assert_eq!(circuits[1].parent_node.as_deref(), Some("Canonical Tower"));
assert_eq!(circuits[2].parent_node.as_deref(), Some("Unknown Parent"));
assert_eq!(circuits[3].parent_node.as_deref(), Some("Effective Parent"));
}
#[test]
fn circuit_bus_readers_release_raw_data_before_resolving_parents() {
let _runtime_guard = runtime_config_test_lock()
.lock()
.expect("runtime config test lock should not be poisoned");
let effective_circuit_id = "deadlock-effective-circuit";
let effective_device_id = "deadlock-effective-device";
let alias_circuit_id = "deadlock-alias-circuit";
let alias_device_id = "deadlock-alias-device";
let effective_circuit_hash = hash_to_i64(effective_circuit_id);
let effective_device_hash = hash_to_i64(effective_device_id);
let alias_circuit_hash = hash_to_i64(alias_circuit_id);
let alias_device_hash = hash_to_i64(alias_device_id);
let mut shaped_devices = ConfigShapedDevices::default();
shaped_devices.replace_with_new_data(vec![
ShapedDevice {
circuit_id: effective_circuit_id.to_string(),
circuit_name: "Effective Parent Circuit".to_string(),
device_id: effective_device_id.to_string(),
device_name: "Effective Parent Device".to_string(),
parent_node: "Tower Alias".to_string(),
circuit_hash: effective_circuit_hash,
device_hash: effective_device_hash,
download_max_mbps: 100.0,
upload_max_mbps: 20.0,
..ShapedDevice::default()
},
ShapedDevice {
circuit_id: alias_circuit_id.to_string(),
circuit_name: "Alias Parent Circuit".to_string(),
device_id: alias_device_id.to_string(),
device_name: "Alias Parent Device".to_string(),
parent_node: "Tower Alias".to_string(),
circuit_hash: alias_circuit_hash,
device_hash: alias_device_hash,
download_max_mbps: 200.0,
upload_max_mbps: 40.0,
..ShapedDevice::default()
},
]);
let old_shaped_devices = lqos_network_devices::swap_shaped_devices_snapshot(
"circuit-lookup-test",
Arc::new(shaped_devices),
);
let old_network_nodes =
lqos_network_devices::with_network_json_read(|network_json| network_json.nodes.clone());
let mut effective_parents = FxHashMap::default();
effective_parents.insert(
effective_circuit_id.to_string(),
RuntimeCircuitParent {
name: "Effective Tower".to_string(),
id: Some("effective-tower-id".to_string()),
},
);
let old_effective_parents = EFFECTIVE_CIRCUIT_PARENTS.swap(Arc::new(effective_parents));
let _state_guard = CircuitLookupStateGuard {
old_shaped_devices: Some(old_shaped_devices),
old_network_nodes: Some(old_network_nodes),
old_effective_parents: Some(old_effective_parents),
};
lqos_network_devices::with_network_json_write(|network_json| {
network_json.nodes = vec![NetworkJsonNode {
name: "Canonical Tower".to_string(),
id: Some("canonical-tower-id".to_string()),
virtual_node: false,
max_throughput: (0.0, 0.0),
current_throughput: DownUpOrder::default(),
current_packets: DownUpOrder::default(),
current_tcp_packets: DownUpOrder::default(),
current_udp_packets: DownUpOrder::default(),
current_icmp_packets: DownUpOrder::default(),
current_tcp_retransmits: DownUpOrder::default(),
current_tcp_retransmit_packets: DownUpOrder::default(),
current_marks: DownUpOrder::default(),
current_drops: DownUpOrder::default(),
rtt_buffer: RttBuffer::default(),
parents: Vec::new(),
immediate_parent: None,
node_type: None,
latitude: None,
longitude: None,
active_attachment_name: Some("Tower Alias".to_string()),
heatmap: None,
qoq_heatmap: None,
}];
});
let _raw_guard = replace_raw_throughput_for_test(
1,
vec![
RawThroughputTestEntry {
ip: XdpIpAddress::from_ip("192.0.2.10".parse().expect("test IP should parse")),
circuit_hash: Some(effective_circuit_hash),
device_hash: Some(effective_device_hash),
most_recent_cycle: 1,
bytes_per_second: DownUpOrder::new(1_000, 200),
tcp_packets: DownUpOrder::default(),
tcp_retransmits: DownUpOrder::default(),
},
RawThroughputTestEntry {
ip: XdpIpAddress::from_ip("192.0.2.11".parse().expect("test IP should parse")),
circuit_hash: Some(alias_circuit_hash),
device_hash: Some(alias_device_hash),
most_recent_cycle: 1,
bytes_per_second: DownUpOrder::new(2_000, 400),
tcp_packets: DownUpOrder::default(),
tcp_retransmits: DownUpOrder::default(),
},
],
);
let (network_locked_tx, network_locked_rx) = mpsc::channel();
let (release_network_tx, release_network_rx) = mpsc::channel();
let network_writer = thread::spawn(move || {
lqos_network_devices::with_network_json_write(|_| {
network_locked_tx
.send(())
.expect("test should signal network lock acquisition");
let _ = release_network_rx.recv_timeout(Duration::from_secs(2));
});
});
network_locked_rx
.recv_timeout(Duration::from_secs(1))
.expect("network writer should acquire the lock");
let (start_reader_tx, start_reader_rx) = mpsc::channel();
let (snapshot_complete_tx, snapshot_complete_rx) = mpsc::channel();
let circuit_reader = thread::spawn(move || {
start_reader_rx
.recv()
.expect("test should start the circuit reader");
get_all_circuits()
});
*CIRCUIT_SNAPSHOT_TEST_HOOK.lock() =
Some((circuit_reader.thread().id(), snapshot_complete_tx));
let _hook_guard = CircuitSnapshotTestHookGuard;
start_reader_tx
.send(())
.expect("test should start the circuit reader");
snapshot_complete_rx
.recv_timeout(Duration::from_secs(1))
.expect("circuit reader should finish its raw-data snapshot");
let raw_data_was_released = THROUGHPUT_TRACKER.raw_data.try_lock().is_some();
release_network_tx
.send(())
.expect("test should release the network writer");
network_writer
.join()
.expect("network writer should finish cleanly");
let BusResponse::CircuitData(all_circuits) = circuit_reader
.join()
.expect("circuit reader should finish cleanly")
else {
panic!("GetAllCircuits should return circuit data");
};
assert!(
raw_data_was_released,
"raw_data must be unlocked before waiting for network.json"
);
assert_eq!(all_circuits.len(), 2);
let effective_circuit = all_circuits
.iter()
.find(|circuit| circuit.circuit_id.as_deref() == Some(effective_circuit_id))
.expect("effective-parent circuit should be present");
assert_eq!(
effective_circuit.parent_node.as_deref(),
Some("Effective Tower")
);
let alias_circuit = all_circuits
.iter()
.find(|circuit| circuit.circuit_id.as_deref() == Some(alias_circuit_id))
.expect("alias-parent circuit should be present");
assert_eq!(
alias_circuit.parent_node.as_deref(),
Some("Canonical Tower")
);
let BusResponse::CircuitData(selected_circuits) =
get_circuit_by_id(alias_circuit_id.to_string())
else {
panic!("GetCircuitById should return circuit data");
};
assert_eq!(selected_circuits.len(), 1);
assert_eq!(
selected_circuits[0].circuit_id.as_deref(),
Some(alias_circuit_id)
);
assert_eq!(
selected_circuits[0].parent_node.as_deref(),
Some("Canonical Tower")
);
}
#[test]
fn effective_circuit_parent_map_uses_effective_parent_fields() {
let shaping_inputs = TopologyShapingInputsFile {