1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
165
166
167
168
169
170
171
172
173
174
175
176
177
178
179
180
181
182
183
184
185
186
187
188
189
190
191
192
193
194
195
196
197
198
199
200
201
202
203
204
205
206
207
208
209
210
211
212
213
214
215
216
217
218
219
220
221
222
223
224
225
226
227
228
229
230
231
232
233
234
235
236
237
238
239
240
241
242
243
244
245
246
247
248
249
250
251
252
253
254
255
256
257
258
259
260
261
262
263
264
265
266
267
268
269
270
271
272
273
274
275
276
277
278
279
280
281
282
283
284
285
286
287
288
289
290
291
292
293
294
295
296
297
298
299
300
301
302
303
304
305
306
307
308
309
310
311
312
313
314
315
316
317
318
319
320
321
322
323
324
325
326
327
328
329
330
331
332
333
334
335
336
337
338
339
340
341
342
343
344
345
346
347
348
349
350
351
352
353
354
355
356
357
358
359
360
361
362
363
364
365
366
367
368
369
370
371
372
373
374
375
376
377
378
379
380
381
382
383
384
385
386
387
388
389
390
391
392
393
394
395
396
397
398
399
400
401
402
403
404
405
406
407
408
409
410
411
412
413
414
415
416
417
418
419
420
421
422
423
424
425
426
427
428
429
430
431
432
433
434
435
436
437
438
439
440
441
442
443
444
445
446
447
448
449
450
451
452
453
454
455
456
457
458
459
460
461
462
463
464
465
466
467
468
469
470
471
472
473
474
475
476
477
478
479
480
481
482
483
484
485
486
487
488
489
490
491
492
493
494
495
496
497
498
499
500
501
502
503
504
505
506
507
508
509
510
511
512
513
514
515
516
517
518
519
520
521
522
523
524
525
526
527
528
529
530
531
532
533
534
535
536
537
538
539
540
541
542
543
544
545
546
547
548
549
550
551
552
553
554
555
556
557
558
559
560
561
562
563
564
565
566
567
568
569
570
571
572
573
574
575
576
577
578
579
580
581
582
583
584
585
586
587
588
589
590
591
592
593
594
595
596
597
598
599
600
601
602
603
604
605
606
607
608
609
610
611
612
613
614
615
616
617
618
619
620
621
622
623
624
625
626
627
628
629
630
631
632
633
634
635
636
637
638
639
640
641
642
643
644
645
646
647
648
649
650
651
652
653
654
655
656
657
658
659
660
661
662
663
664
665
666
667
668
669
670
671
672
673
674
675
676
677
678
679
680
681
682
683
684
685
686
687
688
689
690
691
692
693
694
695
696
697
698
699
700
701
702
703
704
705
706
707
708
709
710
711
712
713
714
715
716
717
718
719
720
721
722
723
724
725
726
727
728
729
730
731
732
733
734
735
736
737
738
739
740
741
742
743
744
745
746
747
748
749
750
751
752
753
754
755
756
757
758
759
760
761
762
763
764
765
766
767
768
769
770
771
772
773
774
775
776
777
778
779
780
781
782
783
784
785
786
787
788
789
790
791
792
793
794
795
796
797
798
799
800
801
802
803
804
805
806
807
808
809
810
811
812
813
814
815
816
817
818
819
820
821
822
823
824
825
826
827
828
829
830
831
832
833
834
835
836
837
838
839
840
841
842
843
844
845
846
847
848
849
850
851
852
853
854
855
856
857
858
859
860
861
862
863
864
865
866
867
868
869
870
871
872
873
874
875
876
877
878
879
880
881
882
883
884
885
886
887
888
889
890
891
892
893
894
895
896
897
898
899
900
901
902
903
904
905
906
907
908
909
910
911
912
913
914
915
916
917
918
919
920
921
922
923
924
925
926
927
928
929
930
931
932
933
934
935
936
937
938
939
940
941
942
943
944
945
946
947
948
949
950
951
952
953
954
955
956
957
958
959
960
961
962
963
964
965
966
967
968
969
970
971
972
973
974
975
976
977
978
979
980
981
982
983
984
985
986
987
988
989
990
991
992
993
994
995
996
997
998
999
1000
1001
1002
1003
1004
1005
1006
1007
1008
1009
1010
1011
1012
1013
1014
1015
1016
1017
1018
1019
1020
1021
1022
1023
1024
1025
1026
1027
1028
1029
1030
1031
1032
1033
1034
1035
1036
1037
1038
1039
1040
1041
1042
1043
1044
1045
1046
1047
1048
1049
1050
1051
1052
1053
1054
1055
1056
1057
1058
1059
1060
1061
1062
1063
1064
1065
1066
1067
1068
1069
1070
1071
1072
1073
//! An asynchronous [counting semaphore].
//!
//! A semaphore limits the number of tasks which may execute concurrently. See
//! the [`Semaphore`] type's documentation for details.
//!
//! [counting semaphore]: https://en.wikipedia.org/wiki/Semaphore_(programming)
use crate::{
    loom::{
        cell::UnsafeCell,
        sync::{
            atomic::{AtomicUsize, Ordering::*},
            spin::{Mutex, MutexGuard},
        },
    },
    util::{fmt, CachePadded, WakeBatch},
    WaitResult,
};
use cordyceps::{
    list::{self, List},
    Linked,
};
use core::{
    cmp,
    future::Future,
    marker::PhantomPinned,
    pin::Pin,
    ptr::{self, NonNull},
    task::{Context, Poll, Waker},
};
use pin_project::{pin_project, pinned_drop};

