dope_core/io/
mod.rs

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

dope_core/io/
mod.rs
