fs: add io_uring open operation

Author: mox692

tokio/src/fs/open_options.rs

@@ -1,17 +1,27 @@
-use crate::fs::{asyncify, File};
+use crate::fs::File;
 
 use std::io;
 use std::path::Path;
 
+cfg_tokio_unstable_uring! {
+    mod uring_open_options;
+    use uring_open_options::UringOpenOptions;
+    use crate::runtime::driver::op::Op;
+}
+
 #[cfg(test)]
 mod mock_open_options;
 #[cfg(test)]
 use mock_open_options::MockOpenOptions as StdOpenOptions;
 #[cfg(not(test))]
 use std::fs::OpenOptions as StdOpenOptions;
 
-#[cfg(unix)]
-use std::os::unix::fs::OpenOptionsExt;
+cfg_not_tokio_unstable_uring! {
+    #[cfg(unix)]
+        use std::os::unix::fs::OpenOptionsExt;
+        use crate::fs::asyncify;
+}
+
 #[cfg(windows)]
 use std::os::windows::fs::OpenOptionsExt;
 
@@ -79,7 +89,22 @@ use std::os::windows::fs::OpenOptionsExt;
 /// }
 /// ```
 #[derive(Clone, Debug)]
-pub struct OpenOptions(StdOpenOptions);
+pub struct OpenOptions(
+    #[cfg(not(all(
+        tokio_unstable_uring,
+        feature = "rt",
+        feature = "fs",
+        target_os = "linux",
+    )))]
+    StdOpenOptions,
+    #[cfg(all(
+        tokio_unstable_uring,
+        feature = "rt",
+        feature = "fs",
+        target_os = "linux",
+    ))]
+    pub(crate) UringOpenOptions,
+);
 
 impl OpenOptions {
     /// Creates a blank new set of options ready for configuration.
@@ -99,7 +124,22 @@ impl OpenOptions {
     /// let future = options.read(true).open("foo.txt");
     /// ```
     pub fn new() -> OpenOptions {
-        OpenOptions(StdOpenOptions::new())
+        OpenOptions(
+            #[cfg(not(all(
+                tokio_unstable_uring,
+                feature = "rt",
+                feature = "fs",
+                target_os = "linux",
+            )))]
+            StdOpenOptions::new(),
+            #[cfg(all(
+                tokio_unstable_uring,
+                feature = "rt",
+                feature = "fs",
+                target_os = "linux",
+            ))]
+            UringOpenOptions::new(),
+        )
     }
 
     /// Sets the option for read access.
@@ -386,17 +426,35 @@ impl OpenOptions {
     /// [`Other`]: std::io::ErrorKind::Other
     /// [`PermissionDenied`]: std::io::ErrorKind::PermissionDenied
     pub async fn open(&self, path: impl AsRef<Path>) -> io::Result<File> {
-        let path = path.as_ref().to_owned();
-        let opts = self.0.clone();
+        self.open_inner(path).await
+    }
 
-        let std = asyncify(move || opts.open(path)).await?;
-        Ok(File::from_std(std))
+    cfg_not_tokio_unstable_uring! {
+        async fn open_inner(&self, path: impl AsRef<Path>) -> io::Result<File> {
+            let path = path.as_ref().to_owned();
+            let opts = self.0.clone();
+
+            let std = asyncify(move || opts.open(path)).await?;
+            Ok(File::from_std(std))
+        }
+
+        /// Returns a mutable reference to the underlying `std::fs::OpenOptions`
+        #[cfg(any(windows, unix))]
+        pub(super) fn as_inner_mut(&mut self) -> &mut StdOpenOptions {
+            &mut self.0
+        }
     }
 
-    /// Returns a mutable reference to the underlying `std::fs::OpenOptions`
-    #[cfg(any(windows, unix))]
-    pub(super) fn as_inner_mut(&mut self) -> &mut StdOpenOptions {
-        &mut self.0
+    cfg_tokio_unstable_uring! {
+        async fn open_inner(&self, path: impl AsRef<Path>) -> io::Result<File> {
+            Op::open(path.as_ref(), self)?.await
+        }
+
+        /// Returns a mutable reference to the underlying `std::fs::OpenOptions`
+        #[cfg(any(windows, unix))]
+        pub(super) fn as_inner_mut(&mut self) -> &mut UringOpenOptions {
+            &mut self.0
+        }
     }
 }
 
@@ -649,9 +707,23 @@ cfg_windows! {
     }
 }
 
