Add support for io_uring's direct descriptors

Cargo.toml

@@ -1,7 +1,8 @@
 [package]
 name          = "a10"
 description   = "A10 is an io_uring library."
-version       = "0.1.9"
+version       = "0.2.0"
+publish       = false # In development.
 authors       = ["Thomas de Zeeuw <[email protected]>"]
 license       = "MIT"
 documentation = "https://docs.rs/a10"

examples/cp.rs

@@ -7,7 +7,7 @@ use std::io;
 
 use a10::fs::OpenOptions;
 use a10::io::ReadBufPool;
-use a10::{Extract, SubmissionQueue};
+use a10::{AsyncFd, Extract, SubmissionQueue};
 
 mod runtime;
 
@@ -43,8 +43,8 @@ fn main() -> io::Result<()> {
 }
 
 async fn cp(sq: SubmissionQueue, source: String, destination: String) -> io::Result<()> {
-    let input = OpenOptions::new().open(sq.clone(), source.into()).await?;
-    let output = OpenOptions::new()
+    let input: AsyncFd = OpenOptions::new().open(sq.clone(), source.into()).await?;
+    let output: AsyncFd = OpenOptions::new()
         .write()
         .create_new()
         .open(sq.clone(), destination.into())

examples/http_client.rs

@@ -2,7 +2,7 @@ use std::net::{SocketAddr, SocketAddrV4};
 use std::{env, io, mem, str};
 
 use a10::net::socket;
-use a10::{Ring, SubmissionQueue};
+use a10::{AsyncFd, Ring, SubmissionQueue};
 
 mod runtime;
 
@@ -56,7 +56,7 @@ async fn request(sq: SubmissionQueue, host: &str, address: SocketAddrV4) -> io::
     let r#type = libc::SOCK_STREAM | libc::SOCK_CLOEXEC;
     let protocol = 0;
     let flags = 0;
-    let socket = socket(sq, domain, r#type, protocol, flags).await?;
+    let socket: AsyncFd = socket(sq, domain, r#type, protocol, flags).await?;
 
     // Connect.
     let addr = to_sockaddr_storage(address);

examples/read_file.rs

@@ -1,7 +1,7 @@
 use std::{env, io, str};
 
 use a10::fs::OpenOptions;
-use a10::{Ring, SubmissionQueue};
+use a10::{AsyncFd, Ring, SubmissionQueue};
 
 mod runtime;
 
@@ -34,7 +34,7 @@ fn main() -> io::Result<()> {
 
 async fn read_file(sq: SubmissionQueue, path: String) -> io::Result<Vec<u8>> {
     // Open a file for reading.
-    let file = OpenOptions::new().open(sq, path.into()).await?;
+    let file: AsyncFd = OpenOptions::new().open(sq, path.into()).await?;
 
     // Read some bytes from the file.
     let buf = file.read(Vec::with_capacity(32 * 1024)).await?;

src/cancel.rs

@@ -8,12 +8,13 @@ use std::io;
 use std::pin::Pin;
 use std::task::{self, Poll};
 
+use crate::fd::{AsyncFd, Descriptor};
 use crate::op::{op_future, poll_state, OpState};
-use crate::{libc, AsyncFd, OpIndex, QueueFull, SubmissionQueue};
+use crate::{libc, OpIndex, QueueFull, SubmissionQueue};
 
 /// Cancelation of operations, also see the [`Cancel`] trait to cancel specific
 /// operations.
-impl AsyncFd {
+impl<D: Descriptor> AsyncFd<D> {
     /// Attempt to cancel all in progress operations on this fd.
     ///
     /// If the I/O operations were succesfully canceled this returns `Ok(n)`,
@@ -32,7 +33,7 @@ impl AsyncFd {
     /// 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.
-    pub const fn cancel_all<'fd>(&'fd self) -> CancelAll<'fd> {
+    pub const fn cancel_all<'fd>(&'fd self) -> CancelAll<'fd, D> {
         CancelAll::new(self, libc::IORING_ASYNC_CANCEL_ALL)
     }
 }
@@ -45,7 +46,7 @@ op_future! {
     },
     setup_state: flags: u32,
     setup: |submission, fd, (), flags| unsafe {
-        submission.cancel(fd.fd(), flags);
+        submission.cancel(fd.fd(), flags | D::cancel_flag());
     },
     map_result: |n| {
         #[allow(clippy::cast_sign_loss)] // Negative values are mapped to errors.

src/config.rs

@@ -25,6 +25,7 @@ pub struct Config<'r> {
     kernel_thread: bool,
     cpu_affinity: Option<u32>,
     idle_timeout: Option<u32>,
+    direct_descriptors: Option<u32>,
     attach: Option<&'r SubmissionQueue>,
 }
 
@@ -67,6 +68,7 @@ impl<'r> Config<'r> {
             kernel_thread: true,
             cpu_affinity: None,
             idle_timeout: None,
+            direct_descriptors: None,
             attach: None,
         }
     }
@@ -194,6 +196,21 @@ impl<'r> Config<'r> {
         self
     }
 
+    /// Enable direct descriptors.
+    ///
+    /// This registers a sparse array of `size` direct descriptor slots enabling
+    /// direct descriptors to be used. If this is not used attempts to create a
+    /// direct descriptor will result in `ENXIO`.
+    ///
+    /// By default direct descriptors are not enabled.
+    #[doc(alias = "IORING_REGISTER_FILES")]
+    #[doc(alias = "IORING_REGISTER_FILES2")]
+    #[doc(alias = "IORING_RSRC_REGISTER_SPARSE")]
+    pub const fn with_direct_descriptors(mut self, size: u32) -> Self {
+        self.direct_descriptors = Some(size);
+        self
+    }
+
     /// Attach the new (to be created) ring to `other_ring`.
     ///
     /// This will cause the `Ring` being created to share the asynchronous
@@ -290,6 +307,22 @@ impl<'r> Config<'r> {
 
         let cq = mmap_completion_queue(fd.as_fd(), &parameters)?;
         let sq = mmap_submission_queue(fd, &parameters)?;
+
+        if let Some(size) = self.direct_descriptors {
+            let register = libc::io_uring_rsrc_register {
+                flags: libc::IORING_RSRC_REGISTER_SPARSE,
+                nr: size,
+                resv2: 0,
+                data: 0,
+                tags: 0,
+            };
+            sq.register(
+                libc::IORING_REGISTER_FILES2,
+                (&register as *const libc::io_uring_rsrc_register).cast(),
+                size_of::<libc::io_uring_rsrc_register>() as _,
+            )?;
+        }
+
         Ok(Ring { cq, sq })
     }
 }

src/extract.rs

@@ -46,12 +46,13 @@ use crate::cancel::{Cancel, CancelOp, CancelResult};
 /// use std::io;
 /// use std::path::PathBuf;
 ///
+/// use a10::fd::File;
 /// use a10::fs::OpenOptions;
 /// use a10::{SubmissionQueue, Extract};
 ///
 /// async fn write_all_to_file(sq: SubmissionQueue, path: PathBuf, buf: Vec<u8>) -> io::Result<(PathBuf, Vec<u8>)> {
 ///     // This `Future` returns just the opened file.
-///     let open_file_future = OpenOptions::new().open(sq, path);
+///     let open_file_future = OpenOptions::new().open::<File>(sq, path);
 ///     // Calling `extract` and awaiting that will return both the file and
 ///     // the path buffer.
 ///     let (file, path) = open_file_future.extract().await?;