Merge 0d3a41f930 into f1c7ae3340

This commit aims to add benchmarks that more realistically reflect
workloads that might happen in the real world.

These benchmarks are as follows:

- "channels", which sets up TASKS tasks, where each task uses a channel
  to wake up the next one.
- "server", which tries to simulate a web server-type scenario.

Signed-off-by: John Nunley <dev@notgull.net>

benches/executor.rs

@@ -1,3 +1,4 @@
+use std::mem;
 use std::thread::available_parallelism;
 
 use async_executor::Executor;
@@ -139,6 +140,122 @@ fn running_benches(c: &mut Criterion) {
                 *multithread,
             );
         });
+
+        group.bench_function("executor::channels", |b| {
+            run(
+                || {
+                    b.iter(move || {
+                        future::block_on(async {
+                            // Create channels.
+                            let mut tasks = Vec::new();
+                            let (first_send, first_recv) = async_channel::bounded(1);
+                            let mut current_recv = first_recv;
+
+                            for _ in 0..TASKS {
+                                let (next_send, next_recv) = async_channel::bounded(1);
+                                let current_recv = mem::replace(&mut current_recv, next_recv);
+
+                                tasks.push(EX.spawn(async move {
+                                    // Send a notification on to the next task.
+                                    for _ in 0..STEPS {
+                                        current_recv.recv().await.unwrap();
+                                        next_send.send(()).await.unwrap();
+                                    }
+                                }));
+                            }
+
+                            for _ in 0..STEPS {
+                                first_send.send(()).await.unwrap();
+                                current_recv.recv().await.unwrap();
+                            }
+
+                            for task in tasks {
+                                task.await;
+                            }
+                        });
+                    });
+                },
+                *multithread,
+            )
+        });
+
+        group.bench_function("executor::web_server", |b| {
+            run(
+                || {
+                    b.iter(move || {
+                        future::block_on(async {
+                            let (db_send, db_recv) =
+                                async_channel::bounded::<async_channel::Sender<_>>(TASKS / 5);
+                            let mut db_rng = fastrand::Rng::with_seed(0x12345678);
+                            let mut web_rng = db_rng.fork();
+
+                            // This task simulates a database.
+                            let db_task = EX.spawn(async move {
+                                loop {
+                                    // Wait for a new task.
+                                    let incoming = match db_recv.recv().await {
+                                        Ok(incoming) => incoming,
+                                        Err(_) => break,
+                                    };
+
+                                    // Process the task. Maybe it takes a while.
+                                    for _ in 0..db_rng.usize(..10) {
+                                        future::yield_now().await;
+                                    }
+
+                                    // Send the data back.
+                                    incoming.send(db_rng.usize(..)).await.ok();
+                                }
+                            });
+
+                            // This task simulates a web server waiting for new tasks.
+                            let server_task = EX.spawn(async move {
+                                for i in 0..TASKS {
+                                    // Get a new connection.
+                                    if web_rng.usize(..=16) == 16 {
+                                        future::yield_now().await;
+                                    }
+
+                                    let mut web_rng = web_rng.fork();
+                                    let db_send = db_send.clone();
+                                    let task = EX.spawn(async move {
+                                        // Check if the data is cached...
+                                        if web_rng.bool() {
+                                            // ...it's in cache!
+                                            future::yield_now().await;
+                                            return;
+                                        }
+
+                                        // Otherwise we have to make a DB call or two.
+                                        for _ in 0..web_rng.usize(STEPS / 2..STEPS) {
+                                            let (resp_send, resp_recv) = async_channel::bounded(1);
+                                            db_send.send(resp_send).await.unwrap();
+                                            criterion::black_box(resp_recv.recv().await.unwrap());
+                                        }
+
+                                        // Send the data back...
+                                        for _ in 0..web_rng.usize(3..16) {
+                                            future::yield_now().await;
+                                        }
+                                    });
+
+                                    task.detach();
+
+                                    if i & 16 == 0 {
+                                        future::yield_now().await;
+                                    }
+                                }
+                            });
+
+                            // Spawn and wait for it to stop.
+                            server_task.await;
+                            db_task.await;
+                        });
+                    })
+                },
+                *multithread,
+            )
+        });
     }
 }