-impl From<StdOpenOptions> for OpenOptions {
-    fn from(options: StdOpenOptions) -> OpenOptions {
-        OpenOptions(options)
+cfg_not_tokio_unstable_uring! {
+    impl From<StdOpenOptions> for OpenOptions {
+        fn from(options: StdOpenOptions) -> OpenOptions {
+            OpenOptions(options)
+        }
+    }
+}
+
+cfg_tokio_unstable_uring! {
+    impl From<StdOpenOptions> for OpenOptions {
+        fn from(_options: StdOpenOptions) -> OpenOptions {
+            // It's not straitforward to convert from std's OpenOptions to io_uring's one.
+            // * https://github.com/rust-lang/rust/issues/74943
+            // * https://github.com/rust-lang/rust/issues/76801
+
+            panic!("Conversion from std's OpenOptions to io_uring's one is not supported")
+        }
     }
 }
 

tokio/src/fs/open_options/uring_open_options.rs

@@ -0,0 +1,103 @@
+use std::io;
+
+#[derive(Debug, Clone)]
+pub(crate) struct UringOpenOptions {
+    pub(crate) read: bool,
+    pub(crate) write: bool,
+    pub(crate) append: bool,
+    pub(crate) truncate: bool,
+    pub(crate) create: bool,
+    pub(crate) create_new: bool,
+    pub(crate) mode: libc::mode_t,
+    pub(crate) custom_flags: libc::c_int,
+}
+
+impl UringOpenOptions {
+    pub(crate) fn new() -> Self {
+        Self {
+            read: false,
+            write: false,
+            append: false,
+            truncate: false,
+            create: false,
+            create_new: false,
+            mode: 0o666,
+            custom_flags: 0,
+        }
+    }
+
+    pub(crate) fn append(&mut self, append: bool) -> &mut Self {
+        self.append = append;
+        self
+    }
+
+    pub(crate) fn create(&mut self, create: bool) -> &mut Self {
+        self.create = create;
+        self
+    }
+
+    pub(crate) fn create_new(&mut self, create_new: bool) -> &mut Self {
+        self.create_new = create_new;
+        self
+    }
+
+    pub(crate) fn read(&mut self, read: bool) -> &mut Self {
+        self.read = read;
+        self
+    }
+
+    pub(crate) fn write(&mut self, write: bool) -> &mut Self {
+        self.write = write;
+        self
+    }
+
+    pub(crate) fn truncate(&mut self, truncate: bool) -> &mut Self {
+        self.truncate = truncate;
+        self
+    }
+
+    pub(crate) fn mode(&mut self, mode: u32) -> &mut Self {
+        self.mode = mode as libc::mode_t;
+        self
+    }
+
+    pub(crate) fn custom_flags(&mut self, flags: i32) -> &mut Self {
+        self.custom_flags = flags;
+        self
+    }
+
+    pub(crate) fn access_mode(&self) -> io::Result<libc::c_int> {
+        match (self.read, self.write, self.append) {
+            (true, false, false) => Ok(libc::O_RDONLY),
+            (false, true, false) => Ok(libc::O_WRONLY),
+            (true, true, false) => Ok(libc::O_RDWR),
+            (false, _, true) => Ok(libc::O_WRONLY | libc::O_APPEND),
+            (true, _, true) => Ok(libc::O_RDWR | libc::O_APPEND),
+            (false, false, false) => Err(io::Error::from_raw_os_error(libc::EINVAL)),
+        }
+    }
+
+    pub(crate) fn creation_mode(&self) -> io::Result<libc::c_int> {
+        match (self.write, self.append) {
+            (true, false) => {}
+            (false, false) => {
+                if self.truncate || self.create || self.create_new {
+                    return Err(io::Error::from_raw_os_error(libc::EINVAL));
+                }
+            }
+            (_, true) => {
+                if self.truncate && !self.create_new {
+                    return Err(io::Error::from_raw_os_error(libc::EINVAL));
+                }
+            }
+        }
+
+        Ok(match (self.create, self.truncate, self.create_new) {
+            (false, false, false) => 0,
+            (true, false, false) => libc::O_CREAT,
+            (false, true, false) => libc::O_TRUNC,
+            (true, true, false) => libc::O_CREAT | libc::O_TRUNC,
+            (_, _, true) => libc::O_CREAT | libc::O_EXCL,
+        })
+    }
+}

tokio/src/io/mod.rs

@@ -293,3 +293,7 @@ cfg_io_blocking! {
         pub(crate) use crate::blocking::JoinHandle as Blocking;
     }
 }
+
+cfg_tokio_unstable_uring! {
+    pub(crate) mod uring;
+}

tokio/src/io/uring/mod.rs

@@ -0,0 +1,2 @@
+pub(crate) mod open;
+pub(crate) mod utils;

tokio/src/io/uring/open.rs

@@ -0,0 +1,44 @@
+use super::utils::cstr;
+use crate::{
+    fs::OpenOptions,
+    runtime::driver::op::{Completable, CqeResult, Op},
+};
+use io_uring::{opcode, types};
+use std::{ffi::CString, io, os::fd::FromRawFd, path::Path};
+
+pub(crate) struct Open {
+    #[allow(dead_code)]
+    path: CString,
+}
+
+impl Completable for Open {
+    type Output = crate::fs::File;
+    fn complete(self, cqe: CqeResult) -> io::Result<Self::Output> {
+        let fd = cqe.result? as i32;
+        let file = unsafe { crate::fs::File::from_raw_fd(fd) };
+        Ok(file)
+    }
+}
+
+impl Op<Open> {
+    /// Submit a request to open a file.
+    pub(crate) fn open(path: &Path, options: &OpenOptions) -> io::Result<Op<Open>> {
+        let inner_opt = options;
+        let path = cstr(path)?;
+
+        let custom_flags = inner_opt.0.custom_flags;
+        let flags = libc::O_CLOEXEC
+            | options.0.access_mode()?
+            | options.0.creation_mode()?
+            | (custom_flags & !libc::O_ACCMODE);
+
+        let open_op = opcode::OpenAt::new(types::Fd(libc::AT_FDCWD), path.as_ptr())
+            .flags(flags)
+            .mode(inner_opt.0.mode)
+            .build();
+
+        // SAFETY: Parameters are valid for the entire duration of the operation
+        let op = unsafe { Op::new(open_op, Open { path }) };
+        Ok(op)
+    }
+}

tokio/src/io/uring/utils.rs

@@ -0,0 +1,6 @@
+use std::{ffi::CString, io, path::Path};
+
+pub(crate) fn cstr(p: &Path) -> io::Result<CString> {
+    use std::os::unix::ffi::OsStrExt;
+    Ok(CString::new(p.as_os_str().as_bytes())?)
+}

tokio/src/macros/cfg.rs

@@ -727,3 +727,17 @@ macro_rules! cfg_tokio_unstable_uring {
         )*
     };
 }
