Rewrite internals

Cargo.toml

@@ -1,7 +1,8 @@
 [package]
 name          = "a10"
 description   = "A10 is an io_uring library."
-version       = "0.2.2"
+version       = "0.3.0"
+publish       = false # In development.
 authors       = ["Thomas de Zeeuw <[email protected]>"]
 license       = "MIT"
 documentation = "https://docs.rs/a10"
@@ -11,6 +12,7 @@ keywords      = ["io_uring", "io", "async", "non-blocking"]
 categories    = ["asynchronous"]
 include       = ["/Cargo.toml", "/src/**/*.rs", "/tests/**/*.rs", "/README.md", "/LICENSE"]
 edition       = "2021"
+rust-version  = "1.84"
 
 [features]
 default = []
@@ -25,6 +27,7 @@ log          = { version = "0.4.21",  default-features = false, features = ["kv_
 
 [dev-dependencies]
 std-logger   = { version = "0.5.3", default-features = false }
+getrandom    = { version = "0.2.15", default-features = false }
 
 [[test]]
 name    = "signals"

Makefile

@@ -50,21 +50,19 @@ clippy:
 		--deny clippy::cargo \
 		--allow clippy::doc-markdown \
 		--allow clippy::equatable-if-let \
-		--allow clippy::manual-c-str-literals \
 		--allow clippy::missing-errors-doc \
 		--allow clippy::missing-panics-doc \
 		--allow clippy::must-use-candidate \
 		--allow clippy::needless-lifetimes \
 		--allow clippy::new-without-default \
 		--allow clippy::option-if-let-else \
 		--allow clippy::redundant-pub-crate \
-		--allow clippy::ref-as-ptr \
 		--allow clippy::single-match-else \
 		--allow clippy::use-self \
-		--allow clippy::manual-inspect \
 		\
 		--allow clippy::cast-possible-truncation \
 		--allow clippy::cast-possible-wrap \
+		--allow clippy::cast-sign-loss \
 
 doc:
 	cargo doc

examples/http_client.rs

@@ -1,5 +1,5 @@
-use std::net::{SocketAddr, SocketAddrV4};
-use std::{env, io, mem, str};
+use std::net::SocketAddr;
+use std::{env, io, str};
 
 use a10::net::socket;
 use a10::{AsyncFd, Ring, SubmissionQueue};
@@ -22,13 +22,8 @@ fn main() -> io::Result<()> {
 
     // Get an IPv4 address for the domain (using blocking I/O).
     let address = std::net::ToSocketAddrs::to_socket_addrs(&addr_host)?
-        .filter(SocketAddr::is_ipv4)
         .next()
         .ok_or_else(|| io::Error::new(io::ErrorKind::Other, "failed to lookup ip"))?;
-    let address = match address {
-        SocketAddr::V4(address) => address,
-        SocketAddr::V6(_) => unreachable!(),
-    };
 
     // Create our future that makes the request.
     let request_future = request(ring.submission_queue().clone(), &host, address);
@@ -50,7 +45,7 @@ fn main() -> io::Result<()> {
 }
 
 /// Make a HTTP GET request to `address`.
-async fn request(sq: SubmissionQueue, host: &str, address: SocketAddrV4) -> io::Result<Vec<u8>> {
+async fn request(sq: SubmissionQueue, host: &str, address: SocketAddr) -> io::Result<Vec<u8>> {
     // Create a new TCP, IPv4 socket.
     let domain = libc::AF_INET;
     let r#type = libc::SOCK_STREAM | libc::SOCK_CLOEXEC;
@@ -59,14 +54,13 @@ async fn request(sq: SubmissionQueue, host: &str, address: SocketAddrV4) -> io::
     let socket: AsyncFd = socket(sq, domain, r#type, protocol, flags).await?;
 
     // Connect.
-    let addr = to_sockaddr_storage(address);
-    socket.connect(addr).await?;
+    socket.connect(address).await?;
 
     // Send a HTTP GET / request to the socket.
     let host = host.split_once(':').map(|(h, _)| h).unwrap_or(host);
     let version = env!("CARGO_PKG_VERSION");
     let request = format!("GET / HTTP/1.1\r\nHost: {host}\r\nUser-Agent: A10-example/{version}\r\nAccept: */*\r\n\r\n");
-    socket.send(request, 0).await?;
+    socket.send_all(request, 0).await?;
 
     // Receiving the response.
     let recv_buf = socket.recv(Vec::with_capacity(8192), 0).await?;
@@ -77,14 +71,3 @@ async fn request(sq: SubmissionQueue, host: &str, address: SocketAddrV4) -> io::
 
     Ok(recv_buf)
 }
-
-fn to_sockaddr_storage(addr: SocketAddrV4) -> libc::sockaddr_in {
-    // SAFETY: a `sockaddr_in` of all zeros is valid.
-    let mut storage: libc::sockaddr_in = unsafe { mem::zeroed() };
-    storage.sin_family = libc::AF_INET as _;
-    storage.sin_port = addr.port().to_be();
-    storage.sin_addr = libc::in_addr {
-        s_addr: u32::from_ne_bytes(addr.ip().octets()),
-    };
-    storage
-}

src/cancel.rs

@@ -4,13 +4,13 @@
 //! [`AsyncFd::cancel_all`] to cancel all operations on a fd.
 
 use std::future::Future;
-use std::io;
 use std::pin::Pin;
 use std::task::{self, Poll};
+use std::{fmt, io};
 
 use crate::fd::{AsyncFd, Descriptor};
-use crate::op::{op_future, poll_state, OpState};
-use crate::{libc, OpIndex, QueueFull, SubmissionQueue};
+use crate::op::{fd_operation, FdOperation, Operation};
+use crate::{sys, OperationId, SubmissionQueue};
 
 /// Cancelation of operations, also see the [`Cancel`] trait to cancel specific
 /// operations.
@@ -33,26 +33,17 @@ impl<D: Descriptor> AsyncFd<D> {
     /// Due to the lazyness of [`Future`]s it is possible that this will return
     /// `Ok(0)` if operations were never polled only to start it after their
     /// first poll.
+    ///
+    /// [`Future`]: std::future::Future
     pub const fn cancel_all<'fd>(&'fd self) -> CancelAll<'fd, D> {
-        CancelAll::new(self, libc::IORING_ASYNC_CANCEL_ALL)
+        CancelAll(FdOperation::new(self, (), ()))
     }
 }
 
-// CancelAll.
-op_future! {
-    fn AsyncFd::cancel_all -> usize,
-    struct CancelAll<'fd> {
-        // Doesn't need any fields.
-    },
-    setup_state: flags: u32,
-    setup: |submission, fd, (), flags| unsafe {
-        submission.cancel(fd.fd(), flags | D::cancel_flag());
-    },
-    map_result: |n| {
-        #[allow(clippy::cast_sign_loss)] // Negative values are mapped to errors.
-        Ok(n as usize)
-    },
-}
+fd_operation!(
+    /// [`Future`] behind [`AsyncFd::cancel_all`].
+    pub struct CancelAll(sys::cancel::CancelAllOp) -> io::Result<usize>;
+);
 
 /// Cancelation of an in progress operations.
 pub trait Cancel {
@@ -67,6 +58,15 @@ pub trait Cancel {
 
     /// Cancel this operation.
     ///
+    /// Once the returned future is completed it will asynchronously cancel the
+    /// related operation. This means that it *may* still return results that
+    /// were created before the operation was actually canceled.
+    ///
+    /// For example using a TCP listener and multishot accept it's possible that
+    /// `MultishotAccept` will return more accepted connections after it's canceled.
+    /// Simply keep accepting the connections and it will return `None` after all
+    /// pending connections have been accepted.
+    ///
     /// If this returns `ENOENT` it means the operation was not found. This can
     /// be caused by the operation never starting, due to the inert nature of
     /// [`Future`]s, or the operation has already been completed.
@@ -79,8 +79,9 @@ pub trait Cancel {
     /// If this is called on an [`AsyncIterator`] it will cause them to return
     /// `None` (eventually, it may still return pending items).
     ///
+    /// [`Future`]: std::future::Future
     /// [`AsyncIterator`]: std::async_iter::AsyncIterator
-    fn cancel(&mut self) -> CancelOp;
+    fn cancel(&mut self) -> CancelOperation;
 }
 
 /// Result of a cancelation attempt.
@@ -101,77 +102,46 @@ pub enum CancelResult {
 }
 
 /// [`Future`] behind [`Cancel::cancel`].
-///
-/// Once this future is completed it will asynchronously cancel the related
-/// operation. This means that it *may* still return results that were created
-/// before the operation was actually canceled.
-///
-/// For example using a TCP listener and multishot accept it's possible that
-/// `MultishotAccept` will return more accepted connections after it's canceled.
-/// Simply keep accepting the connections and it will return `None` after all
-/// pending connections have been accepted.
-///
-///[`MultishotAccept::cancel`]: crate::net::MultishotAccept::cancel
-#[derive(Debug)]
 #[must_use = "`Future`s do nothing unless polled"]
-#[allow(clippy::module_name_repetitions)] // Don't care.
-pub struct CancelOp<'fd> {
-    sq: &'fd SubmissionQueue,
-    state: OpState<Option<OpIndex>>,
-}
+pub struct CancelOperation(pub(crate) CancelOperationState);
 
-impl<'fd> CancelOp<'fd> {
-    /// Create a new `CancelOp`.
-    pub(crate) const fn new(sq: &'fd SubmissionQueue, op_index: Option<OpIndex>) -> CancelOp<'fd> {
-        CancelOp {
-            sq,
-            state: OpState::NotStarted(op_index),
+impl CancelOperation {
+    /// Create a new `CancelOperation`.
+    pub(crate) fn new(sq: SubmissionQueue, op_id: Option<OperationId>) -> CancelOperation {
+        if let Some(op_id) = op_id {
+            let operation = Operation::new(sq, (), op_id);
+            CancelOperation(CancelOperationState::InProgress(operation))
+        } else {
+            CancelOperation(CancelOperationState::Done)
         }
     }
 }
 
-impl<'fd> Future for CancelOp<'fd> {
+/// State of `CancelOperation`.
+pub(crate) enum CancelOperationState {
+    /// Cancellation is already done, or the operation was never started.
+    Done,
+    /// Cancellation is in progress.
+    InProgress(Operation<sys::cancel::CancelOperationOp>),
+}
+
+impl Future for CancelOperation {
     type Output = io::Result<()>;
 
     fn poll(mut self: Pin<&mut Self>, ctx: &mut task::Context<'_>) -> Poll<Self::Output> {
-        // NOTE: don't use `poll_state!` here bacause we return `ENOENT` if it
-        // hasn't started.
-        let op_index = match self.state {
-            OpState::Running(op_index) => op_index,
-            OpState::NotStarted(Some(to_cancel_op_index)) => {
-                // SAFETY: this will not panic as the resources are only removed
-                // after the state is set to `Done`.
-                let result = self
-                    .sq
-                    .add(|submission| unsafe { submission.cancel_op(to_cancel_op_index) });
-                match result {
-                    Ok(op_index) => {
-                        self.state = OpState::Running(op_index);
-                        op_index
-                    }
-                    Err(QueueFull(())) => {
-                        self.sq.wait_for_submission(ctx.waker().clone());
-                        return Poll::Pending;
-                    }
-                }
-            }
-            // If the operation is not started we pretend like we didn't find
-            // it.
-            OpState::NotStarted(None) => {
-                return Poll::Ready(Err(io::Error::from_raw_os_error(libc::ENOENT)))
-            }
-            OpState::Done => poll_state!(__panic CancelOp),
-        };
+        match &mut self.0 {
+            CancelOperationState::Done => Poll::Ready(Ok(())),
+            CancelOperationState::InProgress(op) => Pin::new(op).poll(ctx),
+        }
+    }
+}
 
-        match self.sq.poll_op(ctx, op_index) {
-            Poll::Ready(result) => {
-                self.state = OpState::Done;
-                match result {
-                    Ok((_, _)) => Poll::Ready(Ok(())),
-                    Err(err) => Poll::Ready(Err(err)),
-                }
-            }
-            Poll::Pending => Poll::Pending,
+impl fmt::Debug for CancelOperation {
+    fn fmt(&self, f: &mut ::std::fmt::Formatter<'_>) -> ::std::fmt::Result {
+        const NAME: &str = "a10::CancelOperation";
+        match &self.0 {
+            CancelOperationState::Done => f.debug_struct(NAME).field("state", &"done").finish(),
+            CancelOperationState::InProgress(op) => op.fmt_dbg(NAME, f),
         }
     }
 }