dope_core/platform/
wake.rs

1use std::{cell, marker, mem, os::fd, time};
2
3use dope_sys::io::{
4    fd::Fd as _,
5    pipe,
6    poll::{self, Poll as _},
7};
8
9use crate::{Backend, Errno, platform::Flags};
10
11#[derive(Clone, Copy, Debug, PartialEq, Eq)]
12pub enum CreateError {
13    Open(Errno),
14    CloseOnExec { end: pipe::End, failure: Flags },
15    Nonblocking { end: pipe::End, failure: Flags },
16    Fork(Errno),
17}
18
19#[derive(Clone, Copy, PartialEq, Eq, Debug)]
20pub enum WaitError {
21    Read(Errno),
22    Record {
23        expected: usize,
24        actual: isize,
25    },
26    Poll(Errno),
27    Timeout {
28        timeout: time::Duration,
29        elapsed: time::Duration,
30    },
31    PollResult {
32        result: i32,
33        revents: i16,
34    },
35}
36
37#[derive(Clone, Copy, PartialEq, Eq, Debug)]
38pub enum NotifyError {
39    Write(Errno),
40    Record { expected: usize, actual: isize },
41}
42
43/// An exclusively used wake reader whose ownership may transfer during startup.
44#[repr(transparent)]
45pub struct Wait(fd::OwnedFd, marker::PhantomData<cell::Cell<()>>);
46
47#[repr(transparent)]
48#[derive(Debug)]
49pub struct Notify(fd::OwnedFd, marker::PhantomData<cell::Cell<()>>);
50
51pub struct Ends<R> {
52    read: R,
53    notify: Notify,
54}
55
56impl Ends<Wait> {
57    pub fn blocking() -> Result<Self, CreateError> {
58        let (read, write) = <Backend as pipe::Pipe>::open().map_err(CreateError::from_native)?;
59        Ok(Self {
60            read: Wait(read, marker::PhantomData),
61            notify: Notify(write, marker::PhantomData),
62        })
63    }
64}
65
66impl Ends<fd::OwnedFd> {
67    pub fn event() -> Result<Self, CreateError> {
68        let (read, write) =
69            <Backend as pipe::Pipe>::nonblocking().map_err(CreateError::from_native)?;
70        Ok(Self {
71            read,
72            notify: Notify(write, marker::PhantomData),
73        })
74    }
75}
76
77impl<R> Ends<R> {
78    pub fn split(self) -> (R, Notify) {
79        (self.read, self.notify)
80    }
81}
82
83impl Wait {
84    pub fn wait(&self) -> Result<(), WaitError> {
85        let mut byte = [0u8];
86        loop {
87            match Backend::read::<Errno>(fd::AsFd::as_fd(&self.0), &mut byte) {
88                Ok([_]) => return Ok(()),
89                Ok(bytes) => {
90                    return Err(WaitError::Record {
91                        expected: 1,
92                        actual: bytes.len() as isize,
93                    });
94                }
95                Err(error) if error.is_intr() => {}
96                Err(error) => return Err(WaitError::Read(error)),
97            }
98        }
99    }
100
101    pub fn wait_timeout(&self, timeout: time::Duration) -> Result<(), WaitError> {
102        let started = time::Instant::now();
103        let total = timeout.as_nanos();
104        let mut delay = total.saturating_sub(started.elapsed().as_nanos());
105        loop {
106            let nanos = delay.min(i32::MAX as u128 * 1_000_000) as u64;
107            let millis = nanos.div_ceil(1_000_000) as i32;
108            let mut entries = [poll::Entry::readable(fd::AsFd::as_fd(&self.0))];
109            match Backend::wait::<_, Errno>(&mut entries, millis) {
110                Err(error) if error.is_intr() => {}
111                Err(error) => return Err(WaitError::Poll(error)),
112                Ok(events) => {
113                    let result = events.count() as i32;
114                    let entry = events.entries()[0];
115                    match (result, entry.revents()) {
116                        (1, _) if entry.is_readable() => return self.wait(),
117                        (0, 0) => {}
118                        (_, revents) => return Err(WaitError::PollResult { result, revents }),
119                    }
120                }
121            }
122            let elapsed = started.elapsed();
123            delay = total.saturating_sub(elapsed.as_nanos());
124            if delay == 0 {
125                return Err(WaitError::Timeout { timeout, elapsed });
126            }
127        }
128    }
129}
130
131impl Notify {
132    pub fn notify(&self) -> Result<(), NotifyError> {
133        loop {
134            match Backend::write::<Errno>(fd::AsFd::as_fd(&self.0), &[1]) {
135                Ok([_]) => return Ok(()),
136                Ok(bytes) => {
137                    return Err(NotifyError::Record {
138                        expected: 1,
139                        actual: bytes.len() as isize,
140                    });
141                }
142                Err(error) if error.is_intr() => {}
143                Err(error) => return Err(NotifyError::Write(error)),
144            }
145        }
146    }
147
148    pub fn fork(self) -> Result<(Self, Self), CreateError> {
149        let guard =
150            Backend::duplicate::<Errno>(fd::AsFd::as_fd(&self.0)).map_err(CreateError::Fork)?;
151        Ok((Self(guard, marker::PhantomData), self))
152    }
153}
154
155const _: () = {
156    assert!(mem::size_of::<Wait>() == mem::size_of::<fd::OwnedFd>());
157    assert!(mem::align_of::<Wait>() == mem::align_of::<fd::OwnedFd>());
158    assert!(mem::size_of::<Notify>() == mem::size_of::<fd::OwnedFd>());
159    assert!(mem::align_of::<Notify>() == mem::align_of::<fd::OwnedFd>());
160};
161
162impl CreateError {
163    fn from_native(error: pipe::Error) -> Self {
164        match error {
165            pipe::Error::Open(error) => Self::Open(error),
166            pipe::Error::CloseOnExec { end, failure } => Self::CloseOnExec {
167                end,
168                failure: Flags::cloexec(failure),
169            },
170            pipe::Error::Nonblocking { end, failure } => Self::Nonblocking {
171                end,
172                failure: Flags::nonblock(failure),
173            },
174        }
175    }
176}
177
178impl fd::AsFd for Wait {
179    fn as_fd(&self) -> fd::BorrowedFd<'_> {
180        fd::AsFd::as_fd(&self.0)
181    }
182}
183impl fd::AsFd for Notify {
184    fn as_fd(&self) -> fd::BorrowedFd<'_> {
185        fd::AsFd::as_fd(&self.0)
186    }
187}
dope_core/platform/wake.rs

dope_core/platform/
wake.rs
