Added local queue scheduling and "next_task" optimization

Cargo.toml

@@ -1,6 +1,6 @@
 [package]
 name = "async-executor"
-version = "1.4.0"
+version = "1.4.1"
 authors = ["Stjepan Glavina <[email protected]>"]
 edition = "2018"
 description = "Async executor"
@@ -17,10 +17,25 @@ concurrent-queue = "1.2.2"
 fastrand = "1.3.4"
 futures-lite = "1.11.0"
 once_cell = "1.4.1"
-vec-arena = "1.0.0"
+parking_lot = "0.11.1"
+slab = "0.4.2"
+cache-padded="1"
+crossbeam-deque="0.8"
 
 [dev-dependencies]
+async-oneshot="0.5"
 async-channel = "1.4.1"
 async-io = "1.1.9"
 easy-parallel = "3.1.0"
-num_cpus = "1.13.0"
+scopeguard="1.1"
+criterion = "0.3"
+num_cpus="1"
+
+[profile.release]
+incremental = true
+debug = 2 # Set this to 1 or 2 to get more useful backtraces in debugger.
+lto = 'thin'
+
+[[bench]]
+name = "my_benchmark"
+harness = false

benches/executor.rs

@@ -1,109 +1,165 @@
-#![feature(test)]
+// #![feature(test)]
 
-extern crate test;
+// extern crate test;
 
-use std::future::Future;
+// use std::future::Future;
 
-use async_executor::Executor;
-use futures_lite::{future, prelude::*};
+// use async_executor::{Executor, Task};
+// use futures_lite::{future, prelude::*};
 
-const TASKS: usize = 300;
-const STEPS: usize = 300;
-const LIGHT_TASKS: usize = 25_000;
+// const TASKS: usize = 300;
+// const STEPS: usize = 300;
+// const LIGHT_TASKS: usize = 25_000;
 
-static EX: Executor<'_> = Executor::new();
+// static EX: Executor<'_> = Executor::new();
 
-fn run(f: impl FnOnce()) {
-    let (s, r) = async_channel::bounded::<()>(1);
-    easy_parallel::Parallel::new()
-        .each(0..num_cpus::get(), |_| future::block_on(EX.run(r.recv())))
-        .finish(move || {
-            let _s = s;
-            f()
-        });
-}
+// fn run(f: impl FnOnce()) {
+//     let (s, r) = async_channel::bounded::<()>(1);
+//     easy_parallel::Parallel::new()
+//         .each(0..num_cpus::get(), |_| future::block_on(EX.run(r.recv())))
+//         .finish(move || {
+//             let _s = s;
+//             f()
+//         });
+// }
 
-#[bench]
-fn create(b: &mut test::Bencher) {
-    b.iter(move || {
-        let ex = Executor::new();
-        let task = ex.spawn(async {});
-        future::block_on(ex.run(task));
-    });
-}
+// #[bench]
+// fn create(b: &mut test::Bencher) {
+//     b.iter(move || {
+//         let ex = Executor::new();
+//         let task = ex.spawn(async {});
+//         future::block_on(ex.run(task));
+//     });
+// }
 
-#[bench]
-fn spawn_one(b: &mut test::Bencher) {
-    run(|| {
-        b.iter(move || {
-            future::block_on(async { EX.spawn(async {}).await });
-        });
-    });
-}
+// #[bench]
+// fn spawn_one(b: &mut test::Bencher) {
+//     run(|| {
+//         b.iter(move || {
+//             future::block_on(async { EX.spawn(async {}).await });
+//         });
+//     });
+// }
 
-#[bench]
-fn spawn_many(b: &mut test::Bencher) {
-    run(|| {
-        b.iter(move || {
-            future::block_on(async {
-                let mut tasks = Vec::new();
-                for _ in 0..LIGHT_TASKS {
-                    tasks.push(EX.spawn(async {}));
-                }
-                for task in tasks {
-                    task.await;
-                }
-            });
-        });
-    });
-}
+// #[bench]
+// fn spawn_many(b: &mut test::Bencher) {
+//     run(|| {
+//         b.iter(move || {
+//             future::block_on(async {
+//                 let mut tasks = Vec::new();
+//                 for _ in 0..LIGHT_TASKS {
+//                     tasks.push(EX.spawn(async {}));
+//                 }
+//                 for task in tasks {
+//                     task.await;
+//                 }
+//             });
+//         });
+//     });
+// }
 
