subetha_cxc/
async_ring.rs

1//! `AsyncSpscRing`: `Future`-shaped async adapter on top of
2//! [`crate::blocking_spsc_ring::BlockingSpscRing`].
3//!
4//! Turns the synchronous `send_blocking` / `recv_blocking` API into
5//! `send(...).await` / `recv(...).await` so SubEtha rings compose
6//! with any async executor (tokio, smol, async-std, custom). The
7//! adapter is executor-agnostic: it uses only `std::future::Future`
8//! plus `std::thread` to bridge the cross-process kernel-park to
9//! the Rust `Waker` ecosystem.
10//!
11//! # How the bridge works
12//!
13//! Rust's async model is "Future returns `Pending` and registers a
14//! Waker; something fires the Waker; executor re-polls." The
15//! underlying `CrossProcessWaker` is a kernel-park primitive that
16//! does the wait OFF the async runtime's thread. To bridge:
17//!
18//! 1. First poll calls `try_*` on the inner ring. If immediately
19//!    ready, return `Poll::Ready`.
20//! 2. Otherwise, spawn a std::thread that calls the blocking
21//!    counterpart (`recv_blocking` / `send_blocking`) with the
22//!    caller-supplied timeout. Park the rust Waker.
23//! 3. When the blocking call returns, store the result + fire the
24//!    Waker.
25//! 4. Next poll observes the stored result and returns `Poll::Ready`.
26//!
27//! # Why a bounded timeout is required
28//!
29//! Dropping a pending `AsyncRecv` / `AsyncSend` future does NOT
30//! cancel the spawned worker thread (`std::thread` lacks safe
31//! cancellation). The thread's worst-case lifetime equals the
32//! caller-supplied timeout. Unbounded waits are rejected at the
33//! type level by requiring `Duration` (not `Option<Duration>`).
34//!
35//! # Worker-thread cost model
36//!
37//! Each in-flight future spawns one OS thread. This pattern fits
38//! the substrate's intended use of async (a small number of
39//! long-running consumer tasks per process, not thousands of
40//! short-lived futures). Callers driving high concurrency batch
41//! through one `BlockingSpscRing` per consumer task and call
42//! `recv_blocking` directly inside `tokio::task::spawn_blocking`
43//! (or the executor's equivalent).
44
45use std::future::Future;
46use std::pin::Pin;
47use std::sync::Arc;
48use std::sync::Mutex;
49use std::task::{Context, Poll, Waker};
50use std::time::Duration;
51
52use crate::blocking_spsc_ring::{BlockingError, BlockingSpscRing};
53use crate::shared_ring::RingError;
54
55/// Wrapper providing `.recv(timeout).await` and `.send(timeout).await`.
56pub struct AsyncSpscRing {
57    inner: Arc<BlockingSpscRing>,
58}
59
60impl AsyncSpscRing {
61    /// Wrap an existing `BlockingSpscRing`. Both halves share the
62    /// same underlying ring + wakers; this is just an async-shaped
63    /// view of the same primitive.
64    pub fn new(inner: Arc<BlockingSpscRing>) -> Self {
65        Self { inner }
66    }
67
68    /// Async pop with bounded wait. Returns a future that resolves
69    /// when an item arrives, or `BlockingError::Timeout` after the
70    /// caller-supplied duration.
71    pub fn recv(&self, timeout: Duration) -> AsyncRecv {
72        AsyncRecv {
73            ring: Arc::clone(&self.inner),
74            state: Arc::new(Mutex::new(SlotState::Pending)),
75            timeout,
76            spawned: false,
77        }
78    }
79
80    /// Async push with bounded wait.
81    pub fn send(&self, payload: Vec<u8>, timeout: Duration) -> AsyncSend {
82        AsyncSend {
83            ring: Arc::clone(&self.inner),
84            state: Arc::new(Mutex::new(SlotState::Pending)),
85            timeout,
86            payload: Some(payload),
87            spawned: false,
88        }
89    }
90
91    /// Direct access to the underlying blocking ring.
92    pub fn inner(&self) -> &Arc<BlockingSpscRing> { &self.inner }
93}
94
95/// Shared state between the async future and its worker thread.
96enum SlotState<T> {
97    Pending,
98    Ready(T),
99    Parked(Waker),
100}
101
102type RecvSlot = Arc<Mutex<SlotState<Result<Vec<u8>, BlockingError>>>>;
103type SendSlot = Arc<Mutex<SlotState<Result<(), BlockingError>>>>;
104
105/// Future returned by [`AsyncSpscRing::recv`].
106pub struct AsyncRecv {
107    ring: Arc<BlockingSpscRing>,
108    state: RecvSlot,
109    timeout: Duration,
110    spawned: bool,
111}
112
113impl Future for AsyncRecv {
114    type Output = Result<Vec<u8>, BlockingError>;
115
116    fn poll(self: Pin<&mut Self>, cx: &mut Context<'_>) -> Poll<Self::Output> {
117        let this = self.get_mut();
118
119        // Fast path: data already present, skip the worker thread.
120        if !this.spawned {
121            let mut buf = vec![0u8; 64];
122            match this.ring.try_pop(&mut buf) {
123                Ok(n) => {
124                    buf.truncate(n);
125                    return Poll::Ready(Ok(buf));
126                }
127                Err(RingError::Empty) => {}
128                Err(e) => return Poll::Ready(Err(BlockingError::Ring(e))),
129            }
130        }
131
132        let mut guard = this.state.lock().unwrap();
133        match &mut *guard {
134            SlotState::Ready(_) => {
135                let taken = std::mem::replace(&mut *guard, SlotState::Pending);
136                match taken {
137                    SlotState::Ready(r) => return Poll::Ready(r),
138                    _ => unreachable!(),
139                }
140            }
141            SlotState::Pending | SlotState::Parked(_) => {
142                *guard = SlotState::Parked(cx.waker().clone());
143            }
144        }
145        drop(guard);
146
147        if !this.spawned {
148            this.spawned = true;
149            let ring = Arc::clone(&this.ring);
150            let state = Arc::clone(&this.state);
151            let timeout = this.timeout;
152            std::thread::spawn(move || {
153                let mut buf = vec![0u8; 64];
154                let r = ring.recv_blocking(&mut buf, Some(timeout)).map(|n| {
155                    buf.truncate(n);
156                    buf
157                });
158                finish_slot(&state, r);
159            });
160        }
161
162        Poll::Pending
163    }
164}
165
166/// Future returned by [`AsyncSpscRing::send`].
167pub struct AsyncSend {
168    ring: Arc<BlockingSpscRing>,
169    state: SendSlot,
170    timeout: Duration,
171    payload: Option<Vec<u8>>,
172    spawned: bool,
173}
174
175impl Future for AsyncSend {
176    type Output = Result<(), BlockingError>;
177
178    fn poll(self: Pin<&mut Self>, cx: &mut Context<'_>) -> Poll<Self::Output> {
179        let this = self.get_mut();
180
181        if !this.spawned {
182            let payload_ref = this.payload.as_ref().expect("payload taken twice");
183            match this.ring.try_push(payload_ref) {
184                Ok(()) => return Poll::Ready(Ok(())),
185                Err(RingError::Full) => {}
186                Err(e) => return Poll::Ready(Err(BlockingError::Ring(e))),
187            }
188        }
189
190        let mut guard = this.state.lock().unwrap();
191        match &mut *guard {
192            SlotState::Ready(_) => {
193                let taken = std::mem::replace(&mut *guard, SlotState::Pending);
194                match taken {
195                    SlotState::Ready(r) => return Poll::Ready(r),
196                    _ => unreachable!(),
197                }
198            }
199            SlotState::Pending | SlotState::Parked(_) => {
200                *guard = SlotState::Parked(cx.waker().clone());
201            }
202        }
203        drop(guard);
204
205        if !this.spawned {
206            this.spawned = true;
207            let ring = Arc::clone(&this.ring);
208            let state = Arc::clone(&this.state);
209            let timeout = this.timeout;
210            let payload = this.payload.take().expect("payload taken twice");
211            std::thread::spawn(move || {
212                let r = ring.send_blocking(&payload, Some(timeout));
213                finish_slot(&state, r);
214            });
215        }
216
217        Poll::Pending
218    }
219}
220
221fn finish_slot<T>(state: &Arc<Mutex<SlotState<T>>>, value: T) {
222    let waker_to_fire = {
223        let mut guard = state.lock().unwrap();
224        let prev = std::mem::replace(&mut *guard, SlotState::Ready(value));
225        match prev {
226            SlotState::Parked(w) => Some(w),
227            _ => None,
228        }
229    };
230    if let Some(w) = waker_to_fire {
231        w.wake();
232    }
233}
234
235#[cfg(test)]
236mod tests {
237    use super::*;
238    use std::sync::atomic::{AtomicBool, Ordering};
239    use std::task::Wake;
240
241    /// Minimal hand-rolled block-on for tests: poll the future,
242    /// park on a condvar when it returns Pending, re-poll when
243    /// the waker fires.
244    struct TestWaker {
245        woken: std::sync::Mutex<bool>,
246        cv: std::sync::Condvar,
247    }
248    impl Wake for TestWaker {
249        fn wake(self: Arc<Self>) {
250            let mut g = self.woken.lock().unwrap();
251            *g = true;
252            self.cv.notify_one();
253        }
254    }
255
256    fn block_on<F: Future>(mut fut: F) -> F::Output {
257        let waker_inner = Arc::new(TestWaker {
258            woken: std::sync::Mutex::new(true),
259            cv: std::sync::Condvar::new(),
260        });
261        let waker: Waker = Arc::clone(&waker_inner).into();
262        let mut cx = Context::from_waker(&waker);
263        // SAFETY: future stays on the stack; we never move it again.
264        let mut fut = unsafe { Pin::new_unchecked(&mut fut) };
265        loop {
266            {
267                let mut g = waker_inner.woken.lock().unwrap();
268                while !*g {
269                    g = waker_inner.cv.wait(g).unwrap();
270                }
271                *g = false;
272            }
273            match fut.as_mut().poll(&mut cx) {
274                Poll::Ready(v) => return v,
275                Poll::Pending => continue,
276            }
277        }
278    }
279
280    #[test]
281    fn recv_returns_immediately_when_ring_has_item() {
282        let ring = Arc::new(BlockingSpscRing::create_anon(4).expect("ring"));
283        let mut payload = [0u8; 56];
284        payload[..8].copy_from_slice(&42u64.to_le_bytes());
285        ring.try_push(&payload).expect("push");
286
287        let adapter = AsyncSpscRing::new(Arc::clone(&ring));
288        let got = block_on(adapter.recv(Duration::from_secs(1))).unwrap();
289        let val = u64::from_le_bytes(got[..8].try_into().unwrap());
290        assert_eq!(val, 42);
291    }
292
293    #[test]
294    fn recv_parks_then_completes_when_producer_pushes() {
295        let ring = Arc::new(BlockingSpscRing::create_anon(4).expect("ring"));
296        let r2 = Arc::clone(&ring);
297        let pushed = Arc::new(AtomicBool::new(false));
298        let pushed2 = Arc::clone(&pushed);
299        std::thread::spawn(move || {
300            std::thread::sleep(Duration::from_millis(40));
301            let mut payload = [0u8; 56];
302            payload[..8].copy_from_slice(&7u64.to_le_bytes());
303            // Publish the flag BEFORE the push. `recv` returns the
304            // instant the item is visible, and the ring's Release on
305            // `head` (paired with the consumer's Acquire) carries this
306            // store with it - so the assert below never races a flag
307            // that was published after the data.
308            pushed2.store(true, Ordering::Release);
309            r2.try_push(&payload).expect("push");
310        });
311        let adapter = AsyncSpscRing::new(Arc::clone(&ring));
312        let got = block_on(adapter.recv(Duration::from_secs(2))).unwrap();
313        assert!(pushed.load(Ordering::Acquire), "producer ran");
314        let val = u64::from_le_bytes(got[..8].try_into().unwrap());
315        assert_eq!(val, 7);
316    }
317
318    #[test]
319    fn recv_times_out_when_no_producer() {
320        let ring = Arc::new(BlockingSpscRing::create_anon(4).expect("ring"));
321        let adapter = AsyncSpscRing::new(Arc::clone(&ring));
322        let t0 = std::time::Instant::now();
323        let r = block_on(adapter.recv(Duration::from_millis(80)));
324        assert!(matches!(r, Err(BlockingError::Timeout)));
325        assert!(t0.elapsed() >= Duration::from_millis(60));
326    }
327
328    #[test]
329    fn send_completes_immediately_when_ring_not_full() {
330        let ring = Arc::new(BlockingSpscRing::create_anon(4).expect("ring"));
331        let adapter = AsyncSpscRing::new(Arc::clone(&ring));
332        let payload = (12345u64).to_le_bytes().to_vec();
333        block_on(adapter.send(payload, Duration::from_secs(1))).unwrap();
334        let mut buf = [0u8; 64];
335        let n = ring.try_pop(&mut buf).expect("pop");
336        let v = u64::from_le_bytes(buf[..8].try_into().unwrap());
337        assert_eq!(v, 12345);
338        assert!(n >= 8);
339    }
340}
subetha_cxc/async_ring.rs

subetha_cxc/
async_ring.rs
