tokio-rs/tokio
1
0
Fork 0
mirror of https://github.com/tokio-rs/tokio.git synced 2026-08-20 00:00:08 +02:00
Code Issues Packages Projects Releases Wiki Activity

tokio: update threaded runtime to std::future (#1280)

Browse Source 
re-enables the threaded runtime and sets it (again) as the default.
This commit is contained in:
Carl Lerche
2019-07-10 11:21:06 -07:00
committed by GitHub
parent e5525628cd
commit bd3f3270db
20 changed files with 418 additions and 864 deletions
Download Patch File Download Diff File Expand all files Collapse all files
ci/azure-test-stable.yml
+4 -4
View File
@@ -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 }}
tokio-signal/examples/ctrl-c.rs
+2 -3
View File
@@ -7,13 +7,13 @@ use futures_util::stream::StreamExt;
const STOP_AFTER: u64 = 10;
#[tokio::main]
async fn main() -> Result<(), Box<dyn std::error::Error>> {
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<dyn std::error::Error>> {
}
println!("Stream ended, quiting the program.");
Ok(())
}
tokio-signal/examples/multiple.rs
+6 -15
View File
@@ -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<dyn Error>> {
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<dyn Error>> {
Ok(())
}
pub async fn main() {}
}
#[tokio::main]
async fn main() -> Result<(), Box<dyn Error>> {
async fn main() {
platform::main().await
}
tokio-signal/examples/sighup-example.rs
+4 -12
View File
@@ -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<dyn Error>> {
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<dyn Error>> {
Ok(())
}
pub async fn main() {}
}
#[tokio::main]
async fn main() -> Result<(), Box<dyn Error>> {
async fn main() {
platform::main().await
}
tokio-threadpool/src/thread_pool.rs
+5 -1
View File
@@ -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);
}
}
tokio/Cargo.toml
+3 -2
View File
@@ -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" }
tokio/examples/echo-udp.rs
+5 -6
View File
@@ -50,12 +50,12 @@ impl Server {
}
#[tokio::main]
async fn main() -> Result<(), Box<dyn std::error::Error>> {
async fn main() {
let addr = env::args().nth(1).unwrap_or("127.0.0.1:8080".to_string());
let addr = addr.parse::<SocketAddr>()?;
let addr = addr.parse::<SocketAddr>().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<dyn std::error::Error>> {
};
// This starts the server task.
server.run().await?;
Ok(())
server.run().await.unwrap();
}
tokio/examples/hello_world.rs
+3 -5
View File
@@ -19,16 +19,14 @@ use tokio::io::AsyncWriteExt;
use tokio::net::TcpStream;
#[tokio::main]
pub async fn main() -> Result<(), Box<dyn std::error::Error>> {
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(())
}
tokio/examples/udp-client.rs
+12 -9
View File
@@ -41,29 +41,32 @@ fn get_stdin_data() -> Result<Vec<u8>, Box<dyn std::error::Error>> {
}
#[tokio::main]
async fn main() -> Result<(), Box<dyn std::error::Error>> {
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(())
}
tokio/src/lib.rs
+1 -1
View File
@@ -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;
tokio/src/runtime/current_thread/mod.rs
-31
View File
@@ -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<F>(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<F>(future: F)
where
F: Future<Output = ()> + '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");
}
tokio/src/runtime/mod.rs
+1 -6
View File
@@ -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,
};
*/
tokio/src/runtime/threadpool/builder.rs
+3 -21
View File
@@ -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,
}),
})
tokio/src/runtime/threadpool/mod.rs
+19 -158
View File
@@ -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<Option<Reactor>>,
/// 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>(_: T) -> Box<dyn Future<Item = (), Error = ()> + 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<F>(future: F)
where F: Future<Item = (), Error = ()> + 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<F>(&self, future: F) -> &Self
where F: Future<Item = (), Error = ()> + Send + 'static,
where F: Future<Output = ()> + 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<F, R, E>(&self, future: F) -> Result<R, E>
pub fn block_on<F>(&self, future: F) -> F::Output
where
F: Send + 'static + Future<Item = R, Error = E>,
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<F, R, E>(self, future: F) -> Result<R, E>
where
F: Send + 'static + Future<Item = R, Error = E>,
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();
}
}
}
tokio/src/runtime/threadpool/shutdown.rs
-34
View File
@@ -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<Future<Item = (), Error = ()>>")
.finish()
}
}
tokio/src/runtime/threadpool/task_executor.rs
+13 -17
View File
@@ -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<F>(&self, future: F)
where F: Future<Item = (), Error = ()> + Send + 'static,
where F: Future<Output = ()> + Send + 'static,
{
self.inner.spawn(future).unwrap();
}
}
impl<T> future::Executor<T> for TaskExecutor
where T: Future<Item = (), Error = ()> + Send + 'static,
{
fn execute(&self, future: T) -> Result<(), future::ExecuteError<T>> {
self.inner.execute(future)
}
}
impl crate::executor::Executor for TaskExecutor {
fn spawn(&mut self, future: Box<dyn Future<Item = (), Error = ()> + Send>)
-> Result<(), crate::executor::SpawnError>
{
impl tokio_executor::Executor for TaskExecutor {
fn spawn(
&mut self,
future: Pin<Box<dyn Future<Output = ()> + Send>>,
) -> Result<(), SpawnError> {
self.inner.spawn(future)
}
}
impl<T> crate::executor::TypedExecutor<T> for TaskExecutor
impl<T> tokio_executor::TypedExecutor<T> for TaskExecutor
where
T: Future<Item = (), Error = ()> + Send + 'static,
T: Future<Output = ()> + 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))
}
}
tokio/tests/runtime.rs
-537
View File
@@ -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<dyn Future<Item = (), Error = ()> + 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<F>(spawn: F)
where
F: Fn(Box<dyn Future<Item = (), Error = ()> + 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<F>(spawn: F)
where
F: Fn(
tokio::runtime::current_thread::Handle,
Box<dyn Future<Item = (), Error = ()> + 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<F>(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<T>(x: T) -> Box<dyn Future<Item = T, Error = Error> + 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<F>(spawn: F)
where
F: Fn(Box<dyn Future<Item = (), Error = ()> + 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<F>(spawn: F)
where
F: Fn(&mut Runtime, Box<dyn Future<Item = (), Error = ()> + 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<F>(spawn: F)
where
F: Fn(Box<dyn Future<Item = (), Error = ()> + 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<F1, F2>(first: F1, nested: F2)
where
F1: Fn(Box<dyn Future<Item = (), Error = ()> + Send>) + Send + 'static,
F2: Fn(Box<dyn Future<Item = (), Error = ()> + 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);
}
tokio/tests/runtime_current_thread.rs
+139
View File
@@ -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();
}
tokio/tests/runtime_threaded.rs
+194
View File
@@ -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);
}
tokio/tests/timer.rs
+4 -2
View File
@@ -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();
}