From d04f0f90e6756e8b326d328860ba3ba2504feaa5 Mon Sep 17 00:00:00 2001 From: Darius Clark Date: Mon, 24 Aug 2026 18:48:36 -0400 Subject: [PATCH 01/11] feat: implement compio executor --- Cargo.toml | 5 +- src/error.rs | 10 + src/lib.rs | 125 ++++++++++++- src/rt/compio.rs | 461 +++++++++++++++++++++++++++++++++++++++++++++++ src/rt/mod.rs | 3 + 5 files changed, 602 insertions(+), 2 deletions(-) create mode 100644 src/rt/compio.rs diff --git a/Cargo.toml b/Cargo.toml index 7e497ad..1caa230 100644 --- a/Cargo.toml +++ b/Cargo.toml @@ -18,6 +18,7 @@ A small library designed to utilize async executors through an common API while default = ["tokio"] threadpool = ["futures/thread-pool"] tokio = ["dep:tokio"] +compio = ["dep:compio"] [dependencies] futures = { version = "0.3.31", default-features = false, features = ["std", "alloc", "async-await"] } @@ -28,7 +29,8 @@ either = { version = "1.15.0", optional = true} thiserror = "1.0.69" [target.'cfg(not(target_arch = "wasm32"))'.dependencies] -tokio = { version = "1.49.0", optional = true, features = ["rt", "rt-multi-thread"] } +tokio = { version = "1.49.0", optional = true, features = ["rt", "rt-multi-thread"] } +compio = { version = "0.19", optional = true, default-features = false, features = ["runtime"] } [target.'cfg(target_arch = "wasm32")'.dependencies] wasm-bindgen-futures = "0.4.50" @@ -36,4 +38,5 @@ wasm-bindgen-futures = "0.4.50" [dev-dependencies] futures-timer = "3.0.3" tokio = { version = "1.49.0", features = ["rt", "rt-multi-thread", "macros"] } +compio = { version = "0.19", default-features = false, features = ["runtime", "macros"]} futures = { version = "0.3.31", default-features = false, features = ["std", "alloc", "async-await", "thread-pool", "executor"] } \ No newline at end of file diff --git a/src/error.rs b/src/error.rs index 35cdfa3..e599297 100644 --- a/src/error.rs +++ b/src/error.rs @@ -37,6 +37,16 @@ impl From for JoinError { } } +#[cfg(all(feature = "compio", not(target_arch = "wasm32")))] +impl From for JoinError { + fn from(err: compio::runtime::JoinError) -> Self { + match err { + compio::runtime::JoinError::Cancelled => JoinError::Cancelled, + compio::runtime::JoinError::Panicked(_) => JoinError::Panicked, + } + } +} + /// Error indicating a task did not complete before its timeout elapsed. /// /// Returned as the inner error of a timeout task's output, for example from diff --git a/src/lib.rs b/src/lib.rs index 94e64dd..73a80c2 100644 --- a/src/lib.rs +++ b/src/lib.rs @@ -22,12 +22,20 @@ use futures::future::{AbortHandle, AbortRegistration, Abortable}; use futures::task::AtomicWaker; use futures::{FutureExt, SinkExt, StreamExt, TryFutureExt}; use futures_timeout::Timeout; +#[cfg(all(feature = "compio", not(target_arch = "wasm32")))] +use parking_lot::Mutex; use pollable_map::optional::Optional; use std::future::Future; use std::pin::Pin; use std::sync::Arc; use std::task::{Context, Poll}; +#[cfg(all(feature = "compio", not(target_arch = "wasm32")))] +type CancelFuture = Pin> + Send + 'static>>; + +#[cfg(all(feature = "compio", not(target_arch = "wasm32")))] +type StartCompioCancel = fn(&Mutex>, &AtomicBool); + #[cfg_attr(feature = "tokio", allow(dead_code))] pub(crate) struct CompletionGuard { finished: Arc, @@ -65,10 +73,11 @@ where #[cfg(all( not(feature = "threadpool"), not(feature = "tokio"), + not(feature = "compio"), not(target_arch = "wasm32") ))] compile_error!( - "At least one runtime (i.e 'tokio', 'threadpool', 'wasm-bindgen-futures') must be enabled" + "At least one runtime (i.e `compio`, 'tokio', 'threadpool', 'wasm-bindgen-futures') must be enabled" ); /// An owned permission to join on a task (await its termination). @@ -99,6 +108,12 @@ enum InnerJoinHandle { handle: Optional<::tokio::task::JoinHandle>, abort_requested: AtomicBool, }, + #[cfg(all(feature = "compio", not(target_arch = "wasm32")))] + CompioHandle { + handle: Mutex>, + abort_requested: AtomicBool, + start_cancel: StartCompioCancel, + }, #[allow(dead_code)] CustomHandle { inner: Optional>>, @@ -109,6 +124,71 @@ enum InnerJoinHandle { Empty, } +#[cfg(all(feature = "compio", not(target_arch = "wasm32")))] +enum InnerCompioHandle { + CompioHandle(Option<::compio::runtime::JoinHandle>), + CompioFuture(Optional>), +} + +#[cfg(all(feature = "compio", not(target_arch = "wasm32")))] +impl Future for InnerCompioHandle { + type Output = Result, JoinError>; + fn poll(mut self: Pin<&mut Self>, cx: &mut Context<'_>) -> Poll { + let this = &mut *self; + + match this { + InnerCompioHandle::CompioHandle(handle) => { + let Some(mut handle) = handle.take() else { + return Poll::Ready(Ok(None)); + }; + match Pin::new(&mut handle).poll(cx) { + Poll::Ready(result) => Poll::Ready(result.map(Some).map_err(JoinError::from)), + Poll::Pending => { + *this = InnerCompioHandle::CompioHandle(Some(handle)); + Poll::Pending + } + } + } + InnerCompioHandle::CompioFuture(future) => match Pin::new(future).poll(cx) { + Poll::Ready(result) => Poll::Ready(Ok(result)), + Poll::Pending => Poll::Pending, + }, + } + } +} + +#[cfg(all(feature = "compio", not(target_arch = "wasm32")))] +fn start_compio_cancel( + state: &Mutex>, + requested: &AtomicBool, +) { + let mut state = state.lock(); + + let InnerCompioHandle::CompioHandle(slot) = &mut *state else { + return; + }; + let Some(handle) = slot.take() else { + return; + }; + + requested.store(true, Ordering::Release); + + let future: CancelFuture = Box::pin(handle.cancel()); + *state = InnerCompioHandle::CompioFuture(Optional::new(future)); +} + +#[cfg(all(feature = "compio", not(target_arch = "wasm32")))] +impl InnerCompioHandle { + fn is_finished(&self) -> bool { + match self { + InnerCompioHandle::CompioHandle(handle) => { + handle.as_ref().map(|h| h.is_finished()).unwrap_or(true) + } + InnerCompioHandle::CompioFuture(_) => true, + } + } +} + impl InnerJoinHandle { #[cfg(all(feature = "tokio", not(target_arch = "wasm32")))] fn tokio(handle: ::tokio::task::JoinHandle) -> Self { @@ -117,6 +197,18 @@ impl InnerJoinHandle { abort_requested: AtomicBool::new(false), } } + + #[cfg(all(feature = "compio", not(target_arch = "wasm32")))] + fn compio(handle: ::compio::runtime::JoinHandle) -> Self + where + T: Send + 'static, + { + Self::CompioHandle { + handle: Mutex::new(InnerCompioHandle::CompioHandle(Some(handle))), + abort_requested: AtomicBool::new(false), + start_cancel: start_compio_cancel::, + } + } } impl JoinHandle { @@ -146,6 +238,14 @@ impl JoinHandle { handle.abort(); } } + #[cfg(all(feature = "compio", not(target_arch = "wasm32")))] + InnerJoinHandle::CompioHandle { + handle, + abort_requested, + start_cancel, + } => { + start_cancel(handle, abort_requested); + } InnerJoinHandle::CustomHandle { handle, .. } => handle.abort(), InnerJoinHandle::Empty => {} } @@ -161,6 +261,8 @@ impl JoinHandle { InnerJoinHandle::TokioHandle { handle, .. } => { handle.as_ref().map(|h| h.is_finished()).unwrap_or(true) } + #[cfg(all(feature = "compio", not(target_arch = "wasm32")))] + InnerJoinHandle::CompioHandle { handle, .. } => handle.lock().is_finished(), InnerJoinHandle::CustomHandle { inner, finished, .. } => finished.load(Ordering::Acquire) || inner.is_none(), @@ -214,6 +316,27 @@ impl Future for JoinHandle { Err(e) => Poll::Ready(Err(e.into())), } } + #[cfg(all(feature = "compio", not(target_arch = "wasm32")))] + InnerJoinHandle::CompioHandle { + handle, + abort_requested, + .. + } => { + let handle = &mut *handle.lock(); + let fut = futures::ready!(Pin::new(handle).poll(cx)); + + match fut { + Ok(Some(val)) => Poll::Ready(Ok(val)), + Ok(None) if abort_requested.load(Ordering::Acquire) => { + Poll::Ready(Err(JoinError::Aborted)) + } + Ok(None) => Poll::Ready(Err(JoinError::Empty)), + Err(_) if abort_requested.load(Ordering::Acquire) => { + Poll::Ready(Err(JoinError::Aborted)) + } + Err(e) => Poll::Ready(Err(e.into())), + } + } InnerJoinHandle::CustomHandle { inner, .. } => { let fut = futures::ready!(Pin::new(inner).poll(cx)); match fut { diff --git a/src/rt/compio.rs b/src/rt/compio.rs new file mode 100644 index 0000000..d093bb1 --- /dev/null +++ b/src/rt/compio.rs @@ -0,0 +1,461 @@ +use crate::{Executor, ExecutorBlocking, ExecutorTimeout, InnerJoinHandle, JoinHandle}; +use compio::runtime::Runtime; + +/// Compio executor +#[derive(Clone, Copy, Debug, PartialOrd, PartialEq, Eq)] +pub struct CompioExecutor; + +impl Executor for CompioExecutor { + fn spawn(&self, future: F) -> JoinHandle + where + F: Future + Send + 'static, + F::Output: Send + 'static, + { + let handle = compio::runtime::spawn(future); + let inner = InnerJoinHandle::compio(handle); + JoinHandle { inner } + } +} + +impl ExecutorBlocking for CompioExecutor { + fn spawn_blocking(&self, f: F) -> JoinHandle + where + F: FnOnce() -> R + Send + 'static, + R: Send + 'static, + { + let handle = compio::runtime::spawn_blocking(f); + let inner = InnerJoinHandle::compio(handle); + JoinHandle { inner } + } +} + +impl ExecutorTimeout for CompioExecutor {} + +/// Compio executor with an [`Runtime`] +#[derive(Clone, Debug)] +pub struct CompioRuntimeExecutor { + _runtime: Option, +} + +impl CompioRuntimeExecutor { + /// Creates a compio runtime. + pub fn new() -> std::io::Result { + let runtime = Runtime::builder().build()?; + Ok(Self::with_runtime(runtime)) + } + + /// Create an executor with the supplied [`Runtime`]. + /// + /// Note that the runtime remains alive until the final clone of this executor is + /// dropped. + pub fn with_runtime(runtime: Runtime) -> Self { + Self { + _runtime: Some(runtime), + } + } + + /// Create an executor from the existing Runtime + /// + /// Note that this returns an error when called outside a Compio runtime context. The + /// resulting executor does not prevent that runtime from shutting down. + pub fn from_current_runtime() -> std::io::Result { + let handle = + Runtime::try_current().ok_or_else(|| std::io::Error::other("no runtime running"))?; + Ok(Self::with_runtime(handle)) + } +} + +impl Executor for CompioRuntimeExecutor { + fn spawn(&self, future: F) -> JoinHandle + where + F: Future + Send + 'static, + F::Output: Send + 'static, + { + let handle = self + ._runtime + .as_ref() + .expect("Compio runtime is unavailable") + .spawn(future); + let inner = InnerJoinHandle::compio(handle); + JoinHandle { inner } + } +} + +impl ExecutorBlocking for CompioRuntimeExecutor { + fn spawn_blocking(&self, f: F) -> JoinHandle + where + F: FnOnce() -> R + Send + 'static, + R: Send + 'static, + { + let handle = self + ._runtime + .as_ref() + .expect("Compio runtime is unavailable") + .spawn_blocking(f); + let inner = InnerJoinHandle::compio(handle); + JoinHandle { inner } + } +} + +impl ExecutorTimeout for CompioRuntimeExecutor {} + +#[cfg(test)] +mod tests { + use super::{CompioExecutor, CompioRuntimeExecutor}; + use crate::error::JoinError; + use crate::{Executor, ExecutorBlocking, ExecutorTimeout, TimeoutError}; + use futures::channel::mpsc::{Receiver, UnboundedReceiver}; + use futures_timer::Delay; + + #[compio::test] + async fn explicit_abort_is_reported_as_aborted() { + let handle = CompioExecutor.spawn(futures::future::pending::<()>()); + + handle.abort(); + + assert!(matches!(handle.await, Err(JoinError::Aborted))); + } + + #[cfg(panic = "unwind")] + #[compio::test] + async fn task_panic_is_reported_as_panicked() { + async fn panic_task() -> usize { + panic!("expected task panic"); + } + + let handle = CompioExecutor.spawn(panic_task()); + + assert!(matches!(handle.await, Err(JoinError::Panicked))); + } + + #[test] + fn runtime_shutdown_is_reported_as_cancelled() { + let executor = CompioRuntimeExecutor::new().unwrap(); + let handle = executor.spawn(futures::future::pending::<()>()); + + drop(executor); + + assert!(matches!( + futures::executor::block_on(handle), + Err(JoinError::Cancelled) + )); + } + + #[compio::test] + async fn default_abortable_task() { + let executor = CompioExecutor; + + async fn task(tx: futures::channel::oneshot::Sender<()>) { + futures_timer::Delay::new(std::time::Duration::from_secs(5)).await; + let _ = tx.send(()); + unreachable!(); + } + + let (tx, rx) = futures::channel::oneshot::channel::<()>(); + + let handle = executor.spawn_abortable(task(tx)); + + drop(handle); + let result = rx.await; + assert!(result.is_err()); + } + + #[compio::test] + async fn task_coroutine() { + let executor = CompioExecutor; + + enum Message { + Send(String, futures::channel::oneshot::Sender), + } + + let mut task = executor.spawn_coroutine(|msg: Message| async move { + match msg { + Message::Send(msg, sender) => { + sender.send(msg).unwrap(); + } + } + }); + + let (tx, rx) = futures::channel::oneshot::channel::(); + let msg = Message::Send("Hello".into(), tx); + + task.send(msg).await.unwrap(); + let resp = rx.await.unwrap(); + assert_eq!(resp, "Hello"); + } + + #[compio::test] + async fn task_coroutine_with_context() { + let executor = CompioExecutor; + + type Resp = futures::channel::oneshot::Sender; + + let mut task = + executor.spawn_coroutine_with_context(0usize, |counter: &mut usize, resp: Resp| { + *counter += 1; + let n = *counter; + async move { + resp.send(n).unwrap(); + } + }); + + let (tx1, rx1) = futures::channel::oneshot::channel::(); + let (tx2, rx2) = futures::channel::oneshot::channel::(); + task.send(tx1).await.unwrap(); + task.send(tx2).await.unwrap(); + assert_eq!(rx1.await.unwrap(), 1); + assert_eq!(rx2.await.unwrap(), 2); + } + + #[compio::test] + async fn task_coroutine_with_receiver() { + use futures::stream::StreamExt; + let executor = CompioExecutor; + + enum Message { + Send(String, futures::channel::oneshot::Sender), + } + + let mut task = + executor.spawn_coroutine_with_receiver(|mut rx: Receiver| async move { + while let Some(msg) = rx.next().await { + match msg { + Message::Send(msg, sender) => { + sender.send(msg).unwrap(); + } + } + } + }); + + let (tx, rx) = futures::channel::oneshot::channel::(); + let msg = Message::Send("Hello".into(), tx); + + task.send(msg).await.unwrap(); + let resp = rx.await.unwrap(); + assert_eq!(resp, "Hello"); + } + + #[compio::test] + async fn task_coroutine_with_receiver_and_context() { + use futures::stream::StreamExt; + let executor = CompioExecutor; + + #[derive(Default)] + struct State { + message: String, + } + + enum Message { + Set(String), + Get(futures::channel::oneshot::Sender), + } + + let mut task = executor.spawn_coroutine_with_receiver_and_context( + State::default(), + |mut state, mut rx: Receiver| async move { + while let Some(msg) = rx.next().await { + match msg { + Message::Set(msg) => { + state.message = msg; + } + Message::Get(resp) => { + resp.send(state.message.clone()).unwrap(); + } + } + } + }, + ); + + let msg = Message::Set("Hello".into()); + + task.send(msg).await.unwrap(); + let (tx, rx) = futures::channel::oneshot::channel::(); + let msg = Message::Get(tx); + task.send(msg).await.unwrap(); + let resp = rx.await.unwrap(); + assert_eq!(resp, "Hello"); + } + + #[compio::test] + async fn task_unbounded_coroutine() { + let executor = CompioExecutor; + + enum Message { + Send(String, futures::channel::oneshot::Sender), + } + + let mut task = executor.spawn_unbounded_coroutine(|msg: Message| async move { + match msg { + Message::Send(msg, sender) => { + sender.send(msg).unwrap(); + } + } + }); + + let (tx, rx) = futures::channel::oneshot::channel::(); + let msg = Message::Send("Hello".into(), tx); + + task.send(msg).unwrap(); + let resp = rx.await.unwrap(); + assert_eq!(resp, "Hello"); + } + + #[compio::test] + async fn task_unbounded_coroutine_with_context() { + let executor = CompioExecutor; + + type Resp = futures::channel::oneshot::Sender; + + let mut task = executor.spawn_unbounded_coroutine_with_context( + 0usize, + |counter: &mut usize, resp: Resp| { + *counter += 1; + let n = *counter; + async move { + resp.send(n).unwrap(); + } + }, + ); + + let (tx1, rx1) = futures::channel::oneshot::channel::(); + let (tx2, rx2) = futures::channel::oneshot::channel::(); + task.send(tx1).unwrap(); + task.send(tx2).unwrap(); + assert_eq!(rx1.await.unwrap(), 1); + assert_eq!(rx2.await.unwrap(), 2); + } + + #[compio::test] + async fn task_unbounded_coroutine_with_receiver() { + use futures::stream::StreamExt; + let executor = CompioExecutor; + + enum Message { + Send(String, futures::channel::oneshot::Sender), + } + + let mut task = executor.spawn_unbounded_coroutine_with_receiver( + |mut rx: UnboundedReceiver| async move { + while let Some(msg) = rx.next().await { + match msg { + Message::Send(msg, sender) => { + sender.send(msg).unwrap(); + } + } + } + }, + ); + + let (tx, rx) = futures::channel::oneshot::channel::(); + let msg = Message::Send("Hello".into(), tx); + + task.send(msg).unwrap(); + let resp = rx.await.unwrap(); + assert_eq!(resp, "Hello"); + } + + #[compio::test] + async fn task_unbounded_coroutine_with_receiver_and_context() { + use futures::stream::StreamExt; + let executor = CompioExecutor; + + #[derive(Default)] + struct State { + message: String, + } + + enum Message { + Set(String), + Get(futures::channel::oneshot::Sender), + } + + let mut task = executor.spawn_unbounded_coroutine_with_receiver_and_context( + State::default(), + |mut state, mut rx: UnboundedReceiver| async move { + while let Some(msg) = rx.next().await { + match msg { + Message::Set(msg) => { + state.message = msg; + } + Message::Get(resp) => { + resp.send(state.message.clone()).unwrap(); + } + } + } + }, + ); + + let msg = Message::Set("Hello".into()); + + task.send(msg).unwrap(); + let (tx, rx) = futures::channel::oneshot::channel::(); + let msg = Message::Get(tx); + task.send(msg).unwrap(); + let resp = rx.await.unwrap(); + assert_eq!(resp, "Hello"); + } + + #[compio::test] + async fn timeout_task() { + let executor = CompioExecutor; + + let task = executor.spawn_timeout( + std::time::Duration::from_millis(10), + futures::future::pending::<()>(), + ); + let resp = task.await.unwrap(); + assert!(matches!(resp.unwrap_err(), TimeoutError)); + } + + #[compio::test] + async fn complete_before_timeout_task() { + let executor = CompioExecutor; + + let task = executor.spawn_timeout( + std::time::Duration::from_millis(10), + futures::future::ready("Hello"), + ); + let resp = task.await.unwrap(); + assert!(resp.is_ok()); + let result = resp.unwrap(); + assert_eq!(result, "Hello"); + } + + #[compio::test] + async fn race_before_timeout_task() { + let executor = CompioExecutor; + + let task = executor.spawn_timeout(std::time::Duration::from_millis(500), async { + Delay::new(std::time::Duration::from_millis(10)).await; + "Hello" + }); + let resp = task.await.unwrap(); + assert!(resp.is_ok()); + let result = resp.unwrap(); + assert_eq!(result, "Hello"); + } + + #[compio::test] + async fn abortable_timeout_task() { + let executor = CompioExecutor; + + let task = executor.spawn_abortable_timeout( + std::time::Duration::from_millis(10), + futures::future::pending::<()>(), + ); + let resp = task.await.unwrap(); + assert!(matches!(resp.unwrap_err(), TimeoutError)); + } + + #[compio::test] + async fn blocking_task() { + let executor = CompioExecutor; + + let task = executor.spawn_blocking(|| { + std::thread::sleep(std::time::Duration::from_millis(100)); + "Hello" + }); + let resp = task.await.unwrap(); + assert_eq!(resp, "Hello"); + } +} diff --git a/src/rt/mod.rs b/src/rt/mod.rs index 07b8b10..e4df8d0 100644 --- a/src/rt/mod.rs +++ b/src/rt/mod.rs @@ -6,3 +6,6 @@ pub mod tokio; #[cfg(target_arch = "wasm32")] pub mod wasm; + +#[cfg(all(feature = "compio", not(target_arch = "wasm32")))] +pub mod compio; From b532be26ce7faea4dbe1cf103e86ebc2e92a85b1 Mon Sep 17 00:00:00 2001 From: Darius Clark Date: Mon, 24 Aug 2026 19:59:19 -0400 Subject: [PATCH 02/11] fix: add custom drop to detach task if compio handle is dropped --- src/lib.rs | 15 +++++++++++++++ 1 file changed, 15 insertions(+) diff --git a/src/lib.rs b/src/lib.rs index 73a80c2..2e07df6 100644 --- a/src/lib.rs +++ b/src/lib.rs @@ -130,6 +130,21 @@ enum InnerCompioHandle { CompioFuture(Optional>), } +// Note: We implement a custom drop due to compio `JoinHandle` cancellation of the taks upon drop +// therefore we need to detach the handle upon drop so it will continue running in the +// background to follow the same flow as the other executors implementations. +#[cfg(all(feature = "compio", not(target_arch = "wasm32")))] +impl Drop for InnerCompioHandle { + fn drop(&mut self) { + if let InnerCompioHandle::CompioHandle(handle) = self { + let Some(handle) = handle.take() else { + return; + }; + handle.detach(); + } + } +} + #[cfg(all(feature = "compio", not(target_arch = "wasm32")))] impl Future for InnerCompioHandle { type Output = Result, JoinError>; From 0d5fe3fe4ea79087c8e661876966bfbc04c6db30 Mon Sep 17 00:00:00 2001 From: Darius Clark Date: Mon, 24 Aug 2026 20:06:34 -0400 Subject: [PATCH 03/11] feat: impl compio in GlobalExecutor --- src/global.rs | 30 ++++++++++++++++++++++++++++-- 1 file changed, 28 insertions(+), 2 deletions(-) diff --git a/src/global.rs b/src/global.rs index 899efe9..ac28215 100644 --- a/src/global.rs +++ b/src/global.rs @@ -23,10 +23,22 @@ impl Executor for GlobalExecutor { } #[cfg(all( - feature = "threadpool", + feature = "compio", not(any(feature = "tokio", target_arch = "wasm32")) ))] fn spawn(&self, future: F) -> JoinHandle + where + F: Future + Send + 'static, + F::Output: Send + 'static, + { + crate::rt::compio::CompioExecutor.spawn(future) + } + + #[cfg(all( + feature = "threadpool", + not(any(feature = "tokio", feature = "compio", target_arch = "wasm32")) + ))] + fn spawn(&self, future: F) -> JoinHandle where F: Future + Send + 'static, F::Output: Send + 'static, @@ -46,6 +58,7 @@ impl Executor for GlobalExecutor { #[cfg(all( not(feature = "threadpool"), not(feature = "tokio"), + not(feature = "compio"), not(target_arch = "wasm32") ))] fn spawn(&self, _: F) -> JoinHandle @@ -68,10 +81,22 @@ impl ExecutorBlocking for GlobalExecutor { } #[cfg(all( - feature = "threadpool", + feature = "compio", not(any(feature = "tokio", target_arch = "wasm32")) ))] fn spawn_blocking(&self, f: F) -> JoinHandle + where + F: FnOnce() -> R + Send + 'static, + R: Send + 'static, + { + crate::rt::compio::CompioExecutor.spawn_blocking(f) + } + + #[cfg(all( + feature = "threadpool", + not(any(feature = "tokio", feature = "compio", target_arch = "wasm32")) + ))] + fn spawn_blocking(&self, f: F) -> JoinHandle where F: FnOnce() -> R + Send + 'static, R: Send + 'static, @@ -91,6 +116,7 @@ impl ExecutorBlocking for GlobalExecutor { #[cfg(all( not(feature = "threadpool"), not(feature = "tokio"), + not(feature = "compio"), not(target_arch = "wasm32") ))] fn spawn_blocking(&self, _: F) -> JoinHandle From 063412bce20ccb02b6fa2cabe9ebb64f0c4427b1 Mon Sep 17 00:00:00 2001 From: Darius Clark Date: Mon, 24 Aug 2026 20:09:08 -0400 Subject: [PATCH 04/11] chore: update rustdoc --- src/global.rs | 4 ++-- 1 file changed, 2 insertions(+), 2 deletions(-) diff --git a/src/global.rs b/src/global.rs index ac28215..ed5468f 100644 --- a/src/global.rs +++ b/src/global.rs @@ -3,8 +3,8 @@ use std::future::Future; /// Executor that selects an available runtime backend at compile time. /// -/// * On non-Wasm targets with the `tokio` feature enabled, it uses -/// `TokioExecutor`. +/// * On non-Wasm targets with the `tokio` or `compio` feature enabled, it uses +/// `TokioExecutor` or `CompioExecutor`. /// * On non-Wasm targets with the `threadpool` feature enabled and the `tokio` /// feature disabled, it uses `ThreadPoolExecutor`. /// * On Wasm targets, it uses `WasmExecutor`, backed by From 0f504a9188f50b80064664a37fea1e7170c4f67a Mon Sep 17 00:00:00 2001 From: Darius Clark Date: Mon, 24 Aug 2026 20:11:37 -0400 Subject: [PATCH 05/11] chore: clippy cleanup --- src/lib.rs | 2 +- 1 file changed, 1 insertion(+), 1 deletion(-) diff --git a/src/lib.rs b/src/lib.rs index 2e07df6..5a8fb74 100644 --- a/src/lib.rs +++ b/src/lib.rs @@ -349,7 +349,7 @@ impl Future for JoinHandle { Err(_) if abort_requested.load(Ordering::Acquire) => { Poll::Ready(Err(JoinError::Aborted)) } - Err(e) => Poll::Ready(Err(e.into())), + Err(e) => Poll::Ready(Err(e)), } } InnerJoinHandle::CustomHandle { inner, .. } => { From 15ad92b48ffaabe549d13d21cc4b4baf2b32e1c1 Mon Sep 17 00:00:00 2001 From: Darius Clark Date: Tue, 25 Aug 2026 08:01:21 -0400 Subject: [PATCH 06/11] chore: update comment --- src/rt/compio.rs | 3 --- 1 file changed, 3 deletions(-) diff --git a/src/rt/compio.rs b/src/rt/compio.rs index d093bb1..70a807c 100644 --- a/src/rt/compio.rs +++ b/src/rt/compio.rs @@ -45,9 +45,6 @@ impl CompioRuntimeExecutor { } /// Create an executor with the supplied [`Runtime`]. - /// - /// Note that the runtime remains alive until the final clone of this executor is - /// dropped. pub fn with_runtime(runtime: Runtime) -> Self { Self { _runtime: Some(runtime), From 45cd978c8a691d00760ebc76e69485c48ba5ded9 Mon Sep 17 00:00:00 2001 From: Darius Clark Date: Tue, 25 Aug 2026 10:19:02 -0400 Subject: [PATCH 07/11] fix: impl stronger check for state management of compio for aborting --- src/lib.rs | 101 +++++++++++++++++++++++++++++++++++++++-------- src/rt/compio.rs | 32 +++++++++++++++ 2 files changed, 117 insertions(+), 16 deletions(-) diff --git a/src/lib.rs b/src/lib.rs index 5a8fb74..0a331d7 100644 --- a/src/lib.rs +++ b/src/lib.rs @@ -31,7 +31,7 @@ use std::sync::Arc; use std::task::{Context, Poll}; #[cfg(all(feature = "compio", not(target_arch = "wasm32")))] -type CancelFuture = Pin> + Send + 'static>>; +type BoxCancelFuture = Pin> + Send + 'static>>; #[cfg(all(feature = "compio", not(target_arch = "wasm32")))] type StartCompioCancel = fn(&Mutex>, &AtomicBool); @@ -126,17 +126,67 @@ enum InnerJoinHandle { #[cfg(all(feature = "compio", not(target_arch = "wasm32")))] enum InnerCompioHandle { - CompioHandle(Option<::compio::runtime::JoinHandle>), - CompioFuture(Optional>), + Running(Option<::compio::runtime::JoinHandle>), + Cancelling(CompioCancel), + Ready(Option>), } -// Note: We implement a custom drop due to compio `JoinHandle` cancellation of the taks upon drop +#[cfg(all(feature = "compio", not(target_arch = "wasm32")))] +impl Unpin for InnerCompioHandle {} + +#[cfg(all(feature = "compio", not(target_arch = "wasm32")))] +struct CompioCancelWake { + outer: AtomicWaker, +} + +#[cfg(all(feature = "compio", not(target_arch = "wasm32")))] +impl futures::task::ArcWake for CompioCancelWake { + fn wake_by_ref(arc_self: &Arc) { + arc_self.outer.wake(); + } +} + +#[cfg(all(feature = "compio", not(target_arch = "wasm32")))] +struct CompioCancel { + future: BoxCancelFuture, + wake: Arc, +} + +#[cfg(all(feature = "compio", not(target_arch = "wasm32")))] +impl CompioCancel { + fn new(future: BoxCancelFuture) -> Self { + Self { + future, + wake: Arc::new(CompioCancelWake { + outer: AtomicWaker::new(), + }), + } + } + + fn poll_inner(&mut self) -> Poll> { + let waker = futures::task::waker(self.wake.clone()); + let mut cx = Context::from_waker(&waker); + self.future.as_mut().poll(&mut cx) + } +} + +#[cfg(all(feature = "compio", not(target_arch = "wasm32")))] +impl Future for CompioCancel { + type Output = Option; + + fn poll(mut self: Pin<&mut Self>, cx: &mut Context<'_>) -> Poll { + self.wake.outer.register(cx.waker()); + self.poll_inner() + } +} + +// Note: We implement a custom drop due to compio `JoinHandle` cancellation of the task upon drop // therefore we need to detach the handle upon drop so it will continue running in the // background to follow the same flow as the other executors implementations. #[cfg(all(feature = "compio", not(target_arch = "wasm32")))] impl Drop for InnerCompioHandle { fn drop(&mut self) { - if let InnerCompioHandle::CompioHandle(handle) = self { + if let InnerCompioHandle::Running(handle) = self { let Some(handle) = handle.take() else { return; }; @@ -152,22 +202,29 @@ impl Future for InnerCompioHandle { let this = &mut *self; match this { - InnerCompioHandle::CompioHandle(handle) => { + InnerCompioHandle::Running(handle) => { let Some(mut handle) = handle.take() else { return Poll::Ready(Ok(None)); }; match Pin::new(&mut handle).poll(cx) { Poll::Ready(result) => Poll::Ready(result.map(Some).map_err(JoinError::from)), Poll::Pending => { - *this = InnerCompioHandle::CompioHandle(Some(handle)); + *this = InnerCompioHandle::Running(Some(handle)); Poll::Pending } } } - InnerCompioHandle::CompioFuture(future) => match Pin::new(future).poll(cx) { - Poll::Ready(result) => Poll::Ready(Ok(result)), + InnerCompioHandle::Cancelling(future) => match Pin::new(future).poll(cx) { + Poll::Ready(result) => { + *this = InnerCompioHandle::Ready(None); + Poll::Ready(Ok(result)) + } Poll::Pending => Poll::Pending, }, + InnerCompioHandle::Ready(result) => match result.take() { + Some(result) => Poll::Ready(Ok(result)), + None => Poll::Ready(Err(JoinError::Empty)), + }, } } } @@ -179,7 +236,7 @@ fn start_compio_cancel( ) { let mut state = state.lock(); - let InnerCompioHandle::CompioHandle(slot) = &mut *state else { + let InnerCompioHandle::Running(slot) = &mut *state else { return; }; let Some(handle) = slot.take() else { @@ -188,18 +245,29 @@ fn start_compio_cancel( requested.store(true, Ordering::Release); - let future: CancelFuture = Box::pin(handle.cancel()); - *state = InnerCompioHandle::CompioFuture(Optional::new(future)); + let future: BoxCancelFuture = Box::pin(handle.cancel()); + let mut future = CompioCancel::new(future); + *state = match future.poll_inner() { + Poll::Ready(result) => InnerCompioHandle::Ready(Some(result)), + Poll::Pending => InnerCompioHandle::Cancelling(future), + }; } #[cfg(all(feature = "compio", not(target_arch = "wasm32")))] impl InnerCompioHandle { - fn is_finished(&self) -> bool { + fn is_finished(&mut self) -> bool { match self { - InnerCompioHandle::CompioHandle(handle) => { + InnerCompioHandle::Running(handle) => { handle.as_ref().map(|h| h.is_finished()).unwrap_or(true) } - InnerCompioHandle::CompioFuture(_) => true, + InnerCompioHandle::Cancelling(future) => match future.poll_inner() { + Poll::Ready(result) => { + *self = InnerCompioHandle::Ready(Some(result)); + true + } + Poll::Pending => false, + }, + InnerCompioHandle::Ready(_) => true, } } } @@ -219,7 +287,7 @@ impl InnerJoinHandle { T: Send + 'static, { Self::CompioHandle { - handle: Mutex::new(InnerCompioHandle::CompioHandle(Some(handle))), + handle: Mutex::new(InnerCompioHandle::Running(Some(handle))), abort_requested: AtomicBool::new(false), start_cancel: start_compio_cancel::, } @@ -404,6 +472,7 @@ impl AbortableJoinHandle { /// See [`JoinHandle::abort`] pub fn abort(&self) { self.handle.inner.lock().abort(); + self.handle.wake(); } /// See [`JoinHandle::is_finished`] diff --git a/src/rt/compio.rs b/src/rt/compio.rs index 70a807c..70439ba 100644 --- a/src/rt/compio.rs +++ b/src/rt/compio.rs @@ -103,6 +103,8 @@ mod tests { use crate::{Executor, ExecutorBlocking, ExecutorTimeout, TimeoutError}; use futures::channel::mpsc::{Receiver, UnboundedReceiver}; use futures_timer::Delay; + use std::sync::Arc; + use std::sync::atomic::{AtomicBool, Ordering}; #[compio::test] async fn explicit_abort_is_reported_as_aborted() { @@ -110,6 +112,36 @@ mod tests { handle.abort(); + assert!(handle.is_finished()); + assert!(matches!(handle.await, Err(JoinError::Aborted))); + } + + #[compio::test] + async fn abort_is_requested_before_handle_is_awaited() { + struct DropGuard(Arc); + + impl Drop for DropGuard { + fn drop(&mut self) { + self.0.store(true, Ordering::Release); + } + } + + let dropped = Arc::new(AtomicBool::new(false)); + let task_dropped = dropped.clone(); + let (started_tx, started_rx) = futures::channel::oneshot::channel(); + let handle = CompioExecutor.spawn(async move { + let _guard = DropGuard(task_dropped); + let _ = started_tx.send(()); + futures::future::pending::<()>().await; + }); + + started_rx.await.unwrap(); + handle.abort(); + + crate::task::yield_now().await; + + assert!(dropped.load(Ordering::Acquire)); + assert!(handle.is_finished()); assert!(matches!(handle.await, Err(JoinError::Aborted))); } From a8b8f29b0ce28bcfcdcd8a7bc8dcf3db72af5f15 Mon Sep 17 00:00:00 2001 From: Darius Clark Date: Tue, 25 Aug 2026 15:13:26 -0400 Subject: [PATCH 08/11] chore: add test, document limitations, etc --- README.md | 2 +- src/rt/compio.rs | 23 ++++++++++++++++++++++- 2 files changed, 23 insertions(+), 2 deletions(-) diff --git a/README.md b/README.md index 146ac5e..b8ab852 100644 --- a/README.md +++ b/README.md @@ -4,7 +4,7 @@ async-rt is a small library designed to utilize async executors (i.e Tokio) through a common API while extending functionality for abortable task, tasks that receive messages and being able to switch between two async runtimes at compile time if specific -conditions are met (i.e tokio if the feature is enabled and its a non-wasm32 arch, while wasm32-bindgen-futures is used if the arch is wasm32). +conditions are met (i.e tokio or compio if the feature is enabled and its a non-wasm32 arch, while wasm32-bindgen-futures is used if the arch is wasm32). ## MSRV diff --git a/src/rt/compio.rs b/src/rt/compio.rs index 70439ba..35d9d40 100644 --- a/src/rt/compio.rs +++ b/src/rt/compio.rs @@ -32,6 +32,12 @@ impl ExecutorBlocking for CompioExecutor { impl ExecutorTimeout for CompioExecutor {} /// Compio executor with an [`Runtime`] +/// +/// # Note +/// +/// Creating or supplying this runtime by itself will not drive task to completion when spawned. +/// This will be useful if you are already calling [`CompioRuntimeExecutor::from_current_runtime`] or +/// [`CompioRuntimeExecutor::with_runtime`] with an existing runtime. #[derive(Clone, Debug)] pub struct CompioRuntimeExecutor { _runtime: Option, @@ -39,7 +45,9 @@ pub struct CompioRuntimeExecutor { impl CompioRuntimeExecutor { /// Creates a compio runtime. - pub fn new() -> std::io::Result { + // TODO: make public when ExecutorBlockOn is impl + #[allow(dead_code)] + pub(crate) fn new() -> std::io::Result { let runtime = Runtime::builder().build()?; Ok(Self::with_runtime(runtime)) } @@ -105,6 +113,7 @@ mod tests { use futures_timer::Delay; use std::sync::Arc; use std::sync::atomic::{AtomicBool, Ordering}; + use futures::{SinkExt, StreamExt}; #[compio::test] async fn explicit_abort_is_reported_as_aborted() { @@ -116,6 +125,18 @@ mod tests { assert!(matches!(handle.await, Err(JoinError::Aborted))); } + #[compio::test] + async fn joinhandle_doesnt_abort_on_drop() { + let (mut tx, mut rx) = futures::channel::mpsc::channel(1); + let handle = CompioExecutor.spawn(async move { + let _ = tx.send(()).await; + }); + drop(handle); + + let val = rx.next().await; + assert!(val.is_some()); + } + #[compio::test] async fn abort_is_requested_before_handle_is_awaited() { struct DropGuard(Arc); From d80887e70241a356b84ff320b9276c95b4c91645 Mon Sep 17 00:00:00 2001 From: Darius Clark Date: Tue, 25 Aug 2026 16:20:59 -0400 Subject: [PATCH 09/11] chore: fmt --- src/rt/compio.rs | 20 +++++--------------- 1 file changed, 5 insertions(+), 15 deletions(-) diff --git a/src/rt/compio.rs b/src/rt/compio.rs index 35d9d40..edd367e 100644 --- a/src/rt/compio.rs +++ b/src/rt/compio.rs @@ -40,7 +40,7 @@ impl ExecutorTimeout for CompioExecutor {} /// [`CompioRuntimeExecutor::with_runtime`] with an existing runtime. #[derive(Clone, Debug)] pub struct CompioRuntimeExecutor { - _runtime: Option, + _runtime: Runtime, } impl CompioRuntimeExecutor { @@ -54,9 +54,7 @@ impl CompioRuntimeExecutor { /// Create an executor with the supplied [`Runtime`]. pub fn with_runtime(runtime: Runtime) -> Self { - Self { - _runtime: Some(runtime), - } + Self { _runtime: runtime } } /// Create an executor from the existing Runtime @@ -76,11 +74,7 @@ impl Executor for CompioRuntimeExecutor { F: Future + Send + 'static, F::Output: Send + 'static, { - let handle = self - ._runtime - .as_ref() - .expect("Compio runtime is unavailable") - .spawn(future); + let handle = self._runtime.spawn(future); let inner = InnerJoinHandle::compio(handle); JoinHandle { inner } } @@ -92,11 +86,7 @@ impl ExecutorBlocking for CompioRuntimeExecutor { F: FnOnce() -> R + Send + 'static, R: Send + 'static, { - let handle = self - ._runtime - .as_ref() - .expect("Compio runtime is unavailable") - .spawn_blocking(f); + let handle = self._runtime.spawn_blocking(f); let inner = InnerJoinHandle::compio(handle); JoinHandle { inner } } @@ -110,10 +100,10 @@ mod tests { use crate::error::JoinError; use crate::{Executor, ExecutorBlocking, ExecutorTimeout, TimeoutError}; use futures::channel::mpsc::{Receiver, UnboundedReceiver}; + use futures::{SinkExt, StreamExt}; use futures_timer::Delay; use std::sync::Arc; use std::sync::atomic::{AtomicBool, Ordering}; - use futures::{SinkExt, StreamExt}; #[compio::test] async fn explicit_abort_is_reported_as_aborted() { From 8ad6f4c4b5e94377af612313df893d8af8b2468f Mon Sep 17 00:00:00 2001 From: Darius Clark Date: Tue, 25 Aug 2026 16:36:37 -0400 Subject: [PATCH 10/11] chore: small note --- README.md | 2 +- src/global.rs | 2 +- src/rt/compio.rs | 3 +++ 3 files changed, 5 insertions(+), 2 deletions(-) diff --git a/README.md b/README.md index b8ab852..16ccd15 100644 --- a/README.md +++ b/README.md @@ -3,7 +3,7 @@ ## Overview async-rt is a small library designed to utilize async executors (i.e Tokio) through a common API while extending functionality -for abortable task, tasks that receive messages and being able to switch between two async runtimes at compile time if specific +for abortable task, tasks that receive messages and being able to switch between multiple async runtimes at compile time if specific conditions are met (i.e tokio or compio if the feature is enabled and its a non-wasm32 arch, while wasm32-bindgen-futures is used if the arch is wasm32). ## MSRV diff --git a/src/global.rs b/src/global.rs index ed5468f..9e608ef 100644 --- a/src/global.rs +++ b/src/global.rs @@ -5,7 +5,7 @@ use std::future::Future; /// /// * On non-Wasm targets with the `tokio` or `compio` feature enabled, it uses /// `TokioExecutor` or `CompioExecutor`. -/// * On non-Wasm targets with the `threadpool` feature enabled and the `tokio` +/// * On non-Wasm targets with the `threadpool` feature enabled and the `tokio` or `compio` /// feature disabled, it uses `ThreadPoolExecutor`. /// * On Wasm targets, it uses `WasmExecutor`, backed by /// `wasm-bindgen-futures`. diff --git a/src/rt/compio.rs b/src/rt/compio.rs index edd367e..1095075 100644 --- a/src/rt/compio.rs +++ b/src/rt/compio.rs @@ -53,6 +53,9 @@ impl CompioRuntimeExecutor { } /// Create an executor with the supplied [`Runtime`]. + /// + /// Note that this executor schedules tasks but does not drive the runtime. + /// The supplied runtime must be driven externally. pub fn with_runtime(runtime: Runtime) -> Self { Self { _runtime: runtime } } From 40f0ae18095651c7e2fa4001436c519649b48c9f Mon Sep 17 00:00:00 2001 From: Darius Clark Date: Tue, 25 Aug 2026 16:56:01 -0400 Subject: [PATCH 11/11] chore: change variable usability --- src/rt/compio.rs | 8 ++++---- 1 file changed, 4 insertions(+), 4 deletions(-) diff --git a/src/rt/compio.rs b/src/rt/compio.rs index 1095075..5cb3820 100644 --- a/src/rt/compio.rs +++ b/src/rt/compio.rs @@ -40,7 +40,7 @@ impl ExecutorTimeout for CompioExecutor {} /// [`CompioRuntimeExecutor::with_runtime`] with an existing runtime. #[derive(Clone, Debug)] pub struct CompioRuntimeExecutor { - _runtime: Runtime, + runtime: Runtime, } impl CompioRuntimeExecutor { @@ -57,7 +57,7 @@ impl CompioRuntimeExecutor { /// Note that this executor schedules tasks but does not drive the runtime. /// The supplied runtime must be driven externally. pub fn with_runtime(runtime: Runtime) -> Self { - Self { _runtime: runtime } + Self { runtime } } /// Create an executor from the existing Runtime @@ -77,7 +77,7 @@ impl Executor for CompioRuntimeExecutor { F: Future + Send + 'static, F::Output: Send + 'static, { - let handle = self._runtime.spawn(future); + let handle = self.runtime.spawn(future); let inner = InnerJoinHandle::compio(handle); JoinHandle { inner } } @@ -89,7 +89,7 @@ impl ExecutorBlocking for CompioRuntimeExecutor { F: FnOnce() -> R + Send + 'static, R: Send + 'static, { - let handle = self._runtime.spawn_blocking(f); + let handle = self.runtime.spawn_blocking(f); let inner = InnerJoinHandle::compio(handle); JoinHandle { inner } }