use super::theme::Theme;
use crate::metrics::{self, bytes_to_gbps};
use crate::net::{self, IfStats, NetRate};
use crate::stat::{self, PortStat, EXTRA_COUNTERS};
use crate::trace::{PortMetrics, Recorder};
use std::collections::VecDeque;
use std::time::{Duration, Instant};
use std::collections::HashMap;
#[derive(Clone, Copy, Debug, PartialEq, Eq, Hash)]
pub enum DeviceClass {
Rdma,
Xgmi,
Nvlink,
}
impl DeviceClass {
pub fn label(&self) -> &'static str {
match self {
DeviceClass::Rdma => "RDMA",
DeviceClass::Xgmi => "XGMI",
DeviceClass::Nvlink => "NVLink",
}
}
}
#[derive(Clone, Debug)]
pub struct PortThroughput {
pub dev_name: String,
pub port: u32,
pub link_gbps: Option<f64>,
pub state: Option<String>,
pub tx_gbps: f64,
pub rx_gbps: f64,
pub tx_pkts_per_sec: f64,
pub rx_pkts_per_sec: f64,
pub rx_drops_per_sec: f64,
pub counter_rates: Vec<CounterRate>,
pub port_label: Option<String>,
pub nvlink: Option<NvLinkThroughputMeta>,
pub xgmi: Option<XgmiThroughputMeta>,
pub class: DeviceClass,
}
#[derive(Clone, Debug)]
pub struct XgmiThroughputMeta {
#[allow(dead_code)]
pub gpu_index: u32,
pub gpu_name: String,
pub active_links: u32,
pub links: Vec<crate::xgmi::XgmiLinkSnapshot>,
pub metrics: Option<crate::gpu::GpuMetrics>,
}
#[derive(Clone, Debug)]
pub struct NvLinkThroughputMeta {
#[allow(dead_code)]
pub gpu_index: u32,
pub gpu_name: String,
pub active_links: u32,
pub links: Vec<crate::nvlink::LinkSnapshot>,
pub metrics: Option<crate::gpu::GpuMetrics>,
}
pub use crate::metrics::CounterRate;
#[derive(Clone, Debug, PartialEq, Eq)]
pub enum TableColumn {
Device,
Port,
State,
TxBar,
TxGbps,
RxBar,
RxGbps,
TxPps,
RxPps,
Drops,
Counter(String),
}
impl TableColumn {
pub fn label(&self) -> String {
match self {
Self::Device => "Device".into(),
Self::Port => "Port".into(),
Self::State => "State".into(),
Self::TxBar => "TX ▏".into(),
Self::TxGbps => "TX Gbps".into(),
Self::RxBar => "RX ▏".into(),
Self::RxGbps => "RX Gbps".into(),
Self::TxPps => "TX pps".into(),
Self::RxPps => "RX pps".into(),
Self::Drops => "Drops/s".into(),
Self::Counter(name) => name.clone(),
}
}
pub fn header_label(&self) -> String {
match self {
Self::TxBar => "TX".into(),
Self::RxBar => "RX".into(),
Self::TxGbps | Self::RxGbps => "Gbps".into(),
_ => self.label(),
}
}
pub fn width(&self) -> u16 {
match self {
Self::Device => 16,
Self::Port => 6,
Self::State => 5,
Self::TxBar | Self::RxBar => BAR_WIDTH as u16,
Self::TxGbps | Self::RxGbps => 9,
Self::TxPps | Self::RxPps => 10,
Self::Drops => 9,
Self::Counter(name) => (name.len() as u16 + 2).max(12),
}
}
}
pub const BAR_WIDTH: usize = 12;
pub const REFRESH_DEFAULT_SECS: f64 = 1.0;
const REFRESH_MIN_SECS: f64 = 0.2;
const REFRESH_MAX_SECS: f64 = 10.0;
const REFRESH_STEP_SECS: f64 = 0.5;
pub fn default_columns() -> Vec<TableColumn> {
vec![
TableColumn::Device,
TableColumn::Port,
TableColumn::State,
TableColumn::TxBar,
TableColumn::TxGbps,
TableColumn::RxBar,
TableColumn::RxGbps,
TableColumn::TxPps,
TableColumn::RxPps,
TableColumn::Drops,
]
}
pub fn all_columns() -> Vec<TableColumn> {
let mut cols = default_columns();
for name in EXTRA_COUNTERS {
cols.push(TableColumn::Counter(name.to_string()));
}
cols
}
pub struct RollingAvgState {
samples: HashMap<String, VecDeque<PortThroughput>>,
pub window_secs: usize,
}
pub const ROLLING_AVG_DEFAULT_WINDOW: usize = 5;
const ROLLING_AVG_MIN_WINDOW: usize = 1;
const ROLLING_AVG_MAX_WINDOW: usize = 300;
impl RollingAvgState {
pub fn new(window_secs: usize) -> Self {
Self {
samples: HashMap::new(),
window_secs,
}
}
pub fn push(&mut self, throughputs: &[PortThroughput]) {
let mut seen = std::collections::HashSet::new();
for t in throughputs {
let key = throughput_key(t);
seen.insert(key.clone());
let buf = self
.samples
.entry(key)
.or_insert_with(|| VecDeque::with_capacity(ROLLING_AVG_MAX_WINDOW + 1));
if buf.len() >= ROLLING_AVG_MAX_WINDOW {
buf.pop_front();
}
buf.push_back(t.clone());
}
self.samples.retain(|k, _| seen.contains(k));
}
pub fn averages(&self) -> Vec<PortThroughput> {
self.samples
.values()
.filter_map(|buf| Self::average_window(buf, self.window_secs))
.collect()
}
fn average_window(
buf: &VecDeque<PortThroughput>,
window_secs: usize,
) -> Option<PortThroughput> {
if buf.is_empty() {
return None;
}
let start = buf.len().saturating_sub(window_secs);
let window: Vec<_> = buf.iter().skip(start).collect();
let n = window.len() as f64;
let latest = window.last().unwrap();
let mut avg = PortThroughput {
dev_name: latest.dev_name.clone(),
port: latest.port,
link_gbps: latest.link_gbps,
state: latest.state.clone(),
tx_gbps: window.iter().map(|s| s.tx_gbps).sum::<f64>() / n,
rx_gbps: window.iter().map(|s| s.rx_gbps).sum::<f64>() / n,
tx_pkts_per_sec: window.iter().map(|s| s.tx_pkts_per_sec).sum::<f64>() / n,
rx_pkts_per_sec: window.iter().map(|s| s.rx_pkts_per_sec).sum::<f64>() / n,
rx_drops_per_sec: window.iter().map(|s| s.rx_drops_per_sec).sum::<f64>() / n,
counter_rates: Vec::new(),
port_label: latest.port_label.clone(),
nvlink: latest.nvlink.clone(),
xgmi: latest.xgmi.clone(),
class: latest.class,
};
if let Some(template) = window.last() {
avg.counter_rates = template
.counter_rates
.iter()
.map(|cr| Self::average_counter_rate(cr, &window))
.collect();
}
Some(avg)
}
fn average_counter_rate(cr: &CounterRate, window: &[&PortThroughput]) -> CounterRate {
let present: Vec<&CounterRate> = window
.iter()
.filter_map(|s| s.counter_rates.iter().find(|r| r.name == cr.name))
.collect();
let n = present.len();
if n == 0 {
return CounterRate {
name: cr.name.clone(),
value: cr.value,
delta: 0,
rate: 0.0,
is_bytes: cr.is_bytes,
};
}
let sum_rate: f64 = present.iter().map(|r| r.rate).sum();
let sum_delta: u64 = present.iter().map(|r| r.delta).sum();
let latest_value = present.last().map(|r| r.value).unwrap_or(cr.value);
CounterRate {
name: cr.name.clone(),
value: latest_value,
delta: sum_delta / n as u64,
rate: sum_rate / n as f64,
is_bytes: cr.is_bytes,
}
}
pub fn sample_count(&self) -> usize {
self.samples
.values()
.map(|b| b.len().min(self.window_secs))
.max()
.unwrap_or(0)
}
pub fn increase_window(&mut self) {
self.window_secs = (self.window_secs + 1).min(ROLLING_AVG_MAX_WINDOW);
}
pub fn decrease_window(&mut self) {
self.window_secs = self
.window_secs
.saturating_sub(1)
.max(ROLLING_AVG_MIN_WINDOW);
}
pub fn set_window(&mut self, secs: usize) {
self.window_secs = secs.clamp(ROLLING_AVG_MIN_WINDOW, ROLLING_AVG_MAX_WINDOW);
}
}
pub struct App {
pub should_quit: bool,
pub throughputs: Vec<PortThroughput>,
pub selected_row: usize,
pub show_detail: bool,
pub show_help: bool,
pub processes: Vec<stat::ProcessRdmaInfo>,
pub detail_scroll: u16,
pub detail_max_scroll: u16,
pub theme: Theme,
pub sysinfo: SysInfo,
pub history: HashMap<String, DeviceHistory>,
pub cpu_history: Vec<f32>,
prev_stats: Vec<PortStat>,
prev_ifstats: Vec<IfStats>,
prev_stats_at: Option<Instant>,
prev_ifstats_at: Option<Instant>,
prev_nvlink_at: Option<Instant>,
prev_xgmi_at: Option<Instant>,
prev_taken_at: Option<Instant>,
last_applied_at: Option<Instant>,
rdma_rows: Vec<PortThroughput>,
nvlink_rows: Vec<PortThroughput>,
xgmi_rows: Vec<PortThroughput>,
net_rate: NetRate,
pub has_data: bool,
pub sampler_error: Option<String>,
pub rolling_avg: RollingAvgState,
pub show_rolling_avg: bool,
pub show_window_input: bool,
pub window_input_buf: String,
pub columns: Vec<TableColumn>,
pub show_column_picker: bool,
pub column_picker_cursor: usize,
pub h_scroll: usize,
pub h_scroll_max: usize,
pub table_offset: usize,
pub refresh_interval: Duration,
cached_display: Vec<PortThroughput>,
pub recorder: Option<Recorder>,
pub record_status: Option<String>,
prev_nvlink: Vec<crate::nvlink::NvLinkSnapshot>,
prev_xgmi: Vec<crate::xgmi::XgmiSnapshot>,
pub active_tab: DeviceClass,
pub seen_tabs: Vec<DeviceClass>,
tab_selection: HashMap<DeviceClass, usize>,
}
const HISTORY_LEN: usize = 60;
#[derive(Clone, Debug)]
pub struct DeviceHistory {
pub tx: Vec<f64>,
pub rx: Vec<f64>,
pub util: Vec<f64>,
}
impl DeviceHistory {
fn new() -> Self {
Self {
tx: Vec::with_capacity(HISTORY_LEN),
rx: Vec::with_capacity(HISTORY_LEN),
util: Vec::with_capacity(HISTORY_LEN),
}
}
fn push(&mut self, tx: f64, rx: f64) {
if self.tx.len() >= HISTORY_LEN {
self.tx.remove(0);
self.rx.remove(0);
}
self.tx.push(tx);
self.rx.push(rx);
}
fn push_util(&mut self, util: f64) {
if self.util.len() >= HISTORY_LEN {
self.util.remove(0);
}
self.util.push(util);
}
}
#[derive(Clone)]
pub struct SysInfo {
pub hostname: String,
pub uptime: String,
pub load_avg: String,
pub mem_total_mb: u64,
pub mem_used_mb: u64,
pub mem_pct: f32,
pub cpu_pct: f32,
pub net: NetRate,
}
impl Default for App {
fn default() -> Self {
Self::new()
}
}
impl App {
pub fn new() -> Self {
Self {
should_quit: false,
throughputs: Vec::new(),
selected_row: 0,
theme: Theme::Default,
show_detail: false,
show_help: false,
processes: Vec::new(),
detail_scroll: 0,
detail_max_scroll: 0,
sysinfo: read_sysinfo(NetRate::default()),
history: HashMap::new(),
cpu_history: Vec::with_capacity(HISTORY_LEN),
prev_stats: Vec::new(),
prev_ifstats: Vec::new(),
prev_stats_at: None,
prev_ifstats_at: None,
prev_nvlink_at: None,
prev_xgmi_at: None,
prev_taken_at: None,
last_applied_at: None,
rdma_rows: Vec::new(),
nvlink_rows: Vec::new(),
xgmi_rows: Vec::new(),
net_rate: NetRate::default(),
has_data: false,
sampler_error: None,
rolling_avg: RollingAvgState::new(ROLLING_AVG_DEFAULT_WINDOW),
show_rolling_avg: false,
show_window_input: false,
window_input_buf: String::new(),
columns: default_columns(),
show_column_picker: false,
column_picker_cursor: 0,
h_scroll: 0,
h_scroll_max: 0,
table_offset: 0,
refresh_interval: Duration::from_secs_f64(REFRESH_DEFAULT_SECS),
cached_display: Vec::new(),
recorder: None,
record_status: None,
prev_nvlink: Vec::new(),
prev_xgmi: Vec::new(),
active_tab: DeviceClass::Rdma,
seen_tabs: vec![DeviceClass::Rdma],
tab_selection: HashMap::new(),
}
}
pub fn apply_snapshot(&mut self, snap: crate::sampler::Snapshot) {
if let Some(prev) = self.prev_taken_at {
if snap.taken_at.duration_since(prev).as_secs_f64() < 0.1 {
return;
}
}
let since =
|now: Instant, at: Option<Instant>| at.map(|a| now.duration_since(a).as_secs_f64());
if let Some(s) = snap.stats {
self.rdma_rows = match since(s.taken_at, self.prev_stats_at) {
Some(e) => compute_throughputs(&self.prev_stats, &s.data, e),
None => compute_throughputs(&s.data, &s.data, 1.0), };
self.prev_stats = s.data;
self.prev_stats_at = Some(s.taken_at);
}
if let Some(s) = snap.nvlink {
self.nvlink_rows = match since(s.taken_at, self.prev_nvlink_at) {
Some(e) => compute_nvlink_throughputs(&self.prev_nvlink, &s.data, e),
None => compute_nvlink_throughputs(&s.data, &s.data, 1.0), };
self.prev_nvlink = s.data;
self.prev_nvlink_at = Some(s.taken_at);
}
if let Some(s) = snap.xgmi {
self.xgmi_rows = match since(s.taken_at, self.prev_xgmi_at) {
Some(e) => compute_xgmi_throughputs(&self.prev_xgmi, &s.data, e),
None => compute_xgmi_throughputs(&s.data, &s.data, 1.0), };
self.prev_xgmi = s.data;
self.prev_xgmi_at = Some(s.taken_at);
}
if let Some(s) = snap.ifstats {
if let Some(e) = since(s.taken_at, self.prev_ifstats_at) {
self.net_rate = net::compute_net_rate(&self.prev_ifstats, &s.data, e);
}
self.prev_ifstats = s.data;
self.prev_ifstats_at = Some(s.taken_at);
}
if let Some(processes) = snap.processes {
self.processes = processes;
}
self.throughputs = self.rdma_rows.clone();
self.throughputs.extend(self.nvlink_rows.iter().cloned());
self.throughputs.extend(self.xgmi_rows.iter().cloned());
sort_by_device_order(&mut self.throughputs);
detect_tabs(&mut self.seen_tabs, &self.throughputs);
self.prev_taken_at = Some(snap.taken_at);
self.last_applied_at = Some(Instant::now());
self.has_data = true;
self.clamp_selection();
self.update_history();
self.rolling_avg.push(&self.throughputs);
if let Some(rec) = &mut self.recorder {
rec.push(snap.taken_at, port_metrics(&self.throughputs));
}
self.sysinfo = read_sysinfo(self.net_rate.clone());
if self.cpu_history.len() >= HISTORY_LEN {
self.cpu_history.remove(0);
}
self.cpu_history.push(self.sysinfo.cpu_pct);
self.recompute_display();
}
pub fn stale_secs(&self) -> Option<u64> {
let at = self.last_applied_at?;
let threshold = (self.refresh_interval * 3).max(Duration::from_secs(2));
let age = at.elapsed();
(age > threshold).then_some(age.as_secs())
}
fn recompute_display(&mut self) {
let rows: Vec<PortThroughput> = if self.show_rolling_avg {
let mut avgs = self.rolling_avg.averages();
sort_by_throughput_order(&mut avgs, &self.throughputs);
avgs
} else {
self.throughputs.clone()
};
self.cached_display = rows
.into_iter()
.filter(|t| t.class == self.active_tab)
.collect();
self.selected_row = self
.selected_row
.min(self.cached_display.len().saturating_sub(1));
}
fn update_history(&mut self) {
for t in &self.throughputs {
let util = t
.nvlink
.as_ref()
.and_then(|m| m.metrics.as_ref())
.or_else(|| t.xgmi.as_ref().and_then(|m| m.metrics.as_ref()))
.and_then(|m| m.util_pct);
let entry = self
.history
.entry(t.dev_name.clone())
.or_insert_with(DeviceHistory::new);
entry.push(t.tx_gbps, t.rx_gbps);
if let Some(u) = util {
entry.push_util(u as f64);
}
}
}
pub fn move_up(&mut self) {
self.selected_row = self.selected_row.saturating_sub(1);
}
pub fn move_down(&mut self) {
if !self.cached_display.is_empty() && self.selected_row < self.cached_display.len() - 1 {
self.selected_row += 1;
}
}
pub fn toggle_detail(&mut self) {
self.show_detail = !self.show_detail;
self.detail_scroll = 0;
}
pub fn detail_scroll_up(&mut self) {
if self.detail_scroll > 0 {
self.detail_scroll -= 1;
} else if self.selected_row > 0 {
self.selected_row -= 1;
self.detail_scroll = 0;
}
}
pub fn detail_scroll_down(&mut self, max: u16) {
if self.detail_scroll < max {
self.detail_scroll += 1;
} else if !self.cached_display.is_empty()
&& self.selected_row < self.cached_display.len() - 1
{
self.selected_row += 1;
self.detail_scroll = 0;
}
}
pub fn cycle_theme(&mut self) {
self.theme = self.theme.next();
}
pub fn toggle_rolling_avg(&mut self) {
self.show_rolling_avg = !self.show_rolling_avg;
self.recompute_display();
}
pub fn toggle_recording(&mut self) {
let Some(rec) = self.recorder.take() else {
self.recorder = Some(Recorder::new());
self.record_status = None;
return;
};
if rec.is_empty() {
self.record_status = Some("recording stopped (nothing captured)".to_string());
return;
}
let path = trace_filename();
if let Err(e) = rec.write_to(&path) {
self.record_status = Some(format!("save failed: {} (still recording)", e));
self.recorder = Some(rec);
return;
}
self.record_status = Some(format!("saved {} ({} samples)", path, rec.sample_count()));
}
pub fn recording_progress(&self) -> Option<(u64, usize)> {
self.recorder
.as_ref()
.map(|r| (r.elapsed_secs(), r.sample_count()))
}
pub fn increase_avg_window(&mut self) {
self.rolling_avg.increase_window();
if self.show_rolling_avg {
self.recompute_display();
}
}
pub fn decrease_avg_window(&mut self) {
self.rolling_avg.decrease_window();
if self.show_rolling_avg {
self.recompute_display();
}
}
pub fn increase_refresh_interval(&mut self) {
let secs = (self.refresh_interval.as_secs_f64() + REFRESH_STEP_SECS).min(REFRESH_MAX_SECS);
self.refresh_interval = Duration::from_secs_f64(secs);
}
pub fn decrease_refresh_interval(&mut self) {
let secs = (self.refresh_interval.as_secs_f64() - REFRESH_STEP_SECS).max(REFRESH_MIN_SECS);
self.refresh_interval = Duration::from_secs_f64(secs);
}
pub fn open_window_input(&mut self) {
self.window_input_buf = self.rolling_avg.window_secs.to_string();
self.show_window_input = true;
}
pub fn cancel_window_input(&mut self) {
self.show_window_input = false;
self.window_input_buf.clear();
}
pub fn confirm_window_input(&mut self) {
if let Ok(val) = self.window_input_buf.parse::<usize>() {
self.rolling_avg
.set_window(val.clamp(ROLLING_AVG_MIN_WINDOW, ROLLING_AVG_MAX_WINDOW));
if self.show_rolling_avg {
self.recompute_display();
}
}
self.show_window_input = false;
self.window_input_buf.clear();
}
pub fn open_column_picker(&mut self) {
if self.active_tab != DeviceClass::Rdma {
return;
}
self.show_column_picker = true;
self.column_picker_cursor = 0;
}
pub fn close_column_picker(&mut self) {
self.show_column_picker = false;
}
pub fn column_picker_up(&mut self) {
self.column_picker_cursor = self.column_picker_cursor.saturating_sub(1);
}
pub fn column_picker_down(&mut self) {
let max = all_columns().len().saturating_sub(1);
if self.column_picker_cursor < max {
self.column_picker_cursor += 1;
}
}
pub fn column_picker_toggle(&mut self) {
let all = all_columns();
if let Some(col) = all.get(self.column_picker_cursor) {
if let Some(pos) = self.columns.iter().position(|c| c == col) {
if self.columns.len() > 1 {
self.columns.remove(pos);
}
} else {
self.columns.push(col.clone());
}
}
self.clamp_h_scroll();
}
pub fn scroll_left(&mut self) {
self.h_scroll = self.h_scroll.saturating_sub(1);
}
pub fn scroll_right(&mut self) {
if self.h_scroll < self.h_scroll_max {
self.h_scroll += 1;
}
}
pub fn clamp_h_scroll(&mut self) {
if self.h_scroll >= self.columns.len() {
self.h_scroll = self.columns.len().saturating_sub(1);
}
}
pub fn display_throughputs(&self) -> &[PortThroughput] {
&self.cached_display
}
pub fn selected_throughput(&self) -> Option<&PortThroughput> {
self.cached_display.get(self.selected_row)
}
pub fn selected_device_processes(&self) -> Vec<&stat::ProcessRdmaInfo> {
let Some(t) = self.selected_throughput() else {
return Vec::new();
};
self.processes
.iter()
.filter(|p| p.dev_name == t.dev_name)
.collect()
}
fn clamp_selection(&mut self) {
if !self.cached_display.is_empty() && self.selected_row >= self.cached_display.len() {
self.selected_row = self.cached_display.len() - 1;
}
}
pub fn cycle_tab(&mut self, forward: bool) {
if self.seen_tabs.len() < 2 {
return;
}
let idx = self
.seen_tabs
.iter()
.position(|&t| t == self.active_tab)
.unwrap_or(0);
let n = self.seen_tabs.len();
let next = if forward {
(idx + 1) % n
} else {
(idx + n - 1) % n
};
self.tab_selection
.insert(self.active_tab, self.selected_row);
self.active_tab = self.seen_tabs[next];
self.selected_row = *self.tab_selection.get(&self.active_tab).unwrap_or(&0);
self.table_offset = 0;
self.h_scroll = 0;
self.detail_scroll = 0;
self.recompute_display();
}
}
fn detect_tabs(seen: &mut Vec<DeviceClass>, rows: &[PortThroughput]) {
for class in [DeviceClass::Xgmi, DeviceClass::Nvlink] {
if !seen.contains(&class) && rows.iter().any(|t| t.class == class) {
seen.push(class);
seen.sort_by_key(|c| *c as usize); }
}
}
fn read_file_trimmed(path: &str) -> String {
std::fs::read_to_string(path)
.unwrap_or_default()
.trim()
.to_string()
}
fn read_hostname() -> String {
let h = read_file_trimmed("/etc/hostname");
if h.is_empty() {
"unknown".into()
} else {
h
}
}
fn read_uptime() -> String {
let secs: u64 = read_file_trimmed("/proc/uptime")
.split_whitespace()
.next()
.and_then(|s| s.parse::<f64>().ok())
.unwrap_or(0.0) as u64;
format!(
"up {} days, {}:{:02}",
secs / 86400,
(secs % 86400) / 3600,
(secs % 3600) / 60
)
}
fn read_load_avg() -> String {
read_file_trimmed("/proc/loadavg")
.split_whitespace()
.take(3)
.collect::<Vec<_>>()
.join(", ")
}
fn read_meminfo() -> (u64, u64) {
let content = read_file_trimmed("/proc/meminfo");
let mut total = 0u64;
let mut avail = 0u64;
for line in content.lines() {
let val = || {
line.split_whitespace()
.nth(1)
.and_then(|v| v.parse::<u64>().ok())
.unwrap_or(0)
};
if line.starts_with("MemTotal:") {
total = val();
}
if line.starts_with("MemAvailable:") {
avail = val();
}
}
(total / 1024, (total.saturating_sub(avail)) / 1024)
}
fn read_cpu_usage() -> f32 {
let content = read_file_trimmed("/proc/stat");
let first = content.lines().next().unwrap_or("");
let vals: Vec<u64> = first
.split_whitespace()
.skip(1)
.filter_map(|v| v.parse().ok())
.collect();
if vals.len() < 4 {
return 0.0;
}
let idle = vals[3];
let total: u64 = vals.iter().sum();
if total == 0 {
return 0.0;
}
((total - idle) as f32 / total as f32) * 100.0
}
fn read_sysinfo(net: NetRate) -> SysInfo {
let (mem_total_mb, mem_used_mb) = read_meminfo();
let mem_pct = if mem_total_mb > 0 {
(mem_used_mb as f32 / mem_total_mb as f32) * 100.0
} else {
0.0
};
SysInfo {
hostname: read_hostname(),
uptime: read_uptime(),
load_avg: read_load_avg(),
mem_total_mb,
mem_used_mb,
mem_pct,
cpu_pct: read_cpu_usage(),
net,
}
}
fn port_metrics(throughputs: &[PortThroughput]) -> Vec<PortMetrics> {
throughputs
.iter()
.map(|t| PortMetrics {
dev_name: t.dev_name.clone(),
port: t.port,
tx_gbps: t.tx_gbps,
rx_gbps: t.rx_gbps,
tx_pps: t.tx_pkts_per_sec,
rx_pps: t.rx_pkts_per_sec,
rx_drops_per_sec: t.rx_drops_per_sec,
})
.collect()
}
fn trace_filename() -> String {
let secs = std::time::SystemTime::now()
.duration_since(std::time::UNIX_EPOCH)
.map(|d| d.as_secs())
.unwrap_or(0);
format!("rdmatop-{}.json", secs)
}
fn compute_port_throughput(
curr: &PortStat,
prev: Option<&PortStat>,
elapsed: f64,
) -> PortThroughput {
let counter_rates = metrics::counter_rates(curr, prev, elapsed);
let rates = metrics::port_metrics(curr, prev, elapsed);
PortThroughput {
dev_name: rates.dev_name,
port: rates.port,
link_gbps: curr.link_gbps,
state: curr.state.clone(),
tx_gbps: rates.tx_gbps,
rx_gbps: rates.rx_gbps,
tx_pkts_per_sec: rates.tx_pps,
rx_pkts_per_sec: rates.rx_pps,
rx_drops_per_sec: rates.rx_drops_per_sec,
counter_rates,
port_label: None,
nvlink: None,
xgmi: None,
class: DeviceClass::Rdma,
}
}
fn compute_throughputs(prev: &[PortStat], curr: &[PortStat], elapsed: f64) -> Vec<PortThroughput> {
curr.iter()
.map(|c| compute_port_throughput(c, metrics::find_prev(prev, &c.dev_name, c.port), elapsed))
.collect()
}
fn throughput_key(t: &PortThroughput) -> String {
if t.nvlink.is_some() || t.xgmi.is_some() {
return t.dev_name.clone();
}
format!("{}/{}", t.dev_name, t.port)
}
fn sort_by_device_order(throughputs: &mut [PortThroughput]) {
throughputs.sort_by(|a, b| {
alphanumeric_sort::compare_str(&a.dev_name, &b.dev_name).then(a.port.cmp(&b.port))
});
}
fn sort_by_throughput_order(avgs: &mut [PortThroughput], reference: &[PortThroughput]) {
let index_map: HashMap<String, usize> = reference
.iter()
.enumerate()
.map(|(i, t)| (throughput_key(t), i))
.collect();
avgs.sort_by_key(|t| *index_map.get(&throughput_key(t)).unwrap_or(&usize::MAX));
}
fn compute_nvlink_throughputs(
prev: &[crate::nvlink::NvLinkSnapshot],
curr: &[crate::nvlink::NvLinkSnapshot],
elapsed: f64,
) -> Vec<PortThroughput> {
let delta = |c: Option<u64>, p: Option<u64>| match (c, p) {
(Some(c), Some(p)) => c.saturating_sub(p),
_ => 0,
};
let mut out = Vec::with_capacity(curr.len());
for gpu in curr {
let prev_gpu = prev.iter().find(|p| p.gpu_index == gpu.gpu_index);
let mut counter_rates: Vec<CounterRate> = Vec::new();
let (tx_bytes_total, rx_bytes_total) = if gpu.tx_bytes.is_some() || gpu.rx_bytes.is_some() {
(
delta(gpu.tx_bytes, prev_gpu.and_then(|p| p.tx_bytes)),
delta(gpu.rx_bytes, prev_gpu.and_then(|p| p.rx_bytes)),
)
} else {
let aggregate_link = gpu
.links
.iter()
.find(|l| l.is_active)
.or_else(|| gpu.links.first());
match aggregate_link {
Some(link) => {
let prev_link =
prev_gpu.and_then(|p| p.links.iter().find(|l| l.link_id == link.link_id));
(
delta(link.tx_bytes, prev_link.and_then(|l| l.tx_bytes)),
delta(link.rx_bytes, prev_link.and_then(|l| l.rx_bytes)),
)
}
None => (0, 0),
}
};
let rate = |c: Option<u64>, p: Option<u64>| -> Option<u64> {
c.map(|_| (delta(c, p) as f64 / elapsed) as u64)
};
let mut links = Vec::with_capacity(gpu.links.len());
for link in &gpu.links {
let prev_link =
prev_gpu.and_then(|p| p.links.iter().find(|l| l.link_id == link.link_id));
let mut link_clone = link.clone();
link_clone.tx_bytes = rate(link.tx_bytes, prev_link.and_then(|l| l.tx_bytes));
link_clone.rx_bytes = rate(link.rx_bytes, prev_link.and_then(|l| l.rx_bytes));
links.push(link_clone);
}
for link in &gpu.links {
let prev_link =
prev_gpu.and_then(|p| p.links.iter().find(|l| l.link_id == link.link_id));
if let Some(v) = link.crc_error_count {
let d = delta(
link.crc_error_count,
prev_link.and_then(|l| l.crc_error_count),
);
counter_rates.push(CounterRate {
name: format!("nvlink_crc_l{}", link.link_id),
value: v,
delta: d,
rate: d as f64 / elapsed,
is_bytes: false,
});
}
if let Some(v) = link.replay_error_count {
let d = delta(
link.replay_error_count,
prev_link.and_then(|l| l.replay_error_count),
);
counter_rates.push(CounterRate {
name: format!("nvlink_replay_l{}", link.link_id),
value: v,
delta: d,
rate: d as f64 / elapsed,
is_bytes: false,
});
}
if let Some(v) = link.recovery_error_count {
let d = delta(
link.recovery_error_count,
prev_link.and_then(|l| l.recovery_error_count),
);
counter_rates.push(CounterRate {
name: format!("nvlink_recovery_l{}", link.link_id),
value: v,
delta: d,
rate: d as f64 / elapsed,
is_bytes: false,
});
}
}
let active = gpu.active_links();
out.push(PortThroughput {
dev_name: format!("nvidia{}", gpu.gpu_index),
port: active,
link_gbps: gpu.link_gbps,
state: None,
tx_gbps: bytes_to_gbps(tx_bytes_total as f64 / elapsed),
rx_gbps: bytes_to_gbps(rx_bytes_total as f64 / elapsed),
tx_pkts_per_sec: 0.0,
rx_pkts_per_sec: 0.0,
rx_drops_per_sec: 0.0,
counter_rates,
port_label: Some(format!("{}/{}", active, gpu.link_count)),
nvlink: Some(NvLinkThroughputMeta {
gpu_index: gpu.gpu_index,
gpu_name: gpu.gpu_name.clone(),
active_links: active,
links,
metrics: gpu.metrics.clone(),
}),
xgmi: None,
class: DeviceClass::Nvlink,
});
}
out
}
fn compute_xgmi_throughputs(
prev: &[crate::xgmi::XgmiSnapshot],
curr: &[crate::xgmi::XgmiSnapshot],
elapsed: f64,
) -> Vec<PortThroughput> {
let mut out = Vec::with_capacity(curr.len());
for gpu in curr {
let prev_gpu = prev.iter().find(|p| p.gpu_index == gpu.gpu_index);
let mut tx_total: u64 = 0;
let mut rx_total: u64 = 0;
let mut links = Vec::with_capacity(gpu.links.len());
let delta = |c: Option<u64>, p: Option<u64>| match (c, p) {
(Some(c), Some(p)) => c.saturating_sub(p),
_ => 0,
};
for link in &gpu.links {
let prev_link =
prev_gpu.and_then(|p| p.links.iter().find(|l| l.link_id == link.link_id));
let (tx_delta, rx_delta) = match prev_link {
Some(pl) => (
delta(link.tx_bytes, pl.tx_bytes),
delta(link.rx_bytes, pl.rx_bytes),
),
None => (0, 0),
};
tx_total = tx_total.saturating_add(tx_delta);
rx_total = rx_total.saturating_add(rx_delta);
let mut link_rate = link.clone();
link_rate.tx_bytes = link.tx_bytes.map(|_| (tx_delta as f64 / elapsed) as u64);
link_rate.rx_bytes = link.rx_bytes.map(|_| (rx_delta as f64 / elapsed) as u64);
links.push(link_rate);
}
let mut counter_rates: Vec<CounterRate> = Vec::new();
if let Some(v) = gpu.correctable_errors {
let delta = match prev_gpu.and_then(|p| p.correctable_errors) {
Some(prev_v) => v.saturating_sub(prev_v),
None => 0,
};
counter_rates.push(CounterRate {
name: "xgmi_wafl_ce".to_string(),
value: v,
delta,
rate: delta as f64 / elapsed,
is_bytes: false,
});
}
if let Some(v) = gpu.uncorrectable_errors {
let delta = match prev_gpu.and_then(|p| p.uncorrectable_errors) {
Some(prev_v) => v.saturating_sub(prev_v),
None => 0,
};
counter_rates.push(CounterRate {
name: "xgmi_wafl_ue".to_string(),
value: v,
delta,
rate: delta as f64 / elapsed,
is_bytes: false,
});
}
let active = gpu.active_links();
out.push(PortThroughput {
dev_name: format!("amdgpu{}", gpu.gpu_index),
port: active,
link_gbps: gpu.link_gbps,
state: None,
tx_gbps: bytes_to_gbps(tx_total as f64 / elapsed),
rx_gbps: bytes_to_gbps(rx_total as f64 / elapsed),
tx_pkts_per_sec: 0.0,
rx_pkts_per_sec: 0.0,
rx_drops_per_sec: 0.0,
counter_rates,
port_label: Some(format!("{}/{}", active, gpu.link_count)),
nvlink: None,
xgmi: Some(XgmiThroughputMeta {
gpu_index: gpu.gpu_index,
gpu_name: gpu.gpu_name.clone(),
active_links: active,
links,
metrics: gpu.metrics.clone(),
}),
class: DeviceClass::Xgmi,
});
}
out
}
#[cfg(test)]
mod xgmi_tests {
use super::*;
use crate::xgmi::{XgmiLinkSnapshot, XgmiSnapshot};
fn mk_link(id: u32, active: bool, tx: u64, rx: u64) -> XgmiLinkSnapshot {
XgmiLinkSnapshot {
link_id: id,
is_active: active,
speed_gbps: Some(512.0),
bit_rate_gbps: Some(32.0),
remote_pci_bdf: Some(format!("0000:0{}:00.0", id + 1)),
tx_bytes: Some(tx),
rx_bytes: Some(rx),
}
}
fn mk_gpu(index: u32, links: Vec<XgmiLinkSnapshot>) -> XgmiSnapshot {
let active: u32 = links.iter().filter(|l| l.is_active).count() as u32;
XgmiSnapshot {
gpu_index: index,
gpu_name: "MI325X".to_string(),
link_count: links.len() as u32,
link_gbps: Some(512.0 * active as f64),
correctable_errors: Some(0),
uncorrectable_errors: Some(0),
metrics: None,
links,
}
}
#[test]
fn totals_are_sum_of_link_deltas() {
let prev = vec![mk_gpu(
0,
vec![mk_link(0, true, 1000, 2000), mk_link(1, true, 500, 500)],
)];
let curr = vec![mk_gpu(
0,
vec![mk_link(0, true, 3000, 6000), mk_link(1, true, 1500, 1500)],
)];
let rows = compute_xgmi_throughputs(&prev, &curr, 1.0);
assert_eq!(rows.len(), 1);
let row = &rows[0];
assert_eq!(row.dev_name, "amdgpu0");
assert!((row.tx_gbps - bytes_to_gbps(3000.0)).abs() < 1e-12);
assert!((row.rx_gbps - bytes_to_gbps(5000.0)).abs() < 1e-12);
assert_eq!(row.port_label.as_deref(), Some("2/2"));
let meta = row.xgmi.as_ref().expect("xgmi meta present");
assert_eq!(meta.links[0].tx_bytes, Some(2000));
assert_eq!(meta.links[0].rx_bytes, Some(4000));
assert_eq!(meta.links[1].tx_bytes, Some(1000));
assert_eq!(meta.links[1].rx_bytes, Some(1000));
}
#[test]
fn unseen_gpu_emits_zero_rates() {
let curr = vec![mk_gpu(0, vec![mk_link(0, true, 4096, 8192)])];
let rows = compute_xgmi_throughputs(&[], &curr, 2.0);
let row = &rows[0];
assert_eq!(row.tx_gbps, 0.0);
assert_eq!(row.rx_gbps, 0.0);
let meta = row.xgmi.as_ref().expect("xgmi meta present");
assert_eq!(meta.links[0].tx_bytes, Some(0));
assert_eq!(meta.links[0].rx_bytes, Some(0));
}
#[test]
fn counter_reappearing_after_gap_emits_zero() {
let mut prev_link = mk_link(0, true, 0, 0);
prev_link.tx_bytes = None;
prev_link.rx_bytes = None;
let prev = vec![mk_gpu(0, vec![prev_link])];
let curr = vec![mk_gpu(
0,
vec![mk_link(0, true, u64::MAX / 2, u64::MAX / 2)],
)];
let rows = compute_xgmi_throughputs(&prev, &curr, 1.0);
assert_eq!(rows[0].tx_gbps, 0.0);
assert_eq!(rows[0].rx_gbps, 0.0);
}
#[test]
fn missing_counters_stay_none() {
let mut link = mk_link(0, true, 0, 0);
link.tx_bytes = None;
link.rx_bytes = None;
let prev = vec![mk_gpu(0, vec![link.clone()])];
let curr = vec![mk_gpu(0, vec![link])];
let rows = compute_xgmi_throughputs(&prev, &curr, 1.0);
let meta = rows[0].xgmi.as_ref().expect("xgmi meta present");
assert_eq!(meta.links[0].tx_bytes, None);
assert_eq!(meta.links[0].rx_bytes, None);
}
#[test]
fn counter_going_backwards_clamps_to_zero() {
let prev = vec![mk_gpu(0, vec![mk_link(0, true, 9999, 9999)])];
let curr = vec![mk_gpu(0, vec![mk_link(0, true, 1, 1)])];
let rows = compute_xgmi_throughputs(&prev, &curr, 1.0);
assert_eq!(rows[0].tx_gbps, 0.0);
assert_eq!(rows[0].rx_gbps, 0.0);
}
#[test]
fn inactive_links_excluded_from_active_count() {
let curr = vec![mk_gpu(
0,
vec![mk_link(0, true, 0, 0), mk_link(1, false, 0, 0)],
)];
let rows = compute_xgmi_throughputs(&[], &curr, 1.0);
assert_eq!(rows[0].port, 1);
assert_eq!(rows[0].port_label.as_deref(), Some("1/2"));
}
#[test]
fn wafl_error_counters_emitted_with_deltas() {
let mut prev_gpu = mk_gpu(0, vec![mk_link(0, true, 0, 0)]);
prev_gpu.correctable_errors = Some(5);
prev_gpu.uncorrectable_errors = Some(1);
let mut curr_gpu = mk_gpu(0, vec![mk_link(0, true, 0, 0)]);
curr_gpu.correctable_errors = Some(8);
curr_gpu.uncorrectable_errors = Some(1);
let rows = compute_xgmi_throughputs(&[prev_gpu], &[curr_gpu], 1.0);
let ce = rows[0]
.counter_rates
.iter()
.find(|c| c.name == "xgmi_wafl_ce")
.unwrap();
assert_eq!(ce.value, 8);
assert_eq!(ce.delta, 3);
let ue = rows[0]
.counter_rates
.iter()
.find(|c| c.name == "xgmi_wafl_ue")
.unwrap();
assert_eq!(ue.value, 1);
assert_eq!(ue.delta, 0);
}
#[test]
fn throughput_key_uses_dev_name_for_xgmi_rows() {
let rows = compute_xgmi_throughputs(&[], &[mk_gpu(3, vec![mk_link(0, true, 0, 0)])], 1.0);
assert_eq!(throughput_key(&rows[0]), "amdgpu3");
}
}
#[cfg(test)]
mod nvlink_tests {
use super::*;
use crate::nvlink::{LinkSnapshot, NvLinkSnapshot, RemoteDeviceType};
fn make_link(id: u32, active: bool, tx: Option<u64>, rx: Option<u64>) -> LinkSnapshot {
LinkSnapshot {
link_id: id,
is_active: active,
version: None,
speed_gbps: if active { Some(50.0) } else { None },
remote_device_type: RemoteDeviceType::Gpu,
remote_pci_bdf: None,
tx_bytes: tx,
rx_bytes: rx,
crc_error_count: None,
replay_error_count: None,
recovery_error_count: None,
}
}
#[test]
fn aggregates_only_active_links_and_sets_port_label() {
let elapsed = 1.0_f64;
let prev = vec![NvLinkSnapshot {
gpu_index: 0,
gpu_name: "H100".to_string(),
metrics: None,
link_count: 2,
link_gbps: Some(100.0),
tx_bytes: None,
rx_bytes: None,
links: vec![
make_link(0, true, Some(1_000_000_000), Some(2_000_000_000)),
make_link(1, true, Some(500_000_000), Some(0)),
],
}];
let curr = vec![NvLinkSnapshot {
gpu_index: 0,
gpu_name: "H100".to_string(),
metrics: None,
link_count: 3,
link_gbps: Some(150.0),
tx_bytes: None,
rx_bytes: None,
links: vec![
make_link(0, true, Some(1_500_000_000), Some(2_500_000_000)),
make_link(1, true, Some(700_000_000), Some(100_000_000)),
make_link(2, false, None, None),
],
}];
let rows = compute_nvlink_throughputs(&prev, &curr, elapsed);
assert_eq!(rows.len(), 1);
let row = &rows[0];
assert_eq!(row.dev_name, "nvidia0");
assert_eq!(row.port, 2, "port column = active link count");
assert_eq!(row.port_label.as_deref(), Some("2/3"));
assert_eq!(row.link_gbps, Some(150.0));
assert_eq!(row.tx_pkts_per_sec, 0.0);
assert_eq!(row.rx_pkts_per_sec, 0.0);
assert_eq!(row.rx_drops_per_sec, 0.0);
let expected_tx_gbps = 500_000_000.0_f64 * 8.0 / 1_000_000_000.0;
let expected_rx_gbps = 500_000_000.0_f64 * 8.0 / 1_000_000_000.0;
assert!(
(row.tx_gbps - expected_tx_gbps).abs() < 1e-6,
"tx_gbps {} != {}",
row.tx_gbps,
expected_tx_gbps
);
assert!(
(row.rx_gbps - expected_rx_gbps).abs() < 1e-6,
"rx_gbps {} != {}",
row.rx_gbps,
expected_rx_gbps
);
assert!(
!row.counter_rates
.iter()
.any(|c| c.name.starts_with("nvlink_tx_l")),
"per-link TX throughput counter rates must not be emitted"
);
assert!(
!row.counter_rates
.iter()
.any(|c| c.name.starts_with("nvlink_rx_l")),
"per-link RX throughput counter rates must not be emitted"
);
let nv = row.nvlink.as_ref().expect("nvlink meta present");
assert_eq!(nv.gpu_index, 0);
assert_eq!(nv.gpu_name, "H100");
assert_eq!(nv.active_links, 2);
assert_eq!(nv.links.len(), 3);
}
#[test]
fn aggregate_throughput_is_not_multiplied_by_active_link_count() {
let elapsed = 1.0_f64;
let prev = vec![NvLinkSnapshot {
gpu_index: 0,
gpu_name: "H100".to_string(),
metrics: None,
link_count: 2,
link_gbps: Some(100.0),
tx_bytes: None,
rx_bytes: None,
links: vec![
make_link(0, true, Some(1_000_000_000), Some(2_000_000_000)),
make_link(1, true, Some(1_000_000_000), Some(2_000_000_000)),
],
}];
let curr = vec![NvLinkSnapshot {
gpu_index: 0,
gpu_name: "H100".to_string(),
metrics: None,
link_count: 2,
link_gbps: Some(100.0),
tx_bytes: None,
rx_bytes: None,
links: vec![
make_link(0, true, Some(1_125_000_000), Some(2_125_000_000)),
make_link(1, true, Some(1_125_000_000), Some(2_125_000_000)),
],
}];
let rows = compute_nvlink_throughputs(&prev, &curr, elapsed);
assert_eq!(rows.len(), 1);
let row = &rows[0];
let expected_tx_gbps = 125_000_000.0_f64 * 8.0 / 1_000_000_000.0;
let expected_rx_gbps = 125_000_000.0_f64 * 8.0 / 1_000_000_000.0;
assert!(
(row.tx_gbps - expected_tx_gbps).abs() < 1e-6,
"tx_gbps {} != single-link rate {}",
row.tx_gbps,
expected_tx_gbps
);
assert!(
(row.rx_gbps - expected_rx_gbps).abs() < 1e-6,
"rx_gbps {} != single-link rate {}",
row.rx_gbps,
expected_rx_gbps
);
assert!(
row.tx_gbps < expected_tx_gbps * 1.5,
"tx_gbps {} unexpectedly large; sum-across-links bug?",
row.tx_gbps
);
assert!(
row.rx_gbps < expected_rx_gbps * 1.5,
"rx_gbps {} unexpectedly large; sum-across-links bug?",
row.rx_gbps
);
assert_eq!(row.port, 2);
assert_eq!(row.port_label.as_deref(), Some("2/2"));
}
#[test]
fn aggregate_falls_back_to_first_link_when_no_active() {
let elapsed = 1.0_f64;
let prev = vec![NvLinkSnapshot {
gpu_index: 0,
gpu_name: "H100".to_string(),
metrics: None,
link_count: 2,
link_gbps: Some(100.0),
tx_bytes: None,
rx_bytes: None,
links: vec![
make_link(0, false, Some(0), Some(0)),
make_link(1, false, Some(0), Some(0)),
],
}];
let curr = vec![NvLinkSnapshot {
gpu_index: 0,
gpu_name: "H100".to_string(),
metrics: None,
link_count: 2,
link_gbps: Some(100.0),
tx_bytes: None,
rx_bytes: None,
links: vec![
make_link(0, false, Some(250_000_000), Some(125_000_000)),
make_link(1, false, Some(999_999_999), Some(999_999_999)),
],
}];
let rows = compute_nvlink_throughputs(&prev, &curr, elapsed);
assert_eq!(rows.len(), 1);
let row = &rows[0];
let expected_tx_gbps = 250_000_000.0_f64 * 8.0 / 1_000_000_000.0;
let expected_rx_gbps = 125_000_000.0_f64 * 8.0 / 1_000_000_000.0;
assert!((row.tx_gbps - expected_tx_gbps).abs() < 1e-6);
assert!((row.rx_gbps - expected_rx_gbps).abs() < 1e-6);
assert_eq!(row.port, 0);
assert_eq!(row.port_label.as_deref(), Some("0/2"));
}
#[test]
fn missing_prev_snapshot_emits_zero_rates() {
let elapsed = 2.0_f64;
let curr = vec![NvLinkSnapshot {
gpu_index: 1,
gpu_name: "A100".to_string(),
metrics: None,
link_count: 1,
link_gbps: Some(50.0),
tx_bytes: None,
rx_bytes: None,
links: vec![make_link(0, true, Some(2_000_000_000), Some(1_000_000_000))],
}];
let rows = compute_nvlink_throughputs(&[], &curr, elapsed);
assert_eq!(rows.len(), 1);
let row = &rows[0];
assert_eq!(row.dev_name, "nvidia1");
assert_eq!(row.port_label.as_deref(), Some("1/1"));
assert_eq!(row.tx_gbps, 0.0);
assert_eq!(row.rx_gbps, 0.0);
}
#[test]
fn rolling_avg_keeps_history_across_active_link_changes() {
let elapsed = 1.0_f64;
let links_first = vec![
make_link(0, true, Some(1_000_000_000), Some(2_000_000_000)),
make_link(1, true, Some(500_000_000), Some(250_000_000)),
];
let prev = vec![NvLinkSnapshot {
gpu_index: 0,
gpu_name: "H100".to_string(),
metrics: None,
link_count: 2,
link_gbps: Some(100.0),
tx_bytes: None,
rx_bytes: None,
links: links_first.clone(),
}];
let curr_first = vec![NvLinkSnapshot {
gpu_index: 0,
gpu_name: "H100".to_string(),
metrics: None,
link_count: 3,
link_gbps: Some(150.0),
tx_bytes: None,
rx_bytes: None,
links: vec![
make_link(0, true, Some(1_500_000_000), Some(2_500_000_000)),
make_link(1, true, Some(700_000_000), Some(350_000_000)),
make_link(2, false, None, None),
],
}];
let rows_first = compute_nvlink_throughputs(&prev, &curr_first, elapsed);
assert_eq!(rows_first.len(), 1);
assert_eq!(rows_first[0].port_label.as_deref(), Some("2/3"));
let mut state = RollingAvgState::new(5);
state.push(&rows_first);
assert_eq!(state.samples.len(), 1);
let curr_second = vec![NvLinkSnapshot {
gpu_index: 0,
gpu_name: "H100".to_string(),
metrics: None,
link_count: 3,
link_gbps: Some(150.0),
tx_bytes: None,
rx_bytes: None,
links: vec![
make_link(0, true, Some(2_000_000_000), Some(3_000_000_000)),
make_link(1, true, Some(900_000_000), Some(450_000_000)),
make_link(2, true, Some(100_000_000), Some(50_000_000)),
],
}];
let rows_second = compute_nvlink_throughputs(&curr_first, &curr_second, elapsed);
assert_eq!(rows_second.len(), 1);
assert_eq!(rows_second[0].port_label.as_deref(), Some("3/3"));
state.push(&rows_second);
assert_eq!(
state.samples.len(),
1,
"expected stable key for NVLink rows across active-link changes"
);
let buf = state
.samples
.values()
.next()
.expect("history entry for nvidia0");
assert_eq!(buf.len(), 2);
assert_eq!(buf[0].port_label.as_deref(), Some("2/3"));
assert_eq!(buf[1].port_label.as_deref(), Some("3/3"));
}
#[test]
fn rolling_avg_returns_averaged_nvlink_row() {
let elapsed = 1.0_f64;
let prev = vec![NvLinkSnapshot {
gpu_index: 0,
gpu_name: "H100".to_string(),
metrics: None,
link_count: 1,
link_gbps: Some(50.0),
tx_bytes: None,
rx_bytes: None,
links: vec![make_link(0, true, Some(0), Some(0))],
}];
let curr = vec![NvLinkSnapshot {
gpu_index: 0,
gpu_name: "H100".to_string(),
metrics: None,
link_count: 1,
link_gbps: Some(50.0),
tx_bytes: None,
rx_bytes: None,
links: vec![make_link(0, true, Some(1_000_000_000), Some(500_000_000))],
}];
let rows = compute_nvlink_throughputs(&prev, &curr, elapsed);
assert_eq!(rows.len(), 1);
let mut state = RollingAvgState::new(5);
state.push(&rows);
let avgs = state.averages();
assert_eq!(avgs.len(), 1);
let avg = &avgs[0];
assert_eq!(avg.dev_name, "nvidia0");
assert!(
avg.tx_gbps > 0.0,
"expected non-empty averaged tx_gbps, got {}",
avg.tx_gbps
);
let expected_tx = 1_000_000_000.0_f64 * 8.0 / 1_000_000_000.0;
assert!(
(avg.tx_gbps - expected_tx).abs() < 1e-6,
"tx_gbps {} != {}",
avg.tx_gbps,
expected_tx
);
}
#[test]
fn saturating_sub_protects_against_counter_resets() {
let elapsed = 1.0_f64;
let prev = vec![NvLinkSnapshot {
gpu_index: 2,
gpu_name: "B200".to_string(),
metrics: None,
link_count: 1,
link_gbps: Some(100.0),
tx_bytes: None,
rx_bytes: None,
links: vec![make_link(
0,
true,
Some(10_000_000_000),
Some(20_000_000_000),
)],
}];
let curr = vec![NvLinkSnapshot {
gpu_index: 2,
gpu_name: "B200".to_string(),
metrics: None,
link_count: 1,
link_gbps: Some(100.0),
tx_bytes: None,
rx_bytes: None,
links: vec![make_link(0, true, Some(1_000), Some(2_000))],
}];
let rows = compute_nvlink_throughputs(&prev, &curr, elapsed);
assert_eq!(rows.len(), 1);
assert_eq!(
rows[0].tx_gbps, 0.0,
"counter reset must not produce negative rate"
);
assert_eq!(rows[0].rx_gbps, 0.0);
}
#[test]
fn averaged_row_uses_latest_metadata() {
let elapsed = 1.0_f64;
let prev_a = vec![NvLinkSnapshot {
gpu_index: 0,
gpu_name: "H100".to_string(),
metrics: None,
link_count: 3,
link_gbps: Some(150.0),
tx_bytes: None,
rx_bytes: None,
links: vec![make_link(0, true, Some(0), Some(0))],
}];
let curr_a = vec![NvLinkSnapshot {
gpu_index: 0,
gpu_name: "H100".to_string(),
metrics: None,
link_count: 3,
link_gbps: Some(150.0),
tx_bytes: None,
rx_bytes: None,
links: vec![
make_link(0, true, Some(1_000_000_000), Some(2_000_000_000)),
make_link(1, true, Some(500_000_000), Some(250_000_000)),
make_link(2, false, None, None),
],
}];
let rows_first = compute_nvlink_throughputs(&prev_a, &curr_a, elapsed);
assert_eq!(rows_first.len(), 1);
assert_eq!(rows_first[0].port_label.as_deref(), Some("2/3"));
assert_eq!(rows_first[0].port, 2);
let curr_b = vec![NvLinkSnapshot {
gpu_index: 0,
gpu_name: "H100".to_string(),
metrics: None,
link_count: 3,
link_gbps: Some(150.0),
tx_bytes: None,
rx_bytes: None,
links: vec![
make_link(0, true, Some(2_000_000_000), Some(3_000_000_000)),
make_link(1, true, Some(900_000_000), Some(450_000_000)),
make_link(2, true, Some(100_000_000), Some(50_000_000)),
],
}];
let rows_second = compute_nvlink_throughputs(&curr_a, &curr_b, elapsed);
assert_eq!(rows_second.len(), 1);
assert_eq!(rows_second[0].port_label.as_deref(), Some("3/3"));
assert_eq!(rows_second[0].port, 3);
let mut state = RollingAvgState::new(5);
state.push(&rows_first);
state.push(&rows_second);
let avgs = state.averages();
assert_eq!(avgs.len(), 1);
let avg = &avgs[0];
assert_eq!(
avg.port_label.as_deref(),
Some("3/3"),
"averaged row must use latest sample's port_label"
);
assert_eq!(avg.port, 3, "averaged row must use latest sample's port");
assert_eq!(avg.link_gbps, Some(150.0));
assert_eq!(avg.dev_name, "nvidia0");
}
#[test]
fn aggregates_uses_gpu_wide_fields_when_present() {
let elapsed = 1.0_f64;
let prev = vec![NvLinkSnapshot {
gpu_index: 0,
gpu_name: "H100".to_string(),
metrics: None,
link_count: 2,
link_gbps: Some(100.0),
tx_bytes: Some(10_000_000_000),
rx_bytes: Some(20_000_000_000),
links: vec![
make_link(0, true, Some(0), Some(0)),
make_link(1, true, Some(0), Some(0)),
],
}];
let curr = vec![NvLinkSnapshot {
gpu_index: 0,
gpu_name: "H100".to_string(),
metrics: None,
link_count: 2,
link_gbps: Some(100.0),
tx_bytes: Some(15_000_000_000),
rx_bytes: Some(25_000_000_000),
links: vec![
make_link(0, true, Some(100_000), Some(200_000)), make_link(1, true, Some(300_000), Some(400_000)),
],
}];
let rows = compute_nvlink_throughputs(&prev, &curr, elapsed);
assert_eq!(rows.len(), 1);
let row = &rows[0];
let expected_tx_gbps = 5_000_000_000.0_f64 * 8.0 / 1_000_000_000.0;
let expected_rx_gbps = 5_000_000_000.0_f64 * 8.0 / 1_000_000_000.0;
assert!((row.tx_gbps - expected_tx_gbps).abs() < 1e-6);
assert!((row.rx_gbps - expected_rx_gbps).abs() < 1e-6);
let nv = row.nvlink.as_ref().unwrap();
assert_eq!(nv.links[0].tx_bytes, Some(100_000));
assert_eq!(nv.links[0].rx_bytes, Some(200_000));
assert_eq!(nv.links[1].tx_bytes, Some(300_000));
assert_eq!(nv.links[1].rx_bytes, Some(400_000));
}
#[test]
fn sort_order_stable_for_nvlink() {
let nvlink_ref = PortThroughput {
dev_name: "nvidia0".to_string(),
port: 2,
link_gbps: Some(150.0),
state: None,
tx_gbps: 1.0,
rx_gbps: 2.0,
tx_pkts_per_sec: 0.0,
rx_pkts_per_sec: 0.0,
rx_drops_per_sec: 0.0,
counter_rates: Vec::new(),
port_label: Some("2/3".to_string()),
nvlink: Some(NvLinkThroughputMeta {
gpu_index: 0,
gpu_name: "H100".to_string(),
active_links: 2,
links: Vec::new(),
metrics: None,
}),
xgmi: None,
class: DeviceClass::Nvlink,
};
let rdma_a_ref = PortThroughput {
dev_name: "mlx5_0".to_string(),
port: 1,
link_gbps: Some(100.0),
state: None,
tx_gbps: 3.0,
rx_gbps: 4.0,
tx_pkts_per_sec: 0.0,
rx_pkts_per_sec: 0.0,
rx_drops_per_sec: 0.0,
counter_rates: Vec::new(),
port_label: None,
nvlink: None,
xgmi: None,
class: DeviceClass::Rdma,
};
let rdma_b_ref = PortThroughput {
dev_name: "mlx5_1".to_string(),
port: 1,
link_gbps: Some(100.0),
state: None,
tx_gbps: 5.0,
rx_gbps: 6.0,
tx_pkts_per_sec: 0.0,
rx_pkts_per_sec: 0.0,
rx_drops_per_sec: 0.0,
counter_rates: Vec::new(),
port_label: None,
nvlink: None,
xgmi: None,
class: DeviceClass::Rdma,
};
let reference = vec![rdma_a_ref.clone(), nvlink_ref.clone(), rdma_b_ref.clone()];
let nvlink_ref_pos = 1usize;
let nvlink_avg = PortThroughput {
dev_name: "nvidia0".to_string(),
port: 3,
link_gbps: Some(150.0),
state: None,
tx_gbps: 1.5,
rx_gbps: 2.5,
tx_pkts_per_sec: 0.0,
rx_pkts_per_sec: 0.0,
rx_drops_per_sec: 0.0,
counter_rates: Vec::new(),
port_label: Some("3/3".to_string()),
nvlink: Some(NvLinkThroughputMeta {
gpu_index: 0,
gpu_name: "H100".to_string(),
active_links: 3,
links: Vec::new(),
metrics: None,
}),
xgmi: None,
class: DeviceClass::Nvlink,
};
let rdma_a_avg = PortThroughput {
dev_name: "mlx5_0".to_string(),
port: 1,
link_gbps: Some(100.0),
state: None,
tx_gbps: 3.5,
rx_gbps: 4.5,
tx_pkts_per_sec: 0.0,
rx_pkts_per_sec: 0.0,
rx_drops_per_sec: 0.0,
counter_rates: Vec::new(),
port_label: None,
nvlink: None,
xgmi: None,
class: DeviceClass::Rdma,
};
let rdma_b_avg = PortThroughput {
dev_name: "mlx5_1".to_string(),
port: 1,
link_gbps: Some(100.0),
state: None,
tx_gbps: 5.5,
rx_gbps: 6.5,
tx_pkts_per_sec: 0.0,
rx_pkts_per_sec: 0.0,
rx_drops_per_sec: 0.0,
counter_rates: Vec::new(),
port_label: None,
nvlink: None,
xgmi: None,
class: DeviceClass::Rdma,
};
let mut avgs = vec![rdma_b_avg.clone(), nvlink_avg.clone(), rdma_a_avg.clone()];
sort_by_throughput_order(&mut avgs, &reference);
assert_eq!(avgs[0].dev_name, "mlx5_0");
assert_eq!(avgs[1].dev_name, "nvidia0");
assert_eq!(avgs[2].dev_name, "mlx5_1");
assert_eq!(avgs[1].dev_name, nvlink_ref.dev_name);
assert_ne!(avgs[1].port, nvlink_ref.port);
assert_eq!(avgs[1].port_label, nvlink_avg.port_label);
let _ = nvlink_ref_pos;
}
}
#[cfg(test)]
mod sort_tests {
use super::*;
fn row(dev_name: &str, port: u32) -> PortThroughput {
PortThroughput {
dev_name: dev_name.to_string(),
port,
link_gbps: None,
state: None,
tx_gbps: 0.0,
rx_gbps: 0.0,
tx_pkts_per_sec: 0.0,
rx_pkts_per_sec: 0.0,
rx_drops_per_sec: 0.0,
counter_rates: Vec::new(),
port_label: None,
nvlink: None,
xgmi: None,
class: DeviceClass::Rdma,
}
}
fn names(rows: &[PortThroughput]) -> Vec<(String, u32)> {
rows.iter().map(|t| (t.dev_name.clone(), t.port)).collect()
}
#[test]
fn sort_by_device_order_sorts_index_numerically_not_lexically() {
let mut rows = vec![row("mlx5_10", 1), row("mlx5_2", 1), row("mlx5_9", 1)];
sort_by_device_order(&mut rows);
let sorted: Vec<_> = rows.iter().map(|t| t.dev_name.clone()).collect();
assert_eq!(sorted, vec!["mlx5_2", "mlx5_9", "mlx5_10"]);
}
#[test]
fn sort_by_device_order_orders_by_name_then_port() {
let mut rows = vec![
row("mlx5_10", 1),
row("mlx5_2", 2),
row("mlx5_2", 1),
row("bnxt_0", 1),
];
sort_by_device_order(&mut rows);
assert_eq!(
names(&rows),
vec![
("bnxt_0".to_string(), 1),
("mlx5_2".to_string(), 1),
("mlx5_2".to_string(), 2),
("mlx5_10".to_string(), 1),
]
);
}
}
#[cfg(test)]
mod tab_tests {
use super::*;
fn row(class: DeviceClass, name: &str) -> PortThroughput {
PortThroughput {
dev_name: name.to_string(),
port: 1,
link_gbps: None,
state: None,
tx_gbps: 0.0,
rx_gbps: 0.0,
tx_pkts_per_sec: 0.0,
rx_pkts_per_sec: 0.0,
rx_drops_per_sec: 0.0,
counter_rates: Vec::new(),
port_label: None,
nvlink: None,
xgmi: None,
class,
}
}
#[test]
fn detect_tabs_is_sticky() {
let mut seen = vec![DeviceClass::Rdma];
detect_tabs(&mut seen, &[row(DeviceClass::Xgmi, "amdgpu0")]);
assert_eq!(seen, vec![DeviceClass::Rdma, DeviceClass::Xgmi]);
detect_tabs(&mut seen, &[]);
assert_eq!(seen, vec![DeviceClass::Rdma, DeviceClass::Xgmi]);
detect_tabs(&mut seen, &[row(DeviceClass::Xgmi, "amdgpu0")]);
assert_eq!(seen.len(), 2);
}
#[test]
fn detect_tabs_orders_rdma_xgmi_nvlink() {
let mut seen = vec![DeviceClass::Rdma];
detect_tabs(&mut seen, &[row(DeviceClass::Nvlink, "nvidia0")]);
detect_tabs(&mut seen, &[row(DeviceClass::Xgmi, "amdgpu0")]);
assert_eq!(
seen,
vec![DeviceClass::Rdma, DeviceClass::Xgmi, DeviceClass::Nvlink]
);
}
#[test]
fn display_filters_by_active_tab() {
let mut app = App::new();
app.throughputs = vec![
row(DeviceClass::Rdma, "mlx5_0"),
row(DeviceClass::Rdma, "mlx5_1"),
row(DeviceClass::Xgmi, "amdgpu0"),
];
app.active_tab = DeviceClass::Rdma;
app.recompute_display();
assert_eq!(app.display_throughputs().len(), 2);
app.active_tab = DeviceClass::Xgmi;
app.recompute_display();
assert_eq!(app.display_throughputs().len(), 1);
assert_eq!(app.display_throughputs()[0].dev_name, "amdgpu0");
}
#[test]
fn cycle_wraps_and_remembers_selection() {
let mut app = App::new();
app.throughputs = vec![
row(DeviceClass::Rdma, "mlx5_0"),
row(DeviceClass::Rdma, "mlx5_1"),
row(DeviceClass::Rdma, "mlx5_2"),
row(DeviceClass::Xgmi, "amdgpu0"),
];
app.seen_tabs = vec![DeviceClass::Rdma, DeviceClass::Xgmi];
app.active_tab = DeviceClass::Rdma;
app.recompute_display();
app.selected_row = 2;
app.cycle_tab(true);
assert_eq!(app.active_tab, DeviceClass::Xgmi);
assert_eq!(app.selected_row, 0);
app.cycle_tab(true); assert_eq!(app.active_tab, DeviceClass::Rdma);
assert_eq!(app.selected_row, 2, "remembered per-tab cursor");
app.cycle_tab(false); assert_eq!(app.active_tab, DeviceClass::Xgmi);
}
#[test]
fn single_tab_cycle_is_noop() {
let mut app = App::new();
app.seen_tabs = vec![DeviceClass::Rdma];
app.active_tab = DeviceClass::Rdma;
app.cycle_tab(true);
assert_eq!(app.active_tab, DeviceClass::Rdma);
}
}
#[cfg(test)]
mod apply_snapshot_tests {
use super::*;
use crate::sampler::Snapshot;
use crate::stat::{HwCounter, PortStat};
fn port_stat(dev: &str, tx_bytes: u64) -> PortStat {
PortStat {
dev_name: dev.to_string(),
port: 1,
link_gbps: Some(400.0),
state: None,
counters: vec![HwCounter {
name: "tx_bytes".to_string(),
value: tx_bytes,
}],
}
}
fn sample<T>(data: T, taken_at: Instant) -> Option<crate::sampler::Sample<T>> {
Some(crate::sampler::Sample { data, taken_at })
}
fn snapshot(stats: Vec<PortStat>, taken_at: Instant) -> Snapshot {
Snapshot {
stats: sample(stats, taken_at),
ifstats: sample(Vec::new(), taken_at),
nvlink: sample(Vec::new(), taken_at),
xgmi: sample(Vec::new(), taken_at),
processes: Some(Vec::new()),
taken_at,
}
}
fn gpu_snapshot(tx_bytes: u64, taken_at: Instant) -> Snapshot {
let gpu = crate::nvlink::NvLinkSnapshot {
gpu_index: 0,
gpu_name: "test-gpu".to_string(),
metrics: Some(crate::gpu::GpuMetrics {
util_pct: Some(42),
..Default::default()
}),
link_count: 0,
link_gbps: None,
tx_bytes: Some(tx_bytes),
rx_bytes: Some(0),
links: Vec::new(),
};
Snapshot {
stats: None, ifstats: sample(Vec::new(), taken_at),
nvlink: sample(vec![gpu], taken_at),
xgmi: sample(Vec::new(), taken_at),
processes: Some(Vec::new()),
taken_at,
}
}
#[test]
fn gpu_util_flows_into_history() {
let mut app = App::new();
let t0 = Instant::now();
app.apply_snapshot(gpu_snapshot(0, t0));
app.apply_snapshot(gpu_snapshot(1_000, t0 + Duration::from_secs(2)));
let h = app.history.get("nvidia0").expect("history for nvidia0");
assert_eq!(h.util.len(), 2);
assert_eq!(h.util[0], 42.0);
}
#[test]
fn rates_use_per_subsystem_read_timestamps() {
let mut app = App::new();
let t0 = Instant::now();
let mut first = gpu_snapshot(0, t0);
first.nvlink.as_mut().unwrap().taken_at = t0 + Duration::from_secs(2);
app.apply_snapshot(first);
app.apply_snapshot(gpu_snapshot(1_000_000_000, t0 + Duration::from_secs(4)));
assert!((app.throughputs[0].tx_gbps - 4.0).abs() < 1e-9);
}
#[test]
fn first_snapshot_populates_rows_with_zero_rates() {
let mut app = App::new();
let t0 = Instant::now();
app.apply_snapshot(snapshot(vec![port_stat("mlx5_0", 1_000_000)], t0));
assert!(app.has_data);
assert_eq!(app.throughputs.len(), 1);
assert_eq!(app.throughputs[0].dev_name, "mlx5_0");
assert_eq!(app.throughputs[0].tx_gbps, 0.0); }
#[test]
fn second_snapshot_computes_rates_from_thread_timestamps() {
let mut app = App::new();
let t0 = Instant::now();
app.apply_snapshot(snapshot(vec![port_stat("mlx5_0", 0)], t0));
let t1 = t0 + Duration::from_secs(2);
app.apply_snapshot(snapshot(vec![port_stat("mlx5_0", 1_000_000_000)], t1));
assert!((app.throughputs[0].tx_gbps - 4.0).abs() < 1e-9);
}
#[test]
fn subsecond_duplicate_snapshot_is_skipped() {
let mut app = App::new();
let t0 = Instant::now();
app.apply_snapshot(snapshot(vec![port_stat("mlx5_0", 0)], t0));
app.apply_snapshot(snapshot(
vec![port_stat("mlx5_0", 999)],
t0 + Duration::from_millis(50),
));
assert_eq!(app.throughputs[0].tx_gbps, 0.0);
app.apply_snapshot(snapshot(
vec![port_stat("mlx5_0", 1_000_000_000)],
t0 + Duration::from_secs(2),
));
assert!((app.throughputs[0].tx_gbps - 4.0).abs() < 1e-9);
}
#[test]
fn failed_rdma_read_holds_rdma_state_only() {
let mut app = App::new();
let t0 = Instant::now();
app.apply_snapshot(snapshot(vec![port_stat("mlx5_0", 0)], t0));
let mut bad = snapshot(Vec::new(), t0 + Duration::from_secs(2));
bad.stats = None;
app.apply_snapshot(bad);
assert_eq!(app.throughputs.len(), 1);
app.apply_snapshot(snapshot(
vec![port_stat("mlx5_0", 1_000_000_000)],
t0 + Duration::from_secs(4),
));
assert!((app.throughputs[0].tx_gbps - 2.0).abs() < 1e-9);
}
#[test]
fn gpu_data_flows_when_rdma_read_always_fails() {
let mut app = App::new();
let t0 = Instant::now();
app.apply_snapshot(gpu_snapshot(0, t0));
assert!(app.has_data);
assert_eq!(app.throughputs.len(), 1);
assert_eq!(app.throughputs[0].tx_gbps, 0.0); app.apply_snapshot(gpu_snapshot(1_000_000_000, t0 + Duration::from_secs(2)));
assert!((app.throughputs[0].tx_gbps - 4.0).abs() < 1e-9);
}
#[test]
fn slow_pass_is_not_stale_once_applied() {
let mut app = App::new();
let pass_started = Instant::now() - Duration::from_secs(4);
app.apply_snapshot(snapshot(vec![port_stat("mlx5_0", 0)], pass_started));
assert_eq!(app.stale_secs(), None);
}
}