bench: Add some more filled-out benchmarks

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>

.github/workflows/ci.yml

@@ -37,25 +37,24 @@ jobs:
       - uses: actions/checkout@v4
       - name: Install Rust
         run: rustup update ${{ matrix.rust }} && rustup default ${{ matrix.rust }}
+      - run: rustup target add wasm32-unknown-unknown
+      - uses: taiki-e/install-action@cargo-hack
       - run: cargo build --all --all-features --all-targets
         if: startsWith(matrix.rust, 'nightly')
       - name: Run cargo check (without dev-dependencies to catch missing feature flags)
         if: startsWith(matrix.rust, 'nightly')
         run: cargo check -Z features=dev_dep
       - run: cargo test
+      - run: cargo check --all --all-features --target wasm32-unknown-unknown
+      - run: cargo hack build --all --all-features --target wasm32-unknown-unknown --no-dev-deps
 
   msrv:
     runs-on: ubuntu-latest
-    strategy:
-      matrix:
-        # When updating this, the reminder to update the minimum supported
-        # Rust version in Cargo.toml.
-        rust: ['1.61']
     steps:
       - uses: actions/checkout@v4
-      - name: Install Rust
-        run: rustup update ${{ matrix.rust }} && rustup default ${{ matrix.rust }}
-      - run: cargo build
+      - name: Install cargo-hack
+        uses: taiki-e/install-action@cargo-hack
+      - run: cargo hack build --rust-version
 
   clippy:
     runs-on: ubuntu-latest

CHANGELOG.md

