cocoon_tpm_utils_async/asynchronous/semaphore.rs
1// SPDX-License-Identifier: Apache-2.0
2// Copyright 2023-2025 SUSE LLC
3// Author: Nicolai Stange <nstange@suse.de>
4
5//! Implementation of [`AsyncSemaphore`]
6
7extern crate alloc;
8use crate::{
9 alloc::{SyncVec, SyncVecError},
10 sync_types::{self, Lock as _, WeakSyncRcPtr as _},
11};
12use core::{cell, convert, future, marker, num, ops, pin, task};
13use core::{ptr, sync::atomic};
14
15/// Error returned by [`AsyncSemaphore::new()`] and
16/// [`AsyncSemaphoreExclusiveAllGuard::resize_semaphore()`].
17#[derive(Clone, Copy, PartialEq, Eq, Debug)]
18pub enum AsyncSemaphoreNewError {
19 /// The number of the configured maximum leases is not **strictly** less
20 /// than [`usize::MAX`].
21 UnsupportedCapacity,
22}
23
24/// Error information returned by the [`AsyncSemaphore`] API.
25#[derive(Clone, Copy, Debug)]
26pub enum AsyncSemaphoreError {
27 /// The requested number of leases exceeds the inquired
28 /// [`AsyncSemaphore`]'s total capacity.
29 RequestExceedsSemaphoreCapacity,
30
31 /// A pending operation's associated [`AsyncSemaphore`] has been
32 /// dropped.
33 StaleSemaphore,
34
35 /// Memory allocation failure.
36 MemoryAllocationFailure,
37
38 /// Internal error.
39 Internal,
40}
41
42/// Internal representation of the number/type of leases requested respectively
43/// granted from an [`AsyncSemaphore`].
44#[derive(Clone, Copy, PartialEq, Eq, Debug)]
45enum AsyncSemaphoreLeaseGrantCount {
46 /// "Trivial" grant for zero leases, mutually exclusive with
47 /// [`ExclusiveAll`](Self::ExclusiveAll) grants.
48 TrivialLease,
49 /// A specific, non-zero number of leases.
50 Leases { count: num::NonZeroUsize },
51 /// All of an [`AsyncSemaphore`]'s capacity. Mutually exclusive with any
52 /// other grant of any type, including itself.
53 ExclusiveAll,
54}
55
56impl convert::From<usize> for AsyncSemaphoreLeaseGrantCount {
57 fn from(value: usize) -> Self {
58 if value == 0 {
59 AsyncSemaphoreLeaseGrantCount::TrivialLease
60 } else {
61 AsyncSemaphoreLeaseGrantCount::Leases {
62 count: num::NonZeroUsize::new(value).unwrap(),
63 }
64 }
65 }
66}
67
68/// Internal representation of a waiter enqueued to [`AsyncSemaphoreQueue`].
69struct AsyncSemaphoreQueueEntry {
70 /// The number of leases to acquire.
71 leases_requested: AsyncSemaphoreLeaseGrantCount,
72 /// The waker to invoke once the requested number of leases becomes
73 /// available.
74 waker: Option<task::Waker>,
75 /// The waiter's assigned id.
76 waiter_id: num::NonZeroU64,
77}
78
79/// Internal queue of waiters waiting for leases grants on an
80/// [`AsyncSemaphore`].
81struct AsyncSemaphoreQueue {
82 /// The actual wait queue.
83 queue: SyncVec<AsyncSemaphoreQueueEntry>,
84
85 /// Maximum number of leases that can be granted at a time, may be zero to
86 /// permit only [trivial](AsyncSemaphoreLeaseGrantCount::TrivialLease) (and
87 /// [exclusive](AsyncSemaphoreLeaseGrantCount::ExclusiveAll)) leases.
88 max_leases: usize,
89
90 /// Last waiter id allocated in the course of enqueueing.
91 last_waiter_id: u64,
92
93 /// Number of currently enqueued
94 /// [`TrivialLease`](AsyncSemaphoreLeaseGrantCount::TrivialLease) waiters.
95 enqueued_trivial_lease_waiters: usize,
96 /// Number of currently enqueued
97 /// [`Exclusive`](AsyncSemaphoreLeaseGrantCount::ExclusiveAll) waiters.
98 enqueued_exclusive_all_waiters: usize,
99}
100
101impl AsyncSemaphoreQueue {
102 /// Create a new [`AsyncSemaphoreQueue`] instance.
103 ///
104 /// # Arguments:
105 ///
106 /// * `max_leases` - Semaphore capacity. May be zero to permit only
107 /// [trivial](AsyncSemaphoreLeaseGrantCount::TrivialLease) (and
108 /// [exclusive](AsyncSemaphoreLeaseGrantCount::ExclusiveAll)) leases.
109 fn new(max_leases: usize) -> Self {
110 Self {
111 queue: SyncVec::new(),
112 max_leases,
113 last_waiter_id: 0,
114 enqueued_trivial_lease_waiters: 0,
115 enqueued_exclusive_all_waiters: 0,
116 }
117 }
118
119 /// Match a waiter id to an index in the queue.
120 fn waiter_position(&self, waiter_id: num::NonZeroU64) -> Option<usize> {
121 self.queue.iter().position(|entry| entry.waiter_id == waiter_id)
122 }
123
124 /// Poll the semaphore on behalf of a waiter.
125 ///
126 /// Return `true` if the requested leases have been granted to the waiter,
127 /// `false` otherwise.
128 ///
129 /// In general, uncompleted waiters will remain on the queue, completes ones
130 /// get removed.
131 ///
132 /// # Arguments:
133 ///
134 /// * `waiter_id` - The waiter id, as previously returned by
135 /// [`enqueue_waiter()`](AsyncSemaphoreQueueEntryAllocation::enqueue_waiter)
136 /// of the enqueued waiter instance to check for completion.
137 /// * `waker` - The waker to register at the enqueued waiter entry in case
138 /// the waiter is not ready yet.
139 fn poll_waiter(&mut self, waiter_id: num::NonZeroU64, waker: task::Waker) -> bool {
140 if let Some(index) = self.waiter_position(waiter_id) {
141 self.queue[index].waker = Some(waker);
142 false
143 } else {
144 true
145 }
146 }
147
148 /// Remove a waiter from the queue and update the queue's internal
149 /// bookkeeping, but take no action beyond that.
150 ///
151 /// # Arguments:
152 ///
153 /// * `index` - Index within the queue of the waiter entry to remove.
154 fn remove_entry(&mut self, index: usize) {
155 match self.queue[index].leases_requested {
156 AsyncSemaphoreLeaseGrantCount::TrivialLease => {
157 debug_assert!(self.enqueued_trivial_lease_waiters != 0);
158 self.enqueued_trivial_lease_waiters -= 1;
159 }
160 AsyncSemaphoreLeaseGrantCount::ExclusiveAll => {
161 debug_assert!(self.enqueued_exclusive_all_waiters != 0);
162 self.enqueued_exclusive_all_waiters -= 1;
163 }
164 AsyncSemaphoreLeaseGrantCount::Leases { .. } => (),
165 }
166 self.queue.remove(index);
167 }
168
169 /// Remove an enqueued waiter from the queue by `waiter_id` and update the
170 /// queues internal bookkeeping, but take no action beyond that.
171 ///
172 /// # Arguments:
173 ///
174 /// * `waiter_id` - The waiter id as previously returned by
175 /// [`enqueue_waiter()`](AsyncSemaphoreQueueEntryAllocation::enqueue_waiter)
176 /// of the waiter entry to remove from the queue.
177 fn remove_waiter(&mut self, waiter_id: num::NonZeroU64) {
178 let index = self.waiter_position(waiter_id).unwrap();
179 self.remove_entry(index);
180 }
181
182 /// Prepare for enqueueing a waiter for the semaphore.
183 ///
184 /// Attempt to allocate a queue entry and return the
185 /// [`AsyncSemaphoreQueueEntryAllocation`] on success. The
186 /// [`AsyncSemaphoreQueueEntryAllocation::enqueue_waiter()`] may then
187 /// subsequently get invoked for actually enqueueing a waiter.
188 ///
189 /// `enqueue_waiter_prepare()` may drop and reacquire the supplied
190 /// `this_guard` locking guard.
191 ///
192 /// # Arguments:
193 ///
194 /// * `queue_lock_for_inner_queue` - A
195 /// [`LockForInner`](sync_types::LockForInner) for
196 /// [`AsyncSemaphoreQueue::queue`] derived from a
197 /// [`Lock`](sync_types::Lock) for `Self`.
198 /// * `this_guard` - A [`Lock::Guard`](sync_types::Lock::Guard) on a
199 /// [`Lock`](sync_types::Lock) for `Self`.
200 fn enqueue_waiter_prepare<'a, QL: sync_types::Lock<Self>>(
201 queue_lock_for_inner_queue: &'a sync_types::LockForInner<'a, Self, QL, AsyncSemaphoreQueueDerefInnerQueueTag>,
202 this_guard: QL::Guard<'a>,
203 ) -> Result<AsyncSemaphoreQueueEntryAllocation<'a, QL>, (QL::Guard<'a>, AsyncSemaphoreError)> {
204 let queue_guard_for_inner_queue =
205 sync_types::LockForInnerGuard::<'_, _, _, AsyncSemaphoreQueueDerefInnerQueueTag>::from_outer(this_guard);
206 match SyncVec::try_reserve(queue_lock_for_inner_queue, queue_guard_for_inner_queue, 1) {
207 (queue_guard_for_inner_queue, Ok(())) => Ok(AsyncSemaphoreQueueEntryAllocation {
208 queue_guard: queue_guard_for_inner_queue.into_outer(),
209 }),
210 (queue_guard_for_inner_queue, Err(e)) => Err((
211 queue_guard_for_inner_queue.into_outer(),
212 match e {
213 SyncVecError::MemoryAllocationFailure => AsyncSemaphoreError::MemoryAllocationFailure,
214 },
215 )),
216 }
217 }
218
219 /// Check whether the queue has any uncancelled waiters enqueued.
220 fn has_uncancelled_waiters(&self) -> bool {
221 self.queue.iter().any(|entry| match entry.leases_requested {
222 AsyncSemaphoreLeaseGrantCount::TrivialLease => true,
223 AsyncSemaphoreLeaseGrantCount::Leases { count } => count.get() <= self.max_leases,
224 AsyncSemaphoreLeaseGrantCount::ExclusiveAll => true,
225 })
226 }
227
228 /// Check whether the queue has any
229 /// [`TrivialLease`](AsyncSemaphoreLeaseGrantCount::TrivialLease) waiters
230 /// enqueued.
231 fn has_trivial_lease_waiters(&self) -> bool {
232 let has_trivial_lease_waiters = self.enqueued_trivial_lease_waiters != 0;
233 debug_assert_eq!(
234 has_trivial_lease_waiters,
235 self.queue
236 .iter()
237 .any(|entry| { matches!(entry.leases_requested, AsyncSemaphoreLeaseGrantCount::TrivialLease) })
238 );
239 has_trivial_lease_waiters
240 }
241
242 /// Check whether the queue has any
243 /// [`ExclusiveAll`](AsyncSemaphoreLeaseGrantCount::ExclusiveAll) waiters
244 /// enqueued.
245 fn has_exclusive_all_waiters(&self) -> bool {
246 let has_exclusive_all_waiters = self.enqueued_exclusive_all_waiters != 0;
247 debug_assert_eq!(
248 has_exclusive_all_waiters,
249 self.queue
250 .iter()
251 .any(|entry| { matches!(entry.leases_requested, AsyncSemaphoreLeaseGrantCount::ExclusiveAll) })
252 );
253 has_exclusive_all_waiters
254 }
255}
256
257/// Queue entry reservation returned by
258/// [`AsyncSemaphoreQueue::enqueue_waiter_prepare()`].
259struct AsyncSemaphoreQueueEntryAllocation<'a, QL: sync_types::Lock<AsyncSemaphoreQueue>>
260where
261 QL: 'a,
262{
263 queue_guard: QL::Guard<'a>,
264}
265
266impl<'a, QL: sync_types::Lock<AsyncSemaphoreQueue>> AsyncSemaphoreQueueEntryAllocation<'a, QL>
267where
268 QL: 'a,
269{
270 /// Enqueue a waiter for the semaphore, consuming the queue entry allocation
271 /// associated with `self`.
272 ///
273 /// An id uniquely identifying the wait entry gets returned, alongside a
274 /// guard for the lock initially passed to
275 /// [`AsyncSemaphoreQueue::enqueue_waiter_prepare()`]. Note that this guard
276 /// might have been dropped and reacquired in the course of
277 /// [`AsyncSemaphoreQueue::enqueue_waiter_prepare()`].
278 ///
279 /// # Arguments:
280 ///
281 /// * `leases_requested` - The request number/type of leases to record at
282 /// the newly created waiter entry.
283 fn enqueue_waiter(self, leases_requested: AsyncSemaphoreLeaseGrantCount) -> (QL::Guard<'a>, num::NonZeroU64) {
284 let Self { mut queue_guard } = self;
285 queue_guard.last_waiter_id += 1;
286 let waiter_id = num::NonZeroU64::new(queue_guard.last_waiter_id).unwrap();
287 queue_guard.queue.push(AsyncSemaphoreQueueEntry {
288 leases_requested,
289 waker: None,
290 waiter_id,
291 });
292 match leases_requested {
293 AsyncSemaphoreLeaseGrantCount::TrivialLease => queue_guard.enqueued_trivial_lease_waiters += 1,
294 AsyncSemaphoreLeaseGrantCount::Leases { count: _ } => (),
295 AsyncSemaphoreLeaseGrantCount::ExclusiveAll => queue_guard.enqueued_exclusive_all_waiters += 1,
296 };
297 (queue_guard, waiter_id)
298 }
299}
300
301/// [`DerefInnerByTag`](sync_types::DerefInnerByTag) `TAG` for derefencing
302/// `AsyncSemaphoreQueue::queue`.
303///
304/// Enables presenting the [`Lock`](sync_types::Lock) of the
305/// [`AsyncSemaphoreState::queue`] member as one for the inner, contained
306/// [`AsyncSemaphoreQueue::queue`] via the
307/// [`LockForInner`](sync_types::LockForInner) mechanism.
308struct AsyncSemaphoreQueueDerefInnerQueueTag;
309
310impl sync_types::DerefInnerByTag<AsyncSemaphoreQueueDerefInnerQueueTag> for AsyncSemaphoreQueue {
311 crate::impl_deref_inner_by_tag!(queue, SyncVec<AsyncSemaphoreQueueEntry>);
312}
313
314impl sync_types::DerefMutInnerByTag<AsyncSemaphoreQueueDerefInnerQueueTag> for AsyncSemaphoreQueue {
315 crate::impl_deref_mut_inner_by_tag!(queue);
316}
317
318/// An [`AsyncSemaphore`]'s internal state.
319///
320/// The semaphore state is maintained in a separate `struct` because it,
321/// and the associated maintenance logic in particular, is independent of
322/// of the user-specified generic type of the data wrapped in an
323/// [`AsyncSemaphore`].
324struct AsyncSemaphoreState<ST: sync_types::SyncTypes> {
325 queue: ST::Lock<AsyncSemaphoreQueue>,
326 /// Number of active,
327 /// non-[trivial](AsyncSemaphoreLeaseGrantCount::TrivialLease) lease grants
328 /// handed out. They must eventually get returned
329 /// via [`return_grant()`](Self::return_grant).
330 ///
331 /// For an [`ExclusiveAll`](AsyncSemaphoreLeaseGrantCount::ExclusiveAll)
332 /// grant, this will be one more than the total
333 /// [`max_leases`](Self::max_leases) capacity. This ensures that there
334 /// can only ever be one exlusive owner around at a time, even with a
335 /// capacity of zero.
336 ///
337 /// The zero-sized
338 /// [`TrivialLease`](AsyncSemaphoreLeaseGrantCount::TrivialLease) grants
339 /// are tracked separately in
340 /// [`trivial_leases_granted`](Self::trivial_leases_granted).
341 ///
342 /// Note that `leases_granted` itself is updated only under the
343 /// [`Self::queue`] lock, but read from without that lock being held in
344 /// [`return_trivial_lease_grant()`](Self::return_trivial_lease_grant).
345 leases_granted: atomic::AtomicUsize,
346
347 /// Number of active, zero-sized
348 /// [`TrivialLease`](AsyncSemaphoreLeaseGrantCount::TrivialLease) grants
349 /// handed out.
350 trivial_leases_granted: atomic::AtomicUsize,
351}
352
353impl<ST: sync_types::SyncTypes> AsyncSemaphoreState<ST> {
354 /// Constant offset added to
355 /// [`trivial_leases_granted`](Self::trivial_leases_granted) whenever no
356 /// [`ExclusiveAll`](AsyncSemaphoreLeaseGrantCount::ExclusiveAll) waiter
357 /// is enqueued on the [`queue`](Self::queue).
358 ///
359 /// This is an optimization to disable the wake-up path in
360 /// [`return_trivial_lease_grant()`](Self::return_trivial_lease_grant).
361 /// when not needed.
362 const TRIVIAL_LEASES_GRANTED_NO_EXCLUSIVE_ALL_WAITERS_OFFSET: usize = 1;
363
364 /// Instantiate a [`AsyncSemaphoreState`] for a [`AsyncSemaphore`] with
365 /// specified capacity.
366 ///
367 /// # Arguments:
368 ///
369 /// * `max_leases` - The containg [`AsyncSemaphore`]'s desired capacity.
370 /// Must be **strictly** less than [`usize::MAX`]. May be zero to permit
371 /// only "trivial" (and "exlusive-all") leases.
372 fn new(max_leases: usize) -> Self {
373 debug_assert!(max_leases < usize::MAX);
374 Self {
375 queue: ST::Lock::from(AsyncSemaphoreQueue::new(max_leases)),
376 leases_granted: atomic::AtomicUsize::new(0),
377 trivial_leases_granted: atomic::AtomicUsize::new(
378 Self::TRIVIAL_LEASES_GRANTED_NO_EXCLUSIVE_ALL_WAITERS_OFFSET,
379 ),
380 }
381 }
382
383 /// Maximum number of total semaphore leases which can be active at a time.
384 fn max_leases(&self) -> usize {
385 self.queue.lock().max_leases
386 }
387
388 /// Prepare to asynchronously acquire non-exclusive semaphore leases.
389 ///
390 /// * If the number of requested leases cannot be granted right away, a
391 /// waiter entry will be enqueued and the associated waiter id returned.
392 /// The waiter **must** eventually get removed again either
393 /// - implicitly by getting completed asynchronously through
394 /// [`wake_completed_waiters()`](Self::wake_completed_waiters),
395 /// - explicitly through [`cancel_waiter()`](Self::cancel_waiter) or, upon
396 /// getting asynchronously [failed](Self::wake_failed_waiters), via
397 /// [`remove_waiter`](AsyncSemaphoreQueue::remove_waiter).
398 ///
399 /// It is imperative to wrap the returned waiter id in an RAAIsh type
400 /// as soon as possible without intermediate potential failure points.
401 /// * If the number of leases can get granted right away, `None` will get
402 /// returned. The granted leases must eventually get returned via
403 /// [`return_grant()`](Self::return_grant), or, as an optimization, via
404 /// [`return_trivial_lease_grant()`](Self::return_trivial_lease_grant) if
405 /// applicable.
406 ///
407 /// # Arguments:
408 ///
409 /// * `leases_requested` - The number of leases to obtain. May be zero for
410 /// requesting a [trivial](AsyncSemaphoreLeaseGrantCount::TrivialLease)
411 /// lease.
412 ///
413 /// # Errors:
414 ///
415 /// * [`AsyncSemaphoreError::RequestExceedsSemaphoreCapacity`] - The number
416 /// of leases requested exceeds the [`AsyncSemaphore`]'s total capacity.
417 /// * [`AsyncSemaphoreError::MemoryAllocationFailure`] - Memory allocation
418 /// failure.
419 fn prepare_acquire_leases(&self, leases_requested: usize) -> Result<Option<num::NonZeroU64>, AsyncSemaphoreError> {
420 let queue_guard = self.queue.lock();
421 if self._try_acquire_leases(&queue_guard, leases_requested)? {
422 Ok(None)
423 } else {
424 let queue_lock_for_inner_queue =
425 sync_types::LockForInner::<'_, _, _, AsyncSemaphoreQueueDerefInnerQueueTag>::from_outer(&self.queue);
426
427 match AsyncSemaphoreQueue::enqueue_waiter_prepare(&queue_lock_for_inner_queue, queue_guard) {
428 Ok(entry_allocation) => {
429 // The enqueue_waiter_prepare() might have dropped and reacquired the lock and
430 // some wakeups might have been missed. Recheck.
431 if self._try_acquire_leases(&entry_allocation.queue_guard, leases_requested)? {
432 Ok(None)
433 } else {
434 let (_, waiter_id) =
435 entry_allocation.enqueue_waiter(AsyncSemaphoreLeaseGrantCount::from(leases_requested));
436 Ok(Some(waiter_id))
437 }
438 }
439 Err((_, e)) => Err(e),
440 }
441 }
442 }
443
444 /// Prepare to asynchronously acquire an
445 /// [`ExclusiveAll`](AsyncSemaphoreLeaseGrantCount::ExclusiveAll)
446 /// grant on all of a semaphore's capacity.
447 ///
448 /// * If the [`ExclusiveAll`](AsyncSemaphoreLeaseGrantCount::ExclusiveAll)
449 /// grant cannot get issued right away, a waiter entry will be enqueued
450 /// and the associated waiter id returned. The waiter **must** eventually
451 /// get removed again either
452 /// - implicitly by getting completed asynchronously through
453 /// [`wake_completed_waiters()`](Self::wake_completed_waiters),
454 /// - explicitly through [`cancel_waiter()`](Self::cancel_waiter).
455 ///
456 /// It is imperative to wrap the returned waiter id in an RAAIsh type
457 /// as soon as possible without intermediate potential failure points.
458 /// * If the [`ExclusiveAll`](AsyncSemaphoreLeaseGrantCount::ExclusiveAll)
459 /// grant can get issued right away, `None` will get returned. The grant
460 /// must eventually get returned via
461 /// [`return_grant()`](Self::return_grant).
462 ///
463 /// # Errors:
464 ///
465 /// * [`AsyncSemaphoreError::MemoryAllocationFailure`] - Memory allocation
466 /// failure.
467 fn prepare_acquire_exclusive_all(&self) -> Result<Option<num::NonZeroU64>, AsyncSemaphoreError> {
468 let queue_guard = self.queue.lock();
469 self.acquire_exclusive_all_begin(&queue_guard);
470 if self._try_acquire_exclusive_all(&queue_guard) {
471 self.acquire_exclusive_all_end(&queue_guard);
472 Ok(None)
473 } else {
474 // The enqueue_waiter_prepare() might drop and reacquire the lock.
475 // enqueued_exclusive_all_waiters has not been bumped yet, because that's
476 // possible below only once there's an actual ExlusiveAll waiter
477 // enqueued. Invoke acquire_exclusive_all_end() before dropping the
478 // lock so that a acquire_exclusive_all_begin() issued concurrently
479 // while the lock is dropped wouldn't subtract the
480 // TRIVIAL_LEASES_GRANTED_NO_EXCLUSIVE_ALL_WAITERS_OFFSET once again.
481 self.acquire_exclusive_all_end(&queue_guard);
482 let queue_lock_for_inner_queue =
483 sync_types::LockForInner::<'_, _, _, AsyncSemaphoreQueueDerefInnerQueueTag>::from_outer(&self.queue);
484 match AsyncSemaphoreQueue::enqueue_waiter_prepare(&queue_lock_for_inner_queue, queue_guard) {
485 Ok(entry_allocation) => {
486 self.acquire_exclusive_all_begin(&entry_allocation.queue_guard);
487 // Recheck whether the grant can get acquired now -- we might have missed some
488 // wakeups while the lock was dropped.
489 if self._try_acquire_exclusive_all(&entry_allocation.queue_guard) {
490 self.acquire_exclusive_all_end(&entry_allocation.queue_guard);
491 Ok(None)
492 } else {
493 let (_, waiter_id) =
494 entry_allocation.enqueue_waiter(AsyncSemaphoreLeaseGrantCount::ExclusiveAll);
495 Ok(Some(waiter_id))
496 }
497 }
498 Err((_, e)) => Err(e),
499 }
500 }
501 }
502
503 /// Try to synchronously acquire a non-exclusive leases grant.
504 ///
505 /// Simple wrapper around
506 /// [`_try_acquire_leases()`](Self::_try_acquire_leases) establishing the
507 /// [`queue`](Self::queue) lock.
508 fn try_acquire_leases(&self, leases_requested: usize) -> Result<bool, AsyncSemaphoreError> {
509 let locked_queue = self.queue.lock();
510 self._try_acquire_leases(&locked_queue, leases_requested)
511 }
512
513 /// Try to synchronously acquire an
514 /// [`ExclusiveAll`](AsyncSemaphoreLeaseGrantCount::ExclusiveAll) grant on
515 /// all of a semaphore's capacity with the [`queue`](Self::queue) lock
516 /// held.
517 ///
518 /// Simple wrapper around
519 /// [`_try_acquire_exclusive_all()`](Self::_try_acquire_exclusive_all)
520 /// establishing the [`queue`](Self::queue) lock and issuing a
521 /// [`acquire_exclusive_all_begin()`](Self::acquire_exclusive_all_begin)/
522 /// [`acquire_exclusive_all_end()`](Self::acquire_exclusive_all_end) pair
523 /// around the call.
524 fn try_acquire_exclusive_all(&self) -> bool {
525 let locked_queue = self.queue.lock();
526 self.acquire_exclusive_all_begin(&locked_queue);
527 let granted = self._try_acquire_exclusive_all(&locked_queue);
528 self.acquire_exclusive_all_end(&locked_queue);
529 granted
530 }
531
532 /// Try to synchronously obtain a
533 /// [`TrivialLease`](AsyncSemaphoreLeaseGrantCount::TrivialLease),
534 /// if available, with the [`queue`](Self::queue) lock held.
535 ///
536 /// The operation will only succeed and return `true` if there is no
537 /// [`ExclusiveAll`](AsyncSemaphoreLeaseGrantCount::ExclusiveAll) waiter
538 /// ahead in line. Otherwise `false` will get returned.
539 ///
540 /// # Arguments:
541 ///
542 /// * `locked_queue` - Guard for the locked [`queue`](Self::queue).
543 fn _try_acquire_trivial_lease(
544 &self,
545 locked_queue: &<ST::Lock<AsyncSemaphoreQueue> as sync_types::Lock<AsyncSemaphoreQueue>>::Guard<'_>,
546 ) -> bool {
547 !locked_queue.has_exclusive_all_waiters()
548 && self.try_grant_one(locked_queue, &AsyncSemaphoreLeaseGrantCount::TrivialLease)
549 }
550
551 /// Try to synchronously acquire a leases grant with the
552 /// [`queue`](Self::queue) lock held.
553 ///
554 /// The operation will only succeed and return `true` if the requested
555 /// number of leases is available, no
556 /// [`ExclusiveAll`](AsyncSemaphoreLeaseGrantCount::ExclusiveAll) grant is
557 /// active, and there is no other waiter ahead in line. Otherwise
558 /// `false` will get returned.
559 ///
560 /// In the case of success, the granted leases must eventually get returned
561 /// via [`return_grant()`](Self::return_grant), or, as an optimization,
562 /// via
563 /// [`return_trivial_lease_grant()`](Self::return_trivial_lease_grant) if
564 /// applicable
565 ///
566 /// # Arguments:
567 ///
568 /// * `locked_queue` - Guard for the locked [`queue`](Self::queue).
569 /// * `leases_requested` - The number of leases to obtain. May be zero for
570 /// requesting a [trivial](AsyncSemaphoreLeaseGrantCount::TrivialLease)
571 /// lease.
572 ///
573 /// # Errors:
574 ///
575 /// * [`AsyncSemaphoreError::RequestExceedsSemaphoreCapacity`] - The number
576 /// of leases requested exceeds the [`AsyncSemaphore`]'s total capacity.
577 fn _try_acquire_leases(
578 &self,
579 locked_queue: &<ST::Lock<AsyncSemaphoreQueue> as sync_types::Lock<AsyncSemaphoreQueue>>::Guard<'_>,
580 leases_requested: usize,
581 ) -> Result<bool, AsyncSemaphoreError> {
582 if leases_requested == 0 {
583 Ok(self._try_acquire_trivial_lease(locked_queue))
584 } else if locked_queue.max_leases < leases_requested {
585 Err(AsyncSemaphoreError::RequestExceedsSemaphoreCapacity)
586 } else if !locked_queue.has_uncancelled_waiters()
587 && self.try_grant_one(locked_queue, &AsyncSemaphoreLeaseGrantCount::from(leases_requested))
588 {
589 Ok(true)
590 } else {
591 Ok(false)
592 }
593 }
594
595 /// Try to synchronously acquire an
596 /// [`ExclusiveAll`](AsyncSemaphoreLeaseGrantCount::ExclusiveAll) grant on
597 /// all of a semaphore's capacity with the [`queue`](Self::queue) lock
598 /// held.
599 ///
600 /// The call **must** be wrapped in an
601 /// [`acquire_exclusive_all_begin()`](Self::acquire_exclusive_all_begin)/
602 /// [`acquire_exclusive_all_end()`](Self::acquire_exclusive_all_end) pair or
603 /// it will fail unconditionally.
604 ///
605 /// The operation will only succeed and return `true` if no other grants of
606 /// any kind are active. Otherwise `false` will get returned.
607 ///
608 /// In the case of success, the grant must eventually get returned
609 /// via [`return_grant()`](Self::return_grant).
610 ///
611 /// # Arguments:
612 ///
613 /// * `locked_queue` - Guard for the locked [`queue`](Self::queue).
614 fn _try_acquire_exclusive_all(
615 &self,
616 locked_queue: &<ST::Lock<AsyncSemaphoreQueue> as sync_types::Lock<AsyncSemaphoreQueue>>::Guard<'_>,
617 ) -> bool {
618 !locked_queue.has_uncancelled_waiters()
619 && self.try_grant_one(locked_queue, &AsyncSemaphoreLeaseGrantCount::ExclusiveAll)
620 }
621
622 /// Try to issue a grant assuming the request is due next in
623 /// [`queue`](Self::queue) order.
624 ///
625 /// [`try_grant_one()`](Self::try_grant_one) exclusively considers currently
626 /// issued grants only, and is completely oblivious of any waiters
627 /// enqueued on the [`queue`](Self::queue). In particular, queue order
628 /// logic must get maintained externally. More specifically
629 /// * For [`TrivialLease`](AsyncSemaphoreLeaseGrantCount::TrivialLease)
630 /// requests, no
631 /// [`ExclusiveAll`](AsyncSemaphoreLeaseGrantCount::ExclusiveAll) waiters
632 /// may be enqueued ahead in line.
633 /// * For "reqular" [`Leases`](AsyncSemaphoreLeaseGrantCount::Leases)
634 /// requests, no
635 /// [`ExclusiveAll`](AsyncSemaphoreLeaseGrantCount::ExclusiveAll) or
636 /// (uncancelled) other [`Leases`](AsyncSemaphoreLeaseGrantCount::Leases)
637 /// waiters may be ahead in line.
638 /// * For [`ExclusiveAll`](AsyncSemaphoreLeaseGrantCount::ExclusiveAll)
639 /// requests, no other (uncancelled) waiters of any type may be ahead in
640 /// line.
641 ///
642 /// The operation will succeed with `true` if the requested grant can be
643 /// issued, otherwise `false` will get returned. In the former case, the
644 /// associated grant must eventually get returned via
645 /// [`return_grant()`](Self::return_grant).
646 ///
647 /// # Arguments:
648 ///
649 /// * `locked_queue` - Guard for the locked [`queue`](Self::queue).
650 /// * `leases_requested` - The grant to obtain.
651 fn try_grant_one(
652 &self,
653 locked_queue: &<ST::Lock<AsyncSemaphoreQueue> as sync_types::Lock<AsyncSemaphoreQueue>>::Guard<'_>,
654 leases_requested: &AsyncSemaphoreLeaseGrantCount,
655 ) -> bool {
656 match leases_requested {
657 AsyncSemaphoreLeaseGrantCount::TrivialLease => {
658 if !self.is_exclusive_all_granted(locked_queue) {
659 self.trivial_leases_granted.fetch_add(1, atomic::Ordering::Relaxed);
660 true
661 } else {
662 false
663 }
664 }
665 AsyncSemaphoreLeaseGrantCount::Leases { count } => {
666 let leases_granted = self.leases_granted.load(atomic::Ordering::Relaxed);
667 if leases_granted
668 .checked_add(count.get())
669 .map(|needed_max_leases| needed_max_leases <= locked_queue.max_leases)
670 .unwrap_or(false)
671 {
672 self.leases_granted
673 .store(leases_granted + count.get(), atomic::Ordering::Relaxed);
674 true
675 } else {
676 false
677 }
678 }
679 AsyncSemaphoreLeaseGrantCount::ExclusiveAll => {
680 // The Acquire pairs with Release in return_grant(), it ensures that
681 // 1.) any prior trivial_leases_grant split out from
682 // non-trivial ones (in another thread) are also observed here and
683 // 2.) that the zeroization of leases_granted, if any, happens-before
684 // the SeqCst fence below.
685 // (Both should be achieved by locked_queue held here and in
686 // return_grant() already, but don't rely on
687 // implementation details and make it explicit).
688 let leases_granted = self.leases_granted.load(atomic::Ordering::Acquire);
689 if leases_granted == 0 {
690 // Issue a SeqCst fence between the preceeding zeroization, if any, and the test
691 // of trivial_leases_granted. Pairs with the fence in
692 // return_trivial_lease_grant(), c.f. the comment there.
693 atomic::fence(atomic::Ordering::SeqCst);
694 let trivial_leases_granted = self.trivial_leases_granted.load(atomic::Ordering::Acquire);
695 if trivial_leases_granted == 0 {
696 self.leases_granted
697 .store(locked_queue.max_leases + 1, atomic::Ordering::Relaxed);
698 true
699 } else {
700 false
701 }
702 } else {
703 false
704 }
705 }
706 }
707 }
708
709 /// Preparatory setup for acquiring an
710 /// [`ExclusiveAll`](AsyncSemaphoreLeaseGrantCount::ExclusiveAll) grant,
711 /// either asynchronously or synchronously.
712 ///
713 /// Before attempting to acquire some
714 /// [`ExclusiveAll`](AsyncSemaphoreLeaseGrantCount::ExclusiveAll) grant, the
715 /// [`trivial_leases_granted`](Self::trivial_leases_granted) tracking
716 /// counter must get decremented by
717 /// the [`TRIVIAL_LEASES_GRANTED_NO_EXCLUSIVE_ALL_WAITERS_OFFSET`](Self::TRIVIAL_LEASES_GRANTED_NO_EXCLUSIVE_ALL_WAITERS_OFFSET)
718 /// in order to
719 /// - make the request satisifiable in the first place, as
720 /// [`ExclusiveAll`](AsyncSemaphoreLeaseGrantCount::ExclusiveAll) grants
721 /// will be issued only when
722 /// [`trivial_leases_granted`](Self::trivial_leases_granted) reaches zero
723 /// and
724 /// - to arm the wakeup path in
725 /// [`return_trivial_lease_grant()`](Self::return_trivial_lease_grant).
726 ///
727 /// The setup must eventually be undone by
728 /// [`acquire_exclusive_all_end()`](Self::acquire_exclusive_all_end).
729 ///
730 /// # Arguments:
731 ///
732 /// * `locked_queue` - Guard for the locked [`queue`](Self::queue).
733 fn acquire_exclusive_all_begin(
734 &self,
735 locked_queue: &<ST::Lock<AsyncSemaphoreQueue> as sync_types::Lock<AsyncSemaphoreQueue>>::Guard<'_>,
736 ) {
737 if !locked_queue.has_exclusive_all_waiters() {
738 // The first one, arm the wake-up path in return_trivial_lease_grant(), and
739 // also, enable the ExlusiveAll grant in the first place.
740 debug_assert_ne!(self.trivial_leases_granted.load(atomic::Ordering::Relaxed), 0);
741 self.trivial_leases_granted.fetch_sub(
742 Self::TRIVIAL_LEASES_GRANTED_NO_EXCLUSIVE_ALL_WAITERS_OFFSET,
743 atomic::Ordering::Relaxed,
744 );
745 }
746 }
747
748 /// Teardown the setup established by a previous
749 /// [`acquire_exclusive_all_begin()`](Self::acquire_exclusive_all_begin).
750 ///
751 /// If
752 /// - there had been no waiter enqueued for the associated request ever, the
753 /// `locked_queue` guard must not have been dropped at any point inbetween
754 /// the calls to
755 /// [`acquire_exclusive_all_begin()`](Self::acquire_exclusive_all_begin)
756 /// and [`acquire_exclusive_all_end()`](Self::acquire_exclusive_all_end),
757 /// - there had been a waiter enqueued for the associated request, it must
758 /// have been removed before the call to
759 /// [`acquire_exclusive_all_end()`](Self::acquire_exclusive_all_end) and
760 /// `locked_queue` must not have been dropped at any point in time between
761 /// the removal from the queue and the invocation.
762 ///
763 /// # Arguments:
764 ///
765 /// * `locked_queue` - Guard for the locked [`queue`](Self::queue).
766 fn acquire_exclusive_all_end(
767 &self,
768 locked_queue: &<ST::Lock<AsyncSemaphoreQueue> as sync_types::Lock<AsyncSemaphoreQueue>>::Guard<'_>,
769 ) {
770 if !locked_queue.has_exclusive_all_waiters() {
771 // That's been the last one, disarm the wake-up path in
772 // return_trivial_lease_grant() again.
773 self.trivial_leases_granted.fetch_add(
774 Self::TRIVIAL_LEASES_GRANTED_NO_EXCLUSIVE_ALL_WAITERS_OFFSET,
775 atomic::Ordering::Relaxed,
776 );
777 }
778 }
779
780 /// Return a previously issued
781 /// [`TrivialLease`](AsyncSemaphoreLeaseGrantCount::TrivialLease) grant
782 /// back locklessly.
783 ///
784 /// It is explicitly permitted to invoke this **without** holding the
785 /// [`queue`](Self::queue) lock.
786 ///
787 /// In case the grant return potentially enables some queued
788 /// [`ExclusiveAll`](AsyncSemaphoreLeaseGrantCount::ExclusiveAll)
789 /// waiter to complete, `true` will get returned. In this case, the caller
790 /// **must** wake those via
791 /// [`wake_completed_waiters()`](Self::wake_completed_waiters).
792 fn return_trivial_lease_grant(&self) -> bool {
793 // In the common case of no ExclusiveAll waiter enqueued, the condition
794 // right below always evaluates to false, because trivial_leases_granted is
795 // offset by TRIVIAL_LEASES_GRANTED_NO_EXCLUSIVE_ALL_WAITERS_OFFSET.
796 if self.trivial_leases_granted.fetch_sub(1, atomic::Ordering::Release) == 1 {
797 // Optimistic load w/o a fence.
798 if self.leases_granted.load(atomic::Ordering::Relaxed) == 0 {
799 true
800 } else {
801 // Issue a SeqCst fence pairing with the one from the ExclusiveAll path in
802 // try_grant_one() and redo the test. The argument for correctness proceeds as
803 // follows: If leases_granted != 0 below, then the load is
804 // coherence-ordered before the decrement to zero in
805 // return_grant(), which happens-before the fence in the
806 // ExlusiveAll path in try_grant_one(). It follows that
807 // the fence below is SC-ordered before that latter fence. An analoguous
808 // argument can be made in the other direction for the case that
809 // the ExclusiveAll path in try_grant_one() observes a trivial_leases_granted !=
810 // 0, leading to a contradiction.
811 atomic::fence(atomic::Ordering::SeqCst);
812 self.leases_granted.load(atomic::Ordering::Relaxed) == 0
813 }
814 } else {
815 false
816 }
817 }
818
819 /// Return a previously issued grant back.
820 ///
821 /// Return granted leases back to the pool, making them available to
822 /// other waiters again. If there are some other such waiters queued,
823 /// they will be [woken](Self::wake_completed_waiters) for completion as
824 /// appropriate.
825 ///
826 /// # Arguments:
827 ///
828 /// * `locked_queue` - Guard for the locked [`queue`](Self::queue).
829 /// * `leases` - The grant to return.
830 fn return_grant(
831 &self,
832 locked_queue: &mut <ST::Lock<AsyncSemaphoreQueue> as sync_types::Lock<AsyncSemaphoreQueue>>::Guard<'_>,
833 leases: AsyncSemaphoreLeaseGrantCount,
834 ) {
835 let leases_granted = self.leases_granted.load(atomic::Ordering::Relaxed);
836 let mut wake = true;
837 match leases {
838 AsyncSemaphoreLeaseGrantCount::TrivialLease => {
839 wake = self.return_trivial_lease_grant();
840 }
841 AsyncSemaphoreLeaseGrantCount::Leases { count } => {
842 debug_assert!(count.get() <= leases_granted);
843 let leases_granted = leases_granted - count.get();
844 // Release pairs with Acquire in the ExclusiveAll path from try_grant_one(),
845 // c.f. the comment there.
846 self.leases_granted.store(leases_granted, atomic::Ordering::Release);
847 // Iff leases_granted dropped to zero *and* the next waiter is
848 // an ExclusiveAll one, then a SeqCst fence
849 // pairing with the one in return_trivial_lease_grant() will
850 // be issued before the test of trivial_leases_granted in
851 // wake_completed_waiters() -> try_grant_one().
852 }
853 AsyncSemaphoreLeaseGrantCount::ExclusiveAll => {
854 debug_assert!(self.is_exclusive_all_granted(locked_queue));
855 self.leases_granted.store(0, atomic::Ordering::Relaxed);
856 }
857 }
858 if wake {
859 self.wake_completed_waiters(locked_queue);
860 }
861 }
862
863 /// Complete and wake the maximum possible amount of waiters queued for
864 /// semaphore grants.
865 ///
866 /// Walk the [`queue`](Self::queue) and complete waiters as possible.
867 /// Completed waiters will be woken by means of their associated
868 /// [`wakers`](AsyncSemaphoreQueueEntry::waker), if any, and removed from
869 /// the queue.
870 ///
871 /// # Arguments:
872 ///
873 /// * `locked_queue` - Guard for the locked [`queue`](Self::queue).
874 fn wake_completed_waiters(
875 &self,
876 locked_queue: &mut <ST::Lock<AsyncSemaphoreQueue> as sync_types::Lock<AsyncSemaphoreQueue>>::Guard<'_>,
877 ) {
878 let mut i = 0;
879 let mut trivial_leases_only = false;
880 while i < locked_queue.queue.len() {
881 let entry = &locked_queue.queue[i];
882 if let AsyncSemaphoreLeaseGrantCount::Leases { count } = &entry.leases_requested {
883 if count.get() > locked_queue.max_leases {
884 // The semaphore has been shrunken, skip over this
885 // failed, already woken request.
886 i += 1;
887 continue;
888 }
889 }
890 let is_exclusive_all_waiter =
891 matches!(&entry.leases_requested, AsyncSemaphoreLeaseGrantCount::ExclusiveAll);
892 if (!trivial_leases_only || matches!(&entry.leases_requested, AsyncSemaphoreLeaseGrantCount::TrivialLease))
893 && self.try_grant_one(locked_queue, &entry.leases_requested)
894 {
895 let entry = &mut locked_queue.queue[i];
896 if let Some(waker) = entry.waker.take() {
897 waker.wake();
898 }
899 locked_queue.remove_entry(i);
900 if is_exclusive_all_waiter {
901 self.acquire_exclusive_all_end(locked_queue);
902 break;
903 }
904 } else if !is_exclusive_all_waiter
905 && (trivial_leases_only
906 || (!self.is_exclusive_all_granted(locked_queue) && locked_queue.has_trivial_lease_waiters()))
907 {
908 // It's not an ExlusiveAll waiter and a TrivialLease waiter would have been
909 // woken above. So it must be a Lease waiter.
910 debug_assert!(matches!(
911 &entry.leases_requested,
912 AsyncSemaphoreLeaseGrantCount::Leases { .. }
913 ));
914 // TrivialLease waiters shall be blocked by ExclusiveAll grants ahead in line
915 // only. Continue the search, waking those.
916 // Be careful to not complete subsequent Leases waiters with a possibly smaller
917 // count request.
918 trivial_leases_only = true;
919 i += 1;
920 } else {
921 break;
922 }
923 }
924 }
925
926 /// Wake all queued waiters bound for failure.
927 ///
928 /// Shrinking the [`AsyncSemaphore`] capacity might render some pending
929 /// requests unsatisfiable. Walk the [`queue`](Self::queue) and wake
930 /// waiters bound for failure by means of their associated
931 /// [`wakers`](AsyncSemaphoreQueueEntry::waker), if any. The waiters will be
932 /// left on the queue and subsequently removed from the associated
933 /// [`AsyncSemaphoreLeasesFuture`]'s `poll() once the latter has taken
934 /// notice of the failure.
935 ///
936 /// # Arguments:
937 ///
938 /// * `locked_queue` - Guard for the locked [`queue`](Self::queue).
939 fn wake_failed_waiters(
940 &self,
941 locked_queue: &mut <ST::Lock<AsyncSemaphoreQueue> as sync_types::Lock<AsyncSemaphoreQueue>>::Guard<'_>,
942 ) {
943 let mut failed_some = false;
944 let max_leases = locked_queue.max_leases;
945 for entry in locked_queue.queue.iter_mut() {
946 match entry.leases_requested {
947 AsyncSemaphoreLeaseGrantCount::Leases { count } if max_leases < count.get() => {
948 failed_some = true;
949 if let Some(waker) = entry.waker.take() {
950 waker.wake();
951 }
952 }
953 _ => (),
954 }
955 }
956
957 // Some now failed waiter might have blocked subsequent entries waiting
958 // for a smaller number of leases, which might already be
959 // available. Kick those waiters.
960 if failed_some {
961 self.wake_completed_waiters(locked_queue);
962 }
963 }
964
965 /// Cancel an enqueued waiter.
966 ///
967 /// Cancelling an enqueued waiter might potentially unblock subsequent
968 /// waiters in the queue, so they would get
969 /// [woken](Self::wake_completed_waiters) in the course of cancelling if so.
970 ///
971 /// # Arguments:
972 ///
973 /// * `locked_queue` - Guard for the locked [`queue`](Self::queue).
974 /// * `waiter_id` - The waiter id, as previously returned by
975 /// [`enqueue_waiter()`](AsyncSemaphoreQueueEntryAllocation::enqueue_waiter)
976 /// of the enqueued waiter instance to cancel.
977 /// * `leases_requested` - The grant request associated with the waiter to
978 /// cancel.
979 fn cancel_waiter(
980 &self,
981 locked_queue: &mut <ST::Lock<AsyncSemaphoreQueue> as sync_types::Lock<AsyncSemaphoreQueue>>::Guard<'_>,
982 waiter_id: num::NonZeroU64,
983 leases_requested: AsyncSemaphoreLeaseGrantCount,
984 ) {
985 match locked_queue.waiter_position(waiter_id) {
986 Some(index) => {
987 locked_queue.remove_entry(index);
988 match leases_requested {
989 AsyncSemaphoreLeaseGrantCount::TrivialLease => {
990 // A TrivialLease waiter can only be on the queue if some ExclusiveAll
991 // waiter is ahead of it. In either case, cancelling it won't unblock
992 // anything.
993 debug_assert!(
994 self.is_exclusive_all_granted(locked_queue) || locked_queue.has_exclusive_all_waiters()
995 );
996 }
997 AsyncSemaphoreLeaseGrantCount::Leases { count } => {
998 // This waiter might have blocked subsequent entries waiting for a smaller
999 // number of leases, which might already be available. Kick those waiters.
1000 if count.get() > 1 {
1001 self.wake_completed_waiters(locked_queue);
1002 }
1003 }
1004 AsyncSemaphoreLeaseGrantCount::ExclusiveAll => {
1005 self.acquire_exclusive_all_end(locked_queue);
1006 // This exclusive waiter might have blocked subsequent entries, kick them.
1007 self.wake_completed_waiters(locked_queue);
1008 }
1009 }
1010 }
1011 None => {
1012 // The waiter has previously been removed, which means that it got the requested
1013 // number of semaphore leases granted. The fact that the waiter is attempting to
1014 // cancel itself means that it hasn't noticed yet. Return the grants to the
1015 // pool.
1016 self.return_grant(locked_queue, leases_requested);
1017 }
1018 }
1019 }
1020
1021 /// Check whether some issued
1022 /// [`ExclusiveAll`](AsyncSemaphoreLeaseGrantCount::ExclusiveAll)
1023 /// grant is active.
1024 ///
1025 /// # Arguments:
1026 ///
1027 /// * `locked_queue` - Guard for the locked [`queue`](Self::queue).
1028 fn is_exclusive_all_granted(
1029 &self,
1030 locked_queue: &<ST::Lock<AsyncSemaphoreQueue> as sync_types::Lock<AsyncSemaphoreQueue>>::Guard<'_>,
1031 ) -> bool {
1032 self.leases_granted.load(atomic::Ordering::Relaxed) == locked_queue.max_leases + 1
1033 }
1034}
1035
1036/// A semaphore which can be waited asynchronously for.
1037///
1038/// There are two types of grants which can be issued:
1039/// * regular leases grants - these can be issued within the bounds of the
1040/// `AsyncSemaphore`'s capacity at a time and enable immutable access to the
1041/// wrapped data item. Obtained either asynchrononously via
1042/// [`acquire_leases()`](Self::acquire_leases), or synchronously via
1043/// [`try_acquire_leases()`](Self::try_acquire_leases).
1044/// * "exclusive-all" grants - this type of grant is mutually exclusive with any
1045/// other grant of any kind, that is, it is blocked by and blocks itself any
1046/// other regular leases grant or "exclusive-all" grant. "Exclusive-all"
1047/// grants provide mutable access to the wrapped data item. Obtained either
1048/// asynchrononously via
1049/// [`acquire_exclusive_all()`](Self::acquire_exclusive_all), or synchronously
1050/// via [`try_acquire_exclusive_all()`](Self::try_acquire_exclusive_all).
1051///
1052/// It is worth noting that, perhaps rather uncommon among semaphore
1053/// implementations, the regular leases grants can be zero-sized or "trivial".
1054/// For these "trivial" lease grants, the `AsyncSemaphore`'s capacity imposes
1055/// no effective bound whatsoever, their only significance is that they are
1056/// mutually exclusive with the "exclusive-all" grants. This feature can be used
1057/// to implement e.g. an asynchronous Read-Write-Lock on the grounds of
1058/// `AsyncSemaphore` right away. Furthermore, as the regular leases grants
1059/// can be [split](AsyncSemaphoreLeasesGuard::split_leases) at little cost, it
1060/// enables the instantiation of an indefinite amount of "trivial" lease grant
1061/// instances out of other lease grants, trivial or not, also providing
1062/// immutable access to the wrapped data item without consuming any additional
1063/// semaphore capacity. Observe how this resembles plain immutable Rust
1064/// references to some extent, but without those lifetime or borrow constraints
1065/// respectively which are often a hindrance in asynchronous settings.
1066///
1067/// `AsyncSemaphore` is **not** robust against threads "loosing interest", e.g.
1068/// because the execution environment somehow abandoned them: any leases grant
1069/// waiting [`Future`] acquired from [`acquire_leases()`](Self::acquire_leases)
1070/// or [`acquire_exclusive_all()`](Self::acquire_exclusive_all) must either get
1071/// polled to completion or dropped, otherwise it might block other waiters for
1072/// forever. Similar reasoning applies to the granted leases grants themselves:
1073/// [`AsyncSemaphoreLeasesGuard`] and [`AsyncSemaphoreExclusiveAllGuard`]
1074/// instances may always block other waiters, so it must be made
1075/// sure that progress is always driven forward for a lease holder until the
1076/// respective leases grant gets eventually dropped again.
1077pub struct AsyncSemaphore<ST: sync_types::SyncTypes, T: marker::Send + marker::Sync> {
1078 state: AsyncSemaphoreState<ST>,
1079 data: cell::UnsafeCell<T>,
1080}
1081
1082impl<ST: sync_types::SyncTypes, T: marker::Send + marker::Sync> AsyncSemaphore<ST, T> {
1083 /// Instantiate a new [`AsyncSemaphore`].
1084 ///
1085 /// Note that the caller is supposed to move the returned `AsyncSemaphore`
1086 /// into a [`SyncRcPtr`](sync_types::SyncRcPtr) -- the other parts of
1087 /// the API expect that.
1088 ///
1089 /// # Arguments:
1090 ///
1091 /// * `max_leases` - The semaphore capacity, i.e. the maximum on the total
1092 /// number of regular leases grants which can be active at a time. Must be
1093 /// **strictly** less than [`usize::MAX`]. May be zero to permit only
1094 /// "trivial" and "exlusive-all" leases.
1095 /// * `data` - The data item to wrap and synchronize accesses to.
1096 pub fn new(max_leases: usize, data: T) -> Result<Self, AsyncSemaphoreNewError> {
1097 if max_leases == usize::MAX {
1098 return Err(AsyncSemaphoreNewError::UnsupportedCapacity);
1099 }
1100
1101 Ok(Self {
1102 state: AsyncSemaphoreState::new(max_leases),
1103 data: cell::UnsafeCell::new(data),
1104 })
1105 }
1106
1107 /// Instantiate a new [`AsyncSemaphore`] with a capacity of zero.
1108 ///
1109 /// Equivalent to [`AsyncSemaphore::new(0)`], except that the
1110 /// `new_with_no_capacity()` is infallible.
1111 ///
1112 /// Note that the caller is supposed to move the returned `AsyncSemaphore`
1113 /// into a [`SyncRcPtr`](sync_types::SyncRcPtr) -- the other parts of
1114 /// the API expect that.
1115 ///
1116 /// # Arguments:
1117 ///
1118 /// * `data` - The data item to wrap and synchronize accesses to.
1119 pub fn new_with_no_capacity(data: T) -> Self {
1120 Self {
1121 state: AsyncSemaphoreState::new(0),
1122 data: cell::UnsafeCell::new(data),
1123 }
1124 }
1125
1126 /// The semaphore capacity, i.e. the maximum on the total number of regular
1127 /// leases grants which can be active at a time.
1128 ///
1129 /// Note that a semaphore's capacity can be adjusted after instantiation,
1130 /// either through leases grant
1131 /// [leaking](AsyncSemaphoreLeasesGuard::leak), or
1132 /// [directly](AsyncSemaphoreExclusiveAllGuard::resize_semaphore) through an
1133 /// "exclusive-all" grant, so be careful to stabilize the value as
1134 /// needed.
1135 pub fn max_leases(&self) -> usize {
1136 self.state.max_leases()
1137 }
1138
1139 /// Asynchronously acquire a regular semaphore leases grant.
1140 ///
1141 /// Instantiate an [`AsyncSemaphoreLeasesFuture`] for asynchronous
1142 /// acquisition of the specified number of leases, `leases_requested`.
1143 ///
1144 /// The returned future will only become ready once all previously submitted
1145 /// requests have been completed and the semaphore has the requested
1146 /// number of leases available.
1147 ///
1148 /// In order to avoid deadlocks, care must be taken to not build up a task's
1149 /// granted leases incrementally: if some task already owns some leases,
1150 /// including "trivial" ones, by the time it invokes another
1151 /// [`acquire_leases()`](Self::acquire_leases) operation, it can block some
1152 /// other requests scheduled ahead to make progress, thereby effectively
1153 /// blocking itself from making any progress. If needed, a task can
1154 /// allocate all needed leases at once and then
1155 /// [split](AsyncSemaphoreLeasesGuard::split_leases) the grant into
1156 /// parts as needed.
1157 ///
1158 /// Furthermore, note that the mere existence of a
1159 /// [`AsyncSemaphoreLeasesFuture`] returned from this function may block
1160 /// other waiters -- it **must** always get either polled to completion or
1161 /// dropped again.
1162 ///
1163 /// # Arguments:
1164 ///
1165 /// * `leases_requested` - The number of leases to obtain. May be set to
1166 /// zero for requesting a "trivial" lease grant.
1167 ///
1168 /// # Errors:
1169 ///
1170 /// * [`AsyncSemaphoreError::RequestExceedsSemaphoreCapacity`] - The number
1171 /// of leases requested exceeds the [`AsyncSemaphore`]'s total capacity.
1172 /// * [`AsyncSemaphoreError::MemoryAllocationFailure`] - Memory allocation
1173 /// failure.
1174 pub fn acquire_leases<'a, SP: 'a + sync_types::SyncRcPtr<Self>, SR: sync_types::SyncRcPtrRef<'a, Self, SP>>(
1175 this: &SR,
1176 leases_requested: usize,
1177 ) -> Result<AsyncSemaphoreLeasesFuture<ST, T, SP>, AsyncSemaphoreError> {
1178 match this.state.prepare_acquire_leases(leases_requested)? {
1179 Some(waiter_id) => Ok(AsyncSemaphoreLeasesFuture {
1180 private: AsyncSemaphoreLeasesFuturePriv::Enqueued {
1181 sem: this.make_weak_clone(),
1182 waiter_id,
1183 leases_requested,
1184 },
1185 }),
1186 None => Ok(AsyncSemaphoreLeasesFuture {
1187 private: AsyncSemaphoreLeasesFuturePriv::LeasesGranted {
1188 sem: this.make_weak_clone(),
1189 leases_granted: leases_requested,
1190 },
1191 }),
1192 }
1193 }
1194
1195 /// Asynchronously acquire an "exclusive-all" grant on all of a semaphore's
1196 /// capacity.
1197 ///
1198 /// Instantiate an [`AsyncSemaphoreExclusiveAllFuture`] for asynchronous
1199 /// acquisition of an "exclusive-all" grant on all of the
1200 /// [`AsyncSemaphore`]'s capacity.
1201 ///
1202 /// The returned future will only become ready once all previously submitted
1203 /// requests have been completed and the resulting guards dropped again.
1204 ///
1205 /// Note that the mere existence of a
1206 /// [`AsyncSemaphoreExclusiveAllFuture`] returned from this function may
1207 /// block other waiters -- it **must** always get either polled to
1208 /// completion or dropped again.
1209 ///
1210 /// # Errors:
1211 ///
1212 /// * [`AsyncSemaphoreError::MemoryAllocationFailure`] - Memory allocation
1213 /// failure.
1214 pub fn acquire_exclusive_all<
1215 'a,
1216 SP: 'a + sync_types::SyncRcPtr<Self>,
1217 SR: sync_types::SyncRcPtrRef<'a, Self, SP>,
1218 >(
1219 this: &SR,
1220 ) -> Result<AsyncSemaphoreExclusiveAllFuture<ST, T, SP>, AsyncSemaphoreError> {
1221 match this.state.prepare_acquire_exclusive_all()? {
1222 Some(waiter_id) => Ok(AsyncSemaphoreExclusiveAllFuture {
1223 private: AsyncSemaphoreExclusiveAllFuturePriv::Enqueued {
1224 sem: this.make_weak_clone(),
1225 waiter_id,
1226 },
1227 }),
1228 None => Ok(AsyncSemaphoreExclusiveAllFuture {
1229 private: AsyncSemaphoreExclusiveAllFuturePriv::Granted {
1230 sem: this.make_weak_clone(),
1231 },
1232 }),
1233 }
1234 }
1235
1236 /// Try to synchronously acquire a regular semaphore leases grant.
1237 ///
1238 /// The operation will only succeed and return a
1239 /// [`AsyncSemaphoreLeasesGuard`] if the requested number of leases is
1240 /// available and there is no other, conflicting waiter ahead in line.
1241 /// Otherwise [`None`] will get returned.
1242 ///
1243 /// # Arguments:
1244 ///
1245 /// * `leases_requested` - The number of leases to obtain. May be set to
1246 /// zero for requesting a "trivial" lease grant.
1247 ///
1248 /// # Errors:
1249 ///
1250 /// * [`AsyncSemaphoreError::RequestExceedsSemaphoreCapacity`] - The number
1251 /// of leases requested exceeds the [`AsyncSemaphore`]'s total capacity.
1252 pub fn try_acquire_leases<'a, SP: 'a + sync_types::SyncRcPtr<Self>, SR: sync_types::SyncRcPtrRef<'a, Self, SP>>(
1253 this: &SR,
1254 leases_requested: usize,
1255 ) -> Result<Option<AsyncSemaphoreLeasesGuard<ST, T, SP>>, AsyncSemaphoreError> {
1256 if this.state.try_acquire_leases(leases_requested)? {
1257 Ok(Some(AsyncSemaphoreLeasesGuard::new(
1258 this.make_clone(),
1259 leases_requested,
1260 )))
1261 } else {
1262 Ok(None)
1263 }
1264 }
1265
1266 /// Try to synchronously acquire an "exclusive-all" grant on all of a
1267 /// semaphore's capacity.
1268 ///
1269 /// The operation will only succeed and return a
1270 /// [`AsyncSemaphoreExclusiveAllGuard`] if no other grants of any kind are
1271 /// active. Otherwise [`None`] will get returned.
1272 pub fn try_acquire_exclusive_all<
1273 'a,
1274 SP: 'a + sync_types::SyncRcPtr<Self>,
1275 SR: sync_types::SyncRcPtrRef<'a, Self, SP>,
1276 >(
1277 this: &SR,
1278 ) -> Option<AsyncSemaphoreExclusiveAllGuard<ST, T, SP>> {
1279 if this.state.try_acquire_exclusive_all() {
1280 Some(AsyncSemaphoreExclusiveAllGuard::new(this.make_clone()))
1281 } else {
1282 None
1283 }
1284 }
1285}
1286
1287// A non-Send semaphore would be quite pointless.
1288unsafe impl<ST: sync_types::SyncTypes, T: marker::Send + marker::Sync> marker::Send for AsyncSemaphore<ST, T> {}
1289
1290// A non-Sync semaphore would be quite pointless.
1291unsafe impl<ST: sync_types::SyncTypes, T: marker::Send + marker::Sync> marker::Sync for AsyncSemaphore<ST, T> {}
1292
1293/// [`DerefInnerByTag`](sync_types::DerefInnerByTag) `TAG` for dereferencing
1294/// [`AsyncSemaphore::data`].
1295///
1296/// Enables presenting the [`SyncRcPtr`](sync_types::SyncRcPtr) of the
1297/// [`AsyncSemaphore`] as one for its [`AsyncSemaphore::data`] member
1298/// via the [`SyncRcPtrForInner`](sync_types::SyncRcPtrForInner) mechanism.
1299struct AsyncSemaphoreDerefInnerDataTag;
1300
1301impl<ST: sync_types::SyncTypes, T: marker::Send + marker::Sync>
1302 sync_types::DerefInnerByTag<AsyncSemaphoreDerefInnerDataTag> for AsyncSemaphore<ST, T>
1303{
1304 crate::impl_deref_inner_by_tag!(data, cell::UnsafeCell<T>);
1305}
1306
1307/// Internal [`AsyncSemaphoreLeasesFuture`] state.
1308enum AsyncSemaphoreLeasesFuturePriv<
1309 ST: sync_types::SyncTypes,
1310 T: marker::Send + marker::Sync,
1311 SP: sync_types::SyncRcPtr<AsyncSemaphore<ST, T>>,
1312> {
1313 /// The requested number of semaphore leases had been unavailable at
1314 /// enqueueing time and the waiter got indeed enqueued.
1315 Enqueued {
1316 sem: SP::WeakSyncRcPtr,
1317 waiter_id: num::NonZeroU64,
1318 leases_requested: usize,
1319 },
1320 /// The requested number of semaphore leases had been available at
1321 /// enqueueing time and they got granted right away.
1322 LeasesGranted {
1323 sem: SP::WeakSyncRcPtr,
1324 leases_granted: usize,
1325 },
1326 /// The future is done: the semaphore leases had been acquired at some time
1327 /// and polled out to the user.
1328 Done,
1329}
1330
1331/// Asynchronous wait for [`AsyncSemaphore`] leases.
1332///
1333/// To be obtained through [`AsyncSemaphore::acquire_leases()`].
1334///
1335/// # Note on lifetime management
1336///
1337/// An [`AsyncSemaphoreLeasesFuture`] instance will only maintain a weak
1338/// reference (i.e. a [`WeakSyncRcPtr`](sync_types::WeakSyncRcPtr)) to the
1339/// associated [`AsyncSemaphore`] instance and thus, would not hinder its
1340/// deallocation. In case the semaphore gets dropped before the future had a
1341/// chance to acquire leases from it, its `poll()` would return
1342/// [`AsyncSemaphoreError::StaleSemaphore`].
1343pub struct AsyncSemaphoreLeasesFuture<
1344 ST: sync_types::SyncTypes,
1345 T: marker::Send + marker::Sync,
1346 SP: sync_types::SyncRcPtr<AsyncSemaphore<ST, T>>,
1347> {
1348 private: AsyncSemaphoreLeasesFuturePriv<ST, T, SP>,
1349}
1350
1351impl<ST: sync_types::SyncTypes, T: marker::Send + marker::Sync, SP: sync_types::SyncRcPtr<AsyncSemaphore<ST, T>>>
1352 AsyncSemaphoreLeasesFuture<ST, T, SP>
1353{
1354 /// Obtain the associated [`AsyncSemaphore`].
1355 ///
1356 /// Return the associated [`AsyncSemaphore`] wrapped in `Some` if still
1357 /// alive, `None` otherwise.
1358 pub fn get_semaphore(&self) -> Option<SP> {
1359 let weak_sem = match &self.private {
1360 AsyncSemaphoreLeasesFuturePriv::Enqueued {
1361 sem,
1362 waiter_id: _,
1363 leases_requested: _,
1364 } => sem,
1365 AsyncSemaphoreLeasesFuturePriv::LeasesGranted { sem, leases_granted: _ } => sem,
1366 AsyncSemaphoreLeasesFuturePriv::Done => return None,
1367 };
1368 weak_sem.upgrade()
1369 }
1370}
1371
1372impl<ST: sync_types::SyncTypes, T: marker::Send + marker::Sync, SP: sync_types::SyncRcPtr<AsyncSemaphore<ST, T>>>
1373 marker::Unpin for AsyncSemaphoreLeasesFuture<ST, T, SP>
1374{
1375}
1376
1377impl<ST: sync_types::SyncTypes, T: marker::Send + marker::Sync, SP: sync_types::SyncRcPtr<AsyncSemaphore<ST, T>>>
1378 future::Future for AsyncSemaphoreLeasesFuture<ST, T, SP>
1379{
1380 type Output = Result<AsyncSemaphoreLeasesGuard<ST, T, SP>, AsyncSemaphoreError>;
1381
1382 /// Poll for a regular leases grant from the associated
1383 /// [`AsyncSemaphore`].
1384 ///
1385 /// Upon successful future completion, an [`AsyncSemaphoreLeasesGuard`] is
1386 /// returned, otherwise some error information indicating the cause of the
1387 /// failure.
1388 ///
1389 /// The future must not get polled any further once completed.
1390 ///
1391 /// # Errors:
1392 ///
1393 /// * [`AsyncSemaphoreError::RequestExceedsSemaphoreCapacity`] - The
1394 /// associated [`AsyncSemaphore`]'s capacity has been shrunken below the
1395 /// requested number of leases after the request was made.
1396 /// * [`AsyncSemaphoreError::StaleSemaphore`] - The associated
1397 /// [`AsyncSemaphore`] has been dropped after the request was made.
1398 fn poll(self: pin::Pin<&mut Self>, cx: &mut task::Context<'_>) -> task::Poll<Self::Output> {
1399 let this = self.get_mut();
1400 match &this.private {
1401 AsyncSemaphoreLeasesFuturePriv::Enqueued {
1402 sem,
1403 waiter_id,
1404 leases_requested,
1405 } => {
1406 let sem = match sem.upgrade() {
1407 Some(sem) => sem,
1408 None => {
1409 // The semaphore is gone, indicating some teardown going on.
1410 return task::Poll::Ready(Err(AsyncSemaphoreError::StaleSemaphore));
1411 }
1412 };
1413 let mut locked_queue = sem.state.queue.lock();
1414 if locked_queue.poll_waiter(*waiter_id, cx.waker().clone()) {
1415 drop(locked_queue);
1416 let leases_granted = *leases_requested;
1417 this.private = AsyncSemaphoreLeasesFuturePriv::Done;
1418 task::Poll::Ready(Ok(AsyncSemaphoreLeasesGuard::new(sem, leases_granted)))
1419 } else if locked_queue.max_leases < *leases_requested {
1420 // The semaphore's capacity has been shrunken below the number of requested
1421 // leases in the meanwhile, fail the request.
1422 locked_queue.remove_waiter(*waiter_id);
1423 this.private = AsyncSemaphoreLeasesFuturePriv::Done;
1424 task::Poll::Ready(Err(AsyncSemaphoreError::RequestExceedsSemaphoreCapacity))
1425 } else {
1426 task::Poll::Pending
1427 }
1428 }
1429 AsyncSemaphoreLeasesFuturePriv::LeasesGranted { sem, leases_granted } => {
1430 let sem = match sem.upgrade() {
1431 Some(sem) => sem,
1432 None => {
1433 // The semaphore is gone, indicating some teardown going on.
1434 return task::Poll::Ready(Err(AsyncSemaphoreError::StaleSemaphore));
1435 }
1436 };
1437 let leases_granted = *leases_granted;
1438 this.private = AsyncSemaphoreLeasesFuturePriv::Done;
1439 task::Poll::Ready(Ok(AsyncSemaphoreLeasesGuard::new(sem, leases_granted)))
1440 }
1441 AsyncSemaphoreLeasesFuturePriv::Done => {
1442 // The semaphore leases had been acquired and handed out already.
1443 task::Poll::Ready(Err(AsyncSemaphoreError::Internal))
1444 }
1445 }
1446 }
1447}
1448
1449impl<ST: sync_types::SyncTypes, T: marker::Send + marker::Sync, SP: sync_types::SyncRcPtr<AsyncSemaphore<ST, T>>> Drop
1450 for AsyncSemaphoreLeasesFuture<ST, T, SP>
1451{
1452 fn drop(&mut self) {
1453 match &self.private {
1454 AsyncSemaphoreLeasesFuturePriv::Enqueued {
1455 sem,
1456 waiter_id,
1457 leases_requested,
1458 } => {
1459 if let Some(sem) = sem.upgrade() {
1460 sem.state.cancel_waiter(
1461 &mut sem.state.queue.lock(),
1462 *waiter_id,
1463 AsyncSemaphoreLeaseGrantCount::from(*leases_requested),
1464 );
1465 }
1466 }
1467 AsyncSemaphoreLeasesFuturePriv::LeasesGranted { sem, leases_granted } => {
1468 // The semaphore leases had been granted right from the beginning, but the
1469 // future never got polled for them. Return the grants.
1470 if let Some(sem) = sem.upgrade() {
1471 sem.state.return_grant(
1472 &mut sem.state.queue.lock(),
1473 AsyncSemaphoreLeaseGrantCount::from(*leases_granted),
1474 );
1475 }
1476 }
1477 AsyncSemaphoreLeasesFuturePriv::Done => (),
1478 }
1479 }
1480}
1481
1482/// RAII guard for a regular leases grant acquired from an [`AsyncSemaphore`].
1483///
1484/// To be obtained either synchronously through
1485/// [`AsyncSemaphore::try_acquire_leases()`] or asynchronously via
1486/// [AsyncSemaphore::acquire_leases()].
1487///
1488/// Provides immutable access to the associated [`AsyncSemaphore`]'s wrapped
1489/// data item.
1490///
1491/// Until an [`AsyncSemaphoreLeasesGuard`] gets dropped again, it will block
1492/// other conflicting requests to the associated [`AsyncSemaphore`] from making
1493/// progress:
1494/// - other, subsequent requests for regular leases grants if the total number
1495/// of active leases would exceed the semaphore's capacity any
1496/// - any requests for "exclusive-all" grants.
1497///
1498/// It should be worth noting again, that [`AsyncSemaphoreLeasesGuard`] for
1499/// "trivial" lease grants, i.e. grants for a zero number of semaphore leases,
1500/// do establish mutual exclusion with "exclusive-all" grants.
1501///
1502/// An [`AsyncSemaphoreLeasesGuard`] can be [split](Self::split_leases) into
1503/// two, distributing the currently owned leases among the new instances as
1504/// specified. This is needed for obtaining more than one semaphore lease in a
1505/// deadlock-free manner, c.f. the discussion at
1506/// [`AsyncSemaphore::acquire_leases()`].
1507pub struct AsyncSemaphoreLeasesGuard<
1508 ST: sync_types::SyncTypes,
1509 T: marker::Send + marker::Sync,
1510 SP: sync_types::SyncRcPtr<AsyncSemaphore<ST, T>>,
1511> {
1512 sem: Option<SP>,
1513 leases_granted: usize,
1514 _phantom: marker::PhantomData<fn() -> (ST, T)>,
1515}
1516
1517impl<ST: sync_types::SyncTypes, T: marker::Send + marker::Sync, SP: sync_types::SyncRcPtr<AsyncSemaphore<ST, T>>>
1518 AsyncSemaphoreLeasesGuard<ST, T, SP>
1519{
1520 fn new(sem: SP, leases_granted: usize) -> Self {
1521 Self {
1522 sem: Some(sem),
1523 leases_granted,
1524 _phantom: marker::PhantomData,
1525 }
1526 }
1527
1528 /// Obtain the associated [`AsyncSemaphore`].
1529 pub fn get_semaphore(&self) -> SP::SyncRcPtrRef<'_> {
1530 self.sem.as_ref().unwrap().as_ref()
1531 }
1532
1533 /// Release the lease grant and return the associated [`AsyncSemaphore`].
1534 pub fn into_semaphore(mut self) -> SP {
1535 let sem = self.sem.take().unwrap();
1536 Self::_drop(&sem, self.leases_granted);
1537 sem
1538 }
1539
1540 /// The number of leases acquired from the semaphore.
1541 pub fn leases(&self) -> usize {
1542 self.leases_granted
1543 }
1544
1545 /// Split the leases grant into two.
1546 ///
1547 /// The specified number of `leases` gets split off from `self` and wrapped
1548 /// in a new [`AsyncSemaphoreLeasesGuard`] instance to be returned.
1549 ///
1550 /// This is intended for situations where more than one lease need to get
1551 /// obtained at once, to avoid deadlocks, for example, but then
1552 /// subsequently get managed independently, c.f. the discussion at
1553 /// [`AsyncSemaphore::acquire_leases()`].
1554 ///
1555 /// # Arguments:
1556 ///
1557 /// * `leases` - The number of leases to split off from `self`. May be set
1558 /// to zero for requesting a "trivial" lease grant.
1559 ///
1560 /// # Errors:
1561 /// * [`AsyncSemaphoreError::Internal`] - The number of `leases` to split
1562 /// off exceeds the number of leases owned by `self`.
1563 pub fn split_leases(&mut self, leases: usize) -> Result<Self, AsyncSemaphoreError> {
1564 if leases > self.leases_granted {
1565 Err(AsyncSemaphoreError::Internal)
1566 } else {
1567 if leases == 0 || leases == self.leases_granted {
1568 // Either the new or the old grant will become a trivial one, account for it
1569 // at the semaphore.
1570 //
1571 // At least one of leases_granted or trivial_leases_granted is non-zero at this
1572 // point. If only the former, the Release decrement in
1573 // return_grant() will ensure that the ExclusiveAll
1574 // path in try_grant_one() will see the increment here.
1575 self.sem
1576 .as_ref()
1577 .unwrap()
1578 .state
1579 .trivial_leases_granted
1580 .fetch_add(1, atomic::Ordering::Relaxed);
1581 }
1582
1583 self.leases_granted -= leases;
1584 Ok(Self {
1585 sem: self.sem.clone(),
1586 leases_granted: leases,
1587 _phantom: marker::PhantomData,
1588 })
1589 }
1590 }
1591
1592 /// Split off a zero-sized, "trivial" lease grant.
1593 ///
1594 /// Effectively a special-cased, never failing variant of
1595 /// [`split_leases()`](Self::split_leases) for the case that the latter's
1596 /// `leases` argument is known to be zero, i.e. for splitting off a
1597 /// "trivial" lease grant.
1598 ///
1599 /// Returns a [`AsyncSemaphoreLeasesGuard`] wrapping a zero-sized, i.e.
1600 /// "trivial" lease grant.
1601 pub fn spawn_trivial_lease(&self) -> Self {
1602 self.sem
1603 .as_ref()
1604 .unwrap()
1605 .state
1606 .trivial_leases_granted
1607 .fetch_add(1, atomic::Ordering::Relaxed);
1608 Self {
1609 sem: self.sem.clone(),
1610 leases_granted: 0,
1611 _phantom: marker::PhantomData,
1612 }
1613 }
1614
1615 /// Permanently remove the owned leases from the associated
1616 /// [`AsyncSemaphore`]'s capacity.
1617 ///
1618 /// Note that this might potentially make some other, still enqueued waiters
1619 /// to
1620 /// fail with
1621 /// [`RequestExceedsSemaphoreCapacity`](AsyncSemaphoreError::RequestExceedsSemaphoreCapacity)
1622 /// in case their number of leases requested would now exceed the remaining
1623 /// capacity.
1624 pub fn leak(mut self) {
1625 if self.leases_granted == 0 {
1626 // A trivial lease, just return it.
1627 drop(self);
1628 return;
1629 }
1630 let sem = self.sem.take().unwrap();
1631 let mut locked_queue = sem.state.queue.lock();
1632 debug_assert!(locked_queue.max_leases >= self.leases_granted);
1633 locked_queue.max_leases -= self.leases_granted;
1634 debug_assert!(sem.state.leases_granted.load(atomic::Ordering::Relaxed) >= self.leases_granted);
1635 // Release pairs with Acquire in the ExclusiveAll path from try_grant_one(),
1636 // c.f. the comment there.
1637 sem.state.leases_granted.store(
1638 sem.state.leases_granted.load(atomic::Ordering::Relaxed) - self.leases_granted,
1639 atomic::Ordering::Release,
1640 );
1641
1642 // Shrinking the semaphore capcacity might have rendered some pending request
1643 // unsatisfiable. Wake them.
1644 sem.state.wake_failed_waiters(&mut locked_queue);
1645 // This lease grant could have blocked subsequent ExclusiveAll waiters.
1646 // Wake those as well.
1647 if locked_queue.has_exclusive_all_waiters() {
1648 sem.state.wake_completed_waiters(&mut locked_queue);
1649 }
1650 }
1651
1652 /// Convert into an [`AsyncSemaphoreLeasesWeakGuard`] not hindering
1653 /// destruction of the associated [`AsyncSemaphore`].
1654 pub fn into_weak(mut self) -> AsyncSemaphoreLeasesWeakGuard<ST, T, SP> {
1655 let sem = self.sem.take().unwrap();
1656 AsyncSemaphoreLeasesWeakGuard {
1657 sem: Some(sem.downgrade()),
1658 leases_granted: self.leases_granted,
1659 _phantom: marker::PhantomData,
1660 }
1661 }
1662
1663 /// Convert into a pair of raw pointer to the protected inner value and
1664 /// the granted leases count.
1665 ///
1666 /// Must eventually get converted back with [`from_raw()`](Self::from_raw)
1667 /// or the leases grant will be leaked forever.
1668 pub(super) fn into_raw(mut self) -> (*const T, usize) {
1669 let sem_ptr = SP::into_raw(self.sem.take().unwrap());
1670 let data_ptr =
1671 <AsyncSemaphore<ST, T> as sync_types::DerefInnerByTag<AsyncSemaphoreDerefInnerDataTag>>::to_inner_ptr(
1672 sem_ptr,
1673 );
1674 (cell::UnsafeCell::raw_get(data_ptr), self.leases_granted)
1675 }
1676
1677 /// Convert back from a pair of raw pointer to the protected value and
1678 /// granted leases count obtained from [`from_raw()`](Self::from_raw).
1679 ///
1680 /// # Safety
1681 ///
1682 /// The pair of raw pointer to the protected value and granted leases count
1683 /// must have been previously obtained from
1684 /// [`into_raw()`](Self::into_raw).
1685 pub(super) unsafe fn from_raw(data_ptr: *const T, leases_granted: usize) -> Self {
1686 // UnsafeCell<T> is transmutable to T.
1687 let data_ptr = data_ptr as *const cell::UnsafeCell<T>;
1688 // This is safe, the ptr is required to come from into_raw()
1689 let sem_ptr = unsafe {
1690 <AsyncSemaphore<ST, T> as sync_types::DerefInnerByTag<AsyncSemaphoreDerefInnerDataTag>>::container_of(
1691 data_ptr,
1692 )
1693 };
1694 // Likewise.
1695 let sem = unsafe { SP::from_raw(sem_ptr) };
1696
1697 Self {
1698 sem: Some(sem),
1699 leases_granted,
1700 _phantom: marker::PhantomData,
1701 }
1702 }
1703
1704 /// Return the leases grant to the the [`AsyncSemaphore`], thereby waking
1705 /// other waiters, if any.
1706 fn _drop(sem: &AsyncSemaphore<ST, T>, leases_granted: usize) {
1707 if leases_granted == 0 {
1708 // Fast path for TrivialLease guards.
1709 let wake = sem.state.return_trivial_lease_grant();
1710 if wake {
1711 let mut locked_queue = sem.state.queue.lock();
1712 sem.state.wake_completed_waiters(&mut locked_queue);
1713 }
1714 } else {
1715 sem.state.return_grant(
1716 &mut sem.state.queue.lock(),
1717 AsyncSemaphoreLeaseGrantCount::from(leases_granted),
1718 );
1719 }
1720 }
1721}
1722
1723impl<ST: sync_types::SyncTypes, T: marker::Send + marker::Sync, SP: sync_types::SyncRcPtr<AsyncSemaphore<ST, T>>> Drop
1724 for AsyncSemaphoreLeasesGuard<ST, T, SP>
1725{
1726 fn drop(&mut self) {
1727 if let Some(sem) = self.sem.as_ref() {
1728 Self::_drop(sem, self.leases_granted);
1729 }
1730 }
1731}
1732
1733impl<ST: sync_types::SyncTypes, T: marker::Send + marker::Sync, SP: sync_types::SyncRcPtr<AsyncSemaphore<ST, T>>>
1734 ops::Deref for AsyncSemaphoreLeasesGuard<ST, T, SP>
1735{
1736 type Target = T;
1737
1738 fn deref(&self) -> &Self::Target {
1739 let p = self.sem.as_ref().unwrap().data.get() as *const T;
1740 unsafe { &*p }
1741 }
1742}
1743
1744/// Weak variant of [`AsyncSemaphoreLeasesGuard`] not hindering destruction of
1745/// the associated [`AsyncSemaphore`].
1746///
1747/// In cases were it's desired that long-living lease grants don't prevent a
1748/// destruction of the associated [`AsyncSemaphore`], or to break cycles, a
1749/// `AsyncSemaphoreLeasesWeakGuard` may be used. To be obtained via
1750/// [`AsyncSemaphoreLeasesGuard::into_weak()`].
1751///
1752/// An `AsyncSemaphoreLeasesWeakGuard` may get converted back into a full
1753/// [`AsyncSemaphoreLeasesGuard`] via [`upgrade()`](Self::upgrade).
1754pub struct AsyncSemaphoreLeasesWeakGuard<
1755 ST: sync_types::SyncTypes,
1756 T: marker::Send + marker::Sync,
1757 SP: sync_types::SyncRcPtr<AsyncSemaphore<ST, T>>,
1758> {
1759 sem: Option<SP::WeakSyncRcPtr>,
1760 leases_granted: usize,
1761 _phantom: marker::PhantomData<fn() -> (ST, T)>,
1762}
1763
1764impl<ST: sync_types::SyncTypes, T: marker::Send + marker::Sync, SP: sync_types::SyncRcPtr<AsyncSemaphore<ST, T>>>
1765 AsyncSemaphoreLeasesWeakGuard<ST, T, SP>
1766{
1767 /// Attempt to convert back into an [`AsyncSemaphoreLeasesGuard`].
1768 ///
1769 /// Returns the [`AsyncSemaphoreLeasesGuard`] wrapped in `Some` if the
1770 /// associated [`AsyncSemaphore`] is still alive, `None otherwise.
1771 pub fn upgrade(mut self) -> Option<AsyncSemaphoreLeasesGuard<ST, T, SP>> {
1772 self.sem.take().unwrap().upgrade().map(|sem| AsyncSemaphoreLeasesGuard {
1773 sem: Some(sem),
1774 leases_granted: self.leases_granted,
1775 _phantom: marker::PhantomData,
1776 })
1777 }
1778
1779 /// Try to spawn off a "trivial" lease grant.
1780 ///
1781 /// Returns the [`AsyncSemaphoreLeasesGuard`] for the spawned "trivial"
1782 /// lease grant wrapped in `Some` if the associated [`AsyncSemaphore`]
1783 /// is still alive, `None otherwise.
1784 pub fn try_spawn_trivial_lease(&self) -> Option<AsyncSemaphoreLeasesGuard<ST, T, SP>> {
1785 match self.sem.as_ref().unwrap().upgrade() {
1786 Some(sem) => {
1787 sem.state.trivial_leases_granted.fetch_add(1, atomic::Ordering::Relaxed);
1788 Some(AsyncSemaphoreLeasesGuard {
1789 sem: Some(sem),
1790 leases_granted: 0,
1791 _phantom: marker::PhantomData,
1792 })
1793 }
1794 None => None,
1795 }
1796 }
1797
1798 /// Convert into a pair of raw pointer to the protected inner value and
1799 /// the granted leases count.
1800 ///
1801 /// Must eventually get converted back with [`from_raw()`](Self::from_raw)
1802 /// or the leases grant will be leaked forever.
1803 ///
1804 /// The returned raw pointer **must not** be used for anything except for
1805 /// passing it back to [`from_raw()`](Self::from_raw). In particular it
1806 /// must not get dereferenced, as the associated [`AsyncSemaphore`], and
1807 /// hence its wrapped value, might have been dropped already.
1808 pub(super) fn into_raw(mut self) -> (*const T, usize) {
1809 let sem_ptr = SP::WeakSyncRcPtr::into_raw(self.sem.take().unwrap());
1810 if !sem_ptr.is_null() {
1811 let data_ptr =
1812 <AsyncSemaphore<ST, T> as sync_types::DerefInnerByTag<AsyncSemaphoreDerefInnerDataTag>>::to_inner_ptr(
1813 sem_ptr,
1814 );
1815 (cell::UnsafeCell::raw_get(data_ptr), self.leases_granted)
1816 } else {
1817 (ptr::null(), self.leases_granted)
1818 }
1819 }
1820
1821 /// Convert back from a pair of raw pointer to the protected value and
1822 /// granted leases count.
1823 ///
1824 /// # Safety
1825 ///
1826 /// The pair of raw pointer to the protected value and granted leases count
1827 /// must have been previously obtained from
1828 /// [`into_raw()`](Self::into_raw).
1829 pub(super) unsafe fn from_raw(data_ptr: *const T, leases_granted: usize) -> Self {
1830 let sem_ptr = if !data_ptr.is_null() {
1831 // UnsafeCell<T> is transmutable to T.
1832 let data_ptr = data_ptr as *const cell::UnsafeCell<T>;
1833 // This is safe, the ptr is required to come from into_raw()
1834 unsafe {
1835 <AsyncSemaphore<ST, T> as sync_types::DerefInnerByTag<AsyncSemaphoreDerefInnerDataTag>>::container_of(
1836 data_ptr,
1837 )
1838 }
1839 } else {
1840 ptr::null()
1841 };
1842 // Likewise.
1843 let sem = unsafe { SP::WeakSyncRcPtr::from_raw(sem_ptr) };
1844
1845 Self {
1846 sem: Some(sem),
1847 leases_granted,
1848 _phantom: marker::PhantomData,
1849 }
1850 }
1851}
1852
1853impl<ST: sync_types::SyncTypes, T: marker::Send + marker::Sync, SP: sync_types::SyncRcPtr<AsyncSemaphore<ST, T>>> Drop
1854 for AsyncSemaphoreLeasesWeakGuard<ST, T, SP>
1855{
1856 fn drop(&mut self) {
1857 if let Some(sem) = self.sem.take() {
1858 if let Some(sem) = sem.upgrade() {
1859 drop(AsyncSemaphoreLeasesGuard {
1860 sem: Some(sem),
1861 leases_granted: self.leases_granted,
1862 _phantom: marker::PhantomData,
1863 });
1864 }
1865 }
1866 }
1867}
1868
1869/// Internal [`AsyncSemaphoreExclusiveAllFuture`] state.
1870enum AsyncSemaphoreExclusiveAllFuturePriv<
1871 ST: sync_types::SyncTypes,
1872 T: marker::Send + marker::Sync,
1873 SP: sync_types::SyncRcPtr<AsyncSemaphore<ST, T>>,
1874> {
1875 /// The exclusive grant on all of the semaphore's capcaity had been
1876 /// unavailable at enqueueing time and the waiter got indeed enqueued.
1877 Enqueued {
1878 sem: SP::WeakSyncRcPtr,
1879 waiter_id: num::NonZeroU64,
1880 },
1881 /// All of the semaphore's capacity had been free at enqueueing time and the
1882 /// exclusive grant got issued right away.
1883 Granted { sem: SP::WeakSyncRcPtr },
1884 /// The future is done: the exclusive grant had been acquired at some time
1885 /// and polled out to the user.
1886 Done,
1887}
1888
1889/// Asynchronous wait for an "exclusive-all" grant on all of an
1890/// [`AsyncSemaphore`]'s capacity.
1891///
1892/// To be obtained through [`AsyncSemaphore::acquire_exclusive_all()`].
1893///
1894/// # Note on lifetime management
1895///
1896/// An [`AsyncSemaphoreExclusiveAllFuture`] instance will only maintain a
1897/// weak reference (i.e. a [`WeakSyncRcPtr`](sync_types::WeakSyncRcPtr)) to the
1898/// associated [`AsyncSemaphore`] instance and thus, would not hinder its
1899/// deallocation. In case the semaphore gets dropped before the future had a
1900/// chance to acquire the exclusive grant on it, its `poll()` would return
1901/// [`AsyncSemaphoreError::StaleSemaphore`].
1902pub struct AsyncSemaphoreExclusiveAllFuture<
1903 ST: sync_types::SyncTypes,
1904 T: marker::Send + marker::Sync,
1905 SP: sync_types::SyncRcPtr<AsyncSemaphore<ST, T>>,
1906> {
1907 private: AsyncSemaphoreExclusiveAllFuturePriv<ST, T, SP>,
1908}
1909
1910impl<ST: sync_types::SyncTypes, T: marker::Send + marker::Sync, SP: sync_types::SyncRcPtr<AsyncSemaphore<ST, T>>>
1911 AsyncSemaphoreExclusiveAllFuture<ST, T, SP>
1912{
1913 /// Obtain the associated [`AsyncSemaphore`].
1914 ///
1915 /// Return the associated [`AsyncSemaphore`] wrapped in `Some` if still
1916 /// alive, `None` otherwise.
1917 pub fn get_semaphore(&self) -> Option<SP> {
1918 let weak_sem = match &self.private {
1919 AsyncSemaphoreExclusiveAllFuturePriv::Enqueued { sem, waiter_id: _ } => sem,
1920 AsyncSemaphoreExclusiveAllFuturePriv::Granted { sem } => sem,
1921 AsyncSemaphoreExclusiveAllFuturePriv::Done => return None,
1922 };
1923 weak_sem.upgrade()
1924 }
1925}
1926
1927impl<ST: sync_types::SyncTypes, T: marker::Send + marker::Sync, SP: sync_types::SyncRcPtr<AsyncSemaphore<ST, T>>>
1928 marker::Unpin for AsyncSemaphoreExclusiveAllFuture<ST, T, SP>
1929{
1930}
1931
1932impl<ST: sync_types::SyncTypes, T: marker::Send + marker::Sync, SP: sync_types::SyncRcPtr<AsyncSemaphore<ST, T>>>
1933 future::Future for AsyncSemaphoreExclusiveAllFuture<ST, T, SP>
1934{
1935 type Output = Result<AsyncSemaphoreExclusiveAllGuard<ST, T, SP>, AsyncSemaphoreError>;
1936
1937 /// Poll for an "exclusive-all" grant on all of the associated
1938 /// [`AsyncSemaphore`]'s capacity.
1939 ///
1940 /// Upon successful future completion, an
1941 /// [`AsyncSemaphoreExclusiveAllGuard`] is returned, otherwise some
1942 /// error information indicating the cause of the failure.
1943 ///
1944 /// The future must not get polled any further once completed.
1945 ///
1946 /// # Errors:
1947 ///
1948 /// * [`AsyncSemaphoreError::StaleSemaphore`] - The associated
1949 /// [`AsyncSemaphore`] has been dropped after the request was made.
1950 fn poll(self: pin::Pin<&mut Self>, cx: &mut task::Context<'_>) -> task::Poll<Self::Output> {
1951 let this = self.get_mut();
1952 match &this.private {
1953 AsyncSemaphoreExclusiveAllFuturePriv::Enqueued { sem, waiter_id } => {
1954 let sem = match sem.upgrade() {
1955 Some(sem) => sem,
1956 None => {
1957 // The semaphore is gone, indicating some teardown going on.
1958 return task::Poll::Ready(Err(AsyncSemaphoreError::StaleSemaphore));
1959 }
1960 };
1961 let mut locked_queue = sem.state.queue.lock();
1962 if locked_queue.poll_waiter(*waiter_id, cx.waker().clone()) {
1963 drop(locked_queue);
1964 this.private = AsyncSemaphoreExclusiveAllFuturePriv::Done;
1965 task::Poll::Ready(Ok(AsyncSemaphoreExclusiveAllGuard::new(sem)))
1966 } else {
1967 task::Poll::Pending
1968 }
1969 }
1970 AsyncSemaphoreExclusiveAllFuturePriv::Granted { sem } => {
1971 let sem = match sem.upgrade() {
1972 Some(sem) => sem,
1973 None => {
1974 // The semaphore is gone, indicating some teardown going on.
1975 return task::Poll::Ready(Err(AsyncSemaphoreError::StaleSemaphore));
1976 }
1977 };
1978 this.private = AsyncSemaphoreExclusiveAllFuturePriv::Done;
1979 task::Poll::Ready(Ok(AsyncSemaphoreExclusiveAllGuard::new(sem)))
1980 }
1981 AsyncSemaphoreExclusiveAllFuturePriv::Done => {
1982 // The semaphore leases had been acquired and handed out already.
1983 task::Poll::Ready(Err(AsyncSemaphoreError::Internal))
1984 }
1985 }
1986 }
1987}
1988
1989impl<ST: sync_types::SyncTypes, T: marker::Send + marker::Sync, SP: sync_types::SyncRcPtr<AsyncSemaphore<ST, T>>> Drop
1990 for AsyncSemaphoreExclusiveAllFuture<ST, T, SP>
1991{
1992 fn drop(&mut self) {
1993 match &self.private {
1994 AsyncSemaphoreExclusiveAllFuturePriv::Enqueued { sem, waiter_id } => {
1995 if let Some(sem) = sem.upgrade() {
1996 sem.state.cancel_waiter(
1997 &mut sem.state.queue.lock(),
1998 *waiter_id,
1999 AsyncSemaphoreLeaseGrantCount::ExclusiveAll,
2000 );
2001 }
2002 }
2003 AsyncSemaphoreExclusiveAllFuturePriv::Granted { sem } => {
2004 // The exclusive grant had been issued right from the beginning, but the
2005 // future never got polled for it. Return the grants.
2006 if let Some(sem) = sem.upgrade() {
2007 sem.state
2008 .return_grant(&mut sem.state.queue.lock(), AsyncSemaphoreLeaseGrantCount::ExclusiveAll);
2009 }
2010 }
2011 AsyncSemaphoreExclusiveAllFuturePriv::Done => (),
2012 }
2013 }
2014}
2015
2016/// RAII guard for an "exclusive-all" grant acquired from an [`AsyncSemaphore`].
2017///
2018/// To be obtained either synchronously through
2019/// [`AsyncSemaphore::try_acquire_exclusive_all()`] or asynchronously via
2020/// [AsyncSemaphore::acquire_exclusive_all()].
2021///
2022/// Provides mutable access to the associated [`AsyncSemaphore`]'s wrapped
2023/// data item.
2024///
2025/// Until an [`AsyncSemaphoreExclusiveAllGuard`] gets dropped again, it will
2026/// block any other request, including ones for a zero-sized "trivial" lease,
2027/// from making progress.
2028pub struct AsyncSemaphoreExclusiveAllGuard<
2029 ST: sync_types::SyncTypes,
2030 T: marker::Send + marker::Sync,
2031 SP: sync_types::SyncRcPtr<AsyncSemaphore<ST, T>>,
2032> {
2033 sem: Option<SP>,
2034 _phantom: marker::PhantomData<fn() -> (ST, T)>,
2035}
2036
2037impl<ST: sync_types::SyncTypes, T: marker::Send + marker::Sync, SP: sync_types::SyncRcPtr<AsyncSemaphore<ST, T>>>
2038 AsyncSemaphoreExclusiveAllGuard<ST, T, SP>
2039{
2040 fn new(sem: SP) -> Self {
2041 Self {
2042 sem: Some(sem),
2043 _phantom: marker::PhantomData,
2044 }
2045 }
2046
2047 /// Obtain the associated [`AsyncSemaphore`].
2048 pub fn get_semaphore(&self) -> SP::SyncRcPtrRef<'_> {
2049 self.sem.as_ref().unwrap().as_ref()
2050 }
2051
2052 /// Release the "exclusive-all" lease grant and return the associated
2053 /// [`AsyncSemaphore`].
2054 pub fn into_semaphore(mut self) -> SP {
2055 let sem = self.sem.take().unwrap();
2056 Self::_drop(&sem);
2057 sem
2058 }
2059
2060 /// Return the associated [`AsyncSemaphore`] the "exclusive-all" grant had
2061 /// been issued for.
2062 pub fn semaphore(&self) -> &AsyncSemaphore<ST, T> {
2063 self.sem.as_ref().unwrap()
2064 }
2065
2066 /// Grow or shrink the associated [`AsyncSemaphore`]'s capacity.
2067 ///
2068 /// Independent of whether the capacity is being in- or decreased,
2069 /// all of it will still be exclusively owned by `self`.
2070 ///
2071 /// Note that shrinking the capacity might potentially make some other,
2072 /// still enqueued waiters to fail with
2073 /// [`RequestExceedsSemaphoreCapacity`](AsyncSemaphoreError::RequestExceedsSemaphoreCapacity)
2074 /// in case their number of leases requested would now exceed the
2075 /// remaining capacity.
2076 ///
2077 /// # Arguments:
2078 ///
2079 /// * `max_leases` - The new [`AsyncSemaphore`] capacity. Must be
2080 /// **strictly** less than [`usize::MAX`].
2081 pub fn resize_semaphore(&mut self, max_leases: usize) -> Result<(), AsyncSemaphoreNewError> {
2082 if max_leases == usize::MAX {
2083 return Err(AsyncSemaphoreNewError::UnsupportedCapacity);
2084 }
2085
2086 let sem = self.sem.as_ref().unwrap();
2087 let mut locked_queue = sem.state.queue.lock();
2088 let capacity_shrunken = max_leases < locked_queue.max_leases;
2089 locked_queue.max_leases = max_leases;
2090 sem.state
2091 .leases_granted
2092 .store(max_leases + 1, atomic::Ordering::Relaxed);
2093 // Shrinking the capacity might render some pending requests unsatisfiable.
2094 if capacity_shrunken {
2095 sem.state.wake_failed_waiters(&mut locked_queue);
2096 }
2097
2098 Ok(())
2099 }
2100
2101 /// Downgrade the "exclusive-all" grant on all of a semaphore's capacity to
2102 /// a regular leases grant.
2103 ///
2104 /// The "exclusive-all" grant will be downgraded to a regular grant on all
2105 /// of the associated [`AsyncSemaphore`]'s capacity. This may e.g. be used
2106 /// in case the full grant needs to be
2107 /// [split](AsyncSemaphoreLeasesGuard::split_leases) into smaller ones.
2108 pub fn downgrade(mut self) -> AsyncSemaphoreLeasesGuard<ST, T, SP> {
2109 let sem = self.sem.take().unwrap();
2110 let mut locked_queue = sem.state.queue.lock();
2111 let leases_granted = locked_queue.max_leases;
2112 sem.state
2113 .leases_granted
2114 .store(leases_granted, atomic::Ordering::Relaxed);
2115 if locked_queue.max_leases == 0 {
2116 sem.state.trivial_leases_granted.fetch_add(1, atomic::Ordering::Release);
2117 }
2118 // Turning the ExclusiveAll grant into a regular one might
2119 // unblock TrivialLease waiters.
2120 if locked_queue.has_trivial_lease_waiters() {
2121 sem.state.wake_completed_waiters(&mut locked_queue);
2122 }
2123 drop(locked_queue);
2124 AsyncSemaphoreLeasesGuard::new(sem, leases_granted)
2125 }
2126
2127 /// Convert into an [`AsyncSemaphoreExclusiveAllWeakGuard`] not hindering
2128 /// destruction of the associated [`AsyncSemaphore`].
2129 pub fn into_weak(mut self) -> AsyncSemaphoreExclusiveAllWeakGuard<ST, T, SP> {
2130 let sem = self.sem.take().unwrap();
2131 AsyncSemaphoreExclusiveAllWeakGuard {
2132 sem: Some(sem.downgrade()),
2133 _phantom: marker::PhantomData,
2134 }
2135 }
2136
2137 /// Simultaneously obtain an immutable reference to the associated
2138 /// [`AsyncSemaphore`] as well as a mutable one to its
2139 /// wrapped inner value.
2140 ///
2141 /// Mutably dereferencing the [`AsyncSemaphoreExclusiveAllGuard`] to obtain
2142 /// a `mut` reference on the protected value would result in a borrow on
2143 /// `self` and thus, prohibits any use of
2144 /// [`get_semaphore()`](Self::get_semaphore) over the course of the
2145 /// reference's lifetime.
2146 ///
2147 /// However, due to the [`AsyncSemaphoreExclusiveAllGuard`]'s locking
2148 /// semantics, there isn't anything problematic about immutably
2149 /// referencing the associated [`AsyncSemaphore`] at the same time. It's
2150 /// useful in e.g. situations where the semaphore lives in some
2151 /// zero-overhead [`SyncRcPtrForInner`](sync_types::SyncRcPtrForInner)
2152 /// and one seeks to obtain a reference to the container starting out
2153 /// from the semaphore guard.
2154 pub fn borrow_outer_inner_mut<'a>(&'a mut self) -> (SP::SyncRcPtrRef<'a>, &'a mut T) {
2155 let p = self.sem.as_ref().unwrap().as_ref();
2156 let pdata = p.data.get();
2157 (p, unsafe { &mut *pdata })
2158 }
2159
2160 /// Return the "exclusive-all" grant to the the [`AsyncSemaphore`], thereby
2161 /// waking other waiters, if any.
2162 fn _drop(sem: &AsyncSemaphore<ST, T>) {
2163 sem.state
2164 .return_grant(&mut sem.state.queue.lock(), AsyncSemaphoreLeaseGrantCount::ExclusiveAll);
2165 }
2166}
2167
2168impl<ST: sync_types::SyncTypes, T: marker::Send + marker::Sync, SP: sync_types::SyncRcPtr<AsyncSemaphore<ST, T>>> Drop
2169 for AsyncSemaphoreExclusiveAllGuard<ST, T, SP>
2170{
2171 fn drop(&mut self) {
2172 if let Some(sem) = self.sem.as_ref() {
2173 Self::_drop(sem);
2174 }
2175 }
2176}
2177
2178impl<ST: sync_types::SyncTypes, T: marker::Send + marker::Sync, SP: sync_types::SyncRcPtr<AsyncSemaphore<ST, T>>>
2179 ops::Deref for AsyncSemaphoreExclusiveAllGuard<ST, T, SP>
2180{
2181 type Target = T;
2182
2183 fn deref(&self) -> &Self::Target {
2184 let p = self.sem.as_ref().unwrap().data.get() as *const T;
2185 unsafe { &*p }
2186 }
2187}
2188
2189impl<ST: sync_types::SyncTypes, T: marker::Send + marker::Sync, SP: sync_types::SyncRcPtr<AsyncSemaphore<ST, T>>>
2190 ops::DerefMut for AsyncSemaphoreExclusiveAllGuard<ST, T, SP>
2191{
2192 fn deref_mut(&mut self) -> &mut Self::Target {
2193 let p = self.sem.as_ref().unwrap().data.get();
2194 unsafe { &mut *p }
2195 }
2196}
2197/// Weak variant of [`AsyncSemaphoreExclusiveAllGuard`] not hindering
2198/// destruction of the associated [`AsyncSemaphore`].
2199///
2200/// In cases were it's desired that long-living "exclusive-all" grants don't
2201/// prevent a destruction of the associated [`AsyncSemaphore`], or to break
2202/// cycles, a `AsyncSemaphoreExclusiveAllWeakGuard` may be used. To be obtained
2203/// via [`AsyncSemaphoreExclusiveAllGuard::into_weak()`].
2204///
2205/// An `AsyncSemaphoreExclusiveAllWeakGuard` may get converted back into a full
2206/// [`AsyncSemaphoreExclusiveAllGuard`] via [`upgrade()`](Self::upgrade).
2207pub struct AsyncSemaphoreExclusiveAllWeakGuard<
2208 ST: sync_types::SyncTypes,
2209 T: marker::Send + marker::Sync,
2210 SP: sync_types::SyncRcPtr<AsyncSemaphore<ST, T>>,
2211> {
2212 sem: Option<SP::WeakSyncRcPtr>,
2213 _phantom: marker::PhantomData<fn() -> (ST, T)>,
2214}
2215
2216impl<ST: sync_types::SyncTypes, T: marker::Send + marker::Sync, SP: sync_types::SyncRcPtr<AsyncSemaphore<ST, T>>>
2217 AsyncSemaphoreExclusiveAllWeakGuard<ST, T, SP>
2218{
2219 /// Attempt to convert back into an [`AsyncSemaphoreExclusiveAllGuard`].
2220 ///
2221 /// Returns the [`AsyncSemaphoreExclusiveAllGuard`] wrapped in `Some` if the
2222 /// associated [`AsyncSemaphore`] is still alive, `None otherwise.
2223 pub fn upgrade(mut self) -> Option<AsyncSemaphoreExclusiveAllGuard<ST, T, SP>> {
2224 self.sem
2225 .take()
2226 .unwrap()
2227 .upgrade()
2228 .map(|sem| AsyncSemaphoreExclusiveAllGuard {
2229 sem: Some(sem),
2230 _phantom: marker::PhantomData,
2231 })
2232 }
2233}
2234
2235impl<ST: sync_types::SyncTypes, T: marker::Send + marker::Sync, SP: sync_types::SyncRcPtr<AsyncSemaphore<ST, T>>> Drop
2236 for AsyncSemaphoreExclusiveAllWeakGuard<ST, T, SP>
2237{
2238 fn drop(&mut self) {
2239 if let Some(sem) = self.sem.take() {
2240 if let Some(sem) = sem.upgrade() {
2241 drop(AsyncSemaphoreExclusiveAllGuard {
2242 sem: Some(sem),
2243 _phantom: marker::PhantomData,
2244 });
2245 }
2246 }
2247 }
2248}
2249
2250#[test]
2251fn test_async_semaphore_lease_vs_lease() {
2252 use crate::sync_types::SyncRcPtr as _;
2253 use crate::test::{TestAsyncExecutor, TestNopSyncTypes};
2254 use core::mem;
2255 use ops::DerefMut as _;
2256
2257 type TestAsyncSemaphore = AsyncSemaphore<TestNopSyncTypes, ()>;
2258 type TestAsyncSemaporeSyncRcPtr =
2259 <<TestNopSyncTypes as sync_types::SyncTypes>::SyncRcPtrFactory as sync_types::SyncRcPtrFactory>::SyncRcPtr<
2260 TestAsyncSemaphore,
2261 >;
2262
2263 enum TestFuture {
2264 ExpectLease0Ready {
2265 sem: TestAsyncSemaporeSyncRcPtr,
2266 lease_fut0: AsyncSemaphoreLeasesFuture<TestNopSyncTypes, (), TestAsyncSemaporeSyncRcPtr>,
2267 },
2268 ExpectLease1Ready {
2269 sem: TestAsyncSemaporeSyncRcPtr,
2270 lease_fut1: AsyncSemaphoreLeasesFuture<TestNopSyncTypes, (), TestAsyncSemaporeSyncRcPtr>,
2271 },
2272 ExpectLeases23Ready {
2273 sem: TestAsyncSemaporeSyncRcPtr,
2274 lease_fut2: AsyncSemaphoreLeasesFuture<TestNopSyncTypes, (), TestAsyncSemaporeSyncRcPtr>,
2275 lease_fut3: AsyncSemaphoreLeasesFuture<TestNopSyncTypes, (), TestAsyncSemaporeSyncRcPtr>,
2276 },
2277 Done,
2278 }
2279
2280 impl future::Future for TestFuture {
2281 type Output = ();
2282
2283 fn poll(mut self: pin::Pin<&mut Self>, cx: &mut task::Context<'_>) -> task::Poll<Self::Output> {
2284 let this = mem::replace(self.deref_mut(), Self::Done);
2285 match this {
2286 Self::ExpectLease0Ready { sem, mut lease_fut0 } => {
2287 assert!(matches!(AsyncSemaphore::try_acquire_leases(&sem.as_ref(), 1), Ok(None)));
2288
2289 let mut lease_fut1 = AsyncSemaphore::acquire_leases(&sem.as_ref(), 2).unwrap();
2290 assert!(matches!(pin::Pin::new(&mut lease_fut1).poll(cx), task::Poll::Pending));
2291
2292 let _lease0 = match pin::Pin::new(&mut lease_fut0).poll(cx) {
2293 task::Poll::Ready(guard) => guard.unwrap(),
2294 task::Poll::Pending => unreachable!(),
2295 };
2296
2297 assert!(matches!(AsyncSemaphore::try_acquire_leases(&sem.as_ref(), 1), Ok(None)));
2298 assert!(matches!(pin::Pin::new(&mut lease_fut1).poll(cx), task::Poll::Pending));
2299
2300 // Dropping the lease0 guard is supposed to wake lease_fut1.
2301 *self.deref_mut() = Self::ExpectLease1Ready { sem, lease_fut1 };
2302 task::Poll::Pending
2303 }
2304 Self::ExpectLease1Ready { sem, mut lease_fut1 } => {
2305 assert!(matches!(AsyncSemaphore::try_acquire_leases(&sem.as_ref(), 1), Ok(None)));
2306
2307 let _lease1 = match pin::Pin::new(&mut lease_fut1).poll(cx) {
2308 task::Poll::Ready(guard) => guard.unwrap(),
2309 task::Poll::Pending => unreachable!(),
2310 };
2311
2312 assert!(matches!(AsyncSemaphore::try_acquire_leases(&sem.as_ref(), 1), Ok(None)));
2313 let mut lease_fut2 = AsyncSemaphore::acquire_leases(&sem.as_ref(), 1).unwrap();
2314 assert!(matches!(pin::Pin::new(&mut lease_fut2).poll(cx), task::Poll::Pending));
2315 let mut lease_fut3 = AsyncSemaphore::acquire_leases(&sem.as_ref(), 1).unwrap();
2316 assert!(matches!(pin::Pin::new(&mut lease_fut3).poll(cx), task::Poll::Pending));
2317
2318 // Dropping the lease1 guard is supposed to wake lease_fut2 and lease_fut3.
2319 *self.deref_mut() = Self::ExpectLeases23Ready {
2320 sem,
2321 lease_fut2,
2322 lease_fut3,
2323 };
2324 task::Poll::Pending
2325 }
2326 Self::ExpectLeases23Ready {
2327 sem,
2328 mut lease_fut2,
2329 mut lease_fut3,
2330 } => {
2331 assert!(matches!(AsyncSemaphore::try_acquire_leases(&sem.as_ref(), 1), Ok(None)));
2332
2333 let lease2 = match pin::Pin::new(&mut lease_fut2).poll(cx) {
2334 task::Poll::Ready(guard) => guard.unwrap(),
2335 task::Poll::Pending => unreachable!(),
2336 };
2337 let lease3 = match pin::Pin::new(&mut lease_fut3).poll(cx) {
2338 task::Poll::Ready(guard) => guard.unwrap(),
2339 task::Poll::Pending => unreachable!(),
2340 };
2341
2342 assert!(matches!(AsyncSemaphore::try_acquire_leases(&sem.as_ref(), 1), Ok(None)));
2343 drop(lease2);
2344 let lease4 = AsyncSemaphore::try_acquire_leases(&sem.as_ref(), 1).unwrap().unwrap();
2345
2346 assert!(matches!(AsyncSemaphore::try_acquire_leases(&sem.as_ref(), 1), Ok(None)));
2347 drop(lease3);
2348 let lease5 = AsyncSemaphore::try_acquire_leases(&sem.as_ref(), 1).unwrap().unwrap();
2349
2350 assert!(matches!(AsyncSemaphore::try_acquire_leases(&sem.as_ref(), 2), Ok(None)));
2351 drop(lease4);
2352 assert!(matches!(AsyncSemaphore::try_acquire_leases(&sem.as_ref(), 2), Ok(None)));
2353 drop(lease5);
2354 assert!(matches!(
2355 AsyncSemaphore::try_acquire_leases(&sem.as_ref(), 2),
2356 Ok(Some(..))
2357 ));
2358
2359 let _lease6 = AsyncSemaphore::try_acquire_leases(&sem.as_ref(), 1).unwrap().unwrap();
2360 let mut lease_fut7 = AsyncSemaphore::acquire_leases(&sem.as_ref(), 2).unwrap();
2361 assert!(matches!(pin::Pin::new(&mut lease_fut7).poll(cx), task::Poll::Pending));
2362 let mut lease_fut8 = AsyncSemaphore::acquire_leases(&sem.as_ref(), 1).unwrap();
2363 assert!(matches!(pin::Pin::new(&mut lease_fut8).poll(cx), task::Poll::Pending));
2364 drop(lease_fut7);
2365 let _lease8 = match pin::Pin::new(&mut lease_fut8).poll(cx) {
2366 task::Poll::Ready(guard) => guard.unwrap(),
2367 task::Poll::Pending => unreachable!(),
2368 };
2369
2370 task::Poll::Ready(())
2371 }
2372 Self::Done => unreachable!(),
2373 }
2374 }
2375 }
2376
2377 impl Unpin for TestFuture {}
2378
2379 let e = TestAsyncExecutor::new();
2380 let sem = <<TestNopSyncTypes as sync_types::SyncTypes>::SyncRcPtrFactory as sync_types::SyncRcPtrFactory>::try_new(
2381 TestAsyncSemaphore::new(2, ()).unwrap(),
2382 )
2383 .unwrap();
2384 let lease_fut0 = AsyncSemaphore::acquire_leases(&sem.as_ref(), 2).unwrap();
2385 let w = TestAsyncExecutor::spawn(&e, TestFuture::ExpectLease0Ready { sem, lease_fut0 });
2386 TestAsyncExecutor::run_to_completion(&e);
2387 w.take().unwrap();
2388}
2389
2390#[test]
2391fn test_async_semaphore_lease_vs_trivial() {
2392 use crate::sync_types::SyncRcPtr as _;
2393 use crate::test::{TestAsyncExecutor, TestNopSyncTypes};
2394 use core::mem;
2395 use ops::DerefMut as _;
2396
2397 type TestAsyncSemaphore = AsyncSemaphore<TestNopSyncTypes, ()>;
2398 type TestAsyncSemaporeSyncRcPtr =
2399 <<TestNopSyncTypes as sync_types::SyncTypes>::SyncRcPtrFactory as sync_types::SyncRcPtrFactory>::SyncRcPtr<
2400 TestAsyncSemaphore,
2401 >;
2402
2403 enum TestFuture {
2404 ExpectLease0Ready {
2405 sem: TestAsyncSemaporeSyncRcPtr,
2406 lease_fut0: AsyncSemaphoreLeasesFuture<TestNopSyncTypes, (), TestAsyncSemaporeSyncRcPtr>,
2407 },
2408 ExpectLease1Ready {
2409 sem: TestAsyncSemaporeSyncRcPtr,
2410 trivial_lease: AsyncSemaphoreLeasesGuard<TestNopSyncTypes, (), TestAsyncSemaporeSyncRcPtr>,
2411 lease_fut1: AsyncSemaphoreLeasesFuture<TestNopSyncTypes, (), TestAsyncSemaporeSyncRcPtr>,
2412 },
2413 Done,
2414 }
2415
2416 impl future::Future for TestFuture {
2417 type Output = ();
2418
2419 fn poll(mut self: pin::Pin<&mut Self>, cx: &mut task::Context<'_>) -> task::Poll<Self::Output> {
2420 let this = mem::replace(self.deref_mut(), Self::Done);
2421 match this {
2422 Self::ExpectLease0Ready { sem, mut lease_fut0 } => {
2423 assert!(matches!(AsyncSemaphore::try_acquire_leases(&sem.as_ref(), 0), Ok(..)));
2424 let mut trivial_fut = AsyncSemaphore::acquire_leases(&sem.as_ref(), 0).unwrap();
2425 match pin::Pin::new(&mut trivial_fut).poll(cx) {
2426 task::Poll::Ready(guard) => guard.unwrap(),
2427 task::Poll::Pending => unreachable!(),
2428 };
2429
2430 let _lease0 = match pin::Pin::new(&mut lease_fut0).poll(cx) {
2431 task::Poll::Ready(guard) => guard.unwrap(),
2432 task::Poll::Pending => unreachable!(),
2433 };
2434
2435 assert!(matches!(AsyncSemaphore::try_acquire_leases(&sem.as_ref(), 0), Ok(..)));
2436 let mut trivial_fut = AsyncSemaphore::acquire_leases(&sem.as_ref(), 0).unwrap();
2437 let trivial_lease = match pin::Pin::new(&mut trivial_fut).poll(cx) {
2438 task::Poll::Ready(guard) => guard.unwrap(),
2439 task::Poll::Pending => unreachable!(),
2440 };
2441
2442 assert!(matches!(AsyncSemaphore::try_acquire_leases(&sem.as_ref(), 0), Ok(..)));
2443 let mut trivial_fut = AsyncSemaphore::acquire_leases(&sem.as_ref(), 0).unwrap();
2444 match pin::Pin::new(&mut trivial_fut).poll(cx) {
2445 task::Poll::Ready(guard) => guard.unwrap(),
2446 task::Poll::Pending => unreachable!(),
2447 };
2448
2449 let mut lease_fut1 = AsyncSemaphore::acquire_leases(&sem.as_ref(), 1).unwrap();
2450 assert!(matches!(pin::Pin::new(&mut lease_fut1).poll(cx), task::Poll::Pending));
2451
2452 // Dropping the lease0 guard is supposed to wake lease_fut1.
2453 *self.deref_mut() = Self::ExpectLease1Ready {
2454 sem,
2455 trivial_lease,
2456 lease_fut1,
2457 };
2458 task::Poll::Pending
2459 }
2460 Self::ExpectLease1Ready {
2461 sem,
2462 trivial_lease: _trivial_lease,
2463 mut lease_fut1,
2464 } => {
2465 let lease1 = match pin::Pin::new(&mut lease_fut1).poll(cx) {
2466 task::Poll::Ready(guard) => guard.unwrap(),
2467 task::Poll::Pending => unreachable!(),
2468 };
2469 drop(lease1);
2470 assert!(matches!(
2471 AsyncSemaphore::try_acquire_leases(&sem.as_ref(), 1),
2472 Ok(Some(..))
2473 ));
2474 task::Poll::Ready(())
2475 }
2476 Self::Done => unreachable!(),
2477 }
2478 }
2479 }
2480
2481 impl Unpin for TestFuture {}
2482
2483 let e = TestAsyncExecutor::new();
2484 let sem = <<TestNopSyncTypes as sync_types::SyncTypes>::SyncRcPtrFactory as sync_types::SyncRcPtrFactory>::try_new(
2485 TestAsyncSemaphore::new(1, ()).unwrap(),
2486 )
2487 .unwrap();
2488 let lease_fut0 = AsyncSemaphore::acquire_leases(&sem.as_ref(), 1).unwrap();
2489 let w = TestAsyncExecutor::spawn(&e, TestFuture::ExpectLease0Ready { sem, lease_fut0 });
2490 TestAsyncExecutor::run_to_completion(&e);
2491 w.take().unwrap();
2492}
2493
2494#[test]
2495fn test_async_semaphore_exclusive_vs_exclusive() {
2496 use crate::sync_types::SyncRcPtr as _;
2497 use crate::test::{TestAsyncExecutor, TestNopSyncTypes};
2498 use core::mem;
2499 use ops::DerefMut as _;
2500
2501 type TestAsyncSemaphore = AsyncSemaphore<TestNopSyncTypes, ()>;
2502 type TestAsyncSemaporeSyncRcPtr =
2503 <<TestNopSyncTypes as sync_types::SyncTypes>::SyncRcPtrFactory as sync_types::SyncRcPtrFactory>::SyncRcPtr<
2504 TestAsyncSemaphore,
2505 >;
2506
2507 enum TestFuture {
2508 ExpectExclusiveAll0Ready {
2509 sem: TestAsyncSemaporeSyncRcPtr,
2510 excl_fut0: AsyncSemaphoreExclusiveAllFuture<TestNopSyncTypes, (), TestAsyncSemaporeSyncRcPtr>,
2511 },
2512 ExpectExclusiveAll1Ready {
2513 sem: TestAsyncSemaporeSyncRcPtr,
2514 excl_fut1: AsyncSemaphoreExclusiveAllFuture<TestNopSyncTypes, (), TestAsyncSemaporeSyncRcPtr>,
2515 },
2516 Done,
2517 }
2518
2519 impl future::Future for TestFuture {
2520 type Output = ();
2521
2522 fn poll(mut self: pin::Pin<&mut Self>, cx: &mut task::Context<'_>) -> task::Poll<Self::Output> {
2523 let this = mem::replace(self.deref_mut(), Self::Done);
2524 match this {
2525 Self::ExpectExclusiveAll0Ready { sem, mut excl_fut0 } => {
2526 assert!(AsyncSemaphore::try_acquire_exclusive_all(&sem.as_ref()).is_none());
2527
2528 let mut excl_fut1 = AsyncSemaphore::acquire_exclusive_all(&sem.as_ref()).unwrap();
2529 assert!(matches!(pin::Pin::new(&mut excl_fut1).poll(cx), task::Poll::Pending));
2530
2531 let _excl0 = match pin::Pin::new(&mut excl_fut0).poll(cx) {
2532 task::Poll::Ready(guard) => guard.unwrap(),
2533 task::Poll::Pending => unreachable!(),
2534 };
2535
2536 assert!(matches!(pin::Pin::new(&mut excl_fut1).poll(cx), task::Poll::Pending));
2537
2538 *self.deref_mut() = Self::ExpectExclusiveAll1Ready { sem, excl_fut1 };
2539 task::Poll::Pending
2540 }
2541 Self::ExpectExclusiveAll1Ready { sem, mut excl_fut1 } => {
2542 assert!(AsyncSemaphore::try_acquire_exclusive_all(&sem.as_ref()).is_none());
2543
2544 let excl1 = match pin::Pin::new(&mut excl_fut1).poll(cx) {
2545 task::Poll::Ready(guard) => guard.unwrap(),
2546 task::Poll::Pending => unreachable!(),
2547 };
2548
2549 assert!(AsyncSemaphore::try_acquire_exclusive_all(&sem.as_ref()).is_none());
2550 drop(excl1);
2551 assert!(AsyncSemaphore::try_acquire_exclusive_all(&sem.as_ref()).is_some());
2552
2553 task::Poll::Ready(())
2554 }
2555 Self::Done => unreachable!(),
2556 }
2557 }
2558 }
2559
2560 impl Unpin for TestFuture {}
2561
2562 let e = TestAsyncExecutor::new();
2563 let sem = <<TestNopSyncTypes as sync_types::SyncTypes>::SyncRcPtrFactory as sync_types::SyncRcPtrFactory>::try_new(
2564 TestAsyncSemaphore::new(0, ()).unwrap(),
2565 )
2566 .unwrap();
2567 let excl_fut0 = AsyncSemaphore::acquire_exclusive_all(&sem.as_ref()).unwrap();
2568 let w = TestAsyncExecutor::spawn(&e, TestFuture::ExpectExclusiveAll0Ready { sem, excl_fut0 });
2569 TestAsyncExecutor::run_to_completion(&e);
2570 w.take().unwrap();
2571}
2572
2573#[test]
2574fn test_async_semaphore_exclusive_vs_lease() {
2575 use crate::sync_types::SyncRcPtr as _;
2576 use crate::test::{TestAsyncExecutor, TestNopSyncTypes};
2577 use core::mem;
2578 use ops::DerefMut as _;
2579
2580 type TestAsyncSemaphore = AsyncSemaphore<TestNopSyncTypes, ()>;
2581 type TestAsyncSemaporeSyncRcPtr =
2582 <<TestNopSyncTypes as sync_types::SyncTypes>::SyncRcPtrFactory as sync_types::SyncRcPtrFactory>::SyncRcPtr<
2583 TestAsyncSemaphore,
2584 >;
2585
2586 enum TestFuture {
2587 ExpectExclusiveAll0Ready {
2588 sem: TestAsyncSemaporeSyncRcPtr,
2589 excl_fut0: AsyncSemaphoreExclusiveAllFuture<TestNopSyncTypes, (), TestAsyncSemaporeSyncRcPtr>,
2590 },
2591 ExpectLease1Ready {
2592 sem: TestAsyncSemaporeSyncRcPtr,
2593 lease_fut1: AsyncSemaphoreLeasesFuture<TestNopSyncTypes, (), TestAsyncSemaporeSyncRcPtr>,
2594 },
2595 ExpectExclusiveAll2Ready {
2596 sem: TestAsyncSemaporeSyncRcPtr,
2597 excl_fut2: AsyncSemaphoreExclusiveAllFuture<TestNopSyncTypes, (), TestAsyncSemaporeSyncRcPtr>,
2598 },
2599 Done,
2600 }
2601
2602 impl future::Future for TestFuture {
2603 type Output = ();
2604
2605 fn poll(mut self: pin::Pin<&mut Self>, cx: &mut task::Context<'_>) -> task::Poll<Self::Output> {
2606 let this = mem::replace(self.deref_mut(), Self::Done);
2607 match this {
2608 Self::ExpectExclusiveAll0Ready { sem, mut excl_fut0 } => {
2609 assert!(matches!(AsyncSemaphore::try_acquire_leases(&sem.as_ref(), 1), Ok(None)));
2610
2611 let mut lease_fut1 = AsyncSemaphore::acquire_leases(&sem.as_ref(), 1).unwrap();
2612 assert!(matches!(pin::Pin::new(&mut lease_fut1).poll(cx), task::Poll::Pending));
2613
2614 let _excl0 = match pin::Pin::new(&mut excl_fut0).poll(cx) {
2615 task::Poll::Ready(guard) => guard.unwrap(),
2616 task::Poll::Pending => unreachable!(),
2617 };
2618
2619 assert!(matches!(AsyncSemaphore::try_acquire_leases(&sem.as_ref(), 1), Ok(None)));
2620 assert!(matches!(pin::Pin::new(&mut lease_fut1).poll(cx), task::Poll::Pending));
2621
2622 *self.deref_mut() = Self::ExpectLease1Ready { sem, lease_fut1 };
2623 task::Poll::Pending
2624 }
2625 Self::ExpectLease1Ready { sem, mut lease_fut1 } => {
2626 assert!(AsyncSemaphore::try_acquire_exclusive_all(&sem.as_ref()).is_none());
2627
2628 let mut excl_fut2 = AsyncSemaphore::acquire_exclusive_all(&sem.as_ref()).unwrap();
2629 assert!(matches!(pin::Pin::new(&mut excl_fut2).poll(cx), task::Poll::Pending));
2630
2631 let _lease1 = match pin::Pin::new(&mut lease_fut1).poll(cx) {
2632 task::Poll::Ready(guard) => guard.unwrap(),
2633 task::Poll::Pending => unreachable!(),
2634 };
2635 assert!(AsyncSemaphore::try_acquire_exclusive_all(&sem.as_ref()).is_none());
2636 assert!(matches!(pin::Pin::new(&mut excl_fut2).poll(cx), task::Poll::Pending));
2637
2638 *self.deref_mut() = Self::ExpectExclusiveAll2Ready { sem, excl_fut2 };
2639 task::Poll::Pending
2640 }
2641 Self::ExpectExclusiveAll2Ready { sem, mut excl_fut2 } => {
2642 assert!(matches!(AsyncSemaphore::try_acquire_leases(&sem.as_ref(), 1), Ok(None)));
2643
2644 let excl2 = match pin::Pin::new(&mut excl_fut2).poll(cx) {
2645 task::Poll::Ready(guard) => guard.unwrap(),
2646 task::Poll::Pending => unreachable!(),
2647 };
2648
2649 assert!(matches!(AsyncSemaphore::try_acquire_leases(&sem.as_ref(), 1), Ok(None)));
2650 drop(excl2);
2651 assert!(matches!(
2652 AsyncSemaphore::try_acquire_leases(&sem.as_ref(), 1),
2653 Ok(Some(..))
2654 ));
2655 task::Poll::Ready(())
2656 }
2657 Self::Done => unreachable!(),
2658 }
2659 }
2660 }
2661
2662 impl Unpin for TestFuture {}
2663
2664 let e = TestAsyncExecutor::new();
2665 let sem = <<TestNopSyncTypes as sync_types::SyncTypes>::SyncRcPtrFactory as sync_types::SyncRcPtrFactory>::try_new(
2666 TestAsyncSemaphore::new(1, ()).unwrap(),
2667 )
2668 .unwrap();
2669 let excl_fut0 = AsyncSemaphore::acquire_exclusive_all(&sem.as_ref()).unwrap();
2670 let w = TestAsyncExecutor::spawn(&e, TestFuture::ExpectExclusiveAll0Ready { sem, excl_fut0 });
2671 TestAsyncExecutor::run_to_completion(&e);
2672 w.take().unwrap();
2673}
2674
2675#[test]
2676fn test_async_semaphore_exclusive_vs_trivial() {
2677 use crate::sync_types::SyncRcPtr as _;
2678 use crate::test::{TestAsyncExecutor, TestNopSyncTypes};
2679 use core::mem;
2680 use ops::DerefMut as _;
2681
2682 type TestAsyncSemaphore = AsyncSemaphore<TestNopSyncTypes, ()>;
2683 type TestAsyncSemaporeSyncRcPtr =
2684 <<TestNopSyncTypes as sync_types::SyncTypes>::SyncRcPtrFactory as sync_types::SyncRcPtrFactory>::SyncRcPtr<
2685 TestAsyncSemaphore,
2686 >;
2687
2688 enum TestFuture {
2689 ExpectExclusiveAll0Ready {
2690 sem: TestAsyncSemaporeSyncRcPtr,
2691 excl_fut0: AsyncSemaphoreExclusiveAllFuture<TestNopSyncTypes, (), TestAsyncSemaporeSyncRcPtr>,
2692 },
2693 ExpectTrivial1Ready {
2694 sem: TestAsyncSemaporeSyncRcPtr,
2695 trivial_fut1: AsyncSemaphoreLeasesFuture<TestNopSyncTypes, (), TestAsyncSemaporeSyncRcPtr>,
2696 },
2697 ExpectExclusiveAll2Ready {
2698 sem: TestAsyncSemaporeSyncRcPtr,
2699 excl_fut2: AsyncSemaphoreExclusiveAllFuture<TestNopSyncTypes, (), TestAsyncSemaporeSyncRcPtr>,
2700 },
2701 Done,
2702 }
2703
2704 impl future::Future for TestFuture {
2705 type Output = ();
2706
2707 fn poll(mut self: pin::Pin<&mut Self>, cx: &mut task::Context<'_>) -> task::Poll<Self::Output> {
2708 let this = mem::replace(self.deref_mut(), Self::Done);
2709 match this {
2710 Self::ExpectExclusiveAll0Ready { sem, mut excl_fut0 } => {
2711 assert!(matches!(AsyncSemaphore::try_acquire_leases(&sem.as_ref(), 0), Ok(None)));
2712
2713 let mut trivial_fut1 = AsyncSemaphore::acquire_leases(&sem.as_ref(), 0).unwrap();
2714 assert!(matches!(pin::Pin::new(&mut trivial_fut1).poll(cx), task::Poll::Pending));
2715
2716 let _excl0 = match pin::Pin::new(&mut excl_fut0).poll(cx) {
2717 task::Poll::Ready(guard) => guard.unwrap(),
2718 task::Poll::Pending => unreachable!(),
2719 };
2720
2721 assert!(matches!(AsyncSemaphore::try_acquire_leases(&sem.as_ref(), 0), Ok(None)));
2722 assert!(matches!(pin::Pin::new(&mut trivial_fut1).poll(cx), task::Poll::Pending));
2723
2724 *self.deref_mut() = Self::ExpectTrivial1Ready { sem, trivial_fut1 };
2725 task::Poll::Pending
2726 }
2727 Self::ExpectTrivial1Ready { sem, mut trivial_fut1 } => {
2728 assert!(AsyncSemaphore::try_acquire_exclusive_all(&sem.as_ref()).is_none());
2729
2730 let mut excl_fut2 = AsyncSemaphore::acquire_exclusive_all(&sem.as_ref()).unwrap();
2731 assert!(matches!(pin::Pin::new(&mut excl_fut2).poll(cx), task::Poll::Pending));
2732
2733 let mut trivial1 = match pin::Pin::new(&mut trivial_fut1).poll(cx) {
2734 task::Poll::Ready(guard) => guard.unwrap(),
2735 task::Poll::Pending => unreachable!(),
2736 };
2737
2738 let _trivial = trivial1.split_leases(0).unwrap();
2739 drop(trivial1);
2740
2741 assert!(AsyncSemaphore::try_acquire_exclusive_all(&sem.as_ref()).is_none());
2742 assert!(matches!(pin::Pin::new(&mut excl_fut2).poll(cx), task::Poll::Pending));
2743
2744 *self.deref_mut() = Self::ExpectExclusiveAll2Ready { sem, excl_fut2 };
2745 task::Poll::Pending
2746 }
2747 Self::ExpectExclusiveAll2Ready { sem, mut excl_fut2 } => {
2748 assert!(matches!(AsyncSemaphore::try_acquire_leases(&sem.as_ref(), 0), Ok(None)));
2749
2750 let excl2 = match pin::Pin::new(&mut excl_fut2).poll(cx) {
2751 task::Poll::Ready(guard) => guard.unwrap(),
2752 task::Poll::Pending => unreachable!(),
2753 };
2754
2755 assert!(matches!(AsyncSemaphore::try_acquire_leases(&sem.as_ref(), 0), Ok(None)));
2756 drop(excl2);
2757 assert!(matches!(
2758 AsyncSemaphore::try_acquire_leases(&sem.as_ref(), 0),
2759 Ok(Some(..))
2760 ));
2761
2762 let _trivial3 = AsyncSemaphore::try_acquire_leases(&sem.as_ref(), 0).unwrap().unwrap();
2763 let mut excl_fut4 = AsyncSemaphore::acquire_exclusive_all(&sem.as_ref()).unwrap();
2764 assert!(matches!(pin::Pin::new(&mut excl_fut4).poll(cx), task::Poll::Pending));
2765
2766 let mut trivial_fut5 = AsyncSemaphore::acquire_leases(&sem.as_ref(), 0).unwrap();
2767 let mut lease_fut6 = AsyncSemaphore::acquire_leases(&sem.as_ref(), 1).unwrap();
2768 let mut lease_fut7 = AsyncSemaphore::acquire_leases(&sem.as_ref(), 1).unwrap();
2769 let mut trivial_fut8 = AsyncSemaphore::acquire_leases(&sem.as_ref(), 0).unwrap();
2770 assert!(matches!(pin::Pin::new(&mut trivial_fut5).poll(cx), task::Poll::Pending));
2771 assert!(matches!(pin::Pin::new(&mut lease_fut6).poll(cx), task::Poll::Pending));
2772 assert!(matches!(pin::Pin::new(&mut lease_fut7).poll(cx), task::Poll::Pending));
2773 assert!(matches!(pin::Pin::new(&mut trivial_fut8).poll(cx), task::Poll::Pending));
2774
2775 drop(excl_fut4);
2776
2777 assert!(matches!(
2778 pin::Pin::new(&mut trivial_fut5).poll(cx),
2779 task::Poll::Ready(Ok(..))
2780 ));
2781 let lease6 = match pin::Pin::new(&mut lease_fut6).poll(cx) {
2782 task::Poll::Ready(guard) => guard.unwrap(),
2783 task::Poll::Pending => unreachable!(),
2784 };
2785 assert!(matches!(pin::Pin::new(&mut lease_fut7).poll(cx), task::Poll::Pending));
2786 assert!(matches!(
2787 pin::Pin::new(&mut trivial_fut8).poll(cx),
2788 task::Poll::Ready(Ok(..))
2789 ));
2790 drop(lease6);
2791 assert!(matches!(
2792 pin::Pin::new(&mut lease_fut7).poll(cx),
2793 task::Poll::Ready(Ok(..))
2794 ));
2795
2796 task::Poll::Ready(())
2797 }
2798 Self::Done => unreachable!(),
2799 }
2800 }
2801 }
2802
2803 impl Unpin for TestFuture {}
2804
2805 let e = TestAsyncExecutor::new();
2806 let sem = <<TestNopSyncTypes as sync_types::SyncTypes>::SyncRcPtrFactory as sync_types::SyncRcPtrFactory>::try_new(
2807 TestAsyncSemaphore::new(1, ()).unwrap(),
2808 )
2809 .unwrap();
2810 let excl_fut0 = AsyncSemaphore::acquire_exclusive_all(&sem.as_ref()).unwrap();
2811 let w = TestAsyncExecutor::spawn(&e, TestFuture::ExpectExclusiveAll0Ready { sem, excl_fut0 });
2812 TestAsyncExecutor::run_to_completion(&e);
2813 w.take().unwrap();
2814}