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/README.md b/README.md
index 146ac5e..16ccd15 100644
--- a/README.md
+++ b/README.md
@@ -3,8 +3,8 @@
 ## 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
-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).
+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/error.rs b/src/error.rs
index 35cdfa3..e599297 100644
--- a/src/error.rs
+++ b/src/error.rs
@@ -37,6 +37,16 @@ impl From<tokio::task::JoinError> for JoinError {
     }
 }
 
+#[cfg(all(feature = "compio", not(target_arch = "wasm32")))]
+impl From<compio::runtime::JoinError> 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/global.rs b/src/global.rs
index 899efe9..9e608ef 100644
--- a/src/global.rs
+++ b/src/global.rs
@@ -3,9 +3,9 @@ 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 `threadpool` feature enabled and the `tokio`
+/// * 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` or `compio`
 ///   feature disabled, it uses `ThreadPoolExecutor`.
 /// * On Wasm targets, it uses `WasmExecutor`, backed by
 ///   `wasm-bindgen-futures`.
@@ -23,10 +23,22 @@ impl Executor for GlobalExecutor {
     }
 
     #[cfg(all(
-        feature = "threadpool",
+        feature = "compio",
         not(any(feature = "tokio", target_arch = "wasm32"))
     ))]
     fn spawn<F>(&self, future: F) -> JoinHandle<F::Output>
+    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<F>(&self, future: F) -> JoinHandle<F::Output>
     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<F>(&self, _: F) -> JoinHandle<F::Output>
@@ -68,10 +81,22 @@ impl ExecutorBlocking for GlobalExecutor {
     }
 
     #[cfg(all(
-        feature = "threadpool",
+        feature = "compio",
         not(any(feature = "tokio", target_arch = "wasm32"))
     ))]
     fn spawn_blocking<F, R>(&self, f: F) -> JoinHandle<R>
+    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<F, R>(&self, f: F) -> JoinHandle<R>
     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<F, R>(&self, _: F) -> JoinHandle<R>
