diff --git a/.github/workflows/ci.yml b/.github/workflows/ci.yml index e58b091..547624d 100644 --- a/.github/workflows/ci.yml +++ b/.github/workflows/ci.yml @@ -6,104 +6,179 @@ on: pull_request: branches: ["**"] -concurrency: - group: ci-${{ github.ref }} - cancel-in-progress: true - env: CARGO_TERM_COLOR: always jobs: lint: - name: Lint + name: Lint (clippy + fmt) runs-on: ubuntu-latest steps: - uses: actions/checkout@v4 - - uses: dtolnay/rust-toolchain@nightly + + - name: Install Rust + uses: dtolnay/rust-toolchain@stable with: + toolchain: stable components: clippy, rustfmt - - uses: Swatinem/rust-cache@v2 + + - name: Setup Rust cache + uses: Swatinem/rust-cache@v2 + + - name: Check formatting + run: cargo fmt --all -- --check + + - name: Run clippy + run: cargo clippy --all-targets --all-features -- -D warnings + + test: + name: Test on ${{ matrix.os }} + needs: lint + runs-on: ${{ matrix.os }} + strategy: + matrix: + os: [ubuntu-latest, windows-latest, macos-latest] + + steps: + - uses: actions/checkout@v4 + + - name: Install Rust 1.95 (MSRV) + uses: dtolnay/rust-toolchain@stable with: - shared-key: ci-ubuntu - - run: cargo fmt --all -- --check - - run: cargo clippy --all-targets --all-features -- -D warnings + toolchain: "1.95" + + - name: Setup Rust cache + uses: Swatinem/rust-cache@v2 + + - name: Build + run: cargo build --verbose + + - uses: taiki-e/install-action@nextest + + # Every test needs a backend feature; the default set (`std`) has none. + - name: Run tests + run: cargo nextest run --no-tests=pass + + - name: Run tests with all features + run: cargo nextest run --all-features + + - name: Test documentation + run: cargo test --doc wasm: - name: WASM + name: WASM build + needs: lint runs-on: ubuntu-latest steps: - uses: actions/checkout@v4 - - uses: dtolnay/rust-toolchain@nightly + + - name: Install Rust + uses: dtolnay/rust-toolchain@stable with: + toolchain: stable targets: wasm32-unknown-unknown - - uses: Swatinem/rust-cache@v2 + + - name: Setup Rust cache + uses: Swatinem/rust-cache@v2 + + - name: Build for WASM + run: cargo build --target wasm32-unknown-unknown --all-features + + minimal-versions: + name: Minimal versions + runs-on: ubuntu-latest + steps: + - uses: actions/checkout@v4 + + - name: Install Rust + uses: dtolnay/rust-toolchain@stable with: - shared-key: ci-ubuntu - - run: cargo build --target wasm32-unknown-unknown --features std + toolchain: stable + + - name: Setup Rust cache + uses: Swatinem/rust-cache@v2 + - name: Install cargo-hack + run: cargo install cargo-hack + + - name: Check minimal versions + run: cargo hack check --all --ignore-private coverage: name: Coverage runs-on: ubuntu-latest steps: - uses: actions/checkout@v4 - - uses: dtolnay/rust-toolchain@nightly + + - name: Install Rust + uses: dtolnay/rust-toolchain@stable with: + toolchain: stable components: llvm-tools-preview - - uses: Swatinem/rust-cache@v2 - with: - shared-key: ci-ubuntu - - uses: taiki-e/install-action@cargo-llvm-cov - - run: cargo llvm-cov --all-features --workspace --lcov --output-path lcov.info - - uses: codecov/codecov-action@v4 + + - name: Setup Rust cache + uses: Swatinem/rust-cache@v2 + + - name: Install cargo-llvm-cov + run: cargo install cargo-llvm-cov + + - uses: taiki-e/install-action@nextest + + - name: Generate test coverage + run: cargo llvm-cov nextest --all-features --workspace --lcov --output-path lcov.info + + - name: Upload coverage to Codecov + uses: codecov/codecov-action@v4 with: files: lcov.info fail_ci_if_error: true - typos: - name: Typos + + feature-checks: + name: Feature checks runs-on: ubuntu-latest steps: - uses: actions/checkout@v4 - - uses: crate-ci/typos@master - audit: - name: Security Audit - runs-on: ubuntu-latest - steps: - - uses: actions/checkout@v4 - - uses: Swatinem/rust-cache@v2 - with: - shared-key: ci-ubuntu - - name: Generate Lockfile - run: cargo generate-lockfile - - uses: rustsec/audit-check@v2 - with: - token: ${{ secrets.GITHUB_TOKEN }} + - name: Install Rust + uses: dtolnay/rust-toolchain@stable + with: + toolchain: stable - unused-deps: - name: Unused Dependencies + - name: Setup Rust cache + uses: Swatinem/rust-cache@v2 + + - name: Install cargo-hack + run: cargo install cargo-hack + + - name: Check each feature + run: cargo hack check --each-feature --no-dev-deps + + - name: Check feature combinations + run: cargo hack check --feature-powerset --no-dev-deps + + audit: + name: Security audit runs-on: ubuntu-latest steps: - uses: actions/checkout@v4 - - uses: dtolnay/rust-toolchain@nightly - - uses: Swatinem/rust-cache@v2 + + - name: Generate lockfile + run: cargo generate-lockfile + + - uses: rustsec/audit-check@v2 with: - shared-key: ci-ubuntu - - uses: taiki-e/install-action@cargo-machete - - run: cargo machete + token: ${{ secrets.GITHUB_TOKEN }} - feature-checks: - name: Features + unused-deps: + name: Unused dependencies runs-on: ubuntu-latest steps: - uses: actions/checkout@v4 - - uses: dtolnay/rust-toolchain@nightly - - uses: Swatinem/rust-cache@v2 + + - name: Install Rust + uses: dtolnay/rust-toolchain@stable with: - shared-key: ci-ubuntu - - uses: taiki-e/install-action@cargo-hack - - run: cargo hack check --each-feature --no-dev-deps - - run: cargo hack check --feature-powerset --no-dev-deps + toolchain: stable - test: - name: Test - uses: ./.github/workflows/test.yml + - uses: taiki-e/install-action@cargo-machete + + - run: cargo machete diff --git a/.github/workflows/dep-check.yml b/.github/workflows/dep-check.yml new file mode 100644 index 0000000..c1f00cb --- /dev/null +++ b/.github/workflows/dep-check.yml @@ -0,0 +1,101 @@ +name: Dependency Check + +on: + schedule: + - cron: "0 9 * * 1" + workflow_dispatch: + +permissions: + issues: write + contents: read + +jobs: + check: + runs-on: ubuntu-latest + steps: + - uses: actions/checkout@v4 + + - name: Detect project type + id: detect + run: | + if [ -f Cargo.toml ]; then + echo "type=rust" >> "$GITHUB_OUTPUT" + elif [ -f package.json ]; then + echo "type=node" >> "$GITHUB_OUTPUT" + elif [ -f Package.swift ]; then + echo "type=swift" >> "$GITHUB_OUTPUT" + else + echo "type=unknown" >> "$GITHUB_OUTPUT" + fi + + - name: Setup Rust + if: steps.detect.outputs.type == 'rust' + uses: dtolnay/rust-toolchain@stable + + - name: Rust cache + if: steps.detect.outputs.type == 'rust' + uses: Swatinem/rust-cache@v2 + + - name: Check Rust dependencies + if: steps.detect.outputs.type == 'rust' + run: | + cargo install cargo-outdated --locked + cargo outdated -R --exit-code 1 > outdated.txt 2>&1 || true + cat outdated.txt + + - name: Setup Node + if: steps.detect.outputs.type == 'node' + uses: actions/setup-node@v4 + with: + node-version: lts/* + + - name: Check Node dependencies + if: steps.detect.outputs.type == 'node' + run: | + npm install --ignore-scripts 2>/dev/null || true + npm outdated --long 2>/dev/null > outdated.txt || true + cat outdated.txt + + - name: Check Swift dependencies + if: steps.detect.outputs.type == 'swift' + run: | + swift package show-dependencies 2>/dev/null > outdated.txt || echo "Unable to check Swift dependencies" > outdated.txt + cat outdated.txt + + - name: Report + if: steps.detect.outputs.type != 'unknown' + env: + GH_TOKEN: ${{ github.token }} + run: | + gh label create "dependencies" --description "Dependency updates" --color "0075ca" 2>/dev/null || true + + if [ ! -s outdated.txt ]; then + echo "No outdated dependencies" + existing=$(gh issue list --label "dependencies" --state open --json number --jq '.[0].number' 2>/dev/null || true) + if [ -n "$existing" ] && [ "$existing" != "null" ]; then + gh issue close "$existing" -c "All dependencies are up to date." + fi + exit 0 + fi + + content=$(cat outdated.txt) + body=$(cat </dev/null || true) + + if [ -n "$existing" ] && [ "$existing" != "null" ]; then + gh issue edit "$existing" --body "$body" + echo "Updated issue #$existing" + else + gh issue create --title "Outdated dependencies" --body "$body" --label "dependencies" + echo "Created tracking issue" + fi diff --git a/.github/workflows/pr-base-gate.yml b/.github/workflows/pr-base-gate.yml new file mode 100644 index 0000000..84aae34 --- /dev/null +++ b/.github/workflows/pr-base-gate.yml @@ -0,0 +1,71 @@ +name: PR Base Gate + +on: + pull_request: + branches: [main] + types: [opened, reopened, synchronize, edited] + +permissions: + checks: read + +jobs: + verify-source: + name: PR source gate + runs-on: ubuntu-latest + timeout-minutes: 60 + steps: + - name: main only accepts merges from dev + env: + HEAD_REF: ${{ github.event.pull_request.head.ref }} + HEAD_REPO: ${{ github.event.pull_request.head.repo.full_name }} + BASE_REPO: ${{ github.repository }} + run: | + if [ "${HEAD_REPO}" != "${BASE_REPO}" ]; then + echo "::error::Pull requests into 'main' must come from this repository, not a fork." + exit 1 + fi + if [ "${HEAD_REF}" != "dev" ]; then + echo "::error::Pull requests into 'main' are only allowed from 'dev'. Retarget this pull request at 'dev' instead." + exit 1 + fi + echo "'${HEAD_REF}' -> 'main': source branch allowed." + + # This check doubles as the single required check on main: instead of + # pinning CI job names in the ruleset (which break on every rename), it + # waits for every other check run on the PR's head and merge commits and + # goes green only when all of them did. + - name: wait for every other check on this PR to go green + env: + GH_TOKEN: ${{ secrets.GITHUB_TOKEN }} + HEAD_SHA: ${{ github.event.pull_request.head.sha }} + MERGE_SHA: ${{ github.event.pull_request.merge_commit_sha }} + run: | + set -euo pipefail + deadline=$(( $(date +%s) + 3300 )) + while :; do + { + gh api "repos/${GITHUB_REPOSITORY}/commits/${HEAD_SHA}/check-runs?per_page=100" \ + --jq '.check_runs[] | [.name, (.conclusion // "pending")] | @tsv' + if [ -n "${MERGE_SHA:-}" ] && [ "${MERGE_SHA}" != "null" ]; then + gh api "repos/${GITHUB_REPOSITORY}/commits/${MERGE_SHA}/check-runs?per_page=100" \ + --jq '.check_runs[] | [.name, (.conclusion // "pending")] | @tsv' + fi + } 2>/dev/null | grep -v '^PR source gate\b' > checks.tsv || true + + if grep -E '\t(failure|cancelled|timed_out|action_required|startup_failure|stale)$' checks.tsv; then + echo "::error::a check on this pull request did not pass (see list above)" + exit 1 + fi + pending=$(grep -c $'\tpending$' checks.tsv || true) + if [ "${pending}" -eq 0 ]; then + echo "all checks on this pull request have passed" + exit 0 + fi + if [ "$(date +%s)" -gt "${deadline}" ]; then + echo "::error::timed out after 55 minutes waiting for ${pending} check(s) to conclude" + grep $'\tpending$' checks.tsv + exit 1 + fi + echo "waiting on ${pending} check(s)..." + sleep 30 + done diff --git a/.github/workflows/release.yml b/.github/workflows/release.yml index b32622a..e67c0dc 100644 --- a/.github/workflows/release.yml +++ b/.github/workflows/release.yml @@ -1,26 +1,36 @@ name: Release on: - workflow_run: - workflows: ["CI"] - branches: [main] - types: [completed] + push: + branches: + - main permissions: - contents: write - pull-requests: write + contents: write + pull-requests: write + id-token: write jobs: - release-plz: - name: Release - runs-on: ubuntu-latest - if: github.event.workflow_run.conclusion == 'success' - steps: - - uses: actions/checkout@v4 - with: - fetch-depth: 0 - - uses: dtolnay/rust-toolchain@stable - - uses: MarcoIeni/release-plz-action@v0.5 - env: - GITHUB_TOKEN: ${{ secrets.GITHUB_TOKEN }} - CARGO_REGISTRY_TOKEN: ${{ secrets.CARGO_REGISTRY_TOKEN }} + release-plz: + name: Release-plz + runs-on: ubuntu-latest + steps: + - uses: actions/checkout@v4 + with: + fetch-depth: 0 + + - name: Install Rust + uses: dtolnay/rust-toolchain@v1 + with: + toolchain: stable + + # Mint a short-lived crates.io token from the GitHub OIDC + # identity instead of a stored CARGO_REGISTRY_TOKEN. + - uses: rust-lang/crates-io-auth-action@v1 + id: crates-io-auth + + - name: Run release-plz + uses: MarcoIeni/release-plz-action@v0.5 + env: + GITHUB_TOKEN: ${{ secrets.GITHUB_TOKEN }} + CARGO_REGISTRY_TOKEN: ${{ steps.crates-io-auth.outputs.token }} diff --git a/.github/workflows/test.yml b/.github/workflows/test.yml deleted file mode 100644 index 4ed30f7..0000000 --- a/.github/workflows/test.yml +++ /dev/null @@ -1,23 +0,0 @@ -name: Test - -on: - workflow_call: - -env: - CARGO_TERM_COLOR: always - -jobs: - test: - name: ${{ matrix.os }} - runs-on: ${{ matrix.os }} - strategy: - fail-fast: true - matrix: - os: [ubuntu-latest, windows-latest, macos-latest] - steps: - - uses: actions/checkout@v4 - - uses: dtolnay/rust-toolchain@nightly - - uses: Swatinem/rust-cache@v2 - with: - shared-key: test-${{ matrix.os }} - - run: cargo test --verbose --all-features diff --git a/Cargo.toml b/Cargo.toml index f6d092a..eee1638 100644 --- a/Cargo.toml +++ b/Cargo.toml @@ -3,7 +3,7 @@ name = "executor-core" version = "0.7.1" authors = ["Lexo Liu "] edition = "2024" -rust-version = "1.85.1" +rust-version = "1.95" description = "A zero-cost task executor abstraction layer for Rust async runtimes" repository = "https://github.com/lexoliu/executor-core" documentation = "https://docs.rs/executor-core" @@ -19,12 +19,11 @@ license = { file = "LICENSE" } [package.metadata.docs.rs] -features = ["async-task", "async-executor", "tokio", "std"] +features = ["async-task", "async-executor", "tokio", "compio", "smol", "std"] rustdoc-args = ["--cfg", "docsrs"] [dependencies] async-channel = { version = "2.5.0", default-features = false } -smol = { version = "2.0.2", default-features = false, optional = true } [dependencies.async-task] version = "4.7" @@ -36,12 +35,23 @@ features = [] version = "1.13.2" optional = true -[dependencies.tokio] +[target.'cfg(not(target_arch = "wasm32"))'.dependencies.tokio] version = "1.45" default-features = false features = ["rt", "rt-multi-thread"] optional = true +[target.'cfg(not(target_arch = "wasm32"))'.dependencies.smol] +version = "2.0.2" +default-features = false +optional = true + +[target.'cfg(not(target_arch = "wasm32"))'.dependencies.compio] +version = "0.19.2" +default-features = false +features = ["dispatcher", "runtime"] +optional = true + [features] default = ["std"] std = ["async-task?/std"] @@ -49,9 +59,10 @@ async-task = ["dep:async-task"] async-executor = ["dep:async-executor", "async-task", "std"] smol = ["dep:smol","async-task"] tokio = ["dep:tokio", "std"] +compio = ["dep:compio", "std"] # Feature flag for enabling all compatible features -full = ["std", "async-task", "async-executor", "tokio"] +full = ["std", "async-task", "async-executor", "tokio", "compio"] [dev-dependencies] tokio = { version = "1.45", features = [ diff --git a/src/async_executor.rs b/src/async_executor.rs index d57b879..13d6fb6 100644 --- a/src/async_executor.rs +++ b/src/async_executor.rs @@ -36,23 +36,13 @@ mod tests { extern crate std; use crate::{Executor, LocalExecutor, Task, async_task::AsyncTask}; - use alloc::task::Wake; - use alloc::{format, sync::Arc}; + use alloc::format; use core::future::Future; use core::{ pin::Pin, task::{Context, Poll, Waker}, }; - struct TestWaker; - impl Wake for TestWaker { - fn wake(self: Arc) {} - } - - fn create_waker() -> Waker { - Arc::new(TestWaker).into() - } - async fn sleep_ms(ms: u64) { #[cfg(feature = "std")] { @@ -99,8 +89,7 @@ mod tests { let ex = async_executor::Executor::new(); let mut task: AsyncTask = Executor::spawn(&ex, async { 100 }); - let waker = create_waker(); - let mut cx = Context::from_waker(&waker); + let mut cx = Context::from_waker(Waker::noop()); match Pin::new(&mut task).poll(&mut cx) { Poll::Ready(result) => assert_eq!(result, 100), @@ -116,8 +105,7 @@ mod tests { let ex = async_executor::Executor::new(); let mut task: AsyncTask<&str> = Executor::spawn(&ex, async { "success" }); - let waker = create_waker(); - let mut cx = Context::from_waker(&waker); + let mut cx = Context::from_waker(Waker::noop()); match Pin::new(&mut task).poll_result(&mut cx) { Poll::Ready(Ok(result)) => assert_eq!(result, "success"), @@ -180,8 +168,7 @@ mod tests { let mut task: AsyncTask<&str> = LocalExecutor::spawn_local(&local_ex, async { "local success" }); - let waker = create_waker(); - let mut cx = Context::from_waker(&waker); + let mut cx = Context::from_waker(Waker::noop()); match Pin::new(&mut task).poll_result(&mut cx) { Poll::Ready(Ok(result)) => assert_eq!(result, "local success"), @@ -254,8 +241,7 @@ mod tests { let ex = async_executor::Executor::new(); let mut task: AsyncTask = Executor::spawn(&ex, async { 42 }); - let waker = create_waker(); - let mut cx = Context::from_waker(&waker); + let mut cx = Context::from_waker(Waker::noop()); let _poll_result = Pin::new(&mut task).poll_result(&mut cx); diff --git a/src/compio.rs b/src/compio.rs new file mode 100644 index 0000000..bbef24e --- /dev/null +++ b/src/compio.rs @@ -0,0 +1,422 @@ +//! Integration with the compio async runtime. +//! +//! compio's [`Runtime`] is thread-local (`!Send`, `!Sync`), so the two traits map onto two +//! different compio types: +//! +//! - [`LocalExecutor`] is implemented for [`Runtime`] itself. Spawning goes straight to +//! [`Runtime::spawn`], and the task is driven by whoever runs that runtime +//! (for example [`Runtime::block_on`]). +//! - [`Executor`] (`Send` futures, callable from any thread) is implemented for +//! [`Dispatcher`], compio's thread-per-core pool: each spawned future is sent to a worker +//! thread, built there and driven by that thread's own compio runtime. +//! +//! Dropping a task cancels it on both sides, and [`Task::detach`] lets it run to completion. + +extern crate std; + +use crate::{Error, Executor, LocalExecutor, Task}; +use alloc::boxed::Box; +use async_channel::{Receiver, Sender}; +use core::{ + future::{Future, poll_fn}, + panic::AssertUnwindSafe, + pin::{Pin, pin}, + task::{Context, Poll}, +}; +use std::panic::{catch_unwind, resume_unwind}; + +use compio::runtime::{JoinError, JoinHandle}; +pub use compio::{dispatcher::Dispatcher, runtime::Runtime}; + +fn cancelled() -> Error { + Box::new("Task was cancelled") +} + +/// Task spawned on a compio [`Runtime`] through [`LocalExecutor`]. +/// +/// Dropping it cancels the task. +pub struct CompioLocalTask { + handle: JoinHandle, +} + +impl core::fmt::Debug for CompioLocalTask { + fn fmt(&self, f: &mut core::fmt::Formatter<'_>) -> core::fmt::Result { + f.debug_struct("CompioLocalTask").finish_non_exhaustive() + } +} + +impl Future for CompioLocalTask { + type Output = T; + + fn poll(mut self: Pin<&mut Self>, cx: &mut Context<'_>) -> Poll { + Pin::new(&mut self.handle) + .poll(cx) + .map(|result| match result { + Ok(value) => value, + Err(JoinError::Panicked(payload)) => resume_unwind(payload), + Err(JoinError::Cancelled) => std::panic::panic_any("Task was cancelled"), + }) + } +} + +impl Task for CompioLocalTask { + fn poll_result(mut self: Pin<&mut Self>, cx: &mut Context<'_>) -> Poll> { + Pin::new(&mut self.handle) + .poll(cx) + .map(|result| match result { + Ok(value) => Ok(value), + Err(JoinError::Panicked(payload)) => Err(payload), + Err(JoinError::Cancelled) => Err(cancelled()), + }) + } + + fn detach(self) { + self.handle.detach(); + } +} + +impl LocalExecutor for Runtime { + type Task = CompioLocalTask; + + fn spawn_local(&self, fut: Fut) -> Self::Task + where + Fut: Future + 'static, + { + CompioLocalTask { + handle: self.spawn(fut), + } + } +} + +/// A finished future's output, or its panic payload. +type Outcome = Result; + +/// Task spawned on a compio [`Dispatcher`] through [`Executor`]. +/// +/// Dropping it cancels the future on its worker thread. +pub struct CompioTask { + /// Resolves with the future's output or its panic payload, once the worker is done. + result: Pin>> + Send>>, + /// Tells the worker to drop the future. `None` once detached. + cancel: Option>, +} + +impl core::fmt::Debug for CompioTask { + fn fmt(&self, f: &mut core::fmt::Formatter<'_>) -> core::fmt::Result { + f.debug_struct("CompioTask").finish_non_exhaustive() + } +} + +impl Future for CompioTask { + type Output = T; + + fn poll(mut self: Pin<&mut Self>, cx: &mut Context<'_>) -> Poll { + self.as_mut().poll_result(cx).map(|result| match result { + Ok(value) => value, + Err(payload) => resume_unwind(payload), + }) + } +} + +impl Task for CompioTask { + fn poll_result(mut self: Pin<&mut Self>, cx: &mut Context<'_>) -> Poll> { + // The worker dropped its end without a result: it was shut down mid-flight. + self.result + .as_mut() + .poll(cx) + .map(|result| result.unwrap_or_else(|| Err(cancelled()))) + } + + fn detach(mut self) { + // Closing the channel without a message means "keep running". + drop(self.cancel.take()); + } +} + +impl Drop for CompioTask { + fn drop(&mut self) { + if let Some(cancel) = self.cancel.take() { + // The channel holds one message and is sent to at most once, so this cannot fail + // for lack of room; a closed channel means the worker already finished. + let _ = cancel.try_send(()); + } + } +} + +/// Runs `fut` to completion on the worker, catching its panic, until `cancel` carries a +/// message. A cancel channel that closes empty means the task was detached. +async fn drive(fut: Fut, cancel: Receiver<()>) -> Option> { + let mut fut = pin!(fut); + // Kept across polls: dropping it would drop the wake registration with it. + let mut recv = pin!(cancel.recv()); + let mut detached = false; + poll_fn(|cx| { + if !detached { + match recv.as_mut().poll(cx) { + Poll::Ready(Ok(())) => return Poll::Ready(None), + Poll::Ready(Err(_)) => detached = true, + Poll::Pending => {} + } + } + match catch_unwind(AssertUnwindSafe(|| fut.as_mut().poll(cx))) { + Ok(Poll::Ready(value)) => Poll::Ready(Some(Ok(value))), + Ok(Poll::Pending) => Poll::Pending, + Err(payload) => Poll::Ready(Some(Err(payload))), + } + }) + .await +} + +impl Executor for Dispatcher { + type Task = CompioTask; + + fn spawn(&self, fut: Fut) -> Self::Task + where + Fut: Future + Send + 'static, + { + let (result_tx, result_rx) = async_channel::bounded(1); + let (cancel_tx, cancel_rx) = async_channel::bounded(1); + + let dispatched = self.dispatch(move || async move { + let outcome = drive(fut, cancel_rx).await; + if let Some(outcome) = outcome { + // The task may already be dropped; nobody is left to read the result. + let _ = result_tx.send(outcome).await; + } + }); + if dispatched.is_err() { + panic!("every compio dispatcher worker has panicked; cannot spawn"); + } + + CompioTask { + result: Box::pin(async move { result_rx.recv().await.ok() }), + cancel: Some(cancel_tx), + } + } +} + +#[cfg(test)] +mod tests { + use super::*; + use alloc::{format, sync::Arc}; + use core::sync::atomic::{AtomicBool, Ordering}; + use std::{thread, time::Duration}; + + /// Sets a flag when dropped, proving the future holding it was dropped. + struct DropFlag(Arc); + + impl Drop for DropFlag { + fn drop(&mut self) { + self.0.store(true, Ordering::SeqCst); + } + } + + fn dispatcher() -> Dispatcher { + Dispatcher::new().expect("dispatcher") + } + + fn wait_for(flag: &AtomicBool) { + for _ in 0..500 { + if flag.load(Ordering::SeqCst) { + return; + } + thread::sleep(Duration::from_millis(10)); + } + panic!("condition not reached"); + } + + #[test] + fn spawn_returns_output() { + let ex = dispatcher(); + let task: CompioTask = Executor::spawn(&ex, async { 42 }); + assert_eq!(futures_lite::future::block_on(task), 42); + } + + #[test] + fn spawn_from_other_thread() { + let ex = Arc::new(dispatcher()); + let ex2 = ex.clone(); + let task = thread::spawn(move || Executor::spawn(&*ex2, async { "remote" })) + .join() + .unwrap(); + assert_eq!(futures_lite::future::block_on(task), "remote"); + } + + #[test] + fn spawn_runs_off_the_calling_thread() { + let ex = dispatcher(); + let caller = thread::current().id(); + let task = Executor::spawn(&ex, async move { thread::current().id() }); + assert_ne!(futures_lite::future::block_on(task), caller); + } + + #[test] + fn spawn_async_operation() { + let ex = dispatcher(); + let task = Executor::spawn(&ex, async { + for _ in 0..10 { + futures_lite::future::yield_now().await; + } + "completed" + }); + assert_eq!(futures_lite::future::block_on(task), "completed"); + } + + #[test] + fn spawn_poll_result() { + let ex = dispatcher(); + let task = Executor::spawn(&ex, async { "success" }); + let result = futures_lite::future::block_on(task.result()); + assert_eq!(result.unwrap(), "success"); + } + + #[test] + fn spawn_panic_is_reported_and_worker_survives() { + let ex = dispatcher(); + let task: CompioTask<()> = Executor::spawn(&ex, async { panic!("test panic") }); + let error = futures_lite::future::block_on(task.result()).unwrap_err(); + assert_eq!(*error.downcast::<&str>().unwrap(), "test panic"); + + let task = Executor::spawn(&ex, async { 1 }); + assert_eq!(futures_lite::future::block_on(task), 1); + } + + #[test] + fn spawn_many_tasks_concurrently() { + let ex = dispatcher(); + let tasks: alloc::vec::Vec<_> = (0..64) + .map(|i| Executor::spawn(&ex, async move { i })) + .collect(); + let sum: i32 = tasks.into_iter().map(futures_lite::future::block_on).sum(); + assert_eq!(sum, (0..64).sum()); + } + + #[test] + fn dropping_a_task_cancels_it() { + let ex = dispatcher(); + let started = Arc::new(AtomicBool::new(false)); + let dropped = Arc::new(AtomicBool::new(false)); + + let task = { + let (started, flag) = (started.clone(), DropFlag(dropped.clone())); + Executor::spawn(&ex, async move { + let _flag = flag; + started.store(true, Ordering::SeqCst); + core::future::pending::<()>().await; + }) + }; + wait_for(&started); + assert!(!dropped.load(Ordering::SeqCst)); + + drop(task); + wait_for(&dropped); + } + + #[test] + fn dropping_a_task_cancels_it_before_it_starts() { + let ex = dispatcher(); + let dropped = Arc::new(AtomicBool::new(false)); + let flag = DropFlag(dropped.clone()); + let task = Executor::spawn(&ex, async move { + let _flag = flag; + core::future::pending::<()>().await; + }); + drop(task); + wait_for(&dropped); + } + + #[test] + fn detached_task_runs_to_completion() { + let ex = dispatcher(); + let done = Arc::new(AtomicBool::new(false)); + let flag = done.clone(); + Executor::spawn(&ex, async move { + for _ in 0..10 { + futures_lite::future::yield_now().await; + } + flag.store(true, Ordering::SeqCst); + }) + .detach(); + wait_for(&done); + } + + #[test] + fn task_debug() { + let ex = dispatcher(); + let task = Executor::spawn(&ex, async { 1 }); + assert!(format!("{task:?}").contains("CompioTask")); + } + + #[test] + fn spawn_local_returns_output() { + let rt = Runtime::new().expect("runtime"); + let task: CompioLocalTask = LocalExecutor::spawn_local(&rt, async { 42 }); + assert_eq!(rt.block_on(task), 42); + } + + #[test] + fn spawn_local_accepts_non_send_futures() { + let rt = Runtime::new().expect("runtime"); + let value = alloc::rc::Rc::new(7); + let task = LocalExecutor::spawn_local(&rt, async move { *value }); + assert_eq!(rt.block_on(task), 7); + } + + #[test] + fn spawn_local_poll_result() { + let rt = Runtime::new().expect("runtime"); + let task = LocalExecutor::spawn_local(&rt, async { "local success" }); + assert_eq!(rt.block_on(task.result()).unwrap(), "local success"); + } + + #[test] + fn spawn_local_panic_is_reported() { + let rt = Runtime::new().expect("runtime"); + let task: CompioLocalTask<()> = + LocalExecutor::spawn_local(&rt, async { panic!("local panic") }); + let error = rt.block_on(task.result()).unwrap_err(); + assert_eq!(*error.downcast::<&str>().unwrap(), "local panic"); + } + + #[test] + fn spawn_local_dropping_a_task_cancels_it() { + let rt = Runtime::new().expect("runtime"); + let dropped = Arc::new(AtomicBool::new(false)); + let flag = DropFlag(dropped.clone()); + rt.block_on(async { + let task = LocalExecutor::spawn_local(&rt, async move { + let _flag = flag; + core::future::pending::<()>().await; + }); + futures_lite::future::yield_now().await; + assert!(!dropped.load(Ordering::SeqCst)); + drop(task); + futures_lite::future::yield_now().await; + }); + assert!(dropped.load(Ordering::SeqCst)); + } + + #[test] + fn spawn_local_detached_task_runs_to_completion() { + let rt = Runtime::new().expect("runtime"); + let done = Arc::new(AtomicBool::new(false)); + let flag = done.clone(); + rt.block_on(async { + LocalExecutor::spawn_local(&rt, async move { + flag.store(true, Ordering::SeqCst); + }) + .detach(); + for _ in 0..10 { + futures_lite::future::yield_now().await; + } + }); + assert!(done.load(Ordering::SeqCst)); + } + + #[test] + fn spawn_local_task_debug() { + let rt = Runtime::new().expect("runtime"); + let task = LocalExecutor::spawn_local(&rt, async { 1 }); + assert!(format!("{task:?}").contains("CompioLocalTask")); + } +} diff --git a/src/lib.rs b/src/lib.rs index cf12f1d..578c134 100644 --- a/src/lib.rs +++ b/src/lib.rs @@ -45,12 +45,16 @@ pub mod async_task; #[cfg_attr(docsrs, doc(cfg(feature = "async-executor")))] pub mod async_executor; -#[cfg(feature = "tokio")] +#[cfg(all(feature = "tokio", not(target_arch = "wasm32")))] #[cfg_attr(docsrs, doc(cfg(feature = "tokio")))] pub mod tokio; -#[cfg(feature = "smol")] -#[cfg_attr(docsrs, doc(cfg(feature = "tokio")))] +#[cfg(all(feature = "compio", not(target_arch = "wasm32")))] +#[cfg_attr(docsrs, doc(cfg(feature = "compio")))] +pub mod compio; + +#[cfg(all(feature = "smol", not(target_arch = "wasm32")))] +#[cfg_attr(docsrs, doc(cfg(feature = "smol")))] pub mod smol; use core::{ diff --git a/src/tokio.rs b/src/tokio.rs index 9d5290a..884153e 100644 --- a/src/tokio.rs +++ b/src/tokio.rs @@ -172,8 +172,7 @@ impl LocalExecutor for tokio::task::LocalSet { mod tests { use super::*; use crate::{Executor, LocalExecutor, Task}; - use alloc::task::Wake; - use alloc::{format, sync::Arc}; + use alloc::format; use core::future::Future; use core::{ pin::Pin, @@ -181,15 +180,6 @@ mod tests { }; use tokio::time::{Duration, sleep}; - struct TestWaker; - impl Wake for TestWaker { - fn wake(self: Arc) {} - } - - fn create_waker() -> Waker { - Arc::new(TestWaker).into() - } - #[test] fn test_default_executor_spawn() { let executor = Runtime::new().expect("Failed to create Tokio runtime"); @@ -214,8 +204,7 @@ mod tests { let executor = Runtime::new().expect("Failed to create Tokio runtime"); let mut task: TokioTask = Executor::spawn(&executor, async { 100 }); - let waker = create_waker(); - let mut cx = Context::from_waker(&waker); + let mut cx = Context::from_waker(Waker::noop()); match Pin::new(&mut task).poll(&mut cx) { Poll::Ready(result) => assert_eq!(result, 100), @@ -231,8 +220,7 @@ mod tests { let executor = Runtime::new().expect("Failed to create Tokio runtime"); let mut task: TokioTask<&str> = Executor::spawn(&executor, async { "success" }); - let waker = create_waker(); - let mut cx = Context::from_waker(&waker); + let mut cx = Context::from_waker(Waker::noop()); match Pin::new(&mut task).poll_result(&mut cx) { Poll::Ready(Ok(result)) => assert_eq!(result, "success"), @@ -299,8 +287,7 @@ mod tests { let mut task: TokioLocalTask = LocalExecutor::spawn_local(&local_set, async { 200 }); - let waker = create_waker(); - let mut cx = Context::from_waker(&waker); + let mut cx = Context::from_waker(Waker::noop()); match Pin::new(&mut task).poll(&mut cx) { Poll::Ready(result) => assert_eq!(result, 200), @@ -322,8 +309,7 @@ mod tests { let mut task: TokioLocalTask<&str> = LocalExecutor::spawn_local(&local_set, async { "local success" }); - let waker = create_waker(); - let mut cx = Context::from_waker(&waker); + let mut cx = Context::from_waker(Waker::noop()); match Pin::new(&mut task).poll_result(&mut cx) { Poll::Ready(Ok(result)) => assert_eq!(result, "local success"),