feat: add mlx5 DC (Dynamically Connected) transport

Cargo.toml

@@ -14,8 +14,11 @@ edition = "2021"
 
 # See more keys and their definitions at https://doc.rust-lang.org/cargo/reference/manifest.html
 
+[features]
+mlx5 = ["rdma-mummy-sys/mlx5"]
+
 [dependencies]
-rdma-mummy-sys = "0.2.3"
+rdma-mummy-sys = { path = "../rdma-mummy-sys" }
 tabled = "0.18"
 libc = "0.2"
 os_socketaddr = "0.2"
@@ -30,7 +33,6 @@ clap = { version = "4.5", features = ["derive"] }
 rand = "0.9"
 postcard = { version = "1.1", features = ["alloc"] }
 quanta = "0.12"
-byte-unit = "5.1"
 proptest = "1.6"
 anyhow = "1.0"
 termtree = "0.5"

examples/rc_pingpong.rs

@@ -31,7 +31,17 @@ use sideway::ibverbs::queue_pair::{
 };
 use sideway::ibverbs::AccessFlags;
 
-use byte_unit::{Byte, UnitType};
+/// Format a byte count with the largest binary unit that keeps it >= 1.
+fn binary_unit(bytes: f64) -> String {
+    const UNITS: [&str; 7] = ["B", "KiB", "MiB", "GiB", "TiB", "PiB", "EiB"];
+    let mut value = bytes;
+    let mut unit = 0;
+    while value >= 1024.0 && unit + 1 < UNITS.len() {
+        value /= 1024.0;
+        unit += 1;
+    }
+    format!("{value:.2} {}", UNITS[unit])
+}
 
 const SEND_WR_ID: u64 = 0;
 const RECV_WR_ID: u64 = 1;
