fs: support io-uring in AsyncRead for File (#7907)

spellcheck.dic

@@ -1,4 +1,4 @@
-317
+318
 &
 +
 <
@@ -244,6 +244,7 @@ spawner
 Splitter
 spmc
 spsc
+SQE
 src
 stabilised
 startup

tokio/src/fs/file.rs

@@ -30,6 +30,11 @@ use crate::blocking::{spawn_blocking, spawn_mandatory_blocking};
 #[cfg(not(test))]
 use std::fs::File as StdFile;
 
+cfg_io_uring! {
+    #[cfg(not(test))]
+    use crate::spawn;
+}
+
 /// A reference to an open file on the filesystem.
 ///
 /// This is a specialized version of [`std::fs::File`] for usage from the
@@ -613,13 +618,7 @@ impl AsyncRead for File {
                     let std = me.std.clone();
 
                     let max_buf_size = cmp::min(dst.remaining(), me.max_buf_size);
-                    inner.state = State::Busy(spawn_blocking(move || {
-                        // SAFETY: the `Read` implementation of `std` does not
-                        // read from the buffer it is borrowing and correctly
-                        // reports the length of the data written into the buffer.
-                        let res = unsafe { buf.read_from(&mut &*std, max_buf_size) };
-                        (Operation::Read(res), buf)
-                    }));
+                    inner.state = State::Busy(Inner::poll_read_inner(std, buf, max_buf_size)?);
                 }
                 State::Busy(ref mut rx) => {
                     let (op, mut buf) = ready!(Pin::new(rx).poll(cx))?;
@@ -952,6 +951,125 @@ cfg_windows! {
 }
 
 impl Inner {
+    fn poll_read_inner(
+        std: Arc<StdFile>,
+        buf: Buf,
+        max_buf_size: usize,
+    ) -> io::Result<JoinHandle<(Operation, Buf)>> {
+        // Unit tests use `MockFile` and the mock `spawn_blocking` infrastructure,
+        // which can't drive real io_uring operations. The io_uring read path
+        // is tested through integration tests in `tests/fs_uring_file_read.rs`.
+        #[cfg(all(
+            not(test),
+            tokio_unstable,
+            feature = "io-uring",
+            feature = "rt",
+            feature = "fs",
+            target_os = "linux",
+        ))]
+        {
+            if let Ok(handle) = crate::runtime::Handle::try_current() {
+                let driver_handle = handle.inner.driver().io();
+
+                if driver_handle.is_uring_ready(io_uring::opcode::Read::CODE) {
+                    // Fast path: uring already initialized and Read supported.
+                    let fd: crate::io::uring::utils::ArcFd = std;
+                    return Ok(spawn(Self::uring_read(fd, buf, max_buf_size)));
+                }
+
+                if !driver_handle.is_uring_probed() {
+                    // Not yet probed: lazy init inside an async task so
+                    // `File::from_std()` can still benefit from io-uring.
+                    return Ok(spawn(Self::lazy_init_read(std, buf, max_buf_size)));
+                }
+                // Probed but unsupported: fall through to spawn_blocking.
+            }
+        }
+
+        // Fallback: spawn_blocking
+        let join = Self::spawn_blocking_read(buf, std, max_buf_size);
+        Ok(join)
+    }
+
+    /// Perform an io-uring read with interrupt retry.
+    #[cfg(all(
+        not(test),
+        tokio_unstable,
+        feature = "io-uring",
+        feature = "rt",
+        feature = "fs",
+        target_os = "linux",
+    ))]
+    async fn uring_read(
+        mut fd: crate::io::uring::utils::ArcFd,
+        mut buf: Buf,
+        max_buf_size: usize,
+    ) -> (Operation, Buf) {
+        use crate::runtime::driver::op::Op;
+
+        loop {
+            let (res, r_fd, r_buf) =
+                // u64::MAX to use and advance the file position
+                Op::read_at(fd, buf, max_buf_size, u64::MAX).await;
+            match res {
+                Err(e) if e.kind() == io::ErrorKind::Interrupted => {
+                    buf = r_buf;
+                    fd = r_fd;
+                    continue;
+                }
+                Err(e) => break (Operation::Read(Err(e)), r_buf),
+                Ok(n) => break (Operation::Read(Ok(n as usize)), r_buf),
+            }
+        }
+    }
+
+    /// Attempt lazy io-uring initialization, then read via uring or fall back
+    /// to a blocking read. Covers the `File::from_std()` path where
+    /// `check_and_init()` hasn't been called yet.
+    #[cfg(all(
+        not(test),
+        tokio_unstable,
+        feature = "io-uring",
+        feature = "rt",
+        feature = "fs",
+        target_os = "linux",
+    ))]
+    async fn lazy_init_read(std: Arc<StdFile>, buf: Buf, max_buf_size: usize) -> (Operation, Buf) {
+        let handle = crate::runtime::Handle::current();
+        let driver_handle = handle.inner.driver().io();
+        if driver_handle
+            .check_and_init(io_uring::opcode::Read::CODE)
+            .await
+            .unwrap_or(false)
+        {
+            let fd: crate::io::uring::utils::ArcFd = std;
+            Self::uring_read(fd, buf, max_buf_size).await
+        } else {
+            match Self::spawn_blocking_read(buf, std, max_buf_size).await {
+                Ok(result) => result,
+                Err(e) => (
+                    Operation::Read(Err(io::Error::new(io::ErrorKind::Other, e))),
+                    Buf::with_capacity(0),
+                ),
+            }
+        }
+    }
+
+    fn spawn_blocking_read(
+        buf: Buf,
+        std: Arc<StdFile>,
+        max_buf_size: usize,
+    ) -> JoinHandle<(Operation, Buf)> {
+        spawn_blocking(move || {
+            let mut buf = buf;
+            // SAFETY: the `Read` implementation of `std` does not
+            // read from the buffer it is borrowing and correctly
+            // reports the length of the data written into the buffer.
+            let res = unsafe { buf.read_from(&mut &*std, max_buf_size) };
+            (Operation::Read(res), buf)
+        })
+    }
+
     async fn complete_inflight(&mut self) {
         use std::future::poll_fn;
 

tokio/src/fs/read_uring.rs

@@ -116,7 +116,7 @@ async fn op_read(
     read_len: u32,
 ) -> io::Result<(OwnedFd, Vec<u8>, bool)> {
     loop {
-        let (res, r_fd, r_buf) = Op::read(fd, buf, read_len, *offset).await;
+        let (res, r_fd, r_buf) = Op::read_at(fd, buf, read_len as usize, *offset).await;
 
         match res {
             Err(e) if e.kind() == ErrorKind::Interrupted => {

tokio/src/io/blocking.rs

@@ -277,6 +277,39 @@ impl Buf {
     }
 }
 
+cfg_io_uring! {
+    impl Buf {
+        /// Prepare the internal buffer for an io-uring read operation.
+        ///
+        /// Returns a pointer to the spare capacity and the length available
+        /// for the kernel to write into.
+        pub(crate) fn prepare_uring_read(&mut self, max_buf_size: usize) -> (*mut u8, u32) {
+            assert!(self.is_empty());
+            self.buf.reserve(max_buf_size);
+            let spare = self.buf.spare_capacity_mut();
+            let len = std::cmp::min(spare.len(), max_buf_size);
+            let ptr = spare.as_mut_ptr().cast::<u8>();
+            (ptr, len as u32)
+        }
+
+        /// Complete an io-uring read operation.
+        ///
+        /// # Safety
+        ///
+        /// The caller must ensure that the kernel wrote exactly `n` bytes
+        /// into the buffer that was returned by `prepare_uring_read`.
+        pub(crate) unsafe fn complete_uring_read(&mut self, n: usize) {
+            assert_eq!(self.pos, 0);
+            // SAFETY: `prepare_uring_read` handed out a pointer to
+            // `self.buf.spare_capacity_mut()` after asserting it's empty.
+            // The caller guarantees the kernel initialised exactly `n` bytes
+            // starting at that pointer, so bytes `0..n` are now initialised and
+            // it is sound to set the Vec length to `n`.
+            unsafe { self.buf.set_len(n) };
+        }
+    }
+}
+
 cfg_fs! {
     impl Buf {
         pub(crate) fn discard_read(&mut self) -> i64 {

tokio/src/io/uring/read.rs

@@ -1,27 +1,74 @@
+use crate::io::blocking::Buf;
+use crate::io::uring::utils::{ArcFd, UringFd};
 use crate::runtime::driver::op::{CancelData, Cancellable, Completable, CqeResult, Op};
 
 use io_uring::{opcode, types};
+use std::fmt;
 use std::io::{self, Error};
-use std::os::fd::{AsRawFd, OwnedFd};
+use std::os::fd::OwnedFd;
 
-#[derive(Debug)]
-pub(crate) struct Read {
-    fd: OwnedFd,
-    buf: Vec<u8>,
+/// Trait for buffers that can be used with io-uring read operations.
+pub(crate) trait ReadBuffer: Send + 'static {
+    /// Prepare the buffer for a read operation.
+    /// Returns a pointer and length for the io-uring SQE.
+    fn uring_read_prepare(&mut self, max_len: usize) -> (*mut u8, u32);
+
+    /// Complete a read of `n` bytes.
+    ///
+    /// # Safety
+    ///
+    /// The caller must ensure the kernel wrote exactly `n` bytes
+    /// into the buffer at the pointer returned by `uring_read_prepare`.
+    unsafe fn uring_read_complete(&mut self, n: u32);
 }
 
-impl Completable for Read {
-    type Output = (io::Result<u32>, OwnedFd, Vec<u8>);
+impl ReadBuffer for Vec<u8> {
+    fn uring_read_prepare(&mut self, max_len: usize) -> (*mut u8, u32) {
+        assert!(self.spare_capacity_mut().len() >= max_len);
+        let ptr = self.spare_capacity_mut().as_mut_ptr().cast();
+        (ptr, max_len as u32)
+    }
+
+    unsafe fn uring_read_complete(&mut self, n: u32) {
+        // SAFETY: the kernel wrote `n` bytes into spare capacity starting
+        // at the old self.len(), so self.len() + n bytes are now initialized.
+        unsafe { self.set_len(self.len() + n as usize) };
+    }
+}
+
+impl ReadBuffer for Buf {
+    fn uring_read_prepare(&mut self, max_len: usize) -> (*mut u8, u32) {
+        self.prepare_uring_read(max_len)
+    }
+
+    unsafe fn uring_read_complete(&mut self, n: u32) {
+        // SAFETY: caller guarantees kernel wrote exactly n bytes.
+        unsafe { self.complete_uring_read(n as usize) };
+    }
+}
+
+pub(crate) struct Read<B, F = ArcFd> {
+    fd: F,
+    buf: B,
+}
+
+impl<B: fmt::Debug, F> fmt::Debug for Read<B, F> {
+    fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result {
+        f.debug_struct("Read")
+            .field("buf", &self.buf)
+            .finish_non_exhaustive()
+    }
+}
+
+impl<B: ReadBuffer, F: UringFd> Completable for Read<B, F> {
+    type Output = (io::Result<u32>, F, B);
 
     fn complete(self, cqe: CqeResult) -> Self::Output {
         let mut buf = self.buf;
-
         if let Ok(len) = cqe.result {
-            let new_len = buf.len() + len as usize;
-            // SAFETY: Kernel read len bytes
-            unsafe { buf.set_len(new_len) };
+            // SAFETY: kernel wrote exactly `len` bytes into the prepared buffer.
+            unsafe { buf.uring_read_complete(len) };
         }
-
         (cqe.result, self.fd, buf)
     }
 
@@ -30,32 +77,38 @@ impl Completable for Read {
     }
 }
 
-impl Cancellable for Read {
+impl Cancellable for Read<Vec<u8>, OwnedFd> {
     fn cancel(self) -> CancelData {
-        CancelData::Read(self)
+        CancelData::ReadVec(self)
     }
 }
 
-impl Op<Read> {
-    // Submit a request to read a FD at given length and offset into a
-    // dynamic buffer with uninitialized memory. The read happens on uninitialized
-    // buffer and no overwriting happens.
+impl Cancellable for Read<Buf, ArcFd> {
+    fn cancel(self) -> CancelData {
+        CancelData::ReadBuf(self)
+    }
+}
 
-    // SAFETY: The `len` of the amount to be read and the buffer that is passed
-    // should have capacity > len.
-    //
-    // If `len` read is higher than vector capacity then setting its length by
-    // the caller in terms of size_read can be unsound.
-    pub(crate) fn read(fd: OwnedFd, mut buf: Vec<u8>, len: u32, offset: u64) -> Self {
-        // don't overwrite on already written part
-        assert!(buf.spare_capacity_mut().len() >= len as usize);
-        let buf_mut_ptr = buf.spare_capacity_mut().as_mut_ptr().cast();
+impl<B, F> Op<Read<B, F>>
+where
+    B: ReadBuffer + fmt::Debug,
+    F: UringFd,
+    Read<B, F>: Cancellable,
+{
+    /// Submit a read operation via io-uring.
+    ///
+    /// `max_len` is the maximum number of bytes to read.
+    /// `offset` is the file offset; use `u64::MAX` for the current cursor.
+    pub(crate) fn read_at(fd: F, mut buf: B, max_len: usize, offset: u64) -> Self {
+        let (ptr, len) = buf.uring_read_prepare(max_len);
 
-        let read_op = opcode::Read::new(types::Fd(fd.as_raw_fd()), buf_mut_ptr, len)
+        let sqe = opcode::Read::new(types::Fd(UringFd::as_raw_fd(&fd)), ptr, len)
             .offset(offset)
             .build();
 
-        // SAFETY: Parameters are valid for the entire duration of the operation
-        unsafe { Op::new(read_op, Read { fd, buf }) }
+        // SAFETY: `fd` and `buf`, which owns the heap buffer, are moved into `Read`,
+        // which is held by the `Op` for the entire duration of the io-uring operation.
+        // The buffer pointer remains valid because Vec heap data doesn't move.
+        unsafe { Op::new(sqe, Read { fd, buf }) }
     }
 }

tokio/src/io/uring/utils.rs

@@ -1,6 +1,31 @@
+use std::os::fd::{AsRawFd, OwnedFd, RawFd};
 use std::os::unix::ffi::OsStrExt;
+use std::sync::Arc;
 use std::{ffi::CString, io, path::Path};
 
+pub(crate) type ArcFd = Arc<dyn AsRawFd + Send + Sync + 'static>;
+
+/// Raw file descriptor trait for io-uring operations.
+///
+/// `Arc<dyn AsRawFd>` does not satisfy `AsRawFd` because the blanket impl
+/// for `Arc<T>` requires `T: Sized`. This trait bridges that gap so both
+/// `OwnedFd` and `ArcFd` can be used generically with `Op::read_at`.
+pub(crate) trait UringFd: Send + Sync + 'static {
+    fn as_raw_fd(&self) -> RawFd;
+}
+
+impl UringFd for OwnedFd {
+    fn as_raw_fd(&self) -> RawFd {
+        AsRawFd::as_raw_fd(self)
+    }
+}
+
+impl UringFd for ArcFd {
+    fn as_raw_fd(&self) -> RawFd {
+        (**self).as_raw_fd()
+    }
+}
+
 pub(crate) fn cstr(p: &Path) -> io::Result<CString> {
     Ok(CString::new(p.as_os_str().as_bytes())?)
 }

tokio/src/runtime/driver/op.rs

@@ -1,6 +1,9 @@
+use crate::io::blocking::Buf;
 use crate::io::uring::open::Open;
 use crate::io::uring::read::Read;
+use crate::io::uring::utils::ArcFd;
 use crate::io::uring::write::Write;
+
 use crate::runtime::Handle;
 
 use io_uring::cqueue;
@@ -8,6 +11,7 @@ use io_uring::squeue::Entry;
 use std::future::Future;
 use std::io::{self, Error};
 use std::mem;
+use std::os::fd::OwnedFd;
 use std::pin::Pin;
 use std::task::{Context, Poll, Waker};
 
@@ -18,7 +22,8 @@ use std::task::{Context, Poll, Waker};
 pub(crate) enum CancelData {
     Open(Open),
     Write(Write),
-    Read(Read),
+    ReadVec(Read<Vec<u8>, OwnedFd>),
+    ReadBuf(Read<Buf, ArcFd>),
 }
 
 #[derive(Debug)]

tokio/src/runtime/io/driver/uring.rs

@@ -183,6 +183,26 @@ impl Handle {
         &self.uring_context
     }
 
+    /// Returns `true` if io_uring has already been initialized and the given
+    /// opcode is supported. Returns `false` if io_uring hasn't been
+    /// initialized yet or is unsupported. Unlike `check_and_init`, this
+    /// doesn't attempt initialization.
+    #[cfg_attr(test, allow(dead_code))]
+    pub(crate) fn is_uring_ready(&self, opcode: u8) -> bool {
+        self.uring_probe
+            .get()
+            .and_then(|opt| opt.as_ref())
+            .is_some_and(|probe| probe.is_supported(opcode))
+    }
+
+    /// Returns `true` if the io_uring probe has already been attempted
+    /// (regardless of whether io_uring is supported). Returns `false` if
+    /// no probe has been attempted yet.
+    #[cfg_attr(test, allow(dead_code))]
+    pub(crate) fn is_uring_probed(&self) -> bool {
+        self.uring_probe.get().is_some()
+    }
+
     /// Check if the io_uring context is initialized. If not, it will try to initialize it.
     /// Then, check if the provided opcode is supported.
     ///

tokio/tests/fs_uring_file_read.rs

@@ -0,0 +1,471 @@
+//! Tests for AsyncRead on tokio::fs::File via io-uring.
+
+#![cfg(all(
+    tokio_unstable,
+    feature = "io-uring",
+    feature = "rt",
+    feature = "fs",
+    target_os = "linux"
+))]
+
+use std::io::Write;
+use std::path::PathBuf;
+use std::sync::mpsc;
+use std::time::Duration;
+use tempfile::NamedTempFile;
+use tokio::fs::File;
+use tokio::io::{AsyncReadExt, AsyncSeekExt, AsyncWriteExt};
+use tokio::runtime::{Builder, Runtime};
+use tokio_util::task::TaskTracker;
+
+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(4), multi_rt(64)]
+}
+
+fn create_temp_file(data: &[u8]) -> (NamedTempFile, PathBuf) {
+    let mut tmp = NamedTempFile::new().unwrap();
+    tmp.write_all(data).unwrap();
+    tmp.flush().unwrap();
+    let path = tmp.path().to_path_buf();
+    (tmp, path)
+}
+
+#[tokio::test]
+async fn test_file_read() {
+    let data = b"hello io-uring";
+    let (_tmp, path) = create_temp_file(data);
+
+    let mut file = File::open(&path).await.unwrap();
+    let mut buf = vec![0u8; data.len()];
+    let n = file.read(&mut buf).await.unwrap();
+    assert_eq!(n, data.len());
+    assert_eq!(&buf[..n], data);
+}
+
+#[tokio::test]
+async fn test_file_read_exact() {
+    let data = b"exact read test data";
+    let (_tmp, path) = create_temp_file(data);
+
+    let mut file = File::open(&path).await.unwrap();
+    let mut buf = vec![0u8; data.len()];
+    file.read_exact(&mut buf).await.unwrap();
+    assert_eq!(&buf, data);
+}
+
+#[tokio::test]
+async fn test_file_read_to_end() {
+    // Empty file
+    {
+        let (_tmp, path) = create_temp_file(b"");
+        let mut file = File::open(&path).await.unwrap();
+        let mut buf = Vec::new();
+        let n = file.read_to_end(&mut buf).await.unwrap();
+        assert_eq!(n, 0);
+        assert!(buf.is_empty());
+    }
+
+    // Small file
+    {
+        let data: Vec<u8> = (0..100u8).collect();
+        let (_tmp, path) = create_temp_file(&data);
+        let mut file = File::open(&path).await.unwrap();
+        let mut buf = Vec::new();
+        file.read_to_end(&mut buf).await.unwrap();
+        assert_eq!(buf, data);
+    }
+
+    // File larger than typical buffer
+    {
+        let data: Vec<u8> = (0..100_000u32).map(|i| (i % 256) as u8).collect();
+        let (_tmp, path) = create_temp_file(&data);
+        let mut file = File::open(&path).await.unwrap();
+        let mut buf = Vec::new();
+        file.read_to_end(&mut buf).await.unwrap();
+        assert_eq!(buf, data);
+    }
+}
+
+#[tokio::test]
+async fn test_file_read_to_string() {
+    let text = "hello, io-uring world! 🦀";
+    let (_tmp, path) = create_temp_file(text.as_bytes());
+
+    let mut file = File::open(&path).await.unwrap();
+    let mut s = String::new();
+    file.read_to_string(&mut s).await.unwrap();
+    assert_eq!(s, text);
+}
+
+#[tokio::test]
+async fn test_file_read_empty() {
+    let (_tmp, path) = create_temp_file(b"");
+
+    let mut file = File::open(&path).await.unwrap();
+    let mut buf = vec![0u8; 100];
+    let n = file.read(&mut buf).await.unwrap();
+    assert_eq!(n, 0);
+}
+
+#[tokio::test]
+async fn test_file_read_large() {
+    let data: Vec<u8> = (0..3_000_000u32).map(|i| (i % 256) as u8).collect();
+    let (_tmp, path) = create_temp_file(&data);
+
+    let mut file = File::open(&path).await.unwrap();
+    let mut buf = Vec::new();
+    file.read_to_end(&mut buf).await.unwrap();
+    assert_eq!(buf.len(), data.len());
+    assert_eq!(buf, data);
+}
+
+#[tokio::test]
+async fn test_file_read_custom_buf_size() {
+    let data: Vec<u8> = (0..1000u16).map(|i| (i % 256) as u8).collect();
+    let (_tmp, path) = create_temp_file(&data);
+
+    let mut file = File::open(&path).await.unwrap();
+    file.set_max_buf_size(64);
+    let mut buf = Vec::new();
+    file.read_to_end(&mut buf).await.unwrap();
+    assert_eq!(buf, data);
+}
+
+#[tokio::test]
+async fn test_file_read_partial_buf() {
+    let data: Vec<u8> = (0..100u8).collect();
+    let (_tmp, path) = create_temp_file(&data);
+
+    let mut file = File::open(&path).await.unwrap();
+
+    // Read only 5 bytes
+    let mut buf = vec![0u8; 5];
+    file.read_exact(&mut buf).await.unwrap();
+    assert_eq!(&buf, &data[..5]);
+
+    // Read the rest
+    let mut rest = Vec::new();
+    file.read_to_end(&mut rest).await.unwrap();
+    assert_eq!(rest, data[5..]);
+}
+
+#[tokio::test]
+async fn test_file_read_seek() {
+    let data = b"hello world";
+    let (_tmp, path) = create_temp_file(data);
+
+    let mut file = File::open(&path).await.unwrap();
+    file.seek(std::io::SeekFrom::Start(6)).await.unwrap();
+
+    let mut buf = Vec::new();
+    file.read_to_end(&mut buf).await.unwrap();
+    assert_eq!(buf, b"world");
+}
+
+#[tokio::test]
+async fn test_file_read_seek_back() {
+    let data = b"hello world";
+    let (_tmp, path) = create_temp_file(data);
+
+    let mut file = File::open(&path).await.unwrap();
+
+    // Read first 5 bytes
+    let mut buf = vec![0u8; 5];
+    file.read_exact(&mut buf).await.unwrap();
+    assert_eq!(&buf, b"hello");
+
+    // Seek back to start
+    file.seek(std::io::SeekFrom::Start(0)).await.unwrap();
+
+    // Read full content
+    let mut full = Vec::new();
+    file.read_to_end(&mut full).await.unwrap();
+    assert_eq!(full, data);
+}
+
+#[tokio::test]
+async fn test_file_read_after_write() {
+    let tmp = NamedTempFile::new().unwrap();
+    let path = tmp.path().to_path_buf();
+
+    let mut file = tokio::fs::OpenOptions::new()
+        .read(true)
+        .write(true)
+        .open(&path)
+        .await
+        .unwrap();
+
+    file.write_all(b"hello uring").await.unwrap();
+    file.flush().await.unwrap();
+    file.seek(std::io::SeekFrom::Start(0)).await.unwrap();
+
+    let mut buf = Vec::new();
+    file.read_to_end(&mut buf).await.unwrap();
+    assert_eq!(buf, b"hello uring");
+}
+
+#[tokio::test]
+async fn test_file_read_cancel() {
+    let data: Vec<u8> = (0..10_000).map(|i| (i % 256) as u8).collect();
+    let (_tmp, path) = create_temp_file(&data);
+
+    let path2 = path.clone();
+    let handle = tokio::spawn(async move {
+        let mut file = File::open(&path2).await.unwrap();
+        let mut buf = Vec::new();
+        file.read_to_end(&mut buf).await.unwrap();
+        buf
+    });
+
+    handle.abort();
+    let res = handle.await;
+    assert!(res.unwrap_err().is_cancelled());
+
+    // Verify runtime still works
+    let mut file = File::open(&path).await.unwrap();
+    let mut buf = Vec::new();
+    file.read_to_end(&mut buf).await.unwrap();
+    assert_eq!(buf, data);
+}
+
+#[tokio::test]
+async fn test_file_read_concurrent() {
+    const NUM_FILES: usize = 100;
+
+    let files: Vec<_> = (0..NUM_FILES)
+        .map(|i| {
+            let data: Vec<u8> = (0..1024).map(|j| ((i as u16 + j) % 256) as u8).collect();
+            create_temp_file(&data)
+        })
+        .collect();
+
+    let tracker = TaskTracker::new();
+
+    for (i, (_tmp, path)) in files.iter().enumerate() {
+        let path = path.clone();
+        let expected: Vec<u8> = (0..1024).map(|j| ((i as u16 + j) % 256) as u8).collect();
+        tracker.spawn(async move {
+            let mut file = File::open(&path).await.unwrap();
+            let mut buf = Vec::new();
+            file.read_to_end(&mut buf).await.unwrap();
+            assert_eq!(buf, expected);
+        });
+    }
+
+    tracker.close();
+    tracker.wait().await;
+}
+
+#[test]
+fn test_file_read_multi_runtime() {
+    for rt_factory in rt_combinations() {
+        let rt = rt_factory();
+        let data: Vec<u8> = (0..10_000).map(|i| (i % 256) as u8).collect();
+        let (_tmp, path) = create_temp_file(&data);
+
+        let result = rt.block_on(async {
+            let mut file = File::open(&path).await.unwrap();
+            let mut buf = Vec::new();
+            file.read_to_end(&mut buf).await.unwrap();
+            buf
+        });
+
+        assert_eq!(result, data);
+    }
+}
+
+#[test]
+fn shutdown_runtime_with_pending_reads() {
+    for rt_factory in rt_combinations() {
+        let rt = rt_factory();
+        let (done_tx, done_rx) = mpsc::channel();
+
+        let data: Vec<u8> = (0..10_000).map(|i| (i % 256) as u8).collect();
+        let (_tmp, path) = create_temp_file(&data);
+
+        for _ in 0..50 {
+            let path = path.clone();
+            rt.spawn(async move {
+                let mut file = File::open(&path).await.unwrap();
+                let mut buf = Vec::new();
+                let _ = file.read_to_end(&mut buf).await;
+            });
+        }
+
+        std::thread::spawn(move || {
+            rt.shutdown_timeout(Duration::from_millis(500));
+            done_tx.send(()).unwrap();
+        });
+
+        done_rx.recv().unwrap();
+    }
+}
+
+#[test]
+fn test_file_read_from_std() {
+    let mut tmp = NamedTempFile::new().unwrap();
+    let data = b"hello from_std io-uring";
+    tmp.write_all(data).unwrap();
+    tmp.flush().unwrap();
+    let path = tmp.path().to_owned();
+
+    let rt = Builder::new_multi_thread()
+        .worker_threads(2)
+        .enable_all()
+        .build()
+        .unwrap();
+
+    rt.block_on(async {
+        let std_file = std::fs::File::open(&path).unwrap();
+        let mut file = File::from_std(std_file);
+        let mut buf = Vec::new();
+        file.read_to_end(&mut buf).await.unwrap();
+        assert_eq!(buf, data);
+    });
+}
+
+/// Read with a buffer smaller than the file content. Verifies internal
+/// buffering serves subsequent small reads without issuing new underlying read
+/// operations.
+#[tokio::test]
+async fn test_file_read_with_smaller_buf() {
+    let data: Vec<u8> = (0..1024u16).map(|i| (i % 256) as u8).collect();
+    let (_tmp, path) = create_temp_file(&data);
+
+    let mut file = File::open(&path).await.unwrap();
+
+    // triggers an underlying read that fills the internal buffer with more data
+    // than we consume here
+    let mut buf = vec![0u8; 4];
+    let n = file.read(&mut buf).await.unwrap();
+    assert_eq!(n, 4);
+    assert_eq!(&buf, &data[..4]);
+
+    // Second read: still smaller than what's buffered internally
+    let mut buf = vec![0u8; 32];
+    let n = file.read(&mut buf).await.unwrap();
+    assert!(n > 0);
+    assert_eq!(&buf[..n], &data[4..4 + n]);
+
+    // Read the rest
+    let mut rest = Vec::new();
+    file.read_to_end(&mut rest).await.unwrap();
+    let total = 4 + n + rest.len();
+    assert_eq!(total, data.len());
+}
+
+#[tokio::test]
+async fn test_file_read_with_bigger_buf() {
+    let data = b"hello io-uring";
+    let (_tmp, path) = create_temp_file(data);
+
+    let mut file = File::open(&path).await.unwrap();
+
+    // Read with buffer larger than file's contents
+    let mut buf = vec![0u8; 1024];
+    let n = file.read(&mut buf).await.unwrap();
+    assert!(n > 0 && n <= data.len());
+    assert_eq!(&buf[..n], &data[..n]);
+
+    if n < data.len() {
+        let mut rest = vec![0u8; 1024];
+        let n2 = file.read(&mut rest).await.unwrap();
+        assert_eq!(&rest[..n2], &data[n..n + n2]);
+    }
+}
+
+/// Read a file larger than DEFAULT_MAX_BUF_SIZE (2 MiB). Verifies that
+/// chunked reads across multiple underlying operations produce correct data.
+#[tokio::test]
+async fn test_file_read_buffer_larger_than_max() {
+    // 4 MiB + 1000 bytes to cross multiple chunk boundaries.
+    let size = (4 << 20) + 1000;
+    let data: Vec<u8> = (0..size).map(|i| (i % 256) as u8).collect();
+    let (_tmp, path) = create_temp_file(&data);
+
+    let mut file = File::open(&path).await.unwrap();
+    let mut buf = Vec::new();
+    file.read_to_end(&mut buf).await.unwrap();
+    assert_eq!(buf.len(), data.len());
+    assert_eq!(buf, data);
+}
+
+/// Read some bytes from a file, then write, verifying the implicit seek-back
+/// works correctly.
+#[tokio::test]
+async fn test_file_read_then_write() {
+    let original = b"hello world, io-uring!";
+    let (_tmp, path) = create_temp_file(original);
+
+    let mut file = tokio::fs::OpenOptions::new()
+        .read(true)
+        .write(true)
+        .open(&path)
+        .await
+        .unwrap();
+
+    let mut buf = vec![0u8; 5];
+    file.read_exact(&mut buf).await.unwrap();
+    assert_eq!(&buf, b"hello");
+
+    // Write at the current position
+    file.write_all(b" REPLACED").await.unwrap();
+    file.flush().await.unwrap();
+
+    // Re-read the full file and verify the write landed correctly
+    file.seek(std::io::SeekFrom::Start(0)).await.unwrap();
+    let mut result = Vec::new();
+    file.read_to_end(&mut result).await.unwrap();
+    assert_eq!(&result[..5], b"hello");
+    assert_eq!(&result[5..14], b" REPLACED");
+}
+
+/// Partial read followed by write at a different position. Verifies the
+/// seek-back accounts for partially consumed internal buffer.
+#[tokio::test]
+async fn test_file_partial_read_then_write() {
+    let data = b"abcdefghijklmnopqrstuvwxyz";
+    let (_tmp, path) = create_temp_file(data);
+
+    let mut file = tokio::fs::OpenOptions::new()
+        .read(true)
+        .write(true)
+        .open(&path)
+        .await
+        .unwrap();
+
+    // First read fills internal buffer.
+    let mut buf = vec![0u8; 10];
+    file.read_exact(&mut buf).await.unwrap();
+    assert_eq!(&buf, b"abcdefghij");
+
+    // Second read served from internal buffer
+    let mut buf2 = vec![0u8; 3];
+    file.read_exact(&mut buf2).await.unwrap();
+    assert_eq!(&buf2, b"klm");
+
+    // Write should seek back past unconsumed buffered bytes, then write
+    file.write_all(b"NOP").await.unwrap();
+    file.flush().await.unwrap();
+
+    file.seek(std::io::SeekFrom::Start(0)).await.unwrap();
+    let mut result = Vec::new();
+    file.read_to_end(&mut result).await.unwrap();
+    assert_eq!(&result[..13], b"abcdefghijklm");
+    assert_eq!(&result[13..16], b"NOP");
+    assert_eq!(&result[16..], &data[16..]);
+}