diff --git a/.github/workflows/ci.yml b/.github/workflows/ci.yml new file mode 100644 index 0000000..1a8ceda --- /dev/null +++ b/.github/workflows/ci.yml @@ -0,0 +1,97 @@ +name: CI + +on: + pull_request: + push: + branches: [master] + +env: + CARGO_TERM_COLOR: always + +jobs: + fmt: + name: rustfmt + runs-on: ubuntu-latest + steps: + - uses: actions/checkout@v5 + - uses: dtolnay/rust-toolchain@stable + with: + components: rustfmt + - run: cargo fmt --check + - run: cargo fmt --manifest-path nostd-test/Cargo.toml --check + + clippy: + name: clippy + runs-on: ubuntu-latest + steps: + - uses: actions/checkout@v5 + - uses: dtolnay/rust-toolchain@stable + with: + components: clippy + targets: thumbv7m-none-eabi + - run: cargo clippy --all-targets -- -D warnings + - run: cargo clippy --manifest-path nostd-test/Cargo.toml --target thumbv7m-none-eabi --all-targets -- -D warnings + + test: + name: test, doc, examples + runs-on: ubuntu-latest + steps: + - uses: actions/checkout@v5 + - uses: dtolnay/rust-toolchain@stable + - run: cargo test -- --test-threads=1 + - run: cargo doc --no-deps + - run: cargo run --example helloworld + - run: cargo run --example futures_chain + - run: cargo run --example multi_async_std + - run: cargo run --example signal + + msrv: + name: msrv 1.63 + runs-on: ubuntu-latest + steps: + - uses: actions/checkout@v5 + - uses: dtolnay/rust-toolchain@1.63.0 + with: + targets: wasm32-unknown-unknown,thumbv7m-none-eabi + - name: library and no_std targets on 1.63 + run: | + # `cargo test` also builds the async-std example stack, whose current + # crates need a newer rustc. The library itself does not. + cargo build --lib + cargo check --target wasm32-unknown-unknown --lib + cargo check --target thumbv7m-none-eabi --lib + + nostd: + name: no_std build and qemu + runs-on: ubuntu-latest + steps: + - uses: actions/checkout@v5 + - uses: dtolnay/rust-toolchain@stable + with: + targets: wasm32-unknown-unknown,thumbv7m-none-eabi + - name: install qemu + run: sudo apt-get update && sudo apt-get install -y qemu-system-arm + - name: check library on no_std targets + run: | + cargo check --target wasm32-unknown-unknown --lib + cargo check --target thumbv7m-none-eabi --lib + - name: run bare-metal tests under qemu + run: | + cargo build --manifest-path nostd-test/Cargo.toml --release --target thumbv7m-none-eabi + qemu-system-arm \ + -cpu cortex-m3 \ + -machine lm3s6965evb \ + -nographic \ + -semihosting-config enable=on,target=native \ + -kernel nostd-test/target/thumbv7m-none-eabi/release/executor-nostd-test + + miri: + name: miri + runs-on: ubuntu-latest + steps: + - uses: actions/checkout@v5 + - uses: dtolnay/rust-toolchain@nightly + with: + components: miri + - run: cargo miri setup + - run: cargo miri test -- --test-threads=1 diff --git a/Cargo.toml b/Cargo.toml index 5249895..9b7469b 100644 --- a/Cargo.toml +++ b/Cargo.toml @@ -13,7 +13,11 @@ rust-version="1.63" # See more keys and their definitions at https://doc.rust-lang.org/cargo/reference/manifest.html [dependencies] -spin = "0" +# 0.9.9 is the newest `spin` that still builds on the declared MSRV (1.63). +# Newer 0.x releases pull `lock_api` versions that need Rust 1.71+. +# Only the raw spin mutex is used; `lock_api` stays off so that dependency +# cannot bump the MSRV. +spin = { version = "0.9.9", default-features = false, features = ["mutex", "spin_mutex", "lazy"] } [dev-dependencies] async-std = "1" diff --git a/README.md b/README.md index 2e5276e..fada340 100644 --- a/README.md +++ b/README.md @@ -56,6 +56,47 @@ fn main() { } ``` +## Testing / platform support + +Host tests cover completion, `is_done`, nested `add_async`, `wake` and `wake_by_ref` (including wakes from another thread), waking during poll, drop of finished tasks, and a short many-wake stress run. + +```bash +cargo test -- --test-threads=1 +cargo clippy --all-targets -- -D warnings +cargo fmt --check +cargo doc --no-deps +cargo +nightly miri test -- --test-threads=1 +``` + +The library builds on the declared MSRV, Rust 1.63: + +```bash +cargo +1.63.0 build --lib +cargo +1.63.0 check --target wasm32-unknown-unknown --lib +cargo +1.63.0 check --target thumbv7m-none-eabi --lib +``` + +Behavior tests and the `async-std` examples run on stable. Current `async-std` releases pull crates that need a newer compiler than 1.63; the library does not. + +`no_std` + `alloc` needs pointer-width atomics (`alloc::sync::Arc` and `spin::Mutex`). Checked targets: + +| Target | Support | +| --- | --- | +| `wasm32-unknown-unknown` | `cargo check --target wasm32-unknown-unknown --lib` | +| `thumbv7m-none-eabi` | builds, and `nostd-test` actually runs under QEMU | +| `thumbv6m-none-eabi` (Cortex-M0) | not supported: no pointer-width atomics, so `Arc` is not in `alloc` | + +The QEMU harness is a `#![no_std]` `#![no_main]` binary with a bump allocator. It checks `is_done`, a task that completes and is dropped, a task spawned from another task, a self-wake, several tasks, and a `u32` output. Nothing from `std` is linked. + +```bash +cargo build --manifest-path nostd-test/Cargo.toml --release --target thumbv7m-none-eabi +qemu-system-arm -cpu cortex-m3 -machine lm3s6965evb -nographic \ + -semihosting-config enable=on,target=native \ + -kernel nostd-test/target/thumbv7m-none-eabi/release/executor-nostd-test +``` + +Do not call `update` or `update_woken` from inside a task. Both take a `spin::Mutex` that is not reentrant. Spawn with `add_async`, and wake the waker you were given; the queues exist so that is safe during `update`. + # License This project is licensed under either of diff --git a/examples/futures_chain.rs b/examples/futures_chain.rs index dc9268c..c38f1f1 100644 --- a/examples/futures_chain.rs +++ b/examples/futures_chain.rs @@ -8,7 +8,7 @@ async fn async_sleep(print_name: &str, time: Duration) { println!("{print_name}: after sleep {sleep_time} ms"); } -fn main() -> () { +fn main() { println!("hello"); executor::add_async(async { diff --git a/examples/helloworld.rs b/examples/helloworld.rs index 6a82480..7ed8792 100644 --- a/examples/helloworld.rs +++ b/examples/helloworld.rs @@ -21,7 +21,7 @@ async fn a() { println!("goodbye world"); } -fn main() -> () { +fn main() { executor::add_async(async { a().await; }); diff --git a/examples/signal.rs b/examples/signal.rs index ab3dc2d..a64c4b9 100644 --- a/examples/signal.rs +++ b/examples/signal.rs @@ -40,7 +40,7 @@ fn receive_signal_from_somewhere(state: Arc>) { }); } -fn main() -> () { +fn main() { let state = Arc::new(Mutex::new(AsyncSignalWaiterState { ready: false, waker: None, diff --git a/nostd-test/Cargo.toml b/nostd-test/Cargo.toml new file mode 100644 index 0000000..aa24fe4 --- /dev/null +++ b/nostd-test/Cargo.toml @@ -0,0 +1,23 @@ +[package] +name = "executor-nostd-test" +version = "0.0.0" +edition = "2021" +publish = false +description = "Bare-metal no_std tests for the executor crate" + +[dependencies] +executor = { path = ".." } + +[[bin]] +name = "executor-nostd-test" +path = "src/main.rs" +test = false +bench = false + +[profile.dev] +panic = "abort" + +[profile.release] +panic = "abort" +lto = false +codegen-units = 1 diff --git a/nostd-test/build.rs b/nostd-test/build.rs new file mode 100644 index 0000000..4a44164 --- /dev/null +++ b/nostd-test/build.rs @@ -0,0 +1,5 @@ +fn main() { + let manifest = std::env::var("CARGO_MANIFEST_DIR").unwrap(); + println!("cargo:rustc-link-arg=-T{manifest}/link.ld"); + println!("cargo:rerun-if-changed=link.ld"); +} diff --git a/nostd-test/link.ld b/nostd-test/link.ld new file mode 100644 index 0000000..32e8754 --- /dev/null +++ b/nostd-test/link.ld @@ -0,0 +1,47 @@ +/* LM3S6965 (QEMU lm3s6965evb): 256K flash at 0x0, 64K SRAM at 0x20000000. */ +MEMORY +{ + FLASH : ORIGIN = 0x00000000, LENGTH = 256K + RAM : ORIGIN = 0x20000000, LENGTH = 64K +} + +ENTRY(Reset) + +SECTIONS +{ + .vector_table ORIGIN(FLASH) : + { + LONG(ORIGIN(RAM) + LENGTH(RAM)); + KEEP(*(.vector_table.reset)); + KEEP(*(.vector_table.fault)); + } > FLASH + + .text : + { + *(.text .text.*); + *(.rodata .rodata.*); + } > FLASH + + .data : ALIGN(4) + { + _sdata = .; + *(.data .data.*); + _edata = .; + } > RAM AT > FLASH + _sidata = LOADADDR(.data); + + .bss : ALIGN(4) + { + _sbss = .; + *(.bss .bss.*); + *(COMMON); + _ebss = .; + } > RAM + + /DISCARD/ : + { + *(.ARM.exidx*); + *(.ARM.extab*); + *(.eh_frame*); + } +} diff --git a/nostd-test/src/main.rs b/nostd-test/src/main.rs new file mode 100644 index 0000000..4527671 --- /dev/null +++ b/nostd-test/src/main.rs @@ -0,0 +1,239 @@ +//! Runs without `std`: global allocator, `add_async` / `update` / `is_done`. +//! +//! Build and run under QEMU: +//! `cargo build --manifest-path nostd-test/Cargo.toml --release --target thumbv7m-none-eabi` +//! `qemu-system-arm -cpu cortex-m3 -machine lm3s6965evb -nographic -semihosting-config enable=on,target=native -kernel nostd-test/target/thumbv7m-none-eabi/release/executor-nostd-test` + +#![no_std] +#![no_main] + +extern crate alloc; + +use core::alloc::{GlobalAlloc, Layout}; +use core::future::Future; +use core::panic::PanicInfo; +use core::pin::Pin; +use core::sync::atomic::{AtomicUsize, Ordering}; +use core::task::{Context, Poll}; + +const HEAP_SIZE: usize = 16 * 1024; + +#[repr(C, align(16))] +struct Heap(core::cell::UnsafeCell<[u8; HEAP_SIZE]>); + +unsafe impl Sync for Heap {} + +static HEAP: Heap = Heap(core::cell::UnsafeCell::new([0; HEAP_SIZE])); +static HEAP_NEXT: AtomicUsize = AtomicUsize::new(0); + +struct Bump; + +unsafe impl GlobalAlloc for Bump { + unsafe fn alloc(&self, layout: Layout) -> *mut u8 { + let align = layout.align().max(1); + let size = layout.size(); + let base = HEAP.0.get() as usize; + loop { + let current = HEAP_NEXT.load(Ordering::Relaxed); + let start = (current + align - 1) & !(align - 1); + let end = start.saturating_add(size); + if end > HEAP_SIZE { + return core::ptr::null_mut(); + } + if HEAP_NEXT + .compare_exchange(current, end, Ordering::Relaxed, Ordering::Relaxed) + .is_ok() + { + return (base + start) as *mut u8; + } + } + } + + unsafe fn dealloc(&self, _ptr: *mut u8, _layout: Layout) {} +} + +#[global_allocator] +static ALLOCATOR: Bump = Bump; + +static DROPS: AtomicUsize = AtomicUsize::new(0); +static HITS: AtomicUsize = AtomicUsize::new(0); + +struct Bomb; + +impl Drop for Bomb { + fn drop(&mut self) { + DROPS.fetch_add(1, Ordering::Relaxed); + } +} + +struct SelfWake { + polls: u8, +} + +impl Future for SelfWake { + type Output = (); + + fn poll(mut self: Pin<&mut Self>, cx: &mut Context<'_>) -> Poll<()> { + self.polls = self.polls.wrapping_add(1); + if self.polls == 1 { + cx.waker().wake_by_ref(); + Poll::Pending + } else { + Poll::Ready(()) + } + } +} + +extern "C" { + static mut _sbss: u8; + static mut _ebss: u8; + static mut _sdata: u8; + static mut _edata: u8; + static _sidata: u8; +} + +/// Hardware reset entry. +/// +/// # Safety +/// +/// The CPU must enter here once, with flash and RAM mapped as in `link.ld`, +/// before any other Rust code runs. +#[no_mangle] +pub unsafe extern "C" fn Reset() -> ! { + init_ram(); + if run() { + write0("executor nostd tests ok\n"); + exit(0); + } else { + write0("executor nostd tests FAILED\n"); + exit(1); + } +} + +unsafe fn init_ram() { + let mut bss = core::ptr::addr_of_mut!(_sbss); + let bss_end = core::ptr::addr_of_mut!(_ebss); + while bss < bss_end { + bss.write_volatile(0); + bss = bss.add(1); + } + + let mut data = core::ptr::addr_of_mut!(_sdata); + let data_end = core::ptr::addr_of_mut!(_edata); + let mut src = core::ptr::addr_of!(_sidata); + while data < data_end { + data.write_volatile(src.read_volatile()); + data = data.add(1); + src = src.add(1); + } +} + +fn run() -> bool { + if !executor::is_done() { + return false; + } + + let before = DROPS.load(Ordering::Relaxed); + executor::add_async(async { + let _bomb = Bomb; + }); + if executor::is_done() { + return false; + } + executor::update(); + if !executor::is_done() || DROPS.load(Ordering::Relaxed) != before + 1 { + return false; + } + + executor::add_async(async { + executor::add_async(async { + HITS.fetch_add(1, Ordering::Relaxed); + }); + HITS.fetch_add(1, Ordering::Relaxed); + }); + drive(4); + if HITS.load(Ordering::Relaxed) != 2 || !executor::is_done() { + return false; + } + + executor::add_async(SelfWake { polls: 0 }); + drive(4); + if !executor::is_done() { + return false; + } + + HITS.store(0, Ordering::Relaxed); + for _ in 0..8 { + executor::add_async(async { + HITS.fetch_add(1, Ordering::Relaxed); + }); + } + drive(4); + if HITS.load(Ordering::Relaxed) != 8 || !executor::is_done() { + return false; + } + + HITS.store(0, Ordering::Relaxed); + executor::add_async(async { + let value: u32 = async { 42u32 }.await; + HITS.store(value as usize, Ordering::Relaxed); + value + }); + drive(4); + HITS.load(Ordering::Relaxed) == 42 && executor::is_done() +} + +fn drive(limit: usize) { + for _ in 0..limit { + if executor::is_done() { + return; + } + executor::update(); + } +} + +fn write0(text: &str) { + let mut buf = [0u8; 80]; + let len = text.len().min(buf.len() - 1); + buf[..len].copy_from_slice(&text.as_bytes()[..len]); + unsafe { + semihost(0x04, buf.as_ptr()); + } +} + +fn exit(status: u32) -> ! { + let block = [0x20026u32, status]; + loop { + unsafe { + semihost(0x20, block.as_ptr().cast()); + } + } +} + +unsafe fn semihost(op: u32, arg: *const u8) { + core::arch::asm!( + "bkpt #0xAB", + in("r0") op, + in("r1") arg, + options(nostack) + ); +} + +#[panic_handler] +fn panic(_info: &PanicInfo) -> ! { + write0("panic\n"); + exit(1); +} + +unsafe extern "C" fn fault() -> ! { + write0("fault\n"); + exit(1); +} + +#[link_section = ".vector_table.reset"] +#[no_mangle] +pub static RESET_VECTOR: unsafe extern "C" fn() -> ! = Reset; + +#[link_section = ".vector_table.fault"] +#[no_mangle] +pub static FAULT_VECTORS: [unsafe extern "C" fn() -> !; 2] = [fault, fault]; diff --git a/src/lib.rs b/src/lib.rs index 57f706e..0598c0d 100644 --- a/src/lib.rs +++ b/src/lib.rs @@ -9,7 +9,7 @@ use { sync::atomic::Ordering, task::{Context, Poll, Waker}, }, - spin::Mutex, + spin::{Lazy, Mutex}, }; type TasksList = VecDeque>; @@ -19,11 +19,6 @@ type Future = Pin + Send + 'static>>; struct Executor { /// Tasks collection. Use [`update()`] functon for polling all tasks and continue them progress. tasks: TasksList, - - /// Woken Tasks collection. Contain woken tasks from [`Executor::tasks`]. - /// [`update()`] function aslo can continue progress of this tasks, - /// but [`update_woken()`] function will be preferred, because poll only tasks, that was awakened. - woken_tasks: &'static Mutex, } /// [`Task`] interface for executor @@ -44,10 +39,6 @@ trait Pendable { /// /// Task is our unit of execution and holds a future are waiting on. struct Task { - /// Awaked tasks collection. - /// - /// Shall contain itself to this, when was awaked by ['wake() or wake_by_ref()']. - woken_tasks: &'static Mutex, future: Mutex>, /// Returns `true` if the state is corresponding to [`core::task::Poll::Ready`] otherwise - false. /// @@ -60,11 +51,12 @@ struct Task { // `wake_by_ref` clones the `Arc` and calls this. impl Wake for Task { fn wake(self: Arc) { - let woken_tasks = self.woken_tasks; - // its tempting to call update() "in place", but dont do this for 2 reason: - // 1) for some reason, somethimes cant poll our future, exactly after big latency between update's(), for example, because of sleep() call. - // 2) if call wake() or wake_by_ref() at poll, being at woken_tasks, will produce dead lock state. - woken_tasks.lock().push_back(Box::new(self)); + // Its tempting to call update() here, but dont: + // 1) a wake can arrive long after the last poll (sleep, another thread). + // 2) spin::Mutex is not reentrant. Polling while `update` / `update_woken` + // is already on the stack can lock this same queue. Those functions + // drain the queue and drop the guard before they poll. + WOKEN_TASK_QUEUE.lock().push_back(Box::new(self)); } } @@ -102,8 +94,8 @@ impl Executor { /// Add task for a future to the list of tasks. fn add_asyncs_from_buffer(&mut self) { let mut input_queue = INPUT_TASK_QUEUE.lock(); - while input_queue.len() > 0 { - self.tasks.push_back(input_queue.pop_front().unwrap()); + while let Some(task) = input_queue.pop_front() { + self.tasks.push_back(task); } } @@ -114,7 +106,6 @@ impl Executor { T: Send + 'static, { let task = Arc::new(Task { - woken_tasks: &WOKEN_TASK_QUEUE, future: Mutex::new(future), done: AtomicBool::new(false), }); @@ -129,39 +120,48 @@ impl Executor { self.add_asyncs_from_buffer(); for _ in 0..self.tasks.len() { let task = self.tasks.pop_front().unwrap(); + if task.is_done() { + continue; + } + task.update(); if !task.is_done() { - task.update(); self.tasks.push_back(task); } } } - /// Polls all pending tasks on global executor and remove completed tasks. + /// Polls tasks queued by a waker. Does not remove them from the main list. /// - /// When all tasks will done and we add new task and run them, old completed tasks will be removed from [`Executor::tasks`]. + /// Completed tasks stay in [`Executor::tasks`] until [`Executor::update`] drops them. fn update_woken(&self) { - let mut woken_tasks = self.woken_tasks.lock(); - while !woken_tasks.is_empty() { - woken_tasks.pop_front().unwrap().as_mut().update(); - } + poll_woken_queue(); } } -static DEFAULT_EXECUTOR: Mutex = Mutex::new(Executor { - woken_tasks: &WOKEN_TASK_QUEUE, - tasks: VecDeque::new(), -}); +fn new_executor() -> Mutex { + Mutex::new(Executor { + tasks: VecDeque::new(), + }) +} + +// `VecDeque::new` is not const on the declared MSRV (1.63), so the queues are +// built on first use. `Lazy` is still `no_std`. +static DEFAULT_EXECUTOR: Lazy> = Lazy::new(new_executor); + +fn new_task_queue() -> Mutex { + Mutex::new(VecDeque::new()) +} /// Its tempts to add futuures to executor dirctly, without global container, /// but if we will try add new future, during [`update()`], /// will produse dead lock state, because [`update()`] already lock executor. -static INPUT_TASK_QUEUE: Mutex = Mutex::new(VecDeque::new()); +static INPUT_TASK_QUEUE: Lazy> = Lazy::new(new_task_queue); /// Its tempting to have this container internally, /// but when we will add new tasks, for getting reference on this container, /// we will have to lock executor, /// that will dead lock our program if it perform during [`update()`] function, that aslo lock executor. -static WOKEN_TASK_QUEUE: Mutex = Mutex::new(VecDeque::new()); +static WOKEN_TASK_QUEUE: Lazy> = Lazy::new(new_task_queue); /// Polls all pending tasks on global executor and remove completed tasks. pub fn update() { @@ -170,9 +170,23 @@ pub fn update() { /// Polls all awaked tasks on global executor. pub fn update_woken() { - let mut woken_tasks = WOKEN_TASK_QUEUE.lock(); - while !woken_tasks.is_empty() { - woken_tasks.pop_front().unwrap().as_mut().update(); + poll_woken_queue(); +} + +/// Drain the woken queue, release it, then poll. +/// +/// The guard is dropped before any poll so a `wake` from inside a task can +/// enqueue again. `spin::Mutex` is not reentrant. Tasks queued by that wake +/// wait for the next `update_woken`. Tasks already marked done are not polled. +fn poll_woken_queue() { + let batch = { + let mut woken_tasks = WOKEN_TASK_QUEUE.lock(); + core::mem::take(&mut *woken_tasks) + }; + for task in batch { + if !task.is_done() { + task.update(); + } } } @@ -182,7 +196,6 @@ where T: Send + 'static, { let task = Arc::new(Task { - woken_tasks: &WOKEN_TASK_QUEUE, future: Mutex::new(Box::pin(future)), done: AtomicBool::new(false), }); diff --git a/tests/behavior.rs b/tests/behavior.rs new file mode 100644 index 0000000..9259871 --- /dev/null +++ b/tests/behavior.rs @@ -0,0 +1,609 @@ +//! Behavior of the global executor. +//! +//! The executor is process-global, so every test holds `TEST_LOCK` and leaves +//! the queues empty. Run with `--test-threads=1` as well; the lock is what +//! actually keeps the tests from stepping on each other. + +use core::pin::Pin; +use core::sync::atomic::{AtomicBool, AtomicUsize, Ordering}; +use core::task::{Context, Poll, Waker}; +use std::cell::Cell; +use std::future::Future; +use std::sync::{Arc, Mutex}; +use std::thread; +use std::time::{Duration, Instant}; + +static TEST_LOCK: Mutex<()> = Mutex::new(()); + +fn lock_executor() -> std::sync::MutexGuard<'static, ()> { + let guard = TEST_LOCK.lock().unwrap_or_else(|err| err.into_inner()); + assert!( + executor::is_done(), + "executor was not empty; a previous test left tasks behind" + ); + guard +} + +fn drive(limit: usize) { + for _ in 0..limit { + if executor::is_done() { + return; + } + executor::update(); + } + assert!( + executor::is_done(), + "executor did not finish within {limit} updates" + ); +} + +/// Run `drive` off the test thread so a non-reentrant spin lock cannot hang CI. +fn drive_within(limit: usize, timeout: Duration) { + let (tx, rx) = std::sync::mpsc::channel(); + thread::spawn(move || { + let result = std::panic::catch_unwind(std::panic::AssertUnwindSafe(|| drive(limit))); + let _ = tx.send(result); + }); + match rx.recv_timeout(timeout) { + Ok(Ok(())) => {} + Ok(Err(payload)) => std::panic::resume_unwind(payload), + Err(std::sync::mpsc::RecvTimeoutError::Timeout) => { + panic!("update deadlocked or did not finish within {timeout:?}") + } + Err(std::sync::mpsc::RecvTimeoutError::Disconnected) => { + panic!("driver thread exited without a result") + } + } +} + +struct Slot { + ready: bool, + waker: Option, +} + +struct Wait { + slot: Arc>, +} + +impl Future for Wait { + type Output = (); + + fn poll(self: Pin<&mut Self>, cx: &mut Context<'_>) -> Poll<()> { + let mut slot = self.slot.lock().unwrap(); + if slot.ready { + Poll::Ready(()) + } else { + slot.waker = Some(cx.waker().clone()); + Poll::Pending + } + } +} + +fn fresh_slot() -> Arc> { + Arc::new(Mutex::new(Slot { + ready: false, + waker: None, + })) +} + +/// Poll once so the task can store its waker, and return that waker. +fn arm(slot: &Arc>) -> Waker { + let task_slot = Arc::clone(slot); + executor::add_async(async move { + Wait { slot: task_slot }.await; + }); + executor::update(); + // The task is pending, so the executor is not done yet. + let waker = slot.lock().unwrap().waker.clone().expect("waker stored"); + waker +} + +struct PollCount { + polls: Arc, +} + +impl Future for PollCount { + type Output = (); + + fn poll(self: Pin<&mut Self>, _cx: &mut Context<'_>) -> Poll<()> { + self.polls.fetch_add(1, Ordering::SeqCst); + Poll::Ready(()) + } +} + +struct Gate { + polls: Arc, + go: Arc, +} + +impl Future for Gate { + type Output = (); + + fn poll(self: Pin<&mut Self>, _cx: &mut Context<'_>) -> Poll<()> { + self.polls.fetch_add(1, Ordering::SeqCst); + if self.go.load(Ordering::SeqCst) { + Poll::Ready(()) + } else { + Poll::Pending + } + } +} + +struct Steps { + left: u32, + polls: Arc, +} + +impl Future for Steps { + type Output = (); + + fn poll(mut self: Pin<&mut Self>, cx: &mut Context<'_>) -> Poll<()> { + self.polls.fetch_add(1, Ordering::SeqCst); + if self.left == 0 { + return Poll::Ready(()); + } + self.left -= 1; + cx.waker().wake_by_ref(); + if self.left == 0 { + Poll::Ready(()) + } else { + Poll::Pending + } + } +} + +struct WakeAndSpawn { + polls: u8, + spawned: Arc, +} + +impl Future for WakeAndSpawn { + type Output = (); + + fn poll(mut self: Pin<&mut Self>, cx: &mut Context<'_>) -> Poll<()> { + if self.polls == 0 { + self.polls = 1; + cx.waker().wake_by_ref(); + let spawned = Arc::clone(&self.spawned); + executor::add_async(async move { + spawned.store(true, Ordering::SeqCst); + }); + return Poll::Pending; + } + Poll::Ready(()) + } +} + +struct DropCount(Arc); + +impl Drop for DropCount { + fn drop(&mut self) { + self.0.fetch_add(1, Ordering::SeqCst); + } +} + +#[test] +fn empty_executor_is_done_and_update_is_a_noop() { + let _guard = lock_executor(); + assert!(executor::is_done()); + executor::update(); + executor::update_woken(); + assert!(executor::is_done()); +} + +#[test] +fn ready_task_runs_and_is_done_becomes_true() { + let _guard = lock_executor(); + let ran = Arc::new(AtomicBool::new(false)); + let flag = Arc::clone(&ran); + executor::add_async(async move { + flag.store(true, Ordering::SeqCst); + }); + assert!(!executor::is_done(), "queued task is not done yet"); + executor::update(); + assert!(ran.load(Ordering::SeqCst)); + assert!(executor::is_done()); +} + +#[test] +fn many_tasks_all_complete() { + let _guard = lock_executor(); + let count = Arc::new(AtomicUsize::new(0)); + for _ in 0..64 { + let count = Arc::clone(&count); + executor::add_async(async move { + count.fetch_add(1, Ordering::SeqCst); + }); + } + assert!(!executor::is_done()); + drive(8); + assert_eq!(count.load(Ordering::SeqCst), 64); + assert!(executor::is_done()); +} + +#[test] +fn task_spawned_from_inside_a_task_is_picked_up() { + let _guard = lock_executor(); + let phase = Arc::new(AtomicUsize::new(0)); + let parent = Arc::clone(&phase); + let child = Arc::clone(&phase); + let grand = Arc::clone(&phase); + executor::add_async(async move { + executor::add_async(async move { + executor::add_async(async move { + grand.fetch_add(1, Ordering::SeqCst); + }); + child.fetch_add(1, Ordering::SeqCst); + }); + parent.fetch_add(1, Ordering::SeqCst); + }); + drive(8); + assert_eq!(phase.load(Ordering::SeqCst), 3); + assert!(executor::is_done()); +} + +fn wake_from_thread(waker: Waker, by_ref: bool) { + thread::spawn(move || { + if by_ref { + waker.wake_by_ref(); + } else { + waker.wake(); + } + }) + .join() + .unwrap(); +} + +#[test] +fn wake_by_ref_from_another_thread_completes_via_update_woken() { + let _guard = lock_executor(); + let ran = Arc::new(AtomicBool::new(false)); + let flag = Arc::clone(&ran); + let slot = fresh_slot(); + let slot_for_task = Arc::clone(&slot); + executor::add_async(async move { + Wait { + slot: slot_for_task, + } + .await; + flag.store(true, Ordering::SeqCst); + }); + executor::update(); + let waker = slot.lock().unwrap().waker.clone().expect("waker stored"); + + wake_from_thread(waker, true); + slot.lock().unwrap().ready = true; + executor::update_woken(); + assert!( + ran.load(Ordering::SeqCst), + "update_woken polls the woken task through to the code after await" + ); + assert!( + !executor::is_done(), + "the finished task stays on the main list until update drops it" + ); + executor::update(); + assert!(executor::is_done()); +} + +#[test] +fn consuming_wake_from_another_thread_completes() { + let _guard = lock_executor(); + let ran = Arc::new(AtomicBool::new(false)); + let flag = Arc::clone(&ran); + let slot = fresh_slot(); + let slot_for_task = Arc::clone(&slot); + executor::add_async(async move { + Wait { + slot: slot_for_task, + } + .await; + flag.store(true, Ordering::SeqCst); + }); + executor::update(); + let waker = slot.lock().unwrap().waker.take().expect("waker stored"); + + wake_from_thread(waker, false); + slot.lock().unwrap().ready = true; + executor::update_woken(); + assert!(ran.load(Ordering::SeqCst)); + assert!(!executor::is_done()); + executor::update(); + assert!(executor::is_done()); +} + +#[test] +fn waking_and_spawning_during_poll_does_not_deadlock() { + let _guard = lock_executor(); + let spawned = Arc::new(AtomicBool::new(false)); + let flag = Arc::clone(&spawned); + executor::add_async(async move { + WakeAndSpawn { + polls: 0, + spawned: flag, + } + .await; + }); + + let (tx, rx) = std::sync::mpsc::channel(); + thread::spawn(move || { + let result = std::panic::catch_unwind(std::panic::AssertUnwindSafe(|| { + executor::update(); + executor::update(); + })); + let _ = tx.send(result); + }); + match rx.recv_timeout(Duration::from_secs(2)) { + Ok(Ok(())) => {} + Ok(Err(payload)) => std::panic::resume_unwind(payload), + Err(_) => panic!("update deadlocked while a task woke and spawned during poll"), + } + assert!(spawned.load(Ordering::SeqCst)); + assert!(executor::is_done()); +} + +#[test] +fn self_wake_during_poll_completes_without_deadlock() { + let _guard = lock_executor(); + let polls = Arc::new(AtomicUsize::new(0)); + let seen = Arc::clone(&polls); + executor::add_async(async move { + Steps { + left: 4, + polls: seen, + } + .await; + }); + drive_within(32, Duration::from_secs(2)); + assert!(executor::is_done()); + assert!( + polls.load(Ordering::SeqCst) >= 4, + "each step should be polled" + ); +} + +#[test] +fn ready_task_is_not_polled_again() { + let _guard = lock_executor(); + let polls = Arc::new(AtomicUsize::new(0)); + let seen = Arc::clone(&polls); + executor::add_async(async move { + PollCount { polls: seen }.await; + }); + executor::update(); + assert_eq!(polls.load(Ordering::SeqCst), 1); + assert!(executor::is_done()); + executor::update(); + executor::update_woken(); + executor::update(); + assert_eq!(polls.load(Ordering::SeqCst), 1); +} + +#[test] +fn woken_task_is_not_polled_after_ready() { + let _guard = lock_executor(); + let polls = Arc::new(AtomicUsize::new(0)); + let seen = Arc::clone(&polls); + let slot = fresh_slot(); + let slot_for_task = Arc::clone(&slot); + executor::add_async(async move { + Wait { + slot: slot_for_task, + } + .await; + PollCount { polls: seen }.await; + }); + executor::update(); + let waker = slot.lock().unwrap().waker.clone().expect("waker"); + slot.lock().unwrap().ready = true; + waker.wake_by_ref(); + executor::update_woken(); + // Inner Wait returned Ready and the following PollCount ran in that same + // poll of the async block. One poll of PollCount, then the task is done + // but still listed. + assert_eq!(polls.load(Ordering::SeqCst), 1); + executor::update(); + executor::update(); + assert_eq!(polls.load(Ordering::SeqCst), 1); + assert!(executor::is_done()); +} + +#[test] +fn update_woken_does_not_poll_unrelated_tasks() { + let _guard = lock_executor(); + let polls = Arc::new(AtomicUsize::new(0)); + let go = Arc::new(AtomicBool::new(false)); + let polls_task = Arc::clone(&polls); + let go_task = Arc::clone(&go); + executor::add_async(async move { + Gate { + polls: polls_task, + go: go_task, + } + .await; + }); + let slot = fresh_slot(); + let waker = arm(&slot); + // `arm` already polled every queued task once, including the gate. + assert_eq!(polls.load(Ordering::SeqCst), 1); + + slot.lock().unwrap().ready = true; + waker.wake_by_ref(); + executor::update_woken(); + assert_eq!( + polls.load(Ordering::SeqCst), + 1, + "update_woken must not poll a task that was not woken" + ); + + go.store(true, Ordering::SeqCst); + executor::update(); + assert_eq!(polls.load(Ordering::SeqCst), 2); + assert!(executor::is_done()); +} + +#[test] +fn is_done_tracks_input_queue_and_completed_tasks() { + let _guard = lock_executor(); + assert!(executor::is_done()); + + let slot = fresh_slot(); + let slot_for_task = Arc::clone(&slot); + executor::add_async(async move { + Wait { + slot: slot_for_task, + } + .await; + }); + assert!( + !executor::is_done(), + "a task sitting in the input queue is not done" + ); + + executor::update(); + assert!(!executor::is_done(), "pending task is not done"); + + let waker = slot.lock().unwrap().waker.clone().unwrap(); + slot.lock().unwrap().ready = true; + waker.wake(); + assert!( + !executor::is_done(), + "a woken task is still queued until it is polled and dropped" + ); + + executor::update_woken(); + assert!( + !executor::is_done(), + "update_woken finishes the task but leaves it listed until update" + ); + executor::update(); + assert!(executor::is_done()); +} + +#[test] +fn non_unit_outputs_run() { + let _guard = lock_executor(); + let number = Arc::new(AtomicUsize::new(0)); + let number_task = Arc::clone(&number); + executor::add_async(async move { + let value: u32 = async { 42u32 }.await; + number_task.store(value as usize, Ordering::SeqCst); + value + }); + + let text = Arc::new(Mutex::new(String::new())); + let text_task = Arc::clone(&text); + executor::add_async(async move { + let value = String::from("executor"); + *text_task.lock().unwrap() = value.clone(); + value + }); + + // `Cell` is `Send` and not `Sync`. Output only has to be `Send`. + executor::add_async(async { Cell::new(7u32) }); + + drive(8); + assert_eq!(number.load(Ordering::SeqCst), 42); + assert_eq!(text.lock().unwrap().as_str(), "executor"); + assert!(executor::is_done()); +} + +struct Hold { + drops: Arc, + slot: Arc>, +} + +impl Drop for Hold { + fn drop(&mut self) { + self.drops.fetch_add(1, Ordering::SeqCst); + } +} + +impl Future for Hold { + type Output = (); + + fn poll(self: Pin<&mut Self>, cx: &mut Context<'_>) -> Poll<()> { + let mut slot = self.slot.lock().unwrap(); + if slot.ready { + Poll::Ready(()) + } else { + slot.waker = Some(cx.waker().clone()); + Poll::Pending + } + } +} + +#[test] +fn completed_tasks_are_dropped() { + let _guard = lock_executor(); + let drops = Arc::new(AtomicUsize::new(0)); + let output_drops = Arc::clone(&drops); + executor::add_async(async move { DropCount(output_drops) }); + executor::update(); + assert_eq!( + drops.load(Ordering::SeqCst), + 1, + "Ready output is dropped when the task finishes" + ); + assert!(executor::is_done()); + + // `Hold`'s Drop runs when the future is dropped, not when poll returns Ready. + let held = Arc::new(AtomicUsize::new(0)); + let slot = fresh_slot(); + executor::add_async(Hold { + drops: Arc::clone(&held), + slot: Arc::clone(&slot), + }); + executor::update(); + assert_eq!( + held.load(Ordering::SeqCst), + 0, + "pending task is still alive" + ); + + let waker = slot.lock().unwrap().waker.take().unwrap(); + slot.lock().unwrap().ready = true; + waker.wake_by_ref(); + executor::update_woken(); + assert_eq!( + held.load(Ordering::SeqCst), + 0, + "update_woken finishes the future but the main list still owns it" + ); + drop(waker); + assert_eq!( + held.load(Ordering::SeqCst), + 0, + "the waker was not the last owner; the executor list still is" + ); + executor::update(); + assert_eq!( + held.load(Ordering::SeqCst), + 1, + "update drops the finished task" + ); + assert!(executor::is_done()); +} + +#[test] +fn many_wakes_finish_quickly() { + let _guard = lock_executor(); + let polls = Arc::new(AtomicUsize::new(0)); + const TASKS: usize = 8; + const STEPS: u32 = 16; + for _ in 0..TASKS { + let polls = Arc::clone(&polls); + executor::add_async(async move { + Steps { left: STEPS, polls }.await; + }); + } + let started = Instant::now(); + // Miri interprets every poll, so the wall-clock budget is looser there. + let budget = if cfg!(miri) { 60 } else { 2 }; + drive_within(10_000, Duration::from_secs(budget)); + if !cfg!(miri) { + assert!(started.elapsed() < Duration::from_secs(2)); + } + assert!(executor::is_done()); + assert!(polls.load(Ordering::SeqCst) >= TASKS * STEPS as usize); +}