From f488aa3cd74ebf657ae519e3012d8a7275637f44 Mon Sep 17 00:00:00 2001 From: Lexo Liu Date: Thu, 27 Aug 2026 11:38:30 -0400 Subject: [PATCH 01/12] ci: add weekly dependency check workflow --- .github/workflows/dep-check.yml | 97 +++++++++++++++++++++++++++++++++ 1 file changed, 97 insertions(+) create mode 100644 .github/workflows/dep-check.yml diff --git a/.github/workflows/dep-check.yml b/.github/workflows/dep-check.yml new file mode 100644 index 0000000..b5f1027 --- /dev/null +++ b/.github/workflows/dep-check.yml @@ -0,0 +1,97 @@ +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: 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 From 66f054e64dafc2014b4fe1a35e71b0b3e0f79e56 Mon Sep 17 00:00:00 2001 From: Lexo Liu Date: Thu, 27 Aug 2026 19:52:38 -0400 Subject: [PATCH 02/12] ci: add rust-cache to dep-check workflow --- .github/workflows/dep-check.yml | 4 ++++ 1 file changed, 4 insertions(+) diff --git a/.github/workflows/dep-check.yml b/.github/workflows/dep-check.yml index b5f1027..c1f00cb 100644 --- a/.github/workflows/dep-check.yml +++ b/.github/workflows/dep-check.yml @@ -32,6 +32,10 @@ jobs: 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: | From 7e3bdea211743ced9c421c531df3cb4244da39c5 Mon Sep 17 00:00:00 2001 From: Lexo Liu Date: Thu, 27 Aug 2026 19:53:26 -0400 Subject: [PATCH 03/12] ci: use stable toolchain --- .github/workflows/ci.yml | 22 +++++++++++----------- 1 file changed, 11 insertions(+), 11 deletions(-) diff --git a/.github/workflows/ci.yml b/.github/workflows/ci.yml index ceb9dea..f28f27f 100644 --- a/.github/workflows/ci.yml +++ b/.github/workflows/ci.yml @@ -17,9 +17,9 @@ jobs: - uses: actions/checkout@v4 - name: Install Rust - uses: dtolnay/rust-toolchain@v1 + uses: dtolnay/rust-toolchain@stable with: - toolchain: nightly + toolchain: stable components: clippy, rustfmt - name: Setup Rust cache @@ -43,7 +43,7 @@ jobs: - uses: actions/checkout@v4 - name: Install Rust 1.85.1 (MSRV) - uses: dtolnay/rust-toolchain@v1 + uses: dtolnay/rust-toolchain@stable with: toolchain: 1.85.1 @@ -70,9 +70,9 @@ jobs: - uses: actions/checkout@v4 - name: Install Rust - uses: dtolnay/rust-toolchain@v1 + uses: dtolnay/rust-toolchain@stable with: - toolchain: nightly + toolchain: stable targets: wasm32-unknown-unknown - name: Setup Rust cache @@ -88,9 +88,9 @@ jobs: - uses: actions/checkout@v4 - name: Install Rust - uses: dtolnay/rust-toolchain@v1 + uses: dtolnay/rust-toolchain@stable with: - toolchain: nightly + toolchain: stable - name: Setup Rust cache uses: Swatinem/rust-cache@v2 @@ -107,9 +107,9 @@ jobs: - uses: actions/checkout@v4 - name: Install Rust - uses: dtolnay/rust-toolchain@v1 + uses: dtolnay/rust-toolchain@stable with: - toolchain: nightly + toolchain: stable components: llvm-tools-preview - name: Setup Rust cache @@ -134,9 +134,9 @@ jobs: - uses: actions/checkout@v4 - name: Install Rust - uses: dtolnay/rust-toolchain@v1 + uses: dtolnay/rust-toolchain@stable with: - toolchain: nightly + toolchain: stable - name: Setup Rust cache uses: Swatinem/rust-cache@v2 From 43a19afa7c5d93f9ba0c08c91ac1d7b7674c85ef Mon Sep 17 00:00:00 2001 From: Lexo Liu Date: Fri, 11 Sep 2026 03:32:57 -0400 Subject: [PATCH 04/12] test: use Waker::noop() instead of a hand-rolled TestWaker Stable clippy's manual_noop_waker lint rejects the manual Wake impl; core::task::Waker::noop() has been stable since 1.85, the crate's MSRV. --- src/async_executor.rs | 24 +++++------------------- src/tokio.rs | 24 +++++------------------- 2 files changed, 10 insertions(+), 38 deletions(-) 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/tokio.rs b/src/tokio.rs index 7766eaf..2a86090 100644 --- a/src/tokio.rs +++ b/src/tokio.rs @@ -156,8 +156,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, @@ -165,15 +164,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"); @@ -198,8 +188,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), @@ -215,8 +204,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"), @@ -283,8 +271,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), @@ -306,8 +293,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"), From 60abda02d473d4b78a21b51f0ebccd5731da961f Mon Sep 17 00:00:00 2001 From: Lexo Liu Date: Fri, 11 Sep 2026 03:35:29 -0400 Subject: [PATCH 05/12] fix: gate the tokio backend to non-wasm targets cargo build --target wasm32-unknown-unknown --all-features pulled in tokio with rt-multi-thread, which tokio rejects on wasm. The tokio backend wraps Runtime and LocalSet, neither of which exists there, so declare the dependency under cfg(not(wasm32)) and gate the module the same way; enabling the tokio feature on wasm now resolves but compiles nothing. --- Cargo.toml | 2 +- src/lib.rs | 2 +- 2 files changed, 2 insertions(+), 2 deletions(-) diff --git a/Cargo.toml b/Cargo.toml index 3f56cfb..b4c95d8 100644 --- a/Cargo.toml +++ b/Cargo.toml @@ -37,7 +37,7 @@ 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"] diff --git a/src/lib.rs b/src/lib.rs index ae2b298..bc5feb0 100644 --- a/src/lib.rs +++ b/src/lib.rs @@ -65,7 +65,7 @@ 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; From 95e8f73e6d5ef42036bd13490c4dedfbad9747cb Mon Sep 17 00:00:00 2001 From: Lexo Liu Date: Sat, 12 Sep 2026 02:08:59 -0400 Subject: [PATCH 06/12] ci: gate pull requests into main so only dev may merge (#7) * ci: gate pull requests into main so only dev may merge * ci: gate pull requests into main so only dev may merge * ci: gate pull requests into main so only dev may merge --- .github/workflows/pr-base-gate.yml | 71 ++++++++++++++++++++++++++++++ 1 file changed, 71 insertions(+) create mode 100644 .github/workflows/pr-base-gate.yml 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 From 1fbab8aa87f1a2251eb934119d8e460feef9e0f4 Mon Sep 17 00:00:00 2001 From: Lexo Liu Date: Sat, 12 Sep 2026 02:09:03 -0400 Subject: [PATCH 07/12] ci: publish to crates.io via OIDC trusted publishing (#6) --- .github/workflows/release.yml | 13 ++++++++++++- 1 file changed, 12 insertions(+), 1 deletion(-) diff --git a/.github/workflows/release.yml b/.github/workflows/release.yml index 5e98f7a..e67c0dc 100644 --- a/.github/workflows/release.yml +++ b/.github/workflows/release.yml @@ -4,6 +4,12 @@ on: push: branches: - main + +permissions: + contents: write + pull-requests: write + id-token: write + jobs: release-plz: name: Release-plz @@ -18,8 +24,13 @@ jobs: 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: ${{ secrets.CARGO_REGISTRY_TOKEN }} + CARGO_REGISTRY_TOKEN: ${{ steps.crates-io-auth.outputs.token }} From 58fa8b8af356fd0f68fc6f0bf7bc190ba7cf1b29 Mon Sep 17 00:00:00 2001 From: Lexo Liu Date: Sun, 13 Sep 2026 23:12:23 -0400 Subject: [PATCH 08/12] ci: run tests with cargo nextest (#10) --- .github/workflows/ci.yml | 10 +++++++--- 1 file changed, 7 insertions(+), 3 deletions(-) diff --git a/.github/workflows/ci.yml b/.github/workflows/ci.yml index f28f27f..9f3d8bb 100644 --- a/.github/workflows/ci.yml +++ b/.github/workflows/ci.yml @@ -53,11 +53,13 @@ jobs: - name: Build run: cargo build --verbose + - uses: taiki-e/install-action@nextest + - name: Run tests - run: cargo test --verbose + run: cargo nextest run - name: Run tests with all features - run: cargo test --verbose --all-features + run: cargo nextest run --all-features - name: Test documentation run: cargo test --doc @@ -118,8 +120,10 @@ jobs: - 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 --all-features --workspace --lcov --output-path lcov.info + run: cargo llvm-cov nextest --all-features --workspace --lcov --output-path lcov.info - name: Upload coverage to Codecov uses: codecov/codecov-action@v4 From c22143c953cc92e60e6766952b9fc440794eb8dc Mon Sep 17 00:00:00 2001 From: Lexo Liu Date: Mon, 9 Nov 2026 17:25:57 -0500 Subject: [PATCH 09/12] feat: compio backend Add a compio feature implementing LocalExecutor for compio's Runtime and Executor for its thread-per-core Dispatcher, with cancel-on-drop, detach and panic reporting on both. --- Cargo.toml | 6 +- src/compio.rs | 422 ++++++++++++++++++++++++++++++++++++++++++++++++++ src/lib.rs | 4 + 3 files changed, 430 insertions(+), 2 deletions(-) create mode 100644 src/compio.rs diff --git a/Cargo.toml b/Cargo.toml index b4c95d8..114b3fd 100644 --- a/Cargo.toml +++ b/Cargo.toml @@ -19,11 +19,12 @@ license = { file = "LICENSE" } [package.metadata.docs.rs] -features = ["async-task", "async-executor", "tokio", "std"] +features = ["async-task", "async-executor", "tokio", "compio", "std"] rustdoc-args = ["--cfg", "docsrs"] [dependencies] async-channel = { version = "2.5.0", default-features = false } +compio = { version = "0.19.2", default-features = false, features = ["dispatcher", "runtime"], optional = true } pin-project-lite = { version = "0.2", default-features = false} take_mut = "0.2.2" @@ -49,9 +50,10 @@ std = ["async-task?/std"] async-task = ["dep:async-task"] async-executor = ["dep:async-executor", "async-task", "std"] 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/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 bc5feb0..1f3dbcf 100644 --- a/src/lib.rs +++ b/src/lib.rs @@ -69,6 +69,10 @@ pub mod async_executor; #[cfg_attr(docsrs, doc(cfg(feature = "tokio")))] pub mod tokio; +#[cfg(all(feature = "compio", not(target_arch = "wasm32")))] +#[cfg_attr(docsrs, doc(cfg(feature = "compio")))] +pub mod compio; + use core::{ any::Any, fmt::Debug, From 42676c0abc30e843b7c1c1dc852d8aa68b63a679 Mon Sep 17 00:00:00 2001 From: Lexo Liu Date: Mon, 9 Nov 2026 17:32:37 -0500 Subject: [PATCH 10/12] build: gate compio off wasm32 and raise the MSRV to 1.93 compio-buf uses slice MaybeUninit::assume_init_mut, stable since 1.93. --- .github/workflows/ci.yml | 4 ++-- Cargo.toml | 9 +++++++-- 2 files changed, 9 insertions(+), 4 deletions(-) diff --git a/.github/workflows/ci.yml b/.github/workflows/ci.yml index 9f3d8bb..4a82705 100644 --- a/.github/workflows/ci.yml +++ b/.github/workflows/ci.yml @@ -42,10 +42,10 @@ jobs: steps: - uses: actions/checkout@v4 - - name: Install Rust 1.85.1 (MSRV) + - name: Install Rust 1.93 (MSRV) uses: dtolnay/rust-toolchain@stable with: - toolchain: 1.85.1 + toolchain: "1.93" - name: Setup Rust cache uses: Swatinem/rust-cache@v2 diff --git a/Cargo.toml b/Cargo.toml index 114b3fd..44dfecb 100644 --- a/Cargo.toml +++ b/Cargo.toml @@ -3,7 +3,7 @@ name = "executor-core" version = "0.6.0" authors = ["Lexo Liu "] edition = "2024" -rust-version = "1.85.1" +rust-version = "1.93" 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" @@ -24,7 +24,6 @@ rustdoc-args = ["--cfg", "docsrs"] [dependencies] async-channel = { version = "2.5.0", default-features = false } -compio = { version = "0.19.2", default-features = false, features = ["dispatcher", "runtime"], optional = true } pin-project-lite = { version = "0.2", default-features = false} take_mut = "0.2.2" @@ -44,6 +43,12 @@ default-features = false features = ["rt", "rt-multi-thread"] optional = true +[target.'cfg(not(target_arch = "wasm32"))'.dependencies.compio] +version = "0.19.2" +default-features = false +features = ["dispatcher", "runtime"] +optional = true + [features] default = ["full"] std = ["async-task?/std"] From 36791f1087bfbadb85443d654da857713a20548d Mon Sep 17 00:00:00 2001 From: Lexo Liu Date: Mon, 9 Nov 2026 17:36:06 -0500 Subject: [PATCH 11/12] build: raise the MSRV to 1.95 compio-executor uses cfg_select!, stable since 1.95. --- .github/workflows/ci.yml | 4 ++-- Cargo.toml | 2 +- 2 files changed, 3 insertions(+), 3 deletions(-) diff --git a/.github/workflows/ci.yml b/.github/workflows/ci.yml index 4a82705..a9dc62d 100644 --- a/.github/workflows/ci.yml +++ b/.github/workflows/ci.yml @@ -42,10 +42,10 @@ jobs: steps: - uses: actions/checkout@v4 - - name: Install Rust 1.93 (MSRV) + - name: Install Rust 1.95 (MSRV) uses: dtolnay/rust-toolchain@stable with: - toolchain: "1.93" + toolchain: "1.95" - name: Setup Rust cache uses: Swatinem/rust-cache@v2 diff --git a/Cargo.toml b/Cargo.toml index 44dfecb..010bf40 100644 --- a/Cargo.toml +++ b/Cargo.toml @@ -3,7 +3,7 @@ name = "executor-core" version = "0.6.0" authors = ["Lexo Liu "] edition = "2024" -rust-version = "1.93" +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" From d6ee69993dcb431532107b6290b089955e79ca86 Mon Sep 17 00:00:00 2001 From: Lexo Liu Date: Mon, 9 Nov 2026 17:46:20 -0500 Subject: [PATCH 12/12] ci: accept an empty test run for the default feature set --- .github/workflows/ci.yml | 3 ++- 1 file changed, 2 insertions(+), 1 deletion(-) diff --git a/.github/workflows/ci.yml b/.github/workflows/ci.yml index 4ac128f..547624d 100644 --- a/.github/workflows/ci.yml +++ b/.github/workflows/ci.yml @@ -55,8 +55,9 @@ jobs: - 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 + run: cargo nextest run --no-tests=pass - name: Run tests with all features run: cargo nextest run --all-features