complement missing docs

Signed-off-by: Runji Wang <[email protected]>

Author: wangrunji0408

src/lib.rs

@@ -1,4 +1,7 @@
+//! Asynchronous Rust bindings to UCX.
+
 #![deny(warnings)]
+#![deny(missing_docs)]
 
 use ucx1_sys::ucs_status_ptr_t;
 use ucx1_sys::ucs_status_t;
@@ -25,6 +28,8 @@ macro_rules! spawn_thread {
 
 pub mod ucp;
 
+/// UCX error code.
+#[allow(missing_docs)]
 #[repr(i8)]
 #[derive(thiserror::Error, Debug, PartialEq, Eq)]
 pub enum Error {

src/ucp/endpoint/mod.rs

@@ -73,6 +73,7 @@ impl EndpointInner {
     }
 }
 
+/// Communication endpoint.
 #[derive(Debug, Clone)]
 pub struct Endpoint {
     handle: ucp_ep_h,
@@ -189,10 +190,12 @@ impl Endpoint {
         Ok(endpoint)
     }
 
+    /// Whether the endpoint is closed.
     pub fn is_closed(&self) -> bool {
         self.inner.is_closed()
     }
 
+    /// Get the endpoint status.
     pub fn get_status(&self) -> Result<(), Error> {
         self.inner.check()
     }
@@ -203,6 +206,7 @@ impl Endpoint {
         Ok(self.handle)
     }
 
+    /// Print endpoint information to stderr.
     pub fn print_to_stderr(&self) {
         if !self.inner.is_closed() {
             unsafe { ucp_ep_print_info(self.handle, stderr) };
@@ -274,6 +278,7 @@ impl Endpoint {
         }
     }
 
+    /// Get the worker of the endpoint.
     pub fn worker(&self) -> &Rc<Worker> {
         &self.inner.worker
     }

src/ucp/endpoint/rma.rs

@@ -1,12 +1,15 @@
 use super::*;
 
+/// A memory region allocated through UCP library,
+/// which is optimized for remote memory access operations.
 #[derive(Debug)]
 pub struct MemoryHandle {
     handle: ucp_mem_h,
     context: Arc<Context>,
 }
 
 impl MemoryHandle {
+    /// Register memory region.
     pub fn register(context: &Arc<Context>, region: &mut [u8]) -> Self {
         #[allow(invalid_value)]
         #[allow(clippy::uninit_assumed_init)]
@@ -53,6 +56,7 @@ impl Drop for MemoryHandle {
     }
 }
 
+/// An owned buffer containing remote access key.
 #[derive(Debug)]
 pub struct RKeyBuffer {
     buf: *mut c_void,
@@ -71,6 +75,7 @@ impl Drop for RKeyBuffer {
     }
 }
 
+/// Remote access key.
 #[derive(Debug)]
 pub struct RKey {
     handle: ucp_rkey_h,
@@ -80,6 +85,7 @@ unsafe impl Send for RKey {}
 unsafe impl Sync for RKey {}
 
 impl RKey {
+    /// Create remote access key from packed buffer.
     pub fn unpack(endpoint: &Endpoint, rkey_buffer: &[u8]) -> Self {
         let mut handle = MaybeUninit::uninit();
         let status = unsafe {
@@ -103,6 +109,7 @@ impl Drop for RKey {
 }
 
 impl Endpoint {
+    /// Stores a contiguous block of data into remote memory.
     pub async fn put(&self, buf: &[u8], remote_addr: u64, rkey: &RKey) -> Result<(), Error> {
         trace!("put: endpoint={:?} len={}", self.handle, buf.len());
         unsafe extern "C" fn callback(request: *mut c_void, status: ucs_status_t) {
@@ -134,6 +141,7 @@ impl Endpoint {
         }
     }
 
+    /// Loads a contiguous block of data from remote memory.
     pub async fn get(&self, buf: &mut [u8], remote_addr: u64, rkey: &RKey) -> Result<(), Error> {
         trace!("get: endpoint={:?} len={}", self.handle, buf.len());
         unsafe extern "C" fn callback(request: *mut c_void, status: ucs_status_t) {

src/ucp/endpoint/stream.rs

@@ -1,6 +1,7 @@
 use super::*;
 
 impl Endpoint {
+    /// Sends data through stream.
     pub async fn stream_send(&self, buf: &[u8]) -> Result<usize, Error> {
         trace!("stream_send: endpoint={:?} len={}", self.handle, buf.len());
         unsafe extern "C" fn callback(request: *mut c_void, status: ucs_status_t) {
@@ -36,6 +37,7 @@ impl Endpoint {
         Ok(buf.len())
     }
 
+    /// Receives data from stream.
     pub async fn stream_recv(&self, buf: &mut [MaybeUninit<u8>]) -> Result<usize, Error> {
         trace!("stream_recv: endpoint={:?} len={}", self.handle, buf.len());
         unsafe extern "C" fn callback(request: *mut c_void, status: ucs_status_t, length: u64) {

src/ucp/endpoint/tag.rs

@@ -2,12 +2,14 @@ use super::*;
 use std::io::{IoSlice, IoSliceMut};
 
 impl Worker {
+    /// Receives a message with `tag`.
     pub async fn tag_recv(&self, tag: u64, buf: &mut [MaybeUninit<u8>]) -> Result<usize, Error> {
         self.tag_recv_mask(tag, u64::max_value(), buf)
             .await
             .map(|info| info.1)
     }
 
+    /// Receives a message with `tag` and `tag_mask`.
     pub async fn tag_recv_mask(
         &self,
         tag: u64,
@@ -56,6 +58,7 @@ impl Worker {
         .await
     }
 
+    /// Like `tag_recv`, except that it reads into a slice of buffers.
     pub async fn tag_recv_vectored(
         &self,
         tag: u64,
@@ -103,6 +106,7 @@ impl Worker {
 }
 
 impl Endpoint {
+    /// Sends a messages with `tag`.
     pub async fn tag_send(&self, tag: u64, buf: &[u8]) -> Result<usize, Error> {
         trace!("tag_send: endpoint={:?} len={}", self.handle, buf.len());
         unsafe extern "C" fn callback(request: *mut c_void, status: ucs_status_t) {
@@ -134,6 +138,7 @@ impl Endpoint {
         Ok(buf.len())
     }
 
+    /// Like `tag_send`, except that it reads into a slice of buffers.
     pub async fn tag_send_vectored(&self, tag: u64, iov: &[IoSlice<'_>]) -> Result<usize, Error> {
         trace!(
             "tag_send_vectored: endpoint={:?} iov.len={}",

src/ucp/listener.rs

@@ -6,6 +6,7 @@ use futures::stream::StreamExt;
 use std::mem::MaybeUninit;
 use std::net::SocketAddr;
 
+/// Listening on a specific address and accepting connections from clients.
 #[derive(Debug)]
 pub struct Listener {
     handle: ucp_listener_h,
@@ -14,6 +15,9 @@ pub struct Listener {
     recver: mpsc::UnboundedReceiver<ConnectionRequest>,
 }
 
+/// An incoming connection request.
+///
+/// The request must be explicitly accepted by [Worker::accept] or rejected by [Listener::reject].
 #[derive(Debug)]
 #[must_use = "connection must be accepted or rejected"]
 pub struct ConnectionRequest {
@@ -83,6 +87,7 @@ impl Listener {
         })
     }
 
+    /// Returns the local socket address of this listener.
     pub fn socket_addr(&self) -> Result<SocketAddr, Error> {
         #[allow(clippy::uninit_assumed_init)]
         let mut attr = ucp_listener_attr_t {
@@ -98,6 +103,7 @@ impl Listener {
         Ok(sockaddr.into_addr().unwrap())
     }
 
+    /// Waiting for the next connection request.
     pub async fn next(&mut self) -> ConnectionRequest {
         self.recver.next().await.unwrap()
     }

src/ucp/worker.rs

@@ -82,6 +82,7 @@ impl Worker {
         unsafe { ucp_worker_print_info(self.handle, stderr) };
     }
 
+    /// Thread safe level of the context.
     pub fn thread_mode(&self) -> ucs_thread_mode_t {
         let mut attr = MaybeUninit::<ucp_worker_attr>::uninit();
         unsafe { &mut *attr.as_mut_ptr() }.field_mask =
@@ -111,18 +112,22 @@ impl Worker {
         })
     }
 
+    /// Create a new [`Listener`].
     pub fn create_listener(self: &Rc<Self>, addr: SocketAddr) -> Result<Listener, Error> {
         Listener::new(self, addr)
     }
 
+    /// Connect to a remote worker by address.
     pub fn connect_addr(self: &Rc<Self>, addr: &WorkerAddress) -> Result<Endpoint, Error> {
         Endpoint::connect_addr(self, addr.handle)
     }
 
+    /// Connect to a remote listener.
     pub async fn connect_socket(self: &Rc<Self>, addr: SocketAddr) -> Result<Endpoint, Error> {
         Endpoint::connect_socket(self, addr).await
     }
 
+    /// Accept a connection request.
     pub async fn accept(self: &Rc<Self>, connection: ConnectionRequest) -> Result<Endpoint, Error> {
         Endpoint::accept(self, connection).await
     }