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        for lane in self.iter_mut() {
138            lane.project().state.stop(driver);
139        }
140    }
141
142    pub fn progress<'a>(
143        &self,
144        region: &region::Token<region::Identity<'a>>,
145    ) -> schedule::Progress<'a> {
146        self.lanes
147            .iter()
148            .fold(schedule::Progress::Quiescent, |progress, lane| {
149                progress.reduce(lane.state.progress(region))
150            })
151    }
152
153    pub fn finish(&mut self, proof: &mut finish::Finish<'_, 'd>) {
154        for lane in self.iter_mut() {
155            lane.project().state.finish(proof);
156        }
157        self.active = 0;
158    }
159}
160
161/// An exclusive lane selection retains its owning count through the transition.
162///
163///
164pub struct Entry<'a, 'd, Tag>
165where
166    Tag: route::Tag<Class = kind::Accept>,
167{
168    lane: pin::Pin<&'a mut Lane<'d, Tag>>,
169    active: &'a mut usize,
170}
171
172impl<'a, 'd, Tag> Entry<'a, 'd, Tag>
173where
174    Tag: route::Tag<Class = kind::Accept>,
175{
176    pub fn begin(mut self) -> Option<accept::raw::Arming<'a, 'd, Tag>> {
177        if !self.lane.as_mut().project().state.start() {
178            return None;
179        }
180        Some(accept::raw::Arming::new(self))
181    }
182
183    /// Retires an operation while borrowing its owner's active count.
184    /// The matching lane must have reached native terminal completion.
185    /// Empty owners and pending flights cannot grant terminal access.
186    pub fn retire(mut self, token: route::Erased<'d, kind::Accept>) -> Retire<'a, 'd, Tag> {
187        let lane = self.lane.as_mut().project();
188        let retired = match lane.state.complete(token, Flow::Last) {
189            arm::Disposition::Discard => return Retire::Stale,
190            arm::Disposition::Deliver | arm::Disposition::Probe => {
191                Retired::Terminal(Terminal { lane: self.lane })
192            }
193            arm::Disposition::Retiring => Retired::Cancelled,
194        };
195        // A successful retirement consumes one live lane counted by this owner.
196        *self.active -= 1;
197        Retire::Done(retired)
198    }
199
200    /// Commits a previously idle lane and its owning count together.
201    /// The count cannot exceed the allocated lane slice's length.
202    fn apply(
203        &mut self,
204        result: Result<flight::Flight<'d, kind::Accept>, result::SubmitError>,
205        closed: retire::Completed,
206    ) -> arm::Submission {
207        let lane = self.lane.as_mut().project();
208        let result = lane.state.apply(result);
209        if let arm::Submission::Armed = result {
210            *lane.closed = closed;
211            // Entry owns one distinct idle lane in this count's owned slice.
212            *self.active += 1;
213        }
214        result
215    }
216}
217
218#[must_use]
219pub enum Retire<'a, 'd, Tag>
220where
221    Tag: route::Tag<Class = kind::Accept>,
222{
223    Stale,
224    Done(Retired<'a, 'd, Tag>),
225}
226
227#[must_use]
228pub enum Retired<'a, 'd, Tag>
229where
230    Tag: route::Tag<Class = kind::Accept>,
231{
232    Terminal(Terminal<'a, 'd, Tag>),
233    Cancelled,
234}
235
236#[must_use]
237pub struct Terminal<'a, 'd, Tag>
238where
239    Tag: route::Tag<Class = kind::Accept>,
240{
241    lane: pin::Pin<&'a mut Lane<'d, Tag>>,
242}
243
244impl<'d, Tag> Terminal<'_, 'd, Tag>
245where
246    Tag: route::Tag<Class = kind::Accept>,
247{
248    /// Preserves completed native capacity returns even when their CQE precedes this error.
249    pub fn can_defer(&self, failure: &Failure, driver: &mut driver::Context<'_, 'd>) -> bool {
250        failure.can_defer(driver)
251            || (matches!(failure, Failure::Syscall(_))
252                && failure.is_resource()
253                && self.lane.closed != driver.driver_ref().descriptors().retirement().completed())
254    }
255
256    pub fn into_addr(self) -> Result<net::SocketAddr, socket::Error> {
257        let storage = self.lane.project().addr.get_mut();
258        storage.to_std().map_err(socket::Error::from)
259    }
260}
261
262const _: () = {
263    type Tag = route::KeyTag<1, kind::Accept>;
264    const fn layout<'a, 'd: 'a>() {
265        assert!(
266            mem::size_of::<Lane<'d, Tag>>()
267                == mem::size_of::<(arm::State<'d, Tag>, Storage<Length>, retire::Completed)>()
268        );
269        assert!(
270            mem::align_of::<Lane<'d, Tag>>()
271                == mem::align_of::<(arm::State<'d, Tag>, Storage<Length>, retire::Completed)>()
272        );
273        assert!(
274            mem::size_of::<Slots<'d, Tag>>()
275                == mem::size_of::<(pin::Pin<Box<[Lane<'d, Tag>]>>, usize)>()
276        );
277        assert!(
278            mem::align_of::<Slots<'d, Tag>>() == mem::align_of::<pin::Pin<Box<[Lane<'d, Tag>]>>>()
279        );
280        assert!(mem::size_of::<Terminal<'a, 'd, Tag>>() == mem::size_of::<usize>());
281        assert!(mem::size_of::<Entry<'a, 'd, Tag>>() == 2 * mem::size_of::<usize>());
282        assert!(mem::size_of::<Option<Entry<'a, 'd, Tag>>>() == 2 * mem::size_of::<usize>());
283        assert!(mem::size_of::<accept::raw::Arming<'a, 'd, Tag>>() == 2 * mem::size_of::<usize>());
284    }
285    layout();
286};
dope_core/driver/retained/accept/mod.rs

dope_core/driver/retained/accept/
mod.rs
