diff --git a/tokio-reactor/Cargo.toml b/tokio-reactor/Cargo.toml index 7861204e6..a7117ca77 100644 --- a/tokio-reactor/Cargo.toml +++ b/tokio-reactor/Cargo.toml @@ -36,4 +36,5 @@ tokio-executor = { version = "0.1.1", path = "../tokio-executor" } tokio-io = { version = "0.1.6", path = "../tokio-io" } [dev-dependencies] +num_cpus = "1.8.0" tokio = { version = "0.1.7", path = ".." } diff --git a/tokio-reactor/benches/basic.rs b/tokio-reactor/benches/basic.rs new file mode 100644 index 000000000..4dbf45446 --- /dev/null +++ b/tokio-reactor/benches/basic.rs @@ -0,0 +1,61 @@ +#![feature(test)] +#![deny(warnings)] + +extern crate futures; +extern crate mio; +extern crate num_cpus; +extern crate test; +extern crate tokio; +extern crate tokio_reactor; + +use std::sync::mpsc; + +use futures::{future, Async}; +use self::test::Bencher; +use tokio_reactor::Registration; + +const NUM_YIELD: usize = 1_000; +const TASKS_PER_CPU: usize = 50; + +#[bench] +fn notify_many(b: &mut Bencher) { + let mut rt = tokio::runtime::Runtime::new().unwrap(); + + let tasks = TASKS_PER_CPU * num_cpus::get(); + let (tx, rx) = mpsc::channel(); + + b.iter(|| { + for _ in 0..tasks { + let (r, s) = mio::Registration::new2(); + let registration = Registration::new(); + registration.register(&r).unwrap(); + + let mut rem = NUM_YIELD; + let mut r = Some(r); + let tx = tx.clone(); + + rt.spawn(future::poll_fn(move || { + loop { + let is_ready = registration.poll_read_ready().unwrap().is_ready(); + + if is_ready { + rem -= 1; + + if rem == 0 { + r.take().unwrap(); + tx.send(()).unwrap(); + return Ok(Async::Ready(())); + } + } else { + s.set_readiness(mio::Ready::readable()).unwrap(); + return Ok(Async::NotReady); + } + } + })); + } + + for _ in 0..tasks { + rx.recv().unwrap(); + } + }); +} diff --git a/tokio-reactor/examples/bench-poll.rs b/tokio-reactor/examples/bench-poll.rs deleted file mode 100644 index 922ced5f3..000000000 --- a/tokio-reactor/examples/bench-poll.rs +++ /dev/null @@ -1,45 +0,0 @@ -extern crate futures; -extern crate mio; -extern crate tokio; -extern crate tokio_reactor; - -use futures::Async; -use mio::Ready; -use tokio::prelude::*; -use tokio_reactor::Registration; - -const NUM_FUTURES: usize = 1000; -const NUM_STEPS: usize = 1000; - -fn main() { - tokio::run(future::lazy(|| { - for _ in 0..NUM_FUTURES { - let (r, s) = mio::Registration::new2(); - let registration = Registration::new(); - registration.register(&r).unwrap(); - - let mut r = Some(r); - let mut step = 0; - - tokio::spawn(future::poll_fn(move || { - loop { - let is_ready = registration.poll_read_ready().unwrap().is_ready(); - - if is_ready { - step += 1; - - if step == NUM_STEPS { - r.take().unwrap(); - return Ok(Async::Ready(())); - } - } else { - s.set_readiness(Ready::readable()).unwrap(); - return Ok(Async::NotReady); - } - } - })); - } - - Ok(()) - })); -} diff --git a/tokio-reactor/src/lib.rs b/tokio-reactor/src/lib.rs index eeefb7d02..9a7f112da 100644 --- a/tokio-reactor/src/lib.rs +++ b/tokio-reactor/src/lib.rs @@ -1,8 +1,6 @@ #![doc(html_root_url = "https://docs.rs/tokio-reactor/0.1.5")] #![deny(missing_docs, warnings, missing_debug_implementations)] -#![cfg_attr(feature = "async-await-preview", feature( - pin, - ))] +#![cfg_attr(feature = "async-await-preview", feature(pin))] //! Event loop that drives Tokio I/O resources. //! diff --git a/tokio-threadpool/src/task/mod.rs b/tokio-threadpool/src/task/mod.rs index 1632c23ae..be95b2dc9 100644 --- a/tokio-threadpool/src/task/mod.rs +++ b/tokio-threadpool/src/task/mod.rs @@ -173,7 +173,7 @@ impl Task { /// Notify the task pub fn notify(me: Arc, pool: &Arc) { - if me.schedule(){ + if me.schedule() { let _ = pool.submit(me, pool); } }