Skip to content
New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

fs: add read_at/write_at/seek_read/seek_write for fs::File #6427

Open
wants to merge 6 commits into
base: master
Choose a base branch
from
Open
Show file tree
Hide file tree
Changes from all commits
Commits
File filter

Filter by extension

Filter by extension

Conversations
Failed to load comments.
Jump to
Jump to file
Failed to load files.
Diff view
Diff view
166 changes: 166 additions & 0 deletions tokio/src/fs/file.rs
Original file line number Diff line number Diff line change
Expand Up @@ -538,6 +538,172 @@ impl File {
pub fn set_max_buf_size(&mut self, max_buf_size: usize) {
self.max_buf_size = max_buf_size;
}

/// Reads a number of bytes starting from a given offset.
///
/// Returns the number of bytes read.
///
/// The offset is relative to the start of the file and thus independent
/// from the current cursor.
///
/// The current file cursor is not affected by this function.
///
/// It is not an error to return with a short read.
///
/// # Examples
///
/// ```no_run
/// use tokio::fs::File;
/// use tokio::io::AsyncSeekExt;
///
/// # async fn dox() -> std::io::Result<()> {
/// let mut buf = vec![0_u8; 10];
/// let mut file = File::open("foo.txt").await?;
/// file.read_at(&mut buf, 5).await?;
///
/// assert_eq!(file.stream_position().await?, 0);
/// # Ok(())
/// # }
/// ```
#[cfg(unix)]
#[cfg_attr(docsrs, doc(cfg(unix)))]
pub async fn read_at(&self, buf: &mut [u8], offset: u64) -> io::Result<usize> {
fn _read_at(std: &StdFile, n: usize, offset: u64) -> io::Result<Vec<u8>> {
use std::os::unix::fs::FileExt;
let mut buf: Vec<u8> = vec![0; n];
let n_read = std.read_at(&mut buf, offset)?;
buf.truncate(n_read);

Ok(buf)
}

let std = self.std.clone();
let n = buf.len();
let bytes_read = asyncify(move || _read_at(&std, n, offset)).await?;
let len = bytes_read.len();
buf[..len].copy_from_slice(&bytes_read);
Comment on lines +580 to +584
Copy link
Contributor

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

The Tokio file already has a bunch of logic for keeping track of and reusing a buffer, but none of these functions use that logic. Is there a reason you're doing it this way?

Copy link
Contributor Author

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

The Tokio file already has a bunch of logic for keeping track of and reusing a buffer,

Well, I am not aware of this, do you mean the bytes crate? If not, would you like to show me some examples?

Copy link
Contributor

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

No, I mean the stuff related to these types:

tokio/tokio/src/fs/file.rs

Lines 96 to 118 in c98e22f

struct Inner {
state: State,
/// Errors from writes/flushes are returned in write/flush calls. If a write
/// error is observed while performing a read, it is saved until the next
/// write / flush call.
last_write_err: Option<io::ErrorKind>,
pos: u64,
}
#[derive(Debug)]
enum State {
Idle(Option<Buf>),
Busy(JoinHandle<(Operation, Buf)>),
}
#[derive(Debug)]
enum Operation {
Read(io::Result<usize>),
Write(io::Result<()>),
Seek(io::Result<u64>),
}

Copy link
Contributor Author

@SteveLauC SteveLauC Mar 30, 2024

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

Thanks for showing me this!

I just took another look at other methods implemented with spawn_blocking(), seems that I should add:

        let mut inner = self.inner.lock().await;
        inner.complete_inflight().await;

before involving the actual logic, right?

Update: Well, seems more complicated than I thought, I originally thought we could simply implement it using asyncify()

Copy link
Contributor

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

One could probably argue either way about whether it's even desireable. One advantage of read_at is that you can have several calls in parallel, but if we use the state logic, then they will run one-after-the-other.

I'm not sure what the best answer here is.

Copy link
Contributor Author

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

One advantage of read_at is that you can have several calls in parallel, but if we use the state logic, then they will run one-after-the-other.

Thanks for pointing it out! The capability of enabling concurrent access is indeed the reason why pread/pwrite are added, so I slightly tend to use separate buffers

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

One could probably argue either way about whether it's even desireable. One advantage of read_at is that you can have several calls in parallel, but if we use the state logic, then they will run one-after-the-other.

I'm not sure what the best answer here is.

I have a tentative plan, can we use RwLock instead of Mutex to achieve sharing among multiple readers and allowing them run in parallel?

inner: Mutex<Inner>,

Copy link
Contributor

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

@Chasing1020 A read/write lock doesn't help. Fundamentally, the shared buffer can only hold one piece of data at the time.


Ok(len)
}

/// Writes a number of bytes starting from a given offset.
///
/// Returns the number of bytes written.
///
/// The offset is relative to the start of the file and thus independent from
/// the current cursor.
///
/// The current file cursor is not affected by this function.
///
/// When writing beyond the end of the file, the file is appropriately
/// extended and the intermediate bytes are initialized with the value 0.
///
/// It is not an error to return a short write.
///
/// # Examples
///
/// ```no_run
/// use tokio::fs::File;
/// use tokio::io::AsyncSeekExt;
///
/// # async fn dox() -> std::io::Result<()> {
/// let mut file = File::create("foo.txt").await?;
/// file.write_at(b"foo", 5).await?;
///
/// assert_eq!(file.stream_position().await?, 0);
/// # Ok(())
/// # }
/// ```
#[cfg(unix)]
#[cfg_attr(docsrs, doc(cfg(unix)))]
pub async fn write_at(&self, buf: &[u8], offset: u64) -> io::Result<usize> {
use std::os::unix::fs::FileExt;

let std = self.std.clone();
let buf_clone = buf.to_vec();
asyncify(move || std.write_at(&buf_clone, offset)).await
}

/// Seeks to a given position and reads a number of bytes.
///
/// Returns the number of bytes read.
///
/// The offset is relative to the start of the file and thus independent from
/// the current cursor. The current cursor is affected by this function, it
/// is set to the end of the read.
///
/// Reading beyond the end of the file will always return with a length of 0.
///
/// It is not an error to return with a short read. When returning from
/// such a short read, the file pointer is still updated.
///
/// # Examples
///
/// ```no_run
/// use tokio::fs::File;
///
/// # async fn dox() -> std::io::Result<()> {
/// let file = File::open("foo.txt").await?;
/// let mut buf = vec![0_u8; 10];
/// file.seek_read(&mut buf, 5).await?;
/// # Ok(())
/// # }
/// ```
#[cfg(windows)]
#[cfg_attr(docsrs, doc(cfg(windows)))]
pub async fn seek_read(&self, buf: &mut [u8], offset: u64) -> io::Result<usize> {
fn _read_at(std: &StdFile, n: usize, offset: u64) -> io::Result<Vec<u8>> {
use std::os::windows::fs::FileExt;
let mut buf: Vec<u8> = vec![0; n];
let n_read = std.seek_read(&mut buf, offset)?;
buf.truncate(n_read);

Ok(buf)
}

let std = self.std.clone();
let n = buf.len();
let bytes_read = asyncify(move || _read_at(&std, n, offset)).await?;
let len = bytes_read.len();
buf[..len].copy_from_slice(&bytes_read);

Ok(len)
}

/// Seeks to a given position and writes a number of bytes.
///
/// Returns the number of bytes written.
///
/// The offset is relative to the start of the file and thus independent from
/// the current cursor. The current cursor is affected by this function, it
/// is set to the end of the write.
///
/// When writing beyond the end of the file, the file is appropriately
/// extended and the intermediate bytes are set to zero.
///
/// It is not an error to return a short write. When returning from such a
/// short write, the file pointer is still updated.
///
/// # Examples
///
/// ```no_run
/// use tokio::fs::File;
///
/// # async fn dox() -> std::io::Result<()> {
/// let file = File::create("foo.txt").await?;
/// file.seek_write(b"foo", 5).await?;
/// # Ok(())
/// # }
/// ```
#[cfg(windows)]
#[cfg_attr(docsrs, doc(cfg(windows)))]
pub async fn seek_write(&self, buf: &[u8], offset: u64) -> io::Result<usize> {
use std::os::windows::fs::FileExt;

let std = self.std.clone();
let buf_clone = buf.to_vec();
asyncify(move || std.seek_write(&buf_clone, offset)).await
}
}

