dope_core/io/event/net/
recv.rs

1use std::{mem, ops::ControlFlow};
2
3use dope_sys::table::permit::Queue as _;
4
5use crate::{
6    driver::{self, receive, route, route::kind},
7    io,
8    io::{
9        event,
10        event::{
11            batch::{Flow, Item},
12            entry::Entry,
13        },
14        recv,
15    },
16};
17
18pub enum Outcome<'d> {
19    Data(recv::Lease<'d>),
20    Eof,
21    Cancelled,
22    Exhausted(receive::Exhausted<'d>),
23    Starved(Retryable),
24    Failed(crate::Errno),
25}
26
27#[repr(transparent)]
28pub struct Completion<'d>(Entry<'d, kind::Recv, Item<Outcome<'d>>>);
29
30#[repr(transparent)]
31pub struct Data<'d>(Entry<'d, kind::Recv, Item<recv::Lease<'d>>>);
32
33#[repr(transparent)]
34pub struct Control<'d>(Entry<'d, kind::Recv, Item<ControlKind<'d>>>);
35
36#[derive(Clone, Copy)]
37pub enum Status {
38    Eof,
39    Cancelled,
40    Exhausted,
41    Starved(Retryable),
42    Failed(crate::Errno),
43}
44
45pub enum ControlKind<'d> {
46    Eof,
47    Cancelled,
48    Exhausted(receive::Exhausted<'d>),
49    Starved(Retryable),
50    Failed(crate::Errno),
51}
52
53pub enum Kind<'d> {
54    Data(Data<'d>),
55    Control(Control<'d>),
56}
57
58/// The exact native error proving that a receive operation can retry.
59///
60/// Arbitrary errors cannot enter a retryable completion:
61#[derive(Clone, Copy, Debug, Eq, PartialEq)]
62#[repr(transparent)]
63pub struct Retryable(crate::Errno);
64
65impl From<Retryable> for crate::Errno {
66    fn from(error: Retryable) -> Self {
67        error.0
68    }
69}
70
71impl<'d> Completion<'d> {
72    pub(crate) fn control(
73        target: route::Erased<'d, kind::Recv>,
74        outcome: Result<(), crate::Errno>,
75        flow: Flow,
76        driver: driver::Reference<'d>,
77    ) -> Self {
78        let outcome = match outcome {
79            Ok(()) => Outcome::Eof,
80            Err(errno) if errno == crate::Errno::NO_BUFFER_SPACE => {
81                if flow == Flow::Last && crate::Permits::try_acquire().is_ok() {
82                    Outcome::Exhausted(receive::Exhausted::from_driver(driver))
83                } else {
84                    Outcome::Starved(Retryable(errno))
85                }
86            }
87            Err(errno) if errno == crate::Errno::CANCELED => Outcome::Cancelled,
88            Err(errno) if errno.is_intr() || errno.is_block() => Outcome::Starved(Retryable(errno)),
89            Err(errno) => Outcome::Failed(errno),
90        };
91        Self::new(target, flow, outcome)
92    }
93
94    pub(in crate::io) const fn new(
95        target: route::Erased<'d, kind::Recv>,
96        flow: Flow,
97        outcome: Outcome<'d>,
98    ) -> Self {
99        Self(Entry::new(target, Item::new(flow, outcome)))
100    }
101
102    pub const fn token(&self) -> route::Erased<'d, kind::Recv> {
103        self.0.token()
104    }
105
106    pub const fn flow(&self) -> Flow {
107        self.0.value().flow()
108    }
109
110    /// Transfers the entire undelivered completion to driver reclamation.
111    pub fn retire(self) -> io::Event<'d> {
112        io::Event(event::Kind::RetiredRecv(self))
113    }
114
115    /// Retains the exact completion when its outcome cannot be delivered.
116    pub fn dispatch<T>(
117        self,
118        dispatch: impl FnOnce(
119            route::Erased<'d, kind::Recv>,
120            Flow,
121            Outcome<'d>,
122        ) -> ControlFlow<Outcome<'d>, T>,
123    ) -> ControlFlow<Self, T> {
124        let (target, flow, outcome) = self.into_parts();
125        dispatch(target, flow, outcome).map_break(|outcome| Self::new(target, flow, outcome))
126    }
127
128    pub fn into_parts(self) -> (route::Erased<'d, kind::Recv>, Flow, Outcome<'d>) {
129        let (target, batch) = self.0.split();
130        let (flow, outcome) = batch.into_parts();
131        (target, flow, outcome)
132    }
133
134    pub fn classify(self) -> Kind<'d> {
135        let (target, flow, outcome) = self.into_parts();
136        match outcome {
137            Outcome::Data(lease) => Kind::Data(Data(Entry::new(target, Item::new(flow, lease)))),
138            Outcome::Eof => Kind::Control(Control::new(target, flow, ControlKind::Eof)),
139            Outcome::Cancelled => Kind::Control(Control::new(target, flow, ControlKind::Cancelled)),
140            Outcome::Exhausted(state) => {
141                Kind::Control(Control::new(target, flow, ControlKind::Exhausted(state)))
142            }
143            Outcome::Starved(errno) => {
144                Kind::Control(Control::new(target, flow, ControlKind::Starved(errno)))
145            }
146            Outcome::Failed(errno) => {
147                Kind::Control(Control::new(target, flow, ControlKind::Failed(errno)))
148            }
149        }
150    }
151}
152
153impl<'d> Data<'d> {
154    pub fn into_parts(self) -> (route::Erased<'d, kind::Recv>, Flow, recv::Lease<'d>) {
155        let (target, batch) = self.0.split();
156        let (flow, lease) = batch.into_parts();
157        (target, flow, lease)
158    }
159
160    pub fn rejoin(self) -> Completion<'d> {
161        let (target, flow, lease) = self.into_parts();
162        Completion::new(target, flow, Outcome::Data(lease))
163    }
164}
165
166impl<'d> Control<'d> {
167    fn new(target: route::Erased<'d, kind::Recv>, flow: Flow, outcome: ControlKind<'d>) -> Self {
168        Self(Entry::new(target, Item::new(flow, outcome)))
169    }
170
171    pub const fn token(&self) -> route::Erased<'d, kind::Recv> {
172        self.0.token()
173    }
174
175    pub const fn flow(&self) -> Flow {
176        self.0.value().flow()
177    }
178
179    pub const fn status(&self) -> Status {
180        match self.0.value().value() {
181            ControlKind::Eof => Status::Eof,
182            ControlKind::Cancelled => Status::Cancelled,
183            ControlKind::Exhausted(_) => Status::Exhausted,
184            ControlKind::Starved(error) => Status::Starved(*error),
185            ControlKind::Failed(error) => Status::Failed(*error),
186        }
187    }
188
189    pub fn into_parts(self) -> (route::Erased<'d, kind::Recv>, Flow, Status, ControlKind<'d>) {
190        let status = self.status();
191        let (target, batch) = self.0.split();
192        let (flow, outcome) = batch.into_parts();
193        (target, flow, status, outcome)
194    }
195
196    pub fn rejoin(self) -> Completion<'d> {
197        let (target, batch) = self.0.split();
198        let (flow, outcome) = batch.into_parts();
199        Completion::new(target, flow, outcome.into())
200    }
201
202    pub fn dispatch<T, E>(
203        self,
204        dispatch: impl FnOnce(ControlKind<'d>) -> ControlFlow<(ControlKind<'d>, E), T>,
205    ) -> ControlFlow<(Self, E), T> {
206        let (target, batch) = self.0.split();
207        let (flow, outcome) = batch.into_parts();
208        dispatch(outcome).map_break(|(outcome, cause)| (Self::new(target, flow, outcome), cause))
209    }
210}
211
212impl<'d> From<ControlKind<'d>> for Outcome<'d> {
213    fn from(kind: ControlKind<'d>) -> Self {
214        match kind {
215            ControlKind::Eof => Self::Eof,
216            ControlKind::Cancelled => Self::Cancelled,
217            ControlKind::Exhausted(proof) => Self::Exhausted(proof),
218            ControlKind::Starved(error) => Self::Starved(error),
219            ControlKind::Failed(error) => Self::Failed(error),
220        }
221    }
222}
223
224impl<'d> From<Completion<'d>> for io::Event<'d> {
225    fn from(done: Completion<'d>) -> Self {
226        Self(event::Kind::Recv(done))
227    }
228}
229
230const _: () = {
231    assert!(mem::size_of::<Retryable>() == mem::size_of::<crate::Errno>());
232    assert!(mem::align_of::<Retryable>() == mem::align_of::<crate::Errno>());
233    assert!(mem::size_of::<Option<Retryable>>() == mem::size_of::<Option<crate::Errno>>());
234    assert!(mem::size_of::<[Retryable; 4]>() == mem::size_of::<[crate::Errno; 4]>());
235    assert!(
236        mem::size_of::<Completion<'static>>()
237            == mem::size_of::<(route::Erased<'static, kind::Recv>, Flow, Outcome<'static>)>()
238    );
239    assert!(
240        mem::align_of::<Completion<'static>>()
241            == mem::align_of::<(route::Erased<'static, kind::Recv>, Flow, Outcome<'static>)>()
242    );
243    assert!(
244        mem::size_of::<Data<'static>>()
245            == mem::size_of::<(
246                route::Erased<'static, kind::Recv>,
247                Flow,
248                recv::Lease<'static>,
249            )>()
250    );
251    assert!(
252        mem::align_of::<Data<'static>>()
253            == mem::align_of::<(
254                route::Erased<'static, kind::Recv>,
255                Flow,
256                recv::Lease<'static>,
257            )>()
258    );
259    assert!(
260        mem::size_of::<Control<'static>>()
261            == mem::size_of::<(
262                route::Erased<'static, kind::Recv>,
263                Flow,
264                ControlKind<'static>,
265            )>()
266    );
267    assert!(
268        mem::align_of::<Control<'static>>()
269            == mem::align_of::<(
270                route::Erased<'static, kind::Recv>,
271                Flow,
272                ControlKind<'static>,
273            )>()
274    );
275    assert!(mem::size_of::<Kind<'static>>() <= mem::size_of::<Completion<'static>>());
276};
dope_core/io/event/net/recv.rs

dope_core/io/event/net/
recv.rs
