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-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