tokio/fs/file.rs
1//! Types for working with [`File`].
2//!
3//! [`File`]: File
4
5use crate::fs::{asyncify, FileTimes, OpenOptions};
6use crate::io::blocking::{Buf, DEFAULT_MAX_BUF_SIZE};
7use crate::io::{AsyncRead, AsyncSeek, AsyncWrite, ReadBuf};
8use crate::sync::Mutex;
9
10use std::cmp;
11use std::fmt;
12use std::fs::{Metadata, Permissions};
13use std::future::Future;
14use std::io::{self, Seek, SeekFrom};
15use std::path::Path;
16use std::pin::Pin;
17use std::sync::Arc;
18use std::task::{ready, Context, Poll};
19
20#[cfg(test)]
21use super::mocks::JoinHandle;
22#[cfg(test)]
23use super::mocks::MockFile as StdFile;
24#[cfg(test)]
25use super::mocks::{spawn_blocking, spawn_mandatory_blocking};
26#[cfg(not(test))]
27use crate::blocking::JoinHandle;
28#[cfg(not(test))]
29use crate::blocking::{spawn_blocking, spawn_mandatory_blocking};
30#[cfg(not(test))]
31use std::fs::File as StdFile;
32
33cfg_io_uring! {
34 #[cfg(not(test))]
35 use crate::spawn;
36}
37
38/// A reference to an open file on the filesystem.
39///
40/// This is a specialized version of [`std::fs::File`] for usage from the
41/// Tokio runtime.
42///
43/// An instance of a `File` can be read and/or written depending on what options
44/// it was opened with. Files also implement [`AsyncSeek`] to alter the logical
45/// cursor that the file contains internally.
46///
47/// A file will not be closed immediately when it goes out of scope if there
48/// are any IO operations that have not yet completed. To ensure that a file is
49/// closed immediately when it is dropped, you should call [`flush`] before
50/// dropping it. Note that this does not ensure that the file has been fully
51/// written to disk; the operating system might keep the changes around in an
52/// in-memory buffer. See the [`sync_all`] method for telling the OS to write
53/// the data to disk.
54///
55/// Reading and writing to a `File` is usually done using the convenience
56/// methods found on the [`AsyncReadExt`] and [`AsyncWriteExt`] traits.
57///
58/// [`AsyncSeek`]: trait@crate::io::AsyncSeek
59/// [`flush`]: fn@crate::io::AsyncWriteExt::flush
60/// [`sync_all`]: fn@crate::fs::File::sync_all
61/// [`AsyncReadExt`]: trait@crate::io::AsyncReadExt
62/// [`AsyncWriteExt`]: trait@crate::io::AsyncWriteExt
63///
64/// # Examples
65///
66/// Create a new file and asynchronously write bytes to it:
67///
68/// ```no_run
69/// use tokio::fs::File;
70/// use tokio::io::AsyncWriteExt; // for write_all()
71///
72/// # async fn dox() -> std::io::Result<()> {
73/// let mut file = File::create("foo.txt").await?;
74/// file.write_all(b"hello, world!").await?;
75/// # Ok(())
76/// # }
77/// ```
78///
79/// Read the contents of a file into a buffer:
80///
81/// ```no_run
82/// use tokio::fs::File;
83/// use tokio::io::AsyncReadExt; // for read_to_end()
84///
85/// # async fn dox() -> std::io::Result<()> {
86/// let mut file = File::open("foo.txt").await?;
87///
88/// let mut contents = vec![];
89/// file.read_to_end(&mut contents).await?;
90///
91/// println!("len = {}", contents.len());
92/// # Ok(())
93/// # }
94/// ```
95pub struct File {
96 std: Arc<StdFile>,
97 inner: Mutex<Inner>,
98 max_buf_size: usize,
99}
100
101struct Inner {
102 state: State,
103
104 /// Errors from writes/flushes are returned in write/flush calls. If a write
105 /// error is observed while performing a read, it is saved until the next
106 /// write / flush call.
107 last_write_err: Option<io::ErrorKind>,
108
109 pos: u64,
110}
111
112#[derive(Debug)]
113enum State {
114 Idle(Option<Buf>),
115 Busy(JoinHandle<(Operation, Buf)>),
116}
117
118#[derive(Debug)]
119enum Operation {
120 Read(io::Result<usize>),
121 Write(io::Result<()>),
122 Seek(io::Result<u64>),
123}
124
125impl File {
126 /// Attempts to open a file in read-only mode.
127 ///
128 /// See [`OpenOptions`] for more details.
129 ///
130 /// # Errors
131 ///
132 /// This function will return an error if called from outside of the Tokio
133 /// runtime or if path does not already exist. Other errors may also be
134 /// returned according to `OpenOptions::open`.
135 ///
136 /// # Examples
137 ///
138 /// ```no_run
139 /// use tokio::fs::File;
140 /// use tokio::io::AsyncReadExt;
141 ///
142 /// # async fn dox() -> std::io::Result<()> {
143 /// let mut file = File::open("foo.txt").await?;
144 ///
145 /// let mut contents = vec![];
146 /// file.read_to_end(&mut contents).await?;
147 ///
148 /// println!("len = {}", contents.len());
149 /// # Ok(())
150 /// # }
151 /// ```
152 ///
153 /// The [`read_to_end`] method is defined on the [`AsyncReadExt`] trait.
154 ///
155 /// [`read_to_end`]: fn@crate::io::AsyncReadExt::read_to_end
156 /// [`AsyncReadExt`]: trait@crate::io::AsyncReadExt
157 pub async fn open(path: impl AsRef<Path>) -> io::Result<File> {
158 Self::options().read(true).open(path).await
159 }
160
161 /// Opens a file in write-only mode.
162 ///
163 /// This function will create a file if it does not exist, and will truncate
164 /// it if it does.
165 ///
166 /// See [`OpenOptions`] for more details.
167 ///
168 /// # Errors
169 ///
170 /// Results in an error if called from outside of the Tokio runtime or if
171 /// the underlying [`create`] call results in an error.
172 ///
173 /// [`create`]: std::fs::File::create
174 ///
175 /// # Examples
176 ///
177 /// ```no_run
178 /// use tokio::fs::File;
179 /// use tokio::io::AsyncWriteExt;
180 ///
181 /// # async fn dox() -> std::io::Result<()> {
182 /// let mut file = File::create("foo.txt").await?;
183 /// file.write_all(b"hello, world!").await?;
184 /// # Ok(())
185 /// # }
186 /// ```
187 ///
188 /// The [`write_all`] method is defined on the [`AsyncWriteExt`] trait.
189 ///
190 /// [`write_all`]: fn@crate::io::AsyncWriteExt::write_all
191 /// [`AsyncWriteExt`]: trait@crate::io::AsyncWriteExt
192 pub async fn create(path: impl AsRef<Path>) -> io::Result<File> {
193 Self::options()
194 .write(true)
195 .create(true)
196 .truncate(true)
197 .open(path)
198 .await
199 }
200
201 /// Opens a file in read-write mode.
202 ///
203 /// This function will create a file if it does not exist, or return an error
204 /// if it does. This way, if the call succeeds, the file returned is guaranteed
205 /// to be new.
206 ///
207 /// This option is useful because it is atomic. Otherwise between checking
208 /// whether a file exists and creating a new one, the file may have been
209 /// created by another process (a TOCTOU race condition / attack).
210 ///
211 /// This can also be written using `File::options().read(true).write(true).create_new(true).open(...)`.
212 ///
213 /// See [`OpenOptions`] for more details.
214 ///
215 /// # Examples
216 ///
217 /// ```no_run
218 /// use tokio::fs::File;
219 /// use tokio::io::AsyncWriteExt;
220 ///
221 /// # async fn dox() -> std::io::Result<()> {
222 /// let mut file = File::create_new("foo.txt").await?;
223 /// file.write_all(b"hello, world!").await?;
224 /// # Ok(())
225 /// # }
226 /// ```
227 ///
228 /// The [`write_all`] method is defined on the [`AsyncWriteExt`] trait.
229 ///
230 /// [`write_all`]: fn@crate::io::AsyncWriteExt::write_all
231 /// [`AsyncWriteExt`]: trait@crate::io::AsyncWriteExt
232 pub async fn create_new<P: AsRef<Path>>(path: P) -> std::io::Result<File> {
233 Self::options()
234 .read(true)
235 .write(true)
236 .create_new(true)
237 .open(path)
238 .await
239 }
240
241 /// Returns a new [`OpenOptions`] object.
242 ///
243 /// This function returns a new `OpenOptions` object that you can use to
244 /// open or create a file with specific options if `open()` or `create()`
245 /// are not appropriate.
246 ///
247 /// It is equivalent to `OpenOptions::new()`, but allows you to write more
248 /// readable code. Instead of
249 /// `OpenOptions::new().append(true).open("example.log")`,
250 /// you can write `File::options().append(true).open("example.log")`. This
251 /// also avoids the need to import `OpenOptions`.
252 ///
253 /// See the [`OpenOptions::new`] function for more details.
254 ///
255 /// # Examples
256 ///
257 /// ```no_run
258 /// use tokio::fs::File;
259 /// use tokio::io::AsyncWriteExt;
260 ///
261 /// # async fn dox() -> std::io::Result<()> {
262 /// let mut f = File::options().append(true).open("example.log").await?;
263 /// f.write_all(b"new line\n").await?;
264 /// # Ok(())
265 /// # }
266 /// ```
267 #[must_use]
268 pub fn options() -> OpenOptions {
269 OpenOptions::new()
270 }
271
272 /// Converts a [`std::fs::File`] to a [`tokio::fs::File`](File).
273 ///
274 /// # Examples
275 ///
276 /// ```no_run
277 /// // This line could block. It is not recommended to do this on the Tokio
278 /// // runtime.
279 /// let std_file = std::fs::File::open("foo.txt").unwrap();
280 /// let file = tokio::fs::File::from_std(std_file);
281 /// ```
282 pub fn from_std(std: StdFile) -> File {
283 File {
284 std: Arc::new(std),
285 inner: Mutex::new(Inner {
286 state: State::Idle(Some(Buf::with_capacity(0))),
287 last_write_err: None,
288 pos: 0,
289 }),
290 max_buf_size: DEFAULT_MAX_BUF_SIZE,
291 }
292 }
293
294 /// Attempts to sync all OS-internal metadata to disk.
295 ///
296 /// This function will attempt to ensure that all in-core data reaches the
297 /// filesystem before returning.
298 ///
299 /// # Examples
300 ///
301 /// ```no_run
302 /// use tokio::fs::File;
303 /// use tokio::io::AsyncWriteExt;
304 ///
305 /// # async fn dox() -> std::io::Result<()> {
306 /// let mut file = File::create("foo.txt").await?;
307 /// file.write_all(b"hello, world!").await?;
308 /// file.sync_all().await?;
309 /// # Ok(())
310 /// # }
311 /// ```
312 ///
313 /// The [`write_all`] method is defined on the [`AsyncWriteExt`] trait.
314 ///
315 /// [`write_all`]: fn@crate::io::AsyncWriteExt::write_all
316 /// [`AsyncWriteExt`]: trait@crate::io::AsyncWriteExt
317 pub async fn sync_all(&self) -> io::Result<()> {
318 let mut inner = self.inner.lock().await;
319 inner.complete_inflight().await;
320
321 let std = self.std.clone();
322 asyncify(move || std.sync_all()).await
323 }
324
325 /// This function is similar to `sync_all`, except that it may not
326 /// synchronize file metadata to the filesystem.
327 ///
328 /// This is intended for use cases that must synchronize content, but don't
329 /// need the metadata on disk. The goal of this method is to reduce disk
330 /// operations.
331 ///
332 /// Note that some platforms may simply implement this in terms of `sync_all`.
333 ///
334 /// # Examples
335 ///
336 /// ```no_run
337 /// use tokio::fs::File;
338 /// use tokio::io::AsyncWriteExt;
339 ///
340 /// # async fn dox() -> std::io::Result<()> {
341 /// let mut file = File::create("foo.txt").await?;
342 /// file.write_all(b"hello, world!").await?;
343 /// file.sync_data().await?;
344 /// # Ok(())
345 /// # }
346 /// ```
347 ///
348 /// The [`write_all`] method is defined on the [`AsyncWriteExt`] trait.
349 ///
350 /// [`write_all`]: fn@crate::io::AsyncWriteExt::write_all
351 /// [`AsyncWriteExt`]: trait@crate::io::AsyncWriteExt
352 pub async fn sync_data(&self) -> io::Result<()> {
353 let mut inner = self.inner.lock().await;
354 inner.complete_inflight().await;
355
356 let std = self.std.clone();
357 asyncify(move || std.sync_data()).await
358 }
359
360 /// Truncates or extends the underlying file, updating the size of this file to become size.
361 ///
362 /// If the size is less than the current file's size, then the file will be
363 /// shrunk. If it is greater than the current file's size, then the file
364 /// will be extended to size and have all of the intermediate data filled in
365 /// with 0s.
366 ///
367 /// # Errors
368 ///
369 /// This function will return an error if the file is not opened for
370 /// writing.
371 ///
372 /// # Examples
373 ///
374 /// ```no_run
375 /// use tokio::fs::File;
376 /// use tokio::io::AsyncWriteExt;
377 ///
378 /// # async fn dox() -> std::io::Result<()> {
379 /// let mut file = File::create("foo.txt").await?;
380 /// file.write_all(b"hello, world!").await?;
381 /// file.set_len(10).await?;
382 /// # Ok(())
383 /// # }
384 /// ```
385 ///
386 /// The [`write_all`] method is defined on the [`AsyncWriteExt`] trait.
387 ///
388 /// [`write_all`]: fn@crate::io::AsyncWriteExt::write_all
389 /// [`AsyncWriteExt`]: trait@crate::io::AsyncWriteExt
390 pub async fn set_len(&self, size: u64) -> io::Result<()> {
391 let mut inner = self.inner.lock().await;
392 inner.complete_inflight().await;
393
394 let mut buf = match inner.state {
395 State::Idle(ref mut buf_cell) => buf_cell.take().unwrap(),
396 _ => unreachable!(),
397 };
398
399 let seek = if !buf.is_empty() {
400 Some(SeekFrom::Current(buf.discard_read()))
401 } else {
402 None
403 };
404
405 let std = self.std.clone();
406
407 inner.state = State::Busy(spawn_blocking(move || {
408 let res = if let Some(seek) = seek {
409 (&*std).seek(seek).and_then(|_| std.set_len(size))
410 } else {
411 std.set_len(size)
412 }
413 .map(|()| 0); // the value is discarded later
414
415 // Return the result as a seek
416 (Operation::Seek(res), buf)
417 }));
418
419 let (op, buf) = match inner.state {
420 State::Idle(_) => unreachable!(),
421 State::Busy(ref mut rx) => {
422 let res = rx.await;
423 if res.is_err() {
424 // Restore a valid Idle state before returning the error.
425 inner.state = State::Idle(Some(Buf::with_capacity(0)));
426 }
427 res?
428 }
429 };
430
431 inner.state = State::Idle(Some(buf));
432
433 match op {
434 Operation::Seek(res) => res.map(|pos| {
435 inner.pos = pos;
436 }),
437 _ => unreachable!(),
438 }
439 }
440
441 /// Queries metadata about the underlying file.
442 ///
443 /// # Examples
444 ///
445 /// ```no_run
446 /// use tokio::fs::File;
447 ///
448 /// # async fn dox() -> std::io::Result<()> {
449 /// let file = File::open("foo.txt").await?;
450 /// let metadata = file.metadata().await?;
451 ///
452 /// println!("{:?}", metadata);
453 /// # Ok(())
454 /// # }
455 /// ```
456 pub async fn metadata(&self) -> io::Result<Metadata> {
457 let std = self.std.clone();
458 asyncify(move || std.metadata()).await
459 }
460
461 /// Creates a new `File` instance that shares the same underlying file handle
462 /// as the existing `File` instance. Reads, writes, and seeks will affect both
463 /// File instances simultaneously.
464 ///
465 /// # Examples
466 ///
467 /// ```no_run
468 /// use tokio::fs::File;
469 ///
470 /// # async fn dox() -> std::io::Result<()> {
471 /// let file = File::open("foo.txt").await?;
472 /// let file_clone = file.try_clone().await?;
473 /// # Ok(())
474 /// # }
475 /// ```
476 pub async fn try_clone(&self) -> io::Result<File> {
477 self.inner.lock().await.complete_inflight().await;
478 let std = self.std.clone();
479 let std_file = asyncify(move || std.try_clone()).await?;
480 let mut file = File::from_std(std_file);
481 file.set_max_buf_size(self.max_buf_size);
482 Ok(file)
483 }
484
485 /// Destructures `File` into a [`std::fs::File`]. This function is
486 /// async to allow any in-flight operations to complete.
487 ///
488 /// Use `File::try_into_std` to attempt conversion immediately.
489 ///
490 /// # Examples
491 ///
492 /// ```no_run
493 /// use tokio::fs::File;
494 ///
495 /// # async fn dox() -> std::io::Result<()> {
496 /// let tokio_file = File::open("foo.txt").await?;
497 /// let std_file = tokio_file.into_std().await;
498 /// # Ok(())
499 /// # }
500 /// ```
501 pub async fn into_std(mut self) -> StdFile {
502 self.inner.get_mut().complete_inflight().await;
503 Arc::try_unwrap(self.std).expect("Arc::try_unwrap failed")
504 }
505
506 /// Tries to immediately destructure `File` into a [`std::fs::File`].
507 ///
508 /// # Errors
509 ///
510 /// This function will return an error containing the file if some
511 /// operation is in-flight.
512 ///
513 /// # Examples
514 ///
515 /// ```no_run
516 /// use tokio::fs::File;
517 ///
518 /// # async fn dox() -> std::io::Result<()> {
519 /// let tokio_file = File::open("foo.txt").await?;
520 /// let std_file = tokio_file.try_into_std().unwrap();
521 /// # Ok(())
522 /// # }
523 /// ```
524 #[allow(clippy::result_large_err)]
525 pub fn try_into_std(mut self) -> Result<StdFile, Self> {
526 match Arc::try_unwrap(self.std) {
527 Ok(file) => Ok(file),
528 Err(std_file_arc) => {
529 self.std = std_file_arc;
530 Err(self)
531 }
532 }
533 }
534
535 /// Changes the permissions on the underlying file.
536 ///
537 /// # Platform-specific behavior
538 ///
539 /// This function currently corresponds to the `fchmod` function on Unix and
540 /// the `SetFileInformationByHandle` function on Windows. Note that, this
541 /// [may change in the future][changes].
542 ///
543 /// [changes]: https://doc.rust-lang.org/std/io/index.html#platform-specific-behavior
544 ///
545 /// # Errors
546 ///
547 /// This function will return an error if the user lacks permission change
548 /// attributes on the underlying file. It may also return an error in other
549 /// os-specific unspecified cases.
550 ///
551 /// # Examples
552 ///
553 /// ```no_run
554 /// use tokio::fs::File;
555 ///
556 /// # async fn dox() -> std::io::Result<()> {
557 /// let file = File::open("foo.txt").await?;
558 /// let mut perms = file.metadata().await?.permissions();
559 /// perms.set_readonly(true);
560 /// file.set_permissions(perms).await?;
561 /// # Ok(())
562 /// # }
563 /// ```
564 pub async fn set_permissions(&self, perm: Permissions) -> io::Result<()> {
565 let std = self.std.clone();
566 asyncify(move || std.set_permissions(perm)).await
567 }
568
569 /// Changes the timestamps on the underlying file.
570 ///
571 /// # Errors
572 ///
573 /// This function will return an error if the user lacks permission to change
574 /// timestamps on the underlying file. It may also return an error in other
575 /// os-specific unspecified cases.
576 ///
577 /// # Examples
578 ///
579 /// ```no_run
580 /// use tokio::fs::{File, FileTimes};
581 ///
582 /// # async fn dox() -> std::io::Result<()> {
583 /// let file = File::open("foo.txt").await?;
584 /// let times = FileTimes::new()
585 /// .set_accessed(std::time::SystemTime::now())
586 /// .set_modified(std::time::SystemTime::now());
587 /// file.set_times(times).await?;
588 /// # Ok(())
589 /// # }
590 /// ```
591 pub async fn set_times(&self, times: FileTimes) -> io::Result<()> {
592 // Wait for in-flight reads/writes like `sync_all` does, so they
593 // cannot complete after us and overwrite the timestamps.
594 self.inner.lock().await.complete_inflight().await;
595
596 let std = self.std.clone();
597 let std_times = times.into_std();
598 asyncify(move || std.set_times(std_times)).await
599 }
600
601 /// Changes the modification timestamp on the underlying file.
602 ///
603 /// This is equivalent to `set_times` with only the modification time set.
604 pub async fn set_modified(&self, time: std::time::SystemTime) -> io::Result<()> {
605 self.set_times(FileTimes::new().set_modified(time)).await
606 }
607
608 /// Set the maximum buffer size for the underlying [`AsyncRead`] / [`AsyncWrite`] operation.
609 ///
610 /// Although Tokio uses a sensible default value for this buffer size, this function would be
611 /// useful for changing that default depending on the situation.
612 ///
613 /// # Panics
614 ///
615 /// This function panics if `max_buf_size` is 0.
616 ///
617 /// # Examples
618 ///
619 /// ```no_run
620 /// use tokio::fs::File;
621 /// use tokio::io::AsyncWriteExt;
622 ///
623 /// # async fn dox() -> std::io::Result<()> {
624 /// let mut file = File::open("foo.txt").await?;
625 ///
626 /// // Set maximum buffer size to 8 MiB
627 /// file.set_max_buf_size(8 * 1024 * 1024);
628 ///
629 /// let mut buf = vec![1; 1024 * 1024 * 1024];
630 ///
631 /// // Write the 1 GiB buffer in chunks up to 8 MiB each.
632 /// file.write_all(&mut buf).await?;
633 /// # Ok(())
634 /// # }
635 /// ```
636 #[track_caller]
637 pub fn set_max_buf_size(&mut self, max_buf_size: usize) {
638 assert!(max_buf_size > 0, "`max_buf_size` must be greater than 0");
639 self.max_buf_size = max_buf_size;
640 }
641
642 /// Get the maximum buffer size for the underlying [`AsyncRead`] / [`AsyncWrite`] operation.
643 pub fn max_buf_size(&self) -> usize {
644 self.max_buf_size
645 }
646}
647
648impl AsyncRead for File {
649 fn poll_read(
650 self: Pin<&mut Self>,
651 cx: &mut Context<'_>,
652 dst: &mut ReadBuf<'_>,
653 ) -> Poll<io::Result<()>> {
654 ready!(crate::trace::trace_leaf(cx));
655
656 let me = self.get_mut();
657 let inner = me.inner.get_mut();
658
659 loop {
660 match inner.state {
661 State::Idle(ref mut buf_cell) => {
662 let mut buf = buf_cell.take().unwrap();
663
664 if !buf.is_empty() || dst.remaining() == 0 {
665 buf.copy_to(dst);
666 *buf_cell = Some(buf);
667 return Poll::Ready(Ok(()));
668 }
669
670 let std = me.std.clone();
671
672 let max_buf_size = cmp::min(dst.remaining(), me.max_buf_size);
673 inner.state = State::Busy(Inner::poll_read_inner(std, buf, max_buf_size)?);
674 }
675 State::Busy(ref mut rx) => {
676 let res = ready!(Pin::new(rx).poll(cx));
677 if res.is_err() {
678 // Restore a valid Idle state before returning the error.
679 inner.state = State::Idle(Some(Buf::with_capacity(0)));
680 }
681 let (op, mut buf) = res?;
682
683 match op {
684 Operation::Read(Ok(_)) => {
685 buf.copy_to(dst);
686 inner.state = State::Idle(Some(buf));
687 return Poll::Ready(Ok(()));
688 }
689 Operation::Read(Err(e)) => {
690 assert!(buf.is_empty());
691
692 inner.state = State::Idle(Some(buf));
693 return Poll::Ready(Err(e));
694 }
695 Operation::Write(Ok(())) => {
696 assert!(buf.is_empty());
697 inner.state = State::Idle(Some(buf));
698 continue;
699 }
700 Operation::Write(Err(e)) => {
701 assert!(inner.last_write_err.is_none());
702 inner.last_write_err = Some(e.kind());
703 inner.state = State::Idle(Some(buf));
704 }
705 Operation::Seek(result) => {
706 assert!(buf.is_empty());
707 inner.state = State::Idle(Some(buf));
708 if let Ok(pos) = result {
709 inner.pos = pos;
710 }
711 continue;
712 }
713 }
714 }
715 }
716 }
717 }
718}
719
720impl AsyncSeek for File {
721 fn start_seek(self: Pin<&mut Self>, mut pos: SeekFrom) -> io::Result<()> {
722 let me = self.get_mut();
723 let inner = me.inner.get_mut();
724
725 match inner.state {
726 State::Busy(_) => Err(io::Error::other(
727 "other file operation is pending, call poll_complete before start_seek",
728 )),
729 State::Idle(ref mut buf_cell) => {
730 let mut buf = buf_cell.take().unwrap();
731
732 // Factor in any unread data from the buf
733 let extra_seek = if !buf.is_empty() {
734 let n = buf.discard_read();
735
736 if let SeekFrom::Current(ref mut offset) = pos {
737 match offset.checked_add(n) {
738 Some(new_offset) => {
739 *offset = new_offset;
740 None
741 }
742 None => Some(SeekFrom::Current(n)),
743 }
744 } else {
745 None
746 }
747 } else {
748 None
749 };
750
751 let std = me.std.clone();
752
753 inner.state = State::Busy(spawn_blocking(move || {
754 let res = if let Some(extra_seek) = extra_seek {
755 (&*std).seek(extra_seek).and_then(|_| (&*std).seek(pos))
756 } else {
757 (&*std).seek(pos)
758 };
759 (Operation::Seek(res), buf)
760 }));
761 Ok(())
762 }
763 }
764 }
765
766 fn poll_complete(mut self: Pin<&mut Self>, cx: &mut Context<'_>) -> Poll<io::Result<u64>> {
767 ready!(crate::trace::trace_leaf(cx));
768 let inner = self.inner.get_mut();
769
770 loop {
771 match inner.state {
772 State::Idle(_) => return Poll::Ready(Ok(inner.pos)),
773 State::Busy(ref mut rx) => {
774 let res = ready!(Pin::new(rx).poll(cx));
775 if res.is_err() {
776 // Restore a valid Idle state before returning the error.
777 inner.state = State::Idle(Some(Buf::with_capacity(0)));
778 }
779 let (op, buf) = res?;
780 inner.state = State::Idle(Some(buf));
781
782 match op {
783 Operation::Read(_) => {}
784 Operation::Write(Err(e)) => {
785 assert!(inner.last_write_err.is_none());
786 inner.last_write_err = Some(e.kind());
787 }
788 Operation::Write(_) => {}
789 Operation::Seek(res) => {
790 if let Ok(pos) = res {
791 inner.pos = pos;
792 }
793 return Poll::Ready(res);
794 }
795 }
796 }
797 }
798 }
799 }
800}
801
802impl AsyncWrite for File {
803 fn poll_write(
804 self: Pin<&mut Self>,
805 cx: &mut Context<'_>,
806 src: &[u8],
807 ) -> Poll<io::Result<usize>> {
808 ready!(crate::trace::trace_leaf(cx));
809 let me = self.get_mut();
810 let inner = me.inner.get_mut();
811
812 if let Some(e) = inner.last_write_err.take() {
813 return Poll::Ready(Err(e.into()));
814 }
815
816 loop {
817 match inner.state {
818 State::Idle(ref mut buf_cell) => {
819 let mut buf = buf_cell.take().unwrap();
820
821 let seek = if !buf.is_empty() {
822 Some(SeekFrom::Current(buf.discard_read()))
823 } else {
824 None
825 };
826
827 let n = buf.copy_from(src, me.max_buf_size);
828 let std = me.std.clone();
829
830 let res = spawn_mandatory_blocking(move || {
831 let res = if let Some(seek) = seek {
832 (&*std).seek(seek).and_then(|_| buf.write_to(&mut &*std))
833 } else {
834 buf.write_to(&mut &*std)
835 };
836
837 (Operation::Write(res), buf)
838 })
839 .ok_or_else(|| io::Error::other("background task failed"));
840
841 if res.is_err() {
842 // Restore a valid Idle state before returning the error.
843 inner.state = State::Idle(Some(Buf::with_capacity(0)));
844 }
845 let blocking_task_join_handle = res?;
846
847 inner.state = State::Busy(blocking_task_join_handle);
848
849 return Poll::Ready(Ok(n));
850 }
851 State::Busy(ref mut rx) => {
852 let res = ready!(Pin::new(rx).poll(cx));
853 if res.is_err() {
854 // Restore a valid Idle state before returning the error.
855 inner.state = State::Idle(Some(Buf::with_capacity(0)));
856 }
857 let (op, buf) = res?;
858 inner.state = State::Idle(Some(buf));
859
860 match op {
861 Operation::Read(_) => {
862 // We don't care about the result here. The fact
863 // that the cursor has advanced will be reflected in
864 // the next iteration of the loop
865 continue;
866 }
867 Operation::Write(res) => {
868 // If the previous write was successful, continue.
869 // Otherwise, error.
870 res?;
871 continue;
872 }
873 Operation::Seek(_) => {
874 // Ignore the seek
875 continue;
876 }
877 }
878 }
879 }
880 }
881 }
882
883 fn poll_write_vectored(
884 self: Pin<&mut Self>,
885 cx: &mut Context<'_>,
886 bufs: &[io::IoSlice<'_>],
887 ) -> Poll<Result<usize, io::Error>> {
888 ready!(crate::trace::trace_leaf(cx));
889 let me = self.get_mut();
890 let inner = me.inner.get_mut();
891
892 if let Some(e) = inner.last_write_err.take() {
893 return Poll::Ready(Err(e.into()));
894 }
895
896 loop {
897 match inner.state {
898 State::Idle(ref mut buf_cell) => {
899 let mut buf = buf_cell.take().unwrap();
900
901 let seek = if !buf.is_empty() {
902 Some(SeekFrom::Current(buf.discard_read()))
903 } else {
904 None
905 };
906
907 let n = buf.copy_from_bufs(bufs, me.max_buf_size);
908 let std = me.std.clone();
909
910 let res = spawn_mandatory_blocking(move || {
911 let res = if let Some(seek) = seek {
912 (&*std).seek(seek).and_then(|_| buf.write_to(&mut &*std))
913 } else {
914 buf.write_to(&mut &*std)
915 };
916
917 (Operation::Write(res), buf)
918 })
919 .ok_or_else(|| io::Error::other("background task failed"));
920
921 if res.is_err() {
922 // Restore a valid Idle state before returning the error.
923 inner.state = State::Idle(Some(Buf::with_capacity(0)));
924 }
925 let blocking_task_join_handle = res?;
926
927 inner.state = State::Busy(blocking_task_join_handle);
928
929 return Poll::Ready(Ok(n));
930 }
931 State::Busy(ref mut rx) => {
932 let res = ready!(Pin::new(rx).poll(cx));
933 if res.is_err() {
934 // Restore a valid Idle state before returning the error.
935 inner.state = State::Idle(Some(Buf::with_capacity(0)));
936 }
937 let (op, buf) = res?;
938 inner.state = State::Idle(Some(buf));
939
940 match op {
941 Operation::Read(_) => {
942 // We don't care about the result here. The fact
943 // that the cursor has advanced will be reflected in
944 // the next iteration of the loop
945 continue;
946 }
947 Operation::Write(res) => {
948 // If the previous write was successful, continue.
949 // Otherwise, error.
950 res?;
951 continue;
952 }
953 Operation::Seek(_) => {
954 // Ignore the seek
955 continue;
956 }
957 }
958 }
959 }
960 }
961 }
962
963 fn is_write_vectored(&self) -> bool {
964 true
965 }
966
967 fn poll_flush(mut self: Pin<&mut Self>, cx: &mut Context<'_>) -> Poll<Result<(), io::Error>> {
968 ready!(crate::trace::trace_leaf(cx));
969 let inner = self.inner.get_mut();
970 inner.poll_flush(cx)
971 }
972
973 fn poll_shutdown(self: Pin<&mut Self>, cx: &mut Context<'_>) -> Poll<Result<(), io::Error>> {
974 ready!(crate::trace::trace_leaf(cx));
975 self.poll_flush(cx)
976 }
977}
978
979impl From<StdFile> for File {
980 fn from(std: StdFile) -> Self {
981 Self::from_std(std)
982 }
983}
984
985impl fmt::Debug for File {
986 fn fmt(&self, fmt: &mut fmt::Formatter<'_>) -> fmt::Result {
987 fmt.debug_struct("tokio::fs::File")
988 .field("std", &self.std)
989 .finish()
990 }
991}
992
993#[cfg(unix)]
994impl From<std::os::fd::OwnedFd> for File {
995 fn from(fd: std::os::fd::OwnedFd) -> Self {
996 Self::from_std(StdFile::from(fd))
997 }
998}
999
1000#[cfg(unix)]
1001impl std::os::unix::io::AsRawFd for File {
1002 fn as_raw_fd(&self) -> std::os::unix::io::RawFd {
1003 self.std.as_raw_fd()
1004 }
1005}
1006
1007#[cfg(unix)]
1008impl std::os::unix::io::AsFd for File {
1009 fn as_fd(&self) -> std::os::unix::io::BorrowedFd<'_> {
1010 unsafe {
1011 std::os::unix::io::BorrowedFd::borrow_raw(std::os::unix::io::AsRawFd::as_raw_fd(self))
1012 }
1013 }
1014}
1015
1016#[cfg(unix)]
1017impl std::os::unix::io::FromRawFd for File {
1018 unsafe fn from_raw_fd(fd: std::os::unix::io::RawFd) -> Self {
1019 // Safety: exactly the same safety contract as
1020 // `std::os::unix::io::FromRawFd::from_raw_fd`.
1021 unsafe { StdFile::from_raw_fd(fd).into() }
1022 }
1023}
1024
1025cfg_windows! {
1026 use crate::os::windows::io::{AsRawHandle, FromRawHandle, RawHandle, AsHandle, BorrowedHandle, OwnedHandle};
1027
1028 impl From<OwnedHandle> for File {
1029 fn from(handle: OwnedHandle) -> Self {
1030 Self::from_std(StdFile::from(handle))
1031 }
1032 }
1033
1034 impl AsRawHandle for File {
1035 fn as_raw_handle(&self) -> RawHandle {
1036 self.std.as_raw_handle()
1037 }
1038 }
1039
1040 impl AsHandle for File {
1041 fn as_handle(&self) -> BorrowedHandle<'_> {
1042 unsafe {
1043 BorrowedHandle::borrow_raw(
1044 AsRawHandle::as_raw_handle(self),
1045 )
1046 }
1047 }
1048 }
1049
1050 impl FromRawHandle for File {
1051 unsafe fn from_raw_handle(handle: RawHandle) -> Self {
1052 // Safety: exactly the same safety contract as
1053 // `FromRawHandle::from_raw_handle`.
1054 unsafe { StdFile::from_raw_handle(handle).into() }
1055 }
1056 }
1057}
1058
1059impl Inner {
1060 fn poll_read_inner(
1061 std: Arc<StdFile>,
1062 buf: Buf,
1063 max_buf_size: usize,
1064 ) -> io::Result<JoinHandle<(Operation, Buf)>> {
1065 // Unit tests use `MockFile` and the mock `spawn_blocking` infrastructure,
1066 // which can't drive real io_uring operations. The io_uring read path
1067 // is tested through integration tests in `tests/fs_uring_file_read.rs`.
1068 #[cfg(all(
1069 not(test),
1070 tokio_unstable,
1071 feature = "io-uring",
1072 feature = "rt",
1073 feature = "fs",
1074 target_os = "linux",
1075 ))]
1076 {
1077 if let Ok(handle) = crate::runtime::Handle::try_current() {
1078 if let Some(driver_handle) = handle.inner.driver().io.as_ref() {
1079 if driver_handle.is_uring_ready(io_uring::opcode::Read::CODE) {
1080 // Fast path: uring already initialized and Read supported.
1081 let fd: crate::io::uring::utils::ArcFd = std;
1082 return Ok(spawn(Self::uring_read(fd, buf, max_buf_size)));
1083 }
1084
1085 if !driver_handle.is_uring_probed() {
1086 // Not yet probed: lazy init inside an async task so
1087 // `File::from_std()` can still benefit from io-uring.
1088 return Ok(spawn(Self::lazy_init_read(std, buf, max_buf_size)));
1089 }
1090 }
1091 // No IO driver or probed but unsupported: fall through to spawn_blocking.
1092 }
1093 }
1094
1095 // Fallback: spawn_blocking
1096 let join = Self::spawn_blocking_read(buf, std, max_buf_size);
1097 Ok(join)
1098 }
1099
1100 /// Perform an io-uring read with interrupt retry.
1101 #[cfg(all(
1102 not(test),
1103 tokio_unstable,
1104 feature = "io-uring",
1105 feature = "rt",
1106 feature = "fs",
1107 target_os = "linux",
1108 ))]
1109 async fn uring_read(
1110 mut fd: crate::io::uring::utils::ArcFd,
1111 mut buf: Buf,
1112 max_buf_size: usize,
1113 ) -> (Operation, Buf) {
1114 use crate::runtime::driver::op::Op;
1115
1116 loop {
1117 let (res, r_fd, r_buf) =
1118 // u64::MAX to use and advance the file position
1119 Op::read_at(fd, buf, max_buf_size, u64::MAX).await;
1120 match (res, r_fd, r_buf) {
1121 // Retry an interrupted read with the resources the op handed back.
1122 (Err(e), Some(r_fd), Some(r_buf)) if e.kind() == io::ErrorKind::Interrupted => {
1123 buf = r_buf;
1124 fd = r_fd;
1125 continue;
1126 }
1127 // A cancelled op keeps its buffer in the driver, so the idle slot
1128 // gets a fresh empty `Buf` instead.
1129 (Err(e), _, r_buf) => {
1130 let buf = r_buf.unwrap_or_else(|| Buf::with_capacity(0));
1131 break (Operation::Read(Err(e)), buf);
1132 }
1133 (Ok(n), _, Some(r_buf)) => break (Operation::Read(Ok(n as usize)), r_buf),
1134 (Ok(_), ..) => unreachable!("a successful read returns its buffer"),
1135 }
1136 }
1137 }
1138
1139 /// Attempt lazy io-uring initialization, then read via uring or fall back
1140 /// to a blocking read. Covers the `File::from_std()` path where
1141 /// `check_and_init()` hasn't been called yet.
1142 #[cfg(all(
1143 not(test),
1144 tokio_unstable,
1145 feature = "io-uring",
1146 feature = "rt",
1147 feature = "fs",
1148 target_os = "linux",
1149 ))]
1150 async fn lazy_init_read(std: Arc<StdFile>, buf: Buf, max_buf_size: usize) -> (Operation, Buf) {
1151 if let Ok(handle) = crate::runtime::Handle::try_current() {
1152 if let Some(driver_handle) = handle.inner.driver().io.as_ref() {
1153 if driver_handle
1154 .check_and_init(io_uring::opcode::Read::CODE)
1155 .await
1156 .unwrap_or_default()
1157 {
1158 let fd: crate::io::uring::utils::ArcFd = std;
1159 return Self::uring_read(fd, buf, max_buf_size).await;
1160 }
1161 }
1162 }
1163
1164 match Self::spawn_blocking_read(buf, std, max_buf_size).await {
1165 Ok(result) => result,
1166 Err(e) => (
1167 Operation::Read(Err(io::Error::other(e))),
1168 Buf::with_capacity(0),
1169 ),
1170 }
1171 }
1172
1173 fn spawn_blocking_read(
1174 buf: Buf,
1175 std: Arc<StdFile>,
1176 max_buf_size: usize,
1177 ) -> JoinHandle<(Operation, Buf)> {
1178 spawn_blocking(move || {
1179 let mut buf = buf;
1180 // SAFETY: the `Read` implementation of `std` does not
1181 // read from the buffer it is borrowing and correctly
1182 // reports the length of the data written into the buffer.
1183 let res = unsafe { buf.read_from(&mut &*std, max_buf_size) };
1184 (Operation::Read(res), buf)
1185 })
1186 }
1187
1188 async fn complete_inflight(&mut self) {
1189 use std::future::poll_fn;
1190
1191 poll_fn(|cx| self.poll_complete_inflight(cx)).await;
1192 }
1193
1194 fn poll_complete_inflight(&mut self, cx: &mut Context<'_>) -> Poll<()> {
1195 ready!(crate::trace::trace_leaf(cx));
1196 match self.poll_flush(cx) {
1197 Poll::Ready(Err(e)) => {
1198 self.last_write_err = Some(e.kind());
1199 Poll::Ready(())
1200 }
1201 Poll::Ready(Ok(())) => Poll::Ready(()),
1202 Poll::Pending => Poll::Pending,
1203 }
1204 }
1205
1206 fn poll_flush(&mut self, cx: &mut Context<'_>) -> Poll<Result<(), io::Error>> {
1207 if let Some(e) = self.last_write_err.take() {
1208 return Poll::Ready(Err(e.into()));
1209 }
1210
1211 let (op, buf) = match self.state {
1212 State::Idle(_) => return Poll::Ready(Ok(())),
1213 State::Busy(ref mut rx) => {
1214 let res = ready!(Pin::new(rx).poll(cx));
1215 if res.is_err() {
1216 // Restore a valid Idle state before returning the error.
1217 self.state = State::Idle(Some(Buf::with_capacity(0)));
1218 }
1219 res?
1220 }
1221 };
1222
1223 // The buffer is not used here
1224 self.state = State::Idle(Some(buf));
1225
1226 match op {
1227 Operation::Read(_) => Poll::Ready(Ok(())),
1228 Operation::Write(res) => Poll::Ready(res),
1229 Operation::Seek(_) => Poll::Ready(Ok(())),
1230 }
1231 }
1232}
1233
1234#[cfg(test)]
1235mod tests;