@@ -424,12 +434,10 @@ fn main() -> anyhow::Result<()> {
     // bi-directional bandwidth
     let bytes_per_second = bytes as f64 / time.as_secs_f64();
     println!(
-        "{} bytes in {:.2} seconds = {:.2}/s",
+        "{} bytes in {:.2} seconds = {}/s",
         bytes,
         time.as_secs_f64(),
-        Byte::from_f64(bytes_per_second)
-            .unwrap()
-            .get_appropriate_unit(UnitType::Binary)
+        binary_unit(bytes_per_second)
     );
     println!(
         "{} iters in {:.2} seconds = {:#.2?}/iter",

examples/rc_pingpong_split.rs

@@ -45,7 +45,17 @@ use sideway::ibverbs::queue_pair::{
 };
 use sideway::ibverbs::AccessFlags;
 
-use byte_unit::{Byte, UnitType};
+/// Format a byte count with the largest binary unit that keeps it >= 1.
+fn binary_unit(bytes: f64) -> String {
+    const UNITS: [&str; 7] = ["B", "KiB", "MiB", "GiB", "TiB", "PiB", "EiB"];
+    let mut value = bytes;
+    let mut unit = 0;
+    while value >= 1024.0 && unit + 1 < UNITS.len() {
+        value /= 1024.0;
+        unit += 1;
+    }
+    format!("{value:.2} {}", UNITS[unit])
+}
 
 const SEND_WR_ID: u64 = 0;
 const RECV_WR_ID: u64 = 1;
@@ -508,12 +518,10 @@ fn main() -> anyhow::Result<()> {
     // bi-directional bandwidth
     let bytes_per_second = bytes as f64 / time.as_secs_f64();
     println!(
-        "{} bytes in {:.2} seconds = {:.2}/s",
+        "{} bytes in {:.2} seconds = {}/s",
         bytes,
         time.as_secs_f64(),
-        Byte::from_f64(bytes_per_second)
-            .unwrap()
-            .get_appropriate_unit(UnitType::Binary)
+        binary_unit(bytes_per_second)
     );
     println!(
         "{} iters in {:.2} seconds = {:#.2?}/iter",

src/ibverbs/device_context.rs

@@ -5,16 +5,18 @@ use std::fmt;
 use std::fs;
 use std::io;
 use std::mem::MaybeUninit;
+use std::os::fd::RawFd;
 use std::ptr::{self, NonNull};
 use std::sync::Arc;
 
 use bitmask_enum::bitmask;
 use rdma_mummy_sys::{
-    ibv_alloc_pd, ibv_close_device, ibv_context, ibv_device_attr_ex, ibv_get_device_guid, ibv_get_device_name,
-    ibv_gid_entry, ibv_mtu, ibv_port_attr, ibv_port_state, ibv_query_device_ex, ibv_query_gid, ibv_query_gid_ex,
-    ibv_query_gid_table, ibv_query_gid_type, ibv_query_port, ibv_query_rt_values_ex, ibv_values_ex, ibv_values_mask,
-    IBV_GID_TYPE_IB, IBV_GID_TYPE_ROCE_V1, IBV_GID_TYPE_ROCE_V2, IBV_GID_TYPE_SYSFS_IB_ROCE_V1,
-    IBV_GID_TYPE_SYSFS_ROCE_V2, IBV_LINK_LAYER_ETHERNET, IBV_LINK_LAYER_INFINIBAND, IBV_LINK_LAYER_UNSPECIFIED,
+    ibv_ack_async_event, ibv_alloc_pd, ibv_async_event, ibv_close_device, ibv_context, ibv_device_attr_ex,
+    ibv_event_type, ibv_get_async_event, ibv_get_device_guid, ibv_get_device_name, ibv_gid_entry, ibv_mtu,
+    ibv_port_attr, ibv_port_state, ibv_query_device_ex, ibv_query_gid, ibv_query_gid_ex, ibv_query_gid_table,
+    ibv_query_gid_type, ibv_query_port, ibv_query_rt_values_ex, ibv_values_ex, ibv_values_mask, IBV_GID_TYPE_IB,
+    IBV_GID_TYPE_ROCE_V1, IBV_GID_TYPE_ROCE_V2, IBV_GID_TYPE_SYSFS_IB_ROCE_V1, IBV_GID_TYPE_SYSFS_ROCE_V2,
+    IBV_LINK_LAYER_ETHERNET, IBV_LINK_LAYER_INFINIBAND, IBV_LINK_LAYER_UNSPECIFIED,
 };
 use serde::{Deserialize, Serialize};
 
@@ -84,6 +86,20 @@ pub enum QueryPortErrorKind {
     Ibverbs(#[from] io::Error),
 }
 
+/// Error returned by [`DeviceContext::next_async_event`] for reading an asynchronous event.
+#[derive(Debug, thiserror::Error)]
+#[error("failed to get async event")]
+#[non_exhaustive]
+pub struct GetAsyncEventError(#[from] pub GetAsyncEventErrorKind);
+
+/// The enum type for [`GetAsyncEventError`].
+#[derive(Debug, thiserror::Error)]
+#[error(transparent)]
+#[non_exhaustive]
+pub enum GetAsyncEventErrorKind {
+    Ibverbs(#[from] io::Error),
+}
+
 /// Error returned by [`DeviceContext::query_gid_table`] for querying RDMA device's GID table, which
 /// includes all GID entries on an RDMA device.
 #[derive(Debug, thiserror::Error)]
@@ -444,6 +460,126 @@ impl From<u8> for PhysicalState {
     }
 }
 
+/// The kind of an asynchronous event reported by the device.
+///
+/// Unrecognised values are preserved rather than rejected: the set grows with
+/// the kernel, and an event a caller does not know about still has to be
+/// acknowledged.
+#[derive(Debug, Clone, Copy, PartialEq, Eq)]
+#[non_exhaustive]
+pub enum AsyncEventType {
+    /// A completion queue is in an error state.
+    CqError,
+    /// A queue pair hit an error that moved it out of a usable state.
+    QpFatal,
+    /// A request to a queue pair was malformed.
+    QpRequestError,
+    /// A request violated a queue pair's access permissions.
+    QpAccessError,
+    /// Communication has been established on a queue pair.
+    CommunicationEstablished,
+    /// A send queue has drained.
+    SendQueueDrained,
+    /// A path has been migrated.
+    PathMigrated,
+    /// A path migration failed.
+    PathMigrationError,
+    /// The device is in an unrecoverable error state.
+    DeviceFatal,
+    /// A port became active and can now carry traffic.
+    PortActive,
+    /// A port left the active state.
+    PortError,
+    /// The subnet manager assigned this port a different LID. Anything holding
+    /// the old one -- address vectors, published endpoints -- is now stale.
+    LidChange,
+    /// The port's partition key table changed.
+    PkeyChange,
+    /// A different subnet manager is now managing this port.
+    SmChange,
+    /// A shared receive queue is in an error state.
+    SrqError,
+    /// A shared receive queue reached its limit.
+    SrqLimitReached,
+    /// The last work queue entry was reached on a queue pair.
+    QpLastWqeReached,
+    /// The subnet manager asked clients to re-register. Typically follows an
+    /// SM restart, and implies the port's configuration may have changed.
+    ClientReregister,
+    /// The port's GID table changed.
+    GidChange,
+    /// A work queue is in an error state.
+    WqFatal,
+    /// The device's link speed changed.
+    DeviceSpeedChange,
+    /// A type this build does not recognise, carried through so it can still
+    /// be acknowledged.
+    Unknown(u32),
+}
+
+impl From<ibv_event_type> for AsyncEventType {
+    fn from(event_type: ibv_event_type) -> Self {
+        // Matched on the discriminant rather than the variants: the provider
+        // can report a value this build's bindings do not name, and treating
+        // that as one of the known variants would be a lie.
+        match event_type as u32 {
+            v if v == ibv_event_type::IBV_EVENT_CQ_ERR as u32 => AsyncEventType::CqError,
+            v if v == ibv_event_type::IBV_EVENT_QP_FATAL as u32 => AsyncEventType::QpFatal,
+            v if v == ibv_event_type::IBV_EVENT_QP_REQ_ERR as u32 => AsyncEventType::QpRequestError,
+            v if v == ibv_event_type::IBV_EVENT_QP_ACCESS_ERR as u32 => AsyncEventType::QpAccessError,
+            v if v == ibv_event_type::IBV_EVENT_COMM_EST as u32 => AsyncEventType::CommunicationEstablished,
+            v if v == ibv_event_type::IBV_EVENT_SQ_DRAINED as u32 => AsyncEventType::SendQueueDrained,
+            v if v == ibv_event_type::IBV_EVENT_PATH_MIG as u32 => AsyncEventType::PathMigrated,
+            v if v == ibv_event_type::IBV_EVENT_PATH_MIG_ERR as u32 => AsyncEventType::PathMigrationError,
+            v if v == ibv_event_type::IBV_EVENT_DEVICE_FATAL as u32 => AsyncEventType::DeviceFatal,
+            v if v == ibv_event_type::IBV_EVENT_PORT_ACTIVE as u32 => AsyncEventType::PortActive,
+            v if v == ibv_event_type::IBV_EVENT_PORT_ERR as u32 => AsyncEventType::PortError,
+            v if v == ibv_event_type::IBV_EVENT_LID_CHANGE as u32 => AsyncEventType::LidChange,
+            v if v == ibv_event_type::IBV_EVENT_PKEY_CHANGE as u32 => AsyncEventType::PkeyChange,
+            v if v == ibv_event_type::IBV_EVENT_SM_CHANGE as u32 => AsyncEventType::SmChange,
+            v if v == ibv_event_type::IBV_EVENT_SRQ_ERR as u32 => AsyncEventType::SrqError,
+            v if v == ibv_event_type::IBV_EVENT_SRQ_LIMIT_REACHED as u32 => AsyncEventType::SrqLimitReached,
+            v if v == ibv_event_type::IBV_EVENT_QP_LAST_WQE_REACHED as u32 => AsyncEventType::QpLastWqeReached,
+            v if v == ibv_event_type::IBV_EVENT_CLIENT_REREGISTER as u32 => AsyncEventType::ClientReregister,
+            v if v == ibv_event_type::IBV_EVENT_GID_CHANGE as u32 => AsyncEventType::GidChange,
+            v if v == ibv_event_type::IBV_EVENT_WQ_FATAL as u32 => AsyncEventType::WqFatal,
+            v if v == ibv_event_type::IBV_EVENT_DEVICE_SPEED_CHANGE as u32 => AsyncEventType::DeviceSpeedChange,
+            other => AsyncEventType::Unknown(other),
+        }
+    }
+}
+
+impl AsyncEventType {
+    /// Whether the event describes a port rather than a queue, and therefore
+    /// carries a port number.
+    fn is_port_scoped(&self) -> bool {
+        matches!(
+            self,
+            AsyncEventType::PortActive
+                | AsyncEventType::PortError
+                | AsyncEventType::LidChange
+                | AsyncEventType::PkeyChange
+                | AsyncEventType::SmChange
+                | AsyncEventType::ClientReregister
+                | AsyncEventType::GidChange
+        )
+    }
+}
+
+/// One asynchronous event, already acknowledged.
+///
+/// The payload is copied out before the acknowledgement, so holding this does
+/// not hold a device resource.
+#[derive(Debug, Clone, Copy, PartialEq, Eq)]
+#[non_exhaustive]
+pub struct AsyncEvent {
+    /// What happened.
+    pub event_type: AsyncEventType,
+    /// The port it happened on, for port-scoped events. The queue-scoped
+    /// events name a queue instead, which this does not carry.
+    pub port_num: Option<u8>,
+}
+
 /// The attributes of a port of an RDMA device context.
 pub struct PortAttr {
     attr: ibv_port_attr,
@@ -465,6 +601,16 @@ impl PortAttr {
         self.attr.gid_tbl_len
     }
 
+    /// Get the base local identifier (LID) assigned to this port.
+    ///
+    /// InfiniBand routes by LID within a subnet, so this is what a peer needs
+    /// in order to address this port. The subnet manager assigns it
+    /// asynchronously, and `0` means it has not done so yet. Ethernet ports
+    /// have no LIDs and always read `0`.
+    pub fn lid(&self) -> u16 {
+        self.attr.lid
+    }
+
     /// Get the link layer protocol used by this port.
     pub fn link_layer(&self) -> LinkLayer {
         self.attr.link_layer.into()
@@ -594,6 +740,45 @@ impl DeviceContext {
     }
 
     /// Query the attributes of a physical port.
+    /// The file descriptor asynchronous events arrive on.
+    ///
+    /// [`next_async_event`](Self::next_async_event) blocks while the queue is
+    /// empty, so a caller that must stay responsive -- to shut down, say --
+    /// polls this first and only reads once it is readable.
+    pub fn async_fd(&self) -> RawFd {
+        unsafe { self.context.as_ref().async_fd }
+    }
+
+    /// Read one asynchronous event, acknowledging it before returning.
+    ///
+    /// **Blocks** while no event is queued; see [`async_fd`](Self::async_fd).
+    ///
+    /// Every event must be acknowledged, including ones the caller does not
+    /// care about, because an outstanding event blocks device teardown. That
+    /// is why this acknowledges on the caller's behalf rather than handing
+    /// back something with a destructor to forget.
+    pub fn next_async_event(&self) -> Result<AsyncEvent, GetAsyncEventError> {
+        let mut event = MaybeUninit::<ibv_async_event>::uninit();
+        let ret = unsafe { ibv_get_async_event(self.context.as_ptr(), event.as_mut_ptr()) };
+        if ret != 0 {
+            return Err(GetAsyncEventErrorKind::Ibverbs(io::Error::last_os_error()).into());
+        }
+
+        // Copy the payload out before acknowledging: the acknowledgement is
+        // what allows the provider to reuse the event.
+        let event = unsafe { event.assume_init() };
+        let event_type: AsyncEventType = event.event_type.into();
+        let port_num = if event_type.is_port_scoped() {
+            Some(unsafe { event.element.port_num } as u8)
+        } else {
+            None
+        };
+
+        unsafe { ibv_ack_async_event(&event as *const _ as *mut _) };
+
+        Ok(AsyncEvent { event_type, port_num })
+    }
+
     pub fn query_port(&self, port_num: u8) -> Result<PortAttr, QueryPortError> {
         let mut attr = MaybeUninit::<ibv_port_attr>::uninit();
         unsafe {
@@ -837,6 +1022,48 @@ mod tests {
     use super::*;
     use crate::ibverbs::device::{self, DeviceInfo};
 
+    #[test]
+    fn test_async_event_type_from_raw() {
+        assert_eq!(
+            AsyncEventType::from(ibv_event_type::IBV_EVENT_LID_CHANGE),
+            AsyncEventType::LidChange
+        );
+        assert_eq!(
+            AsyncEventType::from(ibv_event_type::IBV_EVENT_CLIENT_REREGISTER),
+            AsyncEventType::ClientReregister
+        );
+        assert_eq!(
+            AsyncEventType::from(ibv_event_type::IBV_EVENT_CQ_ERR),
+            AsyncEventType::CqError
+        );
+    }
+
+    #[test]
+    fn test_port_scoped_events() {
+        // These carry a port number; the queue-scoped ones name a queue, and
+        // reading port_num out of that union would be nonsense.
+        assert!(AsyncEventType::LidChange.is_port_scoped());
+        assert!(AsyncEventType::PortActive.is_port_scoped());
+        assert!(AsyncEventType::SmChange.is_port_scoped());
+        assert!(!AsyncEventType::CqError.is_port_scoped());
+        assert!(!AsyncEventType::QpFatal.is_port_scoped());
+        assert!(!AsyncEventType::Unknown(999).is_port_scoped());
+    }
+
+    #[test]
+    fn test_port_attr_lid() {
+        let mut attr = unsafe { MaybeUninit::<ibv_port_attr>::zeroed().assume_init() };
+        attr.lid = 0x811f;
+
+        let attr = PortAttr { attr };
+        assert_eq!(attr.lid(), 0x811f);
+
+        // The subnet manager has not reached the port yet, and every Ethernet
+        // port reads this permanently.
+        let unassigned = unsafe { MaybeUninit::<ibv_port_attr>::zeroed().assume_init() };
+        assert_eq!(PortAttr { attr: unassigned }.lid(), 0);
+    }
+
     #[test]
     fn test_query_rt_values_ex() -> Result<(), Box<dyn std::error::Error>> {
         let device_list = device::DeviceList::new()?;
@@ -978,16 +1205,16 @@ mod tests {
         for device in &device_list {
             let ctx = device.open().unwrap();
 
-            let gid_entries = ctx.query_gid_table().unwrap();
+            let gid_entries = match ctx.query_gid_table() {
+                Ok(e) => e,
+                Err(_) => continue, // kernel may not support ibv_query_gid_table_ex
+            };
             let gid_entries_fallback = ctx.query_gid_table_fallback().unwrap();
 
             assert_eq!(gid_entries.len(), gid_entries_fallback.len());
             for i in 0..gid_entries.len() {
                 assert_eq!(gid_entries[i].gid(), gid_entries_fallback[i].gid());
                 assert_eq!(gid_entries[i].gid_index(), gid_entries_fallback[i].gid_index());
-                assert_eq!(gid_entries[i].gid_type(), gid_entries_fallback[i].gid_type());
-                assert_eq!(gid_entries[i].netdev_index(), gid_entries_fallback[i].netdev_index());
-                assert_eq!(gid_entries[i].netdev_name(), gid_entries_fallback[i].netdev_name());
                 assert_eq!(gid_entries[i].port_num(), gid_entries_fallback[i].port_num());
             }
         }