Refactor the task module (#421)

* Refactor the task module

* Fix clippy warning

* Simplify task-local entries

* Reduce the amount of future wrapping

* Cleanup

* Simplify stealing

Author: Stjepan Glavina

Cargo.toml

@@ -27,23 +27,23 @@ unstable = ["broadcaster"]
 [dependencies]
 async-macros = "1.0.0"
 async-task = "1.0.0"
+broadcaster = { version = "0.2.6", optional = true, default-features = false, features = ["default-channels"] }
 crossbeam-channel = "0.3.9"
 crossbeam-deque = "0.7.1"
 crossbeam-utils = "0.6.6"
 futures-core-preview = "=0.3.0-alpha.19"
 futures-io-preview = "=0.3.0-alpha.19"
 futures-timer = "1.0.2"
+kv-log-macro = "1.0.4"
 log = { version = "0.4.8", features = ["kv_unstable"] }
 memchr = "2.2.1"
 mio = "0.6.19"
 mio-uds = "0.6.7"
 num_cpus = "1.10.1"
 once_cell = "1.2.0"
+pin-project-lite = "0.1"
 pin-utils = "0.1.0-alpha.4"
 slab = "0.4.2"
-kv-log-macro = "1.0.4"
-broadcaster = { version = "0.2.6", optional = true, default-features = false, features = ["default-channels"] }
-pin-project-lite = "0.1"
 
 [dev-dependencies]
 femme = "1.2.0"

src/fs/canonicalize.rs

@@ -1,6 +1,6 @@
 use crate::io;
 use crate::path::{Path, PathBuf};
-use crate::task::blocking;
+use crate::task::spawn_blocking;
 
 /// Returns the canonical form of a path.
 ///
@@ -32,5 +32,5 @@ use crate::task::blocking;
 /// ```
 pub async fn canonicalize<P: AsRef<Path>>(path: P) -> io::Result<PathBuf> {
     let path = path.as_ref().to_owned();
-    blocking::spawn(move || std::fs::canonicalize(&path).map(Into::into)).await
+    spawn_blocking(move || std::fs::canonicalize(&path).map(Into::into)).await
 }

src/fs/copy.rs

@@ -1,6 +1,6 @@
 use crate::io;
 use crate::path::Path;
-use crate::task::blocking;
+use crate::task::spawn_blocking;
 
 /// Copies the contents and permissions of a file to a new location.
 ///
@@ -41,5 +41,5 @@ use crate::task::blocking;
 pub async fn copy<P: AsRef<Path>, Q: AsRef<Path>>(from: P, to: Q) -> io::Result<u64> {
     let from = from.as_ref().to_owned();
     let to = to.as_ref().to_owned();
-    blocking::spawn(move || std::fs::copy(&from, &to)).await
+    spawn_blocking(move || std::fs::copy(&from, &to)).await
 }

src/fs/create_dir.rs

@@ -1,6 +1,6 @@
 use crate::io;
 use crate::path::Path;
-use crate::task::blocking;
+use crate::task::spawn_blocking;
 
 /// Creates a new directory.
 ///
@@ -34,5 +34,5 @@ use crate::task::blocking;
 /// ```
 pub async fn create_dir<P: AsRef<Path>>(path: P) -> io::Result<()> {
     let path = path.as_ref().to_owned();
-    blocking::spawn(move || std::fs::create_dir(path)).await
+    spawn_blocking(move || std::fs::create_dir(path)).await
 }

src/fs/create_dir_all.rs

@@ -1,6 +1,6 @@
 use crate::io;
 use crate::path::Path;
-use crate::task::blocking;
+use crate::task::spawn_blocking;
 
 /// Creates a new directory and all of its parents if they are missing.
 ///
@@ -29,5 +29,5 @@ use crate::task::blocking;
 /// ```
 pub async fn create_dir_all<P: AsRef<Path>>(path: P) -> io::Result<()> {
     let path = path.as_ref().to_owned();
-    blocking::spawn(move || std::fs::create_dir_all(path)).await
+    spawn_blocking(move || std::fs::create_dir_all(path)).await
 }

src/fs/dir_builder.rs

@@ -2,7 +2,7 @@ use std::future::Future;
 
 use crate::io;
 use crate::path::Path;
-use crate::task::blocking;
+use crate::task::spawn_blocking;
 
 /// A builder for creating directories with configurable options.
 ///
@@ -107,7 +107,7 @@ impl DirBuilder {
         }
 
         let path = path.as_ref().to_owned();
-        async move { blocking::spawn(move || builder.create(path)).await }
+        async move { spawn_blocking(move || builder.create(path)).await }
     }
 }
 

