From 20213993badc41d45337870c7f28e8e7f98094b2 Mon Sep 17 00:00:00 2001 From: Luke Yue Date: Fri, 5 Jun 2026 00:12:17 +0800 Subject: [PATCH 1/4] ci: remove cirrus ci configuration as they stop providing CI service Signed-off-by: Luke Yue --- .cirrus.yml | 33 ------------------------------ .github/workflows/test.yml | 42 ++++++++++++++++++++++++++++++++++++++ Cargo.toml | 3 ++- 3 files changed, 44 insertions(+), 34 deletions(-) delete mode 100644 .cirrus.yml diff --git a/.cirrus.yml b/.cirrus.yml deleted file mode 100644 index 685a697..0000000 --- a/.cirrus.yml +++ /dev/null @@ -1,33 +0,0 @@ -compute_engine_instance: - image_project: rocky-linux-cloud - image: family/rocky-linux-9 - -task: - env: - PATH: /root/.cargo/bin:${PATH} - CODECOV_TOKEN: ENCRYPTED[5608a167c2ad93fcac429e78e49661525794539aad86af2553c6eb0d1f3dd583f75a9bb4a2864d761bb336f8eec5c68d] - - prepare_script: - - dnf install -y cmake librdmacm libibverbs gcc clang - - rdma link add rxe_eth0 type rxe netdev eth0 - rust_script: - - curl https://sh.rustup.rs -sSf --output rustup.sh - - sh rustup.sh -y - - curl -LsSf https://get.nexte.st/latest/linux | tar zxf - -C ${CARGO_HOME:-~/.cargo}/bin - - curl --proto '=https' --tlsv1.2 -sSf https://just.systems/install.sh | bash -s -- --to ${CARGO_HOME:-~/.cargo}/bin - - cargo install cargo-llvm-cov - codecov_script: - - curl -Os https://cli.codecov.io/latest/linux/codecov - - chmod +x codecov - rdma_core_script: - - dnf install -y git libnl3-devel libudev-devel make pkgconfig valgrind-devel - - git clone https://github.com/linux-rdma/rdma-core.git - - ./rdma-core/build.sh - test_script: - - export LD_LIBRARY_PATH=./rdma-core/build/lib - - just test-basic-with-cov - - just test-rc-pingpong-with-cov - - just test-cmtime-with-cov - - just generate-cov - - sed -i 's#/tmp/cirrus-ci-build/##g' lcov.info - - ./codecov --verbose upload-process --disable-search --fail-on-error --git-service github -f ./lcov.info diff --git a/.github/workflows/test.yml b/.github/workflows/test.yml index d985fad..29c88f6 100644 --- a/.github/workflows/test.yml +++ b/.github/workflows/test.yml @@ -39,3 +39,45 @@ jobs: token: ${{ secrets.CODECOV_TOKEN }} verbose: true file: lcov.info + + test-rxe: + runs-on: ubuntu-latest + steps: + - name: Checkout repository code + uses: actions/checkout@v4 + - name: Install tools required + uses: taiki-e/install-action@v2 + with: + tool: just,cargo-nextest,cargo-llvm-cov + - name: Clone rxe kmod + run: | + uname -a + git clone https://github.com/pizhenwei/rxe.git + make -C rxe + sudo insmod ./rxe/rdma_rxe.ko + ip add + sudo rdma link add rxe_eth0 type rxe netdev eth0 + - name: Build rdma-core + run: | + sudo apt update + sudo apt install -y make pkg-config cmake libnl-3-dev libnl-route-3-dev libnl-genl-3-dev + git clone https://github.com/linux-rdma/rdma-core.git + ./rdma-core/build.sh + - name: Test with RXE + run: | + export LD_LIBRARY_PATH=./rdma-core/build/lib + cargo llvm-cov --no-report run --example ibv_devinfo + if modinfo mlx5_ib >/dev/null 2>&1 && lsmod | grep -q '^mlx5_ib'; then + sudo rmmod mlx5_ib + fi + just test-basic-with-cov + just test-rc-pingpong-with-cov + just test-cmtime-with-cov + just generate-cov + - name: Upload coverage information + uses: codecov/codecov-action@v5 + with: + fail_ci_if_error: false + token: ${{ secrets.CODECOV_TOKEN }} + verbose: true + file: lcov.info diff --git a/Cargo.toml b/Cargo.toml index a1a97a8..fb64918 100644 --- a/Cargo.toml +++ b/Cargo.toml @@ -5,7 +5,8 @@ description = "A better wrapper for using RDMA programming APIs in Rust flavor" license= "MPL-2.0" repository = "https://github.com/RDMA-Rust/sideway" readme = "README.md" -keywords = ["RDMA", "verbs", "cm", "libibverbs", "librdmacm"] +keywords = ["RDMA", "verbs", "libibverbs", "librdmacm", "ibverbs"] +categories = ["network-programming"] authors = [ "Luke Yue ", "FujiZ ", From 912fee4d8d6ce086ce4db481d9133805861597d5 Mon Sep 17 00:00:00 2001 From: Luke Yue Date: Sat, 3 Jan 2026 14:56:06 +0800 Subject: [PATCH 2/4] feat(rdmacm): support setting private data in rdmacm conn param Signed-off-by: Luke Yue --- src/rdmacm/communication_manager.rs | 485 ++++++++++++++++++++++++++-- 1 file changed, 453 insertions(+), 32 deletions(-) diff --git a/src/rdmacm/communication_manager.rs b/src/rdmacm/communication_manager.rs index d938b3e..6f76fc2 100644 --- a/src/rdmacm/communication_manager.rs +++ b/src/rdmacm/communication_manager.rs @@ -166,8 +166,8 @@ use std::{io, mem::MaybeUninit, net::SocketAddr, ptr::NonNull, sync::Arc}; use os_socketaddr::OsSocketAddr; use rdma_mummy_sys::{ - ibv_qp_attr, rdma_accept, rdma_ack_cm_event, rdma_bind_addr, rdma_cm_event, rdma_cm_event_type, rdma_cm_id, - rdma_conn_param, rdma_connect, rdma_create_event_channel, rdma_create_id, rdma_destroy_event_channel, + ibv_qp_attr, ibv_qp_type, rdma_accept, rdma_ack_cm_event, rdma_bind_addr, rdma_cm_event, rdma_cm_event_type, + rdma_cm_id, rdma_conn_param, rdma_connect, rdma_create_event_channel, rdma_create_id, rdma_destroy_event_channel, rdma_destroy_id, rdma_disconnect, rdma_establish, rdma_event_channel, rdma_get_cm_event, rdma_init_qp_attr, rdma_listen, rdma_port_space, rdma_resolve_addr, rdma_resolve_route, }; @@ -245,7 +245,10 @@ pub struct Identifier { /// A connection paramter used for configure the communication when connecting or establishing /// datagram communication. Used in [`Identifier::connect`] and [`Identifier::accept`]. -pub struct ConnectionParameter(rdma_conn_param); +pub struct ConnectionParameter { + conn_param: rdma_conn_param, + private_data: Vec, +} /// The RDMA port space. pub enum PortSpace { @@ -517,6 +520,74 @@ impl Event { unsafe { self.event.as_ref().status } } + /// Get the private data sent by the remote peer. + /// + /// This is available for [`EventType::ConnectRequest`], + /// [`EventType::ConnectResponse`], and [`EventType::Rejected`] events, where + /// RDMA CM uses the `rdma_cm_event.param` union to carry connection or + /// datagram-service private data. + /// + /// Note that the actual amount of data transferred is transport dependent + /// and may be larger than that requested, with trailing zero as padding. + /// + /// **For `AF_IB` connected requests, Linux formats RDMA CM's + /// `struct cma_hdr` at the start of the IB CM REQ private-data area. + /// The first byte is `cma_version`, currently `0`, so byte 0 of + /// the [`EventType::ConnectRequest`] private data is overwritten + /// with `0` and is not the peer's original application byte.** + /// + /// # Returns + /// The private data slice, or an empty slice if the event has no private + /// data. + /// + /// # Example + /// ```ignore + /// match event.event_type() { + /// EventType::ConnectRequest => { + /// let data = event.private_data(); + /// if !data.is_empty() { + /// println!("Received {} bytes of private data", data.len()); + /// } + /// } + /// _ => {} + /// } + /// ``` + pub fn private_data(&self) -> &[u8] { + unsafe { + let event = self.event.as_ref(); + match self.event_type() { + EventType::ConnectRequest | EventType::ConnectResponse | EventType::Rejected => { + let Some(id) = NonNull::new(event.id) else { + return &[]; + }; + + if id.as_ref().qp_type == ibv_qp_type::IBV_QPT_UD { + let param = &event.param.ud; + Self::private_data_slice(event, param.private_data, param.private_data_len) + } else { + let param = &event.param.conn; + Self::private_data_slice(event, param.private_data, param.private_data_len) + } + }, + _ => &[], + } + } + } + + fn private_data_slice( + _event: &rdma_cm_event, private_data: *const std::ffi::c_void, private_data_len: u8, + ) -> &[u8] { + let len = private_data_len as usize; + if len == 0 || private_data.is_null() { + &[] + } else { + // SAFETY: The caller selected the active RDMA CM event union member + // for an event type that carries private data. The returned slice is + // valid until the event is acknowledged. + unsafe { std::slice::from_raw_parts(private_data.cast(), len) } + } + } + /// Acknowledge and free the communication event. /// /// # Note @@ -866,7 +937,7 @@ impl Identifier { /// pub fn connect(&self, mut conn_param: ConnectionParameter) -> Result<(), ConnectError> { let cm_id = self.cm_id; - let ret = unsafe { rdma_connect(cm_id.as_ptr(), &mut conn_param.0) }; + let ret = unsafe { rdma_connect(cm_id.as_ptr(), &mut conn_param.conn_param) }; if ret < 0 { return Err(ConnectErrorKind::Rdmacm(io::Error::last_os_error()).into()); @@ -906,7 +977,7 @@ impl Identifier { pub fn accept(&self, mut conn_param: ConnectionParameter) -> Result<(), AcceptError> { let cm_id = self.cm_id; - let ret = unsafe { rdma_accept(cm_id.as_ptr(), &mut conn_param.0) }; + let ret = unsafe { rdma_accept(cm_id.as_ptr(), &mut conn_param.conn_param) }; if ret < 0 { return Err(AcceptErrorKind::Rdmacm(io::Error::last_os_error()).into()); @@ -962,33 +1033,39 @@ impl Identifier { impl Default for ConnectionParameter { fn default() -> Self { - Self(rdma_conn_param { - private_data: null(), - private_data_len: 0, - responder_resources: 1, - initiator_depth: 1, - flow_control: 0, - retry_count: 7, - rnr_retry_count: 7, - srq: 0, - qp_num: 0, - }) + Self { + conn_param: rdma_conn_param { + private_data: null(), + private_data_len: 0, + responder_resources: 1, + initiator_depth: 1, + flow_control: 0, + retry_count: 7, + rnr_retry_count: 7, + srq: 0, + qp_num: 0, + }, + private_data: Vec::new(), + } } } impl ConnectionParameter { pub fn new() -> Self { - Self(rdma_conn_param { - private_data: null(), - private_data_len: 0, - responder_resources: 0, - initiator_depth: 0, - flow_control: 0, - retry_count: 0, - rnr_retry_count: 0, - srq: 0, - qp_num: 0, - }) + Self { + conn_param: rdma_conn_param { + private_data: null(), + private_data_len: 0, + responder_resources: 0, + initiator_depth: 0, + flow_control: 0, + retry_count: 0, + rnr_retry_count: 0, + srq: 0, + qp_num: 0, + }, + private_data: Vec::new(), + } } /// Setup the QP number of the [`Identifier`]. You should fill in this field when you are @@ -999,7 +1076,101 @@ impl ConnectionParameter { /// [`accept`]: crate::rdmacm::communication_manager::Identifier::accept /// pub fn setup_qp_number(&mut self, qp_number: u32) -> &mut Self { - self.0.qp_num = qp_number; + self.conn_param.qp_num = qp_number; + self + } + + /// Setup the private data to be sent with connect or accept. + /// + /// # Private data size + /// + /// This method copies the provided slice into the [`ConnectionParameter`] + /// and stores that owned buffer's pointer and length in the raw RDMA CM + /// parameter. It does not cap the length to any specific RDMA CM operation. + /// Check the operation limit + /// before calling [`Identifier::connect`] or [`Identifier::accept`], or when + /// using the lower-level [`rdma_reject(3)`] API. + /// + /// | Port space | Service type | [`connect`] | [`accept`] | [`rdma_reject(3)`] | + /// | --- | --- | ---: | ---: | ---: | + /// | [`PortSpace::Tcp`] | connected | 56 | 196 | 148 | + /// | [`PortSpace::Udp`] | datagram | 180 | 136 | 136 | + /// | [`PortSpace::InfiniBand`] | connected | 92 | 196 | 148 | + /// | [`PortSpace::InfiniBand`] | datagram | 216 | 136 | 136 | + /// + /// [`PortSpace::Tcp`] and [`PortSpace::Udp`] values are the user-visible + /// payload sizes documented by the [`rdma_connect(3)`] and + /// [`rdma_accept(3)`] man pages, plus the [`rdma_reject(3)`] sizes implied + /// by Linux's IB CM message constants and RDMA CM routing. + /// [`PortSpace::InfiniBand`] is derived from Linux CMA's + /// `id->qp_type == IB_QPT_UD` branch: connected QPs use IB CM REQ/REP/REJ + /// messages, while datagram services use SIDR REQ/REP. + /// + /// **For IB connected requests, Linux formats RDMA CM's + /// `struct cma_hdr` at the start of the IB CM REQ private-data area. + /// The first byte is `cma_version`, currently `0`, so byte 0 of + /// the [`EventType::ConnectRequest`] private data is overwritten + /// with `0`, please offset one byte when setting private data for IB.** + /// + /// [`connect`]: Identifier::connect + /// [`accept`]: Identifier::accept + /// [`rdma_connect(3)`]: https://man7.org/linux/man-pages/man3/rdma_connect.3.html + /// [`rdma_accept(3)`]: https://man7.org/linux/man-pages/man3/rdma_accept.3.html + /// [`rdma_reject(3)`]: https://man7.org/linux/man-pages/man3/rdma_reject.3.html + /// + /// [`setup_private_data`]: ConnectionParameter::setup_private_data + /// + /// # Panics + /// Panics if `data.len()` does not fit in `u8`, because + /// `rdma_conn_param.private_data_len` is an 8-bit field. + /// + /// # Example + /// ```ignore + /// let my_data = [1u8, 2, 3, 4]; + /// param.setup_private_data(&my_data); + /// id.connect(param)?; + /// ``` + pub fn setup_private_data(&mut self, data: &[u8]) -> &mut Self { + let len = + u8::try_from(data.len()).expect("ConnectionParameter private_data length is limited to u8::MAX bytes"); + self.private_data.clear(); + self.private_data.extend_from_slice(data); + self.conn_param.private_data = if self.private_data.is_empty() { + null() + } else { + self.private_data.as_ptr().cast() + }; + self.conn_param.private_data_len = len; + self + } + + /// Setup responder resources for the connection. + /// This is the maximum number of outstanding RDMA read/atomic operations + /// the local side will accept from the remote side. + pub fn setup_responder_resources(&mut self, resources: u8) -> &mut Self { + self.conn_param.responder_resources = resources; + self + } + + /// Setup initiator depth for the connection. + /// This is the maximum number of outstanding RDMA read/atomic operations + /// that the local side will have pending to the remote side. + pub fn setup_initiator_depth(&mut self, depth: u8) -> &mut Self { + self.conn_param.initiator_depth = depth; + self + } + + /// Setup retry count for the connection. + /// The number of times to retry a connection request or response. + pub fn setup_retry_count(&mut self, count: u8) -> &mut Self { + self.conn_param.retry_count = count; + self + } + + /// Setup RNR retry count for the connection. + /// The number of times to retry a receiver-not-ready error. + pub fn setup_rnr_retry_count(&mut self, count: u8) -> &mut Self { + self.conn_param.rnr_retry_count = count; self } } @@ -1007,11 +1178,122 @@ impl ConnectionParameter { #[cfg(test)] mod tests { use super::*; - use polling::{Event, Events, Poller}; - use std::net::{IpAddr, SocketAddr}; + use crate::ibverbs::address::{GidEntry, GidType}; + use crate::ibverbs::completion::GenericCompletionQueue; + use crate::ibverbs::device; + use crate::ibverbs::queue_pair::{ExtendedQueuePair, QueuePair}; + use polling::{Event as PollingEvent, Events, Poller}; + use std::net::{IpAddr, Ipv6Addr, SocketAddr}; use std::str::FromStr; use std::thread; + const CM_SETUP_PRIVATE_DATA_SIZE: usize = 54; + + #[derive(Clone, Copy)] + struct CmGidTestAddr { + ip: IpAddr, + scope_id: u32, + } + + impl CmGidTestAddr { + fn socket_addr(self, port: u16) -> SocketAddr { + match self.ip { + IpAddr::V4(addr) => SocketAddr::from((addr, port)), + IpAddr::V6(addr) => SocketAddr::V6(std::net::SocketAddrV6::new(addr, port, 0, self.scope_id)), + } + } + } + + fn cm_addr_from_gid_entry(gid_entry: GidEntry) -> Option { + let gid = gid_entry.gid(); + if gid.is_zero() { + return None; + } + + match gid_entry.gid_type() { + GidType::InfiniBand => {}, + GidType::RoceV2 if !gid.is_unicast_link_local() => {}, + _ => return None, + } + + let ipv6 = Ipv6Addr::from(gid); + let scope_id = if ipv6.is_unicast_link_local() { + let scope_id = gid_entry.netdev_index(); + if scope_id == 0 { + return None; + } + scope_id + } else { + 0 + }; + + Some(CmGidTestAddr { + ip: ipv6.to_ipv4_mapped().map_or(IpAddr::V6(ipv6), IpAddr::V4), + scope_id, + }) + } + + fn first_ib_or_roce_v2_gid_addr() -> Option { + let device_list = device::DeviceList::new().ok()?; + + for device in &device_list { + let Ok(ctx) = device.open() else { + continue; + }; + let Ok(gid_entries) = ctx.query_gid_table() else { + continue; + }; + + if let Some(addr) = gid_entries.into_iter().find_map(cm_addr_from_gid_entry) { + return Some(addr); + } + } + + None + } + + fn create_test_qp(id: &Identifier) -> Result { + let ctx = id + .get_device_context() + .ok_or_else(|| "RDMA CM ID has no verbs device context".to_owned())?; + let pd = ctx.alloc_pd().map_err(|err| err.to_string())?; + let cq = GenericCompletionQueue::from( + ctx.create_cq_builder() + .setup_cqe(2) + .build_ex() + .map_err(|err| err.to_string())?, + ); + + let mut qp = pd + .create_qp_builder() + .setup_max_send_wr(1) + .setup_max_send_sge(1) + .setup_max_recv_wr(1) + .setup_max_recv_sge(1) + .setup_send_cq(cq.clone()) + .setup_recv_cq(cq) + .build_ex() + .map_err(|err| err.to_string())?; + + qp.modify(&id.get_qp_attr(QueuePairState::Init).map_err(|err| err.to_string())?) + .map_err(|err| err.to_string())?; + + Ok(qp) + } + + fn move_test_qp_to_rts(id: &Identifier, qp: &mut ExtendedQueuePair) -> Result<(), String> { + qp.modify( + &id.get_qp_attr(QueuePairState::ReadyToReceive) + .map_err(|err| err.to_string())?, + ) + .map_err(|err| err.to_string())?; + qp.modify( + &id.get_qp_attr(QueuePairState::ReadyToSend) + .map_err(|err| err.to_string())?, + ) + .map_err(|err| err.to_string()) + } + #[test] fn test_cm_id_reference_count() -> Result<(), Box> { match EventChannel::new() { @@ -1064,7 +1346,7 @@ mod tests { let key = 233; assert_eq!(Arc::strong_count(&channel), 2); - unsafe { poller.add(&channel, Event::readable(key)).unwrap() }; + unsafe { poller.add(&channel, PollingEvent::readable(key)).unwrap() }; let mut events = Events::new(); events.clear(); @@ -1131,7 +1413,146 @@ mod tests { Ok(channel) => { let _id = channel.create_id(PortSpace::Tcp).unwrap(); - let _param = ConnectionParameter::new(); + let mut data = [0xa5; 196]; + let mut param = ConnectionParameter::new(); + param.setup_private_data(&data); + + assert_eq!(param.conn_param.private_data, param.private_data.as_ptr().cast()); + assert_eq!(param.conn_param.private_data_len, data.len() as u8); + assert_ne!(param.conn_param.private_data, data.as_ptr().cast()); + + data[0] = 0x5a; + let stored_data = unsafe { + std::slice::from_raw_parts( + param.conn_param.private_data as *const u8, + param.conn_param.private_data_len as usize, + ) + }; + assert_eq!(stored_data, &[0xa5; 196]); + assert_eq!(data[0], 0x5a); + + param.setup_private_data(&[]); + assert!(param.conn_param.private_data.is_null()); + assert_eq!(param.conn_param.private_data_len, 0); + + Ok(()) + }, + Err(_) => Ok(()), + } + } + + #[test] + fn test_connect_request_and_response_private_data() -> Result<(), Box> { + match EventChannel::new() { + Ok(channel) => { + let Some(cm_addr) = first_ib_or_roce_v2_gid_addr() else { + eprintln!( + "skipping RDMA CM private-data test: no usable IB GID or non-link-local RoCEv2 GID found" + ); + return Ok(()); + }; + + let listener = channel.create_id(PortSpace::Tcp)?; + let port = 18515; + let server_addr = cm_addr.socket_addr(port); + let client_src_addr = cm_addr.socket_addr(0); + listener.bind_addr(server_addr)?; + listener.listen(1)?; + + // Generate some data with different pattern, keep the first byte zero to make IB happy + let request_payload: [u8; CM_SETUP_PRIVATE_DATA_SIZE] = std::array::from_fn(|index| { + if index == 0 { + 0 + } else { + (index as u8).wrapping_mul(3).wrapping_add(1) + } + }); + + let response_payload: [u8; CM_SETUP_PRIVATE_DATA_SIZE] = + std::array::from_fn(|index| (index as u8).wrapping_mul(5).wrapping_add(3)); + + let server = thread::spawn(move || -> Result, String> { + let event = channel.get_cm_event().map_err(|err| err.to_string())?; + assert_eq!(event.event_type(), EventType::ConnectRequest); + + let request_private_data = event.private_data().to_vec(); + let conn_id = event + .cm_id() + .ok_or_else(|| "CONNECT_REQUEST did not provide a child CM ID".to_owned())?; + + let mut qp = create_test_qp(&conn_id)?; + move_test_qp_to_rts(&conn_id, &mut qp)?; + + let mut accept_param = ConnectionParameter::default(); + accept_param.setup_qp_number(qp.qp_number()); + accept_param.setup_private_data(&response_payload); + conn_id.accept(accept_param).map_err(|err| err.to_string())?; + event.ack().map_err(|err| err.to_string())?; + + let established = channel.get_cm_event().map_err(|err| err.to_string())?; + assert_eq!(established.event_type(), EventType::Established); + established.ack().map_err(|err| err.to_string())?; + + Ok(request_private_data) + }); + + let client_channel = EventChannel::new()?; + let client_id = client_channel.create_id(PortSpace::Tcp)?; + client_id.resolve_addr(Some(client_src_addr), server_addr, Duration::from_secs(2))?; + + let mut client_qp = None; + let response_private_data = loop { + let event = client_channel.get_cm_event()?; + match event.event_type() { + EventType::AddressResolved => { + client_id.resolve_route(Duration::from_secs(2))?; + event.ack()?; + }, + EventType::RouteResolved => { + let qp = create_test_qp(&client_id).map_err(io::Error::other)?; + let mut connect_param = ConnectionParameter::default(); + connect_param.setup_qp_number(qp.qp_number()); + connect_param.setup_private_data(&request_payload); + client_id.connect(connect_param)?; + client_qp = Some(qp); + event.ack()?; + }, + EventType::ConnectResponse => { + let data = event.private_data().to_vec(); + let qp = client_qp + .as_mut() + .expect("client QP must exist before CONNECT_RESPONSE"); + move_test_qp_to_rts(&client_id, qp).map_err(io::Error::other)?; + client_id.establish()?; + event.ack()?; + break data; + }, + event_type => panic!("unexpected client RDMA CM event: {event_type:?}"), + } + }; + + let request_private_data = server + .join() + .map_err(|panic| io::Error::other(format!("server thread panicked: {panic:?}")))? + .map_err(io::Error::other)?; + + // Valid field of private data from the RDMA CM event should be exactly the same as original data. + assert_eq!( + &request_private_data[..CM_SETUP_PRIVATE_DATA_SIZE], + request_payload.as_slice() + ); + assert_eq!( + &response_private_data[..CM_SETUP_PRIVATE_DATA_SIZE], + response_payload.as_slice() + ); + + // The left over should be all zero. + assert!(request_private_data[CM_SETUP_PRIVATE_DATA_SIZE..] + .iter() + .all(|&byte| byte == 0)); + assert!(response_private_data[CM_SETUP_PRIVATE_DATA_SIZE..] + .iter() + .all(|&byte| byte == 0)); Ok(()) }, From 262e1729a53bd110a828d0ff1a02075d0e24eb70 Mon Sep 17 00:00:00 2001 From: Luke Yue Date: Mon, 29 Jun 2026 10:21:29 +0800 Subject: [PATCH 3/4] feat(rdmacm): add rdma_reject support --- src/rdmacm/communication_manager.rs | 96 +++++++++++++++++++++++------ 1 file changed, 78 insertions(+), 18 deletions(-) diff --git a/src/rdmacm/communication_manager.rs b/src/rdmacm/communication_manager.rs index 6f76fc2..22a9271 100644 --- a/src/rdmacm/communication_manager.rs +++ b/src/rdmacm/communication_manager.rs @@ -169,7 +169,7 @@ use rdma_mummy_sys::{ ibv_qp_attr, ibv_qp_type, rdma_accept, rdma_ack_cm_event, rdma_bind_addr, rdma_cm_event, rdma_cm_event_type, rdma_cm_id, rdma_conn_param, rdma_connect, rdma_create_event_channel, rdma_create_id, rdma_destroy_event_channel, rdma_destroy_id, rdma_disconnect, rdma_establish, rdma_event_channel, rdma_get_cm_event, rdma_init_qp_attr, - rdma_listen, rdma_port_space, rdma_resolve_addr, rdma_resolve_route, + rdma_listen, rdma_port_space, rdma_reject, rdma_resolve_addr, rdma_resolve_route, }; use crate::ibverbs::device_context::DeviceContext; @@ -423,6 +423,20 @@ pub enum AcceptErrorKind { Rdmacm(#[from] io::Error), } +/// Error returned by [`Identifier::reject`] for rejecting a connection request. +#[derive(Debug, thiserror::Error)] +#[error("failed to reject")] +#[non_exhaustive] +pub struct RejectError(#[from] pub RejectErrorKind); + +/// The enum type for [`RejectError`]. +#[derive(Debug, thiserror::Error)] +#[error(transparent)] +#[non_exhaustive] +pub enum RejectErrorKind { + Rdmacm(#[from] io::Error), +} + /// Error returned by [`Identifier::establish`] for establishing connection setup. #[derive(Debug, thiserror::Error)] #[error("failed to establish")] @@ -986,6 +1000,35 @@ impl Identifier { Ok(()) } + /// Called from the listening side to reject an incoming connection on the [`Identifier`]. + /// + /// # Note + /// + /// This method is only useful for [`EventType::ConnectRequest`] events. A new [`Identifier`] + /// is automatically created to handle the incoming connection request. This is distinct from + /// the listener [`Identifier`]. The new [`Identifier`] could be obtained by [`Event::cm_id`]. + /// + /// Use [`ConnectionParameter::setup_private_data`] to attach optional rejection private data. + /// + /// [`Event::cm_id`]: crate::rdmacm::communication_manager::Event::cm_id + /// + pub fn reject(&self, conn_param: ConnectionParameter) -> Result<(), RejectError> { + let cm_id = self.cm_id; + let ret = unsafe { + rdma_reject( + cm_id.as_ptr(), + conn_param.conn_param.private_data, + conn_param.conn_param.private_data_len, + ) + }; + + if ret < 0 { + return Err(RejectErrorKind::Rdmacm(io::Error::last_os_error()).into()); + } + + Ok(()) + } + /// Acknowledge an incoming connection response event and complete the connection establishment /// on the [`Identifier`]. /// @@ -1080,18 +1123,17 @@ impl ConnectionParameter { self } - /// Setup the private data to be sent with connect or accept. + /// Setup the private data to be sent with connect, accept, or reject. /// /// # Private data size /// /// This method copies the provided slice into the [`ConnectionParameter`] /// and stores that owned buffer's pointer and length in the raw RDMA CM /// parameter. It does not cap the length to any specific RDMA CM operation. - /// Check the operation limit - /// before calling [`Identifier::connect`] or [`Identifier::accept`], or when - /// using the lower-level [`rdma_reject(3)`] API. + /// Check the operation limit before calling [`Identifier::connect`], + /// [`Identifier::accept`], or [`Identifier::reject`]. /// - /// | Port space | Service type | [`connect`] | [`accept`] | [`rdma_reject(3)`] | + /// | Port space | Service type | [`connect`] | [`accept`] | [`reject`] | /// | --- | --- | ---: | ---: | ---: | /// | [`PortSpace::Tcp`] | connected | 56 | 196 | 148 | /// | [`PortSpace::Udp`] | datagram | 180 | 136 | 136 | @@ -1099,8 +1141,8 @@ impl ConnectionParameter { /// | [`PortSpace::InfiniBand`] | datagram | 216 | 136 | 136 | /// /// [`PortSpace::Tcp`] and [`PortSpace::Udp`] values are the user-visible - /// payload sizes documented by the [`rdma_connect(3)`] and - /// [`rdma_accept(3)`] man pages, plus the [`rdma_reject(3)`] sizes implied + /// payload sizes documented by the [`rdma_connect`] and + /// [`rdma_accept`] man pages, plus the [`rdma_reject`] sizes implied /// by Linux's IB CM message constants and RDMA CM routing. /// [`PortSpace::InfiniBand`] is derived from Linux CMA's /// `id->qp_type == IB_QPT_UD` branch: connected QPs use IB CM REQ/REP/REJ @@ -1114,9 +1156,10 @@ impl ConnectionParameter { /// /// [`connect`]: Identifier::connect /// [`accept`]: Identifier::accept - /// [`rdma_connect(3)`]: https://man7.org/linux/man-pages/man3/rdma_connect.3.html - /// [`rdma_accept(3)`]: https://man7.org/linux/man-pages/man3/rdma_accept.3.html - /// [`rdma_reject(3)`]: https://man7.org/linux/man-pages/man3/rdma_reject.3.html + /// [`reject`]: Identifier::reject + /// [`rdma_connect`]: https://man7.org/linux/man-pages/man3/rdma_connect.3.html + /// [`rdma_accept`]: https://man7.org/linux/man-pages/man3/rdma_accept.3.html + /// [`rdma_reject`]: https://man7.org/linux/man-pages/man3/rdma_reject.3.html /// /// [`setup_private_data`]: ConnectionParameter::setup_private_data /// @@ -1358,7 +1401,11 @@ mod tests { assert_eq!(ev.key, key); let event = channel.get_cm_event().unwrap(); - assert_eq!(event.event_type(), EventType::AddressResolved); + assert!( + matches!(event.event_type(), EventType::AddressResolved | EventType::AddressError), + "unexpected RDMA CM event: {:?}", + event.event_type() + ); assert_eq!(Arc::strong_count(&channel), 2); event.ack().unwrap(); @@ -1582,16 +1629,29 @@ mod tests { fn test_get_device_context_caches_correctly() -> Result<(), Box> { match EventChannel::new() { Ok(channel) => { + let Some(cm_addr) = first_ib_or_roce_v2_gid_addr() else { + eprintln!( + "skipping RDMA CM device-context test: no usable IB GID or non-link-local RoCEv2 GID found" + ); + return Ok(()); + }; + let id = channel.create_id(PortSpace::Tcp)?; - let _ = id.resolve_addr( - None, - SocketAddr::from((IpAddr::from_str("127.0.0.1")?, 0)), - Duration::new(0, 200000000), - ); + if let Err(err) = id.resolve_addr(None, cm_addr.socket_addr(0), Duration::new(0, 200000000)) { + eprintln!("skipping RDMA CM device-context test: resolve_addr failed synchronously: {err}"); + return Ok(()); + } let event = channel.get_cm_event()?; - assert_eq!(event.event_type(), EventType::AddressResolved); + if event.event_type() != EventType::AddressResolved { + eprintln!( + "skipping RDMA CM device-context test: resolve_addr completed with {:?}", + event.event_type() + ); + event.ack()?; + return Ok(()); + } let ctx1 = id.get_device_context(); let ctx2 = id.get_device_context(); From e1d8c52157bca2f75c6068740c4e6f32bb46a3da Mon Sep 17 00:00:00 2001 From: Luke Yue Date: Tue, 30 Jun 2026 09:26:27 +0800 Subject: [PATCH 4/4] feat(rdmacm): add device and address getters --- src/rdmacm/communication_manager.rs | 236 ++++++++++++++++++++++++++-- 1 file changed, 220 insertions(+), 16 deletions(-) diff --git a/src/rdmacm/communication_manager.rs b/src/rdmacm/communication_manager.rs index 22a9271..c687d99 100644 --- a/src/rdmacm/communication_manager.rs +++ b/src/rdmacm/communication_manager.rs @@ -166,9 +166,10 @@ use std::{io, mem::MaybeUninit, net::SocketAddr, ptr::NonNull, sync::Arc}; use os_socketaddr::OsSocketAddr; use rdma_mummy_sys::{ - ibv_qp_attr, ibv_qp_type, rdma_accept, rdma_ack_cm_event, rdma_bind_addr, rdma_cm_event, rdma_cm_event_type, - rdma_cm_id, rdma_conn_param, rdma_connect, rdma_create_event_channel, rdma_create_id, rdma_destroy_event_channel, - rdma_destroy_id, rdma_disconnect, rdma_establish, rdma_event_channel, rdma_get_cm_event, rdma_init_qp_attr, + ibv_context, ibv_qp_attr, ibv_qp_type, rdma_accept, rdma_ack_cm_event, rdma_bind_addr, rdma_cm_event, + rdma_cm_event_type, rdma_cm_id, rdma_conn_param, rdma_connect, rdma_create_event_channel, rdma_create_id, + rdma_destroy_event_channel, rdma_destroy_id, rdma_disconnect, rdma_establish, rdma_event_channel, + rdma_free_devices, rdma_get_cm_event, rdma_get_devices, rdma_get_local_addr, rdma_get_peer_addr, rdma_init_qp_attr, rdma_listen, rdma_port_space, rdma_reject, rdma_resolve_addr, rdma_resolve_route, }; @@ -480,6 +481,24 @@ pub enum GetQueuePairAttributeErrorKind { Rdmacm(#[from] io::Error), } +/// Error returned by [`get_devices`] for getting RDMA devices opened by RDMA CM. +#[derive(Debug, thiserror::Error)] +#[error("failed to get rdma devices")] +#[non_exhaustive] +pub struct GetDevicesError(#[from] pub GetDevicesErrorKind); + +/// The enum type for [`GetDevicesError`]. +#[derive(Debug, thiserror::Error)] +#[error(transparent)] +#[non_exhaustive] +pub enum GetDevicesErrorKind { + Rdmacm(#[from] io::Error), + #[error("rdma_get_devices returned invalid device count: {0}")] + InvalidDeviceCount(i32), + #[error("rdma_get_devices returned null context at index {0}")] + NullDeviceContext(usize), +} + impl Drop for EventChannel { fn drop(&mut self) { unsafe { @@ -637,6 +656,105 @@ impl Drop for Event { } } +struct RdmaDeviceList { + devices: NonNull<*mut ibv_context>, +} + +#[repr(C)] +struct SockAddrIb { + sib_family: libc::sa_family_t, + sib_pkey: u16, + sib_flowinfo: u32, + sib_addr: [u8; 16], + sib_sid: u64, + sib_sid_mask: u64, + sib_scope_id: u64, +} + +impl Drop for RdmaDeviceList { + fn drop(&mut self) { + unsafe { rdma_free_devices(self.devices.as_ptr()) }; + } +} + +fn cached_device_context(context: NonNull) -> Arc { + let mut guard = DEVICE_LISTS.lock().unwrap(); + guard + .entry(context.as_ptr() as usize) + .or_insert_with(|| Arc::new(DeviceContext { context })) + .clone() +} + +fn socket_addr_from_raw(addr: &libc::sockaddr) -> Option { + let len = match addr.sa_family as i32 { + libc::AF_INET => std::mem::size_of::(), + libc::AF_INET6 => std::mem::size_of::(), + _ => return None, + }; + + unsafe { OsSocketAddr::copy_from_raw(addr, len as libc::socklen_t).into_addr() } +} + +fn port_from_raw_addr(addr: &libc::sockaddr) -> u16 { + match addr.sa_family as i32 { + libc::AF_INET => { + let addr = unsafe { &*(std::ptr::from_ref(addr).cast::()) }; + u16::from_be(addr.sin_port) + }, + libc::AF_INET6 => { + let addr = unsafe { &*(std::ptr::from_ref(addr).cast::()) }; + u16::from_be(addr.sin6_port) + }, + libc::AF_IB => { + let addr = unsafe { &*(std::ptr::from_ref(addr).cast::()) }; + u64::from_be(addr.sib_sid) as u16 + }, + _ => 0, + } +} + +/// Get a list of RDMA devices currently available. +/// +/// This wraps [`rdma_get_devices`], which returns a temporary array of +/// RDMA-CM-opened device contexts. The temporary array is released with +/// `rdma_free_devices`, while the returned [`DeviceContext`] handles are reused +/// through this module's global context cache. +/// +/// The cache is intentionally insertion-only. RDMA CM owns these opened +/// contexts, and dropping a [`DeviceContext`] closes its raw context. Keeping a +/// cached [`Arc`] alive prevents Rust from closing a context that librdmacm may +/// still manage and reuse internally. This mirrors [`Identifier::get_device_context`] +/// and does not attempt hot-unplug invalidation. +/// +/// [`rdma_get_devices`]: https://man7.org/linux/man-pages/man3/rdma_get_devices.3.html +pub fn get_devices() -> Result>, GetDevicesError> { + let mut num_devices = 0; + let devices = unsafe { rdma_get_devices(&mut num_devices) }; + + if devices.is_null() || num_devices == 0 { + return Ok(Vec::new()); + } + + let devices = RdmaDeviceList { + devices: unsafe { NonNull::new_unchecked(devices) }, + }; + let num_devices = usize::try_from(num_devices).map_err(|_| GetDevicesErrorKind::InvalidDeviceCount(num_devices))?; + let contexts = unsafe { std::slice::from_raw_parts(devices.devices.as_ptr(), num_devices) }; + let mut guard = DEVICE_LISTS.lock().unwrap(); + + contexts + .iter() + .enumerate() + .map(|(index, &context)| -> Result<_, GetDevicesError> { + let context = NonNull::new(context).ok_or(GetDevicesErrorKind::NullDeviceContext(index))?; + Ok(guard + .entry(context.as_ptr() as usize) + .or_insert_with(|| Arc::new(DeviceContext { context })) + .clone()) + }) + .collect() +} + fn new_cm_id_for_raw(event_channel: Arc, raw: *mut rdma_cm_id) -> Arc { let cm = unsafe { Arc::new(Identifier { @@ -812,6 +930,36 @@ impl Identifier { unsafe { cm_id.as_ref().port_num } } + /// Get the local port number of a bound [`Identifier`] in host byte order. If the + /// [`Identifier`] is not bound to a port, the returned value is 0. + pub fn get_src_port(&self) -> u16 { + unsafe { port_from_raw_addr(rdma_get_local_addr(self.cm_id.as_ref())) } + } + + /// Get the remote port number of a bound [`Identifier`] in host byte order. If the + /// [`Identifier`] is not connected, the returned value is 0. + pub fn get_dst_port(&self) -> u16 { + unsafe { port_from_raw_addr(rdma_get_peer_addr(self.cm_id.as_ref())) } + } + + /// Get the local IP socket address of a bound [`Identifier`]. + /// + /// Returns [`None`] when the RDMA CM address is not representable as + /// [`SocketAddr`], for example when the underlying address family is + /// `AF_IB`, or when the [`Identifier`] is not bound to an address. + pub fn get_local_addr(&self) -> Option { + unsafe { socket_addr_from_raw(rdma_get_local_addr(self.cm_id.as_ref())) } + } + + /// Get the remote IP socket address of the [`Identifier`]. + /// + /// Returns [`None`] when the RDMA CM address is not representable as + /// [`SocketAddr`], for example when the underlying address family is + /// `AF_IB`, or when the [`Identifier`] is not connected. + pub fn get_peer_addr(&self) -> Option { + unsafe { socket_addr_from_raw(rdma_get_peer_addr(self.cm_id.as_ref())) } + } + /// Bind the [`Identifier`] to a specific address. Note that users shouldn't bind to a loopback /// address like `127.0.0.1`, or the connection would fail. /// @@ -927,19 +1075,8 @@ impl Identifier { let cm_id = self.cm_id; unsafe { - if (*cm_id.as_ptr()).verbs.is_null() { - return None; - } - - let mut guard = DEVICE_LISTS.lock().unwrap(); - let device_ctx = guard.entry((*cm_id.as_ptr()).verbs as usize).or_insert_with(|| { - Arc::new(DeviceContext { - // Safe due to the is_null() check above. - context: NonNull::new((*cm_id.as_ptr()).verbs).unwrap(), - }) - }); - - Some(device_ctx.clone()) + let context = NonNull::new((*cm_id.as_ptr()).verbs)?; + Some(cached_device_context(context)) } } @@ -1438,6 +1575,7 @@ mod tests { let res = id.bind_addr(address); assert!(res.is_ok()); + assert_eq!(id.get_src_port(), address.port()); let new_id = channel.create_id(PortSpace::Tcp).unwrap(); @@ -1488,6 +1626,71 @@ mod tests { } } + #[test] + fn test_socket_addr_and_port_from_raw() { + let ipv4 = SocketAddr::from((std::net::Ipv4Addr::new(192, 0, 2, 1), 18515)); + let raw_ipv4 = OsSocketAddr::from(ipv4); + assert_eq!( + unsafe { socket_addr_from_raw(raw_ipv4.as_ptr().as_ref().unwrap()) }, + Some(ipv4) + ); + assert_eq!( + unsafe { port_from_raw_addr(raw_ipv4.as_ptr().as_ref().unwrap()) }, + 18515 + ); + + let ipv6 = SocketAddr::from((std::net::Ipv6Addr::LOCALHOST, 18516)); + let raw_ipv6 = OsSocketAddr::from(ipv6); + assert_eq!( + unsafe { socket_addr_from_raw(raw_ipv6.as_ptr().as_ref().unwrap()) }, + Some(ipv6) + ); + assert_eq!( + unsafe { port_from_raw_addr(raw_ipv6.as_ptr().as_ref().unwrap()) }, + 18516 + ); + + let ib = SockAddrIb { + sib_family: libc::AF_IB as _, + sib_pkey: 0, + sib_flowinfo: 0, + sib_addr: [0; 16], + sib_sid: u64::to_be(18517), + sib_sid_mask: 0, + sib_scope_id: 0, + }; + let ib_addr = unsafe { &*(std::ptr::from_ref(&ib).cast::()) }; + assert_eq!(socket_addr_from_raw(ib_addr), None); + assert_eq!(port_from_raw_addr(ib_addr), 18517); + + let unsupported = libc::sockaddr { + sa_family: libc::AF_UNIX as _, + sa_data: [0; 14], + }; + assert_eq!(socket_addr_from_raw(&unsupported), None); + assert_eq!(port_from_raw_addr(&unsupported), 0); + } + + #[test] + fn test_get_devices_smoke_and_cache() { + let devices = match get_devices() { + Ok(devices) => devices, + Err(err) => { + eprintln!("skipping RDMA CM get_devices smoke test: {err}"); + return; + }, + }; + + let second = get_devices().expect("second rdma_get_devices call should be consistent after first success"); + assert_eq!(devices.len(), second.len()); + for (first, second) in devices.iter().zip(second.iter()) { + assert!( + Arc::ptr_eq(first, second), + "rdma_get_devices contexts should reuse the global DeviceContext cache" + ); + } + } + #[test] fn test_connect_request_and_response_private_data() -> Result<(), Box> { match EventChannel::new() { @@ -1552,6 +1755,7 @@ mod tests { let event = client_channel.get_cm_event()?; match event.event_type() { EventType::AddressResolved => { + assert_eq!(client_id.get_dst_port(), server_addr.port()); client_id.resolve_route(Duration::from_secs(2))?; event.ack()?; },