use std::cmp; use std::fmt; use std::fs::{Metadata, Permissions}; use std::future::Future; use std::io::{self, Seek, SeekFrom}; use std::path::Path; use std::pin::Pin; use std::sync::Arc; use std::task::{ready, Context, Poll};
#[cfg(test)] usesuper::mocks::JoinHandle; #[cfg(test)] usesuper::mocks::MockFile as StdFile; #[cfg(test)] usesuper::mocks::{spawn_blocking, spawn_mandatory_blocking}; #[cfg(not(test))] usecrate::blocking::JoinHandle; #[cfg(not(test))] usecrate::blocking::{spawn_blocking, spawn_mandatory_blocking}; #[cfg(not(test))] use std::fs::File as StdFile;
/// A reference to an open file on the filesystem. /// /// This is a specialized version of [`std::fs::File`] for usage from the /// Tokio runtime. /// /// An instance of a `File` can be read and/or written depending on what options /// it was opened with. Files also implement [`AsyncSeek`] to alter the logical /// cursor that the file contains internally. /// /// A file will not be closed immediately when it goes out of scope if there /// are any IO operations that have not yet completed. To ensure that a file is /// closed immediately when it is dropped, you should call [`flush`] before /// dropping it. Note that this does not ensure that the file has been fully /// written to disk; the operating system might keep the changes around in an /// in-memory buffer. See the [`sync_all`] method for telling the OS to write /// the data to disk. /// /// Reading and writing to a `File` is usually done using the convenience /// methods found on the [`AsyncReadExt`] and [`AsyncWriteExt`] traits. /// /// [`AsyncSeek`]: trait@crate::io::AsyncSeek /// [`flush`]: fn@crate::io::AsyncWriteExt::flush /// [`sync_all`]: fn@crate::fs::File::sync_all /// [`AsyncReadExt`]: trait@crate::io::AsyncReadExt /// [`AsyncWriteExt`]: trait@crate::io::AsyncWriteExt /// /// # Examples /// /// Create a new file and asynchronously write bytes to it: /// /// ```no_run /// use tokio::fs::File; /// use tokio::io::AsyncWriteExt; // for write_all() /// /// # async fn dox() -> std::io::Result<()> { /// let mut file = File::create("foo.txt").await?; /// file.write_all(b"hello, world!").await?; /// # Ok(()) /// # } /// ``` /// /// Read the contents of a file into a buffer: /// /// ```no_run /// use tokio::fs::File; /// use tokio::io::AsyncReadExt; // for read_to_end() /// /// # async fn dox() -> std::io::Result<()> { /// let mut file = File::open("foo.txt").await?; /// /// let mut contents = vec![]; /// file.read_to_end(&mut contents).await?; /// /// println!("len = {}", contents.len()); /// # Ok(()) /// # } /// ``` pubstruct File {
std: Arc<StdFile>,
inner: Mutex<Inner>,
max_buf_size: usize,
}
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)>),
}
impl File { /// Attempts to open a file in read-only mode. /// /// See [`OpenOptions`] for more details. /// /// # Errors /// /// This function will return an error if called from outside of the Tokio /// runtime or if path does not already exist. Other errors may also be /// returned according to `OpenOptions::open`. /// /// # Examples /// /// ```no_run /// use tokio::fs::File; /// use tokio::io::AsyncReadExt; /// /// # async fn dox() -> std::io::Result<()> { /// let mut file = File::open("foo.txt").await?; /// /// let mut contents = vec![]; /// file.read_to_end(&mut contents).await?; /// /// println!("len = {}", contents.len()); /// # Ok(()) /// # } /// ``` /// /// The [`read_to_end`] method is defined on the [`AsyncReadExt`] trait. /// /// [`read_to_end`]: fn@crate::io::AsyncReadExt::read_to_end /// [`AsyncReadExt`]: trait@crate::io::AsyncReadExt pubasyncfn open(path: impl AsRef<Path>) -> io::Result<File> { Self::options().read(true).open(path).await
}
/// Opens a file in write-only mode. /// /// This function will create a file if it does not exist, and will truncate /// it if it does. /// /// See [`OpenOptions`] for more details. /// /// # Errors /// /// Results in an error if called from outside of the Tokio runtime or if /// the underlying [`create`] call results in an error. /// /// [`create`]: std::fs::File::create /// /// # Examples /// /// ```no_run /// use tokio::fs::File; /// use tokio::io::AsyncWriteExt; /// /// # async fn dox() -> std::io::Result<()> { /// let mut file = File::create("foo.txt").await?; /// file.write_all(b"hello, world!").await?; /// # Ok(()) /// # } /// ``` /// /// The [`write_all`] method is defined on the [`AsyncWriteExt`] trait. /// /// [`write_all`]: fn@crate::io::AsyncWriteExt::write_all /// [`AsyncWriteExt`]: trait@crate::io::AsyncWriteExt pubasyncfn create(path: impl AsRef<Path>) -> io::Result<File> { Self::options()
.write(true)
.create(true)
.truncate(true)
.open(path)
.await
}
/// Opens a file in read-write mode. /// /// This function will create a file if it does not exist, or return an error /// if it does. This way, if the call succeeds, the file returned is guaranteed /// to be new. /// /// This option is useful because it is atomic. Otherwise between checking /// whether a file exists and creating a new one, the file may have been /// created by another process (a TOCTOU race condition / attack). /// /// This can also be written using `File::options().read(true).write(true).create_new(true).open(...)`. /// /// See [`OpenOptions`] for more details. /// /// # Examples /// /// ```no_run /// use tokio::fs::File; /// use tokio::io::AsyncWriteExt; /// /// # async fn dox() -> std::io::Result<()> { /// let mut file = File::create_new("foo.txt").await?; /// file.write_all(b"hello, world!").await?; /// # Ok(()) /// # } /// ``` /// /// The [`write_all`] method is defined on the [`AsyncWriteExt`] trait. /// /// [`write_all`]: fn@crate::io::AsyncWriteExt::write_all /// [`AsyncWriteExt`]: trait@crate::io::AsyncWriteExt pubasyncfn create_new<P: AsRef<Path>>(path: P) -> std::io::Result<File> { Self::options()
.read(true)
.write(true)
.create_new(true)
.open(path)
.await
}
/// Returns a new [`OpenOptions`] object. /// /// This function returns a new `OpenOptions` object that you can use to /// open or create a file with specific options if `open()` or `create()` /// are not appropriate. /// /// It is equivalent to `OpenOptions::new()`, but allows you to write more /// readable code. Instead of /// `OpenOptions::new().append(true).open("example.log")`, /// you can write `File::options().append(true).open("example.log")`. This /// also avoids the need to import `OpenOptions`. /// /// See the [`OpenOptions::new`] function for more details. /// /// # Examples /// /// ```no_run /// use tokio::fs::File; /// use tokio::io::AsyncWriteExt; /// /// # async fn dox() -> std::io::Result<()> { /// let mut f = File::options().append(true).open("example.log").await?; /// f.write_all(b"new line\n").await?; /// # Ok(()) /// # } /// ``` #[must_use] pubfn options() -> OpenOptions {
OpenOptions::new()
}
/// Converts a [`std::fs::File`] to a [`tokio::fs::File`](File). /// /// # Examples /// /// ```no_run /// // This line could block. It is not recommended to do this on the Tokio /// // runtime. /// let std_file = std::fs::File::open("foo.txt").unwrap(); /// let file = tokio::fs::File::from_std(std_file); /// ``` pubfn from_std(std: StdFile) -> File {
File {
std: Arc::new(std),
inner: Mutex::new(Inner {
state: State::Idle(Some(Buf::with_capacity(0))),
last_write_err: None,
pos: 0,
}),
max_buf_size: DEFAULT_MAX_BUF_SIZE,
}
}
/// Attempts to sync all OS-internal metadata to disk. /// /// This function will attempt to ensure that all in-core data reaches the /// filesystem before returning. /// /// # Examples /// /// ```no_run /// use tokio::fs::File; /// use tokio::io::AsyncWriteExt; /// /// # async fn dox() -> std::io::Result<()> { /// let mut file = File::create("foo.txt").await?; /// file.write_all(b"hello, world!").await?; /// file.sync_all().await?; /// # Ok(()) /// # } /// ``` /// /// The [`write_all`] method is defined on the [`AsyncWriteExt`] trait. /// /// [`write_all`]: fn@crate::io::AsyncWriteExt::write_all /// [`AsyncWriteExt`]: trait@crate::io::AsyncWriteExt pubasyncfn sync_all(&self) -> io::Result<()> { letmut inner = self.inner.lock().await;
inner.complete_inflight().await;
let std = self.std.clone();
asyncify(move || std.sync_all()).await
}
/// This function is similar to `sync_all`, except that it may not /// synchronize file metadata to the filesystem. /// /// This is intended for use cases that must synchronize content, but don't /// need the metadata on disk. The goal of this method is to reduce disk /// operations. /// /// Note that some platforms may simply implement this in terms of `sync_all`. /// /// # Examples /// /// ```no_run /// use tokio::fs::File; /// use tokio::io::AsyncWriteExt; /// /// # async fn dox() -> std::io::Result<()> { /// let mut file = File::create("foo.txt").await?; /// file.write_all(b"hello, world!").await?; /// file.sync_data().await?; /// # Ok(()) /// # } /// ``` /// /// The [`write_all`] method is defined on the [`AsyncWriteExt`] trait. /// /// [`write_all`]: fn@crate::io::AsyncWriteExt::write_all /// [`AsyncWriteExt`]: trait@crate::io::AsyncWriteExt pubasyncfn sync_data(&self) -> io::Result<()> { letmut inner = self.inner.lock().await;
inner.complete_inflight().await;
let std = self.std.clone();
asyncify(move || std.sync_data()).await
}
/// Truncates or extends the underlying file, updating the size of this file to become size. /// /// If the size is less than the current file's size, then the file will be /// shrunk. If it is greater than the current file's size, then the file /// will be extended to size and have all of the intermediate data filled in /// with 0s. /// /// # Errors /// /// This function will return an error if the file is not opened for /// writing. /// /// # Examples /// /// ```no_run /// use tokio::fs::File; /// use tokio::io::AsyncWriteExt; /// /// # async fn dox() -> std::io::Result<()> { /// let mut file = File::create("foo.txt").await?; /// file.write_all(b"hello, world!").await?; /// file.set_len(10).await?; /// # Ok(()) /// # } /// ``` /// /// The [`write_all`] method is defined on the [`AsyncWriteExt`] trait. /// /// [`write_all`]: fn@crate::io::AsyncWriteExt::write_all /// [`AsyncWriteExt`]: trait@crate::io::AsyncWriteExt pubasyncfn set_len(&self, size: u64) -> io::Result<()> { letmut inner = self.inner.lock().await;
inner.complete_inflight().await;
let seek = if !buf.is_empty() {
Some(SeekFrom::Current(buf.discard_read()))
} else {
None
};
let std = self.std.clone();
inner.state = State::Busy(spawn_blocking(move || { let res = iflet Some(seek) = seek {
(&*std).seek(seek).and_then(|_| std.set_len(size))
} else {
std.set_len(size)
}
.map(|()| 0); // the value is discarded later
// Return the result as a seek
(Operation::Seek(res), buf)
}));
let (op, buf) = match inner.state {
State::Idle(_) => unreachable!(),
State::Busy(refmut rx) => rx.await?,
};
inner.state = State::Idle(Some(buf));
match op {
Operation::Seek(res) => res.map(|pos| {
inner.pos = pos;
}),
_ => unreachable!(),
}
}
/// Creates a new `File` instance that shares the same underlying file handle /// as the existing `File` instance. Reads, writes, and seeks will affect both /// File instances simultaneously. /// /// # Examples /// /// ```no_run /// use tokio::fs::File; /// /// # async fn dox() -> std::io::Result<()> { /// let file = File::open("foo.txt").await?; /// let file_clone = file.try_clone().await?; /// # Ok(()) /// # } /// ``` pubasyncfn try_clone(&self) -> io::Result<File> { self.inner.lock().await.complete_inflight().await; let std = self.std.clone(); let std_file = asyncify(move || std.try_clone()).await?; letmut file = File::from_std(std_file);
file.set_max_buf_size(self.max_buf_size);
Ok(file)
}
/// Destructures `File` into a [`std::fs::File`]. This function is /// async to allow any in-flight operations to complete. /// /// Use `File::try_into_std` to attempt conversion immediately. /// /// # Examples /// /// ```no_run /// use tokio::fs::File; /// /// # async fn dox() -> std::io::Result<()> { /// let tokio_file = File::open("foo.txt").await?; /// let std_file = tokio_file.into_std().await; /// # Ok(()) /// # } /// ``` pubasyncfn into_std(mutself) -> StdFile { self.inner.get_mut().complete_inflight().await;
Arc::try_unwrap(self.std).expect("Arc::try_unwrap failed")
}
/// Tries to immediately destructure `File` into a [`std::fs::File`]. /// /// # Errors /// /// This function will return an error containing the file if some /// operation is in-flight. /// /// # Examples /// /// ```no_run /// use tokio::fs::File; /// /// # async fn dox() -> std::io::Result<()> { /// let tokio_file = File::open("foo.txt").await?; /// let std_file = tokio_file.try_into_std().unwrap(); /// # Ok(()) /// # } /// ``` #[allow(clippy::result_large_err)] pubfn try_into_std(mutself) -> Result<StdFile, Self> { match Arc::try_unwrap(self.std) {
Ok(file) => Ok(file),
Err(std_file_arc) => { self.std = std_file_arc;
Err(self)
}
}
}
/// Changes the permissions on the underlying file. /// /// # Platform-specific behavior /// /// This function currently corresponds to the `fchmod` function on Unix and /// the `SetFileInformationByHandle` function on Windows. Note that, this /// [may change in the future][changes]. /// /// [changes]: https://doc.rust-lang.org/std/io/index.html#platform-specific-behavior /// /// # Errors /// /// This function will return an error if the user lacks permission change /// attributes on the underlying file. It may also return an error in other /// os-specific unspecified cases. /// /// # Examples /// /// ```no_run /// use tokio::fs::File; /// /// # async fn dox() -> std::io::Result<()> { /// let file = File::open("foo.txt").await?; /// let mut perms = file.metadata().await?.permissions(); /// perms.set_readonly(true); /// file.set_permissions(perms).await?; /// # Ok(()) /// # } /// ``` pubasyncfn set_permissions(&self, perm: Permissions) -> io::Result<()> { let std = self.std.clone();
asyncify(move || std.set_permissions(perm)).await
}
/// Set the maximum buffer size for the underlying [`AsyncRead`] / [`AsyncWrite`] operation. /// /// Although Tokio uses a sensible default value for this buffer size, this function would be /// useful for changing that default depending on the situation. /// /// # Examples /// /// ```no_run /// use tokio::fs::File; /// use tokio::io::AsyncWriteExt; /// /// # async fn dox() -> std::io::Result<()> { /// let mut file = File::open("foo.txt").await?; /// /// // Set maximum buffer size to 8 MiB /// file.set_max_buf_size(8 * 1024 * 1024); /// /// let mut buf = vec![1; 1024 * 1024 * 1024]; /// /// // Write the 1 GiB buffer in chunks up to 8 MiB each. /// file.write_all(&mut buf).await?; /// # Ok(()) /// # } /// ``` pubfn set_max_buf_size(&mutself, max_buf_size: usize) { self.max_buf_size = max_buf_size;
}
/// Get the maximum buffer size for the underlying [`AsyncRead`] / [`AsyncWrite`] operation. pubfn max_buf_size(&self) -> usize { self.max_buf_size
}
}
match op {
Operation::Read(_) => { // We don't care about the result here. The fact // that the cursor has advanced will be reflected in // the next iteration of the loop continue;
}
Operation::Write(res) => { // If the previous write was successful, continue. // Otherwise, error.
res?; continue;
}
Operation::Seek(_) => { // Ignore the seek continue;
}
}
}
}
}
}
fn poll_write_vectored( self: Pin<&mutSelf>,
cx: &mut Context<'_>,
bufs: &[io::IoSlice<'_>],
) -> Poll<Result<usize, io::Error>> {
ready!(crate::trace::trace_leaf(cx)); let me = self.get_mut(); let inner = me.inner.get_mut();
match op {
Operation::Read(_) => { // We don't care about the result here. The fact // that the cursor has advanced will be reflected in // the next iteration of the loop continue;
}
Operation::Write(res) => { // If the previous write was successful, continue. // Otherwise, error.
res?; continue;
}
Operation::Seek(_) => { // Ignore the seek continue;
}
}
}
}
}
}
impl FromRawHandle for File { unsafefn from_raw_handle(handle: RawHandle) -> Self { // Safety: exactly the same safety contract as // `FromRawHandle::from_raw_handle`. unsafe { StdFile::from_raw_handle(handle).into() }
}
}
}
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",
))]
{ iflet 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.
}
}
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",
))] asyncfn 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 { matchSelf::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 || { letmut 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)
})
}
asyncfn complete_inflight(&mutself) { use std::future::poll_fn;
Die Informationen auf dieser Webseite wurden
nach bestem Wissen sorgfältig zusammengestellt. Es wird jedoch weder Vollständigkeit, noch Richtigkeit,
noch Qualität der bereit gestellten Informationen zugesichert.
Bemerkung:
Die farbliche Syntaxdarstellung und die Messung sind noch experimentell.