v0.1.x: prepare to release new reactor, executor, and timer (#1751)

## Motivation

The `CurrentThread` runtime's `Handle` is different than the
`CurrentThread` executor's `Handle`. This causes interoperability issues
with custom runtimes that build on the `CurrentThread` executor.

Actix-rt offers a concrete example of the issue: [`System::run_in_executor`][1]
requires a `CurrentThread` executor handle - the change in this PR
allows a `CurrentThread` runtime to be used here.

## Solution

This PR adds `fn into_inner(self)` on the runtime `Handle` that consumes
it and returns the underlying `CurrentThread` executor's `Handle`.

[1]: https://docs.rs/actix-rt/0.2.6/actix_rt/struct.System.html#method.run_in_executor

.cirrus.yml

@@ -32,8 +32,6 @@ task:
   test_script:
     - . $HOME/.cargo/env
     - cargo test --all
-    - (cd tokio-trace/test-log-support && cargo test)
-    - (cd tokio-trace/test_static_max_level_features && cargo test)
     - cargo doc --all
   i686_test_script:
     - . $HOME/.cargo/env

Cargo.toml

@@ -9,7 +9,6 @@ members = [
   "tokio-fs",
   "tokio-futures",
   "tokio-io",
-  "tokio-macros",
   "tokio-reactor",
   "tokio-signal",
   "tokio-sync",
@@ -18,8 +17,6 @@ members = [
   "tokio-timer",
   "tokio-tcp",
   "tokio-tls",
-  "tokio-trace",
-  "tokio-trace/tokio-trace-core",
   "tokio-udp",
   "tokio-uds",
 ]

README.md

@@ -170,6 +170,23 @@ The crates included as part of Tokio are:
 [`tokio-udp`]: tokio-udp
 [`tokio-uds`]: tokio-uds
 
+## Related Projects
+
+In addition to the crates in this repository, the Tokio project also maintains
+several other libraries, including:
+
+* [`tracing`] (formerly `tokio-trace`): A framework for application-level
+  tracing and async-aware diagnostics.
+
+* [`mio`]: A low-level, cross-platform abstraction over OS I/O APIs that powers
+  `tokio`.
+
+* [`bytes`]: Utilities for working with bytes, including efficient byte buffers.
+
+[`tracing`]: https://github.com/tokio-rs/tracing
+[`mio`]: https://github.com/tokio-rs/mio
+[`bytes`]: https://github.com/tokio-rs/bytes
+
 ## Supported Rust Versions
 
 Tokio is built against the latest stable, nightly, and beta Rust releases. The

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

@@ -46,10 +46,6 @@ jobs:
       - tokio-threadpool
       - tokio-timer
       - tokio-test
-      - tokio-trace
-      - tokio-trace/tokio-trace-core
-      - tokio-trace/test-log-support
-      - tokio-trace/test_static_max_level_features
 
 - template: ci/azure-cargo-check.yml
   parameters:
@@ -68,16 +64,10 @@ jobs:
         - udp
         - uds
         - sync
+        - experimental-tracing
       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:
@@ -94,7 +84,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:
@@ -108,7 +98,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-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/patch.toml

@@ -16,7 +16,5 @@ tokio-threadpool = { path = "tokio-threadpool" }
 tokio-timer = { path = "tokio-timer" }
 tokio-tcp = { path = "tokio-tcp" }
 tokio-tls = { path = "tokio-tls" }
-tokio-trace = { path = "tokio-trace" }
-tokio-trace-core = { path = "tokio-trace/tokio-trace-core" }
 tokio-udp = { path = "tokio-udp" }
 tokio-uds = { path = "tokio-uds" }

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/src/lib.rs

@@ -1,6 +1,5 @@
 #![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.
 //!

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/src/lib.rs

@@ -1,4 +1,4 @@
-#![deny(missing_docs, missing_debug_implementations, warnings)]
+#![deny(missing_docs, missing_debug_implementations)]
 #![doc(html_root_url = "https://docs.rs/tokio-codec/0.1.1")]
 
 //! Utilities for encoding and decoding frames.

tokio-current-thread/src/lib.rs

@@ -1,5 +1,5 @@
 #![doc(html_root_url = "https://docs.rs/tokio-current-thread/0.1.6")]
-#![deny(warnings, missing_docs, missing_debug_implementations)]
+#![deny(missing_docs, missing_debug_implementations)]
 
 //! A single-threaded executor which executes tasks on the same thread from which
 //! they are spawned.
@@ -64,7 +64,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 +186,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 +428,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 +633,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 +735,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 +750,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 +795,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 +812,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 +827,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,14 @@
+# 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
+- Add `executor::exit` to allow other executors inside `threadpool::blocking` (#1155).
+
 # 0.1.7 (March 22, 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.7"
-documentation = "https://docs.rs/tokio-executor/0.1.7/tokio_executor"
+version = "0.1.9"
+documentation = "https://docs.rs/tokio-executor/0.1.9/tokio_executor"
 repository = "https://github.com/tokio-rs/tokio"
 homepage = "https://github.com/tokio-rs/tokio"
 license = "MIT"

tokio-executor/README.md

@@ -2,7 +2,7 @@
 
 Task execution related traits and utilities.
 
-[Documentation](https://docs.rs/tokio-executor/0.1.7/tokio_executor)
+[Documentation](https://docs.rs/tokio-executor/0.1.9/tokio_executor)
 
 ## Overview
 
@@ -31,10 +31,10 @@ executor, including:
 
 * [`Park`] abstracts over blocking and unblocking the current thread.
 
-[`Executor`]: https://docs.rs/tokio-executor/0.1.7/tokio_executor/trait.Executor.html
-[`enter`]: https://docs.rs/tokio-executor/0.1.7/tokio_executor/fn.enter.html
-[`DefaultExecutor`]: https://docs.rs/tokio-executor/0.1.7/tokio_executor/struct.DefaultExecutor.html
-[`Park`]: https://docs.rs/tokio-executor/0.1.7/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,
 }
 
@@ -67,6 +67,42 @@ pub fn enter() -> Result<Enter, EnterError> {
     })
 }
 
+// Forces the current "entered" state to be cleared while the closure
+// is executed.
+//
+// # Warning
+//
+// This is hidden for a reason. Do not use without fully understanding
+// executors. Misuing can easily cause your program to deadlock.
+#[doc(hidden)]
+pub fn exit<F: FnOnce() -> R, R>(f: F) -> R {
+    // Reset in case the closure panics
+    struct Reset;
+    impl Drop for Reset {
+        fn drop(&mut self) {
+            ENTERED.with(|c| {
+                c.set(true);
+            });
+        }
+    }
+
+    ENTERED.with(|c| {
+        debug_assert!(c.get());
+        c.set(false);
+    });
+
+    let reset = Reset;
+    let ret = f();
+    ::std::mem::forget(reset);
+
+    ENTERED.with(|c| {
+        assert!(!c.get(), "closure claimed permanent executor");
+        c.set(true);
+    });
+
+    ret
+}
+
 impl Enter {
     /// Register a callback to be invoked if and when the thread
     /// ceased to act as an executor.

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,5 +1,5 @@
-#![deny(missing_docs, missing_debug_implementations, warnings)]
-#![doc(html_root_url = "https://docs.rs/tokio-executor/0.1.7")]
+#![deny(missing_docs, missing_debug_implementations)]
+#![doc(html_root_url = "https://docs.rs/tokio-executor/0.1.9")]
 
 //! Task execution related traits and utilities.
 //!
@@ -61,8 +61,8 @@ mod global;
 pub mod park;
 mod typed;
 
-pub use enter::{enter, Enter, EnterError};
+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/Cargo.toml

@@ -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/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,4 +1,4 @@
-#![deny(missing_docs, missing_debug_implementations, warnings)]
+#![deny(missing_docs, missing_debug_implementations)]
 #![doc(html_root_url = "https://docs.rs/tokio-fs/0.1.6")]
 
 //! 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/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,4 +1,4 @@
-#![deny(missing_docs, missing_debug_implementations, warnings)]
+#![deny(missing_docs, missing_debug_implementations)]
 #![doc(html_root_url = "https://docs.rs/tokio-io/0.1.12")]
 
 //! Core I/O traits and combinators when working with Tokio.
@@ -21,10 +21,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 +65,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-reactor/CHANGELOG.md

@@ -1,3 +1,15 @@
+# 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,13 +8,13 @@ name = "tokio-reactor"
 #   - README.md
 # - Update CHANGELOG.md.
 # - Create "v0.1.x" git tag.
-version = "0.1.9"
+version = "0.1.11"
 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.11/tokio_reactor"
 description = """
 Event loop that drives Tokio I/O resources.
 """
@@ -27,7 +27,7 @@ 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,7 @@
 
 Event loop that drives Tokio I/O resources.
 
-[Documentation](https://docs.rs/tokio-reactor/0.1.9/tokio_reactor)
+[Documentation](https://docs.rs/tokio-reactor/0.1.11/tokio_reactor)
 
 ## Overview
 
@@ -25,10 +25,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,5 +1,5 @@
-#![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.11")]
+#![deny(missing_docs, missing_debug_implementations)]
 
 //! Event loop that drives Tokio I/O resources.
 //!
@@ -133,6 +133,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 +204,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 +631,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 +652,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 +690,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 +743,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/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

@@ -86,9 +86,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 +125,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,8 @@
+# 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.7"
 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.7/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/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,6 +1,5 @@
-#![doc(html_root_url = "https://docs.rs/tokio-sync/0.1.6")]
+#![doc(html_root_url = "https://docs.rs/tokio-sync/0.1.7")]
 #![deny(missing_debug_implementations, missing_docs, unreachable_pub)]
-#![cfg_attr(test, deny(warnings))]
 
 //! Asynchronous synchronization primitives.
 //!

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

@@ -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/src/lib.rs

@@ -1,5 +1,5 @@
 #![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`.
 //!

tokio-test/src/lib.rs

@@ -1,6 +1,5 @@
 #![doc(html_root_url = "https://docs.rs/tokio-test/0.1.0")]
 #![deny(missing_docs, missing_debug_implementations, unreachable_pub)]
-#![cfg_attr(test, deny(warnings))]
 
 //! Tokio and Futures based testing utilites
 //!

tokio-test/src/macros.rs

@@ -127,5 +127,4 @@ mod tests {
         let mut fut = future::ok::<(), ()>(());
         assert_ready_eq!(fut.poll(), ());
     }
-
 }

tokio-threadpool/CHANGELOG.md

@@ -1,3 +1,16 @@
+# 0.1.16 (September 25, 2019)
+
+### Changed
+- Remove last non-dev dependency on rand crate by seeding PRNG via libstd
+  `RandomState` (#1324 backport)
+- Upgrade (dev-only dependency) rand to 0.7.0 (#1302 backport)
+- The minimum supported rust version (MSRV) is now 1.31.0 (#1358)
+
+# 0.1.15 (June 2, 2019)
+
+### Changed
+- Allow other executors inside `threadpool::blocking` (#1155).
+
 # 0.1.14 (April 22, 2019)
 
 ### Added

tokio-threadpool/Cargo.toml

@@ -8,8 +8,8 @@ name = "tokio-threadpool"
 #   - README.md
 # - Update CHANGELOG.md.
 # - Create "v0.1.x" git tag.
-version = "0.1.14"
-documentation = "https://docs.rs/tokio-threadpool/0.1.14/tokio_threadpool"
+version = "0.1.16"
+documentation = "https://docs.rs/tokio-threadpool/0.1.16/tokio_threadpool"
 repository = "https://github.com/tokio-rs/tokio"
 homepage = "https://github.com/tokio-rs/tokio"
 license = "MIT"
@@ -21,18 +21,19 @@ keywords = ["futures", "tokio"]
 categories = ["concurrency", "asynchronous"]
 
 [dependencies]
-tokio-executor = "0.1.7"
+tokio-executor = "0.1.8"
 futures = "0.1.19"
 crossbeam-deque = "0.7.0"
 crossbeam-queue = "0.1.0"
 crossbeam-utils = "0.6.4"
 num_cpus = "1.2"
-rand = "0.6"
 slab = "0.4.1"
 log = "0.4"
+lazy_static = "1"
 
 [dev-dependencies]
-env_logger = "0.5"
+rand = "0.7"
+env_logger = { version = "0.6", default-features = false }
 
 # For comparison benchmarks
 futures-cpupool = "0.1.7"

tokio-threadpool/README.md

@@ -3,7 +3,7 @@
 A library for scheduling execution of futures concurrently across a pool of
 threads.
 
-[Documentation](https://docs.rs/tokio-threadpool/0.1.14/tokio_threadpool)
+[Documentation](https://docs.rs/tokio-threadpool/0.1.16/tokio_threadpool)
 
 ### Why not Rayon?
 

tokio-threadpool/benches/basic.rs

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

tokio-threadpool/benches/blocking.rs

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

tokio-threadpool/benches/depth.rs

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

tokio-threadpool/src/blocking/global.rs

@@ -1,13 +1,55 @@
-use worker::Worker;
-
+use super::{BlockingError, BlockingImpl};
 use futures::Poll;
-
-use std::error::Error;
+use std::cell::Cell;
 use std::fmt;
+use std::marker::PhantomData;
+use tokio_executor::Enter;
 
-/// Error raised by `blocking`.
-pub struct BlockingError {
-    _p: (),
+thread_local! {
+    static CURRENT: Cell<BlockingImpl> = Cell::new(super::default_blocking);
+}
+
+/// Ensures that the executor is removed from the thread-local context
+/// when leaving the scope. This handles cases that involve panicking.
+///
+/// **NOTE:** This is intended specifically for use by `tokio` 0.2's
+/// backwards-compatibility layer. In general, user code should not override the
+/// blocking implementation. If you use this, make sure you know what you're
+/// doing.
+pub struct DefaultGuard<'a> {
+    prior: BlockingImpl,
+    _lifetime: PhantomData<&'a ()>,
+}
+
+/// Set the default blocking implementation, returning a guard that resets the
+/// blocking implementation when dropped.
+///
+/// **NOTE:** This is intended specifically for use by `tokio` 0.2's
+/// backwards-compatibility layer. In general, user code should not override the
+/// blocking implementation. If you use this, make sure you know what you're
+/// doing.
+pub fn set_default<'a>(blocking: BlockingImpl) -> DefaultGuard<'a> {
+    CURRENT.with(|cell| {
+        let prior = cell.replace(blocking);
+        DefaultGuard {
+            prior,
+            _lifetime: PhantomData,
+        }
+    })
+}
+
+/// Set the default blocking implementation for the duration of the closure.
+///
+/// **NOTE:** This is intended specifically for use by `tokio` 0.2's
+/// backwards-compatibility layer. In general, user code should not override the
+/// blocking implementation. If you use this, make sure you know what you're
+/// doing.
+pub fn with_default<F, R>(blocking: BlockingImpl, enter: &mut Enter, f: F) -> R
+where
+    F: FnOnce(&mut Enter) -> R,
+{
+    let _guard = set_default(blocking);
+    f(enter)
 }
 
 /// Enter a blocking section of code.
@@ -125,53 +167,52 @@ pub fn blocking<F, T>(f: F) -> Poll<T, BlockingError>
 where
     F: FnOnce() -> T,
 {
-    let res = Worker::with_current(|worker| {
-        let worker = match worker {
-            Some(worker) => worker,
-            None => {
-                return Err(BlockingError { _p: () });
-            }
-        };
+    CURRENT.with(|cell| {
+        let blocking = cell.get();
 
-        // Transition the worker state to blocking. This will exit the fn early
-        // with `NotReady` if the pool does not have enough capacity to enter
-        // blocking mode.
-        worker.transition_to_blocking()
-    });
+        // Object-safety workaround: the `Blocking` trait must be object-safe,
+        // since we use a trait object in the thread-local. However, a blocking
+        // _operation_ will be generic over the return type of the blocking
+        // function. Therefore, rather than passing a function with a return
+        // type to `Blocking::run_blocking`, we pass a _new_ closure which
+        // doesn't have a return value. That closure invokes the blocking
+        // function and assigns its value to `ret`, which we then unpack when
+        // the blocking call finishes.
+        let mut f = Some(f);
+        let mut ret = None;
+        {
+            let ret2 = &mut ret;
+            let mut run = move || {
+                let f = f
+                    .take()
+                    .expect("blocking closure invoked twice; this is a bug!");
+                *ret2 = Some((f)());
+            };
 
-    // If the transition cannot happen, exit early
-    try_ready!(res);
+            try_ready!((blocking)(&mut run));
+        }
 
-    // Currently in blocking mode, so call the inner closure
-    let ret = f();
-
-    // Try to transition out of blocking mode. This is a fast path that takes
-    // back ownership of the worker if the worker handoff didn't complete yet.
-    Worker::with_current(|worker| {
-        // Worker must be set since it was above.
-        worker.unwrap().transition_from_blocking();
-    });
-
-    // Return the result
-    Ok(ret.into())
+        // Return the result
+        let ret =
+            ret.expect("blocking function finished, but return value was unset; this is a bug!");
+        Ok(ret.into())
+    })
 }
 
-impl fmt::Display for BlockingError {
-    fn fmt(&self, fmt: &mut fmt::Formatter) -> fmt::Result {
-        write!(fmt, "{}", self.description())
-    }
-}
+// === impl DefaultGuard ===
 
-impl fmt::Debug for BlockingError {
+impl<'a> fmt::Debug for DefaultGuard<'a> {
     fn fmt(&self, f: &mut fmt::Formatter) -> fmt::Result {
-        f.debug_struct("BlockingError")
-            .field("reason", &self.description())
-            .finish()
+        f.pad("DefaultGuard { .. }")
     }
 }
 
-impl Error for BlockingError {
-    fn description(&self) -> &str {
-        "`blocking` annotation used from outside the context of a thread pool"
+impl<'a> Drop for DefaultGuard<'a> {
+    fn drop(&mut self) {
+        // if the TLS value has already been torn down, there's nothing else we
+        // can do. we're almost certainly panicking anyway.
+        let _ = CURRENT.try_with(|cell| {
+            cell.set(self.prior);
+        });
     }
 }

tokio-threadpool/src/blocking/mod.rs

@@ -0,0 +1,88 @@
+use worker::Worker;
+
+use futures::{Async, Poll};
+use tokio_executor;
+
+use std::error::Error;
+use std::fmt;
+
+mod global;
+pub use self::global::blocking;
+#[doc(hidden)]
+pub use self::global::{set_default, with_default, DefaultGuard};
+
+/// Error raised by `blocking`.
+pub struct BlockingError {
+    _p: (),
+}
+
+/// A function implementing the behavior run on calls to `blocking`.
+///
+/// **NOTE:** This is intended specifically for use by `tokio` 0.2's
+/// backwards-compatibility layer. In general, user code should not override the
+/// blocking implementation. If you use this, make sure you know what you're
+/// doing.
+#[doc(hidden)]
+pub type BlockingImpl = fn(&mut dyn FnMut()) -> Poll<(), BlockingError>;
+
+fn default_blocking(f: &mut dyn FnMut()) -> Poll<(), BlockingError> {
+    let res = Worker::with_current(|worker| {
+        let worker = match worker {
+            Some(worker) => worker,
+            None => {
+                return Err(BlockingError::new());
+            }
+        };
+
+        // Transition the worker state to blocking. This will exit the fn early
+        // with `NotReady` if the pool does not have enough capacity to enter
+        // blocking mode.
+        worker.transition_to_blocking()
+    });
+
+    // If the transition cannot happen, exit early
+    try_ready!(res);
+
+    // Currently in blocking mode, so call the inner closure.
+    //
+    // "Exit" the current executor in case the blocking function wants
+    // to call a different executor.
+    tokio_executor::exit(move || (f)());
+
+    // Try to transition out of blocking mode. This is a fast path that takes
+    // back ownership of the worker if the worker handoff didn't complete yet.
+    Worker::with_current(|worker| {
+        // Worker must be set since it was above.
+        worker.unwrap().transition_from_blocking();
+    });
+
+    Ok(Async::Ready(()))
+}
+
+impl BlockingError {
+    /// Returns a new `BlockingError`.
+    #[doc(hidden)]
+    pub fn new() -> Self {
+        Self { _p: () }
+    }
+}
+
+impl fmt::Display for BlockingError {
+    fn fmt(&self, fmt: &mut fmt::Formatter) -> fmt::Result {
+        write!(fmt, "{}", self.description())
+    }
+}
+
+impl fmt::Debug for BlockingError {
+    fn fmt(&self, f: &mut fmt::Formatter) -> fmt::Result {
+        f.debug_struct("BlockingError")
+            .field("reason", &self.description())
+            .finish()
+    }
+}
+
+impl Error for BlockingError {
+    fn description(&self) -> &str {
+        "`blocking` annotation used from outside the context of a thread pool"
+    }
+}

tokio-threadpool/src/builder.rs

@@ -67,7 +67,7 @@ pub struct Builder {
     max_blocking: usize,
 
     /// Generates the `Park` instances
-    new_park: Box<Fn(&WorkerId) -> BoxPark>,
+    new_park: Box<dyn Fn(&WorkerId) -> BoxPark>,
 }
 
 impl Builder {
@@ -223,7 +223,7 @@ impl Builder {
     /// ```
     pub fn panic_handler<F>(&mut self, f: F) -> &mut Self
     where
-        F: Fn(Box<Any + Send>) + Send + Sync + 'static,
+        F: Fn(Box<dyn Any + Send>) + Send + Sync + 'static,
     {
         self.config.panic_handler = Some(Arc::new(f));
         self

tokio-threadpool/src/callback.rs

@@ -7,7 +7,7 @@ use tokio_executor::Enter;
 
 #[derive(Clone)]
 pub(crate) struct Callback {
-    f: Arc<Fn(&Worker, &mut Enter) + Send + Sync>,
+    f: Arc<dyn Fn(&Worker, &mut Enter) + Send + Sync>,
 }
 
 impl Callback {

tokio-threadpool/src/config.rs

@@ -13,9 +13,9 @@ pub(crate) struct Config {
     pub name_prefix: Option<String>,
     pub stack_size: Option<usize>,
     pub around_worker: Option<Callback>,
-    pub after_start: Option<Arc<Fn() + Send + Sync>>,
-    pub before_stop: Option<Arc<Fn() + Send + Sync>>,
-    pub panic_handler: Option<Arc<Fn(Box<Any + Send>) + Send + Sync>>,
+    pub after_start: Option<Arc<dyn Fn() + Send + Sync>>,
+    pub before_stop: Option<Arc<dyn Fn() + Send + Sync>>,
+    pub panic_handler: Option<Arc<dyn Fn(Box<dyn Any + Send>) + Send + Sync>>,
 }
 
 /// Max number of workers that can be part of a pool. This is the most that can

tokio-threadpool/src/lib.rs

@@ -1,5 +1,5 @@
-#![doc(html_root_url = "https://docs.rs/tokio-threadpool/0.1.14")]
-#![deny(warnings, missing_docs, missing_debug_implementations)]
+#![doc(html_root_url = "https://docs.rs/tokio-threadpool/0.1.16")]
+#![deny(missing_docs, missing_debug_implementations)]
 
 //! A work-stealing based thread pool for executing futures.
 //!
@@ -84,8 +84,9 @@ extern crate crossbeam_queue;
 extern crate crossbeam_utils;
 #[macro_use]
 extern crate futures;
+#[macro_use]
+extern crate lazy_static;
 extern crate num_cpus;
-extern crate rand;
 extern crate slab;
 
 #[macro_use]
@@ -141,13 +142,13 @@ extern crate log;
 //
 // [Treiber stack]: https://en.wikipedia.org/wiki/Treiber_Stack
 
-pub mod park;
-
-mod blocking;
+#[doc(hidden)]
+pub mod blocking;
 mod builder;
 mod callback;
 mod config;
 mod notifier;
+pub mod park;
 mod pool;
 mod sender;
 mod shutdown;

tokio-threadpool/src/park/boxed.rs

@@ -3,8 +3,8 @@ use tokio_executor::park::{Park, Unpark};
 use std::error::Error;
 use std::time::Duration;
 
-pub(crate) type BoxPark = Box<Park<Unpark = BoxUnpark, Error = ()> + Send>;
-pub(crate) type BoxUnpark = Box<Unpark>;
+pub(crate) type BoxPark = Box<dyn Park<Unpark = BoxUnpark, Error = ()> + Send>;
+pub(crate) type BoxUnpark = Box<dyn Unpark>;
 
 pub(crate) struct BoxedPark<T>(T);
 

tokio-threadpool/src/pool/mod.rs

@@ -17,6 +17,8 @@ use worker::{self, Worker, WorkerId};
 use futures::Poll;
 
 use std::cell::Cell;
+use std::collections::hash_map::RandomState;
+use std::hash::{BuildHasher, Hash, Hasher};
 use std::num::Wrapping;
 use std::sync::atomic::AtomicUsize;
 use std::sync::atomic::Ordering::{AcqRel, Acquire};
@@ -25,7 +27,6 @@ use std::thread;
 
 use crossbeam_deque::Injector;
 use crossbeam_utils::CachePadded;
-use rand;
 
 #[derive(Debug)]
 pub(crate) struct Pool {
@@ -420,11 +421,7 @@ impl Pool {
     /// Uses a thread-local random number generator based on XorShift.
     pub fn rand_usize(&self) -> usize {
         thread_local! {
-            static RNG: Cell<Wrapping<u32>> = {
-                // The initial seed must be non-zero.
-                let init = rand::random::<u32>() | 1;
-                Cell::new(Wrapping(init))
-            }
+            static RNG: Cell<Wrapping<u32>> = Cell::new(Wrapping(prng_seed()));
         }
 
         RNG.with(|rng| {
@@ -448,3 +445,31 @@ impl PartialEq for Pool {
 
 unsafe impl Send for Pool {}
 unsafe impl Sync for Pool {}
+
+// Return a thread-specific, 32-bit, non-zero seed value suitable for a 32-bit
+// PRNG. This uses one libstd RandomState for a default hasher and hashes on
+// the current thread ID to obtain an unpredictable, collision resistant seed.
+fn prng_seed() -> u32 {
+    // This obtains a small number of random bytes from the host system (for
+    // example, on unix via getrandom(2)) in order to seed an unpredictable and
+    // HashDoS resistant 64-bit hash function (currently: `SipHasher13` with
+    // 128-bit state). We only need one of these, to make the seeds for all
+    // process threads different via hashed IDs, collision resistant, and
+    // unpredictable.
+    lazy_static! {
+        static ref RND_STATE: RandomState = RandomState::new();
+    }
+
+    // Hash the current thread ID to produce a u32 value
+    let mut hasher = RND_STATE.build_hasher();
+    thread::current().id().hash(&mut hasher);
+    let hash: u64 = hasher.finish();
+    let seed = (hash as u32) ^ ((hash >> 32) as u32);
+
+    // Ensure non-zero seed (Xorshift yields only zero's for that seed)
+    if seed == 0 {
+        0x9b4e_6d25 // misc bits, could be any non-zero
+    } else {
+        seed
+    }
+}

tokio-threadpool/src/sender.rs

@@ -131,7 +131,7 @@ impl tokio_executor::Executor for Sender {
 
     fn spawn(
         &mut self,
-        future: Box<Future<Item = (), Error = ()> + Send>,
+        future: Box<dyn Future<Item = (), Error = ()> + Send>,
     ) -> Result<(), SpawnError> {
         let mut s = &*self;
         tokio_executor::Executor::spawn(&mut s, future)
@@ -157,7 +157,7 @@ impl<'a> tokio_executor::Executor for &'a Sender {
 
     fn spawn(
         &mut self,
-        future: Box<Future<Item = (), Error = ()> + Send>,
+        future: Box<dyn Future<Item = (), Error = ()> + Send>,
     ) -> Result<(), SpawnError> {
         self.prepare_for_spawn()?;
 

tokio-threadpool/src/task/mod.rs

@@ -60,7 +60,7 @@ pub(crate) enum Run {
     Complete,
 }
 
-type BoxFuture = Box<Future<Item = (), Error = ()> + Send + 'static>;
+type BoxFuture = Box<dyn Future<Item = (), Error = ()> + Send + 'static>;
 
 // ===== impl Task =====
 

tokio-threadpool/tests/blocking.rs

@@ -1,3 +1,4 @@
+extern crate tokio_executor;
 extern crate tokio_threadpool;
 
 extern crate env_logger;
@@ -44,6 +45,28 @@ fn basic() {
     rx2.recv().unwrap();
 }
 
+#[test]
+fn other_executors_can_run_inside_blocking() {
+    let _ = ::env_logger::try_init();
+
+    let pool = Builder::new().pool_size(1).max_blocking(1).build();
+
+    let (tx, rx) = mpsc::channel();
+
+    pool.spawn(lazy(move || {
+        let res = blocking(|| {
+            let _e = tokio_executor::enter().expect("nested blocking enter");
+            tx.send(()).unwrap();
+        })
+        .unwrap();
+
+        assert!(res.is_ready());
+        Ok(().into())
+    }));
+
+    rx.recv().unwrap();
+}
+
 #[test]
 fn notify_task_on_capacity() {
     const BLOCKING: usize = 10;

tokio-threadpool/tests/threadpool.rs

@@ -18,7 +18,9 @@ use std::time::Duration;
 
 thread_local!(static FOO: Cell<u32> = Cell::new(0));
 
-fn ignore_results<F: Future + Send + 'static>(f: F) -> Box<Future<Item = (), Error = ()> + Send> {
+fn ignore_results<F: Future + Send + 'static>(
+    f: F,
+) -> Box<dyn Future<Item = (), Error = ()> + Send> {
     Box::new(f.map(|_| ()).map_err(|_| ()))
 }
 

tokio-timer/CHANGELOG.md

@@ -1,3 +1,11 @@
+# 0.2.12 (November 27, 2019)
+
+### Added
+- `timer::set_default`, which functions like `timer::with_default`, but
+  returns a drop guard (#1725).
+- `clock::set_default`, which functions like `clock::with_default`, but
+  returns a drop guard (#1725).
+
 # 0.2.11 (May 14, 2019)
 
 ### Added

tokio-timer/Cargo.toml

@@ -8,11 +8,11 @@ name = "tokio-timer"
 #   - README.md
 # - Update CHANGELOG.md.
 # - Create "v0.2.x" git tag.
-version = "0.2.11"
+version = "0.2.12"
 authors = ["Carl Lerche <[email protected]>"]
 license = "MIT"
 readme = "README.md"
-documentation = "https://docs.rs/tokio-timer/0.2.11/tokio_timer"
+documentation = "https://docs.rs/tokio-timer/0.2.12/tokio_timer"
 repository = "https://github.com/tokio-rs/tokio"
 homepage = "https://github.com/tokio-rs/tokio"
 description = """
@@ -28,6 +28,6 @@ crossbeam-utils = "0.6.0"
 slab = "0.4.1"
 
 [dev-dependencies]
-rand = "0.6"
+rand = "0.7"
 tokio-mock-task = "0.1.0"
 tokio = "0.1.7"

tokio-timer/README.md

@@ -2,7 +2,7 @@
 
 Timer facilities for Tokio
 
-[Documentation](https://docs.rs/tokio-timer/0.2.11/tokio_timer/)
+[Documentation](https://docs.rs/tokio-timer/0.2.12/tokio_timer/)
 
 ## Overview
 

tokio-timer/src/clock/clock.rs

@@ -3,7 +3,7 @@ use timer;
 
 use tokio_executor::Enter;
 
-use std::cell::Cell;
+use std::cell::RefCell;
 use std::fmt;
 use std::sync::Arc;
 use std::time::Instant;
@@ -17,12 +17,18 @@ use std::time::Instant;
 /// [`Instant::now`]: https://doc.rust-lang.org/std/time/struct.Instant.html#method.now
 #[derive(Default, Clone)]
 pub struct Clock {
-    now: Option<Arc<Now>>,
+    now: Option<Arc<dyn Now>>,
+}
+
+/// A guard that resets the current `Clock` to `None` when dropped.
+#[derive(Debug)]
+pub struct DefaultGuard {
+    _p: (),
 }
 
 thread_local! {
     /// Thread-local tracking the current clock
-    static CLOCK: Cell<Option<*const Clock>> = Cell::new(None)
+    static CLOCK: RefCell<Option<Clock>> = RefCell::new(None)
 }
 
 /// Returns an `Instant` corresponding to "now".
@@ -43,8 +49,8 @@ thread_local! {
 /// let now = clock::now();
 /// ```
 pub fn now() -> Instant {
-    CLOCK.with(|current| match current.get() {
-        Some(ptr) => unsafe { (*ptr).now() },
+    CLOCK.with(|current| match current.borrow().as_ref() {
+        Some(c) => c.now(),
         None => Instant::now(),
     })
 }
@@ -53,8 +59,8 @@ impl Clock {
     /// Return a new `Clock` instance that uses the current execution context's
     /// source of time.
     pub fn new() -> Clock {
-        CLOCK.with(|current| match current.get() {
-            Some(ptr) => unsafe { (*ptr).clone() },
+        CLOCK.with(|current| match current.borrow().as_ref() {
+            Some(c) => c.clone(),
             None => Clock::system(),
         })
     }
@@ -114,26 +120,31 @@ pub fn with_default<F, R>(clock: &Clock, enter: &mut Enter, f: F) -> R
 where
     F: FnOnce(&mut Enter) -> R,
 {
+    let _guard = set_default(clock);
+
+    f(enter)
+}
+
+/// Sets `clock` as the default clock, returning a guard that unsets it on drop.
+///
+/// # Panics
+///
+/// This function panics if there already is a default clock set.
+pub fn set_default(clock: &Clock) -> DefaultGuard {
     CLOCK.with(|cell| {
         assert!(
-            cell.get().is_none(),
+            cell.borrow().is_none(),
             "default clock already set for execution context"
         );
 
-        // Ensure that the clock is removed from the thread-local context
-        // when leaving the scope. This handles cases that involve panicking.
-        struct Reset<'a>(&'a Cell<Option<*const Clock>>);
+        *cell.borrow_mut() = Some(clock.clone());
 
-        impl<'a> Drop for Reset<'a> {
-            fn drop(&mut self) {
-                self.0.set(None);
-            }
-        }
-
-        let _reset = Reset(cell);
-
-        cell.set(Some(clock as *const Clock));
-
-        f(enter)
+        DefaultGuard { _p: () }
     })
 }
+
+impl Drop for DefaultGuard {
+    fn drop(&mut self) {
+        let _ = CLOCK.try_with(|cell| cell.borrow_mut().take());
+    }
+}

tokio-timer/src/clock/mod.rs

@@ -19,5 +19,5 @@
 mod clock;
 mod now;
 
-pub use self::clock::{now, with_default, Clock};
+pub use self::clock::{now, set_default, with_default, Clock, DefaultGuard};
 pub use self::now::Now;

tokio-timer/src/lib.rs

@@ -1,5 +1,5 @@
-#![doc(html_root_url = "https://docs.rs/tokio-timer/0.2.11")]
-#![deny(missing_docs, warnings, missing_debug_implementations)]
+#![doc(html_root_url = "https://docs.rs/tokio-timer/0.2.12")]
+#![deny(missing_docs, missing_debug_implementations)]
 
 //! Utilities for tracking time.
 //!

tokio-timer/src/throttle.rs

@@ -158,7 +158,7 @@ impl<T: StdError + 'static> StdError for ThrottleError<T> {
     // FIXME(taiki-e): When the minimum support version of tokio reaches Rust 1.30,
     // replace this with Error::source.
     #[allow(deprecated)]
-    fn cause(&self) -> Option<&StdError> {
+    fn cause(&self) -> Option<&dyn StdError> {
         match self.0 {
             Either::A(ref err) => Some(err),
             Either::B(ref err) => Some(err),

tokio-timer/src/timer/handle.rs

@@ -44,6 +44,12 @@ pub(crate) struct HandlePriv {
     inner: Weak<Inner>,
 }
 
+/// A guard that resets the current timer to `None` when dropped.
+#[derive(Debug)]
+pub struct DefaultGuard {
+    _p: (),
+}
+
 thread_local! {
     /// Tracks the timer for the current execution context.
     static CURRENT_TIMER: RefCell<Option<HandlePriv>> = RefCell::new(None)
@@ -64,42 +70,32 @@ pub fn with_default<F, R>(handle: &Handle, enter: &mut Enter, f: F) -> R
 where
     F: FnOnce(&mut Enter) -> R,
 {
-    // Ensure that the timer 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_TIMER.with(|current| {
-                let mut current = current.borrow_mut();
-                *current = None;
-            });
-        }
-    }
-
-    // This ensures the value for the current timer gets reset even if there is
-    // a panic.
-    let _r = Reset;
+    let _guard = set_default(handle);
+    f(enter)
+}
 
+/// Sets `handle` as the default timer, returning a guard that unsets it on drop.
+///
+/// # Panics
+///
+/// This function panics if there already is a default timer set.
+pub fn set_default(handle: &Handle) -> DefaultGuard {
     CURRENT_TIMER.with(|current| {
-        {
-            let mut current = current.borrow_mut();
+        let mut current = current.borrow_mut();
 
-            assert!(
-                current.is_none(),
-                "default Tokio timer already set \
-                 for execution context"
-            );
+        assert!(
+            current.is_none(),
+            "default Tokio timer already set \
+             for execution context"
+        );
 
-            let handle = handle
-                .as_priv()
-                .unwrap_or_else(|| panic!("`handle` does not reference a timer"));
+        let handle = handle
+            .as_priv()
+            .unwrap_or_else(|| panic!("`handle` does not reference a timer"));
 
-            *current = Some(handle.clone());
-        }
-
-        f(enter)
-    })
+        *current = Some(handle.clone());
+    });
+    DefaultGuard { _p: () }
 }
 
 impl Handle {
@@ -194,3 +190,12 @@ impl fmt::Debug for HandlePriv {
         write!(f, "HandlePriv")
     }
 }
+
+impl Drop for DefaultGuard {
+    fn drop(&mut self) {
+        let _ = CURRENT_TIMER.try_with(|current| {
+            let mut current = current.borrow_mut();
+            *current = None;
+        });
+    }
+}

tokio-timer/src/timer/mod.rs

@@ -45,7 +45,7 @@ use self::entry::Entry;
 use self::stack::Stack;
 
 pub(crate) use self::handle::HandlePriv;
-pub use self::handle::{with_default, Handle};
+pub use self::handle::{set_default, with_default, DefaultGuard, Handle};
 pub use self::now::{Now, SystemNow};
 pub(crate) use self::registration::Registration;
 
@@ -162,7 +162,7 @@ pub(crate) struct Inner {
     process: AtomicStack,
 
     /// Unparks the timer thread.
-    unpark: Box<Unpark>,
+    unpark: Box<dyn Unpark>,
 }
 
 /// Maximum number of timeouts the system can handle concurrently.
@@ -426,7 +426,7 @@ impl<T, N> Drop for Timer<T, N> {
 // ===== impl Inner =====
 
 impl Inner {
-    fn new(start: Instant, unpark: Box<Unpark>) -> Inner {
+    fn new(start: Instant, unpark: Box<dyn Unpark>) -> Inner {
         Inner {
             num: AtomicUsize::new(0),
             elapsed: AtomicU64::new(0),

tokio-tls/Cargo.toml

@@ -31,7 +31,7 @@ tokio-io = "0.1.7"
 [dev-dependencies]
 tokio = "0.1"
 cfg-if = "0.1"
-env_logger = { version = "0.5", default-features = false }
+env_logger = { version = "0.6", default-features = false }
 
 [target.'cfg(all(not(target_os = "macos"), not(windows), not(target_os = "ios")))'.dev-dependencies]
 openssl = "0.10"

tokio-tls/examples/download-rust-lang.rs

@@ -12,7 +12,7 @@ use native_tls::TlsConnector;
 use tokio::net::TcpStream;
 use tokio::runtime::Runtime;
 
-fn main() -> Result<(), Box<std::error::Error>> {
+fn main() -> Result<(), Box<dyn std::error::Error>> {
     let mut runtime = Runtime::new()?;
     let addr = "www.rust-lang.org:443"
         .to_socket_addrs()?

tokio-tls/examples/echo.rs

@@ -8,7 +8,7 @@ use tokio::io;
 use tokio::net::TcpListener;
 use tokio::prelude::*;
 
-fn main() -> Result<(), Box<std::error::Error>> {
+fn main() -> Result<(), Box<dyn std::error::Error>> {
     // Bind the server's socket
     let addr = "127.0.0.1:12345".parse()?;
     let tcp = TcpListener::bind(&addr)?;