#[cfg(test)]
mod tests;

/// An asynchronous [counting semaphore].
///
/// A semaphore is a synchronization primitive that limits the number of tasks
/// that may run concurrently. It consists of a count of _permits_, which tasks
/// may [`acquire`] in order to execute in some context. When a task acquires a
/// permit from the semaphore, the count of permits held by the semaphore is
/// decreased. When no permits remain in the semaphore, any task that wishes to
/// acquire a permit must (asynchronously) wait until another task has released
/// a permit.
///
/// The [`Permit`] type is a RAII guard representing one or more permits
/// acquired from a `Semaphore`. When a [`Permit`] is dropped, the permits it
/// represents are released back to the `Semaphore`, potentially allowing a
/// waiting task to acquire them.
///
/// # Fairness
///
/// This semaphore is _fair_: as permits become available, they are assigned to
/// waiting tasks in the order that those tasks requested permits (first-in,
/// first-out). This means that all tasks waiting to acquire permits will
/// eventually be allowed to progress, and a single task cannot starve the
/// semaphore of permits (provided that permits are eventually released). The
/// semaphore remains fair even when a call to `acquire` requests more than one
/// permit at a time.
///
/// # Examples
///
/// Using a semaphore to limit concurrency:
///
/// ```
/// # use tokio::task;
/// # #[tokio::main(flavor = "current_thread")]
/// # async fn test() {
/// # use std as alloc;
/// use maitake_sync::Semaphore;
/// use alloc::sync::Arc;
///
/// # let mut tasks = Vec::new();
/// // Allow 4 tasks to run concurrently at a time.
/// let semaphore = Arc::new(Semaphore::new(4));
///
/// for _ in 0..8 {
///     // Clone the `Arc` around the semaphore.
///     let semaphore = semaphore.clone();
///     # let t =
///     task::spawn(async move {
///         // Acquire a permit from the semaphore, returning a RAII guard that
///         // releases the permit back to the semaphore when dropped.
///         //
///         // If all 4 permits have been acquired, the calling task will yield,
///         // and it will be woken when another task releases a permit.
///         let _permit = semaphore
///             .acquire(1)
///             .await
///             .expect("semaphore will not be closed");
///
///         // do some work...
///     });
///     # tasks.push(t);
/// }
/// # for task in tasks { task.await.unwrap() };
/// # }
/// # test();
/// ```
///
/// A semaphore may also be used to cause a task to run once all of a set of
/// tasks have completed. If we want some task _B_ to run only after a fixed
/// number _n_ of tasks _A_ have run, we can have task _B_ try to acquire _n_
/// permits from a semaphore with 0 permits, and have each task _A_ add one
/// permit to the semaphore when it completes.
///
/// For example:
///
/// ```
/// # use tokio::task;
/// # #[tokio::main(flavor = "current_thread")]
/// # async fn test() {
/// # use std as alloc;
/// use maitake_sync::Semaphore;
/// use alloc::sync::Arc;
///
/// // How many tasks will we be waiting for the completion of?
/// const TASKS: usize = 4;
///
/// // Create the semaphore with 0 permits.
/// let semaphore = Arc::new(Semaphore::new(0));
///
/// // Spawn the "B" task that will wait for the 4 "A" tasks to complete.
/// # let b_task =
/// task::spawn({
///     let semaphore = semaphore.clone();
///     async move {
///         println!("Task B starting...");
///
///         // Since the semaphore is created with 0 permits, this will
///         // wait until all 4 "A" tasks have completed.
///        let _permit = semaphore
///             .acquire(TASKS)
///             .await
///             .expect("semaphore will not be closed");
///
///         // ... do some work ...
///
///         println!("Task B done!");
///     }
/// });
///
/// # let mut tasks = Vec::new();
/// for i in 0..TASKS {
///     let semaphore = semaphore.clone();
///     # let t =
///     task::spawn(async move {
///         println!("Task A {i} starting...");
///
///         // Add a single permit to the semaphore. Once all 4 tasks have
///         // completed, the semaphore will have the 4 permits required to
///         // wake the "B" task.
///         semaphore.add_permits(1);
///
///         // ... do some work ...
///
///         println!("Task A {i} done");
///     });
///     # tasks.push(t);
/// }
///
/// # for t in tasks { t.await.unwrap() };
/// # b_task.await.unwrap();
/// # }
/// # test();
/// ```
///
/// [counting semaphore]: https://en.wikipedia.org/wiki/Semaphore_(programming)
/// [`acquire`]: Semaphore::acquire
#[derive(Debug)]
pub struct Semaphore {
    /// The number of permits in the semaphore (or [`usize::MAX] if the
    /// semaphore is closed.
    permits: CachePadded<AtomicUsize>,

