fs: restore File internal state when op fails (#8291)

tokio/src/fs/file.rs

@@ -418,7 +418,14 @@ impl File {
 
         let (op, buf) = match inner.state {
             State::Idle(_) => unreachable!(),
-            State::Busy(ref mut rx) => rx.await?,
+            State::Busy(ref mut rx) => {
+                let res = rx.await;
+                if res.is_err() {
+                    // Restore a valid Idle state before returning the error.
+                    inner.state = State::Idle(Some(Buf::with_capacity(0)));
+                }
+                res?
+            }
         };
 
         inner.state = State::Idle(Some(buf));
@@ -621,7 +628,12 @@ impl AsyncRead for File {
                     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))?;
+                    let res = ready!(Pin::new(rx).poll(cx));
+                    if res.is_err() {
+                        // Restore a valid Idle state before returning the error.
+                        inner.state = State::Idle(Some(Buf::with_capacity(0)));
+                    }
+                    let (op, mut buf) = res?;
 
                     match op {
                         Operation::Read(Ok(_)) => {
@@ -701,7 +713,12 @@ impl AsyncSeek for File {
             match inner.state {
                 State::Idle(_) => return Poll::Ready(Ok(inner.pos)),
                 State::Busy(ref mut rx) => {
-                    let (op, buf) = ready!(Pin::new(rx).poll(cx))?;
+                    let res = ready!(Pin::new(rx).poll(cx));
+                    if res.is_err() {
+                        // Restore a valid Idle state before returning the error.
+                        inner.state = State::Idle(Some(Buf::with_capacity(0)));
+                    }
+                    let (op, buf) = res?;
                     inner.state = State::Idle(Some(buf));
 
                     match op {
@@ -752,7 +769,7 @@ impl AsyncWrite for File {
                     let n = buf.copy_from(src, me.max_buf_size);
                     let std = me.std.clone();
 
-                    let blocking_task_join_handle = spawn_mandatory_blocking(move || {
+                    let res = spawn_mandatory_blocking(move || {
                         let res = if let Some(seek) = seek {
                             (&*std).seek(seek).and_then(|_| buf.write_to(&mut &*std))
                         } else {
@@ -761,16 +778,25 @@ impl AsyncWrite for File {
 
                         (Operation::Write(res), buf)
                     })
-                    .ok_or_else(|| {
+                    .ok_or_else(|| io::Error::new(io::ErrorKind::Other, "background task failed"));
-                        io::Error::new(io::ErrorKind::Other, "background task failed")
+
-                    })?;
+                    if res.is_err() {
+                        // Restore a valid Idle state before returning the error.
+                        inner.state = State::Idle(Some(Buf::with_capacity(0)));
+                    }
+                    let blocking_task_join_handle = res?;
 
                     inner.state = State::Busy(blocking_task_join_handle);
 
                     return Poll::Ready(Ok(n));
                 }
                 State::Busy(ref mut rx) => {
-                    let (op, buf) = ready!(Pin::new(rx).poll(cx))?;
+                    let res = ready!(Pin::new(rx).poll(cx));
+                    if res.is_err() {
+                        // Restore a valid Idle state before returning the error.
+                        inner.state = State::Idle(Some(Buf::with_capacity(0)));
+                    }
+                    let (op, buf) = res?;
                     inner.state = State::Idle(Some(buf));
 
                     match op {
@@ -823,7 +849,7 @@ impl AsyncWrite for File {
                     let n = buf.copy_from_bufs(bufs, me.max_buf_size);
                     let std = me.std.clone();
 
-                    let blocking_task_join_handle = spawn_mandatory_blocking(move || {
+                    let res = spawn_mandatory_blocking(move || {
                         let res = if let Some(seek) = seek {
                             (&*std).seek(seek).and_then(|_| buf.write_to(&mut &*std))
                         } else {
@@ -832,16 +858,25 @@ impl AsyncWrite for File {
 
                         (Operation::Write(res), buf)
                     })
-                    .ok_or_else(|| {
+                    .ok_or_else(|| io::Error::new(io::ErrorKind::Other, "background task failed"));
-                        io::Error::new(io::ErrorKind::Other, "background task failed")
+
-                    })?;
+                    if res.is_err() {
+                        // Restore a valid Idle state before returning the error.
+                        inner.state = State::Idle(Some(Buf::with_capacity(0)));
+                    }
+                    let blocking_task_join_handle = res?;
 
                     inner.state = State::Busy(blocking_task_join_handle);
 
                     return Poll::Ready(Ok(n));
                 }
                 State::Busy(ref mut rx) => {
-                    let (op, buf) = ready!(Pin::new(rx).poll(cx))?;
+                    let res = ready!(Pin::new(rx).poll(cx));
+                    if res.is_err() {
+                        // Restore a valid Idle state before returning the error.
+                        inner.state = State::Idle(Some(Buf::with_capacity(0)));
+                    }
+                    let (op, buf) = res?;
                     inner.state = State::Idle(Some(buf));
 
                     match op {
@@ -1108,7 +1143,14 @@ impl Inner {
 
         let (op, buf) = match self.state {
             State::Idle(_) => return Poll::Ready(Ok(())),
-            State::Busy(ref mut rx) => ready!(Pin::new(rx).poll(cx))?,
+            State::Busy(ref mut rx) => {
+                let res = ready!(Pin::new(rx).poll(cx));
+                if res.is_err() {
+                    // Restore a valid Idle state before returning the error.
+                    self.state = State::Idle(Some(Buf::with_capacity(0)));
+                }
+                res?
+            }
         };
 
         // The buffer is not used here

tokio/tests/fs_file_join_error.rs

@@ -0,0 +1,68 @@
+#![warn(rust_2018_idioms)]
+#![cfg(feature = "full")]
+
+//! A `tokio::fs::File` op whose backing blocking task fails (cancelled/panicked
+//! -> `JoinError`, or fails to spawn) must leave the File in a valid state and
+//! return an `io::Error`, not poison it so a later op panics.
+//!
+//! We run each op on a runtime whose blocking pool is already shut down, that way
+//! the task is cancelled on spawn. A poison panics here and fails the test.
+
+use std::future::Future;
+use std::io::SeekFrom::Start;
+use std::path::Path;
+use std::task::Context;
+use std::time::Duration;
+
+use tokio::fs::File;
+use tokio::io::{AsyncReadExt, AsyncSeekExt, AsyncWriteExt};
+use tokio::runtime::{Builder, Handle};
+
+fn poll_once<F: Future>(fut: F) {
+    let mut fut = Box::pin(fut);
+    let _ = fut
+        .as_mut()
+        .poll(&mut Context::from_waker(futures::task::noop_waker_ref()));
+}
+
+/// A handle to a runtime whose blocking pool has already been shut down.
+fn dead_pool_handle() -> Handle {
+    let rt = Builder::new_multi_thread().enable_all().build().unwrap();
+    let h = rt.handle().clone();
+    rt.shutdown_timeout(Duration::from_secs(0));
+    h
+}
+
+/// Run `ops` in order on a fresh dead-pool File.
+fn run(path: &Path, h: &Handle, ops: &[fn(&mut File)]) {
+    let _enter = h.enter();
+    let opts = std::fs::File::options()
+        .read(true)
+        .write(true)
+        .open(path)
+        .unwrap();
+    let mut file = File::from_std(opts);
+    for &op in ops {
+        op(&mut file);
+    }
+}
+
+#[test]
+fn file_ops_survive_join_error() {
+    let path = std::env::temp_dir().join(format!("tokio_join_error_{}", std::process::id()));
+    std::fs::write(&path, vec![b'x'; 4096]).unwrap();
+    let h = dead_pool_handle();
+
+    let read: fn(&mut File) = |f| poll_once(f.read(&mut [0u8; 8]));
+    let seek: fn(&mut File) = |f| poll_once(f.seek(Start(0)));
+    let write: fn(&mut File) = |f| poll_once(f.write(b"z"));
+    let flush: fn(&mut File) = |f| poll_once(f.flush());
+    let set_len: fn(&mut File) = |f| poll_once(f.set_len(0));
+
+    // Each op fails, however none may panic.
+    run(&path, &h, &[read, read]);
+    run(&path, &h, &[seek, seek]);
+    run(&path, &h, &[set_len, set_len]);
+    run(&path, &h, &[write, write]);
+    run(&path, &h, &[read, flush]);
+}