diff --git a/Cargo.toml b/Cargo.toml index 3e00c4e..6a37478 100644 --- a/Cargo.toml +++ b/Cargo.toml @@ -3,13 +3,16 @@ name = "rdma-cm" version = "0.1.0" edition = "2018" -# See more keys and their definitions at https://doc.rust-lang.org/cargo/reference/manifest.html +[features] +default = ["async"] +async = ["tokio"] [dependencies] tracing = "0.1.26" -nix = "0.21" +nix = "0.23" arrayvec = "0.7.1" thiserror = "1.0.29" +tokio = { version = "1", features = ["net"], optional = true } [build-dependencies] bindgen = "0.59.1" @@ -18,4 +21,4 @@ bindgen = "0.59.1" [dev-dependencies] utilities = { path = "utilities" } -tracing-subscriber = "0.2.19" \ No newline at end of file +tracing-subscriber = "0.2.19" diff --git a/src/error.rs b/src/error.rs index 812359e..13aad37 100644 --- a/src/error.rs +++ b/src/error.rs @@ -33,6 +33,8 @@ pub enum RdmaCmError { RdmaEventChannel(std::io::Error), #[error("Unable to create RDMA Device ID.")] RdmaCreateId(std::io::Error), + #[error("Unable to set O_NONBLOCK on event channel.")] + Fcntl(nix::errno::Errno), #[error("Unable to disconnect RDMA connction.")] Disconnect(std::io::Error), } diff --git a/src/lib.rs b/src/lib.rs index 05ad96f..ed3da08 100644 --- a/src/lib.rs +++ b/src/lib.rs @@ -96,7 +96,7 @@ impl CmEvent { TryFrom::try_from(e).expect("Unable to convert event integer to enum.") } - pub fn get_connection_request_id(&self) -> CommunicationManager { + pub fn get_connection_request_id(&self) -> CommunicationManager { if self.get_event() != RdmaCmEvent::ConnectionRequest { panic!( "{} only makes sense for ConnectRequest event!", @@ -512,15 +512,14 @@ impl QueuePair for (i, (work_id, memory)) in work_requests.enumerate() { // Total number of bytes to send. let length = (R::memory_size(memory) * std::mem::size_of::()) as u32; - sges.push(ffi::ibv_sge { - addr: memory.as_ptr() as u64, - length, - lkey: memory.get_lkey(), - }); - - let wr = - R::make_work_request(*work_id, &mut sges[i] as *mut _, length <= 512, opcode); - requests.push(wr); + sges.push(ffi::ibv_sge { + addr: memory.as_ptr() as u64, + length, + lkey: memory.get_lkey(), + }); + + let wr = R::make_work_request(*work_id, &mut sges[i] as *mut _, length <= 512, opcode); + requests.push(wr); } // Link all entries together. @@ -707,14 +706,14 @@ impl Request for PostRecv { } /// High-level wrapper around an `rdma_cm_id`. -pub struct CommunicationManager { +pub struct CommunicationManager { cm_id: *mut ffi::rdma_cm_id, /// If the CommunicationManager is used for connecting two nodes it needs an event channel. /// We keep a reference to it for deallocation. event_channel: Option<*mut ffi::rdma_event_channel>, } -impl Drop for CommunicationManager { +impl Drop for CommunicationManager { fn drop(&mut self) { debug!("{}", function_name!()); let ret = unsafe { ffi::rdma_destroy_id(self.cm_id) }; @@ -732,35 +731,96 @@ impl Drop for CommunicationManager { } } -impl CommunicationManager { - pub fn new() -> Result { - info!("{}", function_name!()); +/// Make a CommunicationManager, which can be either blocking (the default) or nonblocking (by +/// calling [`Self::async_cm_events`]. +pub struct CommunicationManagerBuilder; - let event_channel: *mut ffi::rdma_event_channel = - unsafe { ffi::rdma_create_event_channel() }; - if event_channel == null_mut() { - return Err(RdmaCmError::RdmaEventChannel(Error::last_os_error())); - } +impl Default for CommunicationManagerBuilder { + fn default() -> Self { + Self + } +} - let mut id: MaybeUninit<*mut ffi::rdma_cm_id> = MaybeUninit::uninit(); - let ret = unsafe { - ffi::rdma_create_id( - event_channel, - id.as_mut_ptr(), - null_mut(), - ffi::rdma_port_space_RDMA_PS_TCP, - ) - }; - if ret == -1 { - return Err(RdmaCmError::RdmaCreateId(Error::last_os_error())); - } +impl CommunicationManagerBuilder { + pub fn async_cm_events(self) -> CommunicationManagerBuilder { + CommunicationManagerBuilder + } + + pub fn build(self) -> Result> { + CommunicationManager::::new() + } +} + +impl CommunicationManagerBuilder { + pub fn build(self) -> Result> { + CommunicationManager::::new() + } +} +fn make_cm_internals() -> Result<(*mut ffi::rdma_cm_id, *mut ffi::rdma_event_channel)> { + let event_channel: *mut ffi::rdma_event_channel = unsafe { ffi::rdma_create_event_channel() }; + if event_channel == null_mut() { + return Err(RdmaCmError::RdmaEventChannel(Error::last_os_error())); + } + + let mut id: MaybeUninit<*mut ffi::rdma_cm_id> = MaybeUninit::uninit(); + let ret = unsafe { + ffi::rdma_create_id( + event_channel, + id.as_mut_ptr(), + null_mut(), + ffi::rdma_port_space_RDMA_PS_TCP, + ) + }; + if ret == -1 { + return Err(RdmaCmError::RdmaCreateId(Error::last_os_error())); + } + + Ok((unsafe { id.assume_init() }, event_channel)) +} + +impl CommunicationManager { + pub fn new() -> Result { + info!("{}", function_name!()); + let (cm_id, event_channel) = make_cm_internals()?; Ok(CommunicationManager { - cm_id: unsafe { id.assume_init() }, + cm_id, event_channel: Some(event_channel), }) } +} +impl CommunicationManager { + /// Make an async CommunicationManager. + /// + /// Almost all of this crate's api is async: we submit events to rdma-cm and the corresponding operations + /// don't block. By default, though, `rdma_get_cm_event` blocks. From the manpage: + /// + /// man rdma_get_cm_event + /// Retrieves a communication event. If no events are pending, by default, the call will block until an event is received. + /// ...The default synchronous behavior of this routine can be changed by modifying the file descriptor associated with the given channel. + /// + /// This is exactly what this function does: set `O_NONBLOCK` on the underlying event channel's + /// fd, via `fcntl`. Correspondingly, [`get_cm_event`] will return a future which will resolve + /// when the event is ready. + /// + /// This function requires the `async` feature, which is activated by default. + #[cfg(feature = "async")] + pub fn new() -> Result> { + let (cm_id, event_channel) = make_cm_internals()?; + nix::fcntl::fcntl( + unsafe { (*event_channel).fd }, + nix::fcntl::FcntlArg::F_SETFL(nix::fcntl::OFlag::O_NONBLOCK), + ) + .map_err(RdmaCmError::Fcntl)?; + Ok(CommunicationManager { + cm_id, + event_channel: Some(event_channel), + }) + } +} + +impl CommunicationManager { /// Convenience method for accessing context and checkingn nullness. Used by other methods. fn get_raw_verbs_context(&self) -> *mut ffi::ibv_context { // Safety: always safe. Our API guarantees dereferencing `cm_id` will always be valid. @@ -800,7 +860,7 @@ impl CommunicationManager { return Err(RdmaCmError::CreateCompletionQueue(Error::last_os_error())); } - Ok(CompletionQueue { cq}) + Ok(CompletionQueue { cq }) } pub fn create_qp( @@ -1003,6 +1063,19 @@ impl CommunicationManager { Ok(()) } + pub fn disconnect(&self) -> Result<()> { + info!("{}", function_name!()); + + let ret = unsafe { ffi::rdma_disconnect(self.cm_id) }; + if ret == -1 { + return Err(RdmaCmError::Disconnect(Error::last_os_error())); + } + + Ok(()) + } +} + +impl CommunicationManager { pub fn get_cm_event(&self) -> Result { info!("{}", function_name!()); @@ -1015,16 +1088,70 @@ impl CommunicationManager { event: unsafe { cm_events.assume_init() }, }) } +} - pub fn disconnect(&self) -> Result<()> { +#[cfg(feature = "async")] +impl CommunicationManager { + pub fn get_cm_event(&self) -> get_cm_event_async::GetCmEventFuture { info!("{}", function_name!()); + get_cm_event_async::GetCmEventFuture::new(unsafe { (*self.cm_id).channel }) + } +} - let ret = unsafe { ffi::rdma_disconnect(self.cm_id) }; - if ret == -1 { - return Err(RdmaCmError::Disconnect(Error::last_os_error())); +#[cfg(feature = "async")] +mod get_cm_event_async { + use super::{ffi, CmEvent, Error, MaybeUninit, Result}; + use crate::RdmaCmError; + use std::{ + future::Future, + pin::Pin, + task::{Context, Poll}, + }; + use tokio::io::unix::AsyncFd; + + struct EventChannel(*mut ffi::rdma_event_channel); + + impl std::os::unix::io::AsRawFd for EventChannel { + fn as_raw_fd(&self) -> std::os::unix::prelude::RawFd { + unsafe { (*self.0).fd } } + } - Ok(()) + pub struct GetCmEventFuture { + cm_events: MaybeUninit<*mut ffi::rdma_cm_event>, + channel: Option>, + } + + impl GetCmEventFuture { + pub fn new(channel: *mut ffi::rdma_event_channel) -> Self { + GetCmEventFuture { + cm_events: MaybeUninit::uninit(), + channel: Some(AsyncFd::new(EventChannel(channel)).expect("make asyncfd")), + } + } + } + + impl Future for GetCmEventFuture { + type Output = Result; + + fn poll(mut self: Pin<&mut Self>, cx: &mut Context<'_>) -> Poll { + match self.channel.as_ref().unwrap().poll_read_ready(cx) { + Poll::Pending => return Poll::Pending, + Poll::Ready(Ok(mut x)) => { + x.clear_ready(); + } + Poll::Ready(Err(e)) => return Poll::Ready(Err(RdmaCmError::GetCmEvent(e))), + } + + let channel = self.channel.take().unwrap().into_inner().0; + let ret = unsafe { ffi::rdma_get_cm_event(channel, self.cm_events.as_mut_ptr()) }; + if ret == -1 { + return Poll::Ready(Err(RdmaCmError::GetCmEvent(Error::last_os_error()))); + } + Poll::Ready(Ok(CmEvent { + event: unsafe { self.cm_events.assume_init() }, + })) + } } }