fs: implement rename using io-uring (#7800)

---------

Co-authored-by: Mattia Pitossi <[email protected]>
Co-authored-by: vrtgs <[email protected]>
Co-authored-by: Daksh <[email protected]>

tokio/src/fs/rename.rs

@@ -10,8 +10,37 @@ use std::path::Path;
 ///
 /// This is an async version of [`std::fs::rename`].
 pub async fn rename(from: impl AsRef<Path>, to: impl AsRef<Path>) -> io::Result<()> {
-    let from = from.as_ref().to_owned();
-    let to = to.as_ref().to_owned();
+    let from = from.as_ref();
+    let to = to.as_ref();
 
+    #[cfg(all(
+        tokio_unstable,
+        feature = "io-uring",
+        feature = "rt",
+        feature = "fs",
+        target_os = "linux",
+    ))]
+    {
+        use crate::io::uring::rename::Rename;
+        use crate::runtime::driver::op::Op;
+
+        let handle = crate::runtime::Handle::current();
+        let driver_handle = handle.inner.driver().io();
+
+        type RenameOp = Op<Rename>;
+
+        if driver_handle
+            .check_and_init(io_uring::opcode::RenameAt::CODE)
+            .await?
+        {
+            return RenameOp::rename(from, to)?.await;
+        }
+    }
+
+    rename_blocking(from, to).await
+}
+
+async fn rename_blocking(from: &Path, to: &Path) -> io::Result<()> {
+    let [from, to] = [from, to].map(Path::to_owned);
     asyncify(move || std::fs::rename(from, to)).await
 }

tokio/src/io/uring/mod.rs

@@ -1,5 +1,6 @@
 pub(crate) mod open;
 pub(crate) mod read;
+pub(crate) mod rename;
 pub(crate) mod statx;
 pub(crate) mod utils;
 pub(crate) mod write;

tokio/src/io/uring/rename.rs

@@ -0,0 +1,60 @@
+use super::utils::cstr;
+
+use crate::runtime::driver::op::{CancelData, Cancellable, Completable, CqeResult, Op};
+
+use io_uring::{opcode, types};
+use std::ffi::CString;
+use std::io;
+use std::path::Path;
+
+#[derive(Debug)]
+pub(crate) struct Rename {
+    /// This field will be read by the kernel during the operation, so we
+    /// need to ensure it is valid for the entire duration of the operation.
+    _from: CString,
+    _to: CString,
+}
+
+impl Completable for Rename {
+    type Output = io::Result<()>;
+
+    fn complete(self, cqe: CqeResult) -> Self::Output {
+        cqe.result.map(drop)
+    }
+
+    fn complete_with_error(self, error: io::Error) -> Self::Output {
+        Err(error)
+    }
+}
+
+impl Cancellable for Rename {
+    fn cancel(self) -> CancelData {
+        CancelData::Rename(self)
+    }
+}
+
+impl Op<Rename> {
+    pub(crate) fn rename(from: &Path, to: &Path) -> io::Result<Op<Rename>> {
+        let from = cstr(from)?;
+        let to = cstr(to)?;
+
+        let rename_op = opcode::RenameAt::new(
+            types::Fd(libc::AT_FDCWD),
+            from.as_ptr(),
+            types::Fd(libc::AT_FDCWD),
+            to.as_ptr(),
+        )
+        .build();
+
+        // SAFETY: Parameters are valid for the entire duration of the operation
+        Ok(unsafe {
+            Op::new(
+                rename_op,
+                Rename {
+                    _from: from,
+                    _to: to,
+                },
+            )
+        })
+    }
+}

tokio/src/runtime/driver/op.rs

@@ -1,6 +1,7 @@
 use crate::io::blocking::Buf;
 use crate::io::uring::open::Open;
 use crate::io::uring::read::Read;
+use crate::io::uring::rename::Rename;
 use crate::io::uring::utils::ArcFd;
 use crate::io::uring::write::Write;
 
@@ -22,7 +23,7 @@ use crate::io::uring::statx::Statx;
 use io_uring::cqueue;
 use io_uring::squeue::Entry;
 use std::future::Future;
-use std::io::{self, Error};
+use std::io;
 use std::mem;
 use std::os::fd::OwnedFd;
 use std::pin::Pin;