impl AsyncRead for File {
Expand Down
12 changes: 12 additions & 0 deletions tokio/src/fs/mocks.rs
Original file line number Diff line number Diff line change
Expand Up @@ -52,6 +52,18 @@ mock! {
impl std::os::unix::io::FromRawFd for File {
unsafe fn from_raw_fd(h: std::os::unix::io::RawFd) -> Self;
}

#[cfg(unix)]
impl std::os::unix::fs::FileExt for File {
fn read_at(&self, buf: &mut [u8], offset: u64) -> io::Result<usize>;
fn write_at(&self, buf: &[u8], offset: u64) -> io::Result<usize>;
}

#[cfg(windows)]
impl std::os::windows::fs::FileExt for File {
fn seek_read(&self, buf: &mut [u8], offset: u64) -> io::Result<usize>;
fn seek_write(&self, buf: &[u8], offset: u64) -> io::Result<usize>;
}
}

impl Read for MockFile {
Expand Down
21 changes: 21 additions & 0 deletions tokio/tests/io_read_at.rs
Original file line number Diff line number Diff line change
@@ -0,0 +1,21 @@
#![warn(rust_2018_idioms)]
#![cfg(all(feature = "full", not(target_os = "wasi")))]

#[tokio::test]
#[cfg(unix)]
async fn read_at() {
use tempfile::tempdir;
use tokio::fs;
use tokio::io::AsyncSeekExt;

let temp_dir = tempdir().unwrap();
let file_path = temp_dir.path().join("a.txt");
fs::write(&file_path, b"HelloWorld").await.unwrap();
let mut file = fs::File::open(file_path.as_path()).await.unwrap();

let mut buf = [0_u8; 10];
assert_eq!(file.read_at(&mut buf, 5).await.unwrap(), 5);
assert_eq!(&buf[..5], b"World");

assert_eq!(file.stream_position().await.unwrap(), 0);
}
21 changes: 21 additions & 0 deletions tokio/tests/io_seek_read.rs
Original file line number Diff line number Diff line change
@@ -0,0 +1,21 @@
#![warn(rust_2018_idioms)]
#![cfg(all(feature = "full", not(target_os = "wasi")))]

#[tokio::test]
#[cfg(windows)]
async fn seek_read() {
use tempfile::tempdir;
use tokio::fs;
use tokio::io::AsyncSeekExt;

let temp_dir = tempdir().unwrap();
let file_path = temp_dir.path().join("a.txt");
fs::write(&file_path, b"HelloWorld").await.unwrap();
let mut file = fs::File::open(file_path.as_path()).await.unwrap();

let mut buf = [0_u8; 10];
assert_eq!(file.seek_read(&mut buf, 5).await.unwrap(), 5);
assert_eq!(&buf[..5], b"World");

assert_eq!(file.stream_position().await.unwrap(), 10);
}
26 changes: 26 additions & 0 deletions tokio/tests/io_seek_write.rs
Original file line number Diff line number Diff line change
@@ -0,0 +1,26 @@
#![warn(rust_2018_idioms)]
#![cfg(all(feature = "full", not(target_os = "wasi")))]

#[tokio::test]
#[cfg(windows)]
async fn seek_write() {
use tempfile::tempdir;
use tokio::fs;
use tokio::fs::OpenOptions;
use tokio::io::AsyncSeekExt;

let temp_dir = tempdir().unwrap();
let file_path = temp_dir.path().join("a.txt");
fs::write(&file_path, b"Hello File").await.unwrap();
let mut file = OpenOptions::new()
.write(true)
.open(file_path.as_path())
.await
.unwrap();

assert_eq!(file.seek_write(b"World", 5).await.unwrap(), 5);
let contents = fs::read(file_path.as_path()).await.unwrap();
assert_eq!(contents, b"HelloWorld");

assert_eq!(file.stream_position().await.unwrap(), 10);
}
26 changes: 26 additions & 0 deletions tokio/tests/io_write_at.rs
Original file line number Diff line number Diff line change
@@ -0,0 +1,26 @@
#![warn(rust_2018_idioms)]
#![cfg(all(feature = "full", not(target_os = "wasi")))]

#[tokio::test]
#[cfg(unix)]
async fn write_at() {
use tempfile::tempdir;
use tokio::fs;
use tokio::fs::OpenOptions;
use tokio::io::AsyncSeekExt;

let temp_dir = tempdir().unwrap();
let file_path = temp_dir.path().join("a.txt");
fs::write(&file_path, b"Hello File").await.unwrap();
let mut file = OpenOptions::new()
.write(true)
.open(file_path.as_path())
.await
.unwrap();

assert_eq!(file.write_at(b"World", 5).await.unwrap(), 5);
let contents = fs::read(file_path.as_path()).await.unwrap();
assert_eq!(contents, b"HelloWorld");

assert_eq!(file.stream_position().await.unwrap(), 0);
}