    /// The queue of tasks waiting to acquire permits.
    ///
    /// A spinlock (from `mycelium_util`) is used here, in order to support
    /// `no_std` platforms; when running `loom` tests, a `loom` mutex is used
    /// instead to simulate the spinlock, because loom doesn't play nice with
    /// real spinlocks.
    waiters: Mutex<SemQueue>,
}

/// A [RAII guard] representing one or more permits acquired from a
/// [`Semaphore`].
///
/// When the `Permit` is dropped, the permits it represents are released back to
/// the [`Semaphore`], potentially waking another task.
///
/// This type is returned by the [`Semaphore::acquire`] and
/// [`Semaphore::try_acquire`] methods.
///
/// [RAII guard]: https://rust-unofficial.github.io/patterns/patterns/behavioural/RAII.html
#[derive(Debug)]
#[must_use = "dropping a `Permit` releases the acquired permits back to the `Semaphore`"]
pub struct Permit<'sem> {
    permits: usize,
    semaphore: &'sem Semaphore,
}

/// The future returned by the [`Semaphore::acquire`] method.
#[derive(Debug)]
#[pin_project(PinnedDrop)]
#[must_use = "futures do nothing unless `.await`ed or `poll`ed"]
pub struct Acquire<'sem> {
    semaphore: &'sem Semaphore,
    queued: bool,
    permits: usize,
    #[pin]
    waiter: Waiter,
}

/// Errors returned by [`Semaphore::try_acquire`].

#[derive(Debug, PartialEq, Eq)]
pub enum TryAcquireError {
    /// The semaphore has been [closed], so additional permits cannot be
    /// acquired.
    ///
    /// [closed]: Semaphore::close
    Closed,
    /// The semaphore does not currently have enough permits to satisfy the
    /// request.
    InsufficientPermits,
}

/// The semaphore's queue of waiters. This is the portion of the semaphore's
/// state stored inside the lock.
#[derive(Debug)]
struct SemQueue {
    /// The linked list of waiters.
    ///
    /// # Safety
    ///
    /// This is protected by a mutex; the mutex *must* be acquired when
    /// manipulating the linked list, OR when manipulating waiter nodes that may
    /// be linked into the list. If a node is known to not be linked, it is safe
    /// to modify that node (such as by waking the stored [`Waker`]) without
    /// holding the lock; otherwise, it may be modified through the list, so the
    /// lock must be held when modifying the
    /// node.
    queue: List<Waiter>,

    /// Has the semaphore closed?
    ///
    /// This is tracked inside of the locked state to avoid a potential race
    /// condition where the semaphore closes while trying to lock the wait queue.
    closed: bool,
}

#[derive(Debug)]
#[pin_project]
struct Waiter {
    #[pin]
    node: UnsafeCell<Node>,

    remaining_permits: RemainingPermits,
}

/// The number of permits needed before this waiter can be woken.
///
/// When this value reaches zero, the waiter has acquired all its needed
/// permits and can be woken. If this value is `usize::max`, then the waiter
/// has not yet been linked into the semaphore queue.
#[derive(Debug)]
struct RemainingPermits(AtomicUsize);

#[derive(Debug)]
struct Node {
    links: list::Links<Waiter>,
    waker: Option<Waker>,

    // This type is !Unpin due to the heuristic from:
    // <https://github.com/rust-lang/rust/pull/82834>
    _pin: PhantomPinned,
}

// === impl Semaphore ===

impl Semaphore {
    /// The maximum number of permits a `Semaphore` may contain.
    pub const MAX_PERMITS: usize = usize::MAX - 1;

    const CLOSED: usize = usize::MAX;

    loom_const_fn! {
        /// Returns a new `Semaphore` with `permits` permits available.
        ///
        /// # Panics
        ///
        /// If `permits` is less than [`MAX_PERMITS`] ([`usize::MAX`] - 1).
        ///
        /// [`MAX_PERMITS`]: Self::MAX_PERMITS
        #[must_use]
        pub fn new(permits: usize) -> Self {
            assert!(
                permits <= Self::MAX_PERMITS,
                "a semaphore may not have more than Semaphore::MAX_PERMITS permits",
        );
            Self {
                permits: CachePadded::new(AtomicUsize::new(permits)),
                waiters: Mutex::new(SemQueue {
                    queue: List::new(),
                    closed: false,
                }),
            }
        }
    }

