From bd3f3270dbf632a7e23e82dd93605db1fe050ee9 Mon Sep 17 00:00:00 2001 From: Carl Lerche Date: Wed, 10 Jul 2019 11:21:06 -0700 Subject: [PATCH] tokio: update threaded runtime to std::future (#1280) re-enables the threaded runtime and sets it (again) as the default. --- ci/azure-test-stable.yml | 8 +- tokio-signal/examples/ctrl-c.rs | 5 +- tokio-signal/examples/multiple.rs | 21 +- tokio-signal/examples/sighup-example.rs | 16 +- tokio-threadpool/src/thread_pool.rs | 6 +- tokio/Cargo.toml | 5 +- tokio/examples/echo-udp.rs | 11 +- tokio/examples/hello_world.rs | 8 +- tokio/examples/udp-client.rs | 21 +- tokio/src/lib.rs | 2 +- tokio/src/runtime/current_thread/mod.rs | 31 - tokio/src/runtime/mod.rs | 7 +- tokio/src/runtime/threadpool/builder.rs | 24 +- tokio/src/runtime/threadpool/mod.rs | 177 +----- tokio/src/runtime/threadpool/shutdown.rs | 34 -- tokio/src/runtime/threadpool/task_executor.rs | 30 +- tokio/tests/runtime.rs | 537 ------------------ tokio/tests/runtime_current_thread.rs | 139 +++++ tokio/tests/runtime_threaded.rs | 194 +++++++ tokio/tests/timer.rs | 6 +- 20 files changed, 418 insertions(+), 864 deletions(-) delete mode 100644 tokio/src/runtime/threadpool/shutdown.rs delete mode 100644 tokio/tests/runtime.rs create mode 100644 tokio/tests/runtime_current_thread.rs create mode 100644 tokio/tests/runtime_threaded.rs diff --git a/ci/azure-test-stable.yml b/ci/azure-test-stable.yml index 996d202ba..67ff590a3 100644 --- a/ci/azure-test-stable.yml +++ b/ci/azure-test-stable.yml @@ -39,10 +39,10 @@ jobs: env: LOOM_MAX_DURATION: 10 CI: 'True' - displayName: cargo test --tests + displayName: ${{ crate.key }} - cargo test --tests workingDirectory: $(Build.SourcesDirectory)/${{ crate.key }} - script: cargo test --examples - displayName: cargo test --examples + displayName: ${{ crate.key }} - cargo test --examples workingDirectory: $(Build.SourcesDirectory)/${{ crate.key }} # Run with each specified feature @@ -51,9 +51,9 @@ jobs: env: LOOM_MAX_DURATION: 10 CI: 'True' - displayName: cargo test --tests --features ${{ feature }} + displayName: ${{ crate.key }} - cargo test --tests --features ${{ feature }} workingDirectory: $(Build.SourcesDirectory)/${{ crate.key }} - script: cargo test --examples --no-default-features --features ${{ feature }} - displayName: cargo test --examples --features ${{ feature }} + displayName: ${{ crate.key }} - cargo test --examples --features ${{ feature }} workingDirectory: $(Build.SourcesDirectory)/${{ crate.key }} diff --git a/tokio-signal/examples/ctrl-c.rs b/tokio-signal/examples/ctrl-c.rs index 0d3b6d21a..dcce6696c 100644 --- a/tokio-signal/examples/ctrl-c.rs +++ b/tokio-signal/examples/ctrl-c.rs @@ -7,13 +7,13 @@ use futures_util::stream::StreamExt; const STOP_AFTER: u64 = 10; #[tokio::main] -async fn main() -> Result<(), Box> { +async fn main() { // tokio_signal provides a convenience builder for Ctrl+C // this even works cross-platform: linux and windows! // // `CtrlC::new()` produces a `Future` of the actual stream-initialisation // so first we await until the signal is ready. - let endless_stream = tokio_signal::CtrlC::new().await?; + let endless_stream = tokio_signal::CtrlC::new().await.unwrap(); // don't keep going forever: convert the endless stream to a bounded one. let mut limited_stream = endless_stream.take(STOP_AFTER); @@ -42,5 +42,4 @@ async fn main() -> Result<(), Box> { } println!("Stream ended, quiting the program."); - Ok(()) } diff --git a/tokio-signal/examples/multiple.rs b/tokio-signal/examples/multiple.rs index 6a640a84e..62d4b8551 100644 --- a/tokio-signal/examples/multiple.rs +++ b/tokio-signal/examples/multiple.rs @@ -3,20 +3,16 @@ //! A small example of how to listen for two signals at the same time -use std::error::Error; - // A trick to not fail build on non-unix platforms when using unix-specific features. #[cfg(unix)] mod platform { - use futures_util::stream::{self, StreamExt}; - use std::error::Error; use tokio_signal::unix::{Signal, SIGINT, SIGTERM}; - pub async fn main() -> Result<(), Box> { + pub async fn main() { // Create a stream for each of the signals we'd like to handle. - let sigint = Signal::new(SIGINT).await?; - let sigterm = Signal::new(SIGTERM).await?; + let sigint = Signal::new(SIGINT).await.unwrap(); + let sigterm = Signal::new(SIGTERM).await.unwrap(); // Use the `select` combinator to merge these two streams into one let stream = stream::select(sigint, sigterm); @@ -31,28 +27,23 @@ mod platform { let (item, _rest) = stream.into_future().await; // Figure out which signal we received - let item = item.ok_or("received no signal")?; + let item = item.ok_or("received no signal").unwrap(); if item == SIGINT { println!("received SIGINT"); } else { assert_eq!(item, SIGTERM); println!("received SIGTERM"); } - - Ok(()) } } #[cfg(not(unix))] mod platform { - use std::error::Error; - pub async fn main() -> Result<(), Box> { - Ok(()) - } + pub async fn main() {} } #[tokio::main] -async fn main() -> Result<(), Box> { +async fn main() { platform::main().await } diff --git a/tokio-signal/examples/sighup-example.rs b/tokio-signal/examples/sighup-example.rs index cd5685be5..08972b9e3 100644 --- a/tokio-signal/examples/sighup-example.rs +++ b/tokio-signal/examples/sighup-example.rs @@ -1,18 +1,15 @@ #![deny(warnings, rust_2018_idioms)] #![feature(async_await)] -use std::error::Error; - // A trick to not fail build on non-unix platforms when using unix-specific features. #[cfg(unix)] mod platform { use futures_util::stream::StreamExt; - use std::error::Error; use tokio_signal::unix::{Signal, SIGHUP}; - pub async fn main() -> Result<(), Box> { + pub async fn main() { // on Unix, we can listen to whatever signal we want, in this case: SIGHUP - let mut stream = Signal::new(SIGHUP).await?; + let mut stream = Signal::new(SIGHUP).await.unwrap(); println!("Waiting for SIGHUPS (Ctrl+C to quit)"); println!( @@ -30,20 +27,15 @@ mod platform { the_signal ); } - - Ok(()) } } #[cfg(not(unix))] mod platform { - use std::error::Error; - pub async fn main() -> Result<(), Box> { - Ok(()) - } + pub async fn main() {} } #[tokio::main] -async fn main() -> Result<(), Box> { +async fn main() { platform::main().await } diff --git a/tokio-threadpool/src/thread_pool.rs b/tokio-threadpool/src/thread_pool.rs index afb9a049e..11da99f9a 100644 --- a/tokio-threadpool/src/thread_pool.rs +++ b/tokio-threadpool/src/thread_pool.rs @@ -188,7 +188,11 @@ impl Drop for ThreadPool { drop(inner); // Wait until all worker threads terminate and the threadpool's resources clean up. - let mut enter = tokio_executor::enter().unwrap(); + let mut enter = match tokio_executor::enter() { + Ok(e) => e, + Err(_) => return, + }; + enter.block_on(shutdown); } } diff --git a/tokio/Cargo.toml b/tokio/Cargo.toml index 1f8823601..2f11aa50a 100644 --- a/tokio/Cargo.toml +++ b/tokio/Cargo.toml @@ -45,11 +45,12 @@ reactor = ["io", "tokio-reactor"] rt-full = [ "num_cpus", "reactor", + "sync", "timer", "tokio-current-thread", "tokio-executor", "tokio-macros", -# "tokio-threadpool", + "tokio-threadpool", "tracing-core", ] sync = ["tokio-sync"] @@ -73,7 +74,7 @@ tokio-executor = { version = "0.2.0", optional = true, path = "../tokio-executor tokio-macros = { version = "0.2.0", optional = true, path = "../tokio-macros" } tokio-reactor = { version = "0.2.0", optional = true, path = "../tokio-reactor" } tokio-sync = { version = "0.2.0", optional = true, path = "../tokio-sync", features = ["async-traits"] } -#tokio-threadpool = { version = "0.2.0", optional = true, path = "../tokio-threadpool" } +tokio-threadpool = { version = "0.2.0", optional = true, path = "../tokio-threadpool" } tokio-tcp = { version = "0.2.0", optional = true, path = "../tokio-tcp" } tokio-udp = { version = "0.2.0", optional = true, path = "../tokio-udp" } tokio-timer = { version = "0.3.0", optional = true, path = "../tokio-timer" } diff --git a/tokio/examples/echo-udp.rs b/tokio/examples/echo-udp.rs index 562f8b430..54bddb30e 100644 --- a/tokio/examples/echo-udp.rs +++ b/tokio/examples/echo-udp.rs @@ -50,12 +50,12 @@ impl Server { } #[tokio::main] -async fn main() -> Result<(), Box> { +async fn main() { let addr = env::args().nth(1).unwrap_or("127.0.0.1:8080".to_string()); - let addr = addr.parse::()?; + let addr = addr.parse::().unwrap(); - let socket = UdpSocket::bind(&addr)?; - println!("Listening on: {}", socket.local_addr()?); + let socket = UdpSocket::bind(&addr).unwrap(); + println!("Listening on: {}", socket.local_addr().unwrap()); let server = Server { socket: socket, @@ -64,6 +64,5 @@ async fn main() -> Result<(), Box> { }; // This starts the server task. - server.run().await?; - Ok(()) + server.run().await.unwrap(); } diff --git a/tokio/examples/hello_world.rs b/tokio/examples/hello_world.rs index 0aa479226..67afa9d1e 100644 --- a/tokio/examples/hello_world.rs +++ b/tokio/examples/hello_world.rs @@ -19,16 +19,14 @@ use tokio::io::AsyncWriteExt; use tokio::net::TcpStream; #[tokio::main] -pub async fn main() -> Result<(), Box> { - let addr = "127.0.0.1:6142".parse()?; +pub async fn main() { + let addr = "127.0.0.1:6142".parse().unwrap(); // Open a TCP stream to the socket address. // // Note that this is the Tokio TcpStream, which is fully async. - let mut stream = TcpStream::connect(&addr).await?; + let mut stream = TcpStream::connect(&addr).await.unwrap(); println!("created stream"); let result = stream.write(b"hello world\n").await; println!("wrote to stream; success={:?}", result.is_ok()); - - Ok(()) } diff --git a/tokio/examples/udp-client.rs b/tokio/examples/udp-client.rs index cc8a81aca..b26c0a37a 100644 --- a/tokio/examples/udp-client.rs +++ b/tokio/examples/udp-client.rs @@ -41,29 +41,32 @@ fn get_stdin_data() -> Result, Box> { } #[tokio::main] -async fn main() -> Result<(), Box> { +async fn main() { let remote_addr: SocketAddr = env::args() .nth(1) .unwrap_or("127.0.0.1:8080".into()) - .parse()?; + .parse() + .unwrap(); + // We use port 0 to let the operating system allocate an available port for us. let local_addr: SocketAddr = if remote_addr.is_ipv4() { "0.0.0.0:0" } else { "[::]:0" } - .parse()?; - let mut socket = UdpSocket::bind(&local_addr)?; + .parse() + .unwrap(); + + let mut socket = UdpSocket::bind(&local_addr).unwrap(); const MAX_DATAGRAM_SIZE: usize = 65_507; - socket.connect(&remote_addr)?; - let data = get_stdin_data()?; - socket.send(&data).await?; + socket.connect(&remote_addr).unwrap(); + let data = get_stdin_data().unwrap(); + socket.send(&data).await.unwrap(); let mut data = vec![0u8; MAX_DATAGRAM_SIZE]; - let len = socket.recv(&mut data).await?; + let len = socket.recv(&mut data).await.unwrap(); println!( "Received {} bytes:\n{}", len, String::from_utf8_lossy(&data[..len]) ); - Ok(()) } diff --git a/tokio/src/lib.rs b/tokio/src/lib.rs index 480b853cc..1e30d756f 100644 --- a/tokio/src/lib.rs +++ b/tokio/src/lib.rs @@ -2,6 +2,7 @@ #![deny(missing_docs, missing_debug_implementations, rust_2018_idioms)] #![cfg_attr(test, deny(warnings))] #![doc(test(no_crate_inject, attr(deny(rust_2018_idioms))))] +#![feature(async_await)] //! A runtime for writing reliable, asynchronous, and slim applications. //! @@ -107,7 +108,6 @@ if_runtime! { pub mod runtime; pub use crate::executor::spawn; - pub use crate::runtime::run; #[cfg(not(test))] // Work around for rust-lang/rust#62127 pub use tokio_macros::main; diff --git a/tokio/src/runtime/current_thread/mod.rs b/tokio/src/runtime/current_thread/mod.rs index 971922a72..d81837eee 100644 --- a/tokio/src/runtime/current_thread/mod.rs +++ b/tokio/src/runtime/current_thread/mod.rs @@ -71,34 +71,3 @@ pub use self::builder::Builder; pub use self::runtime::{Handle, Runtime}; pub use tokio_current_thread::spawn; pub use tokio_current_thread::TaskExecutor; - -use std::future::Future; - -/// Run the provided future to completion using a runtime running on the current thread. -/// -/// This first creates a new [`Runtime`], and calls [`Runtime::block_on`] with the provided future, -/// which blocks the current thread until the provided future completes. It then calls -/// [`Runtime::run`] to wait for any other spawned futures to resolve. -pub fn block_on_all(future: F) -> F::Output -where - F: Future, -{ - let mut r = Runtime::new().expect("failed to start runtime on current thread"); - let v = r.block_on(future); - r.run().expect("failed to resolve remaining futures"); - v -} - -/// Start a current-thread runtime using the supplied future to bootstrap execution. -/// -/// # Panics -/// -/// This function panics if called from the context of an executor. -pub fn run(future: F) -where - F: Future + 'static, -{ - let mut r = Runtime::new().expect("failed to start runtime on current thread"); - r.spawn(future); - r.run().expect("failed to resolve remaining futures"); -} diff --git a/tokio/src/runtime/mod.rs b/tokio/src/runtime/mod.rs index 34df32230..84302a0f4 100644 --- a/tokio/src/runtime/mod.rs +++ b/tokio/src/runtime/mod.rs @@ -109,15 +109,10 @@ //! [`Timer`]: https://docs.rs/tokio-timer/0.2/tokio_timer/timer/struct.Timer.html pub mod current_thread; -//mod threadpool; +mod threadpool; -pub use self::current_thread::{run, Builder, Runtime}; -/* pub use self::threadpool::{ Builder, Runtime, - Shutdown, TaskExecutor, - run, }; -*/ diff --git a/tokio/src/runtime/threadpool/builder.rs b/tokio/src/runtime/threadpool/builder.rs index f8c1c4856..7b35078bf 100644 --- a/tokio/src/runtime/threadpool/builder.rs +++ b/tokio/src/runtime/threadpool/builder.rs @@ -84,18 +84,6 @@ impl Builder { self } - /// Set builder to set up the thread pool instance. - #[deprecated( - since = "0.1.9", - note = "use the `core_threads`, `blocking_threads`, `name_prefix`, \ - `keep_alive`, and `stack_size` functions on `runtime::Builder`, \ - instead")] - #[doc(hidden)] - pub fn threadpool_builder(&mut self, val: ThreadPoolBuilder) -> &mut Self { - self.threadpool_builder = val; - self - } - /// Sets a callback to handle panics in futures. /// /// The callback is triggered when a panic during a future bubbles up to @@ -351,9 +339,9 @@ impl Builder { .around_worker(move |w, enter| { let index = w.id().to_usize(); - tokio_reactor::with_default(&reactor_handles[index], enter, |enter| { - clock::with_default(&clock, enter, |enter| { - timer::with_default(&timer_handles[index], enter, |_| { + tokio_reactor::with_default(&reactor_handles[index], enter, |_| { + clock::with_default(&clock, || { + timer::with_default(&timer_handles[index], || { trace::dispatcher::with_default(&dispatch, || { w.run(); }) @@ -372,14 +360,8 @@ impl Builder { }) .build(); - // To support deprecated `reactor()` function - let reactor = Reactor::new()?; - let reactor_handle = reactor.handle(); - Ok(Runtime { inner: Some(Inner { - reactor_handle, - reactor: Mutex::new(Some(reactor)), pool, }), }) diff --git a/tokio/src/runtime/threadpool/mod.rs b/tokio/src/runtime/threadpool/mod.rs index 9761c1163..e69bd0817 100644 --- a/tokio/src/runtime/threadpool/mod.rs +++ b/tokio/src/runtime/threadpool/mod.rs @@ -1,18 +1,13 @@ mod builder; -mod shutdown; mod task_executor; pub use self::builder::Builder; -pub use self::shutdown::Shutdown; pub use self::task_executor::TaskExecutor; -use crate::reactor::{Handle, Reactor}; -use futures; -use futures::future::Future; use tokio_executor::enter; -use tokio_threadpool as threadpool; + +use std::future::Future; use std::io; -use std::sync::Mutex; /// Handle to the Tokio runtime. /// @@ -35,73 +30,12 @@ pub struct Runtime { #[derive(Debug)] struct Inner { - /// A handle to the reactor in the background thread. - reactor_handle: Handle, - - // TODO: This should go away in 0.2 - reactor: Mutex>, - /// Task execution pool. - pool: threadpool::ThreadPool, + pool: tokio_threadpool::ThreadPool, } // ===== impl Runtime ===== -/// Start the Tokio runtime using the supplied future to bootstrap execution. -/// -/// This function is used to bootstrap the execution of a Tokio application. It -/// does the following: -/// -/// * Start the Tokio runtime using a default configuration. -/// * Spawn the given future onto the thread pool. -/// * Block the current thread until the runtime shuts down. -/// -/// Note that the function will not return immediately once `future` has -/// completed. Instead it waits for the entire runtime to become idle. -/// -/// See the [module level][mod] documentation for more details. -/// -/// # Examples -/// -/// ```rust -/// # use futures::{Future, Stream}; -/// use tokio::net::TcpListener; -/// -/// # fn process(_: T) -> Box + Send> { -/// # unimplemented!(); -/// # } -/// # fn dox() { -/// # let addr = "127.0.0.1:8080".parse().unwrap(); -/// let listener = TcpListener::bind(&addr).unwrap(); -/// -/// let server = listener.incoming() -/// .map_err(|e| println!("error = {:?}", e)) -/// .for_each(|socket| { -/// tokio::spawn(process(socket)) -/// }); -/// -/// tokio::run(server); -/// # } -/// # pub fn main() {} -/// ``` -/// -/// # Panics -/// -/// This function panics if called from the context of an executor. -/// -/// [mod]: ../index.html -pub fn run(future: F) -where F: Future + Send + 'static, -{ - // Check enter before creating a new Runtime... - let mut entered = enter().expect("nested tokio::run"); - let runtime = Runtime::new().expect("failed to start new Runtime"); - runtime.spawn(future); - entered - .block_on(runtime.shutdown_on_idle()) - .expect("shutdown cannot error") -} - impl Runtime { /// Create a new runtime instance with default configuration values. /// @@ -137,43 +71,6 @@ impl Runtime { Builder::new().build() } - #[deprecated(since = "0.1.5", note = "use `reactor` instead")] - #[doc(hidden)] - pub fn handle(&self) -> &Handle { - #[allow(deprecated)] - self.reactor() - } - - /// Return a reference to the reactor handle for this runtime instance. - /// - /// The returned handle reference can be cloned in order to get an owned - /// value of the handle. This handle can be used to initialize I/O resources - /// (like TCP or UDP sockets) that will not be used on the runtime. - /// - /// # Examples - /// - /// ``` - /// use tokio::runtime::Runtime; - /// - /// let rt = Runtime::new() - /// .unwrap(); - /// - /// let reactor_handle = rt.reactor().clone(); - /// - /// // use `reactor_handle` - /// ``` - #[deprecated(since = "0.1.11", note = "there is now a reactor per worker thread")] - pub fn reactor(&self) -> &Handle { - let mut reactor = self.inner().reactor.lock().unwrap(); - if let Some(reactor) = reactor.take() { - if let Ok(background) = reactor.background() { - background.forget(); - } - } - - &self.inner().reactor_handle - } - /// Return a handle to the runtime's executor. /// /// The returned handle can be used to spawn tasks that run on this runtime. @@ -229,7 +126,7 @@ impl Runtime { /// This function panics if the spawn fails. Failure occurs if the executor /// is currently at capacity and is unable to spawn a new future. pub fn spawn(&self, future: F) -> &Self - where F: Future + Send + 'static, + where F: Future + Send + 'static, { self.inner().pool.spawn(future); self @@ -247,48 +144,20 @@ impl Runtime { /// /// This function panics if the executor is at capacity, if the provided /// future panics, or if called within an asynchronous execution context. - pub fn block_on(&self, future: F) -> Result + pub fn block_on(&self, future: F) -> F::Output where - F: Send + 'static + Future, - R: Send + 'static, - E: Send + 'static, + F: Send + 'static + Future, + F::Output: Send + 'static, { let mut entered = enter().expect("nested block_on"); - let (tx, rx) = futures::sync::oneshot::channel(); - self.spawn(future.then(move |r| tx.send(r).map_err(|_| unreachable!()))); - entered.block_on(rx).expect("blocked on future paniced") - } + let (tx, rx) = crate::sync::oneshot::channel(); - /// Run a future to completion on the Tokio runtime, then wait for all - /// background futures to complete too. - /// - /// This runs the given future on the runtime, blocking until it is - /// complete, waiting for background futures to complete, and yielding - /// its resolved result. Any tasks or timers which the future spawns - /// internally will be executed on the runtime and waited for completion. - /// - /// This method should not be called from an asynchronous context. - /// - /// # Panics - /// - /// This function panics if the executor is at capacity, if the provided - /// future panics, or if called within an asynchronous execution context. - pub fn block_on_all(self, future: F) -> Result - where - F: Send + 'static + Future, - R: Send + 'static, - E: Send + 'static, - { - let mut entered = enter().expect("nested block_on_all"); - let (tx, rx) = futures::sync::oneshot::channel(); - self.spawn(future.then(move |r| tx.send(r).map_err(|_| unreachable!()))); - let block = rx - .map_err(|_| panic!("blocked on future paniced")) - .and_then(move |r| { - self.shutdown_on_idle() - .map(move |()| r) - }); - entered.block_on(block).unwrap() + self.spawn(async move { + let res = future.await; + let _ = tx.send(res); + }); + + entered.block_on(rx).expect("blocked on future paniced") } /// Signals the runtime to shutdown once it becomes idle. @@ -323,10 +192,11 @@ impl Runtime { /// ``` /// /// [mod]: index.html - pub fn shutdown_on_idle(mut self) -> Shutdown { + pub async fn shutdown_on_idle(mut self) { let inner = self.inner.take().unwrap(); let inner = inner.pool.shutdown_on_idle(); - Shutdown { inner } + + inner.await; } /// Signals the runtime to shutdown immediately. @@ -364,21 +234,12 @@ impl Runtime { /// ``` /// /// [mod]: index.html - pub fn shutdown_now(mut self) -> Shutdown { + pub async fn shutdown_now(mut self) { let inner = self.inner.take().unwrap(); - Shutdown::shutdown_now(inner) + inner.pool.shutdown_now().await; } fn inner(&self) -> &Inner { self.inner.as_ref().unwrap() } } - -impl Drop for Runtime { - fn drop(&mut self) { - if let Some(inner) = self.inner.take() { - let shutdown = Shutdown::shutdown_now(inner); - let _ = shutdown.wait(); - } - } -} diff --git a/tokio/src/runtime/threadpool/shutdown.rs b/tokio/src/runtime/threadpool/shutdown.rs deleted file mode 100644 index a1b3e7637..000000000 --- a/tokio/src/runtime/threadpool/shutdown.rs +++ /dev/null @@ -1,34 +0,0 @@ -use futures::{try_ready, Future, Poll}; -use tokio_threadpool as threadpool; -use std::fmt; -use super::Inner; - -/// A future that resolves when the Tokio `Runtime` is shut down. -pub struct Shutdown { - pub(super) inner: threadpool::Shutdown, -} - -impl Shutdown { - pub(super) fn shutdown_now(inner: Inner) -> Self { - let inner = inner.pool.shutdown_now(); - Shutdown { inner } - } -} - -impl Future for Shutdown { - type Item = (); - type Error = (); - - fn poll(&mut self) -> Poll<(), ()> { - try_ready!(self.inner.poll()); - Ok(().into()) - } -} - -impl fmt::Debug for Shutdown { - fn fmt(&self, fmt: &mut fmt::Formatter<'_>) -> fmt::Result { - fmt.debug_struct("Shutdown") - .field("inner", &"Box>") - .finish() - } -} diff --git a/tokio/src/runtime/threadpool/task_executor.rs b/tokio/src/runtime/threadpool/task_executor.rs index 8c0c6aa09..f4dfce6be 100644 --- a/tokio/src/runtime/threadpool/task_executor.rs +++ b/tokio/src/runtime/threadpool/task_executor.rs @@ -1,6 +1,9 @@ -use futures::future::{self, Future}; +use tokio_executor::SpawnError; use tokio_threadpool::Sender; +use std::future::Future; +use std::pin::Pin; + /// Executes futures on the runtime /// /// All futures spawned using this executor will be submitted to the associated @@ -48,33 +51,26 @@ impl TaskExecutor { /// This function panics if the spawn fails. Failure occurs if the executor /// is currently at capacity and is unable to spawn a new future. pub fn spawn(&self, future: F) - where F: Future + Send + 'static, + where F: Future + Send + 'static, { self.inner.spawn(future).unwrap(); } } -impl future::Executor for TaskExecutor -where T: Future + Send + 'static, -{ - fn execute(&self, future: T) -> Result<(), future::ExecuteError> { - self.inner.execute(future) - } -} - -impl crate::executor::Executor for TaskExecutor { - fn spawn(&mut self, future: Box + Send>) - -> Result<(), crate::executor::SpawnError> - { +impl tokio_executor::Executor for TaskExecutor { + fn spawn( + &mut self, + future: Pin + Send>>, + ) -> Result<(), SpawnError> { self.inner.spawn(future) } } -impl crate::executor::TypedExecutor for TaskExecutor +impl tokio_executor::TypedExecutor for TaskExecutor where - T: Future + Send + 'static, + T: Future + Send + 'static, { fn spawn(&mut self, future: T) -> Result<(), crate::executor::SpawnError> { - crate::executor::Executor::spawn(self, Box::new(future)) + crate::executor::Executor::spawn(self, Box::pin(future)) } } diff --git a/tokio/tests/runtime.rs b/tokio/tests/runtime.rs deleted file mode 100644 index 780c3c115..000000000 --- a/tokio/tests/runtime.rs +++ /dev/null @@ -1,537 +0,0 @@ -#![cfg(feature = "broken")] -#![deny(warnings, rust_2018_idioms)] - -use env_logger; -use futures; -use futures::sync::oneshot; -use std::sync::{atomic, Arc, Mutex}; -use std::thread; -use tokio; -use tokio::io; -use tokio::net::{TcpListener, TcpStream}; -use tokio::prelude::future::lazy; -use tokio::prelude::*; -use tokio::runtime::Runtime; - -// this import is used in all child modules that have it in scope -// from importing super::*, but the compiler doesn't realise that -// and warns about it. -pub use futures::future::Executor; - -macro_rules! t { - ($e:expr) => { - match $e { - Ok(e) => e, - Err(e) => panic!("{} failed with {:?}", stringify!($e), e), - } - }; -} - -fn create_client_server_future() -> Box + Send> { - let server = t!(TcpListener::bind(&"127.0.0.1:0".parse().unwrap())); - let addr = t!(server.local_addr()); - let client = TcpStream::connect(&addr); - - let server = server - .incoming() - .take(1) - .map_err(|e| panic!("accept err = {:?}", e)) - .for_each(|socket| { - tokio::spawn({ - io::write_all(socket, b"hello") - .map(|_| ()) - .map_err(|e| panic!("write err = {:?}", e)) - }) - }) - .map(|_| ()); - - let client = client - .map_err(|e| panic!("connect err = {:?}", e)) - .and_then(|client| { - // Read all - io::read_to_end(client, vec![]) - .map(|_| ()) - .map_err(|e| panic!("read err = {:?}", e)) - }); - - let future = server.join(client).map(|_| ()); - Box::new(future) -} - -#[test] -fn runtime_tokio_run() { - let _ = env_logger::try_init(); - - tokio::run(create_client_server_future()); -} - -#[test] -fn runtime_single_threaded() { - let _ = env_logger::try_init(); - - let mut runtime = tokio::runtime::current_thread::Runtime::new().unwrap(); - runtime.block_on(create_client_server_future()).unwrap(); - runtime.run().unwrap(); -} - -#[test] -fn runtime_single_threaded_block_on() { - let _ = env_logger::try_init(); - - tokio::runtime::current_thread::block_on_all(create_client_server_future()).unwrap(); -} - -mod runtime_single_threaded_block_on_all { - use super::*; - - fn test(spawn: F) - where - F: Fn(Box + Send>), - { - let cnt = Arc::new(Mutex::new(0)); - let c = cnt.clone(); - - let msg = tokio::runtime::current_thread::block_on_all(lazy(move || { - { - let mut x = c.lock().unwrap(); - *x = 1 + *x; - } - - // Spawn! - spawn(Box::new(lazy(move || { - { - let mut x = c.lock().unwrap(); - *x = 1 + *x; - } - Ok::<(), ()>(()) - }))); - - Ok::<_, ()>("hello") - })) - .unwrap(); - - assert_eq!(2, *cnt.lock().unwrap()); - assert_eq!(msg, "hello"); - } - - #[test] - fn spawn() { - test(|f| { - tokio::spawn(f); - }) - } - - #[test] - fn execute() { - test(|f| { - tokio::executor::DefaultExecutor::current() - .execute(f) - .unwrap(); - }) - } -} - -mod runtime_single_threaded_racy { - use super::*; - fn test(spawn: F) - where - F: Fn( - tokio::runtime::current_thread::Handle, - Box + Send>, - ), - { - let (trigger, exit) = futures::sync::oneshot::channel(); - let (handle_tx, handle_rx) = ::std::sync::mpsc::channel(); - let jh = ::std::thread::spawn(move || { - let mut rt = tokio::runtime::current_thread::Runtime::new().unwrap(); - handle_tx.send(rt.handle()).unwrap(); - - // don't exit until we are told to - rt.block_on(exit.map_err(|_| ())).unwrap(); - - // run until all spawned futures (incl. the "exit" signal future) have completed. - rt.run().unwrap(); - }); - - let (tx, rx) = futures::sync::oneshot::channel(); - - let handle = handle_rx.recv().unwrap(); - spawn( - handle, - Box::new(futures::future::lazy(move || { - tx.send(()).unwrap(); - Ok(()) - })), - ); - - // signal runtime thread to exit - trigger.send(()).unwrap(); - - // wait for runtime thread to exit - jh.join().unwrap(); - - assert_eq!(rx.wait().unwrap(), ()); - } - - #[test] - fn spawn() { - test(|handle, f| { - handle.spawn(f).unwrap(); - }) - } - - #[test] - fn execute() { - test(|handle, f| { - handle.execute(f).unwrap(); - }) - } -} - -mod runtime_multi_threaded { - use super::*; - fn test(spawn: F) - where - F: Fn(&mut Runtime) + Send + 'static, - { - let _ = env_logger::try_init(); - - let mut runtime = tokio::runtime::Builder::new().build().unwrap(); - spawn(&mut runtime); - runtime.shutdown_on_idle().wait().unwrap(); - } - - #[test] - fn spawn() { - test(|rt| { - rt.spawn(create_client_server_future()); - }); - } - - #[test] - fn execute() { - test(|rt| { - rt.executor() - .execute(create_client_server_future()) - .unwrap(); - }); - } -} - -#[test] -fn block_on_timer() { - use std::time::{Duration, Instant}; - use tokio::timer::{Delay, Error}; - - fn after_1s(x: T) -> Box + Send> - where - T: Send + 'static, - { - Box::new(Delay::new(Instant::now() + Duration::from_millis(100)).map(move |_| x)) - } - - let runtime = Runtime::new().unwrap(); - assert_eq!(runtime.block_on(after_1s(42)).unwrap(), 42); - runtime.shutdown_on_idle().wait().unwrap(); -} - -mod from_block_on { - use super::*; - - fn test(spawn: F) - where - F: Fn(Box + Send>) + Send + 'static, - { - let cnt = Arc::new(Mutex::new(0)); - let c = cnt.clone(); - - let runtime = Runtime::new().unwrap(); - let msg = runtime - .block_on(lazy(move || { - { - let mut x = c.lock().unwrap(); - *x = 1 + *x; - } - - // Spawn! - spawn(Box::new(lazy(move || { - { - let mut x = c.lock().unwrap(); - *x = 1 + *x; - } - Ok::<(), ()>(()) - }))); - - Ok::<_, ()>("hello") - })) - .unwrap(); - - runtime.shutdown_on_idle().wait().unwrap(); - assert_eq!(2, *cnt.lock().unwrap()); - assert_eq!(msg, "hello"); - } - - #[test] - fn execute() { - test(|f| { - tokio::executor::DefaultExecutor::current() - .execute(f) - .unwrap(); - }) - } - - #[test] - fn spawn() { - test(|f| { - tokio::spawn(f); - }) - } -} - -#[test] -fn block_waits() { - let (tx, rx) = oneshot::channel(); - - thread::spawn(|| { - use std::time::Duration; - thread::sleep(Duration::from_millis(1000)); - tx.send(()).unwrap(); - }); - - let cnt = Arc::new(Mutex::new(0)); - let c = cnt.clone(); - - let runtime = Runtime::new().unwrap(); - runtime - .block_on(rx.then(move |_| { - { - let mut x = c.lock().unwrap(); - *x = 1 + *x; - } - Ok::<_, ()>(()) - })) - .unwrap(); - - assert_eq!(1, *cnt.lock().unwrap()); - runtime.shutdown_on_idle().wait().unwrap(); -} - -mod many { - use super::*; - - const ITER: usize = 200; - fn test(spawn: F) - where - F: Fn(&mut Runtime, Box + Send>), - { - let cnt = Arc::new(Mutex::new(0)); - let mut runtime = Runtime::new().unwrap(); - - for _ in 0..ITER { - let c = cnt.clone(); - spawn( - &mut runtime, - Box::new(lazy(move || { - { - let mut x = c.lock().unwrap(); - *x = 1 + *x; - } - Ok::<(), ()>(()) - })), - ); - } - - runtime.shutdown_on_idle().wait().unwrap(); - assert_eq!(ITER, *cnt.lock().unwrap()); - } - - #[test] - fn spawn() { - test(|rt, f| { - rt.spawn(f); - }) - } - - #[test] - fn execute() { - test(|rt, f| { - rt.executor().execute(f).unwrap(); - }) - } -} - -mod from_block_on_all { - use super::*; - - fn test(spawn: F) - where - F: Fn(Box + Send>) + Send + 'static, - { - let cnt = Arc::new(Mutex::new(0)); - let c = cnt.clone(); - - let runtime = Runtime::new().unwrap(); - let msg = runtime - .block_on_all(lazy(move || { - { - let mut x = c.lock().unwrap(); - *x = 1 + *x; - } - - // Spawn! - spawn(Box::new(lazy(move || { - { - let mut x = c.lock().unwrap(); - *x = 1 + *x; - } - Ok::<(), ()>(()) - }))); - - Ok::<_, ()>("hello") - })) - .unwrap(); - - assert_eq!(2, *cnt.lock().unwrap()); - assert_eq!(msg, "hello"); - } - - #[test] - fn execute() { - test(|f| { - tokio::executor::DefaultExecutor::current() - .execute(f) - .unwrap(); - }) - } - - #[test] - fn spawn() { - test(|f| { - tokio::spawn(f); - }) - } -} - -mod nested_enter { - use super::*; - use std::panic; - use tokio::runtime::current_thread; - - fn test(first: F1, nested: F2) - where - F1: Fn(Box + Send>) + Send + 'static, - F2: Fn(Box + Send>) + panic::UnwindSafe + Send + 'static, - { - let panicked = Arc::new(Mutex::new(false)); - let panicked2 = panicked.clone(); - - // Since this is testing panics in other threads, printing about panics - // is noisy and can give the impression that the test is ignoring panics. - // - // It *is* ignoring them, but on purpose. - let prev_hook = panic::take_hook(); - panic::set_hook(Box::new(|info| { - let s = info.to_string(); - if s.starts_with("panicked at 'nested ") - || s.starts_with("panicked at 'Multiple executors at once") - { - // expected, noop - } else { - println!("{}", s); - } - })); - - first(Box::new(lazy(move || { - panic::catch_unwind(move || nested(Box::new(lazy(|| Ok::<(), ()>(()))))) - .expect_err("nested should panic"); - *panicked2.lock().unwrap() = true; - Ok::<(), ()>(()) - }))); - - panic::set_hook(prev_hook); - - assert!( - *panicked.lock().unwrap(), - "nested call should have panicked" - ); - } - - fn threadpool_new() -> Runtime { - Runtime::new().expect("rt new") - } - - #[test] - fn run_in_run() { - test(tokio::run, tokio::run); - } - - #[test] - fn threadpool_block_on_in_run() { - test(tokio::run, |fut| { - let rt = threadpool_new(); - rt.block_on(fut).unwrap(); - }); - } - - #[test] - fn threadpool_block_on_all_in_run() { - test(tokio::run, |fut| { - let rt = threadpool_new(); - rt.block_on_all(fut).unwrap(); - }); - } - - #[test] - fn current_thread_block_on_all_in_run() { - test(tokio::run, |fut| { - current_thread::block_on_all(fut).unwrap(); - }); - } -} - -#[test] -fn runtime_reactor_handle() { - #![allow(deprecated)] - - use futures::Stream; - use std::net::{TcpListener as StdListener, TcpStream as StdStream}; - - let rt = Runtime::new().unwrap(); - - let std_listener = StdListener::bind("127.0.0.1:0").unwrap(); - let tk_listener = TcpListener::from_std(std_listener, rt.handle()).unwrap(); - - let addr = tk_listener.local_addr().unwrap(); - - // Spawn a thread since we are avoiding the runtime - let th = thread::spawn(|| for _ in tk_listener.incoming().take(1).wait() {}); - - let _ = StdStream::connect(&addr).unwrap(); - - th.join().unwrap(); -} - -#[test] -fn after_start_and_before_stop_is_called() { - let _ = env_logger::try_init(); - - let after_start = Arc::new(atomic::AtomicUsize::new(0)); - let before_stop = Arc::new(atomic::AtomicUsize::new(0)); - - let after_inner = after_start.clone(); - let before_inner = before_stop.clone(); - let runtime = tokio::runtime::Builder::new() - .after_start(move || { - after_inner.clone().fetch_add(1, atomic::Ordering::Relaxed); - }) - .before_stop(move || { - before_inner.clone().fetch_add(1, atomic::Ordering::Relaxed); - }) - .build() - .unwrap(); - - runtime.block_on_all(create_client_server_future()).unwrap(); - - assert!(after_start.load(atomic::Ordering::Relaxed) > 0); - assert!(before_stop.load(atomic::Ordering::Relaxed) > 0); -} diff --git a/tokio/tests/runtime_current_thread.rs b/tokio/tests/runtime_current_thread.rs new file mode 100644 index 000000000..d5852bf97 --- /dev/null +++ b/tokio/tests/runtime_current_thread.rs @@ -0,0 +1,139 @@ +#![deny(warnings, rust_2018_idioms)] +#![feature(async_await)] + +use tokio::io::{AsyncReadExt, AsyncWriteExt}; +use tokio::net::{TcpListener, TcpStream}; +use tokio::runtime::current_thread::Runtime; +use tokio::sync::oneshot; +use tokio::timer::Delay; +use tokio_test::{assert_err, assert_ok}; + +use env_logger; +use std::sync::mpsc; +use std::time::{Duration, Instant}; + +async fn client_server(tx: mpsc::Sender<()>) { + let addr = assert_ok!("127.0.0.1:0".parse()); + let mut server = assert_ok!(TcpListener::bind(&addr)); + + // Get the assigned address + let addr = assert_ok!(server.local_addr()); + + // Spawn the server + tokio::spawn(async move { + // Accept a socket + let (mut socket, _) = server.accept().await.unwrap(); + + // Write some data + socket.write_all(b"hello").await.unwrap(); + }); + + let mut client = TcpStream::connect(&addr).await.unwrap(); + + let mut buf = vec![]; + client.read_to_end(&mut buf).await.unwrap(); + + assert_eq!(buf, b"hello"); + tx.send(()).unwrap(); +} + +#[test] +fn spawn_run_spawn_root() { + let _ = env_logger::try_init(); + + let mut rt = Runtime::new().unwrap(); + let (tx, rx) = mpsc::channel(); + + let tx2 = tx.clone(); + rt.spawn(async move { + Delay::new(Instant::now() + Duration::from_millis(1000)).await; + tx2.send(()).unwrap(); + }); + + rt.spawn(client_server(tx)); + rt.run().unwrap(); + + assert_ok!(rx.try_recv()); + assert_ok!(rx.try_recv()); +} + +#[test] +fn spawn_run_nested_spawn() { + let _ = env_logger::try_init(); + + let mut rt = Runtime::new().unwrap(); + let (tx, rx) = mpsc::channel(); + + let tx2 = tx.clone(); + rt.spawn(async move { + tokio::spawn(async move { + Delay::new(Instant::now() + Duration::from_millis(1000)).await; + tx2.send(()).unwrap(); + }); + }); + + rt.spawn(client_server(tx)); + rt.run().unwrap(); + + assert_ok!(rx.try_recv()); + assert_ok!(rx.try_recv()); +} + +#[test] +fn block_on() { + let _ = env_logger::try_init(); + + let mut rt = Runtime::new().unwrap(); + let (tx, rx) = mpsc::channel(); + + let tx2 = tx.clone(); + rt.spawn(async move { + Delay::new(Instant::now() + Duration::from_millis(1000)).await; + tx2.send(()).unwrap(); + }); + + rt.block_on(client_server(tx)); + + assert_ok!(rx.try_recv()); + assert_err!(rx.try_recv()); +} + +#[test] +fn racy() { + use std::sync::mpsc; + use std::thread; + + let (trigger, exit) = oneshot::channel(); + let (handle_tx, handle_rx) = mpsc::channel(); + + let jh = thread::spawn(move || { + let mut rt = Runtime::new().unwrap(); + handle_tx.send(rt.handle()).unwrap(); + + // don't exit until we are told to + rt.block_on(async { + exit.await.unwrap(); + }); + + // run until all spawned futures (incl. the "exit" signal future) have completed. + rt.run().unwrap(); + }); + + let (tx, rx) = oneshot::channel(); + + let handle = handle_rx.recv().unwrap(); + handle + .spawn(async { + tx.send(()).unwrap(); + }) + .unwrap(); + + // signal runtime thread to exit + trigger.send(()).unwrap(); + + // wait for runtime thread to exit + jh.join().unwrap(); + + let mut e = tokio_executor::enter().unwrap(); + e.block_on(rx).unwrap(); +} diff --git a/tokio/tests/runtime_threaded.rs b/tokio/tests/runtime_threaded.rs new file mode 100644 index 000000000..ec8423d3b --- /dev/null +++ b/tokio/tests/runtime_threaded.rs @@ -0,0 +1,194 @@ +#![deny(warnings, rust_2018_idioms)] +#![feature(async_await)] + +use tokio; +use tokio::io::{AsyncReadExt, AsyncWriteExt}; +use tokio::net::{TcpListener, TcpStream}; +use tokio::runtime::Runtime; +use tokio::sync::oneshot; +use tokio::timer::Delay; +use tokio_test::{assert_err, assert_ok}; + +use env_logger; +use std::sync::{mpsc, Arc, Mutex}; +use std::thread; +use std::time::{Duration, Instant}; + +async fn client_server(tx: mpsc::Sender<()>) { + let addr = assert_ok!("127.0.0.1:0".parse()); + let mut server = assert_ok!(TcpListener::bind(&addr)); + + // Get the assigned address + let addr = assert_ok!(server.local_addr()); + + // Spawn the server + tokio::spawn(async move { + // Accept a socket + let (mut socket, _) = server.accept().await.unwrap(); + + // Write some data + socket.write_all(b"hello").await.unwrap(); + }); + + let mut client = TcpStream::connect(&addr).await.unwrap(); + + let mut buf = vec![]; + client.read_to_end(&mut buf).await.unwrap(); + + assert_eq!(buf, b"hello"); + tx.send(()).unwrap(); +} + +#[test] +fn spawn_shutdown() { + let _ = env_logger::try_init(); + + let rt = Runtime::new().unwrap(); + let (tx, rx) = mpsc::channel(); + + rt.spawn(client_server(tx.clone())); + + // Use executor trait + let f = Box::pin(client_server(tx)); + tokio_executor::Executor::spawn(&mut rt.executor(), f).unwrap(); + + let mut e = tokio_executor::enter().unwrap(); + e.block_on(rt.shutdown_on_idle()); + + assert_ok!(rx.try_recv()); + assert_ok!(rx.try_recv()); + assert_err!(rx.try_recv()); +} + +#[test] +fn block_on_timer() { + let rt = Runtime::new().unwrap(); + + let v = rt.block_on(async move { + let delay = Delay::new(Instant::now() + Duration::from_millis(100)); + delay.await; + 42 + }); + + assert_eq!(v, 42); + + let mut e = tokio_executor::enter().unwrap(); + e.block_on(rt.shutdown_on_idle()); +} + +#[test] +fn block_waits() { + let (a_tx, a_rx) = oneshot::channel(); + let (b_tx, b_rx) = mpsc::channel(); + + thread::spawn(|| { + use std::time::Duration; + + thread::sleep(Duration::from_millis(1000)); + a_tx.send(()).unwrap(); + }); + + let rt = Runtime::new().unwrap(); + rt.block_on(async move { + a_rx.await.unwrap(); + b_tx.send(()).unwrap(); + }); + + assert_ok!(b_rx.try_recv()); + + let mut e = tokio_executor::enter().unwrap(); + e.block_on(rt.shutdown_on_idle()); +} + +#[test] +fn spawn_many() { + const ITER: usize = 200; + + let cnt = Arc::new(Mutex::new(0)); + let rt = Runtime::new().unwrap(); + + let c = cnt.clone(); + rt.block_on(async move { + for _ in 0..ITER { + let c = c.clone(); + tokio::spawn(async move { + let mut x = c.lock().unwrap(); + *x = 1 + *x; + }); + } + }); + + let mut e = tokio_executor::enter().unwrap(); + e.block_on(rt.shutdown_on_idle()); + + assert_eq!(ITER, *cnt.lock().unwrap()); +} + +#[test] +fn nested_enter() { + use std::panic; + + let rt = Runtime::new().unwrap(); + rt.block_on(async { + assert_err!(tokio_executor::enter()); + + // Since this is testing panics in other threads, printing about panics + // is noisy and can give the impression that the test is ignoring panics. + // + // It *is* ignoring them, but on purpose. + let prev_hook = panic::take_hook(); + panic::set_hook(Box::new(|info| { + let s = info.to_string(); + if s.starts_with("panicked at 'nested ") + || s.starts_with("panicked at 'Multiple executors at once") + { + // expected, noop + } else { + println!("{}", s); + } + })); + + let res = panic::catch_unwind(move || { + let rt = Runtime::new().unwrap(); + rt.block_on(async {}); + }); + + assert_err!(res); + + panic::set_hook(prev_hook); + }); +} + +#[test] +fn after_start_and_before_stop_is_called() { + use std::sync::atomic::{AtomicUsize, Ordering}; + + let _ = env_logger::try_init(); + + let after_start = Arc::new(AtomicUsize::new(0)); + let before_stop = Arc::new(AtomicUsize::new(0)); + + let after_inner = after_start.clone(); + let before_inner = before_stop.clone(); + let rt = tokio::runtime::Builder::new() + .after_start(move || { + after_inner.clone().fetch_add(1, Ordering::Relaxed); + }) + .before_stop(move || { + before_inner.clone().fetch_add(1, Ordering::Relaxed); + }) + .build() + .unwrap(); + + let (tx, rx) = mpsc::channel(); + + rt.block_on(client_server(tx)); + + let mut e = tokio_executor::enter().unwrap(); + e.block_on(rt.shutdown_on_idle()); + + assert_ok!(rx.try_recv()); + + assert!(after_start.load(Ordering::Relaxed) > 0); + assert!(before_stop.load(Ordering::Relaxed) > 0); +} diff --git a/tokio/tests/timer.rs b/tokio/tests/timer.rs index afb9f8367..297a312e5 100644 --- a/tokio/tests/timer.rs +++ b/tokio/tests/timer.rs @@ -13,7 +13,7 @@ use std::time::{Duration, Instant}; fn timer_with_threaded_runtime() { use tokio::runtime::Runtime; - let mut rt = Runtime::new().unwrap(); + let rt = Runtime::new().unwrap(); let (tx, rx) = mpsc::channel(); rt.spawn(async move { @@ -25,7 +25,9 @@ fn timer_with_threaded_runtime() { tx.send(()).unwrap(); }); - rt.run().unwrap(); + let mut e = tokio_executor::enter().unwrap(); + e.block_on(rt.shutdown_on_idle()); + rx.recv().unwrap(); }