diff --git a/src/lib.rs b/src/lib.rs
index 94e64dd..0a331d7 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 BoxCancelFuture<T> = Pin<Box<dyn Future<Output = Option<T>> + Send + 'static>>;
+
+#[cfg(all(feature = "compio", not(target_arch = "wasm32")))]
+type StartCompioCancel<T> = fn(&Mutex<InnerCompioHandle<T>>, &AtomicBool);
+
 #[cfg_attr(feature = "tokio", allow(dead_code))]
 pub(crate) struct CompletionGuard {
     finished: Arc<AtomicBool>,
@@ -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<T> {
         handle: Optional<::tokio::task::JoinHandle<T>>,
         abort_requested: AtomicBool,
     },
+    #[cfg(all(feature = "compio", not(target_arch = "wasm32")))]
+    CompioHandle {
+        handle: Mutex<InnerCompioHandle<T>>,
+        abort_requested: AtomicBool,
+        start_cancel: StartCompioCancel<T>,
+    },
     #[allow(dead_code)]
     CustomHandle {
         inner: Optional<futures::channel::oneshot::Receiver<Result<T, JoinError>>>,
@@ -109,6 +124,154 @@ enum InnerJoinHandle<T> {
     Empty,
 }
 
+#[cfg(all(feature = "compio", not(target_arch = "wasm32")))]
+enum InnerCompioHandle<T> {
+    Running(Option<::compio::runtime::JoinHandle<T>>),
+    Cancelling(CompioCancel<T>),
+    Ready(Option<Option<T>>),
+}
+
+#[cfg(all(feature = "compio", not(target_arch = "wasm32")))]
+impl<T> Unpin for InnerCompioHandle<T> {}
+
+#[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<Self>) {
+        arc_self.outer.wake();
+    }
+}
+
+#[cfg(all(feature = "compio", not(target_arch = "wasm32")))]
+struct CompioCancel<T> {
+    future: BoxCancelFuture<T>,
+    wake: Arc<CompioCancelWake>,
+}
+
+#[cfg(all(feature = "compio", not(target_arch = "wasm32")))]
+impl<T> CompioCancel<T> {
+    fn new(future: BoxCancelFuture<T>) -> Self {
+        Self {
+            future,
+            wake: Arc::new(CompioCancelWake {
+                outer: AtomicWaker::new(),
+            }),
+        }
+    }
+
+    fn poll_inner(&mut self) -> Poll<Option<T>> {
+        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<T> Future for CompioCancel<T> {
+    type Output = Option<T>;
+
+    fn poll(mut self: Pin<&mut Self>, cx: &mut Context<'_>) -> Poll<Self::Output> {
+        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<T> Drop for InnerCompioHandle<T> {
+    fn drop(&mut self) {
+        if let InnerCompioHandle::Running(handle) = self {
+            let Some(handle) = handle.take() else {
+                return;
+            };
+            handle.detach();
+        }
+    }
+}
+
+#[cfg(all(feature = "compio", not(target_arch = "wasm32")))]
+impl<T> Future for InnerCompioHandle<T> {
+    type Output = Result<Option<T>, JoinError>;
+    fn poll(mut self: Pin<&mut Self>, cx: &mut Context<'_>) -> Poll<Self::Output> {
+        let this = &mut *self;
+
+        match this {
+            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::Running(Some(handle));
+                        Poll::Pending
+                    }
+                }
+            }
+            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)),
+            },
+        }
+    }
+}
+
+#[cfg(all(feature = "compio", not(target_arch = "wasm32")))]
+fn start_compio_cancel<T: Send + 'static>(
+    state: &Mutex<InnerCompioHandle<T>>,
+    requested: &AtomicBool,
+) {
+    let mut state = state.lock();
+
+    let InnerCompioHandle::Running(slot) = &mut *state else {
+        return;
+    };
+    let Some(handle) = slot.take() else {
+        return;
+    };
+
+    requested.store(true, Ordering::Release);
+
+    let future: BoxCancelFuture<T> = 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<T> InnerCompioHandle<T> {
+    fn is_finished(&mut self) -> bool {
+        match self {
+            InnerCompioHandle::Running(handle) => {
+                handle.as_ref().map(|h| h.is_finished()).unwrap_or(true)
+            }
+            InnerCompioHandle::Cancelling(future) => match future.poll_inner() {
+                Poll::Ready(result) => {
+                    *self = InnerCompioHandle::Ready(Some(result));
+                    true
+                }
+                Poll::Pending => false,
+            },
+            InnerCompioHandle::Ready(_) => true,
+        }
+    }
+}
+
 impl<T> InnerJoinHandle<T> {
     #[cfg(all(feature = "tokio", not(target_arch = "wasm32")))]
     fn tokio(handle: ::tokio::task::JoinHandle<T>) -> Self {
@@ -117,6 +280,18 @@ impl<T> InnerJoinHandle<T> {
             abort_requested: AtomicBool::new(false),
         }
     }
+
+    #[cfg(all(feature = "compio", not(target_arch = "wasm32")))]
+    fn compio(handle: ::compio::runtime::JoinHandle<T>) -> Self
+    where
+        T: Send + 'static,
+    {
+        Self::CompioHandle {
+            handle: Mutex::new(InnerCompioHandle::Running(Some(handle))),
+            abort_requested: AtomicBool::new(false),
+            start_cancel: start_compio_cancel::<T>,
+        }
+    }
 }
 
 impl<T> JoinHandle<T> {
@@ -146,6 +321,14 @@ impl<T> JoinHandle<T> {
                     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 +344,8 @@ impl<T> JoinHandle<T> {
             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 +399,27 @@ impl<T> Future for JoinHandle<T> {
                     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)),
+                }
+            }
             InnerJoinHandle::CustomHandle { inner, .. } => {
                 let fut = futures::ready!(Pin::new(inner).poll(cx));
                 match fut {
@@ -266,6 +472,7 @@ impl<T> AbortableJoinHandle<T> {
     /// 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
new file mode 100644
index 0000000..5cb3820
--- /dev/null
+++ b/src/rt/compio.rs
@@ -0,0 +1,504 @@
+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<F>(&self, future: F) -> JoinHandle<F::Output>
+    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<F, R>(&self, f: F) -> JoinHandle<R>
+    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`]
+///
+/// # 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: Runtime,
+}
+
+impl CompioRuntimeExecutor {
+    /// Creates a compio runtime.
+    // TODO: make public when ExecutorBlockOn is impl
+    #[allow(dead_code)]
+    pub(crate) fn new() -> std::io::Result<Self> {
+        let runtime = Runtime::builder().build()?;
+        Ok(Self::with_runtime(runtime))
+    }
+
+    /// 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 }
+    }
+
+    /// 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<Self> {
+        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<F>(&self, future: F) -> JoinHandle<F::Output>
+    where
+        F: Future + Send + 'static,
+        F::Output: Send + 'static,
+    {
+        let handle = self.runtime.spawn(future);
+        let inner = InnerJoinHandle::compio(handle);
+        JoinHandle { inner }
+    }
+}
+
+impl ExecutorBlocking for CompioRuntimeExecutor {
+    fn spawn_blocking<F, R>(&self, f: F) -> JoinHandle<R>
+    where
+        F: FnOnce() -> R + Send + 'static,
+        R: Send + 'static,
+    {
+        let handle = self.runtime.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::{SinkExt, StreamExt};
+    use futures_timer::Delay;
+    use std::sync::Arc;
+    use std::sync::atomic::{AtomicBool, Ordering};
+
+    #[compio::test]
+    async fn explicit_abort_is_reported_as_aborted() {
+        let handle = CompioExecutor.spawn(futures::future::pending::<()>());
+
+        handle.abort();
+
+        assert!(handle.is_finished());
+        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<AtomicBool>);
+
+        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)));
+    }
+
+    #[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<String>),
+        }
+
+        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::<String>();
+        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<usize>;
+
+        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::<usize>();
+        let (tx2, rx2) = futures::channel::oneshot::channel::<usize>();
+        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<String>),
+        }
+
+        let mut task =
+            executor.spawn_coroutine_with_receiver(|mut rx: Receiver<Message>| 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::<String>();
+        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<String>),
+        }
+
+        let mut task = executor.spawn_coroutine_with_receiver_and_context(
+            State::default(),
+            |mut state, mut rx: Receiver<Message>| 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::<String>();
+        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<String>),
+        }
+
+        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::<String>();
+        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<usize>;
+
+        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::<usize>();
+        let (tx2, rx2) = futures::channel::oneshot::channel::<usize>();
+        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<String>),
+        }
+
+        let mut task = executor.spawn_unbounded_coroutine_with_receiver(
+            |mut rx: UnboundedReceiver<Message>| 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::<String>();
+        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<String>),
+        }
+
+        let mut task = executor.spawn_unbounded_coroutine_with_receiver_and_context(
+            State::default(),
+            |mut state, mut rx: UnboundedReceiver<Message>| 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::<String>();
+        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;