    /// Returns the number of permits currently available in this semaphore, or
    /// 0 if the semaphore is [closed].
    ///
    /// [closed]: Semaphore::close
    pub fn available_permits(&self) -> usize {
        let permits = self.permits.load(Acquire);
        if permits == Self::CLOSED {
            return 0;
        }

        permits
    }

    /// Acquire `permits` permits from the `Semaphore`, waiting asynchronously
    /// if there are insufficient permits currently available.
    ///
    /// # Returns
    ///
    /// - `Ok(`[`Permit`]`)` with the requested number of permits, if the
    ///   permits were acquired.
    /// - `Err(`[`Closed`]`)` if the semaphore was [closed].
    ///
    /// # Cancellation
    ///
    /// This method uses a queue to fairly distribute permits in the order they
    /// were requested. If an [`Acquire`] future is dropped before it completes,
    /// the task will lose its place in the queue.
    ///
    /// [`Closed`]: crate::Closed
    /// [closed]: Semaphore::close
    pub fn acquire(&self, permits: usize) -> Acquire<'_> {
        Acquire {
            semaphore: self,
            queued: false,
            permits,
            waiter: Waiter::new(permits),
        }
    }

    /// Add `permits` new permits to the semaphore.
    ///
    /// This permanently increases the number of permits available in the
    /// semaphore. The permit count can be permanently *decreased* by calling
    /// [`acquire`] or [`try_acquire`], and [`forget`]ting the returned [`Permit`].
    ///
    /// # Panics
    ///
    /// If adding `permits` permits would cause the permit count to overflow
    /// [`MAX_PERMITS`] ([`usize::MAX`] - 1).
    ///
    /// [`acquire`]: Self::acquire
    /// [`try_acquire`]: Self::try_acquire
    /// [`forget`]: Permit::forget
    /// [`MAX_PERMITS`]: Self::MAX_PERMITS
    #[inline(always)]
    pub fn add_permits(&self, permits: usize) {
        if permits == 0 {
            return;
        }

        self.add_permits_locked(permits, self.waiters.lock());
    }

    /// Try to acquire `permits` permits from the `Semaphore`, without waiting
    /// for additional permits to become available.
    ///
    /// # Returns
    ///
    /// - `Ok(`[`Permit`]`)` with the requested number of permits, if the
    ///   permits were acquired.
    /// - `Err(`[`TryAcquireError::Closed`]`)` if the semaphore was [closed].
    /// - `Err(`[`TryAcquireError::InsufficientPermits`]`)` if the semaphore had
    ///   fewer than `permits` permits available.
    ///
    /// [`Closed`]: crate::Closed
    /// [closed]: Semaphore::close
    pub fn try_acquire(&self, permits: usize) -> Result<Permit<'_>, TryAcquireError> {
        trace!(permits, "Semaphore::try_acquire");
        self.try_acquire_inner(permits).map(|_| Permit {
            permits,
            semaphore: self,
        })
    }

    /// Closes the semaphore.
    ///
    /// This wakes all tasks currently waiting on the semaphore, and prevents
    /// new permits from being acquired.
    pub fn close(&self) {
        let mut waiters = self.waiters.lock();
        self.permits.store(Self::CLOSED, Release);
        waiters.closed = true;
        while let Some(waiter) = waiters.queue.pop_back() {
            if let Some(waker) = Waiter::take_waker(waiter, &mut waiters.queue) {
                waker.wake();
            }
        }
    }

    fn poll_acquire(
        &self,
        mut node: Pin<&mut Waiter>,
        permits: usize,
        queued: bool,
        cx: &mut Context<'_>,
    ) -> Poll<WaitResult<()>> {
        trace!(
            waiter = ?fmt::ptr(node.as_mut()),
            permits,
            queued,
            "Semaphore::poll_acquire"
        );
        // the total number of permits we've acquired so far.
        let mut acquired_permits = 0;
        let waiter = node.as_mut().project();

        // how many permits are currently needed?
        let needed_permits = if queued {
            waiter.remaining_permits.remaining()
        } else {
            permits
        };

        // okay, let's try to consume the requested number of permits from the
        // semaphore.
        let mut sem_curr = self.permits.load(Relaxed);
        let mut lock = None;
        let mut waiters = loop {
            // semaphore has closed
            if sem_curr == Self::CLOSED {
                return crate::closed();
            }

            // the total number of permits currently available to this waiter
            // are the number it has acquired so far plus all the permits
            // in the semaphore.
            let available_permits = sem_curr + acquired_permits;
            let mut remaining = 0;
            let mut sem_next = sem_curr;
            let can_acquire = if available_permits >= needed_permits {
                // there are enough permits available to satisfy this request.

                // the semaphore's next state will be the current number of
                // permits less the amount we have to take from it to satisfy
                // request.
                sem_next -= needed_permits - acquired_permits;
                needed_permits
            } else {
                // the number of permits available in the semaphore is less than
                // number we want to acquire. take all the currently available
                // permits.
                sem_next = 0;
                // how many permits do we still need to acquire?
                remaining = (needed_permits - acquired_permits) - sem_curr;
                sem_curr
            };

            if remaining > 0 && lock.is_none() {
                // we weren't able to acquire enough permits on this poll, so
                // the waiter will probably need to be queued, so we must lock
                // the wait queue.
                //
                // this has to happen *before* the CAS that sets the new value
                // of the semaphore's permits counter. if we subtracted the
                // permits before acquiring the lock, additional permits might
                // be added to the semaphore while we were waiting to lock the
                // wait queue, and we would miss acquiring those permits.
                // therefore, we lock the queue now.
                lock = Some(self.waiters.lock());
            }

            if let Err(actual) = test_dbg!(self.permits.compare_exchange(
                test_dbg!(sem_curr),
                test_dbg!(sem_next),
                AcqRel,
                Acquire
            )) {
                // the semaphore was updated while we were trying to acquire
                // permits.
                sem_curr = actual;
                continue;
            }

            // okay, we took some permits from the semaphore.
            acquired_permits += can_acquire;
            // did we acquire all the permits we needed?
            if test_dbg!(remaining) == 0 {
                if !queued {
                    // the wasn't already in the queue, so we won't need to
                    // remove it --- we're done!
                    trace!(
                        waiter = ?fmt::ptr(node.as_mut()),
                        permits,
                        queued,
                        "Semaphore::poll_acquire -> all permits acquired; done"
                    );
                    return Poll::Ready(Ok(()));
                } else {
                    // we acquired all the permits we needed, but the waiter was
                    // already in the queue, so we need to dequeue it. we may
                    // have already acquired the lock on a previous CAS attempt
                    // that failed, but if not, grab it now.
                    break lock.unwrap_or_else(|| self.waiters.lock());
                }
            }

            // we updated the semaphore, and will need to wait to acquire
            // additional permits.
            break lock.expect("we should have acquired the lock before trying to wait");
        };

        if waiters.closed {
            trace!(
                waiter = ?fmt::ptr(node.as_mut()),
                permits,
                queued,
                "Semaphore::poll_acquire -> semaphore closed"
            );
            return crate::closed();
        }

        // add permits to the waiter, returning whether we added enough to wake
        // it.
        if waiter.remaining_permits.add(&mut acquired_permits) {
            trace!(
                waiter = ?fmt::ptr(node.as_mut()),
                permits,
                queued,
                "Semaphore::poll_acquire -> remaining permits acquired; done"
            );
            // if there are permits left over after waking the node, give the
            // remaining permits back to the semaphore, potentially assigning
            // them to the next waiter in the queue.
            self.add_permits_locked(acquired_permits, waiters);
            return Poll::Ready(Ok(()));
        }

        debug_assert_eq!(
            acquired_permits, 0,
            "if we are enqueueing a waiter, we must have used all the acquired permits"
        );

        // we need to wait --- register the polling task's waker, and enqueue
        // node.
        let node_ptr = unsafe { NonNull::from(Pin::into_inner_unchecked(node)) };
        Waiter::with_node(node_ptr, &mut waiters.queue, |node| {
            let will_wake = node
                .waker
                .as_ref()
                .map_or(false, |waker| waker.will_wake(cx.waker()));
            if !will_wake {
                node.waker = Some(cx.waker().clone())
            }
        });

        // if the waiter is not already in the queue, add it now.
        if !queued {
            waiters.queue.push_front(node_ptr);
            trace!(
                waiter = ?node_ptr,
                permits,
                queued,
                "Semaphore::poll_acquire -> enqueued"
            );
        }

        Poll::Pending
    }

    #[inline(never)]
    fn add_permits_locked<'sem>(
        &'sem self,
        mut permits: usize,
        mut waiters: MutexGuard<'sem, SemQueue>,
    ) {
        trace!(permits, "Semaphore::add_permits");
        if waiters.closed {
            trace!(
                permits,
                "Semaphore::add_permits -> already closed; doing nothing"
            );
            return;
        }

        let mut drained_queue = false;
        while permits > 0 && !drained_queue {
            let mut batch = WakeBatch::new();
            while batch.can_add_waker() {
                // peek the last waiter in the queue to add permits to it; we may not
                // be popping it from the queue if there are not enough permits to
                // wake that waiter.
                match waiters.queue.back() {
                    Some(waiter) => {
                        // try to add enough permits to wake this waiter. if we
                        // can't, break --- we should be out of permits.
                        if !waiter.project_ref().remaining_permits.add(&mut permits) {
                            debug_assert_eq!(permits, 0);
                            break;
                        }
                    }
                    None => {
                        // we've emptied the queue. all done!
                        drained_queue = true;
                        break;
                    }
                };

                // okay, we added enough permits to wake this waiter.
                let waiter = waiters
                    .queue
                    .pop_back()
                    .expect("if `back()` returned `Some`, `pop_back()` will also return `Some`");
                let waker = Waiter::take_waker(waiter, &mut waiters.queue);
                trace!(?waiter, ?waker, permits, "Semaphore::add_permits -> waking");
                if let Some(waker) = waker {
                    batch.add_waker(waker);
                }
            }

            if permits > 0 && drained_queue {
                trace!(
                    permits,
                    "Semaphore::add_permits -> queue drained, assigning remaining permits to semaphore"
                );
                // we drained the queue, but there are still permits left --- add
                // them to the semaphore.
                let prev = self.permits.fetch_add(permits, Release);
                assert!(
                    prev + permits <= Self::MAX_PERMITS,
                    "semaphore overflow adding {permits} permits to {prev}; max permits: {}",
                    Self::MAX_PERMITS
                );
            }

            // wake set is full, drop the lock and wake everyone!
            drop(waiters);
            batch.wake_all();

            // reacquire the lock and continue waking
            waiters = self.waiters.lock();
        }
    }

    /// Drop an `Acquire` future.
    ///
    /// This is factored out into a method on `Semaphore`, because the same code
    /// is run when dropping an `Acquire` future or an `AcquireOwned` future.
    fn drop_acquire(&self, waiter: Pin<&mut Waiter>, permits: usize, queued: bool) {
        // If the future is completed, there is no node in the wait list, so we
        // can skip acquiring the lock.
        if !queued {
            return;
        }

        // This is where we ensure safety. The future is being dropped,
        // which means we must ensure that the waiter entry is no longer stored
        // in the linked list.
        let mut waiters = self.waiters.lock();

        let acquired_permits = permits - waiter.remaining_permits.remaining();

        // Safety: we have locked the wait list.
        unsafe {
            // remove the entry from the list
            let node = NonNull::from(Pin::into_inner_unchecked(waiter));
            waiters.queue.remove(node)
        };

        if acquired_permits > 0 {
            self.add_permits_locked(acquired_permits, waiters);
        }
    }

    /// Try to acquire permits from the semaphore without waiting.
    ///
    /// This method is factored out because it's identical between the
    /// `try_acquire` and `try_acquire_owned` methods, which behave identically
    /// but return different permit types.
    fn try_acquire_inner(&self, permits: usize) -> Result<(), TryAcquireError> {
        let mut available = self.permits.load(Relaxed);
        loop {
            // are there enough permits to satisfy the request?
            match available {
                Self::CLOSED => {
                    trace!(permits, "Semaphore::try_acquire -> closed");
                    return Err(TryAcquireError::Closed);
                }
                available if available < permits => {
                    trace!(
                        permits,
                        available,
                        "Semaphore::try_acquire -> insufficient permits"
                    );
                    return Err(TryAcquireError::InsufficientPermits);
                }
                _ => {}
            }

            let remaining = available - permits;
            match self
                .permits
                .compare_exchange_weak(available, remaining, AcqRel, Acquire)
            {
                Ok(_) => {
                    trace!(permits, remaining, "Semaphore::try_acquire -> acquired");
                    return Ok(());
                }
                Err(actual) => available = actual,
            }
        }
    }
}

