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, FdSlot};
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
152pub enum AcceptEvent<'d> {
153    Accepted(AcceptedSlot<'d>),
154    Failed,
155}
156
157pub enum SocketEvent {
158    Created,
159    Failed(Error),
160}
161
162pub enum ConnectEvent {
163    Connected,
164    Failed(Error),
165}
166
167pub struct Event<'d> {
168    kind: EventKind<'d>,
169    result: i32,
170    operation: u8,
171}
172
173pub enum EventKind<'d> {
174    Accept(Token, bool, AcceptEvent<'d>),
175    Recv(Token, bool, RecvEvent<'d>),
176    Send(Token, SendEvent),
177    Timer(Token),
178    Socket(Token, SocketEvent),
179    Connect(Token, ConnectEvent),
180    Write(Token, WriteEvent),
181    Sync(Token, SyncEvent),
182    Open(Token, OpenEvent),
183    Read(Token, ReadEvent),
184    Stat(Token, StatEvent),
185    Shutdown,
186}
187
188pub enum EventRef<'a, 'd> {
189    Accept(Token, bool, &'a AcceptEvent<'d>),
190    Recv(Token, bool, &'a RecvEvent<'d>),
191    Send(Token, &'a SendEvent),
192    Timer(Token),
193    Socket(Token, &'a SocketEvent),
194    Connect(Token, &'a ConnectEvent),
195    Write(Token, &'a WriteEvent),
196    Sync(Token, &'a SyncEvent),
197    Open(Token, &'a OpenEvent),
198    Read(Token, &'a ReadEvent),
199    Stat(Token, &'a StatEvent),
200    Shutdown,
201}
202
203impl<'d> Event<'d> {
204    pub(crate) fn from_cqe(
205        cqe: Cqe,
206        reference: DriverRef<'d>,
207        provided: impl FnOnce(u32, u16) -> Option<CompletedBuffer>,
208    ) -> Result<Self, DecodeError> {
209        let result = cqe.result;
210        let operation = cqe.kind();
211        let mut provided = cqe
212            .has_buffer()
213            .then(|| provided(result.max(0) as u32, cqe.bid_raw()))
214            .flatten()
215            .map(|completed| ProvidedLease::from_completion(reference, completed));
216        let token = cqe.token;
217        let kind = if token == SHUTDOWN {
218            EventKind::Shutdown
219        } else {
220            match operation {
221                ACCEPT => {
222                    let event = if result >= 0 {
223                        AcceptEvent::Accepted(AcceptedSlot::from_completion(
224                            FdSlot::new(result as u32),
225                            reference,
226                        ))
227                    } else {
228                        AcceptEvent::Failed
229                    };
230                    EventKind::Accept(token, cqe.more(), event)
231                }
232                RECV => {
233                    let event = if result > 0 {
234                        if !cqe.has_buffer() {
235                            debug_assert!(false, "RECV data cqe without buffer flag");
236                            return Err(DecodeError);
237                        }
238                        let lease = provided.take().ok_or(DecodeError)?;
239                        debug_assert_eq!(lease.as_slice().len(), result as usize);
240                        RecvEvent::Data(lease)
241                    } else if result == 0 {
242                        RecvEvent::Eof
243                    } else {
244                        RecvEvent::from_errno(result)
245                    };
246                    EventKind::Recv(token, cqe.more(), event)
247                }
248                RECV_DISCARD => {
249                    let event = if result > 0 {
250                        RecvEvent::Discarded { len: result as u32 }
251                    } else if result == 0 {
252                        RecvEvent::Eof
253                    } else {
254                        RecvEvent::from_errno(result)
255                    };
256                    EventKind::Recv(token, cqe.more(), event)
257                }
258                SEND => EventKind::Send(
259                    token,
260                    if result >= 0 {
261                        SendEvent::Sent(result as u32)
262                    } else {
263                        SendEvent::Failed(-result)
264                    },
265                ),
266                WRITE => EventKind::Write(
267                    token,
268                    if result >= 0 {
269                        WriteEvent::Wrote(result as u32)
270                    } else {
271                        WriteEvent::Failed(-result)
272                    },
273                ),
274                SYNC => EventKind::Sync(
275                    token,
276                    if result >= 0 {
277                        SyncEvent::Synced
278                    } else {
279                        SyncEvent::Failed(-result)
280                    },
281                ),
282                OPEN => EventKind::Open(
283                    token,
284                    if result >= 0 {
285                        OpenEvent::Opened(Handle::take(result).into_owned())
286                    } else {
287                        OpenEvent::Failed(-result)
288                    },
289                ),
290                READ => EventKind::Read(token, ReadEvent::from_result(result)),
291                STAT => EventKind::Stat(
292                    token,
293                    if result >= 0 {
294                        StatEvent::Done
295                    } else {
296                        StatEvent::Failed(-result)
297                    },
298                ),
299                TIMER => EventKind::Timer(token),
300                SOCKET => EventKind::Socket(
301                    token,
302                    if result >= 0 {
303                        SocketEvent::Created
304                    } else {
305                        SocketEvent::Failed(Error::from_raw_os_error(-result))
306                    },
307                ),
308                CONNECT => EventKind::Connect(
309                    token,
310                    if result >= 0 {
311                        ConnectEvent::Connected
312                    } else {
313                        ConnectEvent::Failed(Error::from_raw_os_error(-result))
314                    },
315                ),
316                _ => return Err(DecodeError),
317            }
318        };
319        Ok(Self {
320            kind,
321            result,
322            operation,
323        })
324    }
325
326    pub fn into_kind(self) -> EventKind<'d> {
327        self.kind
328    }
329
330    pub fn as_ref(&self) -> EventRef<'_, 'd> {
331        match &self.kind {
332            EventKind::Accept(t, more, e) => EventRef::Accept(*t, *more, e),
333            EventKind::Recv(t, more, e) => EventRef::Recv(*t, *more, e),
334            EventKind::Send(t, e) => EventRef::Send(*t, e),
335            EventKind::Timer(t) => EventRef::Timer(*t),
336            EventKind::Socket(t, e) => EventRef::Socket(*t, e),
337            EventKind::Connect(t, e) => EventRef::Connect(*t, e),
338            EventKind::Write(t, e) => EventRef::Write(*t, e),
339            EventKind::Sync(t, e) => EventRef::Sync(*t, e),
340            EventKind::Open(t, e) => EventRef::Open(*t, e),
341            EventKind::Read(t, e) => EventRef::Read(*t, e),
342            EventKind::Stat(t, e) => EventRef::Stat(*t, e),
343            EventKind::Shutdown => EventRef::Shutdown,
344        }
345    }
346
347    pub const fn result(&self) -> i32 {
348        self.result
349    }
350
351    pub const fn operation(&self) -> u8 {
352        self.operation
353    }
354
355    pub const fn is_shutdown(&self) -> bool {
356        matches!(self.kind, EventKind::Shutdown)
357    }
358
359    pub const fn token(&self) -> Option<Token> {
360        match &self.kind {
361            EventKind::Accept(token, ..)
362            | EventKind::Recv(token, ..)
363            | EventKind::Send(token, _)
364            | EventKind::Timer(token)
365            | EventKind::Socket(token, _)
366            | EventKind::Connect(token, _)
367            | EventKind::Write(token, _)
368            | EventKind::Sync(token, _)
369            | EventKind::Open(token, _)
370            | EventKind::Read(token, _)
371            | EventKind::Stat(token, _) => Some(*token),
372            EventKind::Shutdown => None,
373        }
374    }
375
376    pub fn route(&self) -> u8 {
377        match &self.kind {
378            EventKind::Accept(t, ..) => t.route(),
379            EventKind::Recv(t, ..) => t.route(),
380            EventKind::Send(t, _) => t.route(),
381            EventKind::Timer(t) => t.route(),
382            EventKind::Socket(t, _) => t.route(),
383            EventKind::Connect(t, _) => t.route(),
384            EventKind::Write(t, _) => t.route(),
385            EventKind::Sync(t, _) => t.route(),
386            EventKind::Open(t, _) => t.route(),
387            EventKind::Read(t, _) => t.route(),
388            EventKind::Stat(t, _) => t.route(),
389            EventKind::Shutdown => SHUTDOWN.route(),
390        }
391    }
392}
dope_core/io/mod.rs

dope_core/io/
mod.rs
