bearer: Inject SubmissionFlags (#41)

* bearer: Inject SubmissionFlags

* Remove dead comments

Author: pinkforest

io-uring-bearer/src/error.rs

@@ -21,6 +21,8 @@ pub enum UringBearerError {
     Submission(String),
     /// Something wrong Slab                                        
     Slab(String),
+    /// Invalid Submission Flags encountered
+    SubmissionFlags,
     /// Slab was able to store the value but not get it? This is a bug.
     SlabBugSetGet(&'static str),
     /// Register handles error
@@ -64,6 +66,7 @@ impl Display for UringBearerError {
             Self::SubmissionPush => write!(f, "Submisionn push error. Is the squeue full?"),
             Self::Submission(s) => write!(f, "Submission: {}", s),
             Self::Slab(s) => write!(f, "Slab: {}", s),
+            Self::SubmissionFlags => write!(f, "Invalid flags set in Submission"),
             Self::SlabBugSetGet(s) => write!(f, "Slab Bug: {}", s),
             Self::RegisterHandles(s) => write!(f, "Register Handles: {}", s),
             Self::Slabbable(e) => write!(f, "Slabbable: {}", e),

io-uring-bearer/src/lib.rs

@@ -33,6 +33,13 @@ pub(crate) mod fixed;
 mod capacity;
 pub use capacity::BearerCapacityKind;
 
+//-----------------------------------------------
+// Submission types
+//-----------------------------------------------
+mod submission;
+#[doc(inline)]
+pub use submission::SubmissionFlags;
+
 //-----------------------------------------------
 // Completion types
 //-----------------------------------------------

io-uring-bearer/src/submission.rs

@@ -0,0 +1,60 @@
+//! Submission related types
+
+use crate::error::UringBearerError;
+use io_uring::squeue::Flags as IoUringFlags;
+
+/// See Linux io_uring_sqe_set_flags(3) for IOSQE_ASYNC for the respective documentation.
+/// These flags may be set for each submission queue entry. By default all flags are Off.
+#[derive(Clone, Debug, Default, PartialEq)]
+pub struct SubmissionFlags {
+    pub(crate) bits: u8,
+}
+
+impl SubmissionFlags {
+    /// Submission of this entry will not start before all the prior entries are completed.
+    /// Submissions after this entry will not complete before this entry is completed.
+    #[inline]
+    pub fn on_io_drain(mut self) -> Self {
+        self.bits |= IoUringFlags::IO_DRAIN.bits();
+        self
+    }
+    /// Submission is linked to the next after this so it will not start before this is completed.
+    #[inline]
+    pub fn on_io_link(mut self) -> Self {
+        self.bits |= IoUringFlags::IO_LINK.bits();
+        self
+    }
+    /// Same as on_io_link but does not sever regardless of completion result.
+    // TODO(doc): what does this actually mean - e.g. will next one get cancelled if this fails ? etc.
+    #[inline]
+    pub fn on_io_hard_link(mut self) -> Self {
+        self.bits |= IoUringFlags::IO_HARDLINK.bits();
+        self
+    }
+    /// Issue direct async submission over non-blocking directly. See Linux io_uring_sqe_set_flags(3).
+    #[inline]
+    pub fn on_async(mut self) -> Self {
+        self.bits |= IoUringFlags::ASYNC.bits();
+        self
+    }
+    /// Signal to kernel to select a buffer from a given group id.
+    #[inline]
+    pub fn on_buffer_select(mut self) -> Self {
+        self.bits |= IoUringFlags::BUFFER_SELECT.bits();
+        self
+    }
+    /// Skip all the completion events when sucessfull out of this submission.
+    #[inline]
+    pub fn on_skip_success(mut self) -> Self {
+        self.bits |= IoUringFlags::SKIP_SUCCESS.bits();
+        self
+    }
+    /// Convert to [`io-uring::squeue::Flags`]
+    #[inline]
+    pub fn to_io_uring_flags(&self) -> Result<io_uring::squeue::Flags, UringBearerError> {
+        match io_uring::squeue::Flags::from_bits(self.bits) {
+            None => Err(UringBearerError::SubmissionFlags),
+            Some(ret) => Ok(ret),
+        }
+    }
+}

io-uring-bearer/src/uring.rs

@@ -22,6 +22,7 @@ use io_uring::IoUring;
 
 use crate::completion::SubmissionRecordStatus;
 use crate::fixed::FixedFdRegister;
+use crate::SubmissionFlags;
 
 use crate::slab::BuffersRec;
 use crate::slab::FutexRec;
@@ -177,34 +178,52 @@ impl<C: core::fmt::Debug + Clone + OpCompletion> UringBearer<C> {
             .map_err(|e| UringBearerError::Submission(e.to_string()))
     }
     /// Push a general Op implementing OpCode trait (see io-uring-opcode)
-    pub fn push_op<Op: OpCode<C>>(&mut self, op: Op) -> Result<usize, UringBearerError> {
+    pub fn push_op<Op: OpCode<C>>(
+        &mut self,
+        op: Op,
+        flags: Option<SubmissionFlags>,
+    ) -> Result<usize, UringBearerError> {
         let key = self
             .fd_slab
             .take_next_with(Completion::Op(op.submission()?))
             .map_err(UringBearerError::Slabbable)?;
 
-        match self._push_to_completion(key) {
+        match self._push_to_completion(key, flags) {
             Err(e) => Err(e),
             Ok(()) => Ok(key),
         }
     }
     /// Push a pending typed Completion directly
-    pub fn push_op_typed(&mut self, op: Completion<C>) -> Result<usize, UringBearerError> {
+    pub fn push_op_typed(
+        &mut self,
+        op: Completion<C>,
+        flags: Option<SubmissionFlags>,
+    ) -> Result<usize, UringBearerError> {
         let key = self
             .fd_slab
             .take_next_with(op)
             .map_err(UringBearerError::Slabbable)?;
 
-        match self._push_to_completion(key) {
+        match self._push_to_completion(key, flags) {
             Err(e) => Err(e),
             Ok(()) => Ok(key),
         }
     }
     #[inline]
-    pub(crate) fn _push_to_completion(&mut self, idx: usize) -> Result<(), UringBearerError> {
+    pub(crate) fn _push_to_completion(
+        &mut self,
+        idx: usize,
+        in_flags: Option<SubmissionFlags>,
+    ) -> Result<(), UringBearerError> {
         let iou = &mut self.io_uring;
         let mut s_queue = iou.submission();
 
+        let flags = match in_flags {
+            None => SubmissionFlags::default(),
+            Some(fl) => fl,
+        }
+        .to_io_uring_flags()?;
+
         let completion_rec = self
             .fd_slab
             .slot_get_mut(idx)
@@ -216,7 +235,7 @@ impl<C: core::fmt::Debug + Clone + OpCompletion> UringBearer<C> {
                     return Err(UringBearerError::InvalidOwnership(completion.owner(), idx));
                 }
                 completion.force_owner_kernel();
-                completion.entry().user_data(idx as u64)
+                completion.entry().flags(flags).user_data(idx as u64)
             }
             _ => return Err(UringBearerError::SlabBugSetGet("Submisison not found?")),
         };

io-uring-bearer/src/uring/accept_multi.rs

@@ -3,13 +3,18 @@
 use super::UringBearer;
 use crate::error::UringBearerError;
 use crate::Completion;
+use crate::SubmissionFlags;
 use io_uring_opcode::OpExtAcceptMulti;
 use io_uring_opcode::{OpCode, OpCompletion};
 use slabbable::Slabbable;
 
 impl<C: core::fmt::Debug + Clone + OpCompletion> UringBearer<C> {
     /// Push a AcceptMulti implementing OpCode +  traits (see io-uring-opcode)
-    pub fn push_accept_multi<Op>(&mut self, op: Op) -> Result<usize, UringBearerError>
+    pub fn push_accept_multi<Op>(
+        &mut self,
+        op: Op,
+        flags: Option<SubmissionFlags>,
+    ) -> Result<usize, UringBearerError>
     where
         Op: OpCode<C> + OpExtAcceptMulti,
     {
@@ -18,7 +23,7 @@ impl<C: core::fmt::Debug + Clone + OpCompletion> UringBearer<C> {
             .take_next_with(Completion::AcceptMulti(op.submission()?))
             .map_err(UringBearerError::Slabbable)?;
 
-        match self._push_to_completion(key) {
+        match self._push_to_completion(key, flags) {
             Err(e) => Err(e),
             Ok(()) => Ok(key),
         }

io-uring-bearer/src/uring/connect.rs

@@ -3,13 +3,18 @@
 use super::UringBearer;
 use crate::error::UringBearerError;
 use crate::Completion;
+use crate::SubmissionFlags;
 use io_uring_opcode::OpExtConnect;
 use io_uring_opcode::{OpCode, OpCompletion};
 use slabbable::Slabbable;
 
 impl<C: core::fmt::Debug + Clone + OpCompletion> UringBearer<C> {
     /// Push a Connect implementing OpCode +  traits (see io-uring-opcode)
-    pub fn push_connect<Op>(&mut self, op: Op) -> Result<usize, UringBearerError>
+    pub fn push_connect<Op>(
+        &mut self,
+        op: Op,
+        flags: Option<SubmissionFlags>,
+    ) -> Result<usize, UringBearerError>
     where
         Op: OpCode<C> + OpExtConnect,
     {
@@ -18,7 +23,7 @@ impl<C: core::fmt::Debug + Clone + OpCompletion> UringBearer<C> {
             .take_next_with(Completion::Connect(op.submission()?))
             .map_err(UringBearerError::Slabbable)?;
 
-        match self._push_to_completion(key) {
+        match self._push_to_completion(key, flags) {
             Err(e) => Err(e),
             Ok(()) => Ok(key),
         }

io-uring-bearer/src/uring/epoll_ctl.rs

@@ -3,13 +3,18 @@
 use super::UringBearer;
 use crate::error::UringBearerError;
 use crate::Completion;
+use crate::SubmissionFlags;
 use io_uring_opcode::OpExtEpollCtl;
 use io_uring_opcode::{OpCode, OpCompletion};
 use slabbable::Slabbable;
 
 impl<C: core::fmt::Debug + Clone + OpCompletion> UringBearer<C> {
     /// Push an EpollCtl implementing OpCode +  traits (see io-uring-opcode)
-    pub fn push_epoll_ctl<Op>(&mut self, op: Op) -> Result<usize, UringBearerError>
+    pub fn push_epoll_ctl<Op>(
+        &mut self,
+        op: Op,
+        flags: Option<SubmissionFlags>,
+    ) -> Result<usize, UringBearerError>
     where
         Op: OpCode<C> + OpExtEpollCtl,
     {
@@ -18,7 +23,7 @@ impl<C: core::fmt::Debug + Clone + OpCompletion> UringBearer<C> {
             .take_next_with(Completion::EpollCtl(op.submission()?))
             .map_err(UringBearerError::Slabbable)?;
 
-        match self._push_to_completion(key) {
+        match self._push_to_completion(key, flags) {
             Err(e) => Err(e),
             Ok(()) => Ok(key),
         }

io-uring-bearer/src/uring/recv.rs

@@ -5,13 +5,19 @@ use crate::Completion;
 use crate::UringBearer;
 
 use crate::slab::{RecvMultiRec, RecvRec};
+use crate::SubmissionFlags;
 
 use io_uring_opcode::OpCompletion;
 use slabbable::Slabbable;
 
 impl<C: core::fmt::Debug + Clone + OpCompletion> UringBearer<C> {
     /// Add Recv pending Completion. The referenced buffers index must hold only one buffer.
-    pub fn add_recv(&mut self, fixed_fd: u32, buf_idx: usize) -> Result<usize, UringBearerError> {
+    pub fn add_recv(
+        &mut self,
+        fixed_fd: u32,
+        buf_idx: usize,
+        flags: Option<SubmissionFlags>,
+    ) -> Result<usize, UringBearerError> {
         let taken_buf = self.take_one_mutable_buffer(buf_idx)?;
         if !self._fixed_fd_validate(fixed_fd) {
             return Err(UringBearerError::FdNotRegistered(fixed_fd));
@@ -21,14 +27,15 @@ impl<C: core::fmt::Debug + Clone + OpCompletion> UringBearer<C> {
             .take_next_with(Completion::Recv(RecvRec::new(fixed_fd as u32, taken_buf)))
             .map_err(UringBearerError::Slabbable)?;
 
-        self._push_to_completion(key)?;
+        self._push_to_completion(key, flags)?;
         Ok(key)
     }
     /// Add RecvMulti pending Completion
     pub fn add_recv_multi(
         &mut self,
         fixed_fd: u32,
         buf_group: u16,
+        flags: Option<SubmissionFlags>,
     ) -> Result<usize, UringBearerError> {
         if !self._fixed_fd_validate(fixed_fd) {
             return Err(UringBearerError::FdNotRegistered(fixed_fd));
@@ -41,7 +48,7 @@ impl<C: core::fmt::Debug + Clone + OpCompletion> UringBearer<C> {
             )))
             .map_err(UringBearerError::Slabbable)?;
 
-        self._push_to_completion(key)?;
+        self._push_to_completion(key, flags)?;
         Ok(key)
     }
 }

io-uring-bearer/src/uring/send_zc.rs

@@ -2,6 +2,7 @@
 
 use crate::uring::UringBearerError;
 use crate::Completion;
+use crate::SubmissionFlags;
 use crate::UringBearer;
 
 use crate::slab::SendZcRec;
@@ -18,6 +19,7 @@ impl<C: core::fmt::Debug + Clone + OpCompletion> UringBearer<C> {
         fixed_fd: u32,
         buf_idx: usize,
         kernel_index: u16,
+        flags: Option<SubmissionFlags>,
     ) -> Result<usize, UringBearerError> {
         let taken_buf = self.take_one_immutable_buffer(buf_idx, kernel_index)?;
         if !self._fixed_fd_validate(fixed_fd) {
@@ -31,7 +33,7 @@ impl<C: core::fmt::Debug + Clone + OpCompletion> UringBearer<C> {
             )))
             .map_err(UringBearerError::Slabbable)?;
 
-        self._push_to_completion(key)?;
+        self._push_to_completion(key, flags)?;
         Ok(key)
     }
 }

io-uring-bearer/src/uring/socket.rs

@@ -3,13 +3,18 @@
 use super::UringBearer;
 use crate::error::UringBearerError;
 use crate::Completion;
+use crate::SubmissionFlags;
 use io_uring_opcode::OpExtSocket;
 use io_uring_opcode::{OpCode, OpCompletion};
 use slabbable::Slabbable;
 
 impl<C: core::fmt::Debug + Clone + OpCompletion> UringBearer<C> {
     /// Push a Socket implementing OpCode +  traits (see io-uring-opcode)
-    pub fn push_socket<Op>(&mut self, op: Op) -> Result<usize, UringBearerError>
+    pub fn push_socket<Op>(
+        &mut self,
+        op: Op,
+        flags: Option<SubmissionFlags>,
+    ) -> Result<usize, UringBearerError>
     where
         Op: OpCode<C> + OpExtSocket,
     {
@@ -18,7 +23,7 @@ impl<C: core::fmt::Debug + Clone + OpCompletion> UringBearer<C> {
             .take_next_with(Completion::Socket(op.submission()?))
             .map_err(UringBearerError::Slabbable)?;
 
-        match self._push_to_completion(key) {
+        match self._push_to_completion(key, flags) {
             Err(e) => Err(e),
             Ok(()) => Ok(key),
         }