// === impl Acquire ===

impl<'sem> Future for Acquire<'sem> {
    type Output = WaitResult<Permit<'sem>>;
    fn poll(self: Pin<&mut Self>, cx: &mut Context<'_>) -> Poll<Self::Output> {
        let this = self.project();
        let poll = this
            .semaphore
            .poll_acquire(this.waiter, *this.permits, *this.queued, cx)
            .map_ok(|_| Permit {
                permits: *this.permits,
                semaphore: this.semaphore,
            });
        *this.queued = poll.is_pending();
        poll
    }
}

#[pinned_drop]
impl PinnedDrop for Acquire<'_> {
    fn drop(self: Pin<&mut Self>) {
        let this = self.project();
        trace!(?this.queued, "Acquire::drop");
        this.semaphore
            .drop_acquire(this.waiter, *this.permits, *this.queued)
    }
}

// safety: the `Acquire` future is not automatically `Sync` because the `Waiter`
// node contains an `UnsafeCell`, which is not `Sync`. this impl is safe because
// the `Acquire` future will only access this `UnsafeCell` when mutably borrowed
// (when polling or dropping the future), so the future itself is safe to share
// immutably between threads.
unsafe impl Sync for Acquire<'_> {}

// === impl Permit ===

impl Permit<'_> {
    /// Forget this permit, dropping it *without* returning the number of
    /// acquired permits to the semaphore.
    ///
    /// This permanently decreases the number of permits in the semaphore by
    /// [`self.permits()`](Self::permits).
    pub fn forget(mut self) {
        self.permits = 0;
    }

    /// Returns the count of semaphore permits owned by this `Permit`.
    #[inline]
    #[must_use]
    pub fn permits(&self) -> usize {
        self.permits
    }
}

