reactor: turn bench-poll into a proper benchmark (#662)

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 = ".." }

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();
+        }
+    });
+}

tokio-reactor/examples/bench-poll.rs

@@ -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(())
-    }));
-}

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.
 //!

tokio-threadpool/src/task/mod.rs

@@ -173,7 +173,7 @@ impl Task {
 
     /// Notify the task
     pub fn notify(me: Arc<Task>, pool: &Arc<Pool>) {
-        if me.schedule(){
+        if me.schedule() {
             let _ = pool.submit(me, pool);
         }
     }