monocoque_core/rt/
compio.rs

1//! compio backend: native `io_uring`. Selected by `runtime-compio`.
2//!
3//! See [`super`](crate::rt) for the facade contract this implements.
4
5use std::future::Future;
6
7pub use compio::net::{TcpListener, TcpStream, ToSocketAddrsAsync as ToSocketAddrs};
8pub use compio::time::{sleep, timeout};
9
10/// Owned read half of a split TCP stream.
11///
12/// compio 0.19's `TcpStream::into_split` returns two owned `TcpStream`s that
13/// share the underlying fd, so the read and write halves are the same type.
14pub type OwnedReadHalf = TcpStream;
15/// Owned write half of a split TCP stream. See [`OwnedReadHalf`].
16pub type OwnedWriteHalf = TcpStream;
17
18#[cfg(unix)]
19pub use compio::net::{UnixListener, UnixStream};
20
21/// Bind a TCP listener with `SO_REUSEPORT` so multiple acceptors can share one
22/// port with in-kernel load balancing. See [`crate::tcp::reuseport_listener`].
23///
24/// # Errors
25///
26/// Returns an error if the socket cannot be created/bound or adopted.
27pub fn bind_reuseport(addr: std::net::SocketAddr) -> std::io::Result<TcpListener> {
28    TcpListener::from_std(crate::tcp::reuseport_listener(addr)?)
29}
30
31/// Handle to a spawned task. Kept alive keeps the task running; dropping it
32/// detaches under compio.
33pub type JoinHandle<T> = compio::runtime::JoinHandle<T>;
34
35/// Spawn a task on the current runtime and return its handle.
36#[inline]
37pub fn spawn<F>(fut: F) -> JoinHandle<F::Output>
38where
39    F: Future + 'static,
40{
41    compio::runtime::spawn(fut)
42}
43
44/// Spawn a task and let it run on its own, discarding the handle.
45#[inline]
46pub fn spawn_detached<F>(fut: F)
47where
48    F: Future + 'static,
49    F::Output: 'static,
50{
51    compio::runtime::spawn(fut).detach();
52}
53
54/// Run a blocking closure off the async executor and await its result.
55#[inline]
56pub async fn spawn_blocking<F, T>(f: F) -> T
57where
58    F: FnOnce() -> T + Send + Sync + 'static,
59    T: Send + 'static,
60{
61    compio::runtime::spawn_blocking(f)
62        .await
63        .expect("blocking task panicked")
64}
65
66/// Await a spawned task and return its output.
67///
68/// Normalizes the difference between backends: compio's `JoinHandle` now awaits
69/// to `Result<T, JoinError>` (like tokio), so unwrap the join error here.
70#[inline]
71pub async fn join<T>(handle: JoinHandle<T>) -> T {
72    handle
73        .await
74        .expect("spawned task panicked or was cancelled")
75}
76
77/// A self-contained runtime owned by a single thread.
78///
79/// Used by worker threads that drive their own event loop independently of
80/// the caller's runtime (for example the publisher's fan-out workers).
81pub struct LocalRuntime {
82    inner: compio::runtime::Runtime,
83}
84
85impl LocalRuntime {
86    /// Build a new single-threaded runtime.
87    pub fn new() -> std::io::Result<Self> {
88        Ok(Self {
89            inner: compio::runtime::Runtime::new()?,
90        })
91    }
92
93    /// Run a future to completion on this runtime, blocking the thread.
94    pub fn block_on<F: Future>(&self, fut: F) -> F::Output {
95        self.inner.block_on(fut)
96    }
97}
monocoque_core/rt/compio.rs

monocoque_core/rt/
compio.rs