src/fs/dir_entry.rs

@@ -5,7 +5,7 @@ use std::sync::Arc;
 use crate::fs::{FileType, Metadata};
 use crate::io;
 use crate::path::PathBuf;
-use crate::task::blocking;
+use crate::task::spawn_blocking;
 
 /// An entry in a directory.
 ///
@@ -87,7 +87,7 @@ impl DirEntry {
     /// ```
     pub async fn metadata(&self) -> io::Result<Metadata> {
         let inner = self.0.clone();
-        blocking::spawn(move || inner.metadata()).await
+        spawn_blocking(move || inner.metadata()).await
     }
 
     /// Reads the file type for this entry.
@@ -125,7 +125,7 @@ impl DirEntry {
     /// ```
     pub async fn file_type(&self) -> io::Result<FileType> {
         let inner = self.0.clone();
-        blocking::spawn(move || inner.file_type()).await
+        spawn_blocking(move || inner.file_type()).await
     }
 
     /// Returns the bare name of this entry without the leading path.

src/fs/file.rs

@@ -12,7 +12,7 @@ use crate::future;
 use crate::io::{self, Read, Seek, SeekFrom, Write};
 use crate::path::Path;
 use crate::prelude::*;
-use crate::task::{self, blocking, Context, Poll, Waker};
+use crate::task::{self, spawn_blocking, Context, Poll, Waker};
 
 /// An open file on the filesystem.
 ///
@@ -112,7 +112,7 @@ impl File {
     /// ```
     pub async fn open<P: AsRef<Path>>(path: P) -> io::Result<File> {
         let path = path.as_ref().to_owned();
-        let file = blocking::spawn(move || std::fs::File::open(&path)).await?;
+        let file = spawn_blocking(move || std::fs::File::open(&path)).await?;
         Ok(File::new(file, true))
     }
 
@@ -147,7 +147,7 @@ impl File {
     /// ```
     pub async fn create<P: AsRef<Path>>(path: P) -> io::Result<File> {
         let path = path.as_ref().to_owned();
-        let file = blocking::spawn(move || std::fs::File::create(&path)).await?;
+        let file = spawn_blocking(move || std::fs::File::create(&path)).await?;
         Ok(File::new(file, true))
     }
 
@@ -180,7 +180,7 @@ impl File {
         })
         .await?;
 
-        blocking::spawn(move || state.file.sync_all()).await
+        spawn_blocking(move || state.file.sync_all()).await
     }
 
     /// Synchronizes OS-internal buffered contents to disk.
@@ -216,7 +216,7 @@ impl File {
         })
         .await?;
 
-        blocking::spawn(move || state.file.sync_data()).await
+        spawn_blocking(move || state.file.sync_data()).await
     }
 
     /// Truncates or extends the file.
@@ -249,7 +249,7 @@ impl File {
         })
         .await?;
 
-        blocking::spawn(move || state.file.set_len(size)).await
+        spawn_blocking(move || state.file.set_len(size)).await
     }
 
     /// Reads the file's metadata.
