aria_primitives/
wakeup.rs

1//! A self-pipe: something that has to interrupt a poll loop writes to it, and
2//! the loop sees an ordinary readable file descriptor.
3//!
4//! Both ends are non-blocking, so neither end can stall the thread it runs on.
5//! What the bytes mean is the caller's business.
6
7use std::io::{self, Read, Write};
8use std::os::unix::net::UnixStream;
9
10const DRAIN_BUFFER_BYTES: usize = 64;
11
12pub struct Wakeup {
13    reader: UnixStream,
14    writer: UnixStream,
15}
16
17impl Wakeup {
18    pub fn new() -> Result<Self, WakeupError> {
19        let (reader, writer) = UnixStream::pair().map_err(WakeupError::Create)?;
20        reader
21            .set_nonblocking(true)
22            .map_err(WakeupError::Configure)?;
23        writer
24            .set_nonblocking(true)
25            .map_err(WakeupError::Configure)?;
26        Ok(Self { reader, writer })
27    }
28
29    /// For an async reactor, and anything else that takes an owned reader.
30    pub fn clone_reader(&self) -> Result<UnixStream, WakeupError> {
31        self.reader.try_clone().map_err(WakeupError::Duplicate)
32    }
33
34    /// For a signal handler, and anything else that takes an owned writer.
35    pub fn clone_writer(&self) -> Result<UnixStream, WakeupError> {
36        self.writer.try_clone().map_err(WakeupError::Duplicate)
37    }
38
39    // A full pipe already carries the message, so a blocked write is success.
40    pub fn signal(&self) -> Result<(), WakeupError> {
41        match (&self.writer).write_all(b"X") {
42            Ok(()) => Ok(()),
43            Err(error) if error.kind() == io::ErrorKind::WouldBlock => Ok(()),
44            Err(error) => Err(WakeupError::Signal(error)),
45        }
46    }
47
48    pub fn drain(&self) -> Result<(), WakeupError> {
49        let mut reader = &self.reader;
50        let mut buffer = [0_u8; DRAIN_BUFFER_BYTES];
51        loop {
52            match reader.read(&mut buffer) {
53                Ok(0) => return Ok(()),
54                Ok(_read) => {}
55                Err(error) if error.kind() == io::ErrorKind::Interrupted => {}
56                Err(error) if error.kind() == io::ErrorKind::WouldBlock => return Ok(()),
57                Err(error) => return Err(WakeupError::Drain(error)),
58            }
59        }
60    }
61}
62
63#[derive(Debug, thiserror::Error)]
64pub enum WakeupError {
65    #[error("failed to create the wakeup socket pair")]
66    Create(#[source] io::Error),
67
68    #[error("failed to configure the wakeup socket pair")]
69    Configure(#[source] io::Error),
70
71    #[error("failed to duplicate the wakeup socket")]
72    Duplicate(#[source] io::Error),
73
74    #[error("failed to signal the wakeup socket")]
75    Signal(#[source] io::Error),
76
77    #[error("failed to drain the wakeup socket")]
78    Drain(#[source] io::Error),
79}
aria_primitives/wakeup.rs

aria_primitives/
wakeup.rs
