orbit_core/ring/
readiness.rs

1//! Native readiness bridge for an SHM ring.
2//!
3//! The shared signal is a generation in the ring header, waited through Linux
4//! futex, FreeBSD umtx, or macOS shared address waits. Each process owns a
5//! private readiness fd (`eventfd`, or a pipe on macOS) and a small
6//! blocking driver thread that converts generation changes into fd readiness.
7//! Async runtimes can therefore wait on their normal reactor without sharing
8//! one drainable eventfd across readers.
9
10#![cfg(any(target_os = "linux", target_os = "freebsd", target_os = "macos"))]
11
12use std::os::fd::{AsFd, AsRawFd, BorrowedFd, RawFd};
13use std::sync::atomic::{AtomicBool, Ordering};
14use std::sync::{Arc, Condvar, Mutex};
15use std::thread::JoinHandle;
16use std::{fmt, io};
17
18use super::shm::ShmRing;
19use crate::readiness::{Readiness, Signal};
20
21/// Process-local fd readiness for one shared Orbit ring.
22///
23/// Every subscribing process creates its own instance. A small driver thread
24/// parks on the ring's generation, counting itself in the ring header while
25/// it is parked, and marks this fd readable when the generation moves. It
26/// then waits for its consumer to [`drain`](Self::drain) before it parks
27/// again. A publisher using [`Fleet::publish_notified`](crate::Fleet) bumps
28/// the generation and wakes only when a driver is parked: while every reader
29/// is busy draining, a publish is a write and an atomic, with no syscall.
30/// Multiple publishes coalesce into one wake.
31///
32/// The fd is backed by an `eventfd` on Linux and FreeBSD and by a nonblocking
33/// pipe on macOS, where native readiness needs 14.4 or later; older systems
34/// return `ErrorKind::Unsupported` from construction so callers can poll.
35///
36/// The contract: drain the fd, then poll the ring through your own cursor.
37/// Draining is what lets the driver park.
38///
39/// **Clear the reactor's readiness before draining, not after.** The driver
40/// signals once per drain: a signal that arrives just after [`drain`](Self::drain)
41/// is the only one until the next drain, and a reactor that clears readiness
42/// after draining (tokio's `clear_ready`) erases it — the consumer then waits
43/// for an edge that never comes, and the driver for a drain that never
44/// happens.
45pub struct RingEventFd {
46    fd: Readiness,
47    ring: Arc<ShmRing>,
48    stop: Arc<AtomicBool>,
49    drained: Arc<Drained>,
50    driver: Option<JoinHandle<()>>,
51}
52
53/// Whether the consumer has taken the last signal. The driver waits here,
54/// in this process, until it has — not on the shared word.
55#[derive(Default)]
56struct Drained {
57    pending: Mutex<bool>,
58    taken: Condvar,
59}
60
61impl RingEventFd {
62    pub(crate) fn new(ring: Arc<ShmRing>) -> io::Result<Self> {
63        let (fd, driver_fd) = local_notification_pair()?;
64        let stop = Arc::new(AtomicBool::new(false));
65        let drained = Arc::new(Drained::default());
66        let (driver_stop, driver_drained, driver_ring) =
67            (stop.clone(), drained.clone(), ring.clone());
68        let mut observed =
69            ring.notification_generation().load(Ordering::Acquire);
70
71        let driver = std::thread::Builder::new()
72            .name(format!("orbit-ring-{}-eventfd", ring.kind()))
73            .spawn(move || {
74                let generation = driver_ring.notification_generation();
75                let waiters = driver_ring.notification_waiters();
76                while !driver_stop.load(Ordering::Acquire) {
77                    // Count in, then look once more. A publisher bumps the
78                    // generation before it reads the count; one of the two
79                    // sees the other, so a publish cannot fall between them.
80                    waiters.fetch_add(1, Ordering::SeqCst);
81                    let current = generation.load(Ordering::SeqCst);
82                    if current == observed
83                        && !driver_stop.load(Ordering::Acquire)
84                    {
85                        let parked =
86                            crate::sync::wait_word(generation, observed);
87                        waiters.fetch_sub(1, Ordering::SeqCst);
88                        if parked.is_err() {
89                            break;
90                        }
91                        continue;
92                    }
93                    waiters.fetch_sub(1, Ordering::SeqCst);
94                    if driver_stop.load(Ordering::Acquire) {
95                        break;
96                    }
97                    observed = current;
98                    *driver_drained.pending.lock().unwrap() = true;
99                    if driver_fd.signal().is_err() {
100                        break;
101                    }
102                    let mut pending = driver_drained.pending.lock().unwrap();
103                    while *pending && !driver_stop.load(Ordering::Acquire) {
104                        pending = driver_drained.taken.wait(pending).unwrap();
105                    }
106                }
107            })?;
108
109        Ok(Self { fd, ring, stop, drained, driver: Some(driver) })
110    }
111
112    /// Wake the ring's parked readers, and only them.
113    pub(crate) fn notify(ring: &ShmRing) -> io::Result<()> {
114        ring.notification_generation().fetch_add(1, Ordering::SeqCst);
115        if ring.notification_waiters().load(Ordering::SeqCst) == 0 {
116            return Ok(());
117        }
118        crate::sync::wake_word(ring.notification_generation())
119    }
120
121    /// Drain the fd and let the driver park again. Poll the ring after this.
122    pub fn drain(&self) -> io::Result<u64> {
123        let tokens = self.fd.drain()?;
124        *self.drained.pending.lock().unwrap() = false;
125        self.drained.taken.notify_one();
126        Ok(tokens)
127    }
128}
129
130impl AsRawFd for RingEventFd {
131    fn as_raw_fd(&self) -> RawFd {
132        self.fd.as_raw_fd()
133    }
134}
135
136impl AsFd for RingEventFd {
137    fn as_fd(&self) -> BorrowedFd<'_> {
138        self.fd.as_fd()
139    }
140}
141
142impl fmt::Debug for RingEventFd {
143    fn fmt(
144        &self,
145        f: &mut fmt::Formatter<'_>,
146    ) -> fmt::Result {
147        f.debug_struct("RingEventFd")
148            .field("fd", &self.fd.as_raw_fd())
149            .field("ring_kind", &self.ring.kind())
150            .finish_non_exhaustive()
151    }
152}
153
154impl Drop for RingEventFd {
155    fn drop(&mut self) {
156        self.stop.store(true, Ordering::Release);
157        // Release a driver waiting for a drain, then one parked on the word:
158        // the generation changes first so a driver between its check and its
159        // wait does not park after the wake.
160        {
161            let _pending = self.drained.pending.lock().unwrap();
162            self.drained.taken.notify_one();
163        }
164        self.ring.notification_generation().fetch_add(1, Ordering::SeqCst);
165        let _ = crate::sync::wake_word(self.ring.notification_generation());
166        if let Some(driver) = self.driver.take() {
167            let _ = driver.join();
168        }
169    }
170}
171
172#[cfg(any(target_os = "linux", target_os = "freebsd"))]
173fn local_notification_pair() -> io::Result<(Readiness, Signal)> {
174    crate::readiness::pair()
175}
176
177/// macOS needs 14.4 for the shared address wait the driver parks on, so a
178/// pair is refused there rather than handed out with nothing to feed it.
179#[cfg(target_os = "macos")]
180fn local_notification_pair() -> io::Result<(Readiness, Signal)> {
181    if crate::sync::macos::api().is_none() {
182        return Err(io::Error::new(
183            io::ErrorKind::Unsupported,
184            "Orbit native readiness requires macOS 14.4 or later",
185        ));
186    }
187    crate::readiness::pair()
188}
orbit_core/ring/readiness.rs

orbit_core/ring/
readiness.rs
