dope_core/io/
mod.rs

1pub mod datagram;
2pub mod fd;
3pub(crate) mod ffi;
4pub mod file;
5pub mod pipe;
6pub mod provided;
7pub mod socket;
8
9use std::error::Error as StdError;
10use std::fmt;
11use std::io::Error;
12use std::os::fd::OwnedFd;
13
14use crate::driver::DriverRef;
15use crate::driver::token::kind::ACCEPT;
16use crate::driver::token::kind::CONNECT;
17use crate::driver::token::kind::OPEN;
18use crate::driver::token::kind::READ;
19use crate::driver::token::kind::RECV;
20use crate::driver::token::kind::RECV_DISCARD;
21use crate::driver::token::kind::SEND;
22use crate::driver::token::kind::SOCKET;
23use crate::driver::token::kind::STAT;
24use crate::driver::token::kind::SYNC;
25use crate::driver::token::kind::TIMER;
26use crate::driver::token::kind::WRITE;
27use crate::driver::token::{SHUTDOWN, Token};
28use fd::AcceptedSlot;
29use ffi::Handle;
30use libc::EAGAIN;
31use libc::ECANCELED;
32use libc::EINTR;
33use libc::ENOBUFS;
34use provided::ProvidedLease;
35use provided::raw::completion::CompletedBuffer;
36use std::fmt::Display;
37use std::fmt::Formatter;
38
39#[derive(Clone, Copy, Debug, PartialEq, Eq)]
40pub struct DecodeError;
41
42impl Display for DecodeError {
43    fn fmt(&self, f: &mut Formatter<'_>) -> fmt::Result {
44        f.write_str("invalid completion")
45    }
46}
47
48impl StdError for DecodeError {}
49
50pub(crate) const BUFFER: u32 = 1 << 0;
51pub(crate) const MORE: u32 = 1 << 1;
52pub(crate) const BUFFER_SHIFT: u32 = 16;
53
54#[derive(Clone, Copy)]
55pub(crate) struct Cqe {
56    token: Token,
57    result: i32,
58    flags: u32,
59}
60
61impl Cqe {
62    pub(crate) const fn new(token: Token, result: i32, flags: u32) -> Self {
63        Self {
64            token,
65            result,
66            flags,
67        }
68    }
69
70    pub(crate) fn kind(self) -> u8 {
71        self.token.kind()
72    }
73
74    fn more(self) -> bool {
75        self.flags & MORE != 0
76    }
77
78    fn bid_raw(self) -> u16 {
79        (self.flags >> BUFFER_SHIFT) as u16
80    }
81
82    fn has_buffer(self) -> bool {
83        self.flags & BUFFER != 0
84    }
85}
86
87pub enum RecvEvent<'d> {
88    Data(ProvidedLease<'d>),
89    Discarded { len: u32 },
90    Eof,
91    Cancelled,
92    Starved,
93    Failed(i32),
94}
95
96impl RecvEvent<'_> {
97    fn from_errno(result: i32) -> Self {
98        match -result {
99            ECANCELED => Self::Cancelled,
100            ENOBUFS | EAGAIN | EINTR => Self::Starved,
101            errno => Self::Failed(errno),
102        }
103    }
104}
105
106#[derive(Clone, Copy)]
107pub enum SendEvent {
108    Sent(u32),
109    Failed(i32),
110}
111
112#[derive(Clone, Copy)]
113pub enum WriteEvent {
114    Wrote(u32),
115    Failed(i32),
116}
117
118#[derive(Clone, Copy)]
119pub enum SyncEvent {
120    Synced,
121    Failed(i32),
122}
123
124pub enum OpenEvent {
125    Opened(OwnedFd),
126    Failed(i32),
127}
128
129#[derive(Clone, Copy)]
130pub enum ReadEvent {
131    Read(u32),
132    Eof,
133    Failed(i32),
134}
135
136impl ReadEvent {
137    fn from_result(result: i32) -> Self {
138        match result {
139            n if n > 0 => Self::Read(n as u32),
140            0 => Self::Eof,
141            n => Self::Failed(-n),
142        }
143    }
144}
145
146#[derive(Clone, Copy)]
147pub enum StatEvent {
148    Done,
149    Failed(i32),
150}
151
152#[derive(Clone, Copy)]
153pub struct TimerEvent(i32);
154
155impl TimerEvent {
156    const fn from_result(result: i32) -> Self {
157        Self(result)
158    }
159
160    pub const fn raw_result(self) -> i32 {
161        self.0
162    }
163
164    pub const fn is_cancelled(self) -> bool {
165        self.0 == -ECANCELED
166    }
167}
168
169pub enum AcceptEvent<'d> {
170    Accepted(AcceptedSlot<'d>),
171    Failed(i32),
172}
173
174pub enum SocketEvent {
175    Created,
176    Failed(Error),
177}
178
179pub enum ConnectEvent {
180    Connected,
181    Failed(Error),
182}
183
184pub enum Event<'d> {
185    Accept(Token, bool, AcceptEvent<'d>),
186    Recv(Token, bool, RecvEvent<'d>),
187    Send(Token, SendEvent),
188    Timer(Token, TimerEvent),
189    Socket(Token, SocketEvent),
190    Connect(Token, ConnectEvent),
191    Write(Token, WriteEvent),
192    Sync(Token, SyncEvent),
193    Open(Token, OpenEvent),
194    Read(Token, ReadEvent),
195    Stat(Token, StatEvent),
196    Shutdown,
197}
198
199impl<'d> Event<'d> {
200    pub(crate) fn from_cqe(
201        cqe: Cqe,
202        reference: DriverRef<'d>,
203        provided: impl FnOnce(u32, u16) -> Option<CompletedBuffer>,
204    ) -> Result<Self, DecodeError> {
205        let result = cqe.result;
206        let operation = cqe.kind();
207        let mut provided = cqe
208            .has_buffer()
209            .then(|| provided(result.max(0) as u32, cqe.bid_raw()))
210            .flatten()
211            .map(|completed| ProvidedLease::from_completion(reference, completed));
212        let token = cqe.token;
213        let event = if token == SHUTDOWN {
214            Event::Shutdown
215        } else {
216            match operation {
217                ACCEPT => {
218                    let event = if result >= 0 {
219                        let slot = reference.fixed_fd_slot(result as u32).ok_or(DecodeError)?;
220                        AcceptEvent::Accepted(AcceptedSlot::from_completion(slot, reference))
221                    } else {
222                        AcceptEvent::Failed(-result)
223                    };
224                    Event::Accept(token, cqe.more(), event)
225                }
226                RECV => {
227                    let event = if result > 0 {
228                        if !cqe.has_buffer() {
229                            debug_assert!(false, "RECV data cqe without buffer flag");
230                            return Err(DecodeError);
231                        }
232                        let lease = provided.take().ok_or(DecodeError)?;
233                        debug_assert_eq!(lease.as_slice().len(), result as usize);
234                        RecvEvent::Data(lease)
235                    } else if result == 0 {
236                        RecvEvent::Eof
237                    } else {
238                        RecvEvent::from_errno(result)
239                    };
240                    Event::Recv(token, cqe.more(), event)
241                }
242                RECV_DISCARD => {
243                    let event = if result > 0 {
244                        RecvEvent::Discarded { len: result as u32 }
245                    } else if result == 0 {
246                        RecvEvent::Eof
247                    } else {
248                        RecvEvent::from_errno(result)
249                    };
250                    Event::Recv(token, cqe.more(), event)
251                }
252                SEND => Event::Send(
253                    token,
254                    if result >= 0 {
255                        SendEvent::Sent(result as u32)
256                    } else {
257                        SendEvent::Failed(-result)
258                    },
259                ),
260                WRITE => Event::Write(
261                    token,
262                    if result >= 0 {
263                        WriteEvent::Wrote(result as u32)
264                    } else {
265                        WriteEvent::Failed(-result)
266                    },
267                ),
268                SYNC => Event::Sync(
269                    token,
270                    if result >= 0 {
271                        SyncEvent::Synced
272                    } else {
273                        SyncEvent::Failed(-result)
274                    },
275                ),
276                OPEN => Event::Open(
277                    token,
278                    if result >= 0 {
279                        OpenEvent::Opened(Handle::take(result).into_owned())
280                    } else {
281                        OpenEvent::Failed(-result)
282                    },
283                ),
284                READ => Event::Read(token, ReadEvent::from_result(result)),
285                STAT => Event::Stat(
286                    token,
287                    if result >= 0 {
288                        StatEvent::Done
289                    } else {
290                        StatEvent::Failed(-result)
291                    },
292                ),
293                TIMER => Event::Timer(token, TimerEvent::from_result(result)),
294                SOCKET => Event::Socket(
295                    token,
296                    if result >= 0 {
297                        SocketEvent::Created
298                    } else {
299                        SocketEvent::Failed(Error::from_raw_os_error(-result))
300                    },
301                ),
302                CONNECT => Event::Connect(
303                    token,
304                    if result >= 0 {
305                        ConnectEvent::Connected
306                    } else {
307                        ConnectEvent::Failed(Error::from_raw_os_error(-result))
308                    },
309                ),
310                _ => return Err(DecodeError),
311            }
312        };
313        Ok(event)
314    }
315
316    pub const fn is_shutdown(&self) -> bool {
317        matches!(self, Event::Shutdown)
318    }
319
320    pub const fn token(&self) -> Option<Token> {
321        match self {
322            Event::Accept(token, ..)
323            | Event::Recv(token, ..)
324            | Event::Send(token, _)
325            | Event::Timer(token, _)
326            | Event::Socket(token, _)
327            | Event::Connect(token, _)
328            | Event::Write(token, _)
329            | Event::Sync(token, _)
330            | Event::Open(token, _)
331            | Event::Read(token, _)
332            | Event::Stat(token, _) => Some(*token),
333            Event::Shutdown => None,
334        }
335    }
336
337    pub fn route(&self) -> u8 {
338        match self {
339            Event::Accept(t, ..) => t.route(),
340            Event::Recv(t, ..) => t.route(),
341            Event::Send(t, _) => t.route(),
342            Event::Timer(t, _) => t.route(),
343            Event::Socket(t, _) => t.route(),
344            Event::Connect(t, _) => t.route(),
345            Event::Write(t, _) => t.route(),
346            Event::Sync(t, _) => t.route(),
347            Event::Open(t, _) => t.route(),
348            Event::Read(t, _) => t.route(),
349            Event::Stat(t, _) => t.route(),
350            Event::Shutdown => SHUTDOWN.route(),
351        }
352    }
353}
dope_core/io/mod.rs

dope_core/io/
mod.rs
