orbit_core/
readiness.rs

1//! A descriptor pair for turning a shared-memory change into fd readiness.
2//!
3//! The waiting primitive in Orbit is a word ([`crate::sync`]), which a
4//! runtime with a reactor of its own cannot park on. The bridge is always
5//! the same: something already watching the word — a ring's driver, a
6//! stream table's driver — writes a token, and the consumer's poll set
7//! wakes. This is that pair, and nothing more: no thread, no policy about
8//! who signals or when.
9//!
10//! An `eventfd` where there is one, a pipe where there is not. Both ends
11//! are nonblocking and close-on-exec, so a signal never blocks its writer
12//! and neither end survives an `exec`.
13//!
14//! Readiness is edge-triggered and coalescing by nature: a token says
15//! something may have changed, never what or how much. A consumer drains
16//! and then re-reads the shared state it cares about.
17
18#![cfg(any(target_os = "linux", target_os = "freebsd", target_os = "macos"))]
19
20use std::mem::size_of;
21use std::os::fd::{AsFd, AsRawFd, BorrowedFd, FromRawFd, OwnedFd, RawFd};
22use std::{fmt, io};
23
24/// The consumer's end: put it in a poll set, drain it, re-read your state.
25pub struct Readiness {
26    fd: OwnedFd
27}
28
29/// The signaller's end, held by whoever watches the shared word.
30pub struct Signal {
31    fd: OwnedFd
32}
33
34impl Readiness {
35    /// Take every token the descriptor holds and return how many there
36    /// were. Never blocks; a drained descriptor answers `Ok(0)`.
37    ///
38    /// The count is a local wake count. It says nothing about how many
39    /// things changed in shared memory, which is what the consumer must
40    /// re-read for itself.
41    pub fn drain(&self) -> io::Result<u64> {
42        let mut total = 0_u64;
43        loop {
44            let mut value = 0_u64;
45            // SAFETY: a nonblocking descriptor this type owns, and a u64
46            // of our own to read into.
47            let read = unsafe {
48                libc::read(self.fd.as_raw_fd(), (&mut value as *mut u64).cast(), size_of::<u64>())
49            };
50            if read == size_of::<u64>() as isize {
51                total = total.saturating_add(value);
52                continue;
53            }
54            if read == 0 {
55                return Err(io::Error::new(
56                    io::ErrorKind::UnexpectedEof,
57                    "Orbit readiness closed while draining"
58                ));
59            }
60            if read < 0 {
61                let error = io::Error::last_os_error();
62                return match error.raw_os_error() {
63                    Some(libc::EINTR) => continue,
64                    Some(libc::EAGAIN) => Ok(total),
65                    _ => Err(error)
66                };
67            }
68            return Err(io::Error::new(
69                io::ErrorKind::InvalidData,
70                "Orbit readiness returned a partial counter"
71            ));
72        }
73    }
74}
75
76impl Signal {
77    /// Make the consumer's end readable. A full descriptor is already
78    /// readable, so a token that cannot be added is not a lost signal.
79    pub fn signal(&self) -> io::Result<()> {
80        let value = 1_u64;
81        loop {
82            // SAFETY: a nonblocking descriptor this type owns, and a u64
83            // of our own to write from.
84            let written = unsafe {
85                libc::write(self.fd.as_raw_fd(), (&value as *const u64).cast(), size_of::<u64>())
86            };
87            if written == size_of::<u64>() as isize {
88                return Ok(());
89            }
90            if written < 0 {
91                let error = io::Error::last_os_error();
92                return match error.raw_os_error() {
93                    Some(libc::EINTR) => continue,
94                    Some(libc::EAGAIN) => Ok(()),
95                    _ => Err(error)
96                };
97            }
98            return Err(io::Error::new(
99                io::ErrorKind::WriteZero,
100                "Orbit readiness accepted a partial token"
101            ));
102        }
103    }
104}
105
106impl AsRawFd for Readiness {
107    fn as_raw_fd(&self) -> RawFd {
108        self.fd.as_raw_fd()
109    }
110}
111
112impl AsFd for Readiness {
113    fn as_fd(&self) -> BorrowedFd<'_> {
114        self.fd.as_fd()
115    }
116}
117
118impl AsRawFd for Signal {
119    fn as_raw_fd(&self) -> RawFd {
120        self.fd.as_raw_fd()
121    }
122}
123
124impl fmt::Debug for Readiness {
125    fn fmt(
126        &self,
127        f: &mut fmt::Formatter<'_>
128    ) -> fmt::Result {
129        f.debug_struct("Readiness").field("fd", &self.fd.as_raw_fd()).finish_non_exhaustive()
130    }
131}
132
133impl fmt::Debug for Signal {
134    fn fmt(
135        &self,
136        f: &mut fmt::Formatter<'_>
137    ) -> fmt::Result {
138        f.debug_struct("Signal").field("fd", &self.fd.as_raw_fd()).finish_non_exhaustive()
139    }
140}
141
142/// One `eventfd`, cloned: both ends are the same object, so the order
143/// they are dropped in does not matter.
144#[cfg(any(target_os = "linux", target_os = "freebsd"))]
145pub fn pair() -> io::Result<(Readiness, Signal)> {
146    // SAFETY: a plain syscall with constant flags.
147    let raw = unsafe { libc::eventfd(0, libc::EFD_CLOEXEC | libc::EFD_NONBLOCK) };
148    if raw < 0 {
149        return Err(io::Error::last_os_error());
150    }
151    // SAFETY: a fresh descriptor this call owns.
152    let fd = unsafe { OwnedFd::from_raw_fd(raw) };
153    let signal = fd.try_clone()?;
154    Ok((Readiness { fd }, Signal { fd: signal }))
155}
156
157/// A pipe, whose two ends are distinct: closing the read end first makes
158/// the write end's `signal` fail, which is how a signaller learns its
159/// consumer is gone.
160#[cfg(target_os = "macos")]
161pub fn pair() -> io::Result<(Readiness, Signal)> {
162    let mut raw = [-1; 2];
163    // SAFETY: a plain syscall writing two descriptors into our array.
164    if unsafe { libc::pipe(raw.as_mut_ptr()) } < 0 {
165        return Err(io::Error::last_os_error());
166    }
167    // Own both ends before any fallible setup, so neither leaks.
168    // SAFETY: two fresh descriptors this call owns.
169    let (read, write) = unsafe { (OwnedFd::from_raw_fd(raw[0]), OwnedFd::from_raw_fd(raw[1])) };
170    for fd in [&read, &write] {
171        // SAFETY: descriptors this call owns.
172        let set = unsafe {
173            libc::fcntl(fd.as_raw_fd(), libc::F_SETFD, libc::FD_CLOEXEC) >= 0
174                && libc::fcntl(fd.as_raw_fd(), libc::F_SETFL, libc::O_NONBLOCK) >= 0
175        };
176        if !set {
177            return Err(io::Error::last_os_error());
178        }
179    }
180    Ok((Readiness { fd: read }, Signal { fd: write }))
181}
182
183#[cfg(test)]
184mod tests {
185    use super::*;
186
187    #[test]
188    fn a_token_survives_the_pair_and_coalesces() {
189        let (readiness, signal) = pair().expect("a pair");
190        assert_eq!(readiness.drain().expect("empty"), 0);
191        signal.signal().expect("signal");
192        signal.signal().expect("signal");
193        assert!(readiness.drain().expect("drain") >= 1);
194        assert_eq!(readiness.drain().expect("drained"), 0);
195    }
196}
197
198#[cfg(all(test, target_os = "macos"))]
199mod pipe_tests {
200    use super::*;
201
202    #[test]
203    fn a_full_pipe_coalesces_and_rearms() {
204        let (read, write) = pair().unwrap();
205        for fd in [read.as_raw_fd(), write.as_raw_fd()] {
206            assert_ne!(unsafe { libc::fcntl(fd, libc::F_GETFL) } & libc::O_NONBLOCK, 0);
207            assert_ne!(unsafe { libc::fcntl(fd, libc::F_GETFD) } & libc::FD_CLOEXEC, 0);
208        }
209        // Fill the pipe deliberately, then exercise the bridge's EAGAIN path.
210        let token = 1u64;
211        loop {
212            let n = unsafe { libc::write(write.as_raw_fd(), (&token as *const u64).cast(), 8) };
213            if n < 0 {
214                assert_eq!(io::Error::last_os_error().raw_os_error(), Some(libc::EAGAIN));
215                break;
216            }
217            assert_eq!(n, 8);
218        }
219        write.signal().unwrap();
220        let mut buffer = [0u64; 128];
221        loop {
222            let n = unsafe {
223                libc::read(read.as_raw_fd(), buffer.as_mut_ptr().cast(), size_of_val(&buffer))
224            };
225            if n < 0 {
226                assert_eq!(io::Error::last_os_error().raw_os_error(), Some(libc::EAGAIN));
227                break;
228            }
229            assert!(n > 0);
230        }
231        write.signal().unwrap();
232        let mut value = 0u64;
233        assert_eq!(unsafe { libc::read(read.as_raw_fd(), (&mut value as *mut u64).cast(), 8) }, 8);
234        assert_eq!(value, 1);
235    }
236}
orbit_core/readiness.rs

orbit_core/
readiness.rs
