From 08e20fcf6a23cbffb7c70f44f7d968bd70b35463 Mon Sep 17 00:00:00 2001 From: Carl Lerche Date: Tue, 27 Aug 2019 12:25:20 -0700 Subject: [PATCH] fs: add support for non-threadpool executors (#1495) Provides a thread pool dedicated to running blocking operations (#588) and update `tokio-fs` to use this pool. In an effort to make incremental progress, this is an initial step towards a final solution. First, it provides a very basic pool implementation with the intend that the pool will be replaced before the final release. Second, it updates `tokio-fs` to always use this blocking pool instead of conditionally using `threadpool::blocking`. Issue #588 contains additional discussion around potential improvements to the "blocking for all" strategy. The implementation provided here builds on work started in #954 and continued in #1045. The general idea is th same as #1045, but the PR improves on some of the details: * The number of explicit operations tracked by `File` is reduced only to the ones that could interact. All other ops are spawned on the blocking pool without being tracked by the `File` instance. * The `seek` implementation is not backed by a trait and `poll_seek` function. This avoids the question of how to model non-blocking seeks on top of a blocking file. In this patch, `seek` is represented as an `async fn`. If the associated future is dropped before the caller observes the return value, we make no effort to define the state in which the file ends up. --- tokio-executor/Cargo.toml | 1 + tokio-executor/src/blocking.rs | 142 +++++ tokio-executor/src/lib.rs | 3 + tokio-fs/Cargo.toml | 7 +- tokio-fs/src/blocking.rs | 252 ++++++++ tokio-fs/src/create_dir.rs | 3 +- tokio-fs/src/create_dir_all.rs | 3 +- tokio-fs/src/file.rs | 334 +++++++++-- tokio-fs/src/hard_link.rs | 5 +- tokio-fs/src/lib.rs | 36 +- tokio-fs/src/metadata.rs | 5 +- tokio-fs/src/open_options.rs | 12 +- tokio-fs/src/os/unix/symlink.rs | 5 +- tokio-fs/src/os/windows/symlink_dir.rs | 5 +- tokio-fs/src/os/windows/symlink_file.rs | 5 +- tokio-fs/src/read.rs | 14 +- tokio-fs/src/read_dir.rs | 65 ++- tokio-fs/src/read_link.rs | 3 +- tokio-fs/src/remove_dir.rs | 3 +- tokio-fs/src/remove_dir_all.rs | 3 +- tokio-fs/src/remove_file.rs | 3 +- tokio-fs/src/rename.rs | 5 +- tokio-fs/src/set_permissions.rs | 3 +- tokio-fs/src/stderr.rs | 25 +- tokio-fs/src/stdin.rs | 14 +- tokio-fs/src/stdout.rs | 25 +- tokio-fs/src/symlink_metadata.rs | 3 +- tokio-fs/src/write.rs | 10 +- tokio-fs/tests/dir.rs | 68 +-- tokio-fs/tests/file.rs | 43 ++ tokio-fs/tests/file_mocked.rs | 736 ++++++++++++++++++++++++ tokio-fs/tests/link.rs | 47 +- tokio-fs/tests/pool/mod.rs | 18 - tokio-fs/tests/sys/file.rs | 265 +++++++++ tokio-fs/tests/sys/pool.rs | 66 +++ 35 files changed, 1974 insertions(+), 263 deletions(-) create mode 100644 tokio-executor/src/blocking.rs create mode 100644 tokio-fs/src/blocking.rs create mode 100644 tokio-fs/tests/file_mocked.rs delete mode 100644 tokio-fs/tests/pool/mod.rs create mode 100644 tokio-fs/tests/sys/file.rs create mode 100644 tokio-fs/tests/sys/pool.rs diff --git a/tokio-executor/Cargo.toml b/tokio-executor/Cargo.toml index 7b579f322..f9089b1e5 100644 --- a/tokio-executor/Cargo.toml +++ b/tokio-executor/Cargo.toml @@ -21,6 +21,7 @@ keywords = ["futures", "tokio"] categories = ["concurrency", "asynchronous"] [features] +blocking = ["tokio-sync"] current-thread = ["crossbeam-channel"] threadpool = [ "tokio-sync", diff --git a/tokio-executor/src/blocking.rs b/tokio-executor/src/blocking.rs new file mode 100644 index 000000000..1c589c741 --- /dev/null +++ b/tokio-executor/src/blocking.rs @@ -0,0 +1,142 @@ +//! Thread pool for blocking operations + +use tokio_sync::oneshot; + +use lazy_static::lazy_static; +use std::collections::VecDeque; +use std::future::Future; +use std::pin::Pin; +use std::sync::{Condvar, Mutex}; +use std::task::{Context, Poll}; +use std::thread; +use std::time::Duration; + +struct Pool { + shared: Mutex, + condvar: Condvar, +} + +struct Shared { + queue: VecDeque>, + num_th: u32, + num_idle: u32, +} + +lazy_static! { + static ref POOL: Pool = Pool::new(); +} + +const MAX_THREADS: u32 = 1_000; +const KEEP_ALIVE: Duration = Duration::from_secs(10); + +/// Result of a blocking operation running on the blocking thread pool. +#[derive(Debug)] +pub struct Blocking { + rx: oneshot::Receiver, +} + +/// Run the provided function on a threadpool dedicated to blocking operations. +pub fn run(f: F) -> Blocking +where + F: FnOnce() -> R + Send + 'static, + R: Send + 'static, +{ + let (tx, rx) = oneshot::channel(); + + let should_spawn = { + let mut shared = POOL.shared.lock().unwrap(); + + shared.queue.push_back(Box::new(move || { + // The receiver may have dropped + let _ = tx.send(f()); + })); + + if shared.num_idle == 0 { + // No threads are able to process the task + + if shared.num_th == MAX_THREADS { + // At max number of threads + false + } else { + shared.num_th += 1; + true + } + } else { + shared.num_idle -= 1; + POOL.condvar.notify_one(); + false + } + }; + + if should_spawn { + spawn_thread(); + } + + Blocking { rx } +} + +impl Future for Blocking { + type Output = T; + + fn poll(mut self: Pin<&mut Self>, cx: &mut Context<'_>) -> Poll { + use std::task::Poll::*; + + match Pin::new(&mut self.rx).poll(cx) { + Ready(Ok(v)) => Ready(v), + Ready(Err(_)) => panic!( + "the blocking operation has been dropped before completing. \ + This should not happen and is a bug." + ), + Pending => Pending, + } + } +} + +fn spawn_thread() { + thread::Builder::new() + .name("tokio-blocking-driver".to_string()) + .spawn(|| { + 'outer: loop { + let mut shared = POOL.shared.lock().unwrap(); + + if let Some(task) = shared.queue.pop_front() { + drop(shared); + run_task(task); + continue; + } + + // IDLE + shared.num_idle += 1; + + loop { + shared = POOL.condvar.wait_timeout(shared, KEEP_ALIVE).unwrap().0; + + if let Some(task) = shared.queue.pop_front() { + drop(shared); + run_task(task); + continue 'outer; + } + } + } + }) + .unwrap(); +} + +fn run_task(f: Box) { + use std::panic::{catch_unwind, AssertUnwindSafe}; + + let _ = catch_unwind(AssertUnwindSafe(|| f())); +} + +impl Pool { + fn new() -> Pool { + Pool { + shared: Mutex::new(Shared { + queue: VecDeque::new(), + num_th: 0, + num_idle: 0, + }), + condvar: Condvar::new(), + } + } +} diff --git a/tokio-executor/src/lib.rs b/tokio-executor/src/lib.rs index e867af4d0..bfdd0fe62 100644 --- a/tokio-executor/src/lib.rs +++ b/tokio-executor/src/lib.rs @@ -67,6 +67,9 @@ mod global; pub mod park; mod typed; +#[cfg(feature = "blocking")] +pub mod blocking; + #[cfg(feature = "current-thread")] pub mod current_thread; diff --git a/tokio-fs/Cargo.toml b/tokio-fs/Cargo.toml index 32f088009..7f499d158 100644 --- a/tokio-fs/Cargo.toml +++ b/tokio-fs/Cargo.toml @@ -23,13 +23,16 @@ categories = ["asynchronous", "network-programming", "filesystem"] [dependencies] tokio-io = { version = "=0.2.0-alpha.2", features = ["util"], path = "../tokio-io" } -tokio-executor = { version = "=0.2.0-alpha.2", features = ["threadpool"], path = "../tokio-executor" } +tokio-executor = { version = "=0.2.0-alpha.2", features = ["blocking"], path = "../tokio-executor" } +tokio-sync = { version = "=0.2.0-alpha.2", path = "../tokio-sync" } futures-core-preview = "=0.3.0-alpha.18" futures-util-preview = "=0.3.0-alpha.18" +lazy_static = "1.3.0" [dev-dependencies] -tokio = { version = "0.2.0-alpha.1", path = "../tokio" } +tokio = { version = "=0.2.0-alpha.2", path = "../tokio" } +tokio-test = { version = "=0.2.0-alpha.2", path = "../tokio-test" } rand = "0.7" tempfile = "3" diff --git a/tokio-fs/src/blocking.rs b/tokio-fs/src/blocking.rs new file mode 100644 index 000000000..57a982e8c --- /dev/null +++ b/tokio-fs/src/blocking.rs @@ -0,0 +1,252 @@ +use crate::sys; + +use tokio_io::{AsyncRead, AsyncWrite}; + +use futures_core::ready; +use std::cmp; +use std::future::Future; +use std::io; +use std::io::prelude::*; +use std::pin::Pin; +use std::task::Poll::*; +use std::task::{Context, Poll}; + +use self::State::*; + +/// `T` should not implement _both_ Read and Write. +#[derive(Debug)] +pub(crate) struct Blocking { + inner: Option, + state: State, +} + +#[derive(Debug)] +pub(crate) struct Buf { + buf: Vec, + pos: usize, +} + +pub(crate) const MAX_BUF: usize = 16 * 1024; + +#[derive(Debug)] +enum State { + Idle(Option), + Busy(sys::Blocking<(io::Result, Buf, T)>), +} + +impl Blocking { + pub(crate) fn new(inner: T) -> Blocking { + Blocking { + inner: Some(inner), + state: State::Idle(Some(Buf::with_capacity(0))), + } + } +} + +impl AsyncRead for Blocking +where + T: Read + Unpin + Send + 'static, +{ + fn poll_read( + mut self: Pin<&mut Self>, + cx: &mut Context<'_>, + dst: &mut [u8], + ) -> Poll> { + loop { + match self.state { + Idle(ref mut buf_cell) => { + let mut buf = buf_cell.take().unwrap(); + + if !buf.is_empty() { + let n = buf.copy_to(dst); + *buf_cell = Some(buf); + return Ready(Ok(n)); + } + + buf.ensure_capacity_for(dst); + let mut inner = self.inner.take().unwrap(); + + self.state = Busy(sys::run(move || { + let res = buf.read_from(&mut inner); + (res, buf, inner) + })); + } + Busy(ref mut rx) => { + let (res, mut buf, inner) = ready!(Pin::new(rx).poll(cx)); + self.inner = Some(inner); + + match res { + Ok(_) => { + let n = buf.copy_to(dst); + self.state = Idle(Some(buf)); + return Ready(Ok(n)); + } + Err(e) => { + assert!(buf.is_empty()); + + self.state = Idle(Some(buf)); + return Ready(Err(e)); + } + } + } + } + } + } +} + +impl AsyncWrite for Blocking +where + T: Write + Unpin + Send + 'static, +{ + fn poll_write( + mut self: Pin<&mut Self>, + cx: &mut Context<'_>, + src: &[u8], + ) -> Poll> { + loop { + match self.state { + Idle(ref mut buf_cell) => { + let mut buf = buf_cell.take().unwrap(); + + assert!(buf.is_empty()); + + let n = buf.copy_from(src); + let mut inner = self.inner.take().unwrap(); + + self.state = Busy(sys::run(move || { + let n = buf.len(); + let res = buf.write_to(&mut inner).map(|_| n); + + (res, buf, inner) + })); + + return Ready(Ok(n)); + } + Busy(ref mut rx) => { + let (res, buf, inner) = ready!(Pin::new(rx).poll(cx)); + self.state = Idle(Some(buf)); + self.inner = Some(inner); + + // If error, return + res?; + } + } + } + } + + fn poll_flush(mut self: Pin<&mut Self>, cx: &mut Context<'_>) -> Poll> { + let (res, buf, inner) = match self.state { + Idle(_) => return Ready(Ok(())), + Busy(ref mut rx) => ready!(Pin::new(rx).poll(cx)), + }; + + // The buffer is not used here + self.state = Idle(Some(buf)); + self.inner = Some(inner); + + Ready(res.map(|_| ())) + } + + fn poll_shutdown(self: Pin<&mut Self>, _cx: &mut Context<'_>) -> Poll> { + Poll::Ready(Ok(())) + } +} + +/// Repeates operations that are interrupted +macro_rules! uninterruptibly { + ($e:expr) => {{ + loop { + match $e { + Err(ref e) if e.kind() == io::ErrorKind::Interrupted => {} + res => break res, + } + } + }}; +} + +impl Buf { + pub(crate) fn with_capacity(n: usize) -> Buf { + Buf { + buf: Vec::with_capacity(n), + pos: 0, + } + } + + pub(crate) fn is_empty(&self) -> bool { + self.len() == 0 + } + + pub(crate) fn len(&self) -> usize { + self.buf.len() - self.pos + } + + pub(crate) fn copy_to(&mut self, dst: &mut [u8]) -> usize { + let n = cmp::min(self.len(), dst.len()); + dst[..n].copy_from_slice(&self.bytes()[..n]); + self.pos += n; + + if self.pos == self.buf.len() { + self.buf.truncate(0); + self.pos = 0; + } + + n + } + + pub(crate) fn copy_from(&mut self, src: &[u8]) -> usize { + assert!(self.is_empty()); + + let n = cmp::min(src.len(), MAX_BUF); + + self.buf.extend_from_slice(&src[..n]); + n + } + + pub(crate) fn bytes(&self) -> &[u8] { + &self.buf[self.pos..] + } + + pub(crate) fn ensure_capacity_for(&mut self, bytes: &[u8]) { + assert!(self.is_empty()); + + let len = cmp::min(bytes.len(), MAX_BUF); + + if self.buf.len() < len { + self.buf.reserve(len - self.buf.len()); + } + + unsafe { + self.buf.set_len(len); + } + } + + pub(crate) fn read_from(&mut self, rd: &mut T) -> io::Result { + let res = uninterruptibly!(rd.read(&mut self.buf)); + + if let Ok(n) = res { + self.buf.truncate(n); + } else { + self.buf.clear(); + } + + assert_eq!(self.pos, 0); + + res + } + + pub(crate) fn write_to(&mut self, wr: &mut T) -> io::Result<()> { + assert_eq!(self.pos, 0); + + // `write_all` already ignores interrupts + let res = wr.write_all(&self.buf); + self.buf.clear(); + res + } + + pub(crate) fn discard_read(&mut self) -> i64 { + let ret = -(self.bytes().len() as i64); + self.pos = 0; + self.buf.truncate(0); + ret + } +} diff --git a/tokio-fs/src/create_dir.rs b/tokio-fs/src/create_dir.rs index 19133c2d9..73a2bf4e1 100644 --- a/tokio-fs/src/create_dir.rs +++ b/tokio-fs/src/create_dir.rs @@ -9,5 +9,6 @@ use std::path::Path; /// /// [std]: https://doc.rust-lang.org/std/fs/fn.create_dir.html pub async fn create_dir>(path: P) -> io::Result<()> { - asyncify(|| std::fs::create_dir(&path)).await + let path = path.as_ref().to_owned(); + asyncify(move || std::fs::create_dir(path)).await } diff --git a/tokio-fs/src/create_dir_all.rs b/tokio-fs/src/create_dir_all.rs index 3bbcbabdd..f9cb3e5fa 100644 --- a/tokio-fs/src/create_dir_all.rs +++ b/tokio-fs/src/create_dir_all.rs @@ -10,5 +10,6 @@ use std::path::Path; /// /// [std]: https://doc.rust-lang.org/std/fs/fn.create_dir_all.html pub async fn create_dir_all>(path: P) -> io::Result<()> { - asyncify(|| std::fs::create_dir_all(&path)).await + let path = path.as_ref().to_owned(); + asyncify(move || std::fs::create_dir_all(path)).await } diff --git a/tokio-fs/src/file.rs b/tokio-fs/src/file.rs index 238dd34fc..828a4adb4 100644 --- a/tokio-fs/src/file.rs +++ b/tokio-fs/src/file.rs @@ -2,17 +2,23 @@ //! //! [`File`]: file/struct.File.html -use crate::{asyncify, blocking_io, OpenOptions}; +use self::State::*; +use crate::blocking::Buf; +use crate::{asyncify, sys}; use tokio_io::{AsyncRead, AsyncWrite}; -use std::convert::TryFrom; +use futures_core::ready; +use std::fmt; use std::fs::{Metadata, Permissions}; -use std::io::{self, Read, Seek, Write}; +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::Context; use std::task::Poll; +use std::task::Poll::*; /// A reference to an open file on the filesystem. /// @@ -58,9 +64,27 @@ use std::task::Poll; /// # Ok(()) /// # } /// ``` -#[derive(Debug)] pub struct File { - std: std::fs::File, + std: Arc, + 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, +} + +#[derive(Debug)] +enum State { + Idle(Option), + Busy(sys::Blocking<(Operation, Buf)>), +} + +#[derive(Debug)] +enum Operation { + Read(io::Result), + Write(io::Result<()>), + Seek(io::Result), } impl File { @@ -94,12 +118,12 @@ impl File { /// ``` pub async fn open

(path: P) -> io::Result where - P: AsRef + Send + Unpin + 'static, + P: AsRef, { - let mut open_options = OpenOptions::new(); - open_options.read(true); + let path = path.as_ref().to_owned(); + let std = asyncify(|| sys::File::open(path)).await?; - open_options.open(path).await + Ok(File::from_std(std)) } /// Opens a file in write-only mode. @@ -132,9 +156,10 @@ impl File { /// ``` pub async fn create

(path: P) -> io::Result where - P: AsRef + Send + Unpin + 'static, + P: AsRef, { - let std_file = asyncify(|| std::fs::File::create(&path)).await?; + let path = path.as_ref().to_owned(); + let std_file = asyncify(move || sys::File::create(path)).await?; Ok(File::from_std(std_file)) } @@ -151,8 +176,12 @@ impl File { /// let std_file = std::fs::File::open("foo.txt").unwrap(); /// let file = tokio::fs::File::from_std(std_file); /// ``` - pub fn from_std(std: std::fs::File) -> File { - File { std } + pub fn from_std(std: sys::File) -> File { + File { + std: Arc::new(std), + state: State::Idle(Some(Buf::with_capacity(0))), + last_write_err: None, + } } /// Seek to an offset, in bytes, in a stream. @@ -174,8 +203,42 @@ impl File { /// # Ok(()) /// # } /// ``` - pub async fn seek(&mut self, pos: io::SeekFrom) -> io::Result { - asyncify(|| self.std.seek(pos)).await + pub async fn seek(&mut self, mut pos: SeekFrom) -> io::Result { + self.complete_inflight().await; + + let mut buf = match self.state { + Idle(ref mut buf_cell) => buf_cell.take().unwrap(), + _ => unreachable!(), + }; + + // Factor in any unread data from the buf + if !buf.is_empty() { + let n = buf.discard_read(); + + if let SeekFrom::Current(ref mut offset) = pos { + *offset += n; + } + } + + let std = self.std.clone(); + + // Start the operation + self.state = Busy(sys::run(move || { + let res = (&*std).seek(pos); + (Operation::Seek(res), buf) + })); + + let (op, buf) = match self.state { + Idle(_) => unreachable!(), + Busy(ref mut rx) => rx.await, + }; + + self.state = Idle(Some(buf)); + + match op { + Operation::Seek(res) => res, + _ => unreachable!(), + } } /// Attempts to sync all OS-internal metadata to disk. @@ -197,7 +260,10 @@ impl File { /// # } /// ``` pub async fn sync_all(&mut self) -> io::Result<()> { - asyncify(|| self.std.sync_all()).await + self.complete_inflight().await; + + let std = self.std.clone(); + asyncify(move || std.sync_all()).await } /// This function is similar to `poll_sync_all`, except that it may not @@ -223,7 +289,10 @@ impl File { /// # } /// ``` pub async fn sync_data(&mut self) -> io::Result<()> { - asyncify(|| self.std.sync_data()).await + self.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. @@ -252,7 +321,44 @@ impl File { /// # } /// ``` pub async fn set_len(&mut self, size: u64) -> io::Result<()> { - asyncify(|| self.std.set_len(size)).await + self.complete_inflight().await; + + let mut buf = match self.state { + Idle(ref mut buf_cell) => buf_cell.take().unwrap(), + _ => unreachable!(), + }; + + let seek = if !buf.is_empty() { + Some(SeekFrom::Current(buf.discard_read())) + } else { + None + }; + + let std = self.std.clone(); + + self.state = Busy(sys::run(move || { + let res = if let 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 self.state { + Idle(_) => unreachable!(), + Busy(ref mut rx) => rx.await, + }; + + self.state = Idle(Some(buf)); + + match op { + Operation::Seek(res) => res.map(|_| ()), + _ => unreachable!(), + } } /// Queries metadata about the underlying file. @@ -271,7 +377,8 @@ impl File { /// # } /// ``` pub async fn metadata(&self) -> io::Result { - asyncify(|| self.std.metadata()).await + let std = self.std.clone(); + asyncify(move || std.metadata()).await } /// Create a new `File` instance that shares the same underlying file handle @@ -290,7 +397,8 @@ impl File { /// # } /// ``` pub async fn try_clone(&self) -> io::Result { - let std_file = asyncify(|| self.std.try_clone()).await?; + let std = self.std.clone(); + let std_file = asyncify(move || std.try_clone()).await?; Ok(File::from_std(std_file)) } @@ -324,54 +432,162 @@ impl File { /// # } /// ``` pub async fn set_permissions(&self, perm: Permissions) -> io::Result<()> { - asyncify(|| self.std.set_permissions(perm)).await + let std = self.std.clone(); + asyncify(move || std.set_permissions(perm)).await } - /// Destructures the `tokio_fs::File` into a [`std::fs::File`][std]. - /// - /// # Panics - /// - /// This function will panic if `shutdown` has been called. - /// - /// [std]: https://doc.rust-lang.org/std/fs/struct.File.html - /// - /// # Examples - /// - /// ```no_run - /// use tokio::fs::File; - /// - /// # async fn dox() -> std::io::Result<()> { - /// let file = File::create("foo.txt").await?; - /// let std_file = file.into_std(); - /// # Ok(()) - /// # } - /// ``` - pub fn into_std(self) -> std::fs::File { - self.std + async fn complete_inflight(&mut self) { + use futures_util::future::poll_fn; + + if let Err(e) = poll_fn(|cx| Pin::new(&mut *self).poll_flush(cx)).await { + self.last_write_err = Some(e.kind()); + } } } impl AsyncRead for File { fn poll_read( - self: Pin<&mut Self>, - _cx: &mut Context<'_>, - buf: &mut [u8], + mut self: Pin<&mut Self>, + cx: &mut Context<'_>, + dst: &mut [u8], ) -> Poll> { - blocking_io(|| (&self.std).read(buf)) + loop { + match self.state { + Idle(ref mut buf_cell) => { + let mut buf = buf_cell.take().unwrap(); + + if !buf.is_empty() { + let n = buf.copy_to(dst); + *buf_cell = Some(buf); + return Ready(Ok(n)); + } + + buf.ensure_capacity_for(dst); + let std = self.std.clone(); + + self.state = Busy(sys::run(move || { + let res = buf.read_from(&mut &*std); + (Operation::Read(res), buf) + })); + } + Busy(ref mut rx) => { + let (op, mut buf) = ready!(Pin::new(rx).poll(cx)); + + match op { + Operation::Read(Ok(_)) => { + let n = buf.copy_to(dst); + self.state = Idle(Some(buf)); + return Ready(Ok(n)); + } + Operation::Read(Err(e)) => { + assert!(buf.is_empty()); + + self.state = Idle(Some(buf)); + return Ready(Err(e)); + } + Operation::Write(Ok(_)) => { + assert!(buf.is_empty()); + self.state = Idle(Some(buf)); + continue; + } + Operation::Write(Err(e)) => { + assert!(self.last_write_err.is_none()); + self.last_write_err = Some(e.kind()); + self.state = Idle(Some(buf)); + } + Operation::Seek(_) => { + assert!(buf.is_empty()); + self.state = Idle(Some(buf)); + continue; + } + } + } + } + } } } impl AsyncWrite for File { fn poll_write( - self: Pin<&mut Self>, - _cx: &mut Context<'_>, - buf: &[u8], + mut self: Pin<&mut Self>, + cx: &mut Context<'_>, + src: &[u8], ) -> Poll> { - blocking_io(|| (&self.std).write(buf)) + if let Some(e) = self.last_write_err.take() { + return Ready(Err(e.into())); + } + + loop { + match self.state { + Idle(ref mut buf_cell) => { + let mut buf = buf_cell.take().unwrap(); + + let seek = if !buf.is_empty() { + Some(SeekFrom::Current(buf.discard_read())) + } else { + None + }; + + let n = buf.copy_from(src); + let std = self.std.clone(); + + self.state = Busy(sys::run(move || { + let res = if let Some(seek) = seek { + (&*std).seek(seek).and_then(|_| buf.write_to(&mut &*std)) + } else { + buf.write_to(&mut &*std) + }; + + (Operation::Write(res), buf) + })); + + return Ready(Ok(n)); + } + Busy(ref mut rx) => { + let (op, buf) = ready!(Pin::new(rx).poll(cx)); + self.state = Idle(Some(buf)); + + 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_flush(self: Pin<&mut Self>, _cx: &mut Context<'_>) -> Poll> { - blocking_io(|| (&self.std).flush()) + fn poll_flush(mut self: Pin<&mut Self>, cx: &mut Context<'_>) -> Poll> { + if let Some(e) = self.last_write_err.take() { + return Ready(Err(e.into())); + } + + let (op, buf) = match self.state { + Idle(_) => return Ready(Ok(())), + Busy(ref mut rx) => ready!(Pin::new(rx).poll(cx)), + }; + + // The buffer is not used here + self.state = Idle(Some(buf)); + + match op { + Operation::Read(_) => Ready(Ok(())), + Operation::Write(res) => Ready(res), + Operation::Seek(_) => Ready(Ok(())), + } } fn poll_shutdown(self: Pin<&mut Self>, _cx: &mut Context<'_>) -> Poll> { @@ -379,16 +595,16 @@ impl AsyncWrite for File { } } -impl From for File { - fn from(std: std::fs::File) -> Self { +impl From for File { + fn from(std: sys::File) -> Self { Self::from_std(std) } } -impl TryFrom for std::fs::File { - type Error = io::Error; - - fn try_from(file: File) -> Result { - Ok(file.std) +impl fmt::Debug for File { + fn fmt(&self, fmt: &mut fmt::Formatter<'_>) -> fmt::Result { + fmt.debug_struct("tokio::fs::File") + .field("std", &self.std) + .finish() } } diff --git a/tokio-fs/src/hard_link.rs b/tokio-fs/src/hard_link.rs index 3d1fe82a4..ff6869410 100644 --- a/tokio-fs/src/hard_link.rs +++ b/tokio-fs/src/hard_link.rs @@ -12,5 +12,8 @@ use std::path::Path; /// /// [std]: https://doc.rust-lang.org/std/fs/fn.hard_link.html pub async fn hard_link, Q: AsRef>(src: P, dst: Q) -> io::Result<()> { - asyncify(|| std::fs::hard_link(&src, &dst)).await + let src = src.as_ref().to_owned(); + let dst = dst.as_ref().to_owned(); + + asyncify(move || std::fs::hard_link(src, dst)).await } diff --git a/tokio-fs/src/lib.rs b/tokio-fs/src/lib.rs index efd07e9c5..46bfa15da 100644 --- a/tokio-fs/src/lib.rs +++ b/tokio-fs/src/lib.rs @@ -34,6 +34,7 @@ //! [`AsyncRead`]: https://docs.rs/tokio-io/0.1/tokio_io/trait.AsyncRead.html //! [tokio-executor]: https://docs.rs/tokio-executor/0.2.0-alpha.2/tokio_executor/threadpool/index.html +mod blocking; mod create_dir; mod create_dir_all; mod file; @@ -76,37 +77,18 @@ pub use crate::symlink_metadata::symlink_metadata; pub use crate::write::write; use std::io; -use std::io::ErrorKind::Other; -use std::task::Poll; -use std::task::Poll::*; - -fn blocking_io(f: F) -> Poll> -where - F: FnOnce() -> io::Result, -{ - use tokio_executor::threadpool::blocking; - - match blocking(f) { - Ready(Ok(v)) => Ready(v), - Ready(Err(_)) => Ready(Err(blocking_err())), - Pending => Pending, - } -} async fn asyncify(f: F) -> io::Result where - F: FnOnce() -> io::Result, + F: FnOnce() -> io::Result + Send + 'static, + T: Send + 'static, { - use futures_util::future::poll_fn; - - let mut f = Some(f); - poll_fn(move |_| blocking_io(|| f.take().unwrap()())).await + sys::run(f).await } -fn blocking_err() -> io::Error { - io::Error::new( - Other, - "`blocking` annotated I/O must be called \ - from the context of the Tokio runtime.", - ) +/// Types in this module can be mocked out in tests. +mod sys { + pub(crate) use std::fs::File; + + pub(crate) use tokio_executor::blocking::{run, Blocking}; } diff --git a/tokio-fs/src/metadata.rs b/tokio-fs/src/metadata.rs index 145c92e0c..668f12a56 100644 --- a/tokio-fs/src/metadata.rs +++ b/tokio-fs/src/metadata.rs @@ -7,7 +7,8 @@ use std::path::Path; /// Queries the file system metadata for a path. pub async fn metadata

(path: P) -> io::Result where - P: AsRef + Send + 'static, + P: AsRef, { - asyncify(|| std::fs::metadata(&path)).await + let path = path.as_ref().to_owned(); + asyncify(|| std::fs::metadata(path)).await } diff --git a/tokio-fs/src/open_options.rs b/tokio-fs/src/open_options.rs index 85b6f1db5..438c0cef2 100644 --- a/tokio-fs/src/open_options.rs +++ b/tokio-fs/src/open_options.rs @@ -1,6 +1,5 @@ -use super::File; +use crate::{asyncify, File}; -use futures_util::future::poll_fn; use std::io; use std::path::Path; @@ -91,10 +90,13 @@ impl OpenOptions { /// [`open`]: https://doc.rust-lang.org/std/fs/struct.OpenOptions.html#method.open pub async fn open

(&self, path: P) -> io::Result where - P: AsRef + Send + Unpin + 'static, + P: AsRef, { - let std_file = poll_fn(|_| crate::blocking_io(|| self.0.open(&path))).await?; - Ok(File::from_std(std_file)) + let path = path.as_ref().to_owned(); + let opts = self.0.clone(); + + let std = asyncify(move || opts.open(path)).await?; + Ok(File::from_std(std)) } } diff --git a/tokio-fs/src/os/unix/symlink.rs b/tokio-fs/src/os/unix/symlink.rs index bfc8a6cfd..b546578ac 100644 --- a/tokio-fs/src/os/unix/symlink.rs +++ b/tokio-fs/src/os/unix/symlink.rs @@ -11,5 +11,8 @@ use std::path::Path; /// /// [std]: https://doc.rust-lang.org/std/os/unix/fs/fn.symlink.html pub async fn symlink, Q: AsRef>(src: P, dst: Q) -> io::Result<()> { - asyncify(|| std::os::unix::fs::symlink(&src, &dst)).await + let src = src.as_ref().to_owned(); + let dst = dst.as_ref().to_owned(); + + asyncify(move || std::os::unix::fs::symlink(src, dst)).await } diff --git a/tokio-fs/src/os/windows/symlink_dir.rs b/tokio-fs/src/os/windows/symlink_dir.rs index d5f72ad3d..8d8e0b434 100644 --- a/tokio-fs/src/os/windows/symlink_dir.rs +++ b/tokio-fs/src/os/windows/symlink_dir.rs @@ -12,5 +12,8 @@ use std::path::Path; /// /// [std]: https://doc.rust-lang.org/std/os/windows/fs/fn.symlink_dir.html pub async fn symlink_dir, Q: AsRef>(src: P, dst: Q) -> io::Result<()> { - asyncify(|| std::os::windows::fs::symlink_dir(&src, &dst)).await + let src = src.as_ref().to_owned(); + let dst = dst.as_ref().to_owned(); + + asyncify(move || std::os::windows::fs::symlink_dir(src, dst)).await } diff --git a/tokio-fs/src/os/windows/symlink_file.rs b/tokio-fs/src/os/windows/symlink_file.rs index 7373355df..c6d9ad802 100644 --- a/tokio-fs/src/os/windows/symlink_file.rs +++ b/tokio-fs/src/os/windows/symlink_file.rs @@ -12,5 +12,8 @@ use std::path::Path; /// /// [std]: https://doc.rust-lang.org/std/os/windows/fs/fn.symlink_file.html pub async fn symlink_file, Q: AsRef>(src: P, dst: Q) -> io::Result<()> { - asyncify(|| std::os::windows::fs::symlink_file(&src, &dst)).await + let src = src.as_ref().to_owned(); + let dst = dst.as_ref().to_owned(); + + asyncify(move || std::os::windows::fs::symlink_file(src, dst)).await } diff --git a/tokio-fs/src/read.rs b/tokio-fs/src/read.rs index 3f496fd6a..21092aaf5 100644 --- a/tokio-fs/src/read.rs +++ b/tokio-fs/src/read.rs @@ -1,6 +1,4 @@ -use crate::File; - -use tokio_io::AsyncReadExt; +use crate::asyncify; use std::{io, path::Path}; @@ -22,12 +20,8 @@ use std::{io, path::Path}; /// ``` pub async fn read

(path: P) -> io::Result> where - P: AsRef + Send + Unpin + 'static, + P: AsRef, { - let mut file = File::open(path).await?; - let metadata = file.metadata().await?; - - let mut contents = Vec::with_capacity(metadata.len() as usize + 1); - file.read_to_end(&mut contents).await?; - Ok(contents) + let path = path.as_ref().to_owned(); + asyncify(move || std::fs::read(path)).await } diff --git a/tokio-fs/src/read_dir.rs b/tokio-fs/src/read_dir.rs index 393574b1c..6428dbda2 100644 --- a/tokio-fs/src/read_dir.rs +++ b/tokio-fs/src/read_dir.rs @@ -1,13 +1,16 @@ -use crate::{asyncify, blocking_io}; +use crate::{asyncify, sys}; +use futures_core::ready; use futures_core::stream::Stream; use std::ffi::OsString; -use std::fs::{DirEntry as StdDirEntry, FileType, Metadata}; +use std::fs::{FileType, Metadata}; +use std::future::Future; use std::io; #[cfg(unix)] use std::os::unix::fs::DirEntryExt; use std::path::{Path, PathBuf}; use std::pin::Pin; +use std::sync::Arc; use std::task::Context; use std::task::Poll; @@ -20,8 +23,10 @@ pub async fn read_dir

(path: P) -> io::Result where P: AsRef + Send + 'static, { - let std = asyncify(|| std::fs::read_dir(&path)).await?; - Ok(ReadDir(std)) + let path = path.as_ref().to_owned(); + let std = asyncify(|| std::fs::read_dir(path)).await?; + + Ok(ReadDir(State::Idle(Some(std)))) } /// Stream of the entries in a directory. @@ -42,22 +47,37 @@ where /// [`Err`]: https://doc.rust-lang.org/std/result/enum.Result.html#variant.Err #[derive(Debug)] #[must_use = "streams do nothing unless polled"] -pub struct ReadDir(std::fs::ReadDir); +pub struct ReadDir(State); + +#[derive(Debug)] +enum State { + Idle(Option), + Pending(sys::Blocking<(Option>, std::fs::ReadDir)>), +} impl Stream for ReadDir { type Item = io::Result; - fn poll_next(mut self: Pin<&mut Self>, _cx: &mut Context<'_>) -> Poll> { - let res = blocking_io(|| match self.0.next() { - Some(Err(err)) => Err(err), - Some(Ok(item)) => Ok(Some(Ok(DirEntry(item)))), - None => Ok(None), - }); + fn poll_next(mut self: Pin<&mut Self>, cx: &mut Context<'_>) -> Poll> { + loop { + match self.0 { + State::Idle(ref mut std) => { + let mut std = std.take().unwrap(); - match res { - Poll::Ready(Err(err)) => Poll::Ready(Some(Err(err))), - Poll::Ready(Ok(v)) => Poll::Ready(v), - Poll::Pending => Poll::Pending, + self.0 = State::Pending(sys::run(move || { + let ret = std.next(); + (ret, std) + })); + } + State::Pending(ref mut rx) => { + let (ret, std) = ready!(Pin::new(rx).poll(cx)); + self.0 = State::Idle(Some(std)); + + let ret = ret.map(|res| res.map(|std| DirEntry(Arc::new(std)))); + + return Poll::Ready(ret); + } + } } } } @@ -75,16 +95,9 @@ impl Stream for ReadDir { /// /// [std]: https://doc.rust-lang.org/std/fs/struct.DirEntry.html #[derive(Debug)] -pub struct DirEntry(StdDirEntry); +pub struct DirEntry(Arc); impl DirEntry { - /// Destructures the `tokio_fs::DirEntry` into a [`std::fs::DirEntry`][std]. - /// - /// [std]: https://doc.rust-lang.org/std/fs/struct.DirEntry.html - pub fn into_std(self) -> StdDirEntry { - self.0 - } - /// Returns the full path to the file that this entry represents. /// /// The full path is created by joining the original path to `read_dir` @@ -177,7 +190,8 @@ impl DirEntry { /// # } /// ``` pub async fn metadata(&self) -> io::Result { - asyncify(|| self.0.metadata()).await + let std = self.0.clone(); + asyncify(move || std.metadata()).await } /// Return the file type for the file that this entry points at. @@ -214,7 +228,8 @@ impl DirEntry { /// # } /// ``` pub async fn file_type(&self) -> io::Result { - asyncify(|| self.0.file_type()).await + let std = self.0.clone(); + asyncify(move || std.file_type()).await } } diff --git a/tokio-fs/src/read_link.rs b/tokio-fs/src/read_link.rs index ac070abe0..1cec06d59 100644 --- a/tokio-fs/src/read_link.rs +++ b/tokio-fs/src/read_link.rs @@ -9,5 +9,6 @@ use std::path::{Path, PathBuf}; /// /// [std]: https://doc.rust-lang.org/std/fs/fn.read_link.html pub async fn read_link>(path: P) -> io::Result { - asyncify(|| std::fs::read_link(&path)).await + let path = path.as_ref().to_owned(); + asyncify(move || std::fs::read_link(path)).await } diff --git a/tokio-fs/src/remove_dir.rs b/tokio-fs/src/remove_dir.rs index 0de7e33b6..853e60b25 100644 --- a/tokio-fs/src/remove_dir.rs +++ b/tokio-fs/src/remove_dir.rs @@ -9,5 +9,6 @@ use std::path::Path; /// /// [std]: https://doc.rust-lang.org/std/fs/fn.remove_dir.html pub async fn remove_dir>(path: P) -> io::Result<()> { - asyncify(|| std::fs::remove_dir(&path)).await + let path = path.as_ref().to_owned(); + asyncify(move || std::fs::remove_dir(path)).await } diff --git a/tokio-fs/src/remove_dir_all.rs b/tokio-fs/src/remove_dir_all.rs index bd50697e1..3a1d8be18 100644 --- a/tokio-fs/src/remove_dir_all.rs +++ b/tokio-fs/src/remove_dir_all.rs @@ -9,5 +9,6 @@ use std::path::Path; /// /// [std]: https://doc.rust-lang.org/std/fs/fn.remove_dir_all.html pub async fn remove_dir_all>(path: P) -> io::Result<()> { - asyncify(|| std::fs::remove_dir_all(&path)).await + let path = path.as_ref().to_owned(); + asyncify(move || std::fs::remove_dir_all(path)).await } diff --git a/tokio-fs/src/remove_file.rs b/tokio-fs/src/remove_file.rs index b5c2216c5..634e7d0b7 100644 --- a/tokio-fs/src/remove_file.rs +++ b/tokio-fs/src/remove_file.rs @@ -13,5 +13,6 @@ use std::path::Path; /// /// [std]: https://doc.rust-lang.org/std/fs/fn.remove_file.html pub async fn remove_file>(path: P) -> io::Result<()> { - asyncify(|| std::fs::remove_file(&path)).await + let path = path.as_ref().to_owned(); + asyncify(move || std::fs::remove_file(path)).await } diff --git a/tokio-fs/src/rename.rs b/tokio-fs/src/rename.rs index e0f138aab..2429d7b87 100644 --- a/tokio-fs/src/rename.rs +++ b/tokio-fs/src/rename.rs @@ -12,5 +12,8 @@ use std::path::Path; /// /// [std]: https://doc.rust-lang.org/std/fs/fn.rename.html pub async fn rename, Q: AsRef>(from: P, to: Q) -> io::Result<()> { - asyncify(|| std::fs::rename(&from, &to)).await + let from = from.as_ref().to_owned(); + let to = to.as_ref().to_owned(); + + asyncify(move || std::fs::rename(from, to)).await } diff --git a/tokio-fs/src/set_permissions.rs b/tokio-fs/src/set_permissions.rs index cd11d2d70..9c663d44a 100644 --- a/tokio-fs/src/set_permissions.rs +++ b/tokio-fs/src/set_permissions.rs @@ -10,5 +10,6 @@ use std::path::Path; /// /// [std]: https://doc.rust-lang.org/std/fs/fn.set_permissions.html pub async fn set_permissions>(path: P, perm: Permissions) -> io::Result<()> { - asyncify(|| std::fs::set_permissions(&path, perm)).await + let path = path.as_ref().to_owned(); + asyncify(|| std::fs::set_permissions(path, perm)).await } diff --git a/tokio-fs/src/stderr.rs b/tokio-fs/src/stderr.rs index d16132b6e..c0e071700 100644 --- a/tokio-fs/src/stderr.rs +++ b/tokio-fs/src/stderr.rs @@ -1,8 +1,8 @@ -use crate::blocking_io; +use crate::blocking::Blocking; use tokio_io::AsyncWrite; -use std::io::{self, Write}; +use std::io; use std::pin::Pin; use std::task::Context; use std::task::Poll; @@ -18,7 +18,7 @@ use std::task::Poll; /// [`AsyncWrite`]: trait.AsyncWrite.html #[derive(Debug)] pub struct Stderr { - std: std::io::Stderr, + std: Blocking, } /// Constructs a new handle to the standard error of the current process. @@ -27,23 +27,28 @@ pub struct Stderr { /// Tokio runtime. pub fn stderr() -> Stderr { let std = io::stderr(); - Stderr { std } + Stderr { + std: Blocking::new(std), + } } impl AsyncWrite for Stderr { fn poll_write( mut self: Pin<&mut Self>, - _cx: &mut Context<'_>, + cx: &mut Context<'_>, buf: &[u8], ) -> Poll> { - blocking_io(|| (&mut self.std).write(buf)) + Pin::new(&mut self.std).poll_write(cx, buf) } - fn poll_flush(mut self: Pin<&mut Self>, _cx: &mut Context<'_>) -> Poll> { - blocking_io(|| (&mut self.std).flush()) + fn poll_flush(mut self: Pin<&mut Self>, cx: &mut Context<'_>) -> Poll> { + Pin::new(&mut self.std).poll_flush(cx) } - fn poll_shutdown(self: Pin<&mut Self>, _cx: &mut Context<'_>) -> Poll> { - Poll::Ready(Ok(())) + fn poll_shutdown( + mut self: Pin<&mut Self>, + cx: &mut Context<'_>, + ) -> Poll> { + Pin::new(&mut self.std).poll_shutdown(cx) } } diff --git a/tokio-fs/src/stdin.rs b/tokio-fs/src/stdin.rs index aa09bd9aa..a0fea9814 100644 --- a/tokio-fs/src/stdin.rs +++ b/tokio-fs/src/stdin.rs @@ -1,8 +1,8 @@ -use crate::blocking_io; +use crate::blocking::Blocking; use tokio_io::AsyncRead; -use std::io::{self, Read}; +use std::io; use std::pin::Pin; use std::task::Context; use std::task::Poll; @@ -24,7 +24,7 @@ use std::task::Poll; /// [`AsyncRead`]: trait.AsyncRead.html #[derive(Debug)] pub struct Stdin { - std: std::io::Stdin, + std: Blocking, } /// Constructs a new handle to the standard input of the current process. @@ -33,15 +33,17 @@ pub struct Stdin { /// Tokio runtime. pub fn stdin() -> Stdin { let std = io::stdin(); - Stdin { std } + Stdin { + std: Blocking::new(std), + } } impl AsyncRead for Stdin { fn poll_read( mut self: Pin<&mut Self>, - _cx: &mut Context<'_>, + cx: &mut Context<'_>, buf: &mut [u8], ) -> Poll> { - blocking_io(|| (&mut self.std).read(buf)) + Pin::new(&mut self.std).poll_read(cx, buf) } } diff --git a/tokio-fs/src/stdout.rs b/tokio-fs/src/stdout.rs index 0c96e960f..04bd8796c 100644 --- a/tokio-fs/src/stdout.rs +++ b/tokio-fs/src/stdout.rs @@ -1,8 +1,8 @@ -use crate::blocking_io; +use crate::blocking::Blocking; use tokio_io::AsyncWrite; -use std::io::{self, Write}; +use std::io; use std::pin::Pin; use std::task::Context; use std::task::Poll; @@ -18,7 +18,7 @@ use std::task::Poll; /// [`AsyncWrite`]: trait.AsyncWrite.html #[derive(Debug)] pub struct Stdout { - std: std::io::Stdout, + std: Blocking, } /// Constructs a new handle to the standard output of the current process. @@ -27,23 +27,28 @@ pub struct Stdout { /// runtime. pub fn stdout() -> Stdout { let std = io::stdout(); - Stdout { std } + Stdout { + std: Blocking::new(std), + } } impl AsyncWrite for Stdout { fn poll_write( mut self: Pin<&mut Self>, - _cx: &mut Context<'_>, + cx: &mut Context<'_>, buf: &[u8], ) -> Poll> { - blocking_io(|| (&mut self.std).write(buf)) + Pin::new(&mut self.std).poll_write(cx, buf) } - fn poll_flush(mut self: Pin<&mut Self>, _cx: &mut Context<'_>) -> Poll> { - blocking_io(|| (&mut self.std).flush()) + fn poll_flush(mut self: Pin<&mut Self>, cx: &mut Context<'_>) -> Poll> { + Pin::new(&mut self.std).poll_flush(cx) } - fn poll_shutdown(self: Pin<&mut Self>, _cx: &mut Context<'_>) -> Poll> { - Poll::Ready(Ok(())) + fn poll_shutdown( + mut self: Pin<&mut Self>, + cx: &mut Context<'_>, + ) -> Poll> { + Pin::new(&mut self.std).poll_shutdown(cx) } } diff --git a/tokio-fs/src/symlink_metadata.rs b/tokio-fs/src/symlink_metadata.rs index 73fd910f9..411572d15 100644 --- a/tokio-fs/src/symlink_metadata.rs +++ b/tokio-fs/src/symlink_metadata.rs @@ -13,5 +13,6 @@ pub async fn symlink_metadata

(path: P) -> io::Result where P: AsRef + Send + 'static, { - asyncify(|| std::fs::symlink_metadata(&path)).await + let path = path.as_ref().to_owned(); + asyncify(|| std::fs::symlink_metadata(path)).await } diff --git a/tokio-fs/src/write.rs b/tokio-fs/src/write.rs index 82e1607a3..9eba86354 100644 --- a/tokio-fs/src/write.rs +++ b/tokio-fs/src/write.rs @@ -1,6 +1,4 @@ -use crate::File; - -use tokio_io::AsyncWriteExt; +use crate::asyncify; use std::{io, path::Path}; @@ -23,8 +21,8 @@ pub async fn write + Unpin>(path: P, contents: C) -> io::Resul where P: AsRef + Send + Unpin + 'static, { - let mut file = File::create(path).await?; - file.write_all(contents.as_ref()).await?; + let path = path.as_ref().to_owned(); + let contents = contents.as_ref().to_owned(); - Ok(()) + asyncify(move || std::fs::write(path, contents)).await } diff --git a/tokio-fs/tests/dir.rs b/tokio-fs/tests/dir.rs index 6d7f79778..cb1da22d8 100644 --- a/tokio-fs/tests/dir.rs +++ b/tokio-fs/tests/dir.rs @@ -1,83 +1,69 @@ #![warn(rust_2018_idioms)] +use tokio::fs; +use tokio_test::assert_ok; + use futures_util::future; use futures_util::try_stream::TryStreamExt; -use std::fs; use std::sync::{Arc, Mutex}; use tempfile::tempdir; -use tokio_fs::*; -mod pool; - -#[test] -fn create() { +#[tokio::test] +async fn create_dir() { let base_dir = tempdir().unwrap(); let new_dir = base_dir.path().join("foo"); let new_dir_2 = new_dir.clone(); - pool::run(async move { - create_dir(new_dir).await?; - Ok(()) - }); + assert_ok!(fs::create_dir(new_dir).await); assert!(new_dir_2.is_dir()); } -#[test] -fn create_all() { +#[tokio::test] +async fn create_all() { let base_dir = tempdir().unwrap(); let new_dir = base_dir.path().join("foo").join("bar"); let new_dir_2 = new_dir.clone(); - pool::run(async move { - create_dir_all(new_dir).await?; - Ok(()) - }); - + assert_ok!(fs::create_dir_all(new_dir).await); assert!(new_dir_2.is_dir()); } -#[test] -fn remove() { +#[tokio::test] +async fn remove() { let base_dir = tempdir().unwrap(); let new_dir = base_dir.path().join("foo"); let new_dir_2 = new_dir.clone(); - fs::create_dir(new_dir.clone()).unwrap(); - - pool::run(async move { - remove_dir(new_dir).await?; - Ok(()) - }); + std::fs::create_dir(new_dir.clone()).unwrap(); + assert_ok!(fs::remove_dir(new_dir).await); assert!(!new_dir_2.exists()); } -#[test] -fn read() { +#[tokio::test] +async fn read() { let base_dir = tempdir().unwrap(); let p = base_dir.path(); - fs::create_dir(p.join("aa")).unwrap(); - fs::create_dir(p.join("bb")).unwrap(); - fs::create_dir(p.join("cc")).unwrap(); + std::fs::create_dir(p.join("aa")).unwrap(); + std::fs::create_dir(p.join("bb")).unwrap(); + std::fs::create_dir(p.join("cc")).unwrap(); let files = Arc::new(Mutex::new(Vec::new())); let f = files.clone(); let p = p.to_path_buf(); - pool::run(async move { - let read_dir_fut = read_dir(p).await?; - read_dir_fut - .try_for_each(move |e| { - let s = e.file_name().to_str().unwrap().to_string(); - f.lock().unwrap().push(s); - future::ok(()) - }) - .await?; - Ok(()) - }); + let read_dir_fut = fs::read_dir(p).await.unwrap(); + read_dir_fut + .try_for_each(move |e| { + let s = e.file_name().to_str().unwrap().to_string(); + f.lock().unwrap().push(s); + future::ok(()) + }) + .await + .unwrap(); let mut files = files.lock().unwrap(); files.sort(); // because the order is not guaranteed diff --git a/tokio-fs/tests/file.rs b/tokio-fs/tests/file.rs index 05e186f46..db26becef 100644 --- a/tokio-fs/tests/file.rs +++ b/tokio-fs/tests/file.rs @@ -1,12 +1,54 @@ #![warn(rust_2018_idioms)] +use tokio::fs::File; +use tokio::prelude::*; + +use std::io::prelude::*; +use tempfile::NamedTempFile; + +/* use rand::{distributions, thread_rng, Rng}; use std::fs; use std::io::SeekFrom; use tempfile::Builder as TmpBuilder; use tokio::io::{AsyncReadExt, AsyncWriteExt}; use tokio_fs::*; +*/ +const HELLO: &[u8] = b"hello world..."; + +#[tokio::test] +async fn basic_read() { + let mut tempfile = tempfile(); + tempfile.write_all(HELLO).unwrap(); + + let mut file = File::open(tempfile.path()).await.unwrap(); + + let mut buf = [0; 1024]; + let n = file.read(&mut buf).await.unwrap(); + + assert_eq!(n, HELLO.len()); + assert_eq!(&buf[..n], HELLO); +} + +#[tokio::test] +async fn basic_write() { + let tempfile = tempfile(); + + let mut file = File::create(tempfile.path()).await.unwrap(); + + file.write_all(HELLO).await.unwrap(); + file.flush().await.unwrap(); + + let file = std::fs::read(tempfile.path()).unwrap(); + assert_eq!(file, HELLO); +} + +fn tempfile() -> NamedTempFile { + NamedTempFile::new().unwrap() +} + +/* mod pool; #[test] @@ -154,3 +196,4 @@ fn clone() { assert_eq!(dst, b"clone successful") } +*/ diff --git a/tokio-fs/tests/file_mocked.rs b/tokio-fs/tests/file_mocked.rs new file mode 100644 index 000000000..422588760 --- /dev/null +++ b/tokio-fs/tests/file_mocked.rs @@ -0,0 +1,736 @@ +#![warn(rust_2018_idioms)] + +mod sys { + mod file; + pub(crate) mod pool; + + pub(crate) use self::file::File; + pub(crate) use self::pool::{run, Blocking}; +} +use sys::pool::{self, asyncify}; + +#[allow(warnings)] +#[path = "../src/file.rs"] +mod file; +use file::File; + +#[allow(warnings)] +#[path = "../src/blocking.rs"] +mod blocking; + +use tokio::prelude::*; +use tokio_test::{assert_pending, assert_ready, assert_ready_err, assert_ready_ok, task}; + +use std::io::SeekFrom; + +const HELLO: &[u8] = b"hello world..."; +const FOO: &[u8] = b"foo bar baz..."; + +#[test] +fn open_read() { + let (mock, file) = sys::File::mock(); + mock.read(HELLO); + + let mut file = File::from_std(file); + + let mut buf = [0; 1024]; + let mut t = task::spawn(file.read(&mut buf)); + + assert_eq!(0, pool::len()); + assert_pending!(t.poll()); + + assert_eq!(1, mock.remaining()); + assert_eq!(1, pool::len()); + + pool::run_one(); + + assert_eq!(0, mock.remaining()); + assert!(t.is_woken()); + + let n = assert_ready_ok!(t.poll()); + assert_eq!(n, HELLO.len()); + assert_eq!(&buf[..n], HELLO); +} + +#[test] +fn read_twice_before_dispatch() { + let (mock, file) = sys::File::mock(); + mock.read(HELLO); + + let mut file = File::from_std(file); + + let mut buf = [0; 1024]; + let mut t = task::spawn(file.read(&mut buf)); + + assert_pending!(t.poll()); + assert_pending!(t.poll()); + + assert_eq!(pool::len(), 1); + pool::run_one(); + + assert!(t.is_woken()); + + let n = assert_ready_ok!(t.poll()); + assert_eq!(&buf[..n], HELLO); +} + +#[test] +fn read_with_smaller_buf() { + let (mock, file) = sys::File::mock(); + mock.read(HELLO); + + let mut file = File::from_std(file); + + { + let mut buf = [0; 32]; + let mut t = task::spawn(file.read(&mut buf)); + assert_pending!(t.poll()); + } + + pool::run_one(); + + { + let mut buf = [0; 4]; + let mut t = task::spawn(file.read(&mut buf)); + let n = assert_ready_ok!(t.poll()); + assert_eq!(n, 4); + assert_eq!(&buf[..], &HELLO[..n]); + } + + // Calling again immediately succeeds with the rest of the buffer + let mut buf = [0; 32]; + let mut t = task::spawn(file.read(&mut buf)); + let n = assert_ready_ok!(t.poll()); + assert_eq!(n, 10); + assert_eq!(&buf[..n], &HELLO[4..]); + + assert_eq!(0, pool::len()); +} + +#[test] +fn read_with_bigger_buf() { + let (mock, file) = sys::File::mock(); + mock.read(&HELLO[..4]).read(&HELLO[4..]); + + let mut file = File::from_std(file); + + { + let mut buf = [0; 4]; + let mut t = task::spawn(file.read(&mut buf)); + assert_pending!(t.poll()); + } + + pool::run_one(); + + { + let mut buf = [0; 32]; + let mut t = task::spawn(file.read(&mut buf)); + let n = assert_ready_ok!(t.poll()); + assert_eq!(n, 4); + assert_eq!(&buf[..n], &HELLO[..n]); + } + + // Calling again immediately succeeds with the rest of the buffer + let mut buf = [0; 32]; + let mut t = task::spawn(file.read(&mut buf)); + + assert_pending!(t.poll()); + + assert_eq!(1, pool::len()); + pool::run_one(); + + assert!(t.is_woken()); + + let n = assert_ready_ok!(t.poll()); + assert_eq!(n, 10); + assert_eq!(&buf[..n], &HELLO[4..]); + + assert_eq!(0, pool::len()); +} + +#[test] +fn read_err_then_read_success() { + let (mock, file) = sys::File::mock(); + mock.read_err().read(&HELLO); + + let mut file = File::from_std(file); + + { + let mut buf = [0; 32]; + let mut t = task::spawn(file.read(&mut buf)); + assert_pending!(t.poll()); + + pool::run_one(); + + assert_ready_err!(t.poll()); + } + + { + let mut buf = [0; 32]; + let mut t = task::spawn(file.read(&mut buf)); + assert_pending!(t.poll()); + + pool::run_one(); + + let n = assert_ready_ok!(t.poll()); + + assert_eq!(n, HELLO.len()); + assert_eq!(&buf[..n], HELLO); + } +} + +#[test] +fn open_write() { + let (mock, file) = sys::File::mock(); + mock.write(HELLO); + + let mut file = File::from_std(file); + + let mut t = task::spawn(file.write(HELLO)); + + assert_eq!(0, pool::len()); + assert_ready_ok!(t.poll()); + + assert_eq!(1, mock.remaining()); + assert_eq!(1, pool::len()); + + pool::run_one(); + + assert_eq!(0, mock.remaining()); + assert!(!t.is_woken()); + + let mut t = task::spawn(file.flush()); + assert_ready_ok!(t.poll()); +} + +#[test] +fn flush_while_idle() { + let (_mock, file) = sys::File::mock(); + + let mut file = File::from_std(file); + + let mut t = task::spawn(file.flush()); + assert_ready_ok!(t.poll()); +} + +#[test] +fn read_with_buffer_larger_than_max() { + // Chunks + let a = 16 * 1024; + let b = a * 2; + let c = a * 3; + let d = a * 4; + + assert_eq!(d / 1024, 64); + + let mut data = vec![]; + for i in 0..(d - 1) { + data.push((i % 151) as u8); + } + + let (mock, file) = sys::File::mock(); + mock.read(&data[0..a]) + .read(&data[a..b]) + .read(&data[b..c]) + .read(&data[c..]); + + let mut file = File::from_std(file); + + let mut actual = vec![0; d]; + let mut pos = 0; + + while pos < data.len() { + let mut t = task::spawn(file.read(&mut actual[pos..])); + + assert_pending!(t.poll()); + pool::run_one(); + assert!(t.is_woken()); + + let n = assert_ready_ok!(t.poll()); + assert!(n <= a); + + pos += n; + } + + assert_eq!(mock.remaining(), 0); + assert_eq!(data, &actual[..data.len()]); +} + +#[test] +fn write_with_buffer_larger_than_max() { + // Chunks + let a = 16 * 1024; + let b = a * 2; + let c = a * 3; + let d = a * 4; + + assert_eq!(d / 1024, 64); + + let mut data = vec![]; + for i in 0..(d - 1) { + data.push((i % 151) as u8); + } + + let (mock, file) = sys::File::mock(); + mock.write(&data[0..a]) + .write(&data[a..b]) + .write(&data[b..c]) + .write(&data[c..]); + + let mut file = File::from_std(file); + + let mut rem = &data[..]; + + let mut first = true; + + while !rem.is_empty() { + let mut t = task::spawn(file.write(rem)); + + if !first { + assert_pending!(t.poll()); + pool::run_one(); + assert!(t.is_woken()); + } + + first = false; + + let n = assert_ready_ok!(t.poll()); + + rem = &rem[n..]; + } + + pool::run_one(); + + assert_eq!(mock.remaining(), 0); +} + +#[test] +fn write_twice_before_dispatch() { + let (mock, file) = sys::File::mock(); + mock.write(HELLO).write(FOO); + + let mut file = File::from_std(file); + + let mut t = task::spawn(file.write(HELLO)); + assert_ready_ok!(t.poll()); + + let mut t = task::spawn(file.write(FOO)); + assert_pending!(t.poll()); + + assert_eq!(pool::len(), 1); + pool::run_one(); + + assert!(t.is_woken()); + + assert_ready_ok!(t.poll()); + + let mut t = task::spawn(file.flush()); + assert_pending!(t.poll()); + + assert_eq!(pool::len(), 1); + pool::run_one(); + + assert!(t.is_woken()); + assert_ready_ok!(t.poll()); +} + +#[test] +fn incomplete_read_followed_by_write() { + let (mock, file) = sys::File::mock(); + mock.read(HELLO) + .seek_current_ok(-(HELLO.len() as i64), 0) + .write(FOO); + + let mut file = File::from_std(file); + + let mut buf = [0; 32]; + + let mut t = task::spawn(file.read(&mut buf)); + assert_pending!(t.poll()); + + pool::run_one(); + + let mut t = task::spawn(file.write(FOO)); + assert_ready_ok!(t.poll()); + + assert_eq!(pool::len(), 1); + pool::run_one(); + + let mut t = task::spawn(file.flush()); + assert_ready_ok!(t.poll()); +} + +#[test] +fn incomplete_partial_read_followed_by_write() { + let (mock, file) = sys::File::mock(); + mock.read(HELLO).seek_current_ok(-10, 0).write(FOO); + + let mut file = File::from_std(file); + + let mut buf = [0; 32]; + let mut t = task::spawn(file.read(&mut buf)); + assert_pending!(t.poll()); + + pool::run_one(); + + let mut buf = [0; 4]; + let mut t = task::spawn(file.read(&mut buf)); + assert_ready_ok!(t.poll()); + + let mut t = task::spawn(file.write(FOO)); + assert_ready_ok!(t.poll()); + + assert_eq!(pool::len(), 1); + pool::run_one(); + + let mut t = task::spawn(file.flush()); + assert_ready_ok!(t.poll()); +} + +#[test] +fn incomplete_read_followed_by_flush() { + let (mock, file) = sys::File::mock(); + mock.read(HELLO) + .seek_current_ok(-(HELLO.len() as i64), 0) + .write(FOO); + + let mut file = File::from_std(file); + + let mut buf = [0; 32]; + + let mut t = task::spawn(file.read(&mut buf)); + assert_pending!(t.poll()); + + pool::run_one(); + + let mut t = task::spawn(file.flush()); + assert_ready_ok!(t.poll()); + + let mut t = task::spawn(file.write(FOO)); + assert_ready_ok!(t.poll()); + + pool::run_one(); +} + +#[test] +fn incomplete_flush_followed_by_write() { + let (mock, file) = sys::File::mock(); + mock.write(HELLO).write(FOO); + + let mut file = File::from_std(file); + + let mut t = task::spawn(file.write(HELLO)); + let n = assert_ready_ok!(t.poll()); + assert_eq!(n, HELLO.len()); + + let mut t = task::spawn(file.flush()); + assert_pending!(t.poll()); + + // TODO: Move under write + pool::run_one(); + + let mut t = task::spawn(file.write(FOO)); + assert_ready_ok!(t.poll()); + + pool::run_one(); + + let mut t = task::spawn(file.flush()); + assert_ready_ok!(t.poll()); +} + +#[test] +fn read_err() { + let (mock, file) = sys::File::mock(); + mock.read_err(); + + let mut file = File::from_std(file); + + let mut buf = [0; 1024]; + let mut t = task::spawn(file.read(&mut buf)); + + assert_pending!(t.poll()); + + pool::run_one(); + assert!(t.is_woken()); + + assert_ready_err!(t.poll()); +} + +#[test] +fn write_write_err() { + let (mock, file) = sys::File::mock(); + mock.write_err(); + + let mut file = File::from_std(file); + + let mut t = task::spawn(file.write(HELLO)); + assert_ready_ok!(t.poll()); + + pool::run_one(); + + let mut t = task::spawn(file.write(FOO)); + assert_ready_err!(t.poll()); +} + +#[test] +fn write_read_write_err() { + let (mock, file) = sys::File::mock(); + mock.write_err().read(HELLO); + + let mut file = File::from_std(file); + + let mut t = task::spawn(file.write(HELLO)); + assert_ready_ok!(t.poll()); + + pool::run_one(); + + let mut buf = [0; 1024]; + let mut t = task::spawn(file.read(&mut buf)); + + assert_pending!(t.poll()); + + pool::run_one(); + + let mut t = task::spawn(file.write(FOO)); + assert_ready_err!(t.poll()); +} + +#[test] +fn write_read_flush_err() { + let (mock, file) = sys::File::mock(); + mock.write_err().read(HELLO); + + let mut file = File::from_std(file); + + let mut t = task::spawn(file.write(HELLO)); + assert_ready_ok!(t.poll()); + + pool::run_one(); + + let mut buf = [0; 1024]; + let mut t = task::spawn(file.read(&mut buf)); + + assert_pending!(t.poll()); + + pool::run_one(); + + let mut t = task::spawn(file.flush()); + assert_ready_err!(t.poll()); +} + +#[test] +fn write_seek_write_err() { + let (mock, file) = sys::File::mock(); + mock.write_err().seek_start_ok(0); + + let mut file = File::from_std(file); + + let mut t = task::spawn(file.write(HELLO)); + assert_ready_ok!(t.poll()); + + pool::run_one(); + + { + let mut t = task::spawn(file.seek(SeekFrom::Start(0))); + assert_pending!(t.poll()); + } + + pool::run_one(); + + let mut t = task::spawn(file.write(FOO)); + assert_ready_err!(t.poll()); +} + +#[test] +fn write_seek_flush_err() { + let (mock, file) = sys::File::mock(); + mock.write_err().seek_start_ok(0); + + let mut file = File::from_std(file); + + let mut t = task::spawn(file.write(HELLO)); + assert_ready_ok!(t.poll()); + + pool::run_one(); + + { + let mut t = task::spawn(file.seek(SeekFrom::Start(0))); + assert_pending!(t.poll()); + } + + pool::run_one(); + + let mut t = task::spawn(file.flush()); + assert_ready_err!(t.poll()); +} + +#[test] +fn sync_all_ordered_after_write() { + let (mock, file) = sys::File::mock(); + mock.write(HELLO).sync_all(); + + let mut file = File::from_std(file); + let mut t = task::spawn(file.write(HELLO)); + assert_ready_ok!(t.poll()); + + let mut t = task::spawn(file.sync_all()); + assert_pending!(t.poll()); + + assert_eq!(1, pool::len()); + pool::run_one(); + + assert!(t.is_woken()); + assert_pending!(t.poll()); + + assert_eq!(1, pool::len()); + pool::run_one(); + + assert!(t.is_woken()); + assert_ready_ok!(t.poll()); +} + +#[test] +fn sync_all_err_ordered_after_write() { + let (mock, file) = sys::File::mock(); + mock.write(HELLO).sync_all_err(); + + let mut file = File::from_std(file); + let mut t = task::spawn(file.write(HELLO)); + assert_ready_ok!(t.poll()); + + let mut t = task::spawn(file.sync_all()); + assert_pending!(t.poll()); + + assert_eq!(1, pool::len()); + pool::run_one(); + + assert!(t.is_woken()); + assert_pending!(t.poll()); + + assert_eq!(1, pool::len()); + pool::run_one(); + + assert!(t.is_woken()); + assert_ready_err!(t.poll()); +} + +#[test] +fn sync_data_ordered_after_write() { + let (mock, file) = sys::File::mock(); + mock.write(HELLO).sync_data(); + + let mut file = File::from_std(file); + let mut t = task::spawn(file.write(HELLO)); + assert_ready_ok!(t.poll()); + + let mut t = task::spawn(file.sync_data()); + assert_pending!(t.poll()); + + assert_eq!(1, pool::len()); + pool::run_one(); + + assert!(t.is_woken()); + assert_pending!(t.poll()); + + assert_eq!(1, pool::len()); + pool::run_one(); + + assert!(t.is_woken()); + assert_ready_ok!(t.poll()); +} + +#[test] +fn sync_data_err_ordered_after_write() { + let (mock, file) = sys::File::mock(); + mock.write(HELLO).sync_data_err(); + + let mut file = File::from_std(file); + let mut t = task::spawn(file.write(HELLO)); + assert_ready_ok!(t.poll()); + + let mut t = task::spawn(file.sync_data()); + assert_pending!(t.poll()); + + assert_eq!(1, pool::len()); + pool::run_one(); + + assert!(t.is_woken()); + assert_pending!(t.poll()); + + assert_eq!(1, pool::len()); + pool::run_one(); + + assert!(t.is_woken()); + assert_ready_err!(t.poll()); +} + +#[test] +fn open_set_len_ok() { + let (mock, file) = sys::File::mock(); + mock.set_len(123); + + let mut file = File::from_std(file); + let mut t = task::spawn(file.set_len(123)); + + assert_pending!(t.poll()); + assert_eq!(1, mock.remaining()); + + pool::run_one(); + assert_eq!(0, mock.remaining()); + + assert!(t.is_woken()); + assert_ready_ok!(t.poll()); +} + +#[test] +fn open_set_len_err() { + let (mock, file) = sys::File::mock(); + mock.set_len_err(123); + + let mut file = File::from_std(file); + let mut t = task::spawn(file.set_len(123)); + + assert_pending!(t.poll()); + assert_eq!(1, mock.remaining()); + + pool::run_one(); + assert_eq!(0, mock.remaining()); + + assert!(t.is_woken()); + assert_ready_err!(t.poll()); +} + +#[test] +fn partial_read_set_len_ok() { + let (mock, file) = sys::File::mock(); + mock.read(HELLO) + .seek_current_ok(-14, 0) + .set_len(123) + .read(FOO); + + let mut buf = [0; 32]; + let mut file = File::from_std(file); + + { + let mut t = task::spawn(file.read(&mut buf)); + assert_pending!(t.poll()); + } + + pool::run_one(); + + { + let mut t = task::spawn(file.set_len(123)); + + assert_pending!(t.poll()); + pool::run_one(); + assert_ready_ok!(t.poll()); + } + + let mut t = task::spawn(file.read(&mut buf)); + assert_pending!(t.poll()); + pool::run_one(); + let n = assert_ready_ok!(t.poll()); + + assert_eq!(n, FOO.len()); + assert_eq!(&buf[..n], FOO); +} diff --git a/tokio-fs/tests/link.rs b/tokio-fs/tests/link.rs index f162684b0..faf6e75e4 100644 --- a/tokio-fs/tests/link.rs +++ b/tokio-fs/tests/link.rs @@ -1,35 +1,30 @@ #![warn(rust_2018_idioms)] -use std::fs; +use tokio::fs; + use std::io::prelude::*; use std::io::BufReader; use tempfile::tempdir; -use tokio_fs::*; -mod pool; - -#[test] -fn test_hard_link() { +#[tokio::test] +async fn test_hard_link() { let dir = tempdir().unwrap(); let src = dir.path().join("src.txt"); let dst = dir.path().join("dst.txt"); { - let mut file = fs::File::create(&src).unwrap(); + let mut file = std::fs::File::create(&src).unwrap(); file.write_all(b"hello").unwrap(); } let dst_2 = dst.clone(); - pool::run(async move { - assert!(hard_link(src, dst_2.clone()).await.is_ok()); - Ok(()) - }); + assert!(fs::hard_link(src, dst_2.clone()).await.is_ok()); let mut content = String::new(); { - let file = fs::File::open(dst).unwrap(); + let file = std::fs::File::open(dst).unwrap(); let mut reader = BufReader::new(file); reader.read_to_string(&mut content).unwrap(); } @@ -38,43 +33,37 @@ fn test_hard_link() { } #[cfg(unix)] -#[test] -fn test_symlink() { +#[tokio::test] +async fn test_symlink() { let dir = tempdir().unwrap(); let src = dir.path().join("src.txt"); let dst = dir.path().join("dst.txt"); { - let mut file = fs::File::create(&src).unwrap(); + let mut file = std::fs::File::create(&src).unwrap(); file.write_all(b"hello").unwrap(); } let src_2 = src.clone(); let dst_2 = dst.clone(); - pool::run(async move { - assert!(os::unix::symlink(src_2.clone(), dst_2.clone()) - .await - .is_ok()); - Ok(()) - }); + assert!(fs::os::unix::symlink(src_2.clone(), dst_2.clone()) + .await + .is_ok()); let mut content = String::new(); { - let file = fs::File::open(dst.clone()).unwrap(); + let file = std::fs::File::open(dst.clone()).unwrap(); let mut reader = BufReader::new(file); reader.read_to_string(&mut content).unwrap(); } assert!(content == "hello"); - pool::run(async move { - let read = read_link(dst.clone()).await.unwrap(); - assert!(read == src); + let read = fs::read_link(dst.clone()).await.unwrap(); + assert!(read == src); - let symlink_meta = symlink_metadata(dst.clone()).await.unwrap(); - assert!(symlink_meta.file_type().is_symlink()); - Ok(()) - }); + let symlink_meta = fs::symlink_metadata(dst.clone()).await.unwrap(); + assert!(symlink_meta.file_type().is_symlink()); } diff --git a/tokio-fs/tests/pool/mod.rs b/tokio-fs/tests/pool/mod.rs deleted file mode 100644 index 39eab7d18..000000000 --- a/tokio-fs/tests/pool/mod.rs +++ /dev/null @@ -1,18 +0,0 @@ -use tokio_executor::threadpool::Builder; - -use std::future::Future; -use std::io; -use std::sync::mpsc; - -pub fn run(f: F) -where - F: Future> + Send + 'static, -{ - let pool = Builder::new().pool_size(1).build(); - let (tx, rx) = mpsc::channel(); - pool.spawn(async move { - f.await.unwrap(); - tx.send(()).unwrap(); - }); - rx.recv().unwrap() -} diff --git a/tokio-fs/tests/sys/file.rs b/tokio-fs/tests/sys/file.rs new file mode 100644 index 000000000..7f3beee84 --- /dev/null +++ b/tokio-fs/tests/sys/file.rs @@ -0,0 +1,265 @@ +use std::collections::VecDeque; +use std::fmt; +use std::fs::{Metadata, Permissions}; +use std::io; +use std::io::prelude::*; +use std::io::SeekFrom; +use std::path::PathBuf; +use std::sync::{Arc, Mutex}; + +pub struct File { + shared: Arc>, +} + +pub struct Handle { + shared: Arc>, +} + +struct Shared { + calls: VecDeque, +} + +#[derive(Debug)] +enum Call { + Read(io::Result>), + Write(io::Result>), + Seek(SeekFrom, io::Result), + SyncAll(io::Result<()>), + SyncData(io::Result<()>), + SetLen(u64, io::Result<()>), +} + +impl Handle { + pub fn read(&self, data: &[u8]) -> &Self { + let mut s = self.shared.lock().unwrap(); + s.calls.push_back(Call::Read(Ok(data.to_owned()))); + self + } + + pub fn read_err(&self) -> &Self { + let mut s = self.shared.lock().unwrap(); + s.calls + .push_back(Call::Read(Err(io::ErrorKind::Other.into()))); + self + } + + pub fn write(&self, data: &[u8]) -> &Self { + let mut s = self.shared.lock().unwrap(); + s.calls.push_back(Call::Write(Ok(data.to_owned()))); + self + } + + pub fn write_err(&self) -> &Self { + let mut s = self.shared.lock().unwrap(); + s.calls + .push_back(Call::Write(Err(io::ErrorKind::Other.into()))); + self + } + + pub fn seek_start_ok(&self, offset: u64) -> &Self { + let mut s = self.shared.lock().unwrap(); + s.calls + .push_back(Call::Seek(SeekFrom::Start(offset), Ok(offset))); + self + } + + pub fn seek_current_ok(&self, offset: i64, ret: u64) -> &Self { + let mut s = self.shared.lock().unwrap(); + s.calls + .push_back(Call::Seek(SeekFrom::Current(offset), Ok(ret))); + self + } + + pub fn sync_all(&self) -> &Self { + let mut s = self.shared.lock().unwrap(); + s.calls.push_back(Call::SyncAll(Ok(()))); + self + } + + pub fn sync_all_err(&self) -> &Self { + let mut s = self.shared.lock().unwrap(); + s.calls + .push_back(Call::SyncAll(Err(io::ErrorKind::Other.into()))); + self + } + + pub fn sync_data(&self) -> &Self { + let mut s = self.shared.lock().unwrap(); + s.calls.push_back(Call::SyncData(Ok(()))); + self + } + + pub fn sync_data_err(&self) -> &Self { + let mut s = self.shared.lock().unwrap(); + s.calls + .push_back(Call::SyncData(Err(io::ErrorKind::Other.into()))); + self + } + + pub fn set_len(&self, size: u64) -> &Self { + let mut s = self.shared.lock().unwrap(); + s.calls.push_back(Call::SetLen(size, Ok(()))); + self + } + + pub fn set_len_err(&self, size: u64) -> &Self { + let mut s = self.shared.lock().unwrap(); + s.calls + .push_back(Call::SetLen(size, Err(io::ErrorKind::Other.into()))); + self + } + + pub fn remaining(&self) -> usize { + let s = self.shared.lock().unwrap(); + s.calls.len() + } +} + +impl Drop for Handle { + fn drop(&mut self) { + if !std::thread::panicking() { + let s = self.shared.lock().unwrap(); + assert_eq!(0, s.calls.len()); + } + } +} + +impl File { + pub fn open(_: PathBuf) -> io::Result { + unimplemented!(); + } + + pub fn create(_: PathBuf) -> io::Result { + unimplemented!(); + } + + pub fn mock() -> (Handle, File) { + let shared = Arc::new(Mutex::new(Shared { + calls: VecDeque::new(), + })); + + let handle = Handle { + shared: shared.clone(), + }; + let file = File { shared }; + + (handle, file) + } + + pub fn sync_all(&self) -> io::Result<()> { + use self::Call::*; + + let mut s = self.shared.lock().unwrap(); + + match s.calls.pop_front() { + Some(SyncAll(ret)) => ret, + Some(op) => panic!("expected next call to be {:?}; was sync_all", op), + None => panic!("did not expect call"), + } + } + + pub fn sync_data(&self) -> io::Result<()> { + use self::Call::*; + + let mut s = self.shared.lock().unwrap(); + + match s.calls.pop_front() { + Some(SyncData(ret)) => ret, + Some(op) => panic!("expected next call to be {:?}; was sync_all", op), + None => panic!("did not expect call"), + } + } + + pub fn set_len(&self, size: u64) -> io::Result<()> { + use self::Call::*; + + let mut s = self.shared.lock().unwrap(); + + match s.calls.pop_front() { + Some(SetLen(arg, ret)) => { + assert_eq!(arg, size); + ret + } + Some(op) => panic!("expected next call to be {:?}; was sync_all", op), + None => panic!("did not expect call"), + } + } + + pub fn metadata(&self) -> io::Result { + unimplemented!(); + } + + pub fn set_permissions(&self, _perm: Permissions) -> io::Result<()> { + unimplemented!(); + } + + pub fn try_clone(&self) -> io::Result { + unimplemented!(); + } +} + +impl Read for &'_ File { + fn read(&mut self, dst: &mut [u8]) -> io::Result { + use self::Call::*; + + let mut s = self.shared.lock().unwrap(); + + match s.calls.pop_front() { + Some(Read(Ok(data))) => { + assert!(dst.len() >= data.len()); + assert!(dst.len() <= 16 * 1024, "actual = {}", dst.len()); // max buffer + + &mut dst[..data.len()].copy_from_slice(&data); + Ok(data.len()) + } + Some(Read(Err(e))) => Err(e), + Some(op) => panic!("expected next call to be {:?}; was a read", op), + None => panic!("did not expect call"), + } + } +} + +impl Write for &'_ File { + fn write(&mut self, src: &[u8]) -> io::Result { + use self::Call::*; + + let mut s = self.shared.lock().unwrap(); + + match s.calls.pop_front() { + Some(Write(Ok(data))) => { + assert_eq!(src, &data[..]); + Ok(src.len()) + } + Some(Write(Err(e))) => Err(e), + Some(op) => panic!("expected next call to be {:?}; was write", op), + None => panic!("did not expect call"), + } + } + + fn flush(&mut self) -> io::Result<()> { + Ok(()) + } +} + +impl Seek for &'_ File { + fn seek(&mut self, pos: SeekFrom) -> io::Result { + use self::Call::*; + + let mut s = self.shared.lock().unwrap(); + + match s.calls.pop_front() { + Some(Seek(expect, res)) => { + assert_eq!(expect, pos); + res + } + Some(op) => panic!("expected call {:?}; was `seek`", op), + None => panic!("did not expect call; was `seek`"), + } + } +} + +impl fmt::Debug for File { + fn fmt(&self, fmt: &mut fmt::Formatter<'_>) -> fmt::Result { + fmt.debug_struct("mock::File").finish() + } +} diff --git a/tokio-fs/tests/sys/pool.rs b/tokio-fs/tests/sys/pool.rs new file mode 100644 index 000000000..501540f15 --- /dev/null +++ b/tokio-fs/tests/sys/pool.rs @@ -0,0 +1,66 @@ +use tokio_sync::oneshot; + +use std::cell::RefCell; +use std::collections::VecDeque; +use std::future::Future; +use std::io; +use std::pin::Pin; +use std::task::{Context, Poll}; + +thread_local! { + static QUEUE: RefCell>> = RefCell::new(VecDeque::new()) +} + +#[derive(Debug)] +pub(crate) struct Blocking { + rx: oneshot::Receiver, +} + +pub(crate) fn run(f: F) -> Blocking +where + F: FnOnce() -> R + Send + 'static, + R: Send + 'static, +{ + let (tx, rx) = oneshot::channel(); + let task = Box::new(move || { + let _ = tx.send(f()); + }); + + QUEUE.with(|cell| cell.borrow_mut().push_back(task)); + + Blocking { rx } +} + +impl Future for Blocking { + type Output = T; + + fn poll(mut self: Pin<&mut Self>, cx: &mut Context<'_>) -> Poll { + use std::task::Poll::*; + + match Pin::new(&mut self.rx).poll(cx) { + Ready(Ok(v)) => Ready(v), + Ready(Err(e)) => panic!("error = {:?}", e), + Pending => Pending, + } + } +} + +pub(crate) async fn asyncify(f: F) -> io::Result +where + F: FnOnce() -> io::Result + Send + 'static, + T: Send + 'static, +{ + run(f).await +} + +pub(crate) fn len() -> usize { + QUEUE.with(|cell| cell.borrow().len()) +} + +pub(crate) fn run_one() { + let task = QUEUE + .with(|cell| cell.borrow_mut().pop_front()) + .expect("expected task to run, but none ready"); + + task(); +}