dope_core/driver/retained/accept/
mod.rs

1use std::{cell, mem, net, pin};
2
3use dope_sys::io::socket::addr::{Length, Storage};
4use o3::base::cell::region;
5
6use crate::{
7    driver,
8    driver::{
9        life::finish,
10        result,
11        retained::{accept, arm, flight},
12        route,
13        route::kind,
14        schedule,
15    },
16    io::{
17        event::{batch::Flow, net::accept::Failure},
18        fd::domain::retire,
19        socket,
20    },
21};
22
23pub mod raw;
24
25/// Current descriptor-retirement barrier for admitting a new accept.
26#[derive(Debug)]
27pub enum Readiness {
28    Ready,
29    Pending,
30    Inflight,
31}
32
33impl From<driver::Reference<'_>> for Readiness {
34    /// Query again after driver mutations; this snapshot is not an admission permit.
35    fn from(driver: driver::Reference<'_>) -> Self {
36        let retired = driver.descriptors().retirement();
37        if retired.has_pending() {
38            Self::Pending
39        } else if retired.has_inflight() {
40            Self::Inflight
41        } else {
42            Self::Ready
43        }
44    }
45}
46
47impl Readiness {
48    pub fn progress<'a>(
49        self,
50        region: &region::Token<region::Identity<'a>>,
51    ) -> schedule::Progress<'a> {
52        match self {
53            Self::Ready | Self::Pending => schedule::Progress::Runnable,
54            Self::Inflight => schedule::Progress::waiting(region),
55        }
56    }
57}
58
59#[pin_project::pin_project(!Unpin)]
60pub struct Lane<'d, Tag>
61where
62    Tag: route::Tag<Class = kind::Accept>,
63{
64    state: arm::State<'d, Tag>,
65    addr: cell::UnsafeCell<Storage<Length>>,
66    closed: retire::Completed,
67}
68
69impl<'d, Tag> Lane<'d, Tag>
70where
71    Tag: route::Tag<Class = kind::Accept>,
72{
73    pub fn new(target: route::Target<'d, Tag>) -> Self {
74        Self {
75            state: arm::State::new(target),
76            addr: cell::UnsafeCell::new(Storage::<Length>::empty()),
77            closed: retire::Completed::EMPTY,
78        }
79    }
80}
81
82/// Pinned accept lanes and their exact in-flight count share one owner.
83///
84pub struct Slots<'d, Tag>
85where
86    Tag: route::Tag<Class = kind::Accept>,
87{
88    lanes: pin::Pin<Box<[Lane<'d, Tag>]>>,
89    active: usize,
90}
91
92impl<'d, Tag> Slots<'d, Tag>
93where
94    Tag: route::Tag<Class = kind::Accept>,
95{
96    pub fn new(lanes: Box<[Lane<'d, Tag>]>) -> Self {
97        Self {
98            lanes: Box::into_pin(lanes),
99            active: 0,
100        }
101    }
102
103    pub fn get_mut(&mut self, index: usize) -> Option<Entry<'_, 'd, Tag>> {
104        // SAFETY: the selected lane remains pinned in this owned allocation.
105        let lane = unsafe { self.lanes.as_mut().get_unchecked_mut() }.get_mut(index)?;
106        Some(Entry {
107            lane: unsafe { pin::Pin::new_unchecked(lane) },
108            active: &mut self.active,
109        })
110    }
111
112    fn iter_mut(
113        &mut self,
114    ) -> impl DoubleEndedIterator<Item = pin::Pin<&mut Lane<'d, Tag>>> + ExactSizeIterator + '_
115    {
116        // SAFETY: the boxed slice keeps every lane at a stable address, and
117        // the iterator exposes each exclusive borrow only as pinned.
118        unsafe { self.lanes.as_mut().get_unchecked_mut() }
119            .iter_mut()
120            .map(|lane| unsafe { pin::Pin::new_unchecked(lane) })
121    }
122
123    pub fn is_empty(&self) -> bool {
124        self.lanes.is_empty()
125    }
126
127    pub fn len(&self) -> usize {
128        self.lanes.len()
129    }
130
131    /// Returns the number of submitted accepts awaiting terminal retirement.
132    pub const fn active(&self) -> usize {
133        self.active
134    }
135
136    pub fn stop(&mut self, driver: &mut driver::Context<'_, 'd>) {
137        let mut active = 0;
138        for lane in self.iter_mut() {
139            let state = lane.project().state;
140            state.stop(driver);
141            active += usize::from(state.is_live());
142        }
143        self.active = active;
144    }
145
146    pub fn progress<'a>(
147        &self,
148        region: &region::Token<region::Identity<'a>>,
149    ) -> schedule::Progress<'a> {
150        self.lanes
151            .iter()
152            .fold(schedule::Progress::Quiescent, |progress, lane| {
153                progress.reduce(lane.state.progress(region))
154            })
155    }
156
157    pub fn finish(&mut self, proof: &mut finish::Finish<'_, 'd>) {
158        for lane in self.iter_mut() {
159            lane.project().state.finish(proof);
160        }
161        self.active = 0;
162    }
163}
164
165/// An exclusive lane selection retains its owning count through the transition.
166///
167///
168pub struct Entry<'a, 'd, Tag>
169where
170    Tag: route::Tag<Class = kind::Accept>,
171{
172    lane: pin::Pin<&'a mut Lane<'d, Tag>>,
173    active: &'a mut usize,
174}
175
176impl<'a, 'd, Tag> Entry<'a, 'd, Tag>
177where
178    Tag: route::Tag<Class = kind::Accept>,
179{
180    pub fn begin(mut self) -> Option<accept::raw::Arming<'a, 'd, Tag>> {
181        if !self.lane.as_mut().project().state.start() {
182            return None;
183        }
184        Some(accept::raw::Arming::new(self))
185    }
186
187    /// Retires an operation while borrowing its owner's active count.
188    /// The matching lane must have reached native terminal completion.
189    /// Empty owners and pending flights cannot grant terminal access.
190    pub fn retire(mut self, token: route::Erased<'d, kind::Accept>) -> Retire<'a, 'd, Tag> {
191        let lane = self.lane.as_mut().project();
192        let retired = match lane.state.complete(token, Flow::Last) {
193            arm::Disposition::Discard => return Retire::Stale,
194            arm::Disposition::Deliver | arm::Disposition::Probe => {
195                Retired::Terminal(Terminal { lane: self.lane })
196            }
197            arm::Disposition::Retiring => Retired::Cancelled,
198        };
199        // A successful retirement consumes one live lane counted by this owner.
200        *self.active -= 1;
201        Retire::Done(retired)
202    }
203
204    /// Commits a previously idle lane and its owning count together.
205    /// The count cannot exceed the allocated lane slice's length.
206    fn apply(
207        &mut self,
208        result: Result<flight::Flight<'d, kind::Accept>, result::SubmitError>,
209        closed: retire::Completed,
210    ) -> arm::Submission {
211        let lane = self.lane.as_mut().project();
212        let result = lane.state.apply(result);
213        if let arm::Submission::Armed = result {
214            *lane.closed = closed;
215            // Entry owns one distinct idle lane in this count's owned slice.
216            *self.active += 1;
217        }
218        result
219    }
220}
221
222#[must_use]
223pub enum Retire<'a, 'd, Tag>
224where
225    Tag: route::Tag<Class = kind::Accept>,
226{
227    Stale,
228    Done(Retired<'a, 'd, Tag>),
229}
230
231#[must_use]
232pub enum Retired<'a, 'd, Tag>
233where
234    Tag: route::Tag<Class = kind::Accept>,
235{
236    Terminal(Terminal<'a, 'd, Tag>),
237    Cancelled,
238}
239
240#[must_use]
241pub struct Terminal<'a, 'd, Tag>
242where
243    Tag: route::Tag<Class = kind::Accept>,
244{
245    lane: pin::Pin<&'a mut Lane<'d, Tag>>,
246}
247
248impl<'d, Tag> Terminal<'_, 'd, Tag>
249where
250    Tag: route::Tag<Class = kind::Accept>,
251{
252    /// Preserves completed native capacity returns even when their CQE precedes this error.
253    pub fn can_defer(&self, failure: &Failure, driver: &mut driver::Context<'_, 'd>) -> bool {
254        failure.can_defer(driver)
255            || (matches!(failure, Failure::Syscall(_))
256                && failure.is_resource()
257                && self.lane.closed != driver.driver_ref().descriptors().retirement().completed())
258    }
259
260    pub fn into_addr(self) -> Result<net::SocketAddr, socket::Error> {
261        let storage = *self.lane.project().addr.get_mut();
262        storage.to_std().map_err(socket::Error::from)
263    }
264
265    pub fn discard(self) {
266        *self.lane.project().addr.get_mut() = Storage::<Length>::empty();
267    }
268}
269
270const _: () = {
271    type Tag = route::KeyTag<1, kind::Accept>;
272    const fn layout<'a, 'd: 'a>() {
273        assert!(
274            mem::size_of::<Lane<'d, Tag>>()
275                == mem::size_of::<(arm::State<'d, Tag>, Storage<Length>, retire::Completed)>()
276        );
277        assert!(
278            mem::align_of::<Lane<'d, Tag>>()
279                == mem::align_of::<(arm::State<'d, Tag>, Storage<Length>, retire::Completed)>()
280        );
281        assert!(
282            mem::size_of::<Slots<'d, Tag>>()
283                == mem::size_of::<(pin::Pin<Box<[Lane<'d, Tag>]>>, usize)>()
284        );
285        assert!(
286            mem::align_of::<Slots<'d, Tag>>() == mem::align_of::<pin::Pin<Box<[Lane<'d, Tag>]>>>()
287        );
288        assert!(mem::size_of::<Terminal<'a, 'd, Tag>>() == mem::size_of::<usize>());
289        assert!(mem::size_of::<Entry<'a, 'd, Tag>>() == 2 * mem::size_of::<usize>());
290        assert!(mem::size_of::<Option<Entry<'a, 'd, Tag>>>() == 2 * mem::size_of::<usize>());
291        assert!(mem::size_of::<accept::raw::Arming<'a, 'd, Tag>>() == 2 * mem::size_of::<usize>());
292    }
293    layout();
294};
dope_core/driver/retained/accept/mod.rs

dope_core/driver/retained/accept/
mod.rs