@@ -268,7 +268,7 @@ impl File {
     /// ```
     pub async fn metadata(&self) -> io::Result<Metadata> {
         let file = self.file.clone();
-        blocking::spawn(move || file.metadata()).await
+        spawn_blocking(move || file.metadata()).await
     }
 
     /// Changes the permissions on the file.
@@ -297,7 +297,7 @@ impl File {
     /// ```
     pub async fn set_permissions(&self, perm: Permissions) -> io::Result<()> {
         let file = self.file.clone();
-        blocking::spawn(move || file.set_permissions(perm)).await
+        spawn_blocking(move || file.set_permissions(perm)).await
     }
 }
 
@@ -692,7 +692,7 @@ impl LockGuard<State> {
         self.register(cx);
 
         // Start a read operation asynchronously.
-        blocking::spawn(move || {
+        spawn_blocking(move || {
             // Read some data from the file into the cache.
             let res = {
                 let State { file, cache, .. } = &mut *self;
@@ -801,7 +801,7 @@ impl LockGuard<State> {
                 self.register(cx);
 
                 // Start a write operation asynchronously.
-                blocking::spawn(move || {
+                spawn_blocking(move || {
                     match (&*self.file).write_all(&self.cache) {
                         Ok(_) => {
                             // Switch to idle mode.
@@ -834,7 +834,7 @@ impl LockGuard<State> {
         self.register(cx);
 
         // Start a flush operation asynchronously.
-        blocking::spawn(move || {
+        spawn_blocking(move || {
             match (&*self.file).flush() {
                 Ok(()) => {
                     // Mark the file as flushed.

src/fs/hard_link.rs

@@ -1,6 +1,6 @@
 use crate::io;
 use crate::path::Path;
-use crate::task::blocking;
+use crate::task::spawn_blocking;
 
 /// Creates a hard link on the filesystem.
 ///
@@ -32,5 +32,5 @@ use crate::task::blocking;
 pub async fn hard_link<P: AsRef<Path>, Q: AsRef<Path>>(from: P, to: Q) -> io::Result<()> {
     let from = from.as_ref().to_owned();
     let to = to.as_ref().to_owned();
-    blocking::spawn(move || std::fs::hard_link(&from, &to)).await
+    spawn_blocking(move || std::fs::hard_link(&from, &to)).await
 }

src/fs/metadata.rs

@@ -1,6 +1,6 @@
 use crate::io;
 use crate::path::Path;
-use crate::task::blocking;
+use crate::task::spawn_blocking;
 
 /// Reads metadata for a path.
 ///
@@ -34,7 +34,7 @@ use crate::task::blocking;
 /// ```
 pub async fn metadata<P: AsRef<Path>>(path: P) -> io::Result<Metadata> {
     let path = path.as_ref().to_owned();
-    blocking::spawn(move || std::fs::metadata(path)).await
+    spawn_blocking(move || std::fs::metadata(path)).await
 }
 
 cfg_not_docs! {

src/fs/open_options.rs

@@ -3,7 +3,7 @@ use std::future::Future;
 use crate::fs::File;
 use crate::io;
 use crate::path::Path;
-use crate::task::blocking;
+use crate::task::spawn_blocking;
 
 /// A builder for opening files with configurable options.
 ///
@@ -285,7 +285,7 @@ impl OpenOptions {
         let path = path.as_ref().to_owned();
         let options = self.0.clone();
         async move {
-            let file = blocking::spawn(move || options.open(path)).await?;
+            let file = spawn_blocking(move || options.open(path)).await?;
             Ok(File::new(file, true))
         }
     }

src/fs/read.rs

@@ -1,6 +1,6 @@
 use crate::io;
 use crate::path::Path;
-use crate::task::blocking;
+use crate::task::spawn_blocking;
 
 /// Reads the entire contents of a file as raw bytes.
 ///
@@ -36,5 +36,5 @@ use crate::task::blocking;
 /// ```
 pub async fn read<P: AsRef<Path>>(path: P) -> io::Result<Vec<u8>> {
     let path = path.as_ref().to_owned();
-    blocking::spawn(move || std::fs::read(path)).await
+    spawn_blocking(move || std::fs::read(path)).await
 }

src/fs/read_dir.rs

@@ -5,7 +5,7 @@ use crate::future::Future;
 use crate::io;
 use crate::path::Path;
 use crate::stream::Stream;
-use crate::task::{blocking, Context, JoinHandle, Poll};
+use crate::task::{spawn_blocking, Context, JoinHandle, Poll};
 
 /// Returns a stream of entries in a directory.
 ///
@@ -45,7 +45,7 @@ use crate::task::{blocking, Context, JoinHandle, Poll};
 /// ```
 pub async fn read_dir<P: AsRef<Path>>(path: P) -> io::Result<ReadDir> {
     let path = path.as_ref().to_owned();
-    blocking::spawn(move || std::fs::read_dir(path))
+    spawn_blocking(move || std::fs::read_dir(path))
         .await
         .map(ReadDir::new)
 }
@@ -91,7 +91,7 @@ impl Stream for ReadDir {
                     let mut inner = opt.take().unwrap();
 
                     // Start the operation asynchronously.
-                    self.0 = State::Busy(blocking::spawn(move || {
+                    self.0 = State::Busy(spawn_blocking(move || {
                         let next = inner.next();
                         (inner, next)
                     }));

src/fs/read_link.rs

@@ -1,6 +1,6 @@
 use crate::io;
 use crate::path::{Path, PathBuf};
-use crate::task::blocking;
+use crate::task::spawn_blocking;
 
 /// Reads a symbolic link and returns the path it points to.
 ///
@@ -28,5 +28,5 @@ use crate::task::blocking;
 /// ```
 pub async fn read_link<P: AsRef<Path>>(path: P) -> io::Result<PathBuf> {
     let path = path.as_ref().to_owned();
-    blocking::spawn(move || std::fs::read_link(path).map(Into::into)).await
+    spawn_blocking(move || std::fs::read_link(path).map(Into::into)).await
 }

src/fs/read_to_string.rs

@@ -1,6 +1,6 @@
 use crate::io;
 use crate::path::Path;
-use crate::task::blocking;
+use crate::task::spawn_blocking;
 
 /// Reads the entire contents of a file as a string.
 ///
@@ -37,5 +37,5 @@ use crate::task::blocking;
 /// ```
 pub async fn read_to_string<P: AsRef<Path>>(path: P) -> io::Result<String> {
     let path = path.as_ref().to_owned();
-    blocking::spawn(move || std::fs::read_to_string(path)).await
+    spawn_blocking(move || std::fs::read_to_string(path)).await
 }

src/fs/remove_dir.rs

@@ -1,6 +1,6 @@
 use crate::io;
 use crate::path::Path;
-use crate::task::blocking;
+use crate::task::spawn_blocking;
 
 /// Removes an empty directory.
 ///
@@ -29,5 +29,5 @@ use crate::task::blocking;
 /// ```
 pub async fn remove_dir<P: AsRef<Path>>(path: P) -> io::Result<()> {
     let path = path.as_ref().to_owned();
-    blocking::spawn(move || std::fs::remove_dir(path)).await
+    spawn_blocking(move || std::fs::remove_dir(path)).await
 }

src/fs/remove_dir_all.rs

@@ -1,6 +1,6 @@
 use crate::io;
 use crate::path::Path;
-use crate::task::blocking;
+use crate::task::spawn_blocking;
 
 /// Removes a directory and all of its contents.
 ///
@@ -29,5 +29,5 @@ use crate::task::blocking;
 /// ```
 pub async fn remove_dir_all<P: AsRef<Path>>(path: P) -> io::Result<()> {
     let path = path.as_ref().to_owned();
-    blocking::spawn(move || std::fs::remove_dir_all(path)).await
+    spawn_blocking(move || std::fs::remove_dir_all(path)).await
 }

src/fs/remove_file.rs

@@ -1,6 +1,6 @@
 use crate::io;
 use crate::path::Path;
-use crate::task::blocking;
+use crate::task::spawn_blocking;
 
 /// Removes a file.
 ///
@@ -29,5 +29,5 @@ use crate::task::blocking;
 /// ```
 pub async fn remove_file<P: AsRef<Path>>(path: P) -> io::Result<()> {
     let path = path.as_ref().to_owned();
-    blocking::spawn(move || std::fs::remove_file(path)).await
+    spawn_blocking(move || std::fs::remove_file(path)).await
 }