chore: prepare tokio-uds v0.2.7 release (#2638)

Signed-off-by: Lucio Franco <[email protected]>

.cirrus.yml

@@ -1,5 +1,5 @@
 freebsd_instance:
-  image: freebsd-12-0-release-amd64
+  image: freebsd-12-1-release-amd64
 
 # Test FreeBSD in a full VM on cirrus-ci.com.  Test the i686 target too, in the
 # same VM.  The binary will be built in 32-bit mode, but will execute on a

Cargo.toml

@@ -9,7 +9,7 @@ members = [
   "tokio-fs",
   "tokio-futures",
   "tokio-io",
-  "tokio-macros",
+  "tokio-process",
   "tokio-reactor",
   "tokio-signal",
   "tokio-sync",

async-await/.cargo/config

@@ -1,2 +0,0 @@
-[build]
-target-dir = "../target"

async-await/Cargo.toml

@@ -1,49 +0,0 @@
-[package]
-name = "examples"
-edition = "2018"
-version = "0.1.0"
-authors = ["Carl Lerche <[email protected]>"]
-license = "MIT"
-
-# Break out of the parent workspace
-[workspace]
-
-[[bin]]
-name = "chat"
-path = "src/chat.rs"
-
-[[bin]]
-name = "echo_client"
-path = "src/echo_client.rs"
-
-[[bin]]
-name = "echo_server"
-path = "src/echo_server.rs"
-
-[[bin]]
-name = "hyper"
-path = "src/hyper.rs"
-
-[dependencies]
-tokio = { version = "0.1.18", features = ["async-await-preview"] }
-futures = "0.1.23"
-bytes = "0.4.9"
-hyper = "0.12.8"
-
-# Avoid using crates.io for Tokio dependencies
-[patch.crates-io]
-tokio = { path = "../tokio" }
-tokio-codec = { path = "../tokio-codec" }
-tokio-current-thread = { path = "../tokio-current-thread" }
-tokio-executor = { path = "../tokio-executor" }
-tokio-fs = { path = "../tokio-fs" }
-tokio-futures  = { path = "../tokio-futures" }
-tokio-io = { path = "../tokio-io" }
-tokio-reactor = { path = "../tokio-reactor" }
-tokio-signal = { path = "../tokio-signal" }
-tokio-tcp = { path = "../tokio-tcp" }
-tokio-threadpool = { path = "../tokio-threadpool" }
-tokio-timer = { path = "../tokio-timer" }
-tokio-tls = { path = "../tokio-tls" }
-tokio-udp = { path = "../tokio-udp" }
-tokio-uds = { path = "../tokio-uds" }

async-await/README.md

@@ -1,5 +0,0 @@
-# Tokio async/await examples
-
-These are a separate crate in order to work around some cargo bugs. It also
-allows `[patch]` to be used in `Cargo.toml` to ensure the correct lib versions
-are being pulled in.

async-await/src/chat.rs

@@ -1,131 +0,0 @@
-#![feature(await_macro, async_await)]
-
-use tokio::await;
-use tokio::codec::{LinesCodec, Decoder};
-use tokio::net::{TcpListener, TcpStream};
-use tokio::prelude::*;
-
-use futures::sync::mpsc;
-
-use std::collections::HashMap;
-use std::io;
-use std::net::SocketAddr;
-use std::sync::{Arc, Mutex};
-
-/// Shorthand for the transmit half of the message channel.
-type Tx = mpsc::UnboundedSender<String>;
-
-struct Shared {
-    peers: HashMap<SocketAddr, Tx>,
-}
-
-impl Shared {
-    /// Create a new, empty, instance of `Shared`.
-    fn new() -> Self {
-        Shared {
-            peers: HashMap::new(),
-        }
-    }
-}
-
-async fn process(stream: TcpStream, state: Arc<Mutex<Shared>>) -> io::Result<()> {
-    let addr = stream.peer_addr().unwrap();
-    let mut lines = LinesCodec::new().framed(stream);
-
-    // Extract the peer's name
-    let name = match await!(lines.next()) {
-        Some(name) => name?,
-        None => {
-            // Disconnected early
-            return Ok(());
-        }
-    };
-
-    println!("`{}` is joining the chat", name);
-
-    let (tx, mut rx) = mpsc::unbounded();
-
-    // Register the socket
-    state.lock().unwrap()
-        .peers.insert(addr, tx);
-
-    // Split the `lines` handle into send and recv handles. This allows spawning
-    // separate tasks.
-    let (mut lines_tx, mut lines_rx) = lines.split();
-
-    // Spawn a task that receives all lines broadcasted to us from other peers
-    // and writes it to the client.
-    tokio::spawn_async(async move {
-        while let Some(line) = await!(rx.next()) {
-            let line = line.unwrap();
-            await!(lines_tx.send_async(line)).unwrap();
-        }
-    });
-
-    // Use the current task to read lines from the socket and broadcast them to
-    // other peers.
-    while let Some(message) = await!(lines_rx.next()) {
-        // TODO: Error handling
-        let message = message.unwrap();
-
-        let mut line = name.clone();
-        line.push_str(": ");
-        line.push_str(&message);
-        line.push_str("\r\n");
-
-        let state = state.lock().unwrap();
-
-        for (peer_addr, tx) in &state.peers {
-            if *peer_addr != addr {
-                // TODO: Error handling
-                tx.unbounded_send(line.clone()).unwrap();
-            }
-        }
-    }
-
-    // Remove the client from the shared state. Doing so will also result in the
-    // tx task to terminate.
-    state.lock().unwrap()
-        .peers.remove(&addr)
-        .expect("bug");
-
-    Ok(())
-}
-
-#[tokio::main]
-async fn main() {
-    // Create the shared state. This is how all the peers communicate.
-    //
-    // The server task will hold a handle to this. For every new client, the
-    // `state` handle is cloned and passed into the task that processes the
-    // client connection.
-    let state = Arc::new(Mutex::new(Shared::new()));
-
-    let addr = "127.0.0.1:6142".parse().unwrap();
-
-    // Bind a TCP listener to the socket address.
-    //
-    // Note that this is the Tokio TcpListener, which is fully async.
-    let listener = TcpListener::bind(&addr).unwrap();
-
-    println!("server running on localhost:6142");
-
-    // Start the Tokio runtime.
-    let mut incoming = listener.incoming();
-
-    while let Some(stream) = await!(incoming.next()) {
-        let stream = match stream {
-            Ok(stream) => stream,
-            Err(_) => continue,
-        };
-
-        let state = state.clone();
-
-        tokio::spawn_async(async move {
-            if let Err(_) = await!(process(stream, state)) {
-                eprintln!("failed to process connection");
-            }
-        });
-    }
-}
-

async-await/src/echo_client.rs

@@ -1,50 +0,0 @@
-#![feature(await_macro, async_await)]
-
-use tokio::await;
-use tokio::net::TcpStream;
-use tokio::prelude::*;
-
-use std::io;
-use std::net::SocketAddr;
-
-const MESSAGES: &[&str] = &[
-    "hello",
-    "world",
-    "one two three",
-];
-
-async fn run_client(addr: &SocketAddr) -> io::Result<()> {
-    let mut stream = await!(TcpStream::connect(addr))?;
-
-    // Buffer to read into
-    let mut buf = [0; 128];
-
-    for msg in MESSAGES {
-        println!(" > write = {:?}", msg);
-
-        // Write the message to the server
-        await!(stream.write_all_async(msg.as_bytes()))?;
-
-        // Read the message back from the server
-        await!(stream.read_exact_async(&mut buf[..msg.len()]))?;
-
-        assert_eq!(&buf[..msg.len()], msg.as_bytes());
-    }
-
-    Ok(())
-}
-
-#[tokio::main]
-async fn main() {
-    use std::env;
-
-    let addr = env::args().nth(1).unwrap_or("127.0.0.1:8080".to_string());
-    let addr = addr.parse::<SocketAddr>().unwrap();
-
-    // Connect to the echo serveer
-
-    match await!(run_client(&addr)) {
-        Ok(_) => println!("done."),
-        Err(e) => eprintln!("echo client failed; error = {:?}", e),
-    }
-}

async-await/src/echo_server.rs

@@ -1,42 +0,0 @@
-#![feature(await_macro, async_await)]
-
-use tokio::await;
-use tokio::net::{TcpListener, TcpStream};
-use tokio::prelude::*;
-
-use std::net::SocketAddr;
-
-fn handle(mut stream: TcpStream) {
-    tokio::spawn_async(async move {
-        let mut buf = [0; 1024];
-
-        loop {
-            match await!(stream.read_async(&mut buf)).unwrap() {
-                0 => break, // Socket closed
-                n => {
-                    // Send the data back
-                    await!(stream.write_all_async(&buf[0..n])).unwrap();
-                }
-            }
-        }
-    });
-}
-
-#[tokio::main]
-async fn main() {
-    use std::env;
-
-    let addr = env::args().nth(1).unwrap_or("127.0.0.1:8080".to_string());
-    let addr = addr.parse::<SocketAddr>().unwrap();
-
-    // Bind the TCP listener
-    let listener = TcpListener::bind(&addr).unwrap();
-    println!("Listening on: {}", addr);
-
-    let mut incoming = listener.incoming();
-
-    while let Some(stream) = await!(incoming.next()) {
-        let stream = stream.unwrap();
-        handle(stream);
-    }
-}

async-await/src/hyper.rs

@@ -1,29 +0,0 @@
-#![feature(await_macro, async_await)]
-
-use tokio::await;
-use tokio::prelude::*;
-use hyper::Client;
-
-use std::time::Duration;
-use std::str;
-
-#[tokio::main]
-async fn main() {
-    let client = Client::new();
-
-    let uri = "http://httpbin.org/ip".parse().unwrap();
-
-    let response = await!({
-        client.get(uri)
-            .timeout(Duration::from_secs(10))
-    }).unwrap();
-
-    println!("Response: {}", response.status());
-
-    let mut body = response.into_body();
-
-    while let Some(chunk) = await!(body.next()) {
-        let chunk = chunk.unwrap();
-        println!("chunk = {}", str::from_utf8(&chunk[..]).unwrap());
-    }
-}

async-await/tests/macros.rs

@@ -1,22 +0,0 @@
-#![feature(await_macro, async_await)]
-
-use tokio::await;
-use tokio::timer::Delay;
-use std::time::{Duration, Instant};
-
-#[tokio::test]
-async fn success_no_async() {
-    assert!(true);
-}
-
-#[tokio::test]
-#[should_panic]
-async fn fail_no_async() {
-    assert!(false);
-}
-
-#[tokio::test]
-async fn use_timer() {
-    let when = Instant::now() + Duration::from_millis(10);
-    await!(Delay::new(when));
-}

azure-pipelines.yml

@@ -24,10 +24,11 @@ jobs:
     cross: true
     crates:
       - tokio-fs
+      - tokio-process
       - tokio-reactor
       - tokio-signal
       - tokio-tcp
-      - tokio-tls
+      # - tokio-tls
       - tokio-udp
       - tokio-uds
 
@@ -68,13 +69,6 @@ jobs:
       tokio-buf:
         - util
 
-# Run async-await tests
-- template: ci/azure-test-nightly.yml
-  parameters:
-    name: test_nightly
-    displayName: Test Async / Await
-    rust: nightly-2019-04-25
-
 # Try cross compiling
 - template: ci/azure-cross-compile.yml
   parameters:
@@ -91,7 +85,7 @@ jobs:
 - template: ci/azure-check-minrust.yml
   parameters:
     name: minrust
-    rust_version: 1.26.0
+    rust_version: 1.31.0
 
 - template: ci/azure-tsan.yml
   parameters:
@@ -105,7 +99,6 @@ jobs:
       - test_sub_cross
       - test_linux
       - features
-      - test_nightly
       - cross_32bit_linux
       - minrust
       - tsan

benches/latency.rs

@@ -1,5 +1,4 @@
 #![feature(test)]
-#![deny(warnings)]
 
 extern crate test;
 #[macro_use]

benches/mio-ops.rs

@@ -1,7 +1,6 @@
 // Measure cost of different operations
 // to get a sense of performance tradeoffs
 #![feature(test)]
-#![deny(warnings)]
 
 extern crate mio;
 extern crate test;

benches/tcp.rs

@@ -1,5 +1,4 @@
 #![feature(test)]
-#![deny(warnings)]
 
 extern crate futures;
 extern crate tokio;

ci/azure-test-nightly.yml

@@ -1,19 +0,0 @@
-jobs:
-- job: ${{ parameters.name }}
-  displayName: ${{ parameters.displayName }}
-  pool:
-    vmImage: ubuntu-16.04
-
-  steps:
-  - template: azure-install-rust.yml
-    parameters:
-      rust_version: ${{ parameters.rust }}
-
-  - template: azure-patch-crates.yml
-
-  - script: cargo check --all
-    displayName: cargo check --all
-
-  # Check benches
-  - script: cargo check --benches --all
-    displayName: Check benchmarks

ci/azure-test-stable.yml

@@ -8,7 +8,7 @@ jobs:
 
       ${{ if parameters.cross }}:
         MacOS:
-          vmImage: macOS-10.13
+          vmImage: macOS-10.14
         Windows:
           vmImage: vs2017-win2016
   pool:

ci/azure-tsan.yml

@@ -12,7 +12,7 @@ jobs:
   steps:
   - template: azure-install-rust.yml
     parameters:
-      rust_version: nightly-2018-11-18
+      rust_version: nightly-2019-07-17
 
   - template: azure-patch-crates.yml
   - script: |

ci/tsan

@@ -35,3 +35,10 @@ race:WorkerEntry::set_next_sleeper
 # This ignores a false positive caused by `thread::park()`/`thread::unpark()`.
 # See: https://github.com/rust-lang/rust/pull/54806#issuecomment-436193353
 race:pthread_cond_destroy
+
+# Recent rand dependency updates and seeding changes have introduced
+# lazy_static's and other racy code. See:
+# https://github.com/tokio-rs/tokio/pull/1358#issuecomment-516172383
+race:RandomState*::build_hasher
+race:lazy_static::
+race:c2_chacha::guts

tokio-buf/README.md

@@ -2,6 +2,8 @@
 
 Asynchronous stream of byte buffers
 
+> **Note:** This crate has been **deprecated in tokio 0.2.x**.
+
 [Documenation](https://docs.rs/tokio-buf)
 
 ## Usage

tokio-buf/src/lib.rs

@@ -1,9 +1,10 @@
 #![doc(html_root_url = "https://docs.rs/tokio-buf/0.1.1")]
 #![deny(missing_docs, missing_debug_implementations, unreachable_pub)]
-#![cfg_attr(test, deny(warnings))]
 
 //! Asynchronous stream of bytes.
 //!
+//! > **Note:** This crate has been **deprecated in tokio 0.2.x**.
+//!
 //! This crate contains the `BufStream` trait and a number of combinators for
 //! this trait. The trait is similar to `Stream` in the `futures` library, but
 //! instead of yielding arbitrary values, it only yields types that implement

tokio-buf/tests/buf_stream.rs

@@ -4,4 +4,4 @@ use tokio_buf::BufStream;
 
 // Ensures that `BufStream` can be a trait object
 #[allow(dead_code)]
-fn obj(_: &mut BufStream<Item = u32, Error = ()>) {}
+fn obj(_: &mut dyn BufStream<Item = u32, Error = ()>) {}

tokio-codec/CHANGELOG.md

@@ -1,3 +1,7 @@
+# 0.1.2 (February 4, 2020)
+
+* Add `tokio 0.2.x` deprecation notice.
+
 # 0.1.1 (September 26, 2018)
 
 * Allow setting max line length with `LinesCodec` (#632)

tokio-codec/Cargo.toml

@@ -8,12 +8,12 @@ name = "tokio-codec"
 #   - README.md
 # - Update CHANGELOG.md.
 # - Create "v0.1.x" git tag.
-version = "0.1.1"
+version = "0.1.2"
 authors = ["Carl Lerche <[email protected]>", "Bryan Burgers <[email protected]>"]
 license = "MIT"
 repository = "https://github.com/tokio-rs/tokio"
 homepage = "https://tokio.rs"
-documentation = "https://docs.rs/tokio-codec/0.1.1/tokio_codec"
+documentation = "https://docs.rs/tokio-codec/0.1.2/tokio_codec"
 description = """
 Utilities for encoding and decoding frames.
 """

tokio-codec/README.md

@@ -2,6 +2,13 @@
 
 Utilities for encoding and decoding frames.
 
+> **Note:** This crate is **deprecated in tokio 0.2.x** and has been moved into
+> [`tokio_util::codec`] of the [`tokio-util` crate] behind the `codec` feature
+> flag.
+
+[`tokio_util::codec`]: https://docs.rs/tokio-util/latest/tokio_util/codec/index.html
+[`tokio-util` crate]: https://docs.rs/tokio-util/latest/tokio_util
+
 [Documentation](https://docs.rs/tokio-codec)
 
 ## Usage

tokio-codec/src/lib.rs

@@ -1,8 +1,15 @@
-#![deny(missing_docs, missing_debug_implementations, warnings)]
-#![doc(html_root_url = "https://docs.rs/tokio-codec/0.1.1")]
+#![deny(missing_docs, missing_debug_implementations)]
+#![doc(html_root_url = "https://docs.rs/tokio-codec/0.1.2")]
 
 //! Utilities for encoding and decoding frames.
 //!
+//! > **Note:** This crate is **deprecated in tokio 0.2.x** and has been moved
+//! into [`tokio_util::codec`] of the [`tokio-util` crate] behind the `codec`
+//! feature flag.
+//!
+//! [`tokio_util::codec`]: https://docs.rs/tokio-util/latest/tokio_util/codec/index.html
+//! [`tokio-util` crate]: https://docs.rs/tokio-util/latest/tokio_util
+//!
 //! Contains adapters to go from streams of bytes, [`AsyncRead`] and
 //! [`AsyncWrite`], to framed streams implementing [`Sink`] and [`Stream`].
 //! Framed streams are also known as [transports].

tokio-current-thread/CHANGELOG.md

@@ -1,3 +1,7 @@
+# 0.1.7 (February 4, 2020)
+ 
+* Add `tokio 0.2.x` deprecation notice.
+
 # 0.1.6 (March 22, 2019)
 
 ### Added

tokio-current-thread/Cargo.toml

@@ -8,8 +8,8 @@ name = "tokio-current-thread"
 #   - README.md
 # - Update CHANGELOG.md.
 # - Create "v0.1.x" git tag.
-version = "0.1.6"
-documentation = "https://docs.rs/tokio-current-thread/0.1.6/tokio_current_thread"
+version = "0.1.7"
+documentation = "https://docs.rs/tokio-current-thread/0.1.7/tokio_current_thread"
 repository = "https://github.com/tokio-rs/tokio"
 homepage = "https://github.com/tokio-rs/tokio"
 license = "MIT"

tokio-current-thread/README.md

@@ -2,6 +2,15 @@
 
 Single threaded executor for Tokio.
 
+> **Note:** This crate is **deprecated in tokio 0.2.x** and has been moved and
+> refactored into various places in the [`tokio`] crate. The closest replacement
+> is to make use of [`tokio::task::LocalSet::block_on`] which requires the
+> [`rt-util` feature].
+
+[`tokio`]: https://docs.rs/tokio/latest/tokio/index.html
+[`tokio::task::LocalSet::block_on`]: https://docs.rs/tokio/latest/tokio/task/struct.LocalSet.html#method.block_on
+[`rt-util` feature]: https://docs.rs/tokio/latest/tokio/index.html#feature-flags
+
 [Documentation](https://docs.rs/tokio-current-thread/0.1.6/tokio_current_thread/)
 
 ## Overview

tokio-current-thread/src/lib.rs

@@ -1,9 +1,17 @@
-#![doc(html_root_url = "https://docs.rs/tokio-current-thread/0.1.6")]
-#![deny(warnings, missing_docs, missing_debug_implementations)]
+#![doc(html_root_url = "https://docs.rs/tokio-current-thread/0.1.7")]
+#![deny(missing_docs, missing_debug_implementations)]
 
 //! A single-threaded executor which executes tasks on the same thread from which
 //! they are spawned.
 //!
+//! > **Note:** This crate is **deprecated in tokio 0.2.x** and has been moved
+//! > and refactored into various places in the [`tokio`] crate. The closest
+//! replacement is to make use of [`tokio::task::LocalSet::block_on`] which
+//! requires the [`rt-util` feature].
+//!
+//! [`tokio`]: https://docs.rs/tokio/latest/tokio/index.html
+//! [`tokio::task::LocalSet::block_on`]: https://docs.rs/tokio/latest/tokio/task/struct.LocalSet.html#method.block_on
+//! [`rt-util` feature]: https://docs.rs/tokio/latest/tokio/index.html#feature-flags
 //!
 //! The crate provides:
 //!
@@ -64,7 +72,7 @@ pub struct CurrentThread<P: Park = ParkThread> {
     spawn_handle: Handle,
 
     /// Receiver for futures spawned from other threads
-    spawn_receiver: mpsc::Receiver<Box<Future<Item = (), Error = ()> + Send + 'static>>,
+    spawn_receiver: mpsc::Receiver<Box<dyn Future<Item = (), Error = ()> + Send + 'static>>,
 
     /// The thread-local ID assigned to this executor.
     id: u64,
@@ -186,11 +194,15 @@ struct Borrow<'a, U: 'a> {
 }
 
 trait SpawnLocal {
-    fn spawn_local(&mut self, future: Box<Future<Item = (), Error = ()>>, already_counted: bool);
+    fn spawn_local(
+        &mut self,
+        future: Box<dyn Future<Item = (), Error = ()>>,
+        already_counted: bool,
+    );
 }
 
 struct CurrentRunner {
-    spawn: Cell<Option<*mut SpawnLocal>>,
+    spawn: Cell<Option<*mut dyn SpawnLocal>>,
     id: Cell<Option<u64>>,
 }
 
@@ -424,7 +436,7 @@ impl<P: Park> Drop for CurrentThread<P> {
 impl tokio_executor::Executor for CurrentThread {
     fn spawn(
         &mut self,
-        future: Box<Future<Item = (), Error = ()> + Send>,
+        future: Box<dyn Future<Item = (), Error = ()> + Send>,
     ) -> Result<(), SpawnError> {
         self.borrow().spawn_local(future, false);
         Ok(())
@@ -629,7 +641,7 @@ impl<'a, P: Park> fmt::Debug for Entered<'a, P> {
 /// Handle to spawn a future on the corresponding `CurrentThread` instance
 #[derive(Clone)]
 pub struct Handle {
-    sender: mpsc::Sender<Box<Future<Item = (), Error = ()> + Send + 'static>>,
+    sender: mpsc::Sender<Box<dyn Future<Item = (), Error = ()> + Send + 'static>>,
     num_futures: Arc<atomic::AtomicUsize>,
     shut_down: Cell<bool>,
     notify: executor::NotifyHandle,
@@ -731,7 +743,7 @@ impl TaskExecutor {
     /// Spawn a future onto the current `CurrentThread` instance.
     pub fn spawn_local(
         &mut self,
-        future: Box<Future<Item = (), Error = ()>>,
+        future: Box<dyn Future<Item = (), Error = ()>>,
     ) -> Result<(), SpawnError> {
         CURRENT.with(|current| match current.spawn.get() {
             Some(spawn) => {
@@ -746,7 +758,7 @@ impl TaskExecutor {
 impl tokio_executor::Executor for TaskExecutor {
     fn spawn(
         &mut self,
-        future: Box<Future<Item = (), Error = ()> + Send>,
+        future: Box<dyn Future<Item = (), Error = ()> + Send>,
     ) -> Result<(), SpawnError> {
         self.spawn_local(future)
     }
@@ -791,7 +803,11 @@ impl<'a, U: Unpark> Borrow<'a, U> {
 }
 
 impl<'a, U: Unpark> SpawnLocal for Borrow<'a, U> {
-    fn spawn_local(&mut self, future: Box<Future<Item = (), Error = ()>>, already_counted: bool) {
+    fn spawn_local(
+        &mut self,
+        future: Box<dyn Future<Item = (), Error = ()>>,
+        already_counted: bool,
+    ) {
         if !already_counted {
             // NOTE: we have a borrow of the Runtime, so we know that it isn't shut down.
             // NOTE: += 2 since LSB is the shutdown bit
@@ -804,7 +820,7 @@ impl<'a, U: Unpark> SpawnLocal for Borrow<'a, U> {
 // ===== impl CurrentRunner =====
 
 impl CurrentRunner {
-    fn set_spawn<F, R>(&self, spawn: &mut SpawnLocal, f: F) -> R
+    fn set_spawn<F, R>(&self, spawn: &mut dyn SpawnLocal, f: F) -> R
     where
         F: FnOnce() -> R,
     {
@@ -819,14 +835,14 @@ impl CurrentRunner {
 
         let _reset = Reset(self);
 
-        let spawn = unsafe { hide_lt(spawn as *mut SpawnLocal) };
+        let spawn = unsafe { hide_lt(spawn as *mut dyn SpawnLocal) };
         self.spawn.set(Some(spawn));
 
         f()
     }
 }
 
-unsafe fn hide_lt<'a>(p: *mut (SpawnLocal + 'a)) -> *mut (SpawnLocal + 'static) {
+unsafe fn hide_lt<'a>(p: *mut (dyn SpawnLocal + 'a)) -> *mut (dyn SpawnLocal + 'static) {
     use std::mem;
     mem::transmute(p)
 }

tokio-current-thread/src/scheduler.rs

@@ -125,7 +125,7 @@ enum Dequeue<U> {
 }
 
 /// Wraps a spawned boxed future
-struct Task(Spawn<Box<Future<Item = (), Error = ()>>>);
+struct Task(Spawn<Box<dyn Future<Item = (), Error = ()>>>);
 
 /// A task that is scheduled. `turn` must be called
 pub struct Scheduled<'a, U: 'a> {
@@ -171,7 +171,7 @@ where
         self.inner.clone().into()
     }
 
-    pub fn schedule(&mut self, item: Box<Future<Item = (), Error = ()>>) {
+    pub fn schedule(&mut self, item: Box<dyn Future<Item = (), Error = ()>>) {
         // Get the current scheduler tick
         let tick_num = self.inner.tick_num.load(SeqCst);
 
@@ -359,7 +359,7 @@ impl<'a, U: Unpark> Scheduled<'a, U> {
 }
 
 impl Task {
-    pub fn new(future: Box<Future<Item = (), Error = ()> + 'static>) -> Self {
+    pub fn new(future: Box<dyn Future<Item = (), Error = ()> + 'static>) -> Self {
         Task(executor::spawn(future))
     }
 }
@@ -687,8 +687,8 @@ unsafe impl<U: Unpark> UnsafeNotify for ArcNode<U> {
     }
 }
 
-unsafe fn hide_lt<U: Unpark>(p: *mut ArcNode<U>) -> *mut UnsafeNotify {
-    mem::transmute(p as *mut UnsafeNotify)
+unsafe fn hide_lt<U: Unpark>(p: *mut ArcNode<U>) -> *mut dyn UnsafeNotify {
+    mem::transmute(p as *mut dyn UnsafeNotify)
 }
 
 impl<U: Unpark> Node<U> {

tokio-current-thread/tests/current_thread.rs

@@ -22,7 +22,7 @@ use futures::sync::oneshot;
 
 mod from_block_on_all {
     use super::*;
-    fn test<F: Fn(Box<Future<Item = (), Error = ()>>) + 'static>(spawn: F) {
+    fn test<F: Fn(Box<dyn Future<Item = (), Error = ()>>) + 'static>(spawn: F) {
         let cnt = Rc::new(Cell::new(0));
         let c = cnt.clone();
 
@@ -102,7 +102,7 @@ fn spawn_many() {
 mod does_not_set_global_executor_by_default {
     use super::*;
 
-    fn test<F: Fn(Box<Future<Item = (), Error = ()> + Send>) -> Result<(), E> + 'static, E>(
+    fn test<F: Fn(Box<dyn Future<Item = (), Error = ()> + Send>) -> Result<(), E> + 'static, E>(
         spawn: F,
     ) {
         block_on_all(lazy(|| {
@@ -127,7 +127,7 @@ mod does_not_set_global_executor_by_default {
 mod from_block_on_future {
     use super::*;
 
-    fn test<F: Fn(Box<Future<Item = (), Error = ()>>)>(spawn: F) {
+    fn test<F: Fn(Box<dyn Future<Item = (), Error = ()>>)>(spawn: F) {
         let cnt = Rc::new(Cell::new(0));
 
         let mut tokio_current_thread = CurrentThread::new();
@@ -181,8 +181,8 @@ mod outstanding_tasks_are_dropped_when_executor_is_dropped {
 
     fn test<F, G>(spawn: F, dotspawn: G)
     where
-        F: Fn(Box<Future<Item = (), Error = ()>>) + 'static,
-        G: Fn(&mut CurrentThread, Box<Future<Item = (), Error = ()>>),
+        F: Fn(Box<dyn Future<Item = (), Error = ()>>) + 'static,
+        G: Fn(&mut CurrentThread, Box<dyn Future<Item = (), Error = ()>>),
     {
         let mut rc = Rc::new(());
 
@@ -383,8 +383,8 @@ mod and_turn {
 
     fn test<F, G>(spawn: F, dotspawn: G)
     where
-        F: Fn(Box<Future<Item = (), Error = ()>>) + 'static,
-        G: Fn(&mut CurrentThread, Box<Future<Item = (), Error = ()>>),
+        F: Fn(Box<dyn Future<Item = (), Error = ()>>) + 'static,
+        G: Fn(&mut CurrentThread, Box<dyn Future<Item = (), Error = ()>>),
     {
         let cnt = Rc::new(Cell::new(0));
         let c = cnt.clone();
@@ -445,7 +445,6 @@ mod and_turn {
             },
         );
     }
-
 }
 
 mod in_drop {
@@ -459,7 +458,7 @@ mod in_drop {
     }
 
     struct MyFuture {
-        _data: Box<Any>,
+        _data: Box<dyn Any>,
     }
 
     impl Future for MyFuture {
@@ -473,8 +472,8 @@ mod in_drop {
 
     fn test<F, G>(spawn: F, dotspawn: G)
     where
-        F: Fn(Box<Future<Item = (), Error = ()>>) + 'static,
-        G: Fn(&mut CurrentThread, Box<Future<Item = (), Error = ()>>),
+        F: Fn(Box<dyn Future<Item = (), Error = ()>>) + 'static,
+        G: Fn(&mut CurrentThread, Box<dyn Future<Item = (), Error = ()>>),
     {
         let mut tokio_current_thread = CurrentThread::new();
 
@@ -520,7 +519,6 @@ mod in_drop {
             },
         );
     }
-
 }
 
 #[test]

tokio-executor/CHANGELOG.md

@@ -1,3 +1,13 @@
+# 0.1.10 (February 4, 2020)
+
+* Add `tokio 0.2.x` deprecation notice.
+
+# 0.1.9 (November 27, 2019)
+
+### Added
+- Add `executor::set_default` which behaves like `with_default` but returns a
+  drop guard (#1725).
+
 # 0.1.8 (June 2, 2019)
 
 ### Added

tokio-executor/Cargo.toml

@@ -8,8 +8,8 @@ name = "tokio-executor"
 #   - README.md
 # - Update CHANGELOG.md.
 # - Create "v0.1.x" git tag.
-version = "0.1.8"
-documentation = "https://docs.rs/tokio-executor/0.1.7/tokio_executor"
+version = "0.1.10"
+documentation = "https://docs.rs/tokio-executor/0.1.10/tokio_executor"
 repository = "https://github.com/tokio-rs/tokio"
 homepage = "https://github.com/tokio-rs/tokio"
 license = "MIT"
@@ -21,7 +21,7 @@ keywords = ["futures", "tokio"]
 categories = ["concurrency", "asynchronous"]
 
 [dependencies]
-crossbeam-utils = "0.6.2"
+crossbeam-utils = "0.7.0"
 futures = "0.1.19"
 
 [dev-dependencies]

tokio-executor/README.md

@@ -2,7 +2,13 @@
 
 Task execution related traits and utilities.
 
-[Documentation](https://docs.rs/tokio-executor/0.1.8/tokio_executor)
+This crate is **deprecated in tokio 0.2.x** and has been moved and refactored
+into various places in the [`tokio::runtime`] module of the [`tokio`] crate.
+
+[`tokio::runtime`]: https://docs.rs/tokio/latest/tokio/runtime/index.html
+[`tokio`]: https://docs.rs/tokio/latest/tokio/index.html
+
+[Documentation](https://docs.rs/tokio-executor/0.1.9/tokio_executor)
 
 ## Overview
 
@@ -31,10 +37,10 @@ executor, including:
 
 * [`Park`] abstracts over blocking and unblocking the current thread.
 
-[`Executor`]: https://docs.rs/tokio-executor/0.1.8/tokio_executor/trait.Executor.html
-[`enter`]: https://docs.rs/tokio-executor/0.1.8/tokio_executor/fn.enter.html
-[`DefaultExecutor`]: https://docs.rs/tokio-executor/0.1.8/tokio_executor/struct.DefaultExecutor.html
-[`Park`]: https://docs.rs/tokio-executor/0.1.8/tokio_executor/park/trait.Park.html
+[`Executor`]: https://docs.rs/tokio-executor/0.1.9/tokio_executor/trait.Executor.html
+[`enter`]: https://docs.rs/tokio-executor/0.1.9/tokio_executor/fn.enter.html
+[`DefaultExecutor`]: https://docs.rs/tokio-executor/0.1.9/tokio_executor/struct.DefaultExecutor.html
+[`Park`]: https://docs.rs/tokio-executor/0.1.9/tokio_executor/park/trait.Park.html
 
 ## License
 

tokio-executor/src/enter.rs

@@ -11,7 +11,7 @@ thread_local!(static ENTERED: Cell<bool> = Cell::new(false));
 ///
 /// For more details, see [`enter` documentation](fn.enter.html)
 pub struct Enter {
-    on_exit: Vec<Box<Callback>>,
+    on_exit: Vec<Box<dyn Callback>>,
     permanent: bool,
 }
 

tokio-executor/src/executor.rs

@@ -94,7 +94,7 @@ pub trait Executor {
     /// ```
     fn spawn(
         &mut self,
-        future: Box<Future<Item = (), Error = ()> + Send>,
+        future: Box<dyn Future<Item = (), Error = ()> + Send>,
     ) -> Result<(), SpawnError>;
 
     /// Provides a best effort **hint** to whether or not `spawn` will succeed.
@@ -140,7 +140,7 @@ pub trait Executor {
 impl<E: Executor + ?Sized> Executor for Box<E> {
     fn spawn(
         &mut self,
-        future: Box<Future<Item = (), Error = ()> + Send>,
+        future: Box<dyn Future<Item = (), Error = ()> + Send>,
     ) -> Result<(), SpawnError> {
         (**self).spawn(future)
     }

tokio-executor/src/global.rs

@@ -19,6 +19,13 @@ pub struct DefaultExecutor {
     _dummy: (),
 }
 
+/// Ensures that the executor is removed from the thread-local context
+/// when leaving the scope. This handles cases that involve panicking.
+#[derive(Debug)]
+pub struct DefaultGuard {
+    _p: (),
+}
+
 impl DefaultExecutor {
     /// Returns a handle to the default executor for the current context.
     ///
@@ -37,7 +44,7 @@ impl DefaultExecutor {
     }
 
     #[inline]
-    fn with_current<F: FnOnce(&mut Executor) -> R, R>(f: F) -> Option<R> {
+    fn with_current<F: FnOnce(&mut dyn Executor) -> R, R>(f: F) -> Option<R> {
         EXECUTOR.with(
             |current_executor| match current_executor.replace(State::Active) {
                 State::Ready(executor_ptr) => {
@@ -57,7 +64,7 @@ enum State {
     // default executor not defined
     Empty,
     // default executor is defined and ready to be used
-    Ready(*mut Executor),
+    Ready(*mut dyn Executor),
     // default executor is currently active (used to detect recursive calls)
     Active,
 }
@@ -72,7 +79,7 @@ thread_local! {
 impl super::Executor for DefaultExecutor {
     fn spawn(
         &mut self,
-        future: Box<Future<Item = (), Error = ()> + Send>,
+        future: Box<dyn Future<Item = (), Error = ()> + Send>,
     ) -> Result<(), SpawnError> {
         DefaultExecutor::with_current(|executor| executor.spawn(future))
             .unwrap_or_else(|| Err(SpawnError::shutdown()))
@@ -175,6 +182,11 @@ where
     T: Executor,
     F: FnOnce(&mut Enter) -> R,
 {
+    unsafe fn hide_lt<'a>(p: *mut (dyn Executor + 'a)) -> *mut (dyn Executor + 'static) {
+        use std::mem;
+        mem::transmute(p)
+    }
+
     EXECUTOR.with(|cell| {
         match cell.get() {
             State::Ready(_) | State::Active => {
@@ -210,9 +222,47 @@ where
     })
 }
 
-unsafe fn hide_lt<'a>(p: *mut (Executor + 'a)) -> *mut (Executor + 'static) {
-    use std::mem;
-    mem::transmute(p)
+/// Sets `executor` as the default executor, returning a guard that unsets it when
+/// dropped.
+///
+/// # Panics
+///
+/// This function panics if there already is a default executor set.
+pub fn set_default<T>(executor: T) -> DefaultGuard
+where
+    T: Executor + 'static,
+{
+    EXECUTOR.with(|cell| {
+        match cell.get() {
+            State::Ready(_) | State::Active => {
+                panic!("default executor already set for execution context")
+            }
+            _ => {}
+        }
+
+        // Ensure that the executor will outlive the call to set_default, even
+        // if the drop guard is never dropped due to calls to `mem::forget` or
+        // similar.
+        let executor = Box::new(executor);
+
+        cell.set(State::Ready(Box::into_raw(executor)));
+    });
+
+    DefaultGuard { _p: () }
+}
+
+impl Drop for DefaultGuard {
+    fn drop(&mut self) {
+        let _ = EXECUTOR.try_with(|cell| {
+            if let State::Ready(prev) = cell.replace(State::Empty) {
+                // drop the previous executor.
+                unsafe {
+                    let prev = Box::from_raw(prev);
+                    drop(prev);
+                };
+            }
+        });
+    }
 }
 
 #[cfg(test)]

tokio-executor/src/lib.rs

@@ -1,7 +1,5 @@
-#![deny(missing_docs, missing_debug_implementations, warnings)]
-#![doc(html_root_url = "https://docs.rs/tokio-executor/0.1.8")]
-// Our MSRV doesn't allow us to fix these warnings yet
-#![allow(rust_2018_idioms)]
+#![deny(missing_docs, missing_debug_implementations)]
+#![doc(html_root_url = "https://docs.rs/tokio-executor/0.1.10")]
 
 //! Task execution related traits and utilities.
 //!
@@ -66,5 +64,5 @@ mod typed;
 pub use enter::{enter, exit, Enter, EnterError};
 pub use error::SpawnError;
 pub use executor::Executor;
-pub use global::{spawn, with_default, DefaultExecutor};
+pub use global::{set_default, spawn, with_default, DefaultExecutor, DefaultGuard};
 pub use typed::TypedExecutor;

tokio-executor/src/park.rs

@@ -128,13 +128,13 @@ pub trait Unpark: Sync + Send + 'static {
     fn unpark(&self);
 }
 
-impl Unpark for Box<Unpark> {
+impl Unpark for Box<dyn Unpark> {
     fn unpark(&self) {
         (**self).unpark()
     }
 }
 
-impl Unpark for Arc<Unpark> {
+impl Unpark for Arc<dyn Unpark> {
     fn unpark(&self) {
         (**self).unpark()
     }

tokio-executor/tests/executor.rs

@@ -10,7 +10,7 @@ mod out_of_executor_context {
 
     fn test<F, E>(spawn: F)
     where
-        F: Fn(Box<Future<Item = (), Error = ()> + Send>) -> Result<(), E>,
+        F: Fn(Box<dyn Future<Item = (), Error = ()> + Send>) -> Result<(), E>,
     {
         let res = spawn(Box::new(lazy(|| Ok(()))));
         assert!(res.is_err());

tokio-fs/CHANGELOG.md

@@ -1,3 +1,7 @@
+# 0.1.7 (February 4, 2020)
+
+* Add `tokio 0.2.x` deprecation notice.
+
 # 0.1.6 (March 1, 2019)
 
 ### Added

tokio-fs/Cargo.toml

@@ -8,13 +8,13 @@ name = "tokio-fs"
 #   - README.md
 # - Update CHANGELOG.md.
 # - Create "v0.1.x" git tag.
-version = "0.1.6"
+version = "0.1.7"
 authors = ["Carl Lerche <[email protected]>"]
 license = "MIT"
 readme = "README.md"
 repository = "https://github.com/tokio-rs/tokio"
 homepage = "https://tokio.rs"
-documentation = "https://docs.rs/tokio-fs/0.1.6/tokio_fs"
+documentation = "https://docs.rs/tokio-fs/0.1.7/tokio_fs"
 description = """
 Filesystem API for Tokio.
 """
@@ -27,9 +27,8 @@ tokio-threadpool = "0.1.3"
 tokio-io = "0.1.6"
 
 [dev-dependencies]
-rand = "0.6"
-tempfile = "3"
-tempdir = "0.3"
+rand = "0.7"
+tempfile = "~3.1.0"
 tokio-io = "0.1.6"
 tokio-codec = "0.1.0"
 tokio = "0.1.7"

tokio-fs/README.md

@@ -2,6 +2,12 @@
 
 Asynchronous filesystem manipulation operations (and stdin, stdout, stderr).
 
+This crate has been **deprecated in tokio 0.2.x** and has been moved into
+[`tokio::fs`] behind the `fs` [feature flag].
+
+[`tokio::fs`]: https://docs.rs/tokio/latest/tokio/fs/index.html
+[feature flag]: https://docs.rs/tokio/latest/tokio/index.html#feature-flags
+
 [Documentation](https://docs.rs/tokio-fs/0.1.6/tokio_fs)
 
 ## Overview

tokio-fs/examples/std-echo.rs

@@ -1,5 +1,5 @@
 //! Echo everything received on STDIN to STDOUT.
-#![deny(deprecated, warnings)]
+#![deny(deprecated)]
 
 extern crate futures;
 extern crate tokio_codec;
@@ -14,7 +14,7 @@ use futures::{Future, Sink, Stream};
 
 use std::io;
 
-pub fn main() -> Result<(), Box<std::error::Error>> {
+pub fn main() -> Result<(), Box<dyn std::error::Error>> {
     let pool = Builder::new().pool_size(1).build();
 
     pool.spawn({

tokio-fs/src/lib.rs

@@ -1,5 +1,5 @@
-#![deny(missing_docs, missing_debug_implementations, warnings)]
-#![doc(html_root_url = "https://docs.rs/tokio-fs/0.1.6")]
+#![deny(missing_docs, missing_debug_implementations)]
+#![doc(html_root_url = "https://docs.rs/tokio-fs/0.1.7")]
 
 //! Asynchronous file and standard stream adaptation.
 //!

tokio-fs/tests/dir.rs

@@ -1,18 +1,18 @@
 extern crate futures;
-extern crate tempdir;
+extern crate tempfile;
 extern crate tokio_fs;
 
 use futures::{Future, Stream};
 use std::fs;
 use std::sync::{Arc, Mutex};
-use tempdir::TempDir;
+use tempfile::tempdir;
 use tokio_fs::*;
 
 mod pool;
 
 #[test]
 fn create() {
-    let base_dir = TempDir::new("base").unwrap();
+    let base_dir = tempdir().unwrap();
     let new_dir = base_dir.path().join("foo");
 
     pool::run({ create_dir(new_dir.clone()) });
@@ -22,7 +22,7 @@ fn create() {
 
 #[test]
 fn create_all() {
-    let base_dir = TempDir::new("base").unwrap();
+    let base_dir = tempdir().unwrap();
     let new_dir = base_dir.path().join("foo").join("bar");
 
     pool::run({ create_dir_all(new_dir.clone()) });
@@ -32,7 +32,7 @@ fn create_all() {
 
 #[test]
 fn remove() {
-    let base_dir = TempDir::new("base").unwrap();
+    let base_dir = tempdir().unwrap();
     let new_dir = base_dir.path().join("foo");
 
     fs::create_dir(new_dir.clone()).unwrap();
@@ -44,7 +44,7 @@ fn remove() {
 
 #[test]
 fn read() {
-    let base_dir = TempDir::new("base").unwrap();
+    let base_dir = tempdir().unwrap();
 
     let p = base_dir.path();
     fs::create_dir(p.join("aa")).unwrap();

tokio-fs/tests/link.rs

@@ -1,19 +1,19 @@
 extern crate futures;
-extern crate tempdir;
+extern crate tempfile;
 extern crate tokio_fs;
 
 use futures::Future;
 use std::fs;
 use std::io::prelude::*;
 use std::io::BufReader;
-use tempdir::TempDir;
+use tempfile::tempdir;
 use tokio_fs::*;
 
 mod pool;
 
 #[test]
 fn test_hard_link() {
-    let dir = TempDir::new("base").unwrap();
+    let dir = tempdir().unwrap();
     let src = dir.path().join("src.txt");
     let dst = dir.path().join("dst.txt");
 
@@ -38,7 +38,7 @@ fn test_hard_link() {
 #[cfg(unix)]
 #[test]
 fn test_symlink() {
-    let dir = TempDir::new("base").unwrap();
+    let dir = tempdir().unwrap();
     let src = dir.path().join("src.txt");
     let dst = dir.path().join("dst.txt");
 

tokio-futures/src/lib.rs

@@ -2,7 +2,6 @@
 #![feature(await_macro)]
 #![doc(html_root_url = "https://docs.rs/tokio-futures/0.1.0")]
 #![deny(missing_docs, missing_debug_implementations)]
-#![cfg_attr(test, deny(warnings))]
 
 //! A preview of Tokio w/ `async` / `await` support.
 

tokio-io/CHANGELOG.md

@@ -1,3 +1,7 @@
+# 0.1.13 (February 4, 2020)
+
+* Add `tokio 0.2.x` deprecation notice.
+
 # 0.1.12 (March 1, 2019)
 
 ### Added

tokio-io/Cargo.toml

@@ -8,12 +8,12 @@ name = "tokio-io"
 #   - README.md
 # - Update CHANGELOG.md.
 # - Create "v0.1.x" git tag.
-version = "0.1.12"
+version = "0.1.13"
 authors = ["Carl Lerche <[email protected]>"]
 license = "MIT"
 repository = "https://github.com/tokio-rs/tokio"
 homepage = "https://tokio.rs"
-documentation = "https://docs.rs/tokio-io/0.1.12/tokio_io"
+documentation = "https://docs.rs/tokio-io/0.1.13/tokio_io"
 description = """
 Core I/O primitives for asynchronous I/O in Rust.
 """

tokio-io/README.md

@@ -4,6 +4,11 @@ Core I/O abstractions for the Tokio stack.
 
 [![Build Status](https://travis-ci.org/tokio-rs/tokio-io.svg?branch=master)](https://travis-ci.org/tokio-rs/tokio-io)
 
+> **Note:** This crate has been **deprecated in tokio 0.2.x** and has been moved
+> into [`tokio::io`].
+
+[`tokio::io`]: https://docs.rs/tokio/latest/tokio/io/index.html
+
 [Documentation](https://docs.rs/tokio-io/0.1.12/tokio_io)
 
 ## Usage

tokio-io/src/_tokio_codec/mod.rs

@@ -10,7 +10,7 @@
 //! [`Stream`]: #
 //! [transports]: #
 
-#![deny(missing_docs, missing_debug_implementations, warnings)]
+#![deny(missing_docs, missing_debug_implementations)]
 #![doc(hidden, html_root_url = "https://docs.rs/tokio-codec/0.1.0")]
 
 // _tokio_codec are the items that belong in the `tokio_codec` crate. However, because we need to

tokio-io/src/lib.rs

@@ -1,8 +1,13 @@
-#![deny(missing_docs, missing_debug_implementations, warnings)]
-#![doc(html_root_url = "https://docs.rs/tokio-io/0.1.12")]
+#![deny(missing_docs, missing_debug_implementations)]
+#![doc(html_root_url = "https://docs.rs/tokio-io/0.1.13")]
 
 //! Core I/O traits and combinators when working with Tokio.
 //!
+//! > **Note:** This crate has been **deprecated in tokio 0.2.x** and has been
+//! > moved into [`tokio::io`].
+//!
+//! [`tokio::io`]: https://docs.rs/tokio/latest/tokio/io/index.html
+//!
 //! A description of the high-level I/O combinators can be [found online] in
 //! addition to a description of the [low level details].
 //!
@@ -21,10 +26,10 @@ use std::io as std_io;
 use futures::{Future, Stream};
 
 /// A convenience typedef around a `Future` whose error component is `io::Error`
-pub type IoFuture<T> = Box<Future<Item = T, Error = std_io::Error> + Send>;
+pub type IoFuture<T> = Box<dyn Future<Item = T, Error = std_io::Error> + Send>;
 
 /// A convenience typedef around a `Stream` whose error component is `io::Error`
-pub type IoStream<T> = Box<Stream<Item = T, Error = std_io::Error> + Send>;
+pub type IoStream<T> = Box<dyn Stream<Item = T, Error = std_io::Error> + Send>;
 
 /// A convenience macro for working with `io::Result<T>` from the `Read` and
 /// `Write` traits.
@@ -65,6 +70,6 @@ pub use self::async_write::AsyncWrite;
 
 fn _assert_objects() {
     fn _assert<T>() {}
-    _assert::<Box<AsyncRead>>();
-    _assert::<Box<AsyncWrite>>();
+    _assert::<Box<dyn AsyncRead>>();
+    _assert::<Box<dyn AsyncWrite>>();
 }

tokio-process/CHANGELOG.md

@@ -0,0 +1,75 @@
+## 0.2.5 - 2020-02-04
+
+* Add `tokio 0.2.x` deprecation notice.
+
+## 0.2.4 - 2019-06-21
+### Fixed
+* Proccesses "leaked" via `Child::forget` now reaped rather than left as zombies
+for the duration of the parent process.
+* Dropping a `Child` process no longer blocks the caller until the process fully
+exits. This avoids a pathological deadlock if the kernel doesn't kill the child.
+
+### Changed
+* Updated the example program for reading lines from a child process to be more
+flexible to be copy/pasted and iterated upon.
+
+## 0.2.3 - 2018-11-01
+### Added
+* `ChildStd{in, out, err}` now implement `AsRawFd`/`AsRawHandle` on Unix/Windows
+systems, respectively.
+
+## 0.2.2 - 2018-05-27
+### Fixed
+- Fixed a pathological situation where a signal could be missed if it arrived
+after polling the child but before registering for a new notification
+
+## 0.2.1 - 2018-05-18
+### Changed
+- **Breaking**: asynchronous spawning of a child process now requires using a
+reactor handle from the `tokio` crate instead of the `tokio-core` crate
+- Child processes may be spawned without specifying a `tokio` handle at all
+(the current/default reactor handle will be used)
+### Removed
+- **Breaking**: removed all previously deprecated items
+
+## 0.1.6 - 2018-05-09
+### Fixed
+- On Unix systems, any child processes that are `kill`ed (or implicitly killed
+via dropping the child without calling `forget`) are no longer left in a zombie
+state, which allows the OS to reclaim the process.
+
+## 0.1.5 - 2018-01-03
+### Changed
+- Minimum required version of `winapi` has been bumped to `0.3`.
+
+## 0.1.4 - 2017-06-25
+### Fixed
+- Added missing `Debug` impls on all types.
+- Added missing `must_use` annotations on all futures.
+- Ensure `status_async` closes child's stdio handles after spawning in order
+to prevent potential deadlocks when attempting to interact with any pipes held
+by the parent process.
+
+## 0.1.3 - 2017-03-15
+### Changed
+- Minimum required version of `futures` has been bumped to `0.1.11`.
+- Minimum required version of `mio` has been bumped to `0.6.5`.
+- Minimum required version of `tokio-core` has been bumped to `0.1.6`.
+
+## 0.1.2 - 2017-01-24
+### Changed
+- Minimum required version of `tokio-signal` has been bumped to `0.1.2`.
+### Fixed
+- The event loop which spawns the first async child no longer needs to be kept
+alive for subsequent child spawns to make progress.
+
+## 0.1.1 - 2016-12-19
+### Added
+- Support performing async I/O operations on the child's stdio handles.
+### Changed
+- Functionality has been reimplemented as the `CommandExt` extension trait
+(implemented directly on `std::process::Command`) instead of going through
+the locally vendored `Command` type.
+
+## 0.1.0 - 2016-09-10
+- First release!

tokio-process/Cargo.toml

@@ -0,0 +1,55 @@
+[package]
+name = "tokio-process"
+# When releasing to crates.io:
+# - Update html_root_url.
+# - Update CHANGELOG.md.
+# - Create "X.Y.Z" git tag.
+version = "0.2.5"
+edition = "2018"
+authors = ["Tokio Contributors <[email protected]>"]
+license = "MIT"
+repository = "https://github.com/tokio-rs/tokio"
+homepage = "https://github.com/tokio-rs/tokio"
+documentation = "https://docs.rs/tokio-process/0.2.5/tokio_process"
+description = """
+An implementation of an asynchronous process management backed futures.
+"""
+categories = ["asynchronous"]
+
+[dependencies]
+futures = "0.1.11"
+tokio-io = "0.1"
+tokio-reactor = "0.1"
+
+[dev-dependencies]
+failure = "0.1"
+log = "0.4"
+
+[dev-dependencies.tokio]
+version = "0.1"
+default-features = false
+features = ["rt-full"]
+
+[target.'cfg(windows)'.dependencies]
+mio-named-pipes = "0.1"
+
+[target.'cfg(windows)'.dependencies.winapi]
+version = "0.3"
+features = [
+  "handleapi",
+  "winerror",
+  "minwindef",
+  "processthreadsapi",
+  "synchapi",
+  "threadpoollegacyapiset",
+  "winbase",
+  "winnt",
+]
+
+[target.'cfg(unix)'.dependencies]
+crossbeam-queue = "0.1.2"
+lazy_static = "1.3"
+libc = "0.2"
+log = "0.4"
+mio = "0.6.5"
+tokio-signal = "0.2.5"

tokio-process/LICENSE

@@ -0,0 +1,25 @@
+Copyright (c) 2019 Tokio Contributors
+
+Permission is hereby granted, free of charge, to any
+person obtaining a copy of this software and associated
+documentation files (the "Software"), to deal in the
+Software without restriction, including without
+limitation the rights to use, copy, modify, merge,
+publish, distribute, sublicense, and/or sell copies of
+the Software, and to permit persons to whom the Software
+is furnished to do so, subject to the following
+conditions:
+
+The above copyright notice and this permission notice
+shall be included in all copies or substantial portions
+of the Software.
+
+THE SOFTWARE IS PROVIDED "AS IS", WITHOUT WARRANTY OF
+ANY KIND, EXPRESS OR IMPLIED, INCLUDING BUT NOT LIMITED
+TO THE WARRANTIES OF MERCHANTABILITY, FITNESS FOR A
+PARTICULAR PURPOSE AND NONINFRINGEMENT. IN NO EVENT
+SHALL THE AUTHORS OR COPYRIGHT HOLDERS BE LIABLE FOR ANY
+CLAIM, DAMAGES OR OTHER LIABILITY, WHETHER IN AN ACTION
+OF CONTRACT, TORT OR OTHERWISE, ARISING FROM, OUT OF OR
+IN CONNECTION WITH THE SOFTWARE OR THE USE OR OTHER
+DEALINGS IN THE SOFTWARE.

tokio-process/README.md

@@ -0,0 +1,54 @@
+# tokio-process
+
+An implementation of process management for Tokio
+
+> This crate has been **deprecated in tokio 0.2.x** and has been moved into
+> [`tokio::process`] behind the `process` [feature flag].
+
+[`tokio::process`]: https://docs.rs/tokio/latest/tokio/process/index.html
+[feature flag]: https://docs.rs/tokio/latest/tokio/index.html#feature-flags
+
+[Documentation](https://docs.rs/tokio-process/0.2.4/tokio_process)
+
+## Usage
+
+First, add this to your `Cargo.toml`:
+
+```toml
+[dependencies]
+tokio-process = "0.2"
+```
+
+Next you can use this in conjunction with the `tokio` and `futures` crates:
+
+```rust,no_run
+use std::process::Command;
+
+use futures::Future;
+use tokio_process::CommandExt;
+
+fn main() {
+    // Use the standard library's `Command` type to build a process and
+    // then execute it via the `CommandExt` trait.
+    let child = Command::new("echo").arg("hello").arg("world").spawn_async();
+
+    // Make sure our child succeeded in spawning and process the result
+    let future = child
+        .expect("failed to spawn")
+        .map(|status| println!("exit status: {}", status))
+        .map_err(|e| panic!("failed to wait for exit: {}", e));
+
+    // Send the future to the tokio runtime for execution
+    tokio::run(future)
+}
+```
+
+## License
+
+This project is licensed under the [MIT license](./LICENSE).
+
+### Contribution
+
+Unless you explicitly state otherwise, any contribution intentionally submitted
+for inclusion in Tokio by you, shall be licensed as MIT, without any additional
+terms or conditions.

tokio-process/src/bin/cat.rs

@@ -0,0 +1,19 @@
+// A cat-like utility that can be used as a subprocess to test I/O
+// stream communication.
+use std::io;
+use std::io::Write;
+
+fn main() {
+    let stdin = io::stdin();
+    let mut stdout = io::stdout();
+    let mut line = String::new();
+    loop {
+        line.clear();
+        stdin.read_line(&mut line).unwrap();
+        if line.is_empty() {
+            break;
+        }
+        stdout.write_all(line.as_bytes()).unwrap();
+    }
+    stdout.flush().unwrap();
+}

tokio-process/src/bin/exit.rs

@@ -0,0 +1,5 @@
+#[allow(dead_code)]
+
+fn main() {
+    std::process::exit(std::env::args().nth(1).unwrap().parse().unwrap());
+}

tokio-process/src/kill.rs

@@ -0,0 +1,13 @@
+use std::io;
+
+/// An interface for killing a running process.
+pub(crate) trait Kill {
+    /// Forcefully kill the process.
+    fn kill(&mut self) -> io::Result<()>;
+}
+
+impl<'a, T: 'a + Kill> Kill for &'a mut T {
+    fn kill(&mut self) -> io::Result<()> {
+        (**self).kill()
+    }
+}

tokio-process/src/lib.rs

@@ -0,0 +1,876 @@
+//! An implementation of asynchronous process management for Tokio.
+//!
+//! > This crate has been **deprecated in tokio 0.2.x** and has been moved into
+//! > [`tokio::process`] behind the `process` [feature flag].
+//!
+//! [`tokio::process`]: https://docs.rs/tokio/latest/tokio/process/index.html
+//! [feature flag]: https://docs.rs/tokio/latest/tokio/index.html#feature-flags
+//!
+//! This crate provides a `CommandExt` trait to enhance the functionality of the
+//! `Command` type in the standard library. The three methods provided by this
+//! trait mirror the "spawning" methods in the standard library. The
+//! `CommandExt` trait in this crate, though, returns "future aware" types that
+//! interoperate with Tokio. The asynchronous process support is provided
+//! through signal handling on Unix and system APIs on Windows.
+//!
+//! # Examples
+//!
+//! Here's an example program which will spawn `echo hello world` and then wait
+//! for it using an event loop.
+//!
+//! ```no_run
+//! extern crate futures;
+//! extern crate tokio;
+//! extern crate tokio_process;
+//!
+//! use std::process::Command;
+//!
+//! use futures::Future;
+//! use tokio_process::CommandExt;
+//!
+//! fn main() {
+//!     // Use the standard library's `Command` type to build a process and
+//!     // then execute it via the `CommandExt` trait.
+//!     let child = Command::new("echo").arg("hello").arg("world")
+//!                         .spawn_async();
+//!
+//!     // Make sure our child succeeded in spawning and process the result
+//!     let future = child.expect("failed to spawn")
+//!         .map(|status| println!("exit status: {}", status))
+//!         .map_err(|e| panic!("failed to wait for exit: {}", e));
+//!
+//!     // Send the future to the tokio runtime for execution
+//!     tokio::run(future)
+//! }
+//! ```
+//!
+//! Next, let's take a look at an example where we not only spawn `echo hello
+//! world` but we also capture its output.
+//!
+//! ```no_run
+//! extern crate futures;
+//! extern crate tokio;
+//! extern crate tokio_process;
+//!
+//! use std::process::Command;
+//!
+//! use futures::Future;
+//! use tokio_process::CommandExt;
+//!
+//! fn main() {
+//!     // Like above, but use `output_async` which returns a future instead of
+//!     // immediately returning the `Child`.
+//!     let output = Command::new("echo").arg("hello").arg("world")
+//!                         .output_async();
+//!
+//!     let future = output.map_err(|e| panic!("failed to collect output: {}", e))
+//!         .map(|output| {
+//!             assert!(output.status.success());
+//!             assert_eq!(output.stdout, b"hello world\n");
+//!         });
+//!
+//!     tokio::run(future);
+//! }
+//! ```
+//!
+//! We can also read input line by line.
+//!
+//! ```no_run
+//! extern crate failure;
+//! extern crate futures;
+//! extern crate tokio;
+//! extern crate tokio_process;
+//! extern crate tokio_io;
+//!
+//! use failure::Error;
+//! use futures::{Future, Stream};
+//! use std::io::BufReader;
+//! use std::process::{Command, Stdio};
+//! use tokio_process::{Child, ChildStdout, CommandExt};
+//!
+//! fn lines_stream(child: &mut Child) -> impl Stream<Item = String, Error = Error> + Send + 'static {
+//!     let stdout = child.stdout().take()
+//!         .expect("child did not have a handle to stdout");
+//!
+//!     tokio_io::io::lines(BufReader::new(stdout))
+//!         // Convert any io::Error into a failure::Error for better flexibility
+//!         .map_err(|e| Error::from(e))
+//!         // We print each line we've received here as an example of a way we can
+//!         // do something with the data. This can be changed to map the data to
+//!         // something else, or to consume it differently.
+//!         .inspect(|line| println!("Line: {}", line))
+//! }
+//!
+//! fn main() {
+//!     // Lazily invoke any code so it can run directly within the tokio runtime
+//!     tokio::run(futures::lazy(|| {
+//!         let mut cmd = Command::new("cat");
+//!
+//!         // Specify that we want the command's standard output piped back to us.
+//!         // By default, standard input/output/error will be inherited from the
+//!         // current process (for example, this means that standard input will
+//!         // come from the keyboard and standard output/error will go directly to
+//!         // the terminal if this process is invoked from the command line).
+//!         cmd.stdout(Stdio::piped());
+//!
+//!         let mut child = cmd.spawn_async()
+//!             .expect("failed to spawn command");
+//!
+//!         let lines = lines_stream(&mut child);
+//!
+//!         // Spawning into the tokio runtime requires that the future's Item and
+//!         // Error are both `()`. This is because tokio doesn't know what to do
+//!         // with any results or errors, so it requires that we've handled them!
+//!         //
+//!         // We can replace these sample usages of the child's exit status (or
+//!         // an encountered error) perform some different actions if needed!
+//!         // For example, log the error, or send a message on a channel, etc.
+//!         let child_future = child
+//!                 .map(|status| println!("child status was: {}", status))
+//!                 .map_err(|e| panic!("error while running child: {}", e));
+//!
+//!         // Ensure the child process can live on within the runtime, otherwise
+//!         // the process will get killed if this handle is dropped
+//!         tokio::spawn(child_future);
+//!
+//!         // Return a future to tokio. This is the same as calling using
+//!         // `tokio::spawn` above, but without having to return a dummy future
+//!         // here.
+//!         lines
+//!             // Convert the stream of values into a future which will resolve
+//!             // once the entire stream has been consumed. In this example we
+//!             // don't need to do anything with the data within the `for_each`
+//!             // call, but you can extend this to do something else (keep in mind
+//!             // that the stream will not produce items until the future returned
+//!             // from the closure resolves).
+//!             .for_each(|_| Ok(()))
+//!             // Similarly we "handle" any errors that arise, as required by tokio.
+//!             .map_err(|e| panic!("error while processing lines: {}", e))
+//!     }));
+//! }
+//! ```
+//!
+//! # Caveats
+//!
+//! While similar to the standard library, this crate's `Child` type differs
+//! importantly in the behavior of `drop`. In the standard library, a child
+//! process will continue running after the instance of `std::process::Child`
+//! is dropped. In this crate, however, because `tokio_process::Child` is a
+//! future of the child's `ExitStatus`, a child process is terminated if
+//! `tokio_process::Child` is dropped. The behavior of the standard library can
+//! be regained with the `Child::forget` method.
+
+#![warn(missing_debug_implementations)]
+#![deny(missing_docs)]
+#![doc(html_root_url = "https://docs.rs/tokio-process/0.2")]
+
+extern crate futures;
+extern crate tokio_io;
+extern crate tokio_reactor;
+
+#[cfg(unix)]
+#[macro_use]
+extern crate lazy_static;
+#[cfg(unix)]
+#[macro_use]
+extern crate log;
+
+use std::io::{self, Read, Write};
+use std::process::{Command, ExitStatus, Output, Stdio};
+
+use crate::kill::Kill;
+use futures::future::{ok, Either};
+use futures::{Async, Future, IntoFuture, Poll};
+use std::fmt;
+use tokio_io::io::read_to_end;
+use tokio_io::{AsyncRead, AsyncWrite, IoFuture};
+use tokio_reactor::Handle;
+
+#[path = "unix/mod.rs"]
+#[cfg(unix)]
+mod imp;
+
+#[path = "windows.rs"]
+#[cfg(windows)]
+mod imp;
+
+mod kill;
+
+/// Extensions provided by this crate to the `Command` type in the standard
+/// library.
+///
+/// This crate primarily enhances the standard library's `Command` type with
+/// asynchronous capabilities. The currently three blocking functions in the
+/// standard library, `spawn`, `status`, and `output`, all have asynchronous
+/// versions through this trait.
+///
+/// Note that the `Child` type spawned is specific to this crate, and that the
+/// I/O handles created from this crate are all asynchronous as well (differing
+/// from their `std` counterparts).
+pub trait CommandExt {
+    /// Executes the command as a child process, returning a handle to it.
+    ///
+    /// By default, stdin, stdout and stderr are inherited from the parent.
+    ///
+    /// This method will spawn the child process synchronously and return a
+    /// handle to a future-aware child process. The `Child` returned implements
+    /// `Future` itself to acquire the `ExitStatus` of the child, and otherwise
+    /// the `Child` has methods to acquire handles to the stdin, stdout, and
+    /// stderr streams.
+    ///
+    /// All I/O this child does will be associated with the current default
+    /// event loop.
+    fn spawn_async(&mut self) -> io::Result<Child> {
+        self.spawn_async_with_handle(&Handle::default())
+    }
+
+    /// Executes the command as a child process, returning a handle to it.
+    ///
+    /// By default, stdin, stdout and stderr are inherited from the parent.
+    ///
+    /// This method will spawn the child process synchronously and return a
+    /// handle to a future-aware child process. The `Child` returned implements
+    /// `Future` itself to acquire the `ExitStatus` of the child, and otherwise
+    /// the `Child` has methods to acquire handles to the stdin, stdout, and
+    /// stderr streams.
+    ///
+    /// The `handle` specified to this method must be a handle to a valid event
+    /// loop, and all I/O this child does will be associated with the specified
+    /// event loop.
+    fn spawn_async_with_handle(&mut self, handle: &Handle) -> io::Result<Child>;
+
+    /// Executes a command as a child process, waiting for it to finish and
+    /// collecting its exit status.
+    ///
+    /// By default, stdin, stdout and stderr are inherited from the parent.
+    ///
+    /// The `StatusAsync` future returned will resolve to the `ExitStatus`
+    /// type in the standard library representing how the process exited. If
+    /// any input/output handles are set to a pipe then they will be immediately
+    ///  closed after the child is spawned.
+    ///
+    /// All I/O this child does will be associated with the current default
+    /// event loop.
+    ///
+    /// If the `StatusAsync` future is dropped before the future resolves, then
+    /// the child will be killed, if it was spawned.
+    ///
+    /// # Errors
+    ///
+    /// This function will return an error immediately if the child process
+    /// cannot be spawned. Otherwise errors obtained while waiting for the child
+    /// are returned through the `StatusAsync` future.
+    fn status_async(&mut self) -> io::Result<StatusAsync> {
+        self.status_async_with_handle(&Handle::default())
+    }
+
+    /// Executes a command as a child process, waiting for it to finish and
+    /// collecting its exit status.
+    ///
+    /// By default, stdin, stdout and stderr are inherited from the parent.
+    ///
+    /// The `StatusAsync` future returned will resolve to the `ExitStatus`
+    /// type in the standard library representing how the process exited. If
+    /// any input/output handles are set to a pipe then they will be immediately
+    ///  closed after the child is spawned.
+    ///
+    /// The `handle` specified must be a handle to a valid event loop, and all
+    /// I/O this child does will be associated with the specified event loop.
+    ///
+    /// If the `StatusAsync` future is dropped before the future resolves, then
+    /// the child will be killed, if it was spawned.
+    ///
+    /// # Errors
+    ///
+    /// This function will return an error immediately if the child process
+    /// cannot be spawned. Otherwise errors obtained while waiting for the child
+    /// are returned through the `StatusAsync` future.
+    fn status_async_with_handle(&mut self, handle: &Handle) -> io::Result<StatusAsync>;
+
+    /// Executes the command as a child process, waiting for it to finish and
+    /// collecting all of its output.
+    ///
+    /// > **Note**: this method, unlike the standard library, will
+    /// > unconditionally configure the stdout/stderr handles to be pipes, even
+    /// > if they have been previously configured. If this is not desired then
+    /// > the `spawn_async` method should be used in combination with the
+    /// > `wait_with_output` method on child.
+    ///
+    /// This method will return a future representing the collection of the
+    /// child process's stdout/stderr. The `OutputAsync` future will resolve to
+    /// the `Output` type in the standard library, containing `stdout` and
+    /// `stderr` as `Vec<u8>` along with an `ExitStatus` representing how the
+    /// process exited.
+    ///
+    /// All I/O this child does will be associated with the current default
+    /// event loop.
+    ///
+    /// If the `OutputAsync` future is dropped before the future resolves, then
+    /// the child will be killed, if it was spawned.
+    fn output_async(&mut self) -> OutputAsync {
+        self.output_async_with_handle(&Handle::default())
+    }
+
+    /// Executes the command as a child process, waiting for it to finish and
+    /// collecting all of its output.
+    ///
+    /// > **Note**: this method, unlike the standard library, will
+    /// > unconditionally configure the stdout/stderr handles to be pipes, even
+    /// > if they have been previously configured. If this is not desired then
+    /// > the `spawn_async` method should be used in combination with the
+    /// > `wait_with_output` method on child.
+    ///
+    /// This method will return a future representing the collection of the
+    /// child process's stdout/stderr. The `OutputAsync` future will resolve to
+    /// the `Output` type in the standard library, containing `stdout` and
+    /// `stderr` as `Vec<u8>` along with an `ExitStatus` representing how the
+    /// process exited.
+    ///
+    /// The `handle` specified must be a handle to a valid event loop, and all
+    /// I/O this child does will be associated with the specified event loop.
+    ///
+    /// If the `OutputAsync` future is dropped before the future resolves, then
+    /// the child will be killed, if it was spawned.
+    fn output_async_with_handle(&mut self, handle: &Handle) -> OutputAsync;
+}
+
+struct SpawnedChild {
+    child: imp::Child,
+    stdin: Option<imp::ChildStdin>,
+    stdout: Option<imp::ChildStdout>,
+    stderr: Option<imp::ChildStderr>,
+}
+
+impl CommandExt for Command {
+    fn spawn_async_with_handle(&mut self, handle: &Handle) -> io::Result<Child> {
+        imp::spawn_child(self, handle).map(|spawned_child| Child {
+            child: ChildDropGuard::new(spawned_child.child),
+            stdin: spawned_child.stdin.map(|inner| ChildStdin { inner }),
+            stdout: spawned_child.stdout.map(|inner| ChildStdout { inner }),
+            stderr: spawned_child.stderr.map(|inner| ChildStderr { inner }),
+        })
+    }
+
+    fn status_async_with_handle(&mut self, handle: &Handle) -> io::Result<StatusAsync> {
+        self.spawn_async_with_handle(handle).map(|mut child| {
+            // Ensure we close any stdio handles so we can't deadlock
+            // waiting on the child which may be waiting to read/write
+            // to a pipe we're holding.
+            child.stdin.take();
+            child.stdout.take();
+            child.stderr.take();
+
+            StatusAsync { inner: child }
+        })
+    }
+
+    fn output_async_with_handle(&mut self, handle: &Handle) -> OutputAsync {
+        self.stdout(Stdio::piped());
+        self.stderr(Stdio::piped());
+
+        let inner = self
+            .spawn_async_with_handle(handle)
+            .into_future()
+            .and_then(Child::wait_with_output);
+
+        OutputAsync {
+            inner: Box::new(inner),
+        }
+    }
+}
+
+/// A drop guard which ensures the child process is killed on drop to maintain
+/// the contract of dropping a Future leads to "cancellation".
+#[derive(Debug)]
+struct ChildDropGuard<T: Kill> {
+    inner: T,
+    kill_on_drop: bool,
+}
+
+impl<T: Kill> ChildDropGuard<T> {
+    fn new(inner: T) -> Self {
+        Self {
+            inner,
+            kill_on_drop: true,
+        }
+    }
+
+    fn forget(&mut self) {
+        self.kill_on_drop = false;
+    }
+}
+
+impl<T: Kill> Kill for ChildDropGuard<T> {
+    fn kill(&mut self) -> io::Result<()> {
+        let ret = self.inner.kill();
+
+        if ret.is_ok() {
+            self.kill_on_drop = false;
+        }
+
+        ret
+    }
+}
+
+impl<T: Kill> Drop for ChildDropGuard<T> {
+    fn drop(&mut self) {
+        if self.kill_on_drop {
+            drop(self.kill());
+        }
+    }
+}
+
+impl<T: Future + Kill> Future for ChildDropGuard<T> {
+    type Item = T::Item;
+    type Error = T::Error;
+
+    fn poll(&mut self) -> Poll<Self::Item, Self::Error> {
+        let ret = self.inner.poll();
+
+        if let Ok(Async::Ready(_)) = ret {
+            // Avoid the overhead of trying to kill a reaped process
+            self.kill_on_drop = false;
+        }
+
+        ret
+    }
+}
+
+/// Representation of a child process spawned onto an event loop.
+///
+/// This type is also a future which will yield the `ExitStatus` of the
+/// underlying child process. A `Child` here also provides access to information
+/// like the OS-assigned identifier and the stdio streams.
+///
+/// > **Note**: The behavior of `drop` on a child in this crate is *different
+/// > than the behavior of the standard library*. If a `tokio_process::Child` is
+/// > dropped before the process finishes then the process will be terminated.
+/// > In the standard library, however, the process continues executing. This is
+/// > done because futures in general take `drop` as a sign of cancellation, and
+/// > this `Child` is itself a future. If you'd like to run a process in the
+/// > background, though, you may use the `forget` method.
+#[must_use = "futures do nothing unless polled"]
+#[derive(Debug)]
+pub struct Child {
+    child: ChildDropGuard<imp::Child>,
+    stdin: Option<ChildStdin>,
+    stdout: Option<ChildStdout>,
+    stderr: Option<ChildStderr>,
+}
+
+impl Child {
+    /// Returns the OS-assigned process identifier associated with this child.
+    pub fn id(&self) -> u32 {
+        self.child.inner.id()
+    }
+
+    /// Forces the child to exit.
+    ///
+    /// This is equivalent to sending a SIGKILL on unix platforms.
+    pub fn kill(&mut self) -> io::Result<()> {
+        self.child.kill()
+    }
+
+    /// Returns a handle for writing to the child's stdin, if it has been
+    /// captured
+    pub fn stdin(&mut self) -> &mut Option<ChildStdin> {
+        &mut self.stdin
+    }
+
+    /// Returns a handle for writing to the child's stdout, if it has been
+    /// captured
+    pub fn stdout(&mut self) -> &mut Option<ChildStdout> {
+        &mut self.stdout
+    }
+
+    /// Returns a handle for writing to the child's stderr, if it has been
+    /// captured
+    pub fn stderr(&mut self) -> &mut Option<ChildStderr> {
+        &mut self.stderr
+    }
+
+    /// Returns a future that will resolve to an `Output`, containing the exit
+    /// status, stdout, and stderr of the child process.
+    ///
+    /// The returned future will simultaneously waits for the child to exit and
+    /// collect all remaining output on the stdout/stderr handles, returning an
+    /// `Output` instance.
+    ///
+    /// The stdin handle to the child process, if any, will be closed before
+    /// waiting. This helps avoid deadlock: it ensures that the child does not
+    /// block waiting for input from the parent, while the parent waits for the
+    /// child to exit.
+    ///
+    /// By default, stdin, stdout and stderr are inherited from the parent. In
+    /// order to capture the output into this `Output` it is necessary to create
+    /// new pipes between parent and child. Use `stdout(Stdio::piped())` or
+    /// `stderr(Stdio::piped())`, respectively, when creating a `Command`.
+    pub fn wait_with_output(mut self) -> WaitWithOutput {
+        drop(self.stdin().take());
+        let stdout = match self.stdout().take() {
+            Some(io) => Either::A(read_to_end(io, Vec::new()).map(|p| p.1)),
+            None => Either::B(ok(Vec::new())),
+        };
+        let stderr = match self.stderr().take() {
+            Some(io) => Either::A(read_to_end(io, Vec::new()).map(|p| p.1)),
+            None => Either::B(ok(Vec::new())),
+        };
+
+        WaitWithOutput {
+            inner: Box::new(
+                self.join3(stdout, stderr)
+                    .map(|(status, stdout, stderr)| Output {
+                        status,
+                        stdout,
+                        stderr,
+                    }),
+            ),
+        }
+    }
+
+    /// Drop this `Child` without killing the underlying process.
+    ///
+    /// Normally a `Child` is killed if it's still alive when dropped, but this
+    /// method will ensure that the child may continue running once the `Child`
+    /// instance is dropped.
+    ///
+    /// > **Note**: this method may leak OS resources depending on your platform.
+    /// > To ensure resources are eventually cleaned up, consider sending the
+    /// > `Child` instance into an event loop as an alternative to this method.
+    ///
+    /// ```no_run
+    /// # extern crate futures;
+    /// # extern crate tokio;
+    /// # extern crate tokio_process;
+    /// #
+    /// # use std::process::Command;
+    /// #
+    /// # use futures::Future;
+    /// # use tokio_process::CommandExt;
+    /// #
+    /// # fn main() {
+    /// let child = Command::new("echo").arg("hello").arg("world")
+    ///                     .spawn_async()
+    ///                     .expect("failed to spawn");
+    ///
+    /// let do_cleanup = child.map(|_| ()) // Ignore result
+    ///                       .map_err(|_| ()); // Ignore errors
+    ///
+    /// tokio::spawn(do_cleanup);
+    /// # }
+    /// ```
+    pub fn forget(mut self) {
+        self.child.forget();
+    }
+}
+
+impl Future for Child {
+    type Item = ExitStatus;
+    type Error = io::Error;
+
+    fn poll(&mut self) -> Poll<ExitStatus, io::Error> {
+        self.child.poll()
+    }
+}
+
+/// Future returned from the `Child::wait_with_output` method.
+///
+/// This future will resolve to the standard library's `Output` type which
+/// contains the exit status, stdout, and stderr of a child process.
+#[must_use = "futures do nothing unless polled"]
+pub struct WaitWithOutput {
+    inner: IoFuture<Output>,
+}
+
+impl fmt::Debug for WaitWithOutput {
+    fn fmt(&self, fmt: &mut fmt::Formatter) -> fmt::Result {
+        fmt.debug_struct("WaitWithOutput")
+            .field("inner", &"..")
+            .finish()
+    }
+}
+
+impl Future for WaitWithOutput {
+    type Item = Output;
+    type Error = io::Error;
+
+    fn poll(&mut self) -> Poll<Output, io::Error> {
+        self.inner.poll()
+    }
+}
+
+#[doc(hidden)]
+#[deprecated(note = "renamed to `StatusAsync`", since = "0.2.1")]
+pub type StatusAsync2 = StatusAsync;
+
+/// Future returned by the `CommandExt::status_async` method.
+///
+/// This future is used to conveniently spawn a child and simply wait for its
+/// exit status. This future will resolves to the `ExitStatus` type in the
+/// standard library.
+#[must_use = "futures do nothing unless polled"]
+#[derive(Debug)]
+pub struct StatusAsync {
+    inner: Child,
+}
+
+impl Future for StatusAsync {
+    type Item = ExitStatus;
+    type Error = io::Error;
+
+    fn poll(&mut self) -> Poll<ExitStatus, io::Error> {
+        self.inner.poll()
+    }
+}
+
+/// Future returned by the `CommandExt::output_async` method.
+///
+/// This future is mostly equivalent to spawning a process and then calling
+/// `wait_with_output` on it internally. This can be useful to simply spawn a
+/// process, collecting all of its output and its exit status.
+#[must_use = "futures do nothing unless polled"]
+pub struct OutputAsync {
+    inner: IoFuture<Output>,
+}
+
+impl fmt::Debug for OutputAsync {
+    fn fmt(&self, fmt: &mut fmt::Formatter) -> fmt::Result {
+        fmt.debug_struct("OutputAsync")
+            .field("inner", &"..")
+            .finish()
+    }
+}
+
+impl Future for OutputAsync {
+    type Item = Output;
+    type Error = io::Error;
+
+    fn poll(&mut self) -> Poll<Output, io::Error> {
+        self.inner.poll()
+    }
+}
+
+/// The standard input stream for spawned children.
+///
+/// This type implements the `Write` trait to pass data to the stdin handle of
+/// a child process. Note that this type is also "futures aware" meaning that it
+/// is both (a) nonblocking and (b) will panic if used off of a future's task.
+#[derive(Debug)]
+pub struct ChildStdin {
+    inner: imp::ChildStdin,
+}
+
+/// The standard output stream for spawned children.
+///
+/// This type implements the `Read` trait to read data from the stdout handle
+/// of a child process. Note that this type is also "futures aware" meaning
+/// that it is both (a) nonblocking and (b) will panic if used off of a
+/// future's task.
+#[derive(Debug)]
+pub struct ChildStdout {
+    inner: imp::ChildStdout,
+}
+
+/// The standard error stream for spawned children.
+///
+/// This type implements the `Read` trait to read data from the stderr handle
+/// of a child process. Note that this type is also "futures aware" meaning
+/// that it is both (a) nonblocking and (b) will panic if used off of a
+/// future's task.
+#[derive(Debug)]
+pub struct ChildStderr {
+    inner: imp::ChildStderr,
+}
+
+impl Write for ChildStdin {
+    fn write(&mut self, bytes: &[u8]) -> io::Result<usize> {
+        self.inner.write(bytes)
+    }
+
+    fn flush(&mut self) -> io::Result<()> {
+        self.inner.flush()
+    }
+}
+
+impl AsyncWrite for ChildStdin {
+    fn shutdown(&mut self) -> Poll<(), io::Error> {
+        self.inner.shutdown()
+    }
+}
+
+impl Read for ChildStdout {
+    fn read(&mut self, bytes: &mut [u8]) -> io::Result<usize> {
+        self.inner.read(bytes)
+    }
+}
+
+impl AsyncRead for ChildStdout {}
+
+impl Read for ChildStderr {
+    fn read(&mut self, bytes: &mut [u8]) -> io::Result<usize> {
+        self.inner.read(bytes)
+    }
+}
+
+impl AsyncRead for ChildStderr {}
+
+#[cfg(unix)]
+mod sys {
+    use super::{ChildStderr, ChildStdin, ChildStdout};
+    use std::os::unix::io::{AsRawFd, RawFd};
+
+    impl AsRawFd for ChildStdin {
+        fn as_raw_fd(&self) -> RawFd {
+            self.inner.get_ref().as_raw_fd()
+        }
+    }
+
+    impl AsRawFd for ChildStdout {
+        fn as_raw_fd(&self) -> RawFd {
+            self.inner.get_ref().as_raw_fd()
+        }
+    }
+
+    impl AsRawFd for ChildStderr {
+        fn as_raw_fd(&self) -> RawFd {
+            self.inner.get_ref().as_raw_fd()
+        }
+    }
+}
+
+#[cfg(windows)]
+mod sys {
+    use super::{ChildStderr, ChildStdin, ChildStdout};
+    use std::os::windows::io::{AsRawHandle, RawHandle};
+
+    impl AsRawHandle for ChildStdin {
+        fn as_raw_handle(&self) -> RawHandle {
+            self.inner.get_ref().as_raw_handle()
+        }
+    }
+
+    impl AsRawHandle for ChildStdout {
+        fn as_raw_handle(&self) -> RawHandle {
+            self.inner.get_ref().as_raw_handle()
+        }
+    }
+
+    impl AsRawHandle for ChildStderr {
+        fn as_raw_handle(&self) -> RawHandle {
+            self.inner.get_ref().as_raw_handle()
+        }
+    }
+}
+
+#[cfg(test)]
+mod test {
+    use super::ChildDropGuard;
+    use crate::kill::Kill;
+    use futures::{Async, Future, Poll};
+    use std::io;
+
+    struct Mock {
+        num_kills: usize,
+        num_polls: usize,
+        poll_result: Poll<(), ()>,
+    }
+
+    impl Mock {
+        fn new() -> Self {
+            Self::with_result(Ok(Async::NotReady))
+        }
+
+        fn with_result(result: Poll<(), ()>) -> Self {
+            Self {
+                num_kills: 0,
+                num_polls: 0,
+                poll_result: result,
+            }
+        }
+    }
+
+    impl Kill for Mock {
+        fn kill(&mut self) -> io::Result<()> {
+            self.num_kills += 1;
+            Ok(())
+        }
+    }
+
+    impl Future for Mock {
+        type Item = ();
+        type Error = ();
+
+        fn poll(&mut self) -> Poll<Self::Item, Self::Error> {
+            self.num_polls += 1;
+            self.poll_result
+        }
+    }
+
+    #[test]
+    fn kills_on_drop() {
+        let mut mock = Mock::new();
+
+        {
+            let guard = ChildDropGuard::new(&mut mock);
+            drop(guard);
+        }
+
+        assert_eq!(1, mock.num_kills);
+        assert_eq!(0, mock.num_polls);
+    }
+
+    #[test]
+    fn no_kill_if_already_killed() {
+        let mut mock = Mock::new();
+
+        {
+            let mut guard = ChildDropGuard::new(&mut mock);
+            let _ = guard.kill();
+            drop(guard);
+        }
+
+        assert_eq!(1, mock.num_kills);
+        assert_eq!(0, mock.num_polls);
+    }
+
+    #[test]
+    fn no_kill_if_reaped() {
+        let mut mock_pending = Mock::with_result(Ok(Async::NotReady));
+        let mut mock_reaped = Mock::with_result(Ok(Async::Ready(())));
+        let mut mock_err = Mock::with_result(Err(()));
+
+        {
+            let mut guard = ChildDropGuard::new(&mut mock_pending);
+            let _ = guard.poll();
+
+            let mut guard = ChildDropGuard::new(&mut mock_reaped);
+            let _ = guard.poll();
+
+            let mut guard = ChildDropGuard::new(&mut mock_err);
+            let _ = guard.poll();
+        }
+
+        assert_eq!(1, mock_pending.num_kills);
+        assert_eq!(1, mock_pending.num_polls);
+
+        assert_eq!(0, mock_reaped.num_kills);
+        assert_eq!(1, mock_reaped.num_polls);
+
+        assert_eq!(1, mock_err.num_kills);
+        assert_eq!(1, mock_err.num_polls);
+    }
+
+    #[test]
+    fn no_kill_on_forget() {
+        let mut mock = Mock::new();
+
+        {
+            let mut guard = ChildDropGuard::new(&mut mock);
+            guard.forget();
+            drop(guard);
+        }
+
+        assert_eq!(0, mock.num_kills);
+        assert_eq!(0, mock.num_polls);
+    }
+}

tokio-process/src/unix/mod.rs

@@ -0,0 +1,221 @@
+//! Unix handling of child processes
+//!
+//! Right now the only "fancy" thing about this is how we implement the
+//! `Future` implementation on `Child` to get the exit status. Unix offers
+//! no way to register a child with epoll, and the only real way to get a
+//! notification when a process exits is the SIGCHLD signal.
+//!
+//! Signal handling in general is *super* hairy and complicated, and it's even
+//! more complicated here with the fact that signals are coalesced, so we may
+//! not get a SIGCHLD-per-child.
+//!
+//! Our best approximation here is to check *all spawned processes* for all
+//! SIGCHLD signals received. To do that we create a `Signal`, implemented in
+//! the `tokio-signal` crate, which is a stream over signals being received.
+//!
+//! Later when we poll the process's exit status we simply check to see if a
+//! SIGCHLD has happened since we last checked, and while that returns "yes" we
+//! keep trying.
+//!
+//! Note that this means that this isn't really scalable, but then again
+//! processes in general aren't scalable (e.g. millions) so it shouldn't be that
+//! bad in theory...
+
+extern crate libc;
+extern crate mio;
+extern crate tokio_signal;
+
+mod orphan;
+mod reap;
+
+use self::mio::event::Evented;
+use self::mio::unix::{EventedFd, UnixReady};
+use self::mio::{Poll as MioPoll, PollOpt, Ready, Token};
+use self::orphan::{AtomicOrphanQueue, OrphanQueue, Wait};
+use self::reap::Reaper;
+use self::tokio_signal::unix::Signal;
+use super::SpawnedChild;
+use crate::kill::Kill;
+use futures::future::FlattenStream;
+use futures::{Future, Poll};
+use std::fmt;
+use std::io;
+use std::os::unix::io::{AsRawFd, RawFd};
+use std::process::{self, ExitStatus};
+use tokio_io::IoFuture;
+use tokio_reactor::{Handle, PollEvented};
+
+impl Wait for process::Child {
+    fn id(&self) -> u32 {
+        self.id()
+    }
+
+    fn try_wait(&mut self) -> io::Result<Option<ExitStatus>> {
+        self.try_wait()
+    }
+}
+
+impl Kill for process::Child {
+    fn kill(&mut self) -> io::Result<()> {
+        self.kill()
+    }
+}
+
+lazy_static! {
+    static ref ORPHAN_QUEUE: AtomicOrphanQueue<process::Child> = AtomicOrphanQueue::new();
+}
+
+struct GlobalOrphanQueue;
+
+impl fmt::Debug for GlobalOrphanQueue {
+    fn fmt(&self, fmt: &mut fmt::Formatter) -> fmt::Result {
+        ORPHAN_QUEUE.fmt(fmt)
+    }
+}
+
+impl OrphanQueue<process::Child> for GlobalOrphanQueue {
+    fn push_orphan(&self, orphan: process::Child) {
+        ORPHAN_QUEUE.push_orphan(orphan)
+    }
+
+    fn reap_orphans(&self) {
+        ORPHAN_QUEUE.reap_orphans()
+    }
+}
+
+#[must_use = "futures do nothing unless polled"]
+pub struct Child {
+    inner: Reaper<process::Child, GlobalOrphanQueue, FlattenStream<IoFuture<Signal>>>,
+}
+
+impl fmt::Debug for Child {
+    fn fmt(&self, fmt: &mut fmt::Formatter) -> fmt::Result {
+        fmt.debug_struct("Child")
+            .field("pid", &self.inner.id())
+            .finish()
+    }
+}
+
+pub(crate) fn spawn_child(cmd: &mut process::Command, handle: &Handle) -> io::Result<SpawnedChild> {
+    let mut child = cmd.spawn()?;
+    let stdin = stdio(child.stdin.take(), handle)?;
+    let stdout = stdio(child.stdout.take(), handle)?;
+    let stderr = stdio(child.stderr.take(), handle)?;
+
+    let signal = Signal::with_handle(libc::SIGCHLD, handle).flatten_stream();
+    Ok(SpawnedChild {
+        child: Child {
+            inner: Reaper::new(child, GlobalOrphanQueue, signal),
+        },
+        stdin,
+        stdout,
+        stderr,
+    })
+}
+
+impl Child {
+    pub fn id(&self) -> u32 {
+        self.inner.id()
+    }
+}
+
+impl Kill for Child {
+    fn kill(&mut self) -> io::Result<()> {
+        self.inner.kill()
+    }
+}
+
+impl Future for Child {
+    type Item = ExitStatus;
+    type Error = io::Error;
+
+    fn poll(&mut self) -> Poll<Self::Item, Self::Error> {
+        self.inner.poll()
+    }
+}
+
+#[derive(Debug)]
+pub struct Fd<T>(T);
+
+impl<T: io::Read> io::Read for Fd<T> {
+    fn read(&mut self, bytes: &mut [u8]) -> io::Result<usize> {
+        self.0.read(bytes)
+    }
+}
+
+impl<T: io::Write> io::Write for Fd<T> {
+    fn write(&mut self, bytes: &[u8]) -> io::Result<usize> {
+        self.0.write(bytes)
+    }
+
+    fn flush(&mut self) -> io::Result<()> {
+        self.0.flush()
+    }
+}
+
+impl<T> AsRawFd for Fd<T>
+where
+    T: AsRawFd,
+{
+    fn as_raw_fd(&self) -> RawFd {
+        self.0.as_raw_fd()
+    }
+}
+
+pub type ChildStdin = PollEvented<Fd<process::ChildStdin>>;
+pub type ChildStdout = PollEvented<Fd<process::ChildStdout>>;
+pub type ChildStderr = PollEvented<Fd<process::ChildStderr>>;
+
+impl<T> Evented for Fd<T>
+where
+    T: AsRawFd,
+{
+    fn register(
+        &self,
+        poll: &MioPoll,
+        token: Token,
+        interest: Ready,
+        opts: PollOpt,
+    ) -> io::Result<()> {
+        EventedFd(&self.as_raw_fd()).register(poll, token, interest | UnixReady::hup(), opts)
+    }
+
+    fn reregister(
+        &self,
+        poll: &MioPoll,
+        token: Token,
+        interest: Ready,
+        opts: PollOpt,
+    ) -> io::Result<()> {
+        EventedFd(&self.as_raw_fd()).reregister(poll, token, interest | UnixReady::hup(), opts)
+    }
+
+    fn deregister(&self, poll: &MioPoll) -> io::Result<()> {
+        EventedFd(&self.as_raw_fd()).deregister(poll)
+    }
+}
+
+fn stdio<T>(option: Option<T>, handle: &Handle) -> io::Result<Option<PollEvented<Fd<T>>>>
+where
+    T: AsRawFd,
+{
+    let io = match option {
+        Some(io) => io,
+        None => return Ok(None),
+    };
+
+    // Set the fd to nonblocking before we pass it to the event loop
+    unsafe {
+        let fd = io.as_raw_fd();
+        let r = libc::fcntl(fd, libc::F_GETFL);
+        if r == -1 {
+            return Err(io::Error::last_os_error());
+        }
+        let r = libc::fcntl(fd, libc::F_SETFL, r | libc::O_NONBLOCK);
+        if r == -1 {
+            return Err(io::Error::last_os_error());
+        }
+    }
+    let io = PollEvented::new_with_handle(Fd(io), handle)?;
+    Ok(Some(io))
+}

tokio-process/src/unix/orphan.rs

@@ -0,0 +1,193 @@
+extern crate crossbeam_queue;
+
+use self::crossbeam_queue::SegQueue;
+use std::io;
+use std::process::ExitStatus;
+
+/// An interface for waiting on a process to exit.
+pub(crate) trait Wait {
+    /// Get the identifier for this process or diagnostics.
+    fn id(&self) -> u32;
+    /// Try waiting for a process to exit in a non-blocking manner.
+    fn try_wait(&mut self) -> io::Result<Option<ExitStatus>>;
+}
+
+impl<'a, T: 'a + Wait> Wait for &'a mut T {
+    fn id(&self) -> u32 {
+        (**self).id()
+    }
+
+    fn try_wait(&mut self) -> io::Result<Option<ExitStatus>> {
+        (**self).try_wait()
+    }
+}
+
+/// An interface for queueing up an orphaned process so that it can be reaped.
+pub(crate) trait OrphanQueue<T> {
+    /// Add an orphan to the queue.
+    fn push_orphan(&self, orphan: T);
+    /// Attempt to reap every process in the queue, ignoring any errors and
+    /// enqueueing any orphans which have not yet exited.
+    fn reap_orphans(&self);
+}
+
+impl<'a, T, O: 'a + OrphanQueue<T>> OrphanQueue<T> for &'a O {
+    fn push_orphan(&self, orphan: T) {
+        (**self).push_orphan(orphan);
+    }
+
+    fn reap_orphans(&self) {
+        (**self).reap_orphans()
+    }
+}
+
+/// An atomic implementation of `OrphanQueue`.
+#[derive(Debug)]
+pub(crate) struct AtomicOrphanQueue<T> {
+    queue: SegQueue<T>,
+}
+
+impl<T> AtomicOrphanQueue<T> {
+    pub(crate) fn new() -> Self {
+        Self {
+            queue: SegQueue::new(),
+        }
+    }
+}
+
+impl<T: Wait> OrphanQueue<T> for AtomicOrphanQueue<T> {
+    fn push_orphan(&self, orphan: T) {
+        self.queue.push(orphan)
+    }
+
+    fn reap_orphans(&self) {
+        let len = self.queue.len();
+
+        if len == 0 {
+            return;
+        }
+
+        let mut orphans = Vec::with_capacity(len);
+        while let Ok(mut orphan) = self.queue.pop() {
+            match orphan.try_wait() {
+                Ok(Some(_)) => {}
+                Err(e) => error!(
+                    "leaking orphaned process {} due to try_wait() error: {}",
+                    orphan.id(),
+                    e,
+                ),
+
+                // Still not done yet, we need to put it back in the queue
+                // when were done draining it, so that we don't get stuck
+                // in an infinite loop here
+                Ok(None) => orphans.push(orphan),
+            }
+        }
+
+        for orphan in orphans {
+            self.queue.push(orphan);
+        }
+    }
+}
+
+#[cfg(test)]
+mod test {
+    use super::Wait;
+    use super::{AtomicOrphanQueue, OrphanQueue};
+    use std::cell::Cell;
+    use std::io;
+    use std::os::unix::process::ExitStatusExt;
+    use std::process::ExitStatus;
+    use std::rc::Rc;
+
+    struct MockWait {
+        total_waits: Rc<Cell<usize>>,
+        num_wait_until_status: usize,
+        return_err: bool,
+    }
+
+    impl MockWait {
+        fn new(num_wait_until_status: usize) -> Self {
+            Self {
+                total_waits: Rc::new(Cell::new(0)),
+                num_wait_until_status,
+                return_err: false,
+            }
+        }
+
+        fn with_err() -> Self {
+            Self {
+                total_waits: Rc::new(Cell::new(0)),
+                num_wait_until_status: 0,
+                return_err: true,
+            }
+        }
+    }
+
+    impl Wait for MockWait {
+        fn id(&self) -> u32 {
+            42
+        }
+
+        fn try_wait(&mut self) -> io::Result<Option<ExitStatus>> {
+            let waits = self.total_waits.get();
+
+            let ret = if self.num_wait_until_status == waits {
+                if self.return_err {
+                    Ok(Some(ExitStatus::from_raw(0)))
+                } else {
+                    Err(io::Error::new(io::ErrorKind::Other, "mock err"))
+                }
+            } else {
+                Ok(None)
+            };
+
+            self.total_waits.set(waits + 1);
+            ret
+        }
+    }
+
+    #[test]
+    fn drain_attempts_a_single_reap_of_all_queued_orphans() {
+        let first_orphan = MockWait::new(0);
+        let second_orphan = MockWait::new(1);
+        let third_orphan = MockWait::new(2);
+        let fourth_orphan = MockWait::with_err();
+
+        let first_waits = first_orphan.total_waits.clone();
+        let second_waits = second_orphan.total_waits.clone();
+        let third_waits = third_orphan.total_waits.clone();
+        let fourth_waits = fourth_orphan.total_waits.clone();
+
+        let orphanage = AtomicOrphanQueue::new();
+        orphanage.push_orphan(first_orphan);
+        orphanage.push_orphan(third_orphan);
+        orphanage.push_orphan(second_orphan);
+        orphanage.push_orphan(fourth_orphan);
+
+        assert_eq!(orphanage.queue.len(), 4);
+
+        orphanage.reap_orphans();
+        assert_eq!(orphanage.queue.len(), 2);
+        assert_eq!(first_waits.get(), 1);
+        assert_eq!(second_waits.get(), 1);
+        assert_eq!(third_waits.get(), 1);
+        assert_eq!(fourth_waits.get(), 1);
+
+        orphanage.reap_orphans();
+        assert_eq!(orphanage.queue.len(), 1);
+        assert_eq!(first_waits.get(), 1);
+        assert_eq!(second_waits.get(), 2);
+        assert_eq!(third_waits.get(), 2);
+        assert_eq!(fourth_waits.get(), 1);
+
+        orphanage.reap_orphans();
+        assert_eq!(orphanage.queue.len(), 0);
+        assert_eq!(first_waits.get(), 1);
+        assert_eq!(second_waits.get(), 2);
+        assert_eq!(third_waits.get(), 3);
+        assert_eq!(fourth_waits.get(), 1);
+
+        orphanage.reap_orphans(); // Safe to reap when empty
+    }
+}

tokio-process/src/unix/reap.rs

@@ -0,0 +1,318 @@
+use super::orphan::{OrphanQueue, Wait};
+use crate::kill::Kill;
+use futures::{Async, Future, Poll, Stream};
+use std::io;
+use std::ops::Deref;
+use std::process::ExitStatus;
+
+/// Orchestrates between registering interest for receiving signals when a
+/// child process has exited, and attempting to poll for process completion.
+#[derive(Debug)]
+pub(crate) struct Reaper<W, Q, S>
+where
+    W: Wait,
+    Q: OrphanQueue<W>,
+{
+    inner: Option<W>,
+    orphan_queue: Q,
+    signal: S,
+}
+
+impl<W, Q, S> Deref for Reaper<W, Q, S>
+where
+    W: Wait,
+    Q: OrphanQueue<W>,
+{
+    type Target = W;
+
+    fn deref(&self) -> &Self::Target {
+        self.inner()
+    }
+}
+
+impl<W, Q, S> Reaper<W, Q, S>
+where
+    W: Wait,
+    Q: OrphanQueue<W>,
+{
+    pub(crate) fn new(inner: W, orphan_queue: Q, signal: S) -> Self {
+        Self {
+            inner: Some(inner),
+            orphan_queue,
+            signal,
+        }
+    }
+
+    fn inner(&self) -> &W {
+        self.inner.as_ref().expect("inner has gone away")
+    }
+
+    fn inner_mut(&mut self) -> &mut W {
+        self.inner.as_mut().expect("inner has gone away")
+    }
+}
+
+impl<W, Q, S> Future for Reaper<W, Q, S>
+where
+    W: Wait,
+    Q: OrphanQueue<W>,
+    S: Stream<Error = io::Error>,
+{
+    type Item = ExitStatus;
+    type Error = io::Error;
+
+    fn poll(&mut self) -> Poll<Self::Item, Self::Error> {
+        loop {
+            // If the child hasn't exited yet, then it's our responsibility to
+            // ensure the current task gets notified when it might be able to
+            // make progress.
+            //
+            // As described in `spawn` above, we just indicate that we can
+            // next make progress once a SIGCHLD is received.
+            //
+            // However, we will register for a notification on the next signal
+            // BEFORE we poll the child. Otherwise it is possible that the child
+            // can exit and the signal can arrive after we last polled the child,
+            // but before we've registered for a notification on the next signal
+            // (this can cause a deadlock if there are no more spawned children
+            // which can generate a different signal for us). A side effect of
+            // pre-registering for signal notifications is that when the child
+            // exits, we will have already registered for an additional
+            // notification we don't need to consume. If another signal arrives,
+            // this future's task will be notified/woken up again. Since the
+            // futures model allows for spurious wake ups this extra wakeup
+            // should not cause significant issues with parent futures.
+            let registered_interest = self.signal.poll()?.is_not_ready();
+
+            self.orphan_queue.reap_orphans();
+            if let Some(status) = self.inner_mut().try_wait()? {
+                return Ok(Async::Ready(status));
+            }
+
+            // If our attempt to poll for the next signal was not ready, then
+            // we've arranged for our task to get notified and we can bail out.
+            if registered_interest {
+                return Ok(Async::NotReady);
+            } else {
+                // Otherwise, if the signal stream delivered a signal to us, we
+                // won't get notified at the next signal, so we'll loop and try
+                // again.
+                continue;
+            }
+        }
+    }
+}
+
+impl<W, Q, S> Kill for Reaper<W, Q, S>
+where
+    W: Kill + Wait,
+    Q: OrphanQueue<W>,
+{
+    fn kill(&mut self) -> io::Result<()> {
+        self.inner_mut().kill()
+    }
+}
+
+impl<W, Q, S> Drop for Reaper<W, Q, S>
+where
+    W: Wait,
+    Q: OrphanQueue<W>,
+{
+    fn drop(&mut self) {
+        if let Ok(Some(_)) = self.inner_mut().try_wait() {
+            return;
+        }
+
+        let orphan = self.inner.take().unwrap();
+        self.orphan_queue.push_orphan(orphan);
+    }
+}
+
+#[cfg(test)]
+mod test {
+    use super::*;
+    use futures::{Async, Poll, Stream};
+    use std::cell::{Cell, RefCell};
+    use std::os::unix::process::ExitStatusExt;
+    use std::process::ExitStatus;
+
+    #[derive(Debug)]
+    struct MockWait {
+        total_kills: usize,
+        total_waits: usize,
+        num_wait_until_status: usize,
+        status: ExitStatus,
+    }
+
+    impl MockWait {
+        fn new(status: ExitStatus, num_wait_until_status: usize) -> Self {
+            Self {
+                total_kills: 0,
+                total_waits: 0,
+                num_wait_until_status,
+                status,
+            }
+        }
+    }
+
+    impl Wait for MockWait {
+        fn id(&self) -> u32 {
+            0
+        }
+
+        fn try_wait(&mut self) -> io::Result<Option<ExitStatus>> {
+            let ret = if self.num_wait_until_status == self.total_waits {
+                Some(self.status)
+            } else {
+                None
+            };
+
+            self.total_waits += 1;
+            Ok(ret)
+        }
+    }
+
+    impl Kill for MockWait {
+        fn kill(&mut self) -> io::Result<()> {
+            self.total_kills += 1;
+            Ok(())
+        }
+    }
+
+    struct MockStream {
+        total_polls: usize,
+        values: Vec<Option<()>>,
+    }
+
+    impl MockStream {
+        fn new(values: Vec<Option<()>>) -> Self {
+            Self {
+                total_polls: 0,
+                values,
+            }
+        }
+    }
+
+    impl Stream for MockStream {
+        type Item = ();
+        type Error = io::Error;
+
+        fn poll(&mut self) -> Poll<Option<Self::Item>, Self::Error> {
+            self.total_polls += 1;
+            match self.values.remove(0) {
+                Some(()) => Ok(Async::Ready(Some(()))),
+                None => Ok(Async::NotReady),
+            }
+        }
+    }
+
+    struct MockQueue<W> {
+        all_enqueued: RefCell<Vec<W>>,
+        total_reaps: Cell<usize>,
+    }
+
+    impl<W> MockQueue<W> {
+        fn new() -> Self {
+            Self {
+                all_enqueued: RefCell::new(Vec::new()),
+                total_reaps: Cell::new(0),
+            }
+        }
+    }
+
+    impl<W: Wait> OrphanQueue<W> for MockQueue<W> {
+        fn push_orphan(&self, orphan: W) {
+            self.all_enqueued.borrow_mut().push(orphan);
+        }
+
+        fn reap_orphans(&self) {
+            self.total_reaps.set(self.total_reaps.get() + 1);
+        }
+    }
+
+    #[test]
+    fn reaper() {
+        let exit = ExitStatus::from_raw(0);
+        let mock = MockWait::new(exit, 3);
+        let mut grim = Reaper::new(
+            mock,
+            MockQueue::new(),
+            MockStream::new(vec![None, Some(()), None, None, None]),
+        );
+
+        // Not yet exited, interest registered
+        assert_eq!(Async::NotReady, grim.poll().expect("failed to wait"));
+        assert_eq!(1, grim.signal.total_polls);
+        assert_eq!(1, grim.total_waits);
+        assert_eq!(1, grim.orphan_queue.total_reaps.get());
+        assert!(grim.orphan_queue.all_enqueued.borrow().is_empty());
+
+        // Not yet exited, couldn't register interest the first time
+        // but managed to register interest the second time around
+        assert_eq!(Async::NotReady, grim.poll().expect("failed to wait"));
+        assert_eq!(3, grim.signal.total_polls);
+        assert_eq!(3, grim.total_waits);
+        assert_eq!(3, grim.orphan_queue.total_reaps.get());
+        assert!(grim.orphan_queue.all_enqueued.borrow().is_empty());
+
+        // Exited
+        assert_eq!(Async::Ready(exit), grim.poll().expect("failed to wait"));
+        assert_eq!(4, grim.signal.total_polls);
+        assert_eq!(4, grim.total_waits);
+        assert_eq!(4, grim.orphan_queue.total_reaps.get());
+        assert!(grim.orphan_queue.all_enqueued.borrow().is_empty());
+    }
+
+    #[test]
+    fn kill() {
+        let exit = ExitStatus::from_raw(0);
+        let mut grim = Reaper::new(
+            MockWait::new(exit, 0),
+            MockQueue::new(),
+            MockStream::new(vec![None]),
+        );
+
+        grim.kill().unwrap();
+        assert_eq!(1, grim.total_kills);
+        assert_eq!(0, grim.orphan_queue.total_reaps.get());
+        assert!(grim.orphan_queue.all_enqueued.borrow().is_empty());
+    }
+
+    #[test]
+    fn drop_reaps_if_possible() {
+        let exit = ExitStatus::from_raw(0);
+        let mut mock = MockWait::new(exit, 0);
+
+        {
+            let queue = MockQueue::new();
+
+            let grim = Reaper::new(&mut mock, &queue, MockStream::new(vec![]));
+
+            drop(grim);
+
+            assert_eq!(0, queue.total_reaps.get());
+            assert!(queue.all_enqueued.borrow().is_empty());
+        }
+
+        assert_eq!(1, mock.total_waits);
+        assert_eq!(0, mock.total_kills);
+    }
+
+    #[test]
+    fn drop_enqueues_orphan_if_wait_fails() {
+        let exit = ExitStatus::from_raw(0);
+        let mut mock = MockWait::new(exit, 2);
+
+        {
+            let queue = MockQueue::<&mut MockWait>::new();
+            let grim = Reaper::new(&mut mock, &queue, MockStream::new(vec![]));
+            drop(grim);
+
+            assert_eq!(0, queue.total_reaps.get());
+            assert_eq!(1, queue.all_enqueued.borrow().len());
+        }
+
+        assert_eq!(1, mock.total_waits);
+        assert_eq!(0, mock.total_kills);
+    }
+}

tokio-process/src/windows.rs

@@ -0,0 +1,192 @@
+//! Windows asynchronous process handling.
+//!
+//! Like with Unix we don't actually have a way of registering a process with an
+//! IOCP object. As a result we similarly need another mechanism for getting a
+//! signal when a process has exited. For now this is implemented with the
+//! `RegisterWaitForSingleObject` function in the kernel32.dll.
+//!
+//! This strategy is the same that libuv takes and essentially just queues up a
+//! wait for the process in a kernel32-specific thread pool. Once the object is
+//! notified (e.g. the process exits) then we have a callback that basically
+//! just completes a `Oneshot`.
+//!
+//! The `poll_exit` implementation will attempt to wait for the process in a
+//! nonblocking fashion, but failing that it'll fire off a
+//! `RegisterWaitForSingleObject` and then wait on the other end of the oneshot
+//! from then on out.
+
+extern crate mio_named_pipes;
+extern crate winapi;
+
+use std::fmt;
+use std::io;
+use std::os::windows::prelude::*;
+use std::os::windows::process::ExitStatusExt;
+use std::process::{self, ExitStatus};
+use std::ptr;
+
+use self::mio_named_pipes::NamedPipe;
+use self::winapi::shared::minwindef::*;
+use self::winapi::shared::winerror::*;
+use self::winapi::um::handleapi::*;
+use self::winapi::um::processthreadsapi::*;
+use self::winapi::um::synchapi::*;
+use self::winapi::um::threadpoollegacyapiset::*;
+use self::winapi::um::winbase::*;
+use self::winapi::um::winnt::*;
+use super::SpawnedChild;
+use crate::kill::Kill;
+use futures::future::Fuse;
+use futures::sync::oneshot;
+use futures::{Async, Future, Poll};
+use tokio_reactor::{Handle, PollEvented};
+
+#[must_use = "futures do nothing unless polled"]
+pub struct Child {
+    child: process::Child,
+    waiting: Option<Waiting>,
+}
+
+impl fmt::Debug for Child {
+    fn fmt(&self, fmt: &mut fmt::Formatter) -> fmt::Result {
+        fmt.debug_struct("Child")
+            .field("pid", &self.id())
+            .field("child", &self.child)
+            .field("waiting", &"..")
+            .finish()
+    }
+}
+
+struct Waiting {
+    rx: Fuse<oneshot::Receiver<()>>,
+    wait_object: HANDLE,
+    tx: *mut Option<oneshot::Sender<()>>,
+}
+
+unsafe impl Sync for Waiting {}
+unsafe impl Send for Waiting {}
+
+pub(crate) fn spawn_child(cmd: &mut process::Command, handle: &Handle) -> io::Result<SpawnedChild> {
+    let mut child = cmd.spawn()?;
+    let stdin = stdio(child.stdin.take(), handle)?;
+    let stdout = stdio(child.stdout.take(), handle)?;
+    let stderr = stdio(child.stderr.take(), handle)?;
+
+    Ok(SpawnedChild {
+        child: Child {
+            child,
+            waiting: None,
+        },
+        stdin,
+        stdout,
+        stderr,
+    })
+}
+
+impl Child {
+    pub fn id(&self) -> u32 {
+        self.child.id()
+    }
+}
+
+impl Kill for Child {
+    fn kill(&mut self) -> io::Result<()> {
+        self.child.kill()
+    }
+}
+
+impl Future for Child {
+    type Item = ExitStatus;
+    type Error = io::Error;
+
+    fn poll(&mut self) -> Poll<Self::Item, Self::Error> {
+        loop {
+            if let Some(ref mut w) = self.waiting {
+                match w.rx.poll().expect("should not be canceled") {
+                    Async::Ready(()) => {}
+                    Async::NotReady => return Ok(Async::NotReady),
+                }
+                let status = try_wait(&self.child)?.expect("not ready yet");
+                return Ok(status.into());
+            }
+
+            if let Some(e) = try_wait(&self.child)? {
+                return Ok(e.into());
+            }
+            let (tx, rx) = oneshot::channel();
+            let ptr = Box::into_raw(Box::new(Some(tx)));
+            let mut wait_object = ptr::null_mut();
+            let rc = unsafe {
+                RegisterWaitForSingleObject(
+                    &mut wait_object,
+                    self.child.as_raw_handle(),
+                    Some(callback),
+                    ptr as *mut _,
+                    INFINITE,
+                    WT_EXECUTEINWAITTHREAD | WT_EXECUTEONLYONCE,
+                )
+            };
+            if rc == 0 {
+                let err = io::Error::last_os_error();
+                drop(unsafe { Box::from_raw(ptr) });
+                return Err(err);
+            }
+            self.waiting = Some(Waiting {
+                rx: rx.fuse(),
+                wait_object,
+                tx: ptr,
+            });
+        }
+    }
+}
+
+impl Drop for Waiting {
+    fn drop(&mut self) {
+        unsafe {
+            let rc = UnregisterWaitEx(self.wait_object, INVALID_HANDLE_VALUE);
+            if rc == 0 {
+                panic!("failed to unregister: {}", io::Error::last_os_error());
+            }
+            drop(Box::from_raw(self.tx));
+        }
+    }
+}
+
+unsafe extern "system" fn callback(ptr: PVOID, _timer_fired: BOOLEAN) {
+    let complete = &mut *(ptr as *mut Option<oneshot::Sender<()>>);
+    let _ = complete.take().unwrap().send(());
+}
+
+pub fn try_wait(child: &process::Child) -> io::Result<Option<ExitStatus>> {
+    unsafe {
+        match WaitForSingleObject(child.as_raw_handle(), 0) {
+            WAIT_OBJECT_0 => {}
+            WAIT_TIMEOUT => return Ok(None),
+            _ => return Err(io::Error::last_os_error()),
+        }
+        let mut status = 0;
+        let rc = GetExitCodeProcess(child.as_raw_handle(), &mut status);
+        if rc == FALSE {
+            Err(io::Error::last_os_error())
+        } else {
+            Ok(Some(ExitStatus::from_raw(status)))
+        }
+    }
+}
+
+pub type ChildStdin = PollEvented<NamedPipe>;
+pub type ChildStdout = PollEvented<NamedPipe>;
+pub type ChildStderr = PollEvented<NamedPipe>;
+
+fn stdio<T>(option: Option<T>, handle: &Handle) -> io::Result<Option<PollEvented<NamedPipe>>>
+where
+    T: IntoRawHandle,
+{
+    let io = match option {
+        Some(io) => io,
+        None => return Ok(None),
+    };
+    let pipe = unsafe { NamedPipe::from_raw_handle(io.into_raw_handle()) };
+    let io = PollEvented::new_with_handle(pipe, handle)?;
+    Ok(Some(io))
+}

tokio-process/tests/issue_42.rs

@@ -0,0 +1,51 @@
+#![cfg(unix)]
+
+extern crate futures;
+extern crate tokio_process;
+
+use futures::{stream, Future, IntoFuture, Stream};
+use std::process::{Command, Stdio};
+use std::sync::atomic::{AtomicBool, Ordering};
+use std::sync::Arc;
+use std::thread;
+use std::time::Duration;
+use tokio_process::CommandExt;
+
+fn run_test() {
+    let finished = Arc::new(AtomicBool::new(false));
+    let finished_clone = finished.clone();
+
+    thread::spawn(move || {
+        let _ = stream::iter_ok(0..2)
+            .map(|i| {
+                Command::new("echo")
+                    .arg(format!("I am spawned process #{}", i))
+                    .stdin(Stdio::null())
+                    .stdout(Stdio::null())
+                    .stderr(Stdio::null())
+                    .spawn_async()
+                    .into_future()
+                    .flatten()
+            })
+            .buffered(2)
+            .collect()
+            .wait();
+
+        finished_clone.store(true, Ordering::SeqCst);
+    });
+
+    thread::sleep(Duration::from_millis(100));
+    assert!(
+        finished.load(Ordering::SeqCst),
+        "FINISHED flag not set, maybe we deadlocked?"
+    );
+}
+
+#[test]
+fn issue_42() {
+    let max = 10;
+    for i in 0..max {
+        println!("running {}/{}", i, max);
+        run_test()
+    }
+}

tokio-process/tests/smoke.rs

@@ -0,0 +1,22 @@
+extern crate tokio_process;
+
+use tokio_process::CommandExt;
+
+mod support;
+
+#[test]
+fn simple() {
+    let mut cmd = support::cmd("exit");
+    cmd.arg("2");
+
+    let mut child = cmd.spawn_async().unwrap();
+
+    let id = child.id();
+    assert!(id > 0);
+
+    let status = support::run_with_timeout(&mut child).expect("failed to run future");
+    assert_eq!(status.code(), Some(2));
+
+    assert_eq!(child.id(), id);
+    drop(child.kill());
+}

tokio-process/tests/stdio.rs

@@ -0,0 +1,113 @@
+extern crate futures;
+#[macro_use]
+extern crate log;
+extern crate tokio_io;
+extern crate tokio_process;
+
+use std::io;
+use std::process::{Command, ExitStatus, Stdio};
+
+use futures::future::Future;
+use futures::stream::{self, Stream};
+use tokio_io::io::{read_until, write_all};
+use tokio_process::{Child, CommandExt};
+
+mod support;
+
+fn cat() -> Command {
+    let mut cmd = support::cmd("cat");
+    cmd.stdin(Stdio::piped()).stdout(Stdio::piped());
+    cmd
+}
+
+fn feed_cat(mut cat: Child, n: usize) -> Box<Future<Item = ExitStatus, Error = io::Error>> {
+    let stdin = cat.stdin().take().unwrap();
+    let stdout = cat.stdout().take().unwrap();
+
+    debug!("starting to feed");
+    // Produce n lines on the child's stdout.
+    let numbers = stream::iter_ok(0..n);
+    let write = numbers
+        .fold(stdin, |stdin, i| {
+            debug!("sending line {} to child", i);
+            write_all(stdin, format!("line {}\n", i).into_bytes()).map(|p| p.0)
+        })
+        .map(|_| ());
+
+    // Try to read `n + 1` lines, ensuring the last one is empty
+    // (i.e. EOF is reached after `n` lines.
+    let reader = io::BufReader::new(stdout);
+    let expected_numbers = stream::iter_ok(0..=n);
+    let read = expected_numbers.fold((reader, 0), move |(reader, i), _| {
+        let done = i >= n;
+        debug!("starting read from child");
+        read_until(reader, b'\n', Vec::new()).and_then(move |(reader, vec)| {
+            debug!(
+                "read line {} from child ({} bytes, done: {})",
+                i,
+                vec.len(),
+                done
+            );
+            match (done, vec.len()) {
+                (false, 0) => Err(io::Error::new(io::ErrorKind::BrokenPipe, "broken pipe")),
+                (true, n) if n != 0 => Err(io::Error::new(io::ErrorKind::Other, "extraneous data")),
+                _ => {
+                    let s = std::str::from_utf8(&vec).unwrap();
+                    let expected = format!("line {}\n", i);
+                    if done || s == expected {
+                        Ok((reader, i + 1))
+                    } else {
+                        Err(io::Error::new(io::ErrorKind::Other, "unexpected data"))
+                    }
+                }
+            }
+        })
+    });
+
+    // Compose reading and writing concurrently.
+    Box::new(write.join(read).and_then(|_| cat))
+}
+
+/// Check for the following properties when feeding stdin and
+/// consuming stdout of a cat-like process:
+///
+/// - A number of lines that amounts to a number of bytes exceeding a
+///   typical OS buffer size can be fed to the child without
+///   deadlock. This tests that we also consume the stdout
+///   concurrently; otherwise this would deadlock.
+///
+/// - We read the same lines from the child that we fed it.
+///
+/// - The child does produce EOF on stdout after the last line.
+#[test]
+fn feed_a_lot() {
+    let child = cat().spawn_async().unwrap();
+    let status = support::run_with_timeout(feed_cat(child, 10000)).unwrap();
+    assert_eq!(status.code(), Some(0));
+}
+
+#[test]
+fn wait_with_output_captures() {
+    let mut child = cat().spawn_async().unwrap();
+    let stdin = child.stdin().take().unwrap();
+    let out = child.wait_with_output();
+
+    let future = write_all(stdin, b"1234").map(|p| p.1).join(out);
+    let ret = support::run_with_timeout(future).unwrap();
+    let (written, output) = ret;
+
+    assert!(output.status.success());
+    assert_eq!(output.stdout, written);
+    assert_eq!(output.stderr.len(), 0);
+}
+
+#[test]
+fn status_closes_any_pipes() {
+    // Cat will open a pipe between the parent and child.
+    // If `status_async` doesn't ensure the handles are closed,
+    // we would end up blocking forever (and time out).
+    let child = cat().status_async().expect("failed to spawn child");
+
+    support::run_with_timeout(child)
+        .expect("time out exceeded! did we get stuck waiting on the child?");
+}

tokio-process/tests/support/mod.rs

@@ -0,0 +1,43 @@
+extern crate futures;
+extern crate tokio;
+
+use self::futures::Future;
+use self::tokio::timer::Timeout;
+use std::env;
+use std::process::Command;
+use std::time::Duration;
+
+pub use self::tokio::runtime::current_thread::Runtime as CurrentThreadRuntime;
+
+pub fn cmd(s: &str) -> Command {
+    let mut me = env::current_exe().unwrap();
+    me.pop();
+    if me.ends_with("deps") {
+        me.pop();
+    }
+    me.push(s);
+    Command::new(me)
+}
+
+pub fn with_timeout<F: Future>(future: F) -> impl Future<Item = F::Item, Error = F::Error> {
+    Timeout::new(future, Duration::from_secs(3)).map_err(|e| {
+        if e.is_timer() {
+            panic!("failed to register timer");
+        } else if e.is_elapsed() {
+            panic!("timed out")
+        } else {
+            e.into_inner().expect("missing inner error")
+        }
+    })
+}
+
+pub fn run_with_timeout<F>(future: F) -> Result<F::Item, F::Error>
+where
+    F: Future,
+{
+    // NB: Timeout requires a timer registration which is provided by
+    // tokio's `current_thread::Runtime`, but isn't available by just using
+    // tokio's default CurrentThread executor which powers `current_thread::block_on_all`.
+    let mut rt = CurrentThreadRuntime::new().expect("failed to get runtime");
+    rt.block_on(with_timeout(future))
+}

tokio-reactor/CHANGELOG.md

@@ -1,3 +1,19 @@
+# 0.1.12 (February 4, 2020)
+
+* Add `tokio 0.2.x` deprecation notice.
+
+# 0.1.11 (November 27, 2019)
+
+### Added
+- `set_default`, which functions like `with_default` but returns a drop
+  guard (#1725)
+
+# 0.1.10 (September 25, 2019)
+
+### Changed
+- Upgrade to parking_lot 0.9.0 (#1298 backport)
+- The minimum supported rust version (MSRV) is now 1.31.0. (#1358)
+
 # 0.1.9 (March 1, 2019)
 
 ### Added

tokio-reactor/Cargo.toml

@@ -8,26 +8,26 @@ name = "tokio-reactor"
 #   - README.md
 # - Update CHANGELOG.md.
 # - Create "v0.1.x" git tag.
-version = "0.1.9"
+version = "0.1.12"
 authors = ["Carl Lerche <[email protected]>"]
 license = "MIT"
 readme = "README.md"
 repository = "https://github.com/tokio-rs/tokio"
 homepage = "https://tokio.rs"
-documentation = "https://docs.rs/tokio-reactor/0.1.9/tokio_reactor"
+documentation = "https://docs.rs/tokio-reactor/0.1.12/tokio_reactor"
 description = """
 Event loop that drives Tokio I/O resources.
 """
 categories = ["asynchronous", "network-programming"]
 
 [dependencies]
-crossbeam-utils = "0.6.0"
+crossbeam-utils = "0.7.0"
 futures = "0.1.19"
 lazy_static = "1.0.2"
 log = "0.4.1"
 mio = "0.6.14"
 num_cpus = "1.8.0"
-parking_lot = "0.7.0"
+parking_lot = "0.9.0"
 slab = "0.4.0"
 tokio-executor = "0.1.1"
 tokio-io = "0.1.6"
@@ -36,4 +36,4 @@ tokio-sync = "0.1.1"
 [dev-dependencies]
 num_cpus = "1.8.0"
 tokio = "0.1.7"
-tokio-io-pool = "0.1.4"
+tokio-io-pool = "=0.1.4"

tokio-reactor/README.md

@@ -2,7 +2,17 @@
 
 Event loop that drives Tokio I/O resources.
 
-[Documentation](https://docs.rs/tokio-reactor/0.1.9/tokio_reactor)
+> **Note:** This crate is **deprecated in tokio 0.2.x** and has been moved and
+> refactored into various places in the [`tokio::runtime`] and [`tokio::io`]
+> modules of the [`tokio`] crate. The Reactor has also been renamed the
+> "I/O Driver".
+
+[`tokio::runtime`]: https://docs.rs/tokio/latest/tokio/runtime/index.html
+[`tokio::io`]: https://docs.rs/tokio/latest/tokio/io/index.html
+[`tokio`]: https://docs.rs/tokio/latest/tokio/index.html
+[`io-driver` feature]: https://docs.rs/tokio/0.2.9/tokio/index.html#feature-flags
+
+[Documentation](https://docs.rs/tokio-reactor/0.1.11/tokio_reactor)
 
 ## Overview
 
@@ -25,10 +35,10 @@ are building a custom I/O resource.
 
 [`mio`]: http://github.com/carllerche/mio
 [`futures`]: http://github.com/rust-lang-nursery/futures-rs
-[`Reactor`]: https://docs.rs/tokio-reactor/0.1.9/tokio_reactor/struct.Reactor.html
-[`Handle`]: https://docs.rs/tokio-reactor/0.1.9/tokio_reactor/struct.Handle.html
-[`Registration`]: https://docs.rs/tokio-reactor/0.1.9/tokio_reactor/struct.Registration.html
-[`PollEvented`]: https://docs.rs/tokio-reactor/0.1.9/tokio_reactor/struct.PollEvented.html
+[`Reactor`]: https://docs.rs/tokio-reactor/0.1.11/tokio_reactor/struct.Reactor.html
+[`Handle`]: https://docs.rs/tokio-reactor/0.1.11/tokio_reactor/struct.Handle.html
+[`Registration`]: https://docs.rs/tokio-reactor/0.1.11/tokio_reactor/struct.Registration.html
+[`PollEvented`]: https://docs.rs/tokio-reactor/0.1.11/tokio_reactor/struct.PollEvented.html
 [`tokio`]: ../
 
 ## License

tokio-reactor/benches/basic.rs

@@ -1,5 +1,4 @@
 #![feature(test)]
-#![deny(warnings)]
 
 extern crate futures;
 extern crate mio;

tokio-reactor/src/lib.rs

@@ -1,8 +1,18 @@
-#![doc(html_root_url = "https://docs.rs/tokio-reactor/0.1.9")]
-#![deny(missing_docs, warnings, missing_debug_implementations)]
+#![doc(html_root_url = "https://docs.rs/tokio-reactor/0.1.12")]
+#![deny(missing_docs, missing_debug_implementations)]
 
 //! Event loop that drives Tokio I/O resources.
 //!
+//! > **Note:** This crate is **deprecated in tokio 0.2.x** and has been moved
+//! > and refactored into various places in the [`tokio::runtime`] and
+//! > [`tokio::io`] modules of the [`tokio`] crate. The Reactor has also been
+//! > renamed the "I/O Driver".
+//!
+//! [`tokio::runtime`]: https://docs.rs/tokio/latest/tokio/runtime/index.html
+//! [`tokio::io`]: https://docs.rs/tokio/latest/tokio/io/index.html
+//! [`tokio`]: https://docs.rs/tokio/latest/tokio/index.html
+//! [`io-driver` feature]: https://docs.rs/tokio/0.2.9/tokio/index.html#feature-flags
+//!
 //! The reactor is the engine that drives asynchronous I/O resources (like TCP and
 //! UDP sockets). It is backed by [`mio`] and acts as a bridge between [`mio`] and
 //! [`futures`].
@@ -133,6 +143,13 @@ pub struct SetFallbackError(());
 #[doc(hidden)]
 pub type SetDefaultError = SetFallbackError;
 
+/// Ensure that the default reactor is removed from the thread-local context
+/// when leaving the scope. This handles cases that involve panicking.
+#[derive(Debug)]
+pub struct DefaultGuard {
+    _p: (),
+}
+
 #[test]
 fn test_handle_size() {
     use std::mem;
@@ -197,45 +214,38 @@ pub fn with_default<F, R>(handle: &Handle, enter: &mut Enter, f: F) -> R
 where
     F: FnOnce(&mut Enter) -> R,
 {
-    // Ensure that the executor is removed from the thread-local context
-    // when leaving the scope. This handles cases that involve panicking.
-    struct Reset;
-
-    impl Drop for Reset {
-        fn drop(&mut self) {
-            CURRENT_REACTOR.with(|current| {
-                let mut current = current.borrow_mut();
-                *current = None;
-            });
-        }
-    }
-
     // This ensures the value for the current reactor gets reset even if there
     // is a panic.
-    let _r = Reset;
+    let _guard = set_default(handle);
+    f(enter)
+}
 
+/// Sets `handle` as the default reactor, returning a guard that unsets it when
+/// dropped.
+///
+/// # Panics
+///
+/// This function panics if there already is a default reactor set.
+pub fn set_default(handle: &Handle) -> DefaultGuard {
     CURRENT_REACTOR.with(|current| {
-        {
-            let mut current = current.borrow_mut();
+        let mut current = current.borrow_mut();
 
-            assert!(
-                current.is_none(),
-                "default Tokio reactor already set \
-                 for execution context"
-            );
+        assert!(
+            current.is_none(),
+            "default Tokio reactor already set \
+             for execution context"
+        );
 
-            let handle = match handle.as_priv() {
-                Some(handle) => handle,
-                None => {
-                    panic!("`handle` does not reference a reactor");
-                }
-            };
+        let handle = match handle.as_priv() {
+            Some(handle) => handle,
+            None => {
+                panic!("`handle` does not reference a reactor");
+            }
+        };
 
-            *current = Some(handle.clone());
-        }
-
-        f(enter)
-    })
+        *current = Some(handle.clone());
+    });
+    DefaultGuard { _p: () }
 }
 
 impl Reactor {
@@ -631,7 +641,7 @@ impl HandlePriv {
     }
 
     unsafe fn from_usize(val: usize) -> HandlePriv {
-        let inner = mem::transmute::<usize, Weak<Inner>>(val);;
+        let inner = mem::transmute::<usize, Weak<Inner>>(val);
         HandlePriv { inner }
     }
 
@@ -652,7 +662,7 @@ impl Inner {
     /// Register an I/O resource with the reactor.
     ///
     /// The registration token is returned.
-    fn add_source(&self, source: &Evented) -> io::Result<usize> {
+    fn add_source(&self, source: &dyn Evented) -> io::Result<usize> {
         // Get an ABA guard value
         let aba_guard = self.next_aba_guard.fetch_add(1 << TOKEN_SHIFT, Relaxed);
 
@@ -690,7 +700,7 @@ impl Inner {
     }
 
     /// Deregisters an I/O resource from the reactor.
-    fn deregister_source(&self, source: &Evented) -> io::Result<()> {
+    fn deregister_source(&self, source: &dyn Evented) -> io::Result<()> {
         self.io.deregister(source)
     }
 
@@ -743,6 +753,15 @@ impl Direction {
     }
 }
 
+impl Drop for DefaultGuard {
+    fn drop(&mut self) {
+        let _ = CURRENT_REACTOR.try_with(|current| {
+            let mut current = current.borrow_mut();
+            *current = None;
+        });
+    }
+}
+
 #[cfg(unix)]
 mod platform {
     use mio::unix::UnixReady;

tokio-signal/CHANGELOG.md

@@ -1,6 +1,9 @@
-# 0.2.9
+# 0.2.9 (February 4, 2020)
+
+* Add `tokio 0.2.x` deprecation notice.
 
 ### Fixed
+
 - `windows::Event` performs internal registrations lazily, so now it can be
 constructed outside of a running task
 - remove usage of deprecated `Handle::current` in default `windows::Event`

tokio-signal/Cargo.toml

@@ -8,12 +8,12 @@ name = "tokio-signal"
 #   - README.md
 # - Update CHANGELOG.md.
 # - Create "v0.2.x" git tag.
-version = "0.2.8"
+version = "0.2.9"
 authors = ["Tokio Contributors <[email protected]>"]
 license = "MIT"
 repository = "https://github.com/tokio-rs/tokio"
 homepage = "https://github.com/tokio-rs/tokio"
-documentation = "https://docs.rs/tokio-signal/0.2.8/tokio_signal"
+documentation = "https://docs.rs/tokio-signal/0.2.9/tokio_signal"
 description = """
 An implementation of an asynchronous Unix signal handling backed futures.
 """

tokio-signal/README.md

@@ -2,6 +2,12 @@
 
 Unix signal handling for Tokio.
 
+> **Note:** This crate is **deprecated in tokio 0.2.x** and has been moved into
+> [`tokio::signal`] behind the `signal` [feature flag].
+
+[`tokio::signal`]: https://docs.rs/tokio/latest/tokio/signal/index.html
+[feature flag]: https://docs.rs/tokio/latest/tokio/index.html#feature-flags
+
 [Documentation](https://docs.rs/tokio-signal/0.2.8/tokio_signal)
 
 ## Usage

tokio-signal/examples/ctrl-c.rs

@@ -7,7 +7,7 @@ use futures::{Future, Stream};
 /// how many signals to handle before exiting
 const STOP_AFTER: u64 = 10;
 
-fn main() -> Result<(), Box<std::error::Error>> {
+fn main() -> Result<(), Box<dyn std::error::Error>> {
     // tokio_signal provides a convenience builder for Ctrl+C
     // this even works cross-platform: linux and windows!
     //

tokio-signal/examples/multiple.rs

@@ -11,7 +11,7 @@ mod platform {
     use futures::{Future, Stream};
     use tokio_signal::unix::{Signal, SIGINT, SIGTERM};
 
-    pub fn main() -> Result<(), Box<::std::error::Error>> {
+    pub fn main() -> Result<(), Box<dyn std::error::Error>> {
         // Create a stream for each of the signals we'd like to handle.
         let sigint = Signal::new(SIGINT).flatten_stream();
         let sigterm = Signal::new(SIGTERM).flatten_stream();
@@ -39,7 +39,6 @@ mod platform {
         }
         Ok(())
     }
-
 }
 
 #[cfg(not(unix))]
@@ -49,6 +48,6 @@ mod platform {
     }
 }
 
-fn main() -> Result<(), Box<std::error::Error>> {
+fn main() -> Result<(), Box<dyn std::error::Error>> {
     platform::main()
 }

tokio-signal/examples/sighup-example.rs

@@ -9,7 +9,7 @@ mod platform {
     use futures::{Future, Stream};
     use tokio_signal::unix::{Signal, SIGHUP};
 
-    pub fn main() -> Result<(), Box<::std::error::Error>> {
+    pub fn main() -> Result<(), Box<dyn std::error::Error>> {
         // on Unix, we can listen to whatever signal we want, in this case: SIGHUP
         let stream = Signal::new(SIGHUP).flatten_stream();
 
@@ -38,7 +38,6 @@ mod platform {
         ::tokio::runtime::current_thread::block_on_all(future)?;
         Ok(())
     }
-
 }
 
 #[cfg(not(unix))]
@@ -48,6 +47,6 @@ mod platform {
     }
 }
 
-fn main() -> Result<(), Box<std::error::Error>> {
+fn main() -> Result<(), Box<dyn std::error::Error>> {
     platform::main()
 }

tokio-signal/src/lib.rs

@@ -1,8 +1,14 @@
-#![doc(html_root_url = "https://docs.rs/tokio-signal/0.2.8")]
+#![doc(html_root_url = "https://docs.rs/tokio-signal/0.2.9")]
 #![deny(missing_docs)]
 
 //! Asynchronous signal handling for Tokio
 //!
+//! > **Note:** This crate is **deprecated in tokio 0.2.x** and has been moved
+//! > into [`tokio::signal`] behind the `signal` [feature flag].
+//!
+//! [`tokio::signal`]: https://docs.rs/tokio/latest/tokio/signal/index.html
+//! [feature flag]: https://docs.rs/tokio/latest/tokio/index.html#feature-flags
+//!
 //! This crate implements asynchronous signal handling for Tokio, an
 //! asynchronous I/O framework in Rust. The primary type exported from this
 //! crate, `unix::Signal`, allows listening for arbitrary signals on Unix
@@ -86,9 +92,9 @@ pub mod unix;
 pub mod windows;
 
 /// A future whose error is `io::Error`
-pub type IoFuture<T> = Box<Future<Item = T, Error = io::Error> + Send>;
+pub type IoFuture<T> = Box<dyn Future<Item = T, Error = io::Error> + Send>;
 /// A stream whose error is `io::Error`
-pub type IoStream<T> = Box<Stream<Item = T, Error = io::Error> + Send>;
+pub type IoStream<T> = Box<dyn Stream<Item = T, Error = io::Error> + Send>;
 
 /// Creates a stream which receives "ctrl-c" notifications sent to a process.
 ///
@@ -125,7 +131,7 @@ pub fn ctrl_c_handle(handle: &Handle) -> IoFuture<IoStream<()>> {
         let handle = handle.clone();
         Box::new(future::lazy(move || {
             unix::Signal::with_handle(unix::libc::SIGINT, &handle)
-                .map(|x| Box::new(x.map(|_| ())) as Box<Stream<Item = _, Error = _> + Send>)
+                .map(|x| Box::new(x.map(|_| ())) as Box<dyn Stream<Item = _, Error = _> + Send>)
         }))
     }
 

tokio-sync/CHANGELOG.md

@@ -1,3 +1,12 @@
+# 0.1.8 (February 4, 2020)
+
+* Add `tokio 0.2.x` deprecation notice.
+
+# 0.1.7 (October 10, 2019)
+
+### Fixed
+- memory leak when polling oneshot handle from more than one task (#1649).
+
 # 0.1.6 (June 4, 2019)
 
 ### Added

tokio-sync/Cargo.toml

@@ -8,12 +8,12 @@ name = "tokio-sync"
 #   - README.md
 # - Update CHANGELOG.md.
 # - Create "v0.1.x" git tag.
-version = "0.1.6"
+version = "0.1.8"
 authors = ["Carl Lerche <[email protected]>"]
 license = "MIT"
 repository = "https://github.com/tokio-rs/tokio"
 homepage = "https://tokio.rs"
-documentation = "https://docs.rs/tokio-sync/0.1.6/tokio_sync"
+documentation = "https://docs.rs/tokio-sync/0.1.8/tokio_sync"
 description = """
 Synchronization utilities.
 """
@@ -24,7 +24,7 @@ fnv = "1.0.6"
 futures = "0.1.19"
 
 [dev-dependencies]
-env_logger = { version = "0.5", default-features = false }
+env_logger = { version = "0.6", default-features = false }
 tokio = { version = "0.1.15", path = "../tokio" }
 tokio-mock-task = "0.1.1"
 loom = { version = "0.1.1", features = ["futures"] }

tokio-sync/README.md

@@ -2,6 +2,12 @@
 
 Synchronization utilities
 
+> **Note:** This crate is **deprecated in tokio 0.2.x** and has been moved into
+> [`tokio::sync`] behind the `sync` [feature flag].
+
+[`tokio::sync`]: https://docs.rs/tokio/latest/tokio/sync/index.html
+[feature flag]: https://docs.rs/tokio/latest/tokio/index.html#feature-flags
+
 [Documentation](https://docs.rs/tokio-sync/0.1.6/tokio_sync/)
 
 ## Overview

tokio-sync/benches/mpsc.rs

@@ -1,5 +1,4 @@
 #![feature(test)]
-#![cfg_attr(test, deny(warnings))]
 
 extern crate futures;
 extern crate test;

tokio-sync/benches/oneshot.rs

@@ -1,5 +1,4 @@
 #![feature(test)]
-#![cfg_attr(test, deny(warnings))]
 
 extern crate futures;
 extern crate test;

tokio-sync/src/lib.rs

@@ -1,9 +1,14 @@
-#![doc(html_root_url = "https://docs.rs/tokio-sync/0.1.6")]
+#![doc(html_root_url = "https://docs.rs/tokio-sync/0.1.8")]
 #![deny(missing_debug_implementations, missing_docs, unreachable_pub)]
-#![cfg_attr(test, deny(warnings))]
 
 //! Asynchronous synchronization primitives.
 //!
+//! > **Note:** This crate is **deprecated in tokio 0.2.x** and has been moved into
+//! > [`tokio::sync`] behind the `sync` [feature flag].
+//!
+//! [`tokio::sync`]: https://docs.rs/tokio/latest/tokio/sync/index.html
+//! [feature flag]: https://docs.rs/tokio/latest/tokio/index.html#feature-flags
+//!
 //! This crate provides primitives for synchronizing asynchronous tasks.
 
 extern crate fnv;

tokio-sync/src/oneshot.rs

@@ -198,6 +198,8 @@ impl<T> Sender<T> {
                 state = State::unset_tx_task(&inner.state);
 
                 if state.is_closed() {
+                    // Set the flag again so that the waker is released in drop
+                    State::set_tx_task(&inner.state);
                     return Ok(Async::Ready(()));
                 } else {
                     unsafe { inner.drop_tx_task() };
@@ -363,6 +365,9 @@ impl<T> Inner<T> {
                     // Unset the task
                     state = State::unset_rx_task(&self.state);
                     if state.is_complete() {
+                        // Set the flag again so that the waker is released in drop
+                        State::set_rx_task(&self.state);
+
                         return match unsafe { self.consume_value() } {
                             Some(value) => Ok(Ready(value)),
                             None => Err(RecvError(())),

tokio-sync/tests/atomic_task.rs

@@ -1,5 +1,3 @@
-#![deny(warnings)]
-
 extern crate futures;
 extern crate tokio_mock_task;
 extern crate tokio_sync;

tokio-sync/tests/errors.rs

@@ -1,5 +1,3 @@
-#![deny(warnings)]
-
 extern crate tokio_sync;
 
 fn is_error<T: ::std::error::Error + Send + Sync>() {}

tokio-sync/tests/fuzz_list.rs

@@ -1,5 +1,3 @@
-#![deny(warnings)]
-
 extern crate futures;
 #[macro_use]
 extern crate loom;

tokio-sync/tests/fuzz_oneshot.rs

@@ -1,5 +1,3 @@
-#![deny(warnings)]
-
 extern crate futures;
 extern crate loom;
 

tokio-sync/tests/fuzz_semaphore.rs

@@ -1,5 +1,3 @@
-#![deny(warnings)]
-
 #[macro_use]
 extern crate futures;
 #[macro_use]

tokio-sync/tests/lock.rs

@@ -1,5 +1,3 @@
-#![deny(warnings)]
-
 extern crate futures;
 extern crate tokio_mock_task;
 extern crate tokio_sync;

tokio-sync/tests/mpsc.rs

@@ -1,5 +1,3 @@
-#![deny(warnings)]
-
 extern crate futures;
 extern crate tokio_mock_task;
 extern crate tokio_sync;

tokio-sync/tests/oneshot.rs

@@ -1,5 +1,3 @@
-#![deny(warnings)]
-
 extern crate futures;
 extern crate tokio_mock_task;
 extern crate tokio_sync;

tokio-sync/tests/semaphore.rs

@@ -1,5 +1,3 @@
-#![deny(warnings)]
-
 extern crate futures;
 extern crate tokio_mock_task;
 extern crate tokio_sync;

tokio-tcp/Cargo.toml

@@ -8,12 +8,12 @@ name = "tokio-tcp"
 #   - README.md
 # - Update CHANGELOG.md.
 # - Create "v0.1.x" git tag.
-version = "0.1.3"
+version = "0.1.4"
 authors = ["Carl Lerche <[email protected]>"]
 license = "MIT"
 repository = "https://github.com/tokio-rs/tokio"
 homepage = "https://tokio.rs"
-documentation = "https://docs.rs/tokio-tcp/0.1.3/tokio_tcp"
+documentation = "https://docs.rs/tokio-tcp/0.1.4/tokio_tcp"
 description = """
 TCP bindings for tokio.
 """
@@ -28,6 +28,6 @@ iovec = "0.1"
 futures = "0.1.19"
 
 [dev-dependencies]
-env_logger = { version = "0.5", default-features = false }
+env_logger = { version = "0.6", default-features = false }
 net2 = "0.2"
 tokio = "0.1.13"

tokio-tcp/README.md

@@ -2,6 +2,12 @@
 
 TCP bindings for `tokio`.
 
+> **Note:** This crate is **deprecated in tokio 0.2.x** and has been moved into
+> [`tokio::net`] behind the `tcp` [feature flag].
+
+[`tokio::net`]: https://docs.rs/tokio/latest/tokio/net/index.html
+[feature flag]: https://docs.rs/tokio/latest/tokio/index.html#feature-flags
+
 [Documentation](https://docs.rs/tokio-tcp/0.1.3/tokio_tcp)
 
 ## License

tokio-tcp/src/lib.rs

@@ -1,8 +1,14 @@
 #![doc(html_root_url = "https://docs.rs/tokio-tcp/0.1.3")]
-#![deny(missing_docs, warnings, missing_debug_implementations)]
+#![deny(missing_docs, missing_debug_implementations)]
 
 //! TCP bindings for `tokio`.
 //!
+//! > **Note:** This crate is **deprecated in tokio 0.2.x** and has been moved
+//! > into [`tokio::net`] behind the `tcp` [feature flag].
+//!
+//! [`tokio::net`]: https://docs.rs/tokio/latest/tokio/net/index.html
+//! [feature flag]: https://docs.rs/tokio/latest/tokio/index.html#feature-flags
+//!
 //! This module contains the TCP networking types, similar to the standard
 //! library, which can be used to implement networking protocols.
 //!