diff --git a/tokio/src/fs/file.rs b/tokio/src/fs/file.rs index a172d2eee..d0275535b 100644 --- a/tokio/src/fs/file.rs +++ b/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(|| { - io::Error::new(io::ErrorKind::Other, "background task failed") - })?; + .ok_or_else(|| 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(|| { - io::Error::new(io::ErrorKind::Other, "background task failed") - })?; + .ok_or_else(|| 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 diff --git a/tokio/tests/fs_file_join_error.rs b/tokio/tests/fs_file_join_error.rs new file mode 100644 index 000000000..20b1c86cc --- /dev/null +++ b/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(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]); +}