Skip to main content

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}