+
+macro_rules! cfg_not_tokio_unstable_uring {
+    ($($item:item)*) => {
+        $(
+            #[cfg(not(all(
+                tokio_unstable_uring,
+                feature = "rt",
+                feature = "fs",
+                target_os = "linux",
+            )))]
+            $item
+        )*
+    };
+}

tokio/src/runtime/builder.rs

@@ -338,6 +338,15 @@ impl Builder {
             all(unix, feature = "signal")
         ))]
         self.enable_io();
+
+        #[cfg(all(
+            tokio_unstable_uring,
+            feature = "rt",
+            feature = "fs",
+            target_os = "linux",
+        ))]
+        self.enable_uring();
+
         #[cfg(feature = "time")]
         self.enable_time();
 
@@ -1575,6 +1584,30 @@ cfg_time! {
     }
 }
 
+cfg_tokio_unstable_uring! {
+    impl Builder {
+        /// Enables the `tokio-uring` driver.
+        ///
+        /// Doing this enables using io_uring operations on the runtime.
+        ///
+        /// # Examples
+        ///
+        /// ```
+        /// use tokio::runtime;
+        ///
+        /// let rt = runtime::Builder::new_multi_thread()
+        ///     .enable_uring()
+        ///     .build()
+        ///     .unwrap();
+        /// ```
+        pub fn enable_uring(&mut self) -> &mut Self {
+            // Currently, the uring flag is represented as `enable_io`.
+            self.enable_io = true;
+            self
+        }
+    }
+}
+
 cfg_test_util! {
     impl Builder {
         /// Controls if the runtime's clock starts paused or advancing.

tokio/src/runtime/mod.rs

@@ -323,7 +323,7 @@ pub(crate) mod context;
 
 pub(crate) mod park;
 
-mod driver;
+pub(crate) mod driver;
 
 pub(crate) mod scheduler;