simu/resource/priority.rs
1// SPDX-FileCopyrightText: Copyright (c) Siemens 2026 contributed by Christoph Kuhmuench christoph.kuhmuench@gmail.com
2//
3// SPDX-License-Identifier: MIT OR Apache-2.0
4
5//! Priority-scheduled resource pool.
6//!
7//! [`PriorityResource`] serves blocked waiters by priority level — lower
8//! number first, FIFO within a level. For a pool whose *holders* can be
9//! evicted by more urgent requests, see
10//! [`PreemptiveResource`](crate::PreemptiveResource).
11
12use std::cell::{Cell, RefCell};
13use std::future::Future;
14use std::pin::Pin;
15use std::rc::Rc;
16use std::task::{Context, Poll};
17
18use super::wait_queue::WaitQueue;
19
20// ---------------------------------------------------------------------------
21// Public API
22// ---------------------------------------------------------------------------
23
24/// A cloneable handle to a capacity-limited resource pool with priority
25/// scheduling.
26///
27/// Like [`Resource`](crate::Resource), units are acquired by calling
28/// [`request`](PriorityResource::request) and awaiting the returned future.
29/// Unlike `Resource`, waiters are served in **priority order**: the waiter with
30/// the lowest priority number is served first. Within the same priority level,
31/// waiters are served FIFO.
32///
33/// **Priority convention:** lower number = higher priority (`0` is highest).
34///
35/// `PriorityResource` wraps an `Rc<RefCell<>>` internally, so cloning is cheap
36/// and all clones share the same pool. It is `!Send + !Sync` — consistent with
37/// `SimEnv`.
38///
39/// An urgent case overtakes a routine one that queued first:
40///
41/// ```
42/// use std::cell::RefCell;
43/// use std::rc::Rc;
44/// use simu::{SimEnv, PriorityResource};
45///
46/// let mut env = SimEnv::with_seed(0);
47/// let doctor = PriorityResource::new(1);
48/// let seen = Rc::new(RefCell::new(Vec::new()));
49///
50/// // Occupy the doctor until t = 1.
51/// let h = env.handle();
52/// let d = doctor.clone();
53/// env.spawn(async move {
54/// let _g = d.request(5).await;
55/// h.timeout(1.0).await;
56/// });
57///
58/// // Two patients queue while the doctor is busy — routine (10) arrives
59/// // before urgent (0), but urgent is served first.
60/// for (name, priority) in [("routine", 10), ("urgent", 0)] {
61/// let d = doctor.clone();
62/// let seen = Rc::clone(&seen);
63/// env.spawn(async move {
64/// let _g = d.request(priority).await;
65/// seen.borrow_mut().push(name);
66/// });
67/// }
68///
69/// env.run();
70/// assert_eq!(*seen.borrow(), ["urgent", "routine"]); // lower number wins
71/// ```
72///
73/// Internally this is a `WaitQueue<u32>` (a `pub(crate)` helper): the
74/// `u32` priority is the ordering key, and the queue's internal sequence
75/// counter provides FIFO tie-breaking within a level.
76#[derive(Clone)]
77pub struct PriorityResource {
78 state: Rc<RefCell<WaitQueue<u32>>>,
79}
80
81impl std::fmt::Debug for PriorityResource {
82 fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
83 let mut d = f.debug_struct("PriorityResource");
84 if let Ok(q) = self.state.try_borrow() {
85 d.field("in_use", &q.in_use())
86 .field("capacity", &q.capacity())
87 .field("queue_len", &q.live_waiters());
88 }
89 d.finish_non_exhaustive()
90 }
91}
92
93impl PriorityResource {
94 /// Create a new priority resource pool with the given capacity.
95 ///
96 /// # Panics
97 ///
98 /// Panics if `capacity` is zero.
99 #[must_use]
100 pub fn new(capacity: usize) -> Self {
101 assert!(capacity > 0, "PriorityResource capacity must be at least 1");
102 PriorityResource {
103 state: Rc::new(RefCell::new(WaitQueue::new(capacity))),
104 }
105 }
106
107 /// Request one unit at the given `priority` (lower = higher priority).
108 ///
109 /// Resolves immediately if a unit is available; otherwise suspends the
110 /// calling process and wakes it before any lower-priority waiter when a
111 /// unit becomes free.
112 ///
113 /// The returned [`PriorityResourceGuard`] releases the unit when dropped.
114 #[must_use = "futures do nothing unless awaited"]
115 pub fn request(&self, priority: u32) -> PriorityResourceRequest {
116 PriorityResourceRequest {
117 state: Rc::clone(&self.state),
118 priority,
119 registered: false,
120 consumed: false,
121 canceled: Rc::new(Cell::new(false)),
122 granted: Rc::new(Cell::new(false)),
123 }
124 }
125
126 /// Number of units currently in use.
127 #[must_use]
128 pub fn in_use(&self) -> usize {
129 self.state.borrow().in_use()
130 }
131
132 /// Total capacity of this resource pool.
133 #[must_use]
134 pub fn capacity(&self) -> usize {
135 self.state.borrow().capacity()
136 }
137
138 /// Number of processes currently queued waiting for a unit, across all
139 /// priority levels. Excludes abandoned (canceled) requests.
140 #[must_use]
141 pub fn queue_len(&self) -> usize {
142 self.state.borrow().live_waiters()
143 }
144}
145
146/// Future returned by [`PriorityResource::request`].
147///
148/// Resolves to a [`PriorityResourceGuard`] once a unit is acquired.
149pub struct PriorityResourceRequest {
150 state: Rc<RefCell<WaitQueue<u32>>>,
151 priority: u32,
152 /// Prevents double-queuing on repeated polls (same pattern as `ResourceRequest`).
153 registered: bool,
154 /// Set once a granted/acquired unit has become a guard; `Drop` then must
155 /// not release (the guard owns the unit).
156 consumed: bool,
157 /// Shared with the queue entry; set to `true` on drop if the request was
158 /// registered but never granted.
159 canceled: Rc<Cell<bool>>,
160 /// Shared with the queue entry; set to `true` by `WaitQueue::release` when
161 /// the unit is handed directly to this request. Checked first in `poll`.
162 granted: Rc<Cell<bool>>,
163}
164
165impl std::fmt::Debug for PriorityResourceRequest {
166 fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
167 f.debug_struct("PriorityResourceRequest")
168 .field("priority", &self.priority)
169 .field("registered", &self.registered)
170 .field("granted", &self.granted.get())
171 .finish_non_exhaustive()
172 }
173}
174
175impl Future for PriorityResourceRequest {
176 type Output = PriorityResourceGuard;
177
178 fn poll(mut self: Pin<&mut Self>, cx: &mut Context<'_>) -> Poll<PriorityResourceGuard> {
179 // Direct handoff: a released unit was transferred to us (see `Resource`).
180 if self.granted.get() {
181 self.consumed = true;
182 return Poll::Ready(PriorityResourceGuard {
183 state: Rc::clone(&self.state),
184 });
185 }
186 // Drop the borrow before writing self.registered to satisfy the borrow checker.
187 let acquired = {
188 let mut state = self.state.borrow_mut();
189 if state.try_acquire() {
190 true
191 } else {
192 if !self.registered {
193 state.register(
194 self.priority,
195 cx.waker().clone(),
196 Rc::clone(&self.canceled),
197 Rc::clone(&self.granted),
198 );
199 }
200 false
201 }
202 };
203 if acquired {
204 self.consumed = true;
205 return Poll::Ready(PriorityResourceGuard {
206 state: Rc::clone(&self.state),
207 });
208 }
209 self.registered = true;
210 Poll::Pending
211 }
212}
213
214impl Drop for PriorityResourceRequest {
215 fn drop(&mut self) {
216 if self.consumed {
217 return; // the guard owns the unit and will release it
218 }
219 if self.granted.get() {
220 // Handed a unit but never consumed it — pass it on (see `Resource`).
221 self.state.borrow_mut().release();
222 } else if self.registered {
223 self.canceled.set(true);
224 }
225 }
226}
227
228/// RAII guard that holds one unit of a [`PriorityResource`].
229///
230/// The unit is released automatically when this value is dropped, waking the
231/// highest-priority suspended requester (if any).
232pub struct PriorityResourceGuard {
233 state: Rc<RefCell<WaitQueue<u32>>>,
234}
235
236impl std::fmt::Debug for PriorityResourceGuard {
237 fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
238 f.debug_struct("PriorityResourceGuard")
239 .finish_non_exhaustive()
240 }
241}
242
243impl Drop for PriorityResourceGuard {
244 fn drop(&mut self) {
245 // Release one unit and wake the highest-priority live waiter; the
246 // WaitQueue skips canceled (abandoned) waiters automatically.
247 self.state.borrow_mut().release();
248 }
249}