impl Drop for Permit<'_> {
    fn drop(&mut self) {
        trace!(?self.permits, "Permit::drop");
        self.semaphore.add_permits(self.permits);
    }
}

// === impl TryAcquireError ===

impl fmt::Display for TryAcquireError {
    fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result {
        match self {
            Self::Closed => f.pad("semaphore closed"),
            Self::InsufficientPermits => f.pad("semaphore has insufficient permits"),
        }
    }
}

feature! {
    #![feature = "core-error"]
    impl core::error::Error for TryAcquireError {}
}

// === Owned variants when `Arc` is available ===

feature! {
    #![feature = "alloc"]

    use alloc::sync::Arc;

    /// Future returned from [`Semaphore::acquire_owned()`].
    ///
    /// This is identical to the [`Acquire`] future, except that it takes an
    /// [`Arc`] reference to the [`Semaphore`], allowing the returned future to
    /// live for the `'static` lifetime, and returns an [`OwnedPermit`] (rather
    /// than a [`Permit`]), which is also valid for the `'static` lifetime.
    #[derive(Debug)]
    #[pin_project(PinnedDrop)]
    #[must_use = "futures do nothing unless `.await`ed or `poll`ed"]
    pub struct AcquireOwned {
        semaphore: Arc<Semaphore>,
        queued: bool,
        permits: usize,
        #[pin]
        waiter: Waiter,
    }

    /// An owned [RAII guard] representing one or more permits acquired from a
    /// [`Semaphore`].
    ///
    /// When the `OwnedPermit` is dropped, the permits it represents are
    /// released back to  the [`Semaphore`], potentially waking another task.
    ///
    /// This type is identical to the [`Permit`] type, except that it holds an
    /// [`Arc`] clone of the [`Semaphore`], rather than borrowing it. This
    /// allows the guard to be valid for the `'static` lifetime.
    ///
    /// This type is returned by the [`Semaphore::acquire_owned`] and
    /// [`Semaphore::try_acquire_owned`] methods.
    ///
    /// [RAII guard]: https://rust-unofficial.github.io/patterns/patterns/behavioural/RAII.html
    #[derive(Debug)]
    #[must_use = "dropping an `OwnedPermit` releases the acquired permits back to the `Semaphore`"]
    pub struct OwnedPermit {
        permits: usize,
        semaphore: Arc<Semaphore>,
    }

    impl Semaphore {
        /// Acquire `permits` permits from the `Semaphore`, waiting asynchronously
        /// if there are insufficient permits currently available, and returning
        /// an [`OwnedPermit`].
        ///
        /// This method behaves identically to [`acquire`], except that it
        /// requires the `Semaphore` to be wrapped in an [`Arc`], and returns an
        /// [`OwnedPermit`] which clones the [`Arc`] rather than borrowing the
        /// semaphore. This allows the returned [`OwnedPermit`] to be valid for
        /// the `'static` lifetime.
        ///
        /// # Returns
        ///
        /// - `Ok(`[`OwnedPermit`]`)` with the requested number of permits, if the
        ///   permits were acquired.
        /// - `Err(`[`Closed`]`)` if the semaphore was [closed].
        ///
        /// # Cancellation
        ///
        /// This method uses a queue to fairly distribute permits in the order they
        /// were requested. If an [`AcquireOwned`] future is dropped before it
        /// completes,   the task will lose its place in the queue.
        ///
        /// [`acquire`]: Semaphore::acquire
        /// [`Closed`]: crate::Closed
        /// [closed]: Semaphore::close
        pub fn acquire_owned(self: &Arc<Self>, permits: usize) -> AcquireOwned {
            AcquireOwned {
                semaphore: self.clone(),
                queued: false,
                permits,
                waiter: Waiter::new(permits),
            }
        }

        /// Try to acquire `permits` permits from the `Semaphore`, without waiting
        /// for additional permits to become available, and returning an [`OwnedPermit`].
        ///
        /// This method behaves identically to [`try_acquire`], except that it
        /// requires the `Semaphore` to be wrapped in an [`Arc`], and returns an
        /// [`OwnedPermit`] which clones the [`Arc`] rather than borrowing the
        /// semaphore. This allows the returned [`OwnedPermit`] to be valid for
        /// the `'static` lifetime.
        ///
        /// # Returns
        ///
        /// - `Ok(`[`OwnedPermit`]`)` with the requested number of permits, if the
        ///   permits were acquired.
        /// - `Err(`[`TryAcquireError::Closed`]`)` if the semaphore was [closed].
        /// - `Err(`[`TryAcquireError::InsufficientPermits`]`)` if the semaphore
        ///   had fewer than `permits` permits available.
        ///
        ///
        /// [`try_acquire`]: Semaphore::try_acquire
        /// [`Closed`]: crate::Closed
        /// [closed]: Semaphore::close
        pub fn try_acquire_owned(self: &Arc<Self>, permits: usize) -> Result<OwnedPermit, TryAcquireError> {
            trace!(permits, "Semaphore::try_acquire_owned");
            self.try_acquire_inner(permits).map(|_| OwnedPermit {
                permits,
                semaphore: self.clone(),
            })
        }
    }

    // === impl AcquireOwned ===

    impl Future for AcquireOwned {
        type Output = WaitResult<OwnedPermit>;

        fn poll(self: Pin<&mut Self>, cx: &mut Context<'_>) -> Poll<Self::Output> {
            let this = self.project();
            let poll = this
                .semaphore
                .poll_acquire(this.waiter, *this.permits, *this.queued, cx)
                .map_ok(|_| OwnedPermit {
                    permits: *this.permits,
                    // TODO(eliza): might be nice to not have to bump the
                    // refcount here...
                    semaphore: this.semaphore.clone(),
                });
            *this.queued = poll.is_pending();
            poll
        }
    }

    #[pinned_drop]
    impl PinnedDrop for AcquireOwned {
        fn drop(mut self: Pin<&mut Self>) {
            let this = self.project();
            trace!(?this.queued, "AcquireOwned::drop");
            this.semaphore
                .drop_acquire(this.waiter, *this.permits, *this.queued)
        }
    }

    // safety: this is safe for the same reasons as the `Sync` impl for the
    // `Acquire` future.
    unsafe impl Sync for AcquireOwned {}

    // === impl OwnedPermit ===

    impl OwnedPermit {
        /// Forget this permit, dropping it *without* returning the number of
        /// acquired permits to the semaphore.
        ///
        /// This permanently decreases the number of permits in the semaphore by
        /// [`self.permits()`](Self::permits).
        pub fn forget(mut self) {
            self.permits = 0;
        }

        /// Returns the count of semaphore permits owned by this `OwnedPermit`.
        #[inline]
        #[must_use]
        pub fn permits(&self) -> usize {
            self.permits
        }
    }

    impl Drop for OwnedPermit {
        fn drop(&mut self) {
            trace!(?self.permits, "OwnedPermit::drop");
            self.semaphore.add_permits(self.permits);
        }
    }

}