-#[bench]
-fn spawn_recursively(b: &mut test::Bencher) {
-    fn go(i: usize) -> impl Future<Output = ()> + Send + 'static {
-        async move {
-            if i != 0 {
-                EX.spawn(async move {
-                    let fut = go(i - 1).boxed();
-                    fut.await;
-                })
-                .await;
-            }
-        }
-    }
+// #[bench]
+// fn spawn_recursively(b: &mut test::Bencher) {
+//     fn go(i: usize) -> impl Future<Output = ()> + Send + 'static {
+//         async move {
+//             if i != 0 {
+//                 EX.spawn(async move {
+//                     let fut = go(i - 1).boxed();
+//                     fut.await;
+//                 })
+//                 .await;
+//             }
+//         }
+//     }
 
-    run(|| {
-        b.iter(move || {
-            future::block_on(async {
-                let mut tasks = Vec::new();
-                for _ in 0..TASKS {
-                    tasks.push(EX.spawn(go(STEPS)));
-                }
-                for task in tasks {
-                    task.await;
-                }
-            });
-        });
-    });
-}
+//     run(|| {
+//         b.iter(move || {
+//             future::block_on(async {
+//                 let mut tasks = Vec::new();
+//                 for _ in 0..TASKS {
+//                     tasks.push(EX.spawn(go(STEPS)));
+//                 }
+//                 for task in tasks {
+//                     task.await;
+//                 }
+//             });
+//         });
+//     });
+// }
 
-#[bench]
-fn yield_now(b: &mut test::Bencher) {
-    run(|| {
-        b.iter(move || {
-            future::block_on(async {
-                let mut tasks = Vec::new();
-                for _ in 0..TASKS {
-                    tasks.push(EX.spawn(async move {
-                        for _ in 0..STEPS {
-                            future::yield_now().await;
-                        }
-                    }));
-                }
-                for task in tasks {
-                    task.await;
-                }
-            });
-        });
-    });
-}
+// #[bench]
+// fn yield_now(b: &mut test::Bencher) {
+//     run(|| {
+//         b.iter(move || {
+//             future::block_on(async {
+//                 let mut tasks = Vec::new();
+//                 for _ in 0..TASKS {
+//                     tasks.push(EX.spawn(async move {
+//                         for _ in 0..STEPS {
+//                             future::yield_now().await;
+//                         }
+//                     }));
+//                 }
+//                 for task in tasks {
+//                     task.await;
+//                 }
+//             });
+//         });
+//     });
+// }
+
+// #[bench]
+// fn context_switch_quiet(b: &mut test::Bencher) {
+//     let (send, mut recv) = async_channel::bounded::<usize>(1);
+//     let mut tasks: Vec<Task<Option<()>>> = vec![];
+//     for _ in 0..TASKS {
+//         let old_recv = recv.clone();
+//         let (new_send, new_recv) = async_channel::bounded(1);
+//         tasks.push(EX.spawn(async move {
+//             loop {
+//                 new_send.send(old_recv.recv().await.ok()?).await.ok()?
+//             }
+//         }));
+//         recv = new_recv;
+//     }
+//     run(|| {
+//         b.iter(move || {
+//             future::block_on(async {
+//                 send.send(1).await.unwrap();
+//                 recv.recv().await.unwrap();
+//             });
+//         });
+//     });
+// }
+
+// #[bench]
+// fn context_switch_busy(b: &mut test::Bencher) {
+//     let (send, mut recv) = async_channel::bounded::<usize>(1);
+//     let mut tasks: Vec<Task<Option<()>>> = vec![];
+//     for num in 0..TASKS {
+//         let old_recv = recv.clone();
+//         let (new_send, new_recv) = async_channel::bounded(1);
+//         tasks.push(EX.spawn(async move {
+//             loop {
+//                 // eprintln!("forward {}", num);
+//                 new_send.send(old_recv.recv().await.ok()?).await.ok()?;
+//             }
+//         }));
+//         recv = new_recv;
+//     }
+//     for _ in 0..TASKS {
+//         tasks.push(EX.spawn(async move {
+//             loop {
+//                 future::yield_now().await;
+//             }
+//         }))
+//     }
+//     run(|| {
+//         b.iter(move || {
+//             future::block_on(async {
+//                 send.send(1).await.unwrap();
+//                 recv.recv().await.unwrap();
+//             });
+//         });
+//     });
+// }