Repository: Thomasdezeeuw/a10
Pull request #109 (2024-04-05 17:47:10)

Add additional configuration options


Patch:
diff --git a/src/config.rs b/src/config.rs
index 9913a1c..67a8ac1 100644
--- a/src/config.rs
+++ b/src/config.rs
@@ -14,10 +14,13 @@ use crate::{libc, AtomicBitMap, CompletionQueue, Ring, SharedSubmissionQueue, Su
 /// Created by calling [`Ring::config`].
 #[derive(Debug, Clone)]
 #[must_use = "no ring is created until `a10::Config::build` is called"]
+#[allow(clippy::struct_excessive_bools)] // This is just stupid.
 pub struct Config<'r> {
     submission_entries: u32,
     completion_entries: Option<u32>,
     disabled: bool,
+    single_issuer: bool,
+    defer_taskrun: bool,
     clamp: bool,
     kernel_thread: bool,
     cpu_affinity: Option<u32>,
@@ -58,6 +61,8 @@ impl<'r> Config<'r> {
             submission_entries: entries,
             completion_entries: None,
             disabled: false,
+            single_issuer: false,
+            defer_taskrun: false,
             clamp: false,
             kernel_thread: true,
             cpu_affinity: None,
@@ -76,6 +81,45 @@ impl<'r> Config<'r> {
         self
     }
 
+    /// Enable single issuer.
+    ///
+    /// This hints to the kernel that only a single thread will submit requests,
+    /// which is used for optimisations within the kernel. This means that only
+    /// the thread that [`build`] the ring or [`enabled`] it (after starting in
+    /// disable mode) may register resources with the ring, resources such as
+    /// the [`ReadBufPool`].
+    ///
+    /// This optimisation is enforces by the kernel, which will return `EEXIST`
+    /// or `AlreadyExists` if another thread attempt to register resource or
+    /// otherwise use the [`Ring`] in a way that is not allowed.
+    ///
+    /// [`build`]: Config::build
+    /// [`enabled`]: Ring::enable
+    /// [`ReadBufPool`]: crate::io::ReadBufPool
+    #[doc(alias = "IORING_SETUP_SINGLE_ISSUER")]
+    pub const fn single_issuer(mut self) -> Config<'r> {
+        self.single_issuer = true;
+        self
+    }
+
+    /// Defer task running.
+    ///
+    /// By default, kernel will process all outstanding work at the end of any
+    /// system call or thread interrupt. This can delay the application from
+    /// making other progress.
+    ///
+    /// Enabling this option will hint to kernel that it should defer work until
+    /// [`Ring::poll`] is called. This way the work is done in the
+    /// [`Ring::poll`].
+    ///
+    /// This options required [`Config::single_issuer`] to be set. This option
+    /// does not work with [`Config::with_kernel_thread`] set.
+    #[doc(alias = "IORING_SETUP_DEFER_TASKRUN")]
+    pub const fn defer_task_run(mut self) -> Config<'r> {
+        self.defer_taskrun = true;
+        self
+    }
+
     /// Set the size of the completion queue.
     ///
     /// By default the kernel will use a completion queue twice as large as the
@@ -176,16 +220,22 @@ impl<'r> Config<'r> {
         let mut parameters: libc::io_uring_params = unsafe { mem::zeroed() };
         parameters.flags = libc::IORING_SETUP_SUBMIT_ALL; // Submit all submissions on error.
         if self.kernel_thread {
-            parameters.flags |= libc::IORING_SETUP_SQPOLL // Kernel thread for polling.
-            // Using `IORING_SETUP_SQPOLL` we always have one issuer.
-            | libc::IORING_SETUP_SINGLE_ISSUER;
+            parameters.flags |= libc::IORING_SETUP_SQPOLL; // Kernel thread for polling.
         } else {
             // Don't interrupt userspace, the user must call `Ring::poll` any way.
             parameters.flags |= libc::IORING_SETUP_COOP_TASKRUN;
         }
         if self.disabled {
+            // Start the ring in disabled mode.
             parameters.flags |= libc::IORING_SETUP_R_DISABLED;
         }
+        if self.single_issuer {
+            // Only allow access from a single thread.
+            parameters.flags |= libc::IORING_SETUP_SINGLE_ISSUER;
+        }
+        if self.defer_taskrun {
+            parameters.flags |= libc::IORING_SETUP_DEFER_TASKRUN;
+        }
         if let Some(completion_entries) = self.completion_entries {
             parameters.cq_entries = completion_entries;
             parameters.flags |= libc::IORING_SETUP_CQSIZE;


Test patch:
diff --git a/tests/ring.rs b/tests/ring.rs
index ee4d0a6..f901ddb 100644
--- a/tests/ring.rs
+++ b/tests/ring.rs
@@ -17,6 +17,7 @@ use std::time::{Duration, Instant};
 
 use a10::cancel::Cancel;
 use a10::fs::OpenOptions;
+use a10::io::ReadBufPool;
 use a10::msg::{MsgListener, MsgToken, SendMsg};
 use a10::poll::{MultishotPoll, OneshotPoll};
 use a10::{mem, process, AsyncFd, Config, Ring, SubmissionQueue};
@@ -28,6 +29,7 @@ use util::{
 };
 
 const DATA: &[u8] = b"Hello, World!";
+const BUF_SIZE: usize = 4096;
 
 #[test]
 fn polling_timeout() -> io::Result<()> {
@@ -151,6 +153,7 @@ fn submission_queue_full_is_handle_internally() {
 
 #[test]
 fn config_disabled() {
+    require_kernel!(5, 10);
     init();
     let mut ring = Ring::config(1).disable().build().unwrap();
 
@@ -163,6 +166,61 @@ fn config_disabled() {
     ring.poll(Some(Duration::from_millis(1))).unwrap();
 }
 
+#[test]
+fn config_single_issuer() {
+    require_kernel!(6, 0);
+    init();
+
+    let ring = Ring::config(1).single_issuer().build().unwrap();
+
+    // This is fine.
+    let buf_pool = ReadBufPool::new(ring.submission_queue().clone(), 2, BUF_SIZE as u32).unwrap();
+    drop(buf_pool);
+
+    thread::spawn(move || {
+        // This is not (we're on a different thread).
+        let err =
+            ReadBufPool::new(ring.submission_queue().clone(), 2, BUF_SIZE as u32).unwrap_err();
+        assert_eq!(err.raw_os_error(), Some(libc::EEXIST));
+    })
+    .join()
+    .unwrap();
+}
+
+#[test]
+fn config_single_issuer_disabled_ring() {
+    require_kernel!(6, 0);
+    init();
+
+    let mut ring = Ring::config(1).single_issuer().disable().build().unwrap();
+
+    thread::spawn(move || {
+        ring.enable().unwrap();
+
+        // Since this thread enabled the ring, this is now fine.
+        let buf_pool =
+            ReadBufPool::new(ring.submission_queue().clone(), 2, BUF_SIZE as u32).unwrap();
+        drop(buf_pool);
+    })
+    .join()
+    .unwrap();
+}
+
+#[test]
+fn config_defer_task_run() {
+    require_kernel!(6, 1);
+    init();
+
+    let mut ring = Ring::config(1)
+        .single_issuer()
+        .defer_task_run()
+        .with_kernel_thread(false)
+        .build()
+        .unwrap();
+
+    ring.poll(Some(Duration::ZERO)).unwrap();
+}
+
 #[test]
 fn wake_ring() {
     init();