use crate::netlink::*;
use crate::rdma::*;
use std::io;
#[derive(Clone, Debug)]
pub struct HwCounter {
pub name: String,
pub value: u64,
}
#[derive(Clone, Debug)]
pub struct PortStat {
pub dev_name: String,
pub port: u32,
pub link_gbps: Option<f64>,
pub state: Option<String>,
pub counters: Vec<HwCounter>,
}
impl PortStat {
pub fn new(dev_name: String, port: u32) -> Self {
Self {
dev_name,
port,
link_gbps: None,
state: None,
counters: Vec::new(),
}
}
pub fn counter_value(&self, name: &str) -> Option<u64> {
self.counters
.iter()
.find(|c| c.name == name)
.map(|c| c.value)
}
}
#[cfg(test)]
pub(crate) fn port_stat(dev: &str, port: u32, counters: &[(&str, u64)]) -> PortStat {
let mut stat = PortStat::new(dev.to_string(), port);
for (name, value) in counters {
stat.counters.push(HwCounter {
name: name.to_string(),
value: *value,
});
}
stat
}
pub const THROUGHPUT_COUNTERS: &[&str] =
&["tx_bytes", "rx_bytes", "tx_pkts", "rx_pkts", "rx_drops"];
pub const EXTRA_COUNTERS: &[&str] = &[
"send_bytes",
"send_wrs",
"recv_bytes",
"recv_wrs",
"rdma_write_bytes",
"rdma_write_wrs",
"rdma_write_wr_err",
"rdma_write_recv_bytes",
"rdma_read_bytes",
"rdma_read_wrs",
"rdma_read_wr_err",
"rdma_read_resp_bytes",
"retrans_bytes",
"retrans_pkts",
"retrans_timeout_events",
"unresponsive_remote_events",
"impaired_remote_conn_events",
];
struct RdmaDev {
idx: u32,
name: String,
num_ports: u32,
}
fn parse_dev(nlmsg: &NlMsg) -> Option<RdmaDev> {
let mut dev = RdmaDev {
idx: 0,
name: String::new(),
num_ports: 1,
};
for nla in nlmsg.attrs() {
match nla.attr_type {
RDMA_NLDEV_ATTR_DEV_INDEX => dev.idx = nla.u32(),
RDMA_NLDEV_ATTR_DEV_NAME => dev.name = nla.str().to_string(),
RDMA_NLDEV_ATTR_PORT_INDEX => dev.num_ports = nla.u32(),
_ => {}
}
}
(!dev.name.is_empty()).then_some(dev)
}
fn parse_hw_counter(entry: &Nla) -> HwCounter {
let mut name = String::new();
let mut value = 0u64;
for f in entry.nested() {
match f.attr_type {
RDMA_NLDEV_ATTR_STAT_HWCOUNTER_ENTRY_NAME => name = f.str().to_string(),
RDMA_NLDEV_ATTR_STAT_HWCOUNTER_ENTRY_VALUE => value = f.u64(),
_ => {}
}
}
HwCounter { name, value }
}
fn parse_port_counters(nlmsg: &NlMsg) -> Option<PortStat> {
let mut stat = PortStat::new(String::new(), 0);
for nla in nlmsg.attrs() {
match nla.attr_type {
RDMA_NLDEV_ATTR_DEV_NAME => stat.dev_name = nla.str().to_string(),
RDMA_NLDEV_ATTR_PORT_INDEX => stat.port = nla.u32(),
RDMA_NLDEV_ATTR_STAT_HWCOUNTERS => {
stat.counters = nla
.nested()
.filter(|e| e.attr_type == RDMA_NLDEV_ATTR_STAT_HWCOUNTER_ENTRY)
.map(|e| parse_hw_counter(&e))
.collect();
}
_ => {}
}
}
if stat.dev_name.is_empty() {
return None;
}
Some(stat)
}
fn parse_port_stat(nlmsg: &NlMsg) -> Option<PortStat> {
let mut stat = parse_port_counters(nlmsg)?;
fill_missing_from_sysfs(&mut stat);
stat.link_gbps = read_port_link_gbps(&stat.dev_name, stat.port);
Some(stat)
}
pub struct CounterReader {
sock: NlSocket,
dev_name: String,
dev_idx: u32,
port: u32,
seq: u32,
}
impl CounterReader {
pub fn open(dev_name: &str, port: u32) -> io::Result<Self> {
let sock = NlSocket::open(NETLINK_RDMA)?;
let dev = enumerate_devices(&sock)?
.into_iter()
.find(|dev| dev.name == dev_name && (1..=dev.num_ports).contains(&port))
.ok_or_else(|| {
io::Error::new(
io::ErrorKind::NotFound,
format!("RDMA port {dev_name}/{port}"),
)
})?;
Ok(Self {
sock,
dev_name: dev.name,
dev_idx: dev.idx,
port,
seq: 2,
})
}
pub fn read(&mut self) -> io::Result<PortStat> {
self.seq = self.seq.wrapping_add(1);
let msg = port_request(RDMA_NLDEV_CMD_STAT_GET, self.dev_idx, self.port, self.seq);
let mut stat = None;
for buf in self.sock.request(msg)? {
for response in NlMsgIter::new(&buf) {
if response.is_error() {
counter_response_error(response.payload)?;
} else if !response.is_done() {
stat = parse_port_counters(&response).or(stat);
}
}
}
finish_counter_sample(stat, &self.dev_name, self.port, fill_missing_from_sysfs)
}
}
fn counter_response_error(payload: &[u8]) -> io::Result<()> {
let code = payload
.get(..4)
.map(|bytes| i32::from_ne_bytes(bytes.try_into().unwrap()))
.ok_or_else(|| io::Error::new(io::ErrorKind::InvalidData, "truncated netlink error"))?;
if code == 0 || code == -libc::EOPNOTSUPP {
Ok(())
} else {
Err(io::Error::from_raw_os_error(code.saturating_neg()))
}
}
fn finish_counter_sample(
stat: Option<PortStat>,
dev_name: &str,
port: u32,
fallback: impl FnOnce(&mut PortStat),
) -> io::Result<PortStat> {
let mut stat = stat.unwrap_or_else(|| PortStat::new(dev_name.to_string(), port));
if stat.counter_value("tx_bytes").is_none() || stat.counter_value("rx_bytes").is_none() {
fallback(&mut stat);
}
if stat.counters.is_empty() {
return Err(io::Error::other(format!(
"no counters for {dev_name}/{port}"
)));
}
Ok(stat)
}
fn port_state_name(v: u8) -> Option<String> {
let s = match v {
1 => "DOWN",
2 => "INIT",
3 => "ARMED",
4 => "ACTIVE",
5 => "ACTIVE_DEFER",
_ => return None,
};
Some(s.to_string())
}
fn read_port_state_sysfs(dev_name: &str, port: u32) -> Option<String> {
let path = format!("/sys/class/infiniband/{}/ports/{}/state", dev_name, port);
let raw = std::fs::read_to_string(path).ok()?;
Some(raw.trim().rsplit(": ").next()?.to_string())
}
fn parse_port_state(nlmsg: &NlMsg) -> Option<String> {
nlmsg
.attrs()
.find(|nla| nla.attr_type == RDMA_NLDEV_ATTR_PORT_STATE)
.and_then(|nla| port_state_name(nla.u8()))
}
fn port_request(cmd: u32, dev_idx: u32, port: u32, seq: u32) -> Vec<u8> {
NlMsgBuilder::new(
rdma_nl_get_type(RDMA_NL_NLDEV, cmd),
NLM_F_REQUEST | NLM_F_ACK,
seq,
)
.put_u32(RDMA_NLDEV_ATTR_DEV_INDEX, dev_idx)
.put_u32(RDMA_NLDEV_ATTR_PORT_INDEX, port)
.build()
}
fn query_port_state(sock: &NlSocket, dev: &RdmaDev, port: u32, seq: u32) -> Option<String> {
let msg = port_request(RDMA_NLDEV_CMD_PORT_GET, dev.idx, port, seq);
collect_responses(sock, msg, parse_port_state)
.ok()
.and_then(|v| v.into_iter().next())
.or_else(|| read_port_state_sysfs(&dev.name, port))
}
fn read_port_link_gbps(dev_name: &str, port: u32) -> Option<f64> {
let path = format!("/sys/class/infiniband/{}/ports/{}/rate", dev_name, port);
let raw = std::fs::read_to_string(path).ok()?;
raw.split_whitespace().next()?.parse::<f64>().ok()
}
const SYSFS_DATA_COUNTERS: &[&str] = &["port_xmit_data", "port_rcv_data"];
const SYSFS_SYNTH: &[(&str, &str, u64)] = &[
("tx_bytes", "port_xmit_data", 4),
("rx_bytes", "port_rcv_data", 4),
("tx_pkts", "port_xmit_packets", 1),
("rx_pkts", "port_rcv_packets", 1),
];
const HW_COUNTER_ALIASES: &[(&str, &str)] = &[
("tx_bytes", "hw_tx_bytes_cnt"),
("rx_bytes", "hw_rx_bytes_cnt"),
("tx_pkts", "hw_tx_packets_cnt"),
("rx_pkts", "hw_rx_packets_cnt"),
("rx_drops", "hw_rx_disable_drop_cnt"),
("tx_bytes", "tx_bytes_cnt"),
("rx_bytes", "rx_bytes_cnt"),
("tx_pkts", "tx_packets_cnt"),
("rx_pkts", "rx_packets_cnt"),
("rx_drops", "rx_disable_drop_cnt"),
];
fn fill_missing_from_sysfs(stat: &mut PortStat) {
let dir = format!(
"/sys/class/infiniband/{}/ports/{}/counters",
stat.dev_name, stat.port
);
if let Ok(entries) = std::fs::read_dir(&dir) {
for entry in entries.flatten() {
let Some(name) = entry.file_name().to_str().map(String::from) else {
continue;
};
if stat.counters.iter().any(|c| c.name == name) {
continue;
}
let Ok(raw) = std::fs::read_to_string(entry.path()) else {
continue;
};
let Ok(v) = raw.trim().parse::<u64>() else {
continue;
};
let value = if SYSFS_DATA_COUNTERS.contains(&name.as_str()) {
v.saturating_mul(4)
} else {
v
};
stat.counters.push(HwCounter { name, value });
}
}
for (synth_name, src_name, mult) in SYSFS_SYNTH {
if stat.counters.iter().any(|c| c.name == *synth_name) {
continue;
}
let Some(src) = stat.counters.iter().find(|c| c.name == *src_name) else {
continue;
};
let base = if SYSFS_DATA_COUNTERS.contains(src_name) {
src.value
} else {
src.value.saturating_mul(*mult)
};
stat.counters.push(HwCounter {
name: (*synth_name).to_string(),
value: base,
});
}
for (canonical, alias) in HW_COUNTER_ALIASES {
if stat.counters.iter().any(|c| c.name == *canonical) {
continue;
}
if let Some(src) = stat.counters.iter().find(|c| c.name == *alias) {
let value = src.value;
stat.counters.push(HwCounter {
name: canonical.to_string(),
value,
});
}
}
}
fn enumerate_devices(sock: &NlSocket) -> io::Result<Vec<RdmaDev>> {
let msg = NlMsgBuilder::new(
rdma_nl_get_type(RDMA_NL_NLDEV, RDMA_NLDEV_CMD_GET),
NLM_F_REQUEST | NLM_F_ACK | NLM_F_DUMP,
1,
)
.build();
collect_responses(sock, msg, parse_dev)
}
fn query_port_stats(
sock: &NlSocket,
dev_idx: u32,
port: u32,
seq: u32,
) -> io::Result<Vec<PortStat>> {
let msg = port_request(RDMA_NLDEV_CMD_STAT_GET, dev_idx, port, seq);
collect_responses(sock, msg, parse_port_stat)
}
pub fn read_all_stats() -> io::Result<Vec<PortStat>> {
let sock = NlSocket::open(NETLINK_RDMA)?;
let devs = enumerate_devices(&sock)?;
let sock = NlSocket::open(NETLINK_RDMA)?;
let mut all = Vec::new();
let mut seq = 100u32;
for dev in &devs {
for port in 1..=dev.num_ports {
let state = query_port_state(&sock, dev, port, seq);
seq += 1;
for mut stat in query_port_stats(&sock, dev.idx, port, seq)? {
stat.state = state.clone();
all.push(stat);
}
seq += 1;
}
}
Ok(all)
}
#[derive(Clone, Debug)]
pub struct QpInfo {
pub dev_name: String,
pub lqpn: u32,
pub qp_type: u8,
pub state: u8,
pub pid: u32,
pub comm: String,
}
fn parse_qp_response(nlmsg: &NlMsg, dev_name: &str) -> Vec<QpInfo> {
let mut name = dev_name.to_string();
let mut qps = Vec::new();
for nla in nlmsg.attrs() {
match nla.attr_type {
RDMA_NLDEV_ATTR_DEV_NAME => name = nla.str().to_string(),
RDMA_NLDEV_ATTR_RES_QP => {
for entry in nla.nested() {
if entry.attr_type == RDMA_NLDEV_ATTR_RES_QP_ENTRY {
if let Some(qp) = parse_single_qp(&entry, &name) {
qps.push(qp);
}
}
}
}
_ => {}
}
}
qps
}
fn parse_single_qp(nla: &Nla, dev_name: &str) -> Option<QpInfo> {
let mut qp = QpInfo {
dev_name: dev_name.to_string(),
lqpn: 0,
qp_type: 0,
state: 0,
pid: 0,
comm: String::new(),
};
for attr in nla.nested() {
match attr.attr_type {
RDMA_NLDEV_ATTR_RES_LQPN => qp.lqpn = attr.u32(),
RDMA_NLDEV_ATTR_RES_TYPE if !attr.data.is_empty() => {
qp.qp_type = attr.data[0];
}
RDMA_NLDEV_ATTR_RES_STATE if !attr.data.is_empty() => {
qp.state = attr.data[0];
}
RDMA_NLDEV_ATTR_RES_PID => {
qp.pid = attr.u32();
qp.comm = read_proc_comm(qp.pid);
}
RDMA_NLDEV_ATTR_RES_KERN_NAME => qp.comm = attr.str().to_string(),
_ => {}
}
}
(qp.pid > 0).then_some(qp)
}
fn read_proc_comm(pid: u32) -> String {
std::fs::read_to_string(format!("/proc/{}/comm", pid))
.unwrap_or_default()
.trim()
.to_string()
}
pub fn read_all_qps() -> io::Result<Vec<QpInfo>> {
let sock = NlSocket::open(NETLINK_RDMA)?;
let devs = enumerate_devices(&sock)?;
drop(sock);
let mut all = Vec::new();
let mut seq = 200u32;
for dev in &devs {
let sock = NlSocket::open(NETLINK_RDMA)?;
let msg = NlMsgBuilder::new(
rdma_nl_get_type(RDMA_NL_NLDEV, RDMA_NLDEV_CMD_RES_QP_GET),
NLM_F_REQUEST | NLM_F_ACK | NLM_F_DUMP,
seq,
)
.put_u32(RDMA_NLDEV_ATTR_DEV_INDEX, dev.idx)
.build();
if let Ok(bufs) = sock.request(msg) {
for buf in bufs {
for nlmsg in NlMsgIter::new(&buf) {
if nlmsg.is_done() || nlmsg.is_error() {
continue;
}
all.extend(parse_qp_response(&nlmsg, &dev.name));
}
}
}
seq += 1;
}
Ok(all)
}
#[derive(Clone, Debug)]
pub struct ProcessRdmaInfo {
pub pid: u32,
pub dev_name: String,
pub qp_count: usize,
pub user: String,
pub nice: i32,
pub state: char,
pub virt_kb: u64,
pub res_kb: u64,
pub shr_kb: u64,
pub mem_pct: f32,
pub threads: u32,
pub cmdline: String,
}
struct ProcStat {
state: char,
nice: i32,
vsize: u64,
rss_pages: u64,
num_threads: u32,
}
fn read_proc_stat(pid: u32) -> Option<ProcStat> {
let data = std::fs::read_to_string(format!("/proc/{}/stat", pid)).ok()?;
let rest = data.rsplit_once(')')?.1;
let fields: Vec<&str> = rest.split_whitespace().collect();
if fields.len() < 22 {
return None;
}
Some(ProcStat {
state: fields[0].chars().next().unwrap_or('?'),
nice: fields[16].parse().unwrap_or(0),
num_threads: fields[17].parse().unwrap_or(1),
vsize: fields[20].parse().unwrap_or(0),
rss_pages: fields[21].parse().unwrap_or(0),
})
}
fn read_proc_shr_kb(pid: u32) -> u64 {
std::fs::read_to_string(format!("/proc/{}/statm", pid))
.ok()
.and_then(|s| s.split_whitespace().nth(2)?.parse::<u64>().ok())
.unwrap_or(0)
* 4 }
fn read_proc_uid(pid: u32) -> u32 {
std::fs::read_to_string(format!("/proc/{}/status", pid))
.ok()
.and_then(|s| {
s.lines()
.find(|l| l.starts_with("Uid:"))
.and_then(|l| l.split_whitespace().nth(1)?.parse().ok())
})
.unwrap_or(0)
}
fn uid_to_name(uid: u32) -> String {
std::fs::read_to_string("/etc/passwd")
.ok()
.and_then(|s| {
s.lines()
.find(|l| l.split(':').nth(2) == Some(&uid.to_string()))
.map(|l| l.split(':').next().unwrap_or("?").to_string())
})
.unwrap_or_else(|| uid.to_string())
}
fn read_proc_cmdline(pid: u32) -> String {
std::fs::read_to_string(format!("/proc/{}/cmdline", pid))
.unwrap_or_default()
.replace('\0', " ")
.trim()
.to_string()
}
fn total_memory_kb() -> u64 {
std::fs::read_to_string("/proc/meminfo")
.ok()
.and_then(|s| {
s.lines()
.find(|l| l.starts_with("MemTotal:"))
.and_then(|l| l.split_whitespace().nth(1)?.parse().ok())
})
.unwrap_or(1)
}
fn proc_mem_stats(ps: &Option<ProcStat>) -> (char, i32, u64, u64, u32) {
match ps {
Some(s) => (
s.state,
s.nice,
s.vsize / 1024,
s.rss_pages * 4,
s.num_threads,
),
None => ('?', 0, 0, 0, 0),
}
}
fn mem_percent(res_kb: u64) -> f32 {
let total = total_memory_kb();
if total > 0 {
(res_kb as f32 / total as f32) * 100.0
} else {
0.0
}
}
fn resolve_user(uid: u32, ps: &Option<ProcStat>) -> String {
if uid > 0 || ps.is_some() {
uid_to_name(uid)
} else {
"?".into()
}
}
fn resolve_cmdline(pid: u32, comm: &str) -> String {
let cmdline = read_proc_cmdline(pid);
if !cmdline.is_empty() {
cmdline
} else {
comm.to_string()
}
}
fn enrich_process(pid: u32, comm: &str, dev_name: &str, qp_count: usize) -> ProcessRdmaInfo {
let ps = read_proc_stat(pid);
let (state, nice, virt_kb, res_kb, threads) = proc_mem_stats(&ps);
ProcessRdmaInfo {
pid,
dev_name: dev_name.to_string(),
qp_count,
user: resolve_user(read_proc_uid(pid), &ps),
nice,
state,
virt_kb,
res_kb,
shr_kb: read_proc_shr_kb(pid),
mem_pct: mem_percent(res_kb),
threads,
cmdline: resolve_cmdline(pid, comm),
}
}
pub fn aggregate_by_process(qps: &[QpInfo]) -> Vec<ProcessRdmaInfo> {
use std::collections::HashMap;
let mut counts: HashMap<(u32, String), (&QpInfo, usize)> = HashMap::new();
for qp in qps {
if qp.pid == 0 {
continue;
}
let key = (qp.pid, qp.dev_name.clone());
let entry = counts.entry(key).or_insert((qp, 0));
entry.1 += 1;
}
let mut result: Vec<ProcessRdmaInfo> = counts
.into_values()
.map(|(qp, count)| enrich_process(qp.pid, &qp.comm, &qp.dev_name, count))
.collect();
result.sort_by(|a, b| b.qp_count.cmp(&a.qp_count).then(a.pid.cmp(&b.pid)));
result
}
#[cfg(test)]
mod counter_recording_tests {
use super::*;
#[test]
fn empty_netlink_response_can_use_sysfs_counters() {
let stat = finish_counter_sample(None, "test0", 1, |stat| {
assert_eq!(stat.dev_name, "test0");
assert_eq!(stat.port, 1);
stat.counters.push(HwCounter {
name: "tx_bytes".into(),
value: 42,
});
})
.unwrap();
assert_eq!(stat.counter_value("tx_bytes"), Some(42));
assert!(finish_counter_sample(None, "test0", 1, |_| {}).is_err());
}
#[test]
fn unsupported_stats_allow_fallback_but_permission_errors_surface() {
assert!(counter_response_error(&(-libc::EOPNOTSUPP).to_ne_bytes()).is_ok());
assert!(counter_response_error(&0_i32.to_ne_bytes()).is_ok());
assert_eq!(
counter_response_error(&(-libc::EACCES).to_ne_bytes())
.unwrap_err()
.kind(),
io::ErrorKind::PermissionDenied
);
assert!(counter_response_error(&[]).is_err());
}
}