// === impl Waiter ===

impl Waiter {
    fn new(permits: usize) -> Self {
        Self {
            node: UnsafeCell::new(Node {
                links: list::Links::new(),
                waker: None,
                _pin: PhantomPinned,
            }),
            remaining_permits: RemainingPermits(AtomicUsize::new(permits)),
        }
    }

    #[inline(always)]
    #[cfg_attr(loom, track_caller)]
    fn take_waker(this: NonNull<Self>, list: &mut List<Self>) -> Option<Waker> {
        Self::with_node(this, list, |node| node.waker.take())
    }

    /// # Safety
    ///
    /// This is only safe to call while the list is locked. The dummy `_list`
    /// parameter ensures this method is only called while holding the lock, so
    /// this can be safe.
    ///
    /// Of course, that must be the *same* list that this waiter is a member of,
    /// and currently, there is no way to ensure that...
    #[inline(always)]
    #[cfg_attr(loom, track_caller)]
    fn with_node<T>(
        mut this: NonNull<Self>,
        _list: &mut List<Self>,
        f: impl FnOnce(&mut Node) -> T,
    ) -> T {
        unsafe {
            // safety: this is only called while holding the lock on the queue,
            // so it's safe to mutate the waiter.
            this.as_mut().node.with_mut(|node| f(&mut *node))
        }
    }
}

