base_pool/
singleplex.rs

1// Released under the MIT License.
2// Copyright, 2026, by Samuel Williams.
3
4use crate::{
5    Acquisition, Error, Options, Pool, ResourceId, Retirement, Stats,
6    core::{Acquired, Core, Slot},
7};
8use base_scheduler::Cancellation;
9use std::{
10    ops::{Deref, DerefMut},
11    sync::{Arc, Weak},
12};
13
14struct SingleSlot<R> {
15    resource: Option<R>,
16    draining: bool,
17}
18impl<R: Send> Slot for SingleSlot<R> {
19    type Resource = R;
20    type Acquired = R;
21    fn available(&self) -> bool {
22        !self.draining && self.resource.is_some()
23    }
24    fn acquire(&mut self) -> R {
25        self.resource.take().unwrap()
26    }
27    fn uses(&self) -> usize {
28        usize::from(self.resource.is_none())
29    }
30    fn draining(&self) -> bool {
31        self.draining
32    }
33    fn drain(&mut self) {
34        self.draining = true;
35    }
36    fn into_resource(self) -> R {
37        self.resource.unwrap()
38    }
39}
40
41/// Exclusive resource ownership. Resources need `Send`, without requiring `Sync`.
42/// Clean reuse is an explicit lease release; unfinished lease drop retires it.
43pub struct SingleplexPool<R: Send> {
44    core: Arc<Core<SingleSlot<R>>>,
45}
46impl<R: Send> Clone for SingleplexPool<R> {
47    fn clone(&self) -> Self {
48        Self {
49            core: self.core.clone(),
50        }
51    }
52}
53impl<R: Send> SingleplexPool<R> {
54    pub fn new(options: Options) -> Result<Self, Error> {
55        Ok(Self {
56            core: Core::new(options)?,
57        })
58    }
59    pub async fn acquire_or_reserve(
60        &self,
61        cancel: &Cancellation,
62    ) -> Result<Acquisition<SingleplexLease<R>, SingleplexBuildPermit<R>>, Error> {
63        Ok(match self.core.acquire(cancel).await? {
64            Acquired::Existing(id, resource) => Acquisition::Existing(SingleplexLease {
65                core: self.core.clone(),
66                id,
67                resource: Some(resource),
68            }),
69            Acquired::Build(id) => Acquisition::Build(SingleplexBuildPermit {
70                core: self.core.clone(),
71                id,
72                pending: true,
73            }),
74        })
75    }
76    pub fn begin_close(&self) {
77        self.core.begin_close();
78    }
79    pub fn stats(&self) -> Stats {
80        self.core.stats()
81    }
82    /// Retire unused registrations until at most `retain` live resources remain,
83    /// or until only in-use resources remain. No I/O occurs here.
84    pub fn prune(&self, retain: usize) -> usize {
85        self.core.prune(retain)
86    }
87    pub fn try_retirement(&self) -> Option<Retirement<R>> {
88        self.core.retirements.pop()
89    }
90    /// Wait for cleanup work; `None` means admission is closed and all resources
91    /// and construction/retirement obligations have finished.
92    pub async fn next_retirement(
93        &self,
94        cancel: &Cancellation,
95    ) -> Result<Option<Retirement<R>>, Error> {
96        self.core.next_retirement(cancel).await
97    }
98    /// Call `begin_close`, finish active work and drain retirements concurrently.
99    /// This waits for acknowledgments; it does not execute resource cleanup.
100    pub async fn wait_closed(&self, cancel: &Cancellation) -> Result<(), Error> {
101        self.core.wait_closed(cancel).await
102    }
103}
104impl<R: Send> Pool for SingleplexPool<R> {
105    type Permit = SingleplexLease<R>;
106    type BuildPermit = SingleplexBuildPermit<R>;
107    async fn acquire_or_reserve(
108        &self,
109        cancel: &Cancellation,
110    ) -> Result<Acquisition<Self::Permit, Self::BuildPermit>, Error> {
111        self.acquire_or_reserve(cancel).await
112    }
113    fn begin_close(&self) {
114        self.begin_close();
115    }
116    fn stats(&self) -> Stats {
117        self.stats()
118    }
119}
120
121/// A reserved construction slot. Register the fully established resource, then
122/// acquire again to use it. Driver registration must precede this commit.
123/// Dropping a permit returns its construction reservation exactly once.
124pub struct SingleplexBuildPermit<R: Send> {
125    core: Arc<Core<SingleSlot<R>>>,
126    id: ResourceId,
127    pending: bool,
128}
129impl<R: Send> SingleplexBuildPermit<R> {
130    pub fn id(&self) -> ResourceId {
131        self.id
132    }
133    /// A commit racing shutdown transfers the resource into the retirement queue
134    /// and returns `Closed`; the cleanup owner must still drain that queue.
135    pub fn commit(mut self, resource: R) -> Result<SingleplexResource<R>, Error> {
136        self.core.commit(
137            self.id,
138            SingleSlot {
139                resource: Some(resource),
140                draining: false,
141            },
142        )?;
143        self.pending = false;
144        Ok(SingleplexResource {
145            core: Arc::downgrade(&self.core),
146            id: self.id,
147        })
148    }
149}
150impl<R: Send> Drop for SingleplexBuildPermit<R> {
151    fn drop(&mut self) {
152        if self.pending {
153            self.core.abort_build(self.id);
154        }
155    }
156}
157
158/// A weak registration handle for health changes and retirement requests.
159/// Holding it does not keep the pool alive or give access to a leased resource.
160pub struct SingleplexResource<R: Send> {
161    core: Weak<Core<SingleSlot<R>>>,
162    id: ResourceId,
163}
164impl<R: Send> Clone for SingleplexResource<R> {
165    fn clone(&self) -> Self {
166        Self {
167            core: self.core.clone(),
168            id: self.id,
169        }
170    }
171}
172impl<R: Send> SingleplexResource<R> {
173    pub fn id(&self) -> ResourceId {
174        self.id
175    }
176    /// Stop admission immediately; an outstanding lease retains cleanup ownership
177    /// until it returns. Returns false for an already-draining/removed resource.
178    pub fn retire(&self) -> bool {
179        self.core.upgrade().is_some_and(|core| core.retire(self.id))
180    }
181}
182
183/// Exclusive non-cloneable resource lease. Release only after proving that the
184/// whole operation is synchronized and reusable, including both HTTP directions.
185pub struct SingleplexLease<R: Send> {
186    core: Arc<Core<SingleSlot<R>>>,
187    id: ResourceId,
188    resource: Option<R>,
189}
190impl<R: Send> SingleplexLease<R> {
191    pub fn id(&self) -> ResourceId {
192        self.id
193    }
194    pub fn release(mut self) {
195        self.finish(false);
196    }
197    pub fn retire(mut self) {
198        self.finish(true);
199    }
200    fn finish(&mut self, retire: bool) {
201        let Some(resource) = self.resource.take() else {
202            return;
203        };
204        let retired = {
205            let mut state = self.core.lock();
206            let slot = state.slots.get_mut(&self.id).unwrap();
207            if retire {
208                slot.drain();
209            }
210            debug_assert!(slot.resource.is_none());
211            slot.resource = Some(resource);
212            self.core.remove_if_drained(&mut state, self.id)
213        };
214        self.core.enqueue(self.id, retired);
215    }
216}
217impl<R: Send> Deref for SingleplexLease<R> {
218    type Target = R;
219    fn deref(&self) -> &R {
220        self.resource.as_ref().unwrap()
221    }
222}
223impl<R: Send> DerefMut for SingleplexLease<R> {
224    fn deref_mut(&mut self) -> &mut R {
225        self.resource.as_mut().unwrap()
226    }
227}
228impl<R: Send> Drop for SingleplexLease<R> {
229    fn drop(&mut self) {
230        self.finish(true);
231    }
232}
base_pool/singleplex.rs

base_pool/
singleplex.rs