@@ -35,6 +36,7 @@ use std::task::{Context, Poll, Waker};
 pub(crate) enum CancelData {
     Open(Open),
     Write(Write),
+    Rename(Rename),
     ReadVec(Read<Vec<u8>, OwnedFd>),
     ReadBuf(Read<Buf, ArcFd>),
     #[cfg(
@@ -69,7 +71,7 @@ pub(crate) enum Lifecycle {
     ),
 
     /// The operation has completed with a single cqe result
-    Completed(io_uring::cqueue::Entry),
+    Completed(cqueue::Entry),
 }
 
 pub(crate) enum State {
@@ -149,7 +151,7 @@ pub(crate) trait Completable {
     //
     // The `Op` type that implements this trait can return the passed error
     // upstream by embedding it in the `Output`.
-    fn complete_with_error(self, error: Error) -> Self::Output;
+    fn complete_with_error(self, error: io::Error) -> Self::Output;
 }
 
 /// Extracts the `CancelData` needed to safely cancel an in-flight io_uring operation.

tokio/tests/fs_uring_rename.rs

@@ -0,0 +1,259 @@
+//! Uring file operations tests.
+
+#![cfg(all(
+    tokio_unstable,
+    feature = "io-uring",
+    feature = "rt",
+    feature = "fs",
+    target_os = "linux"
+))]
+
+use futures::future::Future;
+use futures::future::FutureExt;
+use libc::PATH_MAX;
+use std::future::poll_fn;
+use std::io::Write;
+use std::path::PathBuf;
+use std::sync::mpsc;
+use std::task::Poll;
+use std::time::Duration;
+use tempfile::{tempdir, NamedTempFile};
+use tokio::fs::{rename, try_exists};
+use tokio::runtime::{Builder, Runtime};
+use tokio_test::assert_pending;
+use tokio_util::task::TaskTracker;
+
+use crate::support::io_uring::io_uring_supported;
+
+mod support {
+    pub(crate) mod io_uring;
+}
+
+fn multi_rt(n: usize) -> Box<dyn Fn() -> Runtime> {
+    Box::new(move || {
+        Builder::new_multi_thread()
+            .worker_threads(n)
+            .enable_all()
+            .build()
+            .unwrap()
+    })
+}
+
+fn current_rt() -> Box<dyn Fn() -> Runtime> {
+    Box::new(|| Builder::new_current_thread().enable_all().build().unwrap())
+}
+
+fn rt_combinations() -> Vec<Box<dyn Fn() -> Runtime>> {
+    vec![
+        current_rt(),
+        multi_rt(1),
+        multi_rt(2),
+        multi_rt(8),
+        multi_rt(64),
+        multi_rt(256),
+    ]
+}
+
+#[test]
+fn shutdown_runtime_while_performing_io_uring_ops() {
+    if !io_uring_supported() {
+        return;
+    }
+
+    fn run(rt: Runtime) {
+        let (done_tx, done_rx) = mpsc::channel();
+        let (_tmp, path) = create_tmp_files(1);
+        // keep 100 permits
+        const N: i32 = 100;
+        rt.spawn(async move {
+            let path = path[0].clone();
+            let dst = path.with_extension("renamed");
+
+            // spawning a bunch of uring operations.
+            let mut futs = vec![];
+
+            // spawning a bunch of uring operations.
+            for _ in 0..N {
+                let mut fut = Box::pin(rename(path.clone(), dst.clone()));
+
+                poll_fn(|cx| {
+                    assert_pending!(fut.as_mut().poll(cx));
+                    Poll::<()>::Pending
+                })
+                .await;
+
+                futs.push(fut);
+            }
+
+            tokio::task::yield_now().await;
+        });
+
+        std::thread::spawn(move || {
+            rt.shutdown_timeout(Duration::from_millis(300));
+            done_tx.send(()).unwrap();
+        });
+
+        done_rx.recv().unwrap();
+    }
+
+    for rt in rt_combinations() {
+        run(rt());
+    }
+}
+
+#[test]
+fn rename_many_files() {
+    fn run(rt: Runtime) {
+        const NUM_FILES: usize = 512;
+
+        let dir = tempdir().unwrap();
+
+        rt.block_on(async move {
+            let tracker = TaskTracker::new();
+
+            for i in 0..NUM_FILES {
+                let src = dir.path().join(format!("src-{i}"));
+                let dst = dir.path().join(format!("dst-{i}"));
+                tokio::fs::write(&src, b"contents").await.unwrap();
+
+                tracker.spawn(async move {
+                    rename(&src, &dst).await.unwrap();
+
+                    assert!(!try_exists(&src).await.unwrap());
+                    assert!(try_exists(&dst).await.unwrap());
+                });
+            }
+            tracker.close();
+            tracker.wait().await;
+        });
+    }
+
+    for rt in rt_combinations() {
+        run(rt());
+    }
+}
+
+#[tokio::test]
+async fn rename_file() {
+    let dir = tempdir().unwrap();
+    let src = dir.path().join("src.txt");
+    let dst = dir.path().join("dst.txt");
+
+    tokio::fs::write(&src, b"Hello File!").await.unwrap();
+
+    rename(&src, &dst).await.unwrap();
+
+    assert!(!try_exists(&src).await.unwrap());
+    assert!(try_exists(&dst).await.unwrap());
+    assert_eq!(tokio::fs::read(&dst).await.unwrap(), b"Hello File!");
+}
+
+#[tokio::test]
+async fn rename_replaces_destination() {
+    let dir = tempdir().unwrap();
+    let src = dir.path().join("src.txt");
+    let dst = dir.path().join("dst.txt");
+
+    tokio::fs::write(&src, b"source").await.unwrap();
+    tokio::fs::write(&dst, b"destination").await.unwrap();
+
+    rename(&src, &dst).await.unwrap();
+
+    assert!(!try_exists(&src).await.unwrap());
+    assert_eq!(tokio::fs::read(&dst).await.unwrap(), b"source");
+}
+
+#[tokio::test]
+async fn rename_directory() {
+    let dir = tempdir().unwrap();
+    let src = dir.path().join("src_dir");
+    let dst = dir.path().join("dst_dir");
+
+    tokio::fs::create_dir(&src).await.unwrap();
+    tokio::fs::write(src.join("file.txt"), b"contents")
+        .await
+        .unwrap();
+
+    rename(&src, &dst).await.unwrap();
+
+    assert!(!try_exists(&src).await.unwrap());
+    assert!(try_exists(&dst).await.unwrap());
+    assert_eq!(
+        tokio::fs::read(dst.join("file.txt")).await.unwrap(),
+        b"contents"
+    );
+}
+
+#[tokio::test]
+async fn rename_nonexistent_source() {
+    let dir = tempdir().unwrap();
+    let src = dir.path().join("nonexistent");
+    let dst = dir.path().join("dst");
+
+    let result = rename(&src, &dst).await;
+    assert_eq!(result.err().unwrap().raw_os_error().unwrap(), libc::ENOENT);
+}
+
+#[tokio::test]
+async fn rename_path_name_too_long() {
+    let dir = tempdir().unwrap();
+    let src = dir.path().join("src.txt");
+    tokio::fs::write(&src, b"Hello File!").await.unwrap();
+
+    // if the destination name is above PATH_MAX (Linux is 4096 bytes), we should
+    // receive a `std::io::ErrorKind::InvalidFilename` error.
+    let long_dst = dir.path().join(vec!["a"; (PATH_MAX + 1) as usize].join(""));
+    let result = rename(&src, &long_dst).await;
+    assert_eq!(
+        result.err().unwrap().kind(),
+        std::io::ErrorKind::InvalidFilename
+    );
+}
+
+#[tokio::test]
+async fn cancel_op_future() {
+    if !io_uring_supported() {
+        return;
+    }
+
+    let (_tmp_file, path): (Vec<NamedTempFile>, Vec<PathBuf>) = create_tmp_files(1);
+    let path = path[0].clone();
+    let dst = path.with_extension("renamed");
+
+    let (tx, mut rx) = tokio::sync::mpsc::unbounded_channel();
+
+    let handle = tokio::spawn(async move {
+        poll_fn(|cx| {
+            let fut = rename(path.clone(), dst.clone());
+
+            // the first poll should return Pending.
+            assert_pending!(Box::pin(fut).poll_unpin(cx));
+            tx.send(true).unwrap();
+
+            Poll::<()>::Pending
+        })
+        .await;
+    });
+
+    // Wait for the first poll
+    let val = rx.recv().await;
+    assert!(val.unwrap());
+
+    handle.abort();
+
+    let res = handle.await.unwrap_err();
+    assert!(res.is_cancelled());
+}
+
+fn create_tmp_files(num_files: usize) -> (Vec<NamedTempFile>, Vec<PathBuf>) {
+    let mut files = Vec::with_capacity(num_files);
+    for _ in 0..num_files {
+        let mut tmp = NamedTempFile::new().unwrap();
+        let buf = vec![20; 1023];
+        tmp.write_all(&buf).unwrap();
+        let path = tmp.path().to_path_buf();
+        files.push((tmp, path));
+    }
+
+    files.into_iter().unzip()
+}