mirror of
https://github.com/tokio-rs/tokio.git
synced 2026-08-29 00:00:11 +02:00
Add some missing future::Executor implementations (#563)
This adds an implementation of future::Executor for `executor::DefaultExecutor` and `runtime::current_thread::Handle`.
This commit is contained in:
Carl Lerche
+229
-110
@@ -11,6 +11,11 @@ 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,
|
||||
@@ -72,78 +77,130 @@ fn runtime_single_threaded_block_on() {
|
||||
tokio::runtime::current_thread::block_on_all(create_client_server_future()).unwrap();
|
||||
}
|
||||
|
||||
#[test]
|
||||
fn runtime_single_threaded_block_on_all() {
|
||||
let cnt = Arc::new(Mutex::new(0));
|
||||
let c = cnt.clone();
|
||||
mod runtime_single_threaded_block_on_all {
|
||||
use super::*;
|
||||
|
||||
let msg = tokio::runtime::current_thread::block_on_all(lazy(move || {
|
||||
{
|
||||
let mut x = c.lock().unwrap();
|
||||
*x = 1 + *x;
|
||||
}
|
||||
fn test<F>(spawn: F)
|
||||
where
|
||||
F: Fn(Box<Future<Item=(), Error=()> + Send>),
|
||||
{
|
||||
let cnt = Arc::new(Mutex::new(0));
|
||||
let c = cnt.clone();
|
||||
|
||||
// Spawn!
|
||||
tokio::spawn(lazy(move || {
|
||||
let msg = tokio::runtime::current_thread::block_on_all(lazy(move || {
|
||||
{
|
||||
let mut x = c.lock().unwrap();
|
||||
*x = 1 + *x;
|
||||
}
|
||||
Ok::<(), ()>(())
|
||||
}));
|
||||
|
||||
Ok::<_, ()>("hello")
|
||||
})).unwrap();
|
||||
// Spawn!
|
||||
spawn(Box::new(lazy(move || {
|
||||
{
|
||||
let mut x = c.lock().unwrap();
|
||||
*x = 1 + *x;
|
||||
}
|
||||
Ok::<(), ()>(())
|
||||
})));
|
||||
|
||||
assert_eq!(2, *cnt.lock().unwrap());
|
||||
assert_eq!(msg, "hello");
|
||||
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();
|
||||
})
|
||||
}
|
||||
}
|
||||
|
||||
#[test]
|
||||
fn runtime_single_threaded_racy_spawn() {
|
||||
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();
|
||||
mod runtime_single_threaded_racy {
|
||||
use super::*;
|
||||
fn test<F>(spawn: F)
|
||||
where
|
||||
F: Fn(
|
||||
tokio::runtime::current_thread::Handle,
|
||||
Box<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();
|
||||
// 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();
|
||||
});
|
||||
// run until all spawned futures (incl. the "exit" signal future) have completed.
|
||||
rt.run().unwrap();
|
||||
});
|
||||
|
||||
let (tx, rx) = futures::sync::oneshot::channel();
|
||||
let (tx, rx) = futures::sync::oneshot::channel();
|
||||
|
||||
let handle = handle_rx.recv().unwrap();
|
||||
handle
|
||||
.spawn(futures::future::lazy(move || {
|
||||
let handle = handle_rx.recv().unwrap();
|
||||
spawn(handle, Box::new(futures::future::lazy(move || {
|
||||
tx.send(()).unwrap();
|
||||
Ok(())
|
||||
}))
|
||||
.unwrap();
|
||||
})));
|
||||
|
||||
// signal runtime thread to exit
|
||||
trigger.send(()).unwrap();
|
||||
// signal runtime thread to exit
|
||||
trigger.send(()).unwrap();
|
||||
|
||||
// wait for runtime thread to exit
|
||||
jh.join().unwrap();
|
||||
// wait for runtime thread to exit
|
||||
jh.join().unwrap();
|
||||
|
||||
assert_eq!(rx.wait().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(); })
|
||||
}
|
||||
}
|
||||
|
||||
#[test]
|
||||
fn runtime_multi_threaded() {
|
||||
let _ = env_logger::try_init();
|
||||
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();
|
||||
runtime.spawn(create_client_server_future());
|
||||
runtime.shutdown_on_idle().wait().unwrap();
|
||||
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};
|
||||
@@ -161,35 +218,57 @@ fn block_on_timer() {
|
||||
runtime.shutdown_on_idle().wait().unwrap();
|
||||
}
|
||||
|
||||
#[test]
|
||||
fn spawn_from_block_on() {
|
||||
let cnt = Arc::new(Mutex::new(0));
|
||||
let c = cnt.clone();
|
||||
mod from_block_on {
|
||||
use super::*;
|
||||
|
||||
let mut runtime = Runtime::new().unwrap();
|
||||
let msg = runtime
|
||||
.block_on(lazy(move || {
|
||||
{
|
||||
let mut x = c.lock().unwrap();
|
||||
*x = 1 + *x;
|
||||
}
|
||||
fn test<F>(spawn: F)
|
||||
where
|
||||
F: Fn(Box<Future<Item=(), Error=()> + Send>) + Send + 'static,
|
||||
{
|
||||
let cnt = Arc::new(Mutex::new(0));
|
||||
let c = cnt.clone();
|
||||
|
||||
// Spawn!
|
||||
tokio::spawn(lazy(move || {
|
||||
let mut runtime = Runtime::new().unwrap();
|
||||
let msg = runtime
|
||||
.block_on(lazy(move || {
|
||||
{
|
||||
let mut x = c.lock().unwrap();
|
||||
*x = 1 + *x;
|
||||
}
|
||||
Ok::<(), ()>(())
|
||||
}));
|
||||
|
||||
Ok::<_, ()>("hello")
|
||||
}))
|
||||
.unwrap();
|
||||
// Spawn!
|
||||
spawn(Box::new(lazy(move || {
|
||||
{
|
||||
let mut x = c.lock().unwrap();
|
||||
*x = 1 + *x;
|
||||
}
|
||||
Ok::<(), ()>(())
|
||||
})));
|
||||
|
||||
runtime.shutdown_on_idle().wait().unwrap();
|
||||
assert_eq!(2, *cnt.lock().unwrap());
|
||||
assert_eq!(msg, "hello");
|
||||
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]
|
||||
@@ -220,54 +299,94 @@ fn block_waits() {
|
||||
runtime.shutdown_on_idle().wait().unwrap();
|
||||
}
|
||||
|
||||
#[test]
|
||||
fn spawn_many() {
|
||||
mod many {
|
||||
use super::*;
|
||||
|
||||
const ITER: usize = 200;
|
||||
fn test<F>(spawn: F)
|
||||
where
|
||||
F: Fn(&mut Runtime, Box<Future<Item=(), Error=()> + Send>),
|
||||
{
|
||||
let cnt = Arc::new(Mutex::new(0));
|
||||
let mut runtime = Runtime::new().unwrap();
|
||||
|
||||
let cnt = Arc::new(Mutex::new(0));
|
||||
let mut runtime = Runtime::new().unwrap();
|
||||
|
||||
for _ in 0..ITER {
|
||||
let c = cnt.clone();
|
||||
runtime.spawn(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_from_block_on_all() {
|
||||
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!
|
||||
tokio::spawn(lazy(move || {
|
||||
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::<(), ()>(())
|
||||
}));
|
||||
})));
|
||||
}
|
||||
|
||||
Ok::<_, ()>("hello")
|
||||
}))
|
||||
.unwrap();
|
||||
runtime.shutdown_on_idle().wait().unwrap();
|
||||
assert_eq!(ITER, *cnt.lock().unwrap());
|
||||
}
|
||||
|
||||
assert_eq!(2, *cnt.lock().unwrap());
|
||||
assert_eq!(msg, "hello");
|
||||
#[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<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); })
|
||||
}
|
||||
}
|
||||
|
||||
Reference in New Issue
Block a user