unsafe impl Linked<list::Links<Waiter>> for Waiter {
    type Handle = NonNull<Waiter>;

    fn into_ptr(r: Self::Handle) -> NonNull<Self> {
        r
    }

    unsafe fn from_ptr(ptr: NonNull<Self>) -> Self::Handle {
        ptr
    }

    unsafe fn links(target: NonNull<Self>) -> NonNull<list::Links<Waiter>> {
        // Safety: using `ptr::addr_of!` avoids creating a temporary
        // reference, which stacked borrows dislikes.
        let node = ptr::addr_of!((*target.as_ptr()).node);
        (*node).with_mut(|node| {
            let links = ptr::addr_of_mut!((*node).links);
            // Safety: since the `target` pointer is `NonNull`, we can assume
            // that pointers to its members are also not null, making this use
            // of `new_unchecked` fine.
            NonNull::new_unchecked(links)
        })
    }
}

// === impl RemainingPermits ===

impl RemainingPermits {
    /// Add an acquisition of permits to the waiter, returning whether or not
    /// the waiter has acquired enough permits to be woken.
    #[inline]
    #[cfg_attr(loom, track_caller)]
    fn add(&self, permits: &mut usize) -> bool {
        let mut curr = self.0.load(Relaxed);
        loop {
            let taken = cmp::min(curr, *permits);
            let remaining = curr - taken;
            match self
                .0
                .compare_exchange_weak(curr, remaining, AcqRel, Acquire)
            {
                // added the permits to the waiter!
                Ok(_) => {
                    *permits -= taken;
                    return remaining == 0;
                }
                Err(actual) => curr = actual,
            }
        }
    }

    #[inline]
    fn remaining(&self) -> usize {
        self.0.load(Acquire)
    }
}