use std::sync::atomic::Ordering;
use crate::config::StatsLevel;
use super::{EdgeId, GraphInner, InputQueueStatsSnapshot, NodeStatsSnapshot};
impl GraphInner {
pub(super) fn queue_depth(&self, edge: EdgeId) -> usize {
self.edges[edge]
.consumers
.iter()
.map(|&(n, p)| self.nodes[n].queue_len(p))
.sum()
}
pub(super) fn node_stats(&self, i: usize) -> Option<NodeStatsSnapshot> {
let node = self.nodes.get(i)?;
let st = &node.stats;
let running = st.in_flight.load(Ordering::Relaxed) > 0;
let running_for_us = if running && self.stats_level == StatsLevel::Full {
let now_us = self.epoch.elapsed().as_micros() as i64;
(now_us - st.started_us.load(Ordering::Relaxed)).max(0)
} else {
0
};
Some(NodeStatsSnapshot {
node_name: node.name.clone(),
kernel_name: node.kernel_name.clone(),
running,
running_for_us,
processed: st.processed.load(Ordering::Relaxed),
errors: st.errors.load(Ordering::Relaxed),
total_process_us: st.total_us.load(Ordering::Relaxed),
max_process_us: st.max_us.load(Ordering::Relaxed),
packets_in: st.packets_in.load(Ordering::Relaxed),
packets_out: st.packets_out.load(Ordering::Relaxed),
peak_queue_depth: st.peak_queue_depth.load(Ordering::Relaxed),
queued: (0..node.input_queues.len())
.map(|p| node.queue_len(p))
.sum(),
})
}
pub(super) fn input_queue_stats(
&self,
node_id: usize,
port: usize,
) -> Option<InputQueueStatsSnapshot> {
self.input_queue_stats_at(node_id, port, self.epoch_us())
}
pub(super) fn input_queue_stats_at(
&self,
node_id: usize,
port: usize,
now_us: i64,
) -> Option<InputQueueStatsSnapshot> {
let node = self.nodes.get(node_id)?;
let port_name = node.in_ports.name(port)?.to_string();
let stats = node.input_queue_stats.get(port)?;
let queued_packets = node.queue_len(port);
let queued_bytes = node.input_queue_bytes[port].load(Ordering::SeqCst);
let reserved_packets = node.input_queue_reserved[port].load(Ordering::SeqCst);
let since = stats.blocked_since_us.load(Ordering::SeqCst);
let blocked_for_us = if since == 0 {
0
} else {
now_us.saturating_sub(since.saturating_sub(1)).max(0) as u64
};
let edge = node.inputs[port];
let producer_name = self.edges[edge]
.producer
.map(|producer| self.nodes[producer].name.clone());
Some(InputQueueStatsSnapshot {
node_name: node.name.clone(),
port_name,
producer_name,
packet_capacity: node.input_queue_capacity[port],
queued_packets,
queued_bytes,
reserved_packets,
peak_queued_packets: stats.peak_packets.load(Ordering::Relaxed),
peak_queued_bytes: stats.peak_bytes.load(Ordering::Relaxed),
blocked: since != 0,
blocked_for_us,
block_events: stats.block_events.load(Ordering::Relaxed),
total_blocked_us: stats
.blocked_total_us
.load(Ordering::Relaxed)
.saturating_add(blocked_for_us),
})
}
pub(super) fn dump(&self) -> String {
let mut s = String::new();
s.push_str(&format!(
"graph state={:?} nodes={} edges={} queued={} ({} bytes)\n",
self.state(),
self.nodes.len(),
self.edges.len(),
self.shared.total_queued(),
self.shared.total_queued_bytes()
));
s.push_str("node state queued processed errors avg_us max_us\n");
for i in 0..self.nodes.len() {
let st = self.node_stats(i).expect("node exists");
let sched = self.nodes[i].sched.lock().expect("scheduler lock poisoned");
let state = if st.running {
format!("RUNNING {}ms", st.running_for_us / 1000)
} else if sched.closed {
"closed".to_string()
} else if sched.opened {
"idle".to_string()
} else {
"new".to_string()
};
let avg = if st.processed > 0 {
st.total_process_us / st.processed as i64
} else {
0
};
s.push_str(&format!(
"{:<17} {:<12} {:>6} {:>9} {:>6} {:>6} {:>6}\n",
st.node_name, state, st.queued, st.processed, st.errors, avg, st.max_process_us
));
for port in 0..self.nodes[i].input_queues.len() {
let queue = self.input_queue_stats(i, port).expect("input port exists");
let capacity = queue
.packet_capacity
.map_or_else(|| "unbounded".to_string(), |value| value.to_string());
s.push_str(&format!(
" input `{}` capacity={} packets queued={}/{}B reserved={} peak={}/{}B blocked={} events={} total={}us\n",
queue.port_name,
capacity,
queue.queued_packets,
queue.queued_bytes,
queue.reserved_packets,
queue.peak_queued_packets,
queue.peak_queued_bytes,
queue.blocked,
queue.block_events,
queue.total_blocked_us,
));
}
}
for e in &self.edges {
s.push_str(&format!(
"edge `{}` closed={} dropped={}\n",
e.name,
e.is_closed(),
e.dropped_count()
));
if e.is_graph_input {
let watermark = e.watermark_backpressure.snapshot(self.epoch_us());
s.push_str(&format!(
" watermark limit={} queued={} blocked={} waiters={} events={} total={}us\n",
self.shared.config.max_queued_packets,
self.shared.total_queued(),
watermark.blocked,
watermark.active_waiters,
watermark.block_events,
watermark.total_blocked_us,
));
}
for poller in e.pollers.lock().expect("poller list lock poisoned").iter() {
let block = poller.block_backpressure.snapshot(self.epoch_us());
let queued = poller.queue.lock().expect("poller lock poisoned").len();
s.push_str(&format!(
" poller policy={:?} capacity={} queued={} dropped={} blocked={} waiters={} events={} total={}us\n",
poller.overflow,
poller
.capacity
.map_or_else(|| "unbounded".to_string(), |value| value.to_string()),
queued,
poller.dropped.load(Ordering::Relaxed),
block.blocked,
block.active_waiters,
block.block_events,
block.total_blocked_us,
));
}
}
s
}
}