o3/depot/queue/pool/fair/
mod.rs

1//! Region-owned FIFO lanes with fair unit admission embedded in their sidecars.
2
3use std::{cell, mem, num};
4
5use credit::fair;
6pub use lanes::Lanes;
7
8use crate::{
9    base::cell::region,
10    depot::{
11        arena::{linked, linked::slots},
12        capacity,
13        queue::pool,
14    },
15    mem::{Lease, Return, credit, lease},
16};
17
18mod lanes;
19
20#[derive(Clone, Copy, Eq, PartialEq)]
21pub struct Plan {
22    slots: capacity::Count,
23    lanes: num::NonZero<u32>,
24}
25
26#[derive(Debug, Clone, Copy, Eq, PartialEq)]
27pub enum Error {
28    ZeroLanes { capacity: usize },
29    CapacityBelowLanes { capacity: usize, lanes: usize },
30    Capacity(capacity::Error),
31}
32
33const _: () = assert!(std::mem::size_of::<Plan>() == 2 * std::mem::size_of::<u32>());
34
35impl Plan {
36    pub fn new(capacity: usize, lanes: usize) -> Result<Self, Error> {
37        if capacity < lanes {
38            return Err(Error::CapacityBelowLanes { capacity, lanes });
39        }
40        let Some(count) = num::NonZero::<u32>::new(lanes as u32) else {
41            return Err(if lanes == 0 {
42                Error::ZeroLanes { capacity }
43            } else {
44                Error::Capacity(capacity::Error::new(capacity))
45            });
46        };
47        let slots = capacity::Count::try_from(capacity).map_err(Error::Capacity)?;
48        Ok(Self {
49            slots,
50            lanes: count,
51        })
52    }
53}
54
55pub struct State<S> {
56    value: cell::UnsafeCell<S>,
57    held: cell::Cell<capacity::Count>,
58    index: capacity::Count,
59}
60
61impl<S: Default> Default for State<S> {
62    fn default() -> Self {
63        Self {
64            value: cell::UnsafeCell::new(S::default()),
65            held: cell::Cell::new(capacity::Count::EMPTY),
66            index: capacity::Count::EMPTY,
67        }
68    }
69}
70
71impl<S> State<S> {
72    pub const fn value(&self) -> &S {
73        // SAFETY: Lane retains the shared region borrow that excludes exclusive value access.
74        unsafe { &*self.value.get() }
75    }
76}
77
78/// One fair lane retaining its exact owner and exclusive region brand.
79pub struct Binding<'a, 'd, T, S = ()> {
80    owner: &'a Lanes<'d, T, S>,
81    slot: &'a cell::UnsafeCell<crate::depot::raw::QueueSlot<State<S>>>,
82}
83
84impl<'a, 'd, T, S> Binding<'a, 'd, T, S> {
85    /// Projects an admitted lane while retaining its exact owner and region.
86    pub fn from_source(
87        source: crate::depot::raw::Proof<(&'a Lanes<'d, T, S>, usize), capacity::Physical>,
88    ) -> Self {
89        let (owner, index) = source.into_inner();
90        // SAFETY: the retained physical proof bounds this index in the original
91        // owner, whose fixed lanes stay initialized for its complete borrow.
92        let slot = unsafe { &*owner.inner.slots.as_ptr().add(index).cast() };
93        Self { owner, slot }
94    }
95
96    pub const fn index(self) -> usize {
97        // SAFETY: bind retains this initialized slot's immutable original index.
98        unsafe { (*self.slot.get()).state.index.get() }
99    }
100
101    pub fn view<'t>(
102        self,
103        _: &'t region::Token<region::Identity<'d>>,
104    ) -> pool::View<'t, T, State<S>, false>
105    where
106        'a: 't,
107    {
108        // SAFETY: the matching shared token excludes exclusive lane loans for 't.
109        unsafe { pool::View::new(&self.owner.inner.store, &*self.slot.get()) }
110    }
111
112    fn lane_mut<'t>(
113        self,
114        _: &'t mut region::Token<region::Identity<'d>>,
115    ) -> pool::Lane<'t, T, State<S>, false>
116    where
117        'a: 't,
118    {
119        // SAFETY: the matching exclusive token excludes every competing lane
120        // loan; only fair transitions may consume slots or access the ledger.
121        unsafe { pool::Lane::new(&self.owner.inner.store, &mut *self.slot.get()) }
122    }
123
124    fn state(self) -> &'a State<S> {
125        // SAFETY: fair admission touches only the Cell ledger; value loans stay
126        // inside its disjoint UnsafeCell and require the matching region token.
127        unsafe { &(*self.slot.get()).state }
128    }
129
130    pub fn reserve(self) -> Result<Permit<'a, 'd, T, S>, fair::Cause<usize>> {
131        let state = self.state();
132        let held = state.held.get();
133        let shared = self.owner.shared.get();
134        let balance = crate::mem::raw::Balance::new([held.raw()], [shared.raw()]);
135        let [required] = balance.demand([1], [1]);
136        if required > shared.raw() {
137            return Err(fair::Cause::Exhausted {
138                axis: fair::Axis::at::<0>(),
139                required: 1,
140                available: 0,
141            });
142        }
143        // SAFETY: each lane retains its first slot, and the remaining shared
144        // demand was admitted above. Total storage fits u32, so success cannot
145        // overflow held; both counters still describe the original allocation.
146        let balance = unsafe { balance.acquire_raw([1], [1]) };
147        let ([held], [shared]) = balance.into_parts();
148        state.held.set(capacity::Count::new(held));
149        self.owner.shared.set(capacity::Count::new(shared));
150        Ok(Permit::new(self))
151    }
152
153    pub fn take_front(
154        self,
155        token: &mut region::Token<region::Identity<'d>>,
156    ) -> Option<(T, Permit<'a, 'd, T, S>)> {
157        let lane = self.lane_mut(token);
158        let index = lane.slot.queue.front_index()?;
159        // SAFETY: the original Binding owns this lane's unchanged live front.
160        let (value, vacant) = unsafe { lane.slot.queue.take(self, index) };
161        Some((value, Permit::new(vacant.release())))
162    }
163}
164
165impl<T, S> Copy for Binding<'_, '_, T, S> {}
166
167impl<T, S> Clone for Binding<'_, '_, T, S> {
168    fn clone(&self) -> Self {
169        *self
170    }
171}
172
173// SAFETY: Binding retains the same fair allocation for its complete lifetime.
174impl<T, S> linked::Sealed<T> for Binding<'_, '_, T, S> {}
175
176impl<T, S> linked::Source<T> for Binding<'_, '_, T, S> {
177    const TRACK: bool = false;
178
179    fn store(&self) -> &slots::Store<T> {
180        &self.owner.inner.store
181    }
182}
183
184/// One unused queue slot entitlement bound to its original fair lane.
185#[must_use = "a fair queue reservation must be published or returned"]
186pub struct Permit<'a, 'd, T, S = ()> {
187    lease: Lease<Refund<'a, 'd, T, S>, lease::Open>,
188}
189
190struct Refund<'a, 'd, T, S> {
191    binding: Binding<'a, 'd, T, S>,
192}
193
194impl<'a, 'd, T, S> Permit<'a, 'd, T, S> {
195    fn new(binding: Binding<'a, 'd, T, S>) -> Self {
196        Self {
197            lease: Lease::open(Refund { binding }, ()),
198        }
199    }
200
201    pub fn state_mut<'b>(
202        &'b self,
203        token: &'b mut region::Token<region::Identity<'d>>,
204    ) -> &'b mut S {
205        self.lease
206            .sink()
207            .binding
208            .lane_mut(token)
209            .into_state()
210            .value
211            .get_mut()
212    }
213
214    pub fn push_back(self, token: &mut region::Token<region::Identity<'d>>, value: T) {
215        self.into_claim(token).append(value);
216    }
217
218    /// Restores a deferred head using its retained fair-lane entitlement.
219    pub fn push_front(self, token: &mut region::Token<region::Identity<'d>>, value: T) {
220        crate::mem::Return::return_item(self.into_claim(token), value);
221    }
222
223    fn into_claim<'t>(
224        self,
225        token: &'t mut region::Token<region::Identity<'d>>,
226    ) -> slots::Vacant<T, slots::Bound<'t, T, (), false>>
227    where
228        'a: 't,
229    {
230        let (Refund { binding }, ()) = self.lease.into_parts();
231        let lane = binding.lane_mut(token);
232        // SAFETY: the exclusive token retains the original fair lane and its
233        // untracked slot policy; the fair ledger owns this affine entitlement.
234        let owner = unsafe { lane.slot.queue.bind(lane.store) };
235        // SAFETY: this affine permit owns one unused slot in this exact allocation.
236        unsafe { slots::Vacant::reserve(owner) }
237    }
238}
239
240impl<T, S> Return for Refund<'_, '_, T, S> {
241    type Item = ();
242
243    fn return_item(self, (): ()) {
244        let binding = self.binding;
245        let state = binding.state();
246        let held = state.held.get();
247        let shared = binding.owner.shared.get();
248        // SAFETY: this affine refund owns exactly one held unit in the original
249        // lane. Returning it cannot exceed the allocation's shared capacity.
250        let balance = unsafe {
251            crate::mem::raw::Balance::new([held.raw()], [shared.raw()]).release_raw([1], [1])
252        };
253        let ([held], [shared]) = balance.into_parts();
254        state.held.set(capacity::Count::new(held));
255        binding.owner.shared.set(capacity::Count::new(shared));
256    }
257}
258
259const _: () = {
260    const WORD: usize = mem::size_of::<usize>();
261
262    assert!(mem::size_of::<State<u64>>() == 2 * WORD);
263    assert!(mem::size_of::<Binding<'static, 'static, u64>>() == 2 * WORD);
264    assert!(mem::size_of::<Permit<'static, 'static, u64>>() == 2 * WORD);
265    assert!(mem::size_of::<Option<Permit<'static, 'static, u64>>>() == 2 * WORD);
266};
o3/depot/queue/pool/fair/mod.rs

o3/depot/queue/pool/fair/
mod.rs
