From f943312865b9d5007f25d2fd5bd8efa3f89d1541 Mon Sep 17 00:00:00 2001 From: Carter Green Date: Mon, 13 Apr 2026 13:36:23 -0500 Subject: [PATCH] fs: support io-uring in `AsyncRead` for `File` (#7907) --- spellcheck.dic | 3 +- tokio/src/fs/file.rs | 132 +++++++- tokio/src/fs/read_uring.rs | 2 +- tokio/src/io/blocking.rs | 33 ++ tokio/src/io/uring/read.rs | 113 +++++-- tokio/src/io/uring/utils.rs | 25 ++ tokio/src/runtime/driver/op.rs | 7 +- tokio/src/runtime/io/driver/uring.rs | 20 ++ tokio/tests/fs_uring_file_read.rs | 471 +++++++++++++++++++++++++++ 9 files changed, 766 insertions(+), 40 deletions(-) create mode 100644 tokio/tests/fs_uring_file_read.rs diff --git a/spellcheck.dic b/spellcheck.dic index 6e162d39c..e3cbfe077 100644 --- a/spellcheck.dic +++ b/spellcheck.dic @@ -1,4 +1,4 @@ -317 +318 & + < @@ -244,6 +244,7 @@ spawner Splitter spmc spsc +SQE src stabilised startup diff --git a/tokio/src/fs/file.rs b/tokio/src/fs/file.rs index 3579f11ba..12c1f2dbe 100644 --- a/tokio/src/fs/file.rs +++ b/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, + buf: Buf, + max_buf_size: usize, + ) -> io::Result> { + // 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, 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, + 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; diff --git a/tokio/src/fs/read_uring.rs b/tokio/src/fs/read_uring.rs index 5b38c2122..67d709a2c 100644 --- a/tokio/src/fs/read_uring.rs +++ b/tokio/src/fs/read_uring.rs @@ -116,7 +116,7 @@ async fn op_read( read_len: u32, ) -> io::Result<(OwnedFd, Vec, 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 => { diff --git a/tokio/src/io/blocking.rs b/tokio/src/io/blocking.rs index 1af506545..2aefd2cc0 100644 --- a/tokio/src/io/blocking.rs +++ b/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::(); + (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 { diff --git a/tokio/src/io/uring/read.rs b/tokio/src/io/uring/read.rs index e8ee633ac..7eb81fe1b 100644 --- a/tokio/src/io/uring/read.rs +++ b/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, +/// 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, OwnedFd, Vec); +impl ReadBuffer for Vec { + 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 { + fd: F, + buf: B, +} + +impl fmt::Debug for Read { + fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result { + f.debug_struct("Read") + .field("buf", &self.buf) + .finish_non_exhaustive() + } +} + +impl Completable for Read { + type Output = (io::Result, 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, OwnedFd> { fn cancel(self) -> CancelData { - CancelData::Read(self) + CancelData::ReadVec(self) } } -impl Op { - // 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 { + 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, 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 Op> +where + B: ReadBuffer + fmt::Debug, + F: UringFd, + Read: 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 }) } } } diff --git a/tokio/src/io/uring/utils.rs b/tokio/src/io/uring/utils.rs index e30e7a5dd..651859366 100644 --- a/tokio/src/io/uring/utils.rs +++ b/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; + +/// Raw file descriptor trait for io-uring operations. +/// +/// `Arc` does not satisfy `AsRawFd` because the blanket impl +/// for `Arc` 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 { Ok(CString::new(p.as_os_str().as_bytes())?) } diff --git a/tokio/src/runtime/driver/op.rs b/tokio/src/runtime/driver/op.rs index d2b9289ce..f5fe4c37b 100644 --- a/tokio/src/runtime/driver/op.rs +++ b/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, OwnedFd>), + ReadBuf(Read), } #[derive(Debug)] diff --git a/tokio/src/runtime/io/driver/uring.rs b/tokio/src/runtime/io/driver/uring.rs index 1b946c1b3..55475adf9 100644 --- a/tokio/src/runtime/io/driver/uring.rs +++ b/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. /// diff --git a/tokio/tests/fs_uring_file_read.rs b/tokio/tests/fs_uring_file_read.rs new file mode 100644 index 000000000..7e4b2e866 --- /dev/null +++ b/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 Runtime> { + Box::new(move || { + Builder::new_multi_thread() + .worker_threads(n) + .enable_all() + .build() + .unwrap() + }) +} + +fn current_rt() -> Box Runtime> { + Box::new(|| Builder::new_current_thread().enable_all().build().unwrap()) +} + +fn rt_combinations() -> Vec 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 = (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 = (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 = (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 = (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 = (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 = (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 = (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 = (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 = (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 = (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 = (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 = (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..]); +}