@@ -1,3 +1,39 @@
+# Version 1.11.0
+
+- Re-export the `async_task::FallibleTask` primitive. (#113)
+- Support racy initialization of the executor state. This should allow the executor to be
+  initialized on web targets without any issues. (#108)
+
+# Version 1.10.0
+
+- Add a function `spawn_batch` that allows users to spawn multiple tasks while only locking the executor once. (#92)
+
+# Version 1.9.1
+
+- Remove the thread-local optimization due to the bugs that it introduces. (#106)
+
+# Version 1.9.0
+
+- Re-introduce the thread-local task push optimization to the executor. (#93)
+- Bump `async-task` to v4.4.0. (#90)
+- Replace some unnecessary atomic operations with non-atomic operations. (#94)
+- Use weaker atomic orderings for notifications. (#95)
+- When spawning a future, avoid looking up the ID to assign to that future twice. (#96)
+
+# Version 1.8.0
+
+- When spawned tasks panic, the panic is caught and then surfaced in the spawned
+ `Task`. Previously, the panic would be surfaced in `tick()` or `run()`. (#78)
+
+# Version 1.7.2
+
+- Fix compilation under WebAssembly targets (#77).
+
+# Version 1.7.1
+
+- Fix compilation under WebAssembly targets (#75).
+- Add a disclaimer indicating that this is a reference executor (#74).
+
 # Version 1.7.0
 
 - Bump `async-lock` and `futures-lite` to their latest versions. (#70)

Cargo.toml

@@ -3,10 +3,10 @@ name = "async-executor"
 # When publishing a new version:
 # - Update CHANGELOG.md
 # - Create "v1.x.y" git tag
-version = "1.7.0"
-authors = ["Stjepan Glavina <stjepang@gmail.com>"]
-edition = "2018"
-rust-version = "1.61"
+version = "1.11.0"
+authors = ["Stjepan Glavina <stjepang@gmail.com>", "John Nunley <dev@notgull.net>"]
+edition = "2021"
+rust-version = "1.60"
 description = "Async executor"
 license = "Apache-2.0 OR MIT"
 repository = "https://github.com/smol-rs/async-executor"
@@ -15,19 +15,21 @@ categories = ["asynchronous", "concurrency"]
 exclude = ["/.*"]
 
 [dependencies]
-async-lock = "3.0.0"
-async-task = "4.0.0"
+async-task = "4.4.0"
 concurrent-queue = "2.0.0"
 fastrand = "2.0.0"
 futures-lite = { version = "2.0.0", default-features = false }
 slab = "0.4.4"
 
+[target.'cfg(target_family = "wasm")'.dependencies]
+futures-lite = { version = "2.0.0", default-features = false, features = ["std"] }
+
 [dev-dependencies]
 async-channel = "2.0.0"
 async-io = "2.1.0"
-criterion = { version = "0.4.0", default-features = false, features = ["cargo_bench_support"] }
+async-lock = "3.0.0"
+criterion = { version = "0.5", default-features = false, features = ["cargo_bench_support"] }
 easy-parallel = "3.1.0"
-event-listener = "3.0.0"
 fastrand = "2.0.0"
 futures-lite = "2.0.0"
 once_cell = "1.16.0"

README.md

@@ -11,6 +11,13 @@ https://docs.rs/async-executor)
 
 Async executors.
 
+This crate provides two reference executors that trade performance for
+functionality. They should be considered reference executors that are "good
+enough" for most use cases. For more specialized use cases, consider writing
+your own executor on top of [`async-task`].
+
+[`async-task`]: https://crates.io/crates/async-task
+
 ## Examples
 
 ```rust

benches/executor.rs

@@ -1,4 +1,4 @@
-use std::future::Future;
+use std::mem;
 use std::thread::available_parallelism;
 
 use async_executor::Executor;
@@ -52,6 +52,21 @@ fn running_benches(c: &mut Criterion) {
             );
         });
 
+        group.bench_function("executor::spawn_batch", |b| {
+            run(
+                || {
+                    let mut handles = vec![];
+
+                    b.iter(|| {
+                        EX.spawn_many((0..250).map(|_| future::yield_now()), &mut handles);
+                    });
+
+                    handles.clear();
+                },
+                *multithread,
+            )
+        });
+
         group.bench_function("executor::spawn_many_local", |b| {
             run(
                 || {
@@ -125,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,
+            )
+        });
     }
 }
 

examples/limit.rs

@@ -1,46 +1,23 @@
 //! An executor where you can only push a limited number of tasks.
 
 use async_executor::{Executor, Task};
-use event_listener::{Event, EventListener};
-use futures_lite::pin;
-use std::{
-    future::Future,
-    sync::{
-        atomic::{AtomicUsize, Ordering},
-        Arc,
-    },
-    time::Duration,
-};
+use async_lock::Semaphore;
+use std::{future::Future, sync::Arc, time::Duration};
 
 /// An executor where you can only push a limited number of tasks.
 struct LimitedExecutor {
     /// Inner running executor.
     executor: Executor<'static>,
 
-    /// Shared state.
-    shared: Arc<SharedState>,
-}
-
-struct SharedState {
-    /// The maximum number of tasks that can be pushed.
-    max: usize,
-
-    /// The current number of active tasks.
-    active: AtomicUsize,
-
-    /// Event listeners for when a new task is available.
-    slot_available: Event,
+    /// Semaphore limiting the number of tasks.
+    semaphore: Arc<Semaphore>,
 }
 
 impl LimitedExecutor {
     fn new(max: usize) -> Self {
         Self {
             executor: Executor::new(),
-            shared: Arc::new(SharedState {
-                max,
-                active: AtomicUsize::new(0),
-                slot_available: Event::new(),
-            }),
+            semaphore: Semaphore::new(max).into(),
         }
     }
 
@@ -49,67 +26,18 @@ impl LimitedExecutor {
     where
         F::Output: Send + 'static,
     {
-        let listener = EventListener::new(&self.shared.slot_available);
-        pin!(listener);
+        // Wait for a semaphore permit.
+        let permit = self.semaphore.acquire_arc().await;
 
-        // Load the current number of active tasks.
-        let mut active = self.shared.active.load(Ordering::Acquire);
+        // Wrap it into a new future.
+        let future = async move {
+            let result = future.await;
+            drop(permit);
+            result
+        };
 
-        loop {
-            // Check if there is a slot available.
-            if active < self.shared.max {
-                // Try to set the slot to what would be the new number of tasks.
-                let new_active = active + 1;
-                match self.shared.active.compare_exchange(
-                    active,
-                    new_active,
-                    Ordering::SeqCst,
-                    Ordering::SeqCst,
-                ) {
-                    Ok(_) => {
-                        // Wrap the future in another future that decrements the active count
-                        // when it's done.
-                        let future = {
-                            let shared = self.shared.clone();
-                            async move {
-                                struct DecOnDrop(Arc<SharedState>);
-
-                                impl Drop for DecOnDrop {
-                                    fn drop(&mut self) {
-                                        // Decrement the count and notify someone.
-                                        self.0.active.fetch_sub(1, Ordering::SeqCst);
-                                        self.0.slot_available.notify(usize::MAX);
-                                    }
-                                }
-
-                                let _dec = DecOnDrop(shared);
-                                future.await
-                            }
-                        };
-
-                        // Wake up another waiter, in case there is one.
-                        self.shared.slot_available.notify(1);
-
-                        // Spawn the task.
-                        return self.executor.spawn(future);
-                    }
-
-                    Err(actual) => {
-                        // Try again.
-                        active = actual;
-                    }
-                }
-            } else {
-                // Start waiting for a slot to become available.
-                if listener.as_ref().is_listening() {
-                    listener.as_mut().await;
-                } else {
-                    listener.as_mut().listen();
-                }
-
-                active = self.shared.active.load(Ordering::Acquire);
-            }
-        }
+        // Spawn the task.
+        self.executor.spawn(future)
     }
 
     /// Run a future to completion.

examples/priority.rs

@@ -1,6 +1,5 @@
 //! An executor with task priorities.
 
-use std::future::Future;
 use std::thread;
 
 use async_executor::{Executor, Task};

src/lib.rs

@@ -1,5 +1,12 @@
 //! Async executors.
 //!
+//! This crate provides two reference executors that trade performance for
+//! functionality. They should be considered reference executors that are "good
+//! enough" for most use cases. For more specialized use cases, consider writing
+//! your own executor on top of [`async-task`].
+//!
+//! [`async-task`]: https://crates.io/crates/async-task
+//!
 //! # Examples
 //!
 //! ```
@@ -18,7 +25,12 @@
 //! future::block_on(ex.run(task));
 //! ```
 
-#![warn(missing_docs, missing_debug_implementations, rust_2018_idioms)]
+#![warn(
+    missing_docs,
+    missing_debug_implementations,
+    rust_2018_idioms,
+    clippy::undocumented_unsafe_blocks
+)]
 #![doc(
     html_favicon_url = "https://raw.githubusercontent.com/smol-rs/smol/master/assets/images/logo_fullsize_transparent.png"
 )]
@@ -27,22 +39,20 @@
 )]
 
 use std::fmt;
-use std::future::Future;
 use std::marker::PhantomData;
 use std::panic::{RefUnwindSafe, UnwindSafe};
 use std::rc::Rc;
-use std::sync::atomic::{AtomicBool, AtomicUsize, Ordering};
+use std::sync::atomic::{AtomicBool, AtomicPtr, Ordering};
 use std::sync::{Arc, Mutex, RwLock, TryLockError};
 use std::task::{Poll, Waker};
 
-use async_lock::OnceCell;
-use async_task::Runnable;
+use async_task::{Builder, Runnable};
 use concurrent_queue::ConcurrentQueue;
 use futures_lite::{future, prelude::*};
 use slab::Slab;
 
 #[doc(no_inline)]
-pub use async_task::Task;
+pub use async_task::{FallibleTask, Task};
 
 /// An async executor.
 ///
@@ -70,13 +80,15 @@ pub use async_task::Task;
 /// ```
 pub struct Executor<'a> {
     /// The executor state.
-    state: OnceCell<Arc<State>>,
+    state: AtomicPtr<State>,
 
     /// Makes the `'a` lifetime invariant.
     _marker: PhantomData<std::cell::UnsafeCell<&'a ()>>,
 }
 
+// SAFETY: Executor stores no thread local state that can be accessed via other thread.
 unsafe impl Send for Executor<'_> {}
+// SAFETY: Executor internally synchronizes all of it's operations internally.
 unsafe impl Sync for Executor<'_> {}
 
 impl UnwindSafe for Executor<'_> {}
@@ -100,7 +112,7 @@ impl<'a> Executor<'a> {
     /// ```
     pub const fn new() -> Executor<'a> {
         Executor {
-            state: OnceCell::new(),
+            state: AtomicPtr::new(std::ptr::null_mut()),
             _marker: PhantomData,
         }
     }
@@ -143,17 +155,120 @@ impl<'a> Executor<'a> {
     pub fn spawn<T: Send + 'a>(&self, future: impl Future<Output = T> + Send + 'a) -> Task<T> {
         let mut active = self.state().active.lock().unwrap();
 
+        // SAFETY: `T` and the future are `Send`.
+        unsafe { self.spawn_inner(future, &mut active) }
+    }
+
+    /// Spawns many tasks onto the executor.
+    ///
+    /// As opposed to the [`spawn`] method, this locks the executor's inner task lock once and
+    /// spawns all of the tasks in one go. With large amounts of tasks this can improve
+    /// contention.
+    ///
+    /// For very large numbers of tasks the lock is occasionally dropped and re-acquired to
+    /// prevent runner thread starvation. It is assumed that the iterator provided does not
+    /// block; blocking iterators can lock up the internal mutex and therefore the entire
+    /// executor.
+    ///
+    /// ## Example
+    ///
+    /// ```
+    /// use async_executor::Executor;
+    /// use futures_lite::{stream, prelude::*};
+    /// use std::future::ready;
+    ///
+    /// # futures_lite::future::block_on(async {
+    /// let mut ex = Executor::new();
+    ///
+    /// let futures = [
+    ///     ready(1),
+    ///     ready(2),
+    ///     ready(3)
+    /// ];
+    ///
+    /// // Spawn all of the futures onto the executor at once.
+    /// let mut tasks = vec![];
+    /// ex.spawn_many(futures, &mut tasks);
+    ///
+    /// // Await all of them.
+    /// let results = ex.run(async move {
+    ///     stream::iter(tasks).then(|x| x).collect::<Vec<_>>().await
+    /// }).await;
+    /// assert_eq!(results, [1, 2, 3]);
+    /// # });
+    /// ```
+    ///
+    /// [`spawn`]: Executor::spawn
+    pub fn spawn_many<T: Send + 'a, F: Future<Output = T> + Send + 'a>(
+        &self,
+        futures: impl IntoIterator<Item = F>,
+        handles: &mut impl Extend<Task<F::Output>>,
+    ) {
+        let mut active = Some(self.state().active.lock().unwrap());
+
+        // Convert the futures into tasks.
+        let tasks = futures.into_iter().enumerate().map(move |(i, future)| {
+            // SAFETY: `T` and the future are `Send`.
+            let task = unsafe { self.spawn_inner(future, active.as_mut().unwrap()) };
+
+            // Yield the lock every once in a while to ease contention.
+            if i.wrapping_sub(1) % 500 == 0 {
+                drop(active.take());
+                active = Some(self.state().active.lock().unwrap());
+            }
+
+            task
+        });
+
+        // Push the tasks to the user's collection.
+        handles.extend(tasks);
+    }
+
+    /// Spawn a future while holding the inner lock.
+    ///
+    /// # Safety
+    ///
+    /// If this is an `Executor`, `F` and `T` must be `Send`.
+    unsafe fn spawn_inner<T: 'a>(
+        &self,
+        future: impl Future<Output = T> + 'a,
+        active: &mut Slab<Waker>,
+    ) -> Task<T> {
         // Remove the task from the set of active tasks when the future finishes.
-        let index = active.vacant_entry().key();
-        let state = self.state().clone();
+        let entry = active.vacant_entry();
+        let index = entry.key();
+        let state = self.state_as_arc();
         let future = async move {
             let _guard = CallOnDrop(move || drop(state.active.lock().unwrap().try_remove(index)));
             future.await
         };
 
         // Create the task and register it in the set of active tasks.
-        let (runnable, task) = unsafe { async_task::spawn_unchecked(future, self.schedule()) };
-        active.insert(runnable.waker());
+        //
+        // SAFETY:
+        //
+        // If `future` is not `Send`, this must be a `LocalExecutor` as per this
+        // function's unsafe precondition. Since `LocalExecutor` is `!Sync`,
+        // `try_tick`, `tick` and `run` can only be called from the origin
+        // thread of the `LocalExecutor`. Similarly, `spawn` can only  be called
+        // from the origin thread, ensuring that `future` and the executor share
+        // the same origin thread. The `Runnable` can be scheduled from other
+        // threads, but because of the above `Runnable` can only be called or
+        // dropped on the origin thread.
+        //
+        // `future` is not `'static`, but we make sure that the `Runnable` does
+        // not outlive `'a`. When the executor is dropped, the `active` field is
+        // drained and all of the `Waker`s are woken. Then, the queue inside of
+        // the `Executor` is drained of all of its runnables. This ensures that
+        // runnables are dropped and this precondition is satisfied.
+        //
+        // `self.schedule()` is `Send`, `Sync` and `'static`, as checked below.
+        // Therefore we do not need to worry about what is done with the
+        // `Waker`.
+        let (runnable, task) = Builder::new()
+            .propagate_panic(true)
+            .spawn_unchecked(|()| future, self.schedule());
+        entry.insert(runnable.waker());
 
         runnable.schedule();
         task
@@ -232,7 +347,7 @@ impl<'a> Executor<'a> {
     /// assert_eq!(res, 6);
     /// ```
     pub async fn run<T>(&self, future: impl Future<Output = T>) -> T {
-        let runner = Runner::new(self.state());
+        let mut runner = Runner::new(self.state());
         let mut rng = fastrand::Rng::new();
 
         // A future that runs tasks forever.
@@ -252,7 +367,7 @@ impl<'a> Executor<'a> {
 
     /// Returns a function that schedules a runnable task when it gets woken up.
     fn schedule(&self) -> impl Fn(Runnable) + Send + Sync + 'static {
-        let state = self.state().clone();
+        let state = self.state_as_arc();
 
         // TODO: If possible, push into the current local queue and notify the ticker.
         move |runnable| {
@@ -261,23 +376,73 @@ impl<'a> Executor<'a> {
         }
     }
 
+    /// Returns a pointer to the inner state.
+    #[inline]
+    fn state_ptr(&self) -> *const State {
+        #[cold]
+        fn alloc_state(atomic_ptr: &AtomicPtr<State>) -> *mut State {
+            let state = Arc::new(State::new());
+            // TODO: Switch this to use cast_mut once the MSRV can be bumped past 1.65
+            let ptr = Arc::into_raw(state) as *mut State;
+            if let Err(actual) = atomic_ptr.compare_exchange(
+                std::ptr::null_mut(),
+                ptr,
+                Ordering::AcqRel,
+                Ordering::Acquire,
+            ) {
+                // SAFETY: This was just created from Arc::into_raw.
+                drop(unsafe { Arc::from_raw(ptr) });
+                actual
+            } else {
+                ptr
+            }
+        }
+
+        let mut ptr = self.state.load(Ordering::Acquire);
+        if ptr.is_null() {
+            ptr = alloc_state(&self.state);
+        }
+        ptr
+    }
+
     /// Returns a reference to the inner state.
-    fn state(&self) -> &Arc<State> {
-        self.state.get_or_init_blocking(|| Arc::new(State::new()))
+    #[inline]
+    fn state(&self) -> &State {
+        // SAFETY: So long as an Executor lives, it's state pointer will always be valid
+        // when accessed through state_ptr.
+        unsafe { &*self.state_ptr() }
+    }
+
+    // Clones the inner state Arc
+    #[inline]
+    fn state_as_arc(&self) -> Arc<State> {
+        // SAFETY: So long as an Executor lives, it's state pointer will always be a valid
+        // Arc when accessed through state_ptr.
+        let arc = unsafe { Arc::from_raw(self.state_ptr()) };
+        let clone = arc.clone();
+        std::mem::forget(arc);
+        clone
     }
 }
 
 impl Drop for Executor<'_> {
     fn drop(&mut self) {
-        if let Some(state) = self.state.get() {
-            let mut active = state.active.lock().unwrap();
-            for w in active.drain() {
-                w.wake();
-            }
-            drop(active);
-
-            while state.queue.pop().is_ok() {}
+        let ptr = *self.state.get_mut();
+        if ptr.is_null() {
+            return;
         }
+
+        // SAFETY: As ptr is not null, it was allocated via Arc::new and converted
+        // via Arc::into_raw in state_ptr.
+        let state = unsafe { Arc::from_raw(ptr) };
+
+        let mut active = state.active.lock().unwrap_or_else(|e| e.into_inner());
+        for w in active.drain() {
+            w.wake();
+        }
+        drop(active);
+
+        while state.queue.pop().is_ok() {}
     }
 }
 
@@ -375,20 +540,70 @@ impl<'a> LocalExecutor<'a> {
     pub fn spawn<T: 'a>(&self, future: impl Future<Output = T> + 'a) -> Task<T> {
         let mut active = self.inner().state().active.lock().unwrap();
 
-        // Remove the task from the set of active tasks when the future finishes.
-        let index = active.vacant_entry().key();
-        let state = self.inner().state().clone();
-        let future = async move {
-            let _guard = CallOnDrop(move || drop(state.active.lock().unwrap().try_remove(index)));
-            future.await
-        };
+        // SAFETY: This executor is not thread safe, so the future and its result
+        //         cannot be sent to another thread.
+        unsafe { self.inner().spawn_inner(future, &mut active) }
+    }
 
-        // Create the task and register it in the set of active tasks.
-        let (runnable, task) = unsafe { async_task::spawn_unchecked(future, self.schedule()) };
-        active.insert(runnable.waker());
+    /// Spawns many tasks onto the executor.
+    ///
+    /// As opposed to the [`spawn`] method, this locks the executor's inner task lock once and
+    /// spawns all of the tasks in one go. With large amounts of tasks this can improve
+    /// contention.
+    ///
+    /// It is assumed that the iterator provided does not block; blocking iterators can lock up
+    /// the internal mutex and therefore the entire executor. Unlike [`Executor::spawn`], the
+    /// mutex is not released, as there are no other threads that can poll this executor.
+    ///
+    /// ## Example
+    ///
+    /// ```
+    /// use async_executor::LocalExecutor;
+    /// use futures_lite::{stream, prelude::*};
+    /// use std::future::ready;
+    ///
+    /// # futures_lite::future::block_on(async {
+    /// let mut ex = LocalExecutor::new();
+    ///
+    /// let futures = [
+    ///     ready(1),
+    ///     ready(2),
+    ///     ready(3)
+    /// ];
+    ///
+    /// // Spawn all of the futures onto the executor at once.
+    /// let mut tasks = vec![];
+    /// ex.spawn_many(futures, &mut tasks);
+    ///
+    /// // Await all of them.
+    /// let results = ex.run(async move {
+    ///     stream::iter(tasks).then(|x| x).collect::<Vec<_>>().await
+    /// }).await;
+    /// assert_eq!(results, [1, 2, 3]);
+    /// # });
+    /// ```
+    ///
+    /// [`spawn`]: LocalExecutor::spawn
+    /// [`Executor::spawn_many`]: Executor::spawn_many
+    pub fn spawn_many<T: Send + 'a, F: Future<Output = T> + Send + 'a>(
+        &self,
+        futures: impl IntoIterator<Item = F>,
+        handles: &mut impl Extend<Task<F::Output>>,
+    ) {
+        let mut active = self.inner().state().active.lock().unwrap();
 
-        runnable.schedule();
-        task
+        // Convert all of the futures to tasks.
+        let tasks = futures.into_iter().map(|future| {
+            // SAFETY: This executor is not thread safe, so the future and its result
+            //         cannot be sent to another thread.
+            unsafe { self.inner().spawn_inner(future, &mut active) }
+
+            // As only one thread can spawn or poll tasks at a time, there is no need
+            // to release lock contention here.
+        });
+
+        // Push them to the user's collection.
+        handles.extend(tasks);
     }
 
     /// Attempts to run a task if at least one is scheduled.
@@ -454,16 +669,6 @@ impl<'a> LocalExecutor<'a> {
         self.inner().run(future).await
     }
 
-    /// Returns a function that schedules a runnable task when it gets woken up.
-    fn schedule(&self) -> impl Fn(Runnable) + Send + Sync + 'static {
-        let state = self.inner().state().clone();
-
-        move |runnable| {
-            state.queue.push(runnable).unwrap();
-            state.notify();
-        }
-    }
-
     /// Returns a reference to the inner executor.
     fn inner(&self) -> &Executor<'a> {
         &self.inner
@@ -515,7 +720,7 @@ impl State {
     fn notify(&self) {
         if self
             .notified
-            .compare_exchange(false, true, Ordering::SeqCst, Ordering::SeqCst)
+            .compare_exchange(false, true, Ordering::AcqRel, Ordering::Acquire)
             .is_ok()
         {
             let waker = self.sleepers.lock().unwrap().notify();
@@ -558,9 +763,7 @@ impl Sleepers {
     fn update(&mut self, id: usize, waker: &Waker) -> bool {
         for item in &mut self.wakers {
             if item.0 == id {
-                if !item.1.will_wake(waker) {
-                    item.1 = waker.clone();
-                }
+                item.1.clone_from(waker);
                 return false;
             }
         }
@@ -613,29 +816,26 @@ struct Ticker<'a> {
     /// 1) Woken.
     /// 2a) Sleeping and unnotified.
     /// 2b) Sleeping and notified.
-    sleeping: AtomicUsize,
+    sleeping: usize,
 }
 
 impl Ticker<'_> {
     /// Creates a ticker.
     fn new(state: &State) -> Ticker<'_> {
-        Ticker {
-            state,
-            sleeping: AtomicUsize::new(0),
-        }
+        Ticker { state, sleeping: 0 }
     }
 
     /// Moves the ticker into sleeping and unnotified state.
     ///
     /// Returns `false` if the ticker was already sleeping and unnotified.
-    fn sleep(&self, waker: &Waker) -> bool {
+    fn sleep(&mut self, waker: &Waker) -> bool {
         let mut sleepers = self.state.sleepers.lock().unwrap();
 
-        match self.sleeping.load(Ordering::SeqCst) {
+        match self.sleeping {
             // Move to sleeping state.
-            0 => self
-                .sleeping
-                .store(sleepers.insert(waker), Ordering::SeqCst),
+            0 => {
+                self.sleeping = sleepers.insert(waker);
+            }
 
             // Already sleeping, check if notified.
             id => {
@@ -647,31 +847,31 @@ impl Ticker<'_> {
 
         self.state
             .notified
-            .swap(sleepers.is_notified(), Ordering::SeqCst);
+            .store(sleepers.is_notified(), Ordering::Release);
 
         true
     }
 
     /// Moves the ticker into woken state.
-    fn wake(&self) {
-        let id = self.sleeping.swap(0, Ordering::SeqCst);
-        if id != 0 {
+    fn wake(&mut self) {
+        if self.sleeping != 0 {
             let mut sleepers = self.state.sleepers.lock().unwrap();
-            sleepers.remove(id);
+            sleepers.remove(self.sleeping);
 
             self.state
                 .notified
-                .swap(sleepers.is_notified(), Ordering::SeqCst);
+                .store(sleepers.is_notified(), Ordering::Release);
         }
+        self.sleeping = 0;
     }
 
     /// Waits for the next runnable task to run.
-    async fn runnable(&self) -> Runnable {
+    async fn runnable(&mut self) -> Runnable {
         self.runnable_with(|| self.state.queue.pop().ok()).await
     }
 
     /// Waits for the next runnable task to run, given a function that searches for a task.
-    async fn runnable_with(&self, mut search: impl FnMut() -> Option<Runnable>) -> Runnable {
+    async fn runnable_with(&mut self, mut search: impl FnMut() -> Option<Runnable>) -> Runnable {
         future::poll_fn(|cx| {
             loop {
                 match search() {
@@ -702,14 +902,13 @@ impl Ticker<'_> {
 impl Drop for Ticker<'_> {
     fn drop(&mut self) {
         // If this ticker is in sleeping state, it must be removed from the sleepers list.
-        let id = self.sleeping.swap(0, Ordering::SeqCst);
-        if id != 0 {
+        if self.sleeping != 0 {
             let mut sleepers = self.state.sleepers.lock().unwrap();
-            let notified = sleepers.remove(id);
+            let notified = sleepers.remove(self.sleeping);
 
             self.state
                 .notified
-                .swap(sleepers.is_notified(), Ordering::SeqCst);
+                .store(sleepers.is_notified(), Ordering::Release);
 
             // If this ticker was notified, then notify another ticker.
             if notified {
@@ -734,7 +933,7 @@ struct Runner<'a> {
     local: Arc<ConcurrentQueue<Runnable>>,
 
     /// Bumped every time a runnable task is found.
-    ticks: AtomicUsize,
+    ticks: usize,
 }
 
 impl Runner<'_> {
@@ -744,7 +943,7 @@ impl Runner<'_> {
             state,
             ticker: Ticker::new(state),
             local: Arc::new(ConcurrentQueue::bounded(512)),
-            ticks: AtomicUsize::new(0),
+            ticks: 0,
         };
         state
             .local_queues
@@ -755,7 +954,7 @@ impl Runner<'_> {
     }
 
     /// Waits for the next runnable task to run.
-    async fn runnable(&self, rng: &mut fastrand::Rng) -> Runnable {
+    async fn runnable(&mut self, rng: &mut fastrand::Rng) -> Runnable {
         let runnable = self
             .ticker
             .runnable_with(|| {
@@ -798,9 +997,9 @@ impl Runner<'_> {
             .await;
 
         // Bump the tick counter.
-        let ticks = self.ticks.fetch_add(1, Ordering::SeqCst);
+        self.ticks = self.ticks.wrapping_add(1);
 
-        if ticks % 64 == 0 {
+        if self.ticks % 64 == 0 {
             // Steal tasks from the global queue to ensure fair task scheduling.
             steal(&self.state.queue, &self.local);
         }
@@ -850,21 +1049,24 @@ fn steal<T>(src: &ConcurrentQueue<T>, dest: &ConcurrentQueue<T>) {
 /// Debug implementation for `Executor` and `LocalExecutor`.
 fn debug_executor(executor: &Executor<'_>, name: &str, f: &mut fmt::Formatter<'_>) -> fmt::Result {
     // Get a reference to the state.
-    let state = match executor.state.get() {
-        Some(state) => state,
-        None => {
-            // The executor has not been initialized.
-            struct Uninitialized;
+    let ptr = executor.state.load(Ordering::Acquire);
+    if ptr.is_null() {
+        // The executor has not been initialized.
+        struct Uninitialized;
 
-            impl fmt::Debug for Uninitialized {
-                fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result {
-                    f.write_str("<uninitialized>")
-                }
+        impl fmt::Debug for Uninitialized {
+            fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result {
+                f.write_str("<uninitialized>")
             }
-
-            return f.debug_tuple(name).field(&Uninitialized).finish();
         }
-    };
+
+        return f.debug_tuple(name).field(&Uninitialized).finish();
+    }
+
+    // SAFETY: If the state pointer is not null, it must have been
+    // allocated properly by Arc::new and converted via Arc::into_raw
+    // in state_ptr.
+    let state = unsafe { &*ptr };
 
     /// Debug wrapper for the number of active tasks.
     struct ActiveTasks<'a>(&'a Mutex<Slab<Waker>>);
@@ -930,6 +1132,7 @@ fn _ensure_send_and_sync() {
 
     fn is_send<T: Send>(_: T) {}
     fn is_sync<T: Sync>(_: T) {}
+    fn is_static<T: 'static>(_: T) {}
 
     is_send::<Executor<'_>>(Executor::new());
     is_sync::<Executor<'_>>(Executor::new());
@@ -939,6 +1142,9 @@ fn _ensure_send_and_sync() {
     is_sync(ex.run(pending::<()>()));
     is_send(ex.tick());
     is_sync(ex.tick());
+    is_send(ex.schedule());
+    is_sync(ex.schedule());
+    is_static(ex.schedule());
 
     /// ```compile_fail
     /// use async_executor::LocalExecutor;

tests/drop.rs

@@ -121,6 +121,20 @@ fn drop_finished_task_and_then_drop_executor() {
     assert_eq!(DROP.load(Ordering::SeqCst), 1);
 }
 
+#[test]
+fn iterator_panics_mid_run() {
+    let ex = Executor::new();
+
+    let panic = std::panic::catch_unwind(|| {
+        let mut handles = vec![];
+        ex.spawn_many(
+            (0..50).map(|i| if i == 25 { panic!() } else { future::ready(i) }),
+            &mut handles,
+        )
+    });
+    assert!(panic.is_err());
+}
+
 struct CallOnDrop<F: Fn()>(F);
 
 impl<F: Fn()> Drop for CallOnDrop<F> {

tests/larger_tasks.rs

@@ -0,0 +1,99 @@
+//! Test for larger tasks.
+
+use async_executor::Executor;
+use futures_lite::future::{self, block_on};
+use futures_lite::prelude::*;
+
+use std::sync::Arc;
+use std::thread;
+use std::time::Duration;
+
+fn do_run<Fut: Future<Output = ()>>(mut f: impl FnMut(Arc<Executor<'static>>) -> Fut) {
+    // This should not run for longer than two minutes.
+    #[cfg(not(miri))]
+    let _stop_timeout = {
+        let (stop_timeout, stopper) = async_channel::bounded::<()>(1);
+        thread::spawn(move || {
+            block_on(async move {
+                let timeout = async {
+                    async_io::Timer::after(Duration::from_secs(2 * 60)).await;
+                    eprintln!("test timed out after 2m");
+                    std::process::exit(1)
+                };
+
+                let _ = stopper.recv().or(timeout).await;
+            })
+        });
+        stop_timeout
+    };
+
+    let ex = Arc::new(Executor::new());
+
+    // Test 1: Use the `run` command.
+    block_on(ex.run(f(ex.clone())));
+
+    // Test 2: Loop on `tick`.
+    block_on(async {
+        let ticker = async {
+            loop {
+                ex.tick().await;
+            }
+        };
+
+        f(ex.clone()).or(ticker).await
+    });
+
+    // Test 3: Run on many threads.
+    thread::scope(|scope| {
+        let (_signal, shutdown) = async_channel::bounded::<()>(1);
+
+        for _ in 0..16 {
+            let shutdown = shutdown.clone();
+            let ex = &ex;
+            scope.spawn(move || block_on(ex.run(shutdown.recv())));
+        }
+
+        block_on(f(ex.clone()));
+    });
+
+    // Test 4: Tick loop on many threads.
+    thread::scope(|scope| {
+        let (_signal, shutdown) = async_channel::bounded::<()>(1);
+
+        for _ in 0..16 {
+            let shutdown = shutdown.clone();
+            let ex = &ex;
+            scope.spawn(move || {
+                block_on(async move {
+                    let ticker = async {
+                        loop {
+                            ex.tick().await;
+                        }
+                    };
+
+                    shutdown.recv().or(ticker).await
+                })
+            });
+        }
+
+        block_on(f(ex.clone()));
+    });
+}
+
+#[test]
+fn smoke() {
+    do_run(|ex| async move { ex.spawn(async {}).await });
+}
+
+#[test]
+fn yield_now() {
+    do_run(|ex| async move { ex.spawn(future::yield_now()).await })
+}
+
+#[test]
+fn timer() {
+    do_run(|ex| async move {
+        ex.spawn(async_io::Timer::after(Duration::from_millis(5)))
+            .await;
+    })
+}

tests/panic_prop.rs

@@ -0,0 +1,14 @@
+use async_executor::Executor;
+use futures_lite::{future, prelude::*};
+
+#[test]
+fn test_panic_propagation() {
+    let ex = Executor::new();
+    let task = ex.spawn(async { panic!("should be caught by the task") });
+
+    // Running the executor should not panic.
+    assert!(ex.try_tick());
+
+    // Polling the task should.
+    assert!(future::block_on(task.catch_unwind()).is_err());
+}

tests/spawn_many.rs

@@ -0,0 +1,45 @@
+use async_executor::{Executor, LocalExecutor};
+use futures_lite::future;
+
+#[cfg(not(miri))]
+const READY_COUNT: usize = 50_000;
+#[cfg(miri)]
+const READY_COUNT: usize = 505;
+
+#[test]
+fn spawn_many() {
+    future::block_on(async {
+        let ex = Executor::new();
+
+        // Spawn a lot of tasks.
+        let mut tasks = vec![];
+        ex.spawn_many((0..READY_COUNT).map(future::ready), &mut tasks);
+
+        // Run all of the tasks in parallel.
+        ex.run(async move {
+            for (i, task) in tasks.into_iter().enumerate() {
+                assert_eq!(task.await, i);
+            }
+        })
+        .await;
+    });
+}
+
+#[test]
+fn spawn_many_local() {
+    future::block_on(async {
+        let ex = LocalExecutor::new();
+
+        // Spawn a lot of tasks.
+        let mut tasks = vec![];
+        ex.spawn_many((0..READY_COUNT).map(future::ready), &mut tasks);
+
+        // Run all of the tasks in parallel.
+        ex.run(async move {
+            for (i, task) in tasks.into_iter().enumerate() {
+                assert_eq!(task.await, i);
+            }
+        })
+        .await;
+    });
+}