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
1074
1075
1076
1077
1078
1079
1080
1081
1082
1083
1084
1085
1086
1087
1088
1089
1090
1091
1092
1093
1094
1095
1096
1097
1098
1099
1100
1101
1102
1103
1104
1105
1106
1107
1108
1109
1110
1111
1112
1113
1114
1115
1116
1117
1118
1119
1120
1121
1122
1123
1124
1125
1126
1127
1128
1129
1130
1131
1132
1133
1134
1135
1136
1137
1138
1139
1140
1141
1142
1143
1144
1145
1146
1147
1148
1149
1150
1151
1152
1153
1154
1155
1156
1157
1158
1159
1160
1161
1162
1163
1164
1165
1166
1167
1168
1169
1170
1171
1172
1173
1174
1175
1176
1177
1178
1179
1180
1181
1182
1183
1184
1185
1186
1187
1188
1189
1190
1191
1192
1193
1194
1195
1196
1197
1198
1199
1200
1201
1202
1203
1204
1205
1206
1207
1208
1209
1210
1211
1212
1213
1214
1215
1216
1217
1218
1219
1220
1221
1222
1223
1224
1225
1226
1227
1228
1229
1230
1231
1232
1233
1234
1235
1236
1237
1238
1239
1240
1241
1242
1243
1244
1245
1246
1247
1248
1249
1250
1251
1252
1253
1254
1255
1256
1257
1258
1259
1260
1261
1262
1263
1264
1265
1266
1267
1268
1269
1270
1271
1272
1273
1274
1275
1276
1277
1278
1279
1280
1281
1282
1283
1284
1285
1286
1287
1288
1289
1290
1291
1292
1293
1294
1295
1296
1297
1298
1299
1300
1301
1302
1303
1304
1305
1306
1307
1308
1309
1310
1311
1312
1313
1314
1315
1316
1317
1318
1319
1320
1321
1322
1323
1324
1325
1326
1327
1328
1329
1330
1331
1332
1333
1334
1335
1336
1337
1338
1339
1340
1341
1342
1343
1344
1345
1346
1347
1348
1349
1350
1351
1352
1353
1354
1355
1356
1357
1358
1359
1360
1361
1362
1363
1364
1365
1366
1367
1368
1369
1370
1371
1372
1373
1374
1375
1376
1377
1378
1379
1380
1381
1382
1383
1384
1385
1386
1387
1388
1389
1390
1391
1392
1393
1394
1395
1396
1397
1398
1399
1400
1401
1402
1403
1404
1405
1406
1407
1408
1409
1410
1411
1412
1413
1414
1415
1416
1417
1418
1419
1420
1421
1422
1423
1424
1425
1426
1427
1428
1429
1430
1431
1432
1433
1434
1435
1436
1437
1438
1439
1440
1441
1442
1443
1444
1445
1446
1447
1448
1449
1450
1451
1452
1453
1454
1455
1456
1457
1458
1459
1460
1461
1462
1463
1464
1465
1466
1467
1468
1469
1470
1471
1472
1473
1474
1475
1476
1477
1478
1479
1480
1481
1482
1483
1484
1485
1486
1487
1488
1489
1490
1491
1492
1493
1494
1495
1496
1497
1498
1499
1500
1501
1502
1503
1504
1505
1506
1507
1508
1509
1510
1511
1512
1513
1514
1515
1516
1517
1518
1519
1520
1521
1522
1523
1524
1525
1526
1527
1528
1529
1530
1531
1532
1533
1534
1535
1536
1537
1538
1539
1540
1541
1542
1543
1544
1545
1546
1547
1548
1549
1550
1551
1552
1553
1554
1555
1556
1557
1558
1559
1560
1561
1562
1563
1564
1565
1566
1567
1568
1569
1570
1571
1572
1573
1574
1575
1576
1577
1578
1579
1580
1581
1582
1583
1584
1585
1586
1587
1588
1589
1590
1591
1592
1593
1594
1595
1596
1597
1598
1599
1600
1601
1602
1603
1604
1605
1606
1607
1608
1609
1610
1611
1612
1613
1614
1615
1616
1617
1618
1619
1620
1621
1622
1623
1624
1625
1626
1627
1628
1629
1630
1631
1632
1633
1634
1635
1636
1637
1638
1639
1640
1641
1642
1643
1644
1645
1646
1647
1648
1649
1650
1651
1652
1653
1654
1655
1656
1657
1658
1659
1660
1661
1662
1663
1664
1665
1666
1667
1668
1669
1670
1671
1672
1673
1674
1675
1676
1677
1678
1679
1680
1681
1682
1683
1684
1685
1686
1687
1688
1689
1690
1691
1692
1693
1694
1695
1696
1697
1698
1699
1700
1701
1702
1703
1704
1705
//! A queue of waiting tasks that can be woken in first-in, first-out order (or
//! all at once).
//!
//! See the [`WaitQueue`] type's documentation for details.
#[cfg(any(test, maitake_ultraverbose))]
use crate::util::fmt;
use crate::{
    blocking::{DefaultMutex, Mutex, ScopedRawMutex},
    loom::{
        cell::UnsafeCell,
        sync::atomic::{AtomicUsize, Ordering::*},
    },
    util::{CachePadded, WakeBatch},
    WaitResult,
};
use cordyceps::{
    list::{self, List},
    Linked,
};
use core::{
    future::Future,
    marker::PhantomPinned,
    mem,
    pin::Pin,
    ptr::{self, NonNull},
    task::{Context, Poll, Waker},
};
use mycelium_bitfield::{bitfield, enum_from_bits, FromBits};
use pin_project::{pin_project, pinned_drop};

#[cfg(test)]
mod tests;

/// A queue of waiting tasks which can be [woken in first-in, first-out
/// order][wake], or [all at once][wake_all].
///
/// A `WaitQueue` allows any number of tasks to [wait] asynchronously and be
/// woken when some event occurs, either [individually][wake] in first-in,
/// first-out order, or [all at once][wake_all]. This makes it a vital building
/// block of runtime services (such as timers or I/O resources), where it may be
/// used to wake a set of tasks when a timer completes or when a resource
/// becomes available. It can be equally useful for implementing higher-level
/// synchronization primitives: for example, a `WaitQueue` plus an
/// [`UnsafeCell`] is essentially [an entire implementation of a fair
/// asynchronous mutex][mutex]. Finally, a `WaitQueue` can be a useful
/// synchronization primitive on its own: sometimes, you just need to have a
/// bunch of tasks wait for something and then wake them all up.
///
/// # Overriding the blocking mutex
///
/// This type uses a [blocking `Mutex`](crate::blocking::Mutex) internally to
/// synchronize access to its wait list. By default, this is a [`DefaultMutex`]. To
/// use an alternative [`ScopedRawMutex`] implementation, use the
/// [`new_with_raw_mutex`](Self::new_with_raw_mutex) constructor. See [the documentation
/// on overriding mutex
/// implementations](crate::blocking#overriding-mutex-implementations) for more
/// details.
///
/// # Examples
///
/// Waking a single task at a time by calling [`wake`][wake]:
///
/// ```ignore
/// use std::sync::Arc;
/// use maitake::scheduler::Scheduler;
/// use maitake_sync::WaitQueue;
///
/// const TASKS: usize = 10;
///
/// // In order to spawn tasks, we need a `Scheduler` instance.
/// let scheduler = Scheduler::new();
///
/// // Construct a new `WaitQueue`.
/// let q = Arc::new(WaitQueue::new());
///
/// // Spawn some tasks that will wait on the queue.
/// for _ in 0..TASKS {
///     let q = q.clone();
///     scheduler.spawn(async move {
///         // Wait to be woken by the queue.
///         q.wait().await.expect("queue is not closed");
///     });
/// }
///
/// // Tick the scheduler once.
/// let tick = scheduler.tick();
///
/// // No tasks should complete on this tick, as they are all waiting
/// // to be woken by the queue.
/// assert_eq!(tick.completed, 0, "no tasks have been woken");
///
/// let mut completed = 0;
/// for i in 1..=TASKS {
///     // Wake the next task from the queue.
///     q.wake();
///
///     // Tick the scheduler.
///     let tick = scheduler.tick();
///
///     // A single task should have completed on this tick.
///     completed += tick.completed;
///     assert_eq!(completed, i);
/// }
///
/// assert_eq!(completed, TASKS, "all tasks should have completed");
/// ```
///
/// Waking all tasks using [`wake_all`][wake_all]:
///
/// ```ignore
/// use std::sync::Arc;
/// use maitake::scheduler::Scheduler;
/// use maitake_sync::WaitQueue;
///
/// const TASKS: usize = 10;
///
/// // In order to spawn tasks, we need a `Scheduler` instance.
/// let scheduler = Scheduler::new();
///
/// // Construct a new `WaitQueue`.
/// let q = Arc::new(WaitQueue::new());
///
/// // Spawn some tasks that will wait on the queue.
/// for _ in 0..TASKS {
///     let q = q.clone();
///     scheduler.spawn(async move {
///         // Wait to be woken by the queue.
///         q.wait().await.expect("queue is not closed");
///     });
/// }
///
/// // Tick the scheduler once.
/// let tick = scheduler.tick();
///
/// // No tasks should complete on this tick, as they are all waiting
/// // to be woken by the queue.
/// assert_eq!(tick.completed, 0, "no tasks have been woken");
///
/// // Wake all tasks waiting for the queue.
/// q.wake_all();
///
/// // Tick the scheduler again to run the woken tasks.
/// let tick = scheduler.tick();
///
/// // All tasks have now completed, since they were woken by the
/// // queue.
/// assert_eq!(tick.completed, TASKS, "all tasks should have completed");
/// ```
///
/// # Implementation Notes
///
/// This type is implemented using an [intrusive doubly-linked list][ilist].
///
/// The *[intrusive]* aspect of this list is important, as it means that it does
/// not allocate memory. Instead, nodes in the linked list are stored in the
/// futures of tasks trying to wait for capacity. This means that it is not
/// necessary to allocate any heap memory for each task waiting to be woken.
///
/// However, the intrusive linked list introduces one new danger: because
/// futures can be *cancelled*, and the linked list nodes live within the
/// futures trying to wait on the queue, we *must* ensure that the node
/// is unlinked from the list before dropping a cancelled future. Failure to do
/// so would result in the list containing dangling pointers. Therefore, we must
/// use a *doubly-linked* list, so that nodes can edit both the previous and
/// next node when they have to remove themselves. This is kind of a bummer, as
/// it means we can't use something nice like this [intrusive queue by Dmitry
/// Vyukov][2], and there are not really practical designs for lock-free
/// doubly-linked lists that don't rely on some kind of deferred reclamation
/// scheme such as hazard pointers or QSBR.
///
/// Instead, we just stick a [`Mutex`] around the linked list, which must be
/// acquired to pop nodes from it, or for nodes to remove themselves when
/// futures are cancelled. This is a bit sad, but the critical sections for this
/// mutex are short enough that we still get pretty good performance despite it.
///
/// [`Waker`]: core::task::Waker
/// [wait]: WaitQueue::wait
/// [wake]: WaitQueue::wake
/// [wake_all]: WaitQueue::wake_all
/// [`UnsafeCell`]: core::cell::UnsafeCell
/// [ilist]: cordyceps::List
/// [intrusive]: https://fuchsia.dev/fuchsia-src/development/languages/c-cpp/fbl_containers_guide/introduction
/// [mutex]: crate::Mutex
/// [2]: https://www.1024cores.net/home/lock-free-algorithms/queues/intrusive-mpsc-node-based-queue
#[derive(Debug)]
pub struct WaitQueue<Lock: ScopedRawMutex = DefaultMutex> {
    /// The wait queue's state variable.
    state: CachePadded<AtomicUsize>,

    /// 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.
    ///
    /// A spinlock 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.
    queue: Mutex<List<Waiter>, Lock>,
}

/// Future returned from [`WaitQueue::wait()`].
///
/// This future is fused, so once it has completed, any future calls to poll
/// will immediately return [`Poll::Ready`].
///
/// # Notes
///
/// This future is `!Unpin`, as it is unsafe to [`core::mem::forget`] a `Wait`
/// future once it has been polled. For instance, the following code must not
/// compile:
///
///```compile_fail
/// use maitake_sync::wait_queue::Wait;
///
/// // Calls to this function should only compile if `T` is `Unpin`.
/// fn assert_unpin<T: Unpin>() {}
///
/// assert_unpin::<Wait<'_>>();
/// ```
#[derive(Debug)]
#[pin_project(PinnedDrop)]
#[must_use = "futures do nothing unless `.await`ed or `poll`ed"]
pub struct Wait<'a, Lock: ScopedRawMutex = DefaultMutex> {
    /// The [`WaitQueue`] being waited on.
    queue: &'a WaitQueue<Lock>,

    /// Entry in the wait queue linked list.
    #[pin]
    waiter: Waiter,
}

/// A waiter node which may be linked into a wait queue.
#[derive(Debug)]
#[repr(C)]
#[pin_project]
struct Waiter {
    /// The intrusive linked list node.
    ///
    /// This *must* be the first field in the struct in order for the `Linked`
    /// implementation to be sound.
    #[pin]
    node: UnsafeCell<Node>,

    /// The future's state.
    state: WaitStateBits,
}

#[derive(Debug)]
struct Node {
    /// Intrusive linked list pointers.
    links: list::Links<Waiter>,

    /// The node's waker
    waker: Wakeup,

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

bitfield! {
    #[derive(Eq, PartialEq)]
    struct QueueState<usize> {
        /// The queue's state.
        const STATE: State;

        /// The number of times [`WaitQueue::wake_all`] has been called.
        const WAKE_ALLS = ..;
    }
}

bitfield! {
    #[derive(Eq, PartialEq)]
    struct WaitStateBits<usize> {
        /// The waiter's state.
        const STATE: WaitState;

        /// The number of times [`WaitQueue::wake_all`] has been called.
        const WAKE_ALLS = ..;
    }
}

enum_from_bits! {
    /// The state of a [`Waiter`] node in a [`WaitQueue`].
    #[derive(Debug, Eq, PartialEq)]
    enum WaitState<u8> {
        /// The waiter has not yet been enqueued.
        ///
        /// The number of times [`WaitQueue::wake_all`] has been called is stored
        /// when the node is created, in order to determine whether it was woken by
        /// a stored wakeup when enqueueing.
        ///
        /// When in this state, the node is **not** part of the linked list, and
        /// can be dropped without removing it from the list.
        Start = 0b00,

        /// The waiter is waiting.
        ///
        /// When in this state, the node **is** part of the linked list. If the
        /// node is dropped in this state, it **must** be removed from the list
        /// before dropping it. Failure to ensure this will result in dangling
        /// pointers in the linked list!
        Waiting = 0b01,

        /// The waiter has been woken.
        ///
        /// When in this state, the node is **not** part of the linked list, and
        /// can be dropped without removing it from the list.
        Woken = 0b10,
    }
}

/// The queue's current state.
#[derive(Debug, Copy, Clone, Eq, PartialEq)]
#[repr(u8)]
enum State {
    /// No waiters are queued, and there is no pending notification.
    /// Waiting while the queue is in this state will enqueue the waiter;
    /// notifying while in this state will store a pending notification in the
    /// queue, transitioning to [`State::Woken`].
    Empty = 0b00,

    /// There are one or more waiters in the queue. Waiting while
    /// the queue is in this state will not transition the state. Waking while
    /// in this state will wake the first waiter in the queue; if this empties
    /// the queue, then the queue will transition to [`State::Empty`].
    Waiting = 0b01,

    /// The queue has a stored notification. Waiting while the queue
    /// is in this state will consume the pending notification *without*
    /// enqueueing the waiter and transition the queue to [`State::Empty`].
    /// Waking while in this state will leave the queue in this state.
    Woken = 0b10,

    /// The queue is closed. Waiting while in this state will return
    /// [`Closed`] without transitioning the queue's state.
    ///
    /// *Note*: This *must* correspond to all state bits being set, as it's set
    /// via a [`fetch_or`].
    ///
    /// [`Closed`]: crate::Closed
    /// [`fetch_or`]: core::sync::atomic::AtomicUsize::fetch_or
    Closed = 0b11,
}

#[derive(Clone, Debug)]
enum Wakeup {
    Empty,
    Waiting(Waker),
    One,
    All,
    Closed,
}

// === impl WaitQueue ===

impl WaitQueue {
    loom_const_fn! {
        /// Returns a new `WaitQueue`.
        ///
        /// This constructor returns a `WaitQueue` that uses a [`DefaultMutex`] as
        /// the [`ScopedRawMutex`] implementation for wait list synchronization.
        /// To use a different [`ScopedRawMutex`] implementation, use the
        /// [`new_with_raw_mutex`](Self::new_with_raw_mutex) constructor, instead. See
        /// [the documentation on overriding mutex
        /// implementations](crate::blocking#overriding-mutex-implementations)
        /// for more details.
        #[must_use]
        pub fn new() -> Self {
            Self::new_with_raw_mutex(DefaultMutex::new())
        }
    }
}

impl<Lock> WaitQueue<Lock>
where
    Lock: ScopedRawMutex,
{
    loom_const_fn! {
        /// Returns a new `WaitQueue`, using the provided [`ScopedRawMutex`]
        /// implementation for wait-list synchronization.
        ///
        /// This constructor allows a `WaitQueue` to be constructed with any type that
        /// implements [`ScopedRawMutex`] as the underlying raw blocking mutex
        /// implementation. See [the documentation on overriding mutex
        /// implementations](crate::blocking#overriding-mutex-implementations)
        /// for more details.
        #[must_use]
        pub fn new_with_raw_mutex(lock: Lock) -> Self {
            Self::make(State::Empty, lock)
        }
    }

    loom_const_fn! {
        /// Returns a new `WaitQueue` with a single stored wakeup.
        ///
        /// The first call to [`wait`] on this queue will immediately succeed.
        ///
        /// [`wait`]: Self::wait
        // TODO(eliza): should this be a public API?
        #[must_use]
        pub(crate) fn new_woken(lock: Lock) -> Self {
            Self::make(State::Woken, lock)
        }
    }

    loom_const_fn! {
        #[must_use]
        fn make(state: State, lock: Lock) -> Self {
            Self {
                state: CachePadded::new(AtomicUsize::new(state.into_usize())),
                queue: Mutex::new_with_raw_mutex(List::new(), lock),
            }
        }
    }

    /// Wake the next task in the queue.
    ///
    /// If the queue is empty, a wakeup is stored in the `WaitQueue`, and the
    /// **next** call to [`wait().await`] will complete immediately. If one or more
    /// tasks are currently in the queue, the first task in the queue is woken.
    ///
    /// At most one wakeup will be stored in the queue at any time. If `wake()`
    /// is called many times while there are no tasks in the queue, only a
    /// single wakeup is stored.
    ///
    /// [`wait().await`]: Self::wait()
    ///
    /// # Examples
    ///
    /// # Examples
    ///
    /// ```
    /// # use tokio::task;
    /// # #[tokio::main(flavor = "current_thread")]
    /// # async fn test() {
    /// use std::sync::Arc;
    /// use maitake_sync::WaitQueue;
    ///
    /// let queue = Arc::new(WaitQueue::new());
    ///
    /// let waiter = task::spawn({
    ///     // clone the queue to move into the spawned task
    ///     let queue = queue.clone();
    ///     async move {
    ///         queue.wait().await;
    ///         println!("received wakeup!");
    ///     }
    /// });
    ///
    /// println!("waking task...");
    /// queue.wake();
    ///
    /// waiter.await.unwrap();
    /// # }
    /// # test();
    /// ```
    #[inline]
    pub fn wake(&self) {
        // snapshot the queue's current state.
        let mut state = self.load();

        // check if any tasks are currently waiting on this queue. if there are
        // no waiting tasks, store the wakeup to be consumed by the next call to
        // `wait`.
        loop {
            match state.get(QueueState::STATE) {
                // if the queue is closed, bail.
                State::Closed => return,
                // if there are waiting tasks, break out of the loop and wake one.
                State::Waiting => break,
                _ => {}
            }

            let next = state.with_state(State::Woken);
            // advance the state to `Woken`, and return (if we did so
            // successfully)
            match self.compare_exchange(state, next) {
                Ok(_) => return,
                Err(actual) => state = actual,
            }
        }

        // okay, there are tasks waiting on the queue; we must acquire the lock
        // on the linked list and wake the next task from the queue.
        let waker = self.queue.with_lock(|queue| {
            test_debug!("wake: -> locked");

            // the queue's state may have changed while we were waiting to acquire
            // the lock, so we need to acquire a new snapshot before we take the
            // waker.
            state = self.load();
            self.wake_locked(queue, state)
        });

        //now that we've released the lock, wake the waiting task (if we
        //actually deuqueued one).
        if let Some(waker) = waker {
            waker.wake();
        }
    }

    /// Wake *all* tasks currently in the queue.
    ///
    /// All tasks currently waiting on the queue are woken. Unlike [`wake()`], a
    /// wakeup is *not* stored in the queue to wake the next call to [`wait()`]
    /// if the queue is empty. Instead, this method only wakes all currently
    /// registered waiters. Registering a task to be woken is done by `await`ing
    /// the [`Future`] returned by the [`wait()`] method on this queue.
    ///
    /// # Examples
    ///
    /// ```
    /// # use tokio::task;
    /// # #[tokio::main(flavor = "current_thread")]
    /// # async fn test() {
    /// use maitake_sync::WaitQueue;
    /// use std::sync::Arc;
    ///
    /// let queue = Arc::new(WaitQueue::new());
    ///
    /// // spawn multiple tasks to wait on the queue.
    /// let task1 = task::spawn({
    ///     let queue = queue.clone();
    ///     async move {
    ///         println!("task 1 waiting...");
    ///         queue.wait().await;
    ///         println!("task 1 woken")
    ///     }
    /// });
    ///
    /// let task2 = task::spawn({
    ///     let queue = queue.clone();
    ///     async move {
    ///         println!("task 2 waiting...");
    ///         queue.wait().await;
    ///         println!("task 2 woken")
    ///     }
    /// });
    ///
    /// // yield to the scheduler so that both tasks register
    /// // themselves to wait on the queue.
    /// task::yield_now().await;
    ///
    /// // neither task will have been woken.
    /// assert!(!task1.is_finished());
    /// assert!(!task2.is_finished());
    ///
    /// // wake all tasks waiting on the queue.
    /// queue.wake_all();
    ///
    /// // yield to the scheduler again so that the tasks can execute.
    /// task::yield_now().await;
    ///
    /// assert!(task1.is_finished());
    /// assert!(task2.is_finished());
    /// # }
    /// # test();
    /// ```
    ///
    /// [`wake()`]: Self::wake
    /// [`wait()`]: Self::wait
    pub fn wake_all(&self) {
        let mut batch = WakeBatch::new();
        let mut waiters_remaining = true;

        // This is a little bit contorted: we must load the state inside the
        // lock, but for all states except for `Waiting`, we just need to bail
        // out...but we can't `return` from the outer function inside the lock
        // closure. Therefore, we just return a `bool` and, if it's `true`,
        // return instead of doing more work.
        let done = self.queue.with_lock(|queue| {
            let state = self.load();

            match test_dbg!(state.get(QueueState::STATE)) {
                // If there are no waiters in the queue, increment the number of
                // `wake_all` calls and return. incrementing the `wake_all` count
                // must be performed inside the lock, so we do it here.
                State::Woken | State::Empty => {
                    self.state.fetch_add(QueueState::ONE_WAKE_ALL, SeqCst);
                    true
                }
                // If the queue is already closed, this is a no-op. Just bail.
                State::Closed => true,
                // Okay, there are waiters in the queue. Transition to the empty
                // state inside the lock and start draining the queue.
                State::Waiting => {
                    let next_state = QueueState::new()
                        .with_state(State::Empty)
                        .with(QueueState::WAKE_ALLS, state.get(QueueState::WAKE_ALLS) + 1);
                    self.compare_exchange(state, next_state)
                        .expect("state should not have transitioned while locked");

                    // Drain the first batch of waiters from the queue.
                    waiters_remaining =
                        test_dbg!(Self::drain_to_wake_batch(&mut batch, queue, Wakeup::All));

                    false
                }
            }
        });

        if done {
            return;
        }

        batch.wake_all();
        // As long as there are waiters remaining to wake, lock the queue, drain
        // another batch, release the lock, and wake them.
        while waiters_remaining {
            self.queue.with_lock(|queue| {
                waiters_remaining = Self::drain_to_wake_batch(&mut batch, queue, Wakeup::All);
            });
            batch.wake_all();
        }
    }

    /// Close the queue, indicating that it may no longer be used.
    ///
    /// Once a queue is closed, all [`wait()`] calls (current or future) will
    /// return an error.
    ///
    /// This method is generally used when implementing higher-level
    /// synchronization primitives or resources: when an event makes a resource
    /// permanently unavailable, the queue can be closed.
    ///
    /// [`wait()`]: Self::wait
    pub fn close(&self) {
        let state = self.state.fetch_or(State::Closed.into_usize(), SeqCst);
        let state = test_dbg!(QueueState::from_bits(state));
        if state.get(QueueState::STATE) != State::Waiting {
            return;
        }

        let mut batch = WakeBatch::new();
        let mut waking = true;
        while waking {
            waking = self
                .queue
                .with_lock(|queue| Self::drain_to_wake_batch(&mut batch, queue, Wakeup::Closed));
            batch.wake_all();
        }
    }

    /// Wait to be woken up by this queue.
    ///
    /// Equivalent to:
    ///
    /// ```ignore
    /// async fn wait(&self);
    /// ```
    ///
    /// This returns a [`Wait`] [`Future`] that will complete when the task is
    /// woken by a call to [`wake()`] or [`wake_all()`], or when the `WaitQueue`
    /// is dropped.
    ///
    /// Each `WaitQueue` holds a single wakeup. If [`wake()`] was previously
    /// called while no tasks were waiting on the queue, then `wait().await`
    /// will complete immediately, consuming the stored wakeup. Otherwise,
    /// `wait().await` waits to be woken by the next call to [`wake()`] or
    /// [`wake_all()`].
    ///
    /// The [`Wait`] future is not guaranteed to receive wakeups from calls to
    /// [`wake()`] if it has not yet been polled. See the documentation for the
    /// [`Wait::subscribe()`] method for details on receiving wakeups from the
    /// queue prior to polling the `Wait` future for the first time.
    ///
    /// A `Wait` future **is** is guaranteed to recieve wakeups from calls to
    /// [`wake_all()`] as soon as it is created, even if it has not yet been
    /// polled.
    ///
    /// # Returns
    ///
    /// The [`Future`] returned by this method completes with one of the
    /// following [outputs](Future::Output):
    ///
    /// - [`Ok`]`(())` if the task was woken by a call to [`wake()`] or
    ///   [`wake_all()`].
    /// - [`Err`]`(`[`Closed`]`)` if the task was woken by the `WaitQueue` being
    ///   [`close`d](WaitQueue::close).
    ///
    /// # Cancellation
    ///
    /// A `WaitQueue` fairly distributes wakeups to waiting tasks in the order
    /// that they started to wait. If a [`Wait`] future is dropped, the task
    /// will forfeit its position in the queue.
    ///
    /// # Examples
    ///
    /// ```
    /// # use tokio::task;
    /// # #[tokio::main(flavor = "current_thread")]
    /// # async fn test() {
    /// use std::sync::Arc;
    /// use maitake_sync::WaitQueue;
    ///
    /// let queue = Arc::new(WaitQueue::new());
    ///
    /// let waiter = task::spawn({
    ///     // clone the queue to move into the spawned task
    ///     let queue = queue.clone();
    ///     async move {
    ///         queue.wait().await;
    ///         println!("received wakeup!");
    ///     }
    /// });
    ///
    /// println!("waking task...");
    /// queue.wake();
    ///
    /// waiter.await.unwrap();
    /// # }
    /// # test();
    /// ```
    ///
    /// [`wake()`]: Self::wake
    /// [`wake_all()`]: Self::wake_all
    /// [`Closed`]: crate::Closed
    pub fn wait(&self) -> Wait<'_, Lock> {
        Wait {
            queue: self,
            waiter: self.waiter(),
        }
    }

    pub(crate) fn try_wait(&self) -> Poll<WaitResult<()>> {
        let mut state = self.load();
        let initial_wake_alls = state.get(QueueState::WAKE_ALLS);
        while state.get(QueueState::STATE) == State::Woken {
            match self.compare_exchange(state, state.with_state(State::Empty)) {
                Ok(_) => return Poll::Ready(Ok(())),
                Err(actual) => state = actual,
            }
        }

        match state.get(QueueState::STATE) {
            State::Closed => crate::closed(),
            _ if state.get(QueueState::WAKE_ALLS) > initial_wake_alls => Poll::Ready(Ok(())),
            State::Empty | State::Waiting => Poll::Pending,
            State::Woken => Poll::Ready(Ok(())),
        }
    }

    /// Asynchronously poll the given function `f` until a condition occurs,
    /// using the [`WaitQueue`] to only re-poll when notified.
    ///
    /// This can be used to implement a "wait loop", turning a "try" function
    /// (e.g. "try_recv" or "try_send") into an asynchronous function (e.g.
    /// "recv" or "send").
    ///
    /// In particular, this function correctly *registers* interest in the [`WaitQueue`]
    /// prior to polling the function, ensuring that there is not a chance of a race
    /// where the condition occurs AFTER checking but BEFORE registering interest
    /// in the [`WaitQueue`], which could lead to deadlock.
    ///
    /// This is intended to have similar behavior to `Condvar` in the standard library,
    /// but asynchronous, and not requiring operating system intervention (or existence).
    ///
    /// In particular, this can be used in cases where interrupts or events are used
    /// to signify readiness or completion of some task, such as the completion of a
    /// DMA transfer, or reception of an ethernet frame. In cases like this, the interrupt
    /// can wake the queue, allowing the polling function to check status fields for
    /// partial progress or completion.
    ///
    /// Consider using [`Self::wait_for_value()`] if your function does return a value.
    ///
    /// Consider using [`WaitCell::wait_for()`](super::wait_cell::WaitCell::wait_for)
    /// if you do not need multiple waiters.
    ///
    /// # Returns
    ///
    /// * [`Ok`]`(())` if the closure returns `true`.
    /// * [`Err`]`(`[`Closed`](crate::Closed)`)` if the [`WaitQueue`] is closed.
    ///
    /// # Examples
    ///
    /// ```
    /// # use tokio::task;
    /// # #[tokio::main(flavor = "current_thread")]
    /// # async fn test() {
    /// use std::sync::Arc;
    /// use maitake_sync::WaitQueue;
    /// use std::sync::atomic::{AtomicU8, Ordering};
    ///
    /// let queue = Arc::new(WaitQueue::new());
    /// let num = Arc::new(AtomicU8::new(0));
    ///
    /// let waiter1 = task::spawn({
    ///     // clone items to move into the spawned task
    ///     let queue = queue.clone();
    ///     let num = num.clone();
    ///     async move {
    ///         queue.wait_for(|| num.load(Ordering::Relaxed) > 5).await;
    ///         println!("received wakeup!");
    ///     }
    /// });
    ///
    /// let waiter2 = task::spawn({
    ///     // clone items to move into the spawned task
    ///     let queue = queue.clone();
    ///     let num = num.clone();
    ///     async move {
    ///         queue.wait_for(|| num.load(Ordering::Relaxed) > 10).await;
    ///         println!("received wakeup!");
    ///     }
    /// });
    ///
    /// println!("poking task...");
    ///
    /// for i in 0..20 {
    ///     num.store(i, Ordering::Relaxed);
    ///     queue.wake();
    /// }
    ///
    /// waiter1.await.unwrap();
    /// waiter2.await.unwrap();
    /// # }
    /// # test();
    /// ```
    pub async fn wait_for<F: FnMut() -> bool>(&self, mut f: F) -> WaitResult<()> {
        loop {
            let wait = self.wait();
            let mut wait = core::pin::pin!(wait);
            let _ = wait.as_mut().subscribe()?;
            if f() {
                return Ok(());
            }
            wait.await?;
        }
    }

    /// Asynchronously poll the given function `f` until a condition occurs,
    /// using the [`WaitQueue`] to only re-poll when notified.
    ///
    /// This can be used to implement a "wait loop", turning a "try" function
    /// (e.g. "try_recv" or "try_send") into an asynchronous function (e.g.
    /// "recv" or "send").
    ///
    /// In particular, this function correctly *registers* interest in the [`WaitQueue`]
    /// prior to polling the function, ensuring that there is not a chance of a race
    /// where the condition occurs AFTER checking but BEFORE registering interest
    /// in the [`WaitQueue`], which could lead to deadlock.
    ///
    /// This is intended to have similar behavior to `Condvar` in the standard library,
    /// but asynchronous, and not requiring operating system intervention (or existence).
    ///
    /// In particular, this can be used in cases where interrupts or events are used
    /// to signify readiness or completion of some task, such as the completion of a
    /// DMA transfer, or reception of an ethernet frame. In cases like this, the interrupt
    /// can wake the queue, allowing the polling function to check status fields for
    /// partial progress or completion, and also return the status flags at the same time.
    ///
    /// Consider using [`Self::wait_for()`] if your function does not return a value.
    ///
    /// Consider using [`WaitCell::wait_for_value()`](super::wait_cell::WaitCell::wait_for_value)
    /// if you do not need multiple waiters.
    ///
    /// * [`Ok`]`(T)` if the closure returns [`Some`]`(T)`.
    /// * [`Err`]`(`[`Closed`](crate::Closed)`)` if the [`WaitQueue`] is closed.
    ///
    /// # Examples
    ///
    /// ```
    /// # use tokio::task;
    /// # #[tokio::main(flavor = "current_thread")]
    /// # async fn test() {
    /// use std::sync::Arc;
    /// use maitake_sync::WaitQueue;
    /// use std::sync::atomic::{AtomicU8, Ordering};
    ///
    /// let queue = Arc::new(WaitQueue::new());
    /// let num = Arc::new(AtomicU8::new(0));
    ///
    /// let waiter1 = task::spawn({
    ///     // clone items to move into the spawned task
    ///     let queue = queue.clone();
    ///     let num = num.clone();
    ///     async move {
    ///         let rxd = queue.wait_for_value(|| {
    ///             let val = num.load(Ordering::Relaxed);
    ///             if val > 5 {
    ///                 return Some(val);
    ///             }
    ///             None
    ///         }).await.unwrap();
    ///         assert!(rxd > 5);
    ///         println!("received wakeup with value: {rxd}");
    ///     }
    /// });
    ///
    /// let waiter2 = task::spawn({
    ///     // clone items to move into the spawned task
    ///     let queue = queue.clone();
    ///     let num = num.clone();
    ///     async move {
    ///         let rxd = queue.wait_for_value(|| {
    ///             let val = num.load(Ordering::Relaxed);
    ///             if val > 10 {
    ///                 return Some(val);
    ///             }
    ///             None
    ///         }).await.unwrap();
    ///         assert!(rxd > 10);
    ///         println!("received wakeup with value: {rxd}");
    ///     }
    /// });
    ///
    /// println!("poking task...");
    ///
    /// for i in 0..20 {
    ///     num.store(i, Ordering::Relaxed);
    ///     queue.wake();
    /// }
    ///
    /// waiter1.await.unwrap();
    /// waiter2.await.unwrap();
    /// # }
    /// # test();
    /// ```
    pub async fn wait_for_value<T, F: FnMut() -> Option<T>>(&self, mut f: F) -> WaitResult<T> {
        loop {
            let wait = self.wait();
            let mut wait = core::pin::pin!(wait);
            match wait.as_mut().subscribe() {
                Poll::Ready(wr) => wr?,
                Poll::Pending => {}
            }
            if let Some(t) = f() {
                return Ok(t);
            }
            wait.await?;
        }
    }

    /// Returns `true` if this `WaitQueue` is [closed](Self::close).
    #[must_use]
    pub fn is_closed(&self) -> bool {
        self.load().get(QueueState::STATE) == State::Closed
    }

    /// Returns a [`Waiter`] entry in this queue.
    ///
    /// This is factored out into a separate function because it's used by both
    /// [`WaitQueue::wait`] and [`WaitQueue::wait_owned`].
    fn waiter(&self) -> Waiter {
        // how many times has `wake_all` been called when this waiter is created?
        let current_wake_alls = test_dbg!(self.load().get(QueueState::WAKE_ALLS));
        let state = WaitStateBits::new()
            .with(WaitStateBits::WAKE_ALLS, current_wake_alls)
            .with(WaitStateBits::STATE, WaitState::Start);
        Waiter {
            state,
            node: UnsafeCell::new(Node {
                links: list::Links::new(),
                waker: Wakeup::Empty,
                _pin: PhantomPinned,
            }),
        }
    }

    #[cfg_attr(test, track_caller)]
    fn load(&self) -> QueueState {
        #[allow(clippy::let_and_return)]
        let state = QueueState::from_bits(self.state.load(SeqCst));
        test_debug!("state.load() = {state:?}");
        state
    }

    #[cfg_attr(test, track_caller)]
    fn store(&self, state: QueueState) {
        test_debug!("state.store({state:?}");
        self.state.store(state.0, SeqCst);
    }

    #[cfg_attr(test, track_caller)]
    fn compare_exchange(
        &self,
        current: QueueState,
        new: QueueState,
    ) -> Result<QueueState, QueueState> {
        #[allow(clippy::let_and_return)]
        let res = self
            .state
            .compare_exchange(current.0, new.0, SeqCst, SeqCst)
            .map(QueueState::from_bits)
            .map_err(QueueState::from_bits);
        test_debug!("state.compare_exchange({current:?}, {new:?}) = {res:?}");
        res
    }

    #[cold]
    #[inline(never)]
    fn wake_locked(&self, queue: &mut List<Waiter>, curr: QueueState) -> Option<Waker> {
        let state = curr.get(QueueState::STATE);

        // is the queue still in the `Waiting` state? it is possible that we
        // transitioned to a different state while locking the queue.
        if test_dbg!(state) != State::Waiting {
            // if there are no longer any queued tasks, try to store the
            // wakeup in the queue and bail.
            if let Err(actual) = self.compare_exchange(curr, curr.with_state(State::Woken)) {
                debug_assert!(actual.get(QueueState::STATE) != State::Waiting);
                self.store(actual.with_state(State::Woken));
            }

            return None;
        }

        // otherwise, we have to dequeue a task and wake it.
        let node = queue
            .pop_back()
            .expect("if we are in the Waiting state, there must be waiters in the queue");
        let waker = Waiter::wake(node, queue, Wakeup::One);

        // if we took the final waiter currently in the queue, transition to the
        // `Empty` state.
        if test_dbg!(queue.is_empty()) {
            self.store(curr.with_state(State::Empty));
        }

        waker
    }

    /// Drain waiters from `queue` and add them to `batch`. Returns `true` if
    /// the batch was filled while more waiters remain in the queue, indicating
    /// that this function must be called again to wake all waiters.
    fn drain_to_wake_batch(
        batch: &mut WakeBatch,
        queue: &mut List<Waiter>,
        wakeup: Wakeup,
    ) -> bool {
        while let Some(node) = queue.pop_back() {
            let Some(waker) = Waiter::wake(node, queue, wakeup.clone()) else {
                // this waiter was enqueued by `Wait::register` and doesn't have
                // a waker, just keep going.
                continue;
            };

            if batch.add_waker(waker) {
                // there's still room in the wake set, just keep adding to it.
                continue;
            }

            // wake set is full, drop the lock and wake everyone!
            break;
        }

        !queue.is_empty()
    }
}

// === impl Waiter ===

impl Waiter {
    /// Returns the [`Waker`] for the task that owns this `Waiter`.
    ///
    /// # Safety
    ///
    /// This is only safe to call while the list is locked. The `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 wake(this: NonNull<Self>, list: &mut List<Self>, wakeup: Wakeup) -> Option<Waker> {
        Waiter::with_node(this, list, |node| {
            let waker = test_dbg!(mem::replace(&mut node.waker, wakeup));
            match waker {
                // the node has a registered waker, so wake the task.
                Wakeup::Waiting(waker) => Some(waker),
                // do nothing: the node was registered by `Wait::register`
                // without a waker, so the future will already be woken when it is
                // actually polled.
                Wakeup::Empty => None,
                // the node was already woken? this should not happen and
                // probably indicates a race!
                _ => unreachable!("tried to wake a waiter in the {:?} state!", waker),
            }
        })
    }

    /// # 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))
        }
    }

    fn poll_wait<Lock>(
        mut self: Pin<&mut Self>,
        queue: &WaitQueue<Lock>,
        waker: Option<&Waker>,
    ) -> Poll<WaitResult<()>>
    where
        Lock: ScopedRawMutex,
    {
        test_debug!(ptr = ?fmt::ptr(self.as_mut()), "Waiter::poll_wait");
        let ptr = unsafe { NonNull::from(Pin::into_inner_unchecked(self.as_mut())) };
        let mut this = self.as_mut().project();

        match test_dbg!(this.state.get(WaitStateBits::STATE)) {
            WaitState::Start => {
                let queue_state = queue.load();

                // can we consume a pending wakeup?
                if queue
                    .compare_exchange(
                        queue_state.with_state(State::Woken),
                        queue_state.with_state(State::Empty),
                    )
                    .is_ok()
                {
                    this.state.set(WaitStateBits::STATE, WaitState::Woken);
                    return Poll::Ready(Ok(()));
                }

                // okay, no pending wakeups. try to wait...

                test_debug!("poll_wait: locking...");
                queue.queue.with_lock(move |waiters| {
                    test_debug!("poll_wait: -> locked");
                    let mut queue_state = queue.load();

                    // the whole queue was woken while we were trying to acquire
                    // the lock!
                    if queue_state.get(QueueState::WAKE_ALLS)
                        != this.state.get(WaitStateBits::WAKE_ALLS)
                    {
                        this.state.set(WaitStateBits::STATE, WaitState::Woken);
                        return Poll::Ready(Ok(()));
                    }

                    // transition the queue to the waiting state
                    'to_waiting: loop {
                        match test_dbg!(queue_state.get(QueueState::STATE)) {
                            // the queue is `Empty`, transition to `Waiting`
                            State::Empty => {
                                match queue.compare_exchange(
                                    queue_state,
                                    queue_state.with_state(State::Waiting),
                                ) {
                                    Ok(_) => break 'to_waiting,
                                    Err(actual) => queue_state = actual,
                                }
                            }
                            // the queue is already `Waiting`
                            State::Waiting => break 'to_waiting,
                            // the queue was woken, consume the wakeup.
                            State::Woken => {
                                match queue.compare_exchange(
                                    queue_state,
                                    queue_state.with_state(State::Empty),
                                ) {
                                    Ok(_) => {
                                        this.state.set(WaitStateBits::STATE, WaitState::Woken);
                                        return Poll::Ready(Ok(()));
                                    }
                                    Err(actual) => queue_state = actual,
                                }
                            }
                            State::Closed => return crate::closed(),
                        }
                    }

                    // enqueue the node
                    this.state.set(WaitStateBits::STATE, WaitState::Waiting);
                    if let Some(waker) = waker {
                        this.node.as_mut().with_mut(|node| {
                            unsafe {
                                // safety: we may mutate the node because we are
                                // holding the lock.
                                debug_assert!(matches!((*node).waker, Wakeup::Empty));
                                (*node).waker = Wakeup::Waiting(waker.clone());
                            }
                        });
                    }

                    waiters.push_front(ptr);

                    Poll::Pending
                })
            }
            WaitState::Waiting => {
                queue.queue.with_lock(|_waiters| {
                    this.node.with_mut(|node| unsafe {
                        // safety: we may mutate the node because we are
                        // holding the lock.
                        let node = &mut *node;
                        match node.waker {
                            Wakeup::Waiting(ref mut curr_waker) => {
                                match waker {
                                    Some(waker) if !curr_waker.will_wake(waker) => {
                                        *curr_waker = waker.clone()
                                    }
                                    _ => {}
                                }
                                Poll::Pending
                            }
                            Wakeup::All | Wakeup::One => {
                                this.state.set(WaitStateBits::STATE, WaitState::Woken);
                                Poll::Ready(Ok(()))
                            }
                            Wakeup::Closed => {
                                this.state.set(WaitStateBits::STATE, WaitState::Woken);
                                crate::closed()
                            }
                            Wakeup::Empty => {
                                if let Some(waker) = waker {
                                    node.waker = Wakeup::Waiting(waker.clone());
                                }

                                Poll::Pending
                            }
                        }
                    })
                })
            }
            WaitState::Woken => Poll::Ready(Ok(())),
        }
    }

    /// Release this `Waiter` from the queue.
    ///
    /// This is called from the `drop` implementation for the [`Wait`] and
    /// [`WaitOwned`] futures.
    fn release<Lock>(mut self: Pin<&mut Self>, queue: &WaitQueue<Lock>)
    where
        Lock: ScopedRawMutex,
    {
        let state = *(self.as_mut().project().state);
        let ptr = NonNull::from(unsafe { Pin::into_inner_unchecked(self) });
        test_debug!(self = ?fmt::ptr(ptr), ?state, ?queue.state, "Waiter::release");

        // if we're not enqueued, we don't have to do anything else.
        if state.get(WaitStateBits::STATE) != WaitState::Waiting {
            return;
        }

        let next_waiter = queue.queue.with_lock(|waiters| {
            let state = queue.load();
            // remove the node
            unsafe {
                // safety: we have the lock on the queue, so this is safe.
                waiters.remove(ptr);
            };

            // if we removed the last waiter from the queue, transition the state to
            // `Empty`.
            if test_dbg!(waiters.is_empty()) && state.get(QueueState::STATE) == State::Waiting {
                queue.store(state.with_state(State::Empty));
            }

            // if the node has an unconsumed wakeup, it must be assigned to the next
            // node in the queue.
            if Waiter::with_node(ptr, waiters, |node| matches!(&node.waker, Wakeup::One)) {
                queue.wake_locked(waiters, state)
            } else {
                None
            }
        });

        if let Some(next) = next_waiter {
            next.wake();
        }
    }
}

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 Wait ===

impl<Lock: ScopedRawMutex> Wait<'_, Lock> {
    /// Returns `true` if this `Wait` future is waiting for a notification from
    /// the provided [`WaitQueue`].
    ///
    /// # Examples
    ///
    /// ```
    /// use maitake_sync::WaitQueue;
    ///
    /// let queue1 = WaitQueue::new();
    /// let queue2 = WaitQueue::new();
    ///
    /// let wait = queue1.wait();
    /// assert!(wait.waits_on(&queue1));
    /// assert!(!wait.waits_on(&queue2));
    /// ```
    #[inline]
    #[must_use]
    pub fn waits_on(&self, queue: &WaitQueue<Lock>) -> bool {
        ptr::eq(self.queue, queue)
    }

    /// Returns `true` if `self` and `other` are waiting on a notification from
    /// the same [`WaitQueue`].
    ///
    /// # Examples
    ///
    /// Two [`Wait`] futures waiting on the same [`WaitQueue`] return `true`:
    ///
    /// ```
    /// use maitake_sync::WaitQueue;
    ///
    /// let queue = WaitQueue::new();
    ///
    /// let wait1 = queue.wait();
    /// let wait2 = queue.wait();
    /// assert!(wait1.same_queue(&wait2));
    /// assert!(wait2.same_queue(&wait1));
    /// ```
    ///
    /// Two [`Wait`] futures waiting on different [`WaitQueue`]s return `false`:
    ///
    /// ```
    /// use maitake_sync::WaitQueue;
    ///
    /// let queue1 = WaitQueue::new();
    /// let queue2 = WaitQueue::new();
    ///
    /// let wait1 = queue1.wait();
    /// let wait2 = queue2.wait();
    /// assert!(!wait1.same_queue(&wait2));
    /// assert!(!wait2.same_queue(&wait1));
    /// ```
    #[inline]
    #[must_use]
    pub fn same_queue(&self, other: &Wait<'_, Lock>) -> bool {
        ptr::eq(self.queue, other.queue)
    }

    /// Eagerly subscribe this future to wakeups from [`WaitQueue::wake()`].
    ///
    /// Polling a `Wait` future adds that future to the list of waiters that may
    /// receive a wakeup from a `WaitQueue`. However, in some cases, it is
    /// desirable to subscribe to wakeups *prior* to actually waiting for one.
    /// This method should be used when it is necessary to ensure a `Wait`
    /// future is in the list of waiters before the future is `poll`ed for the
    /// first time.
    ///
    /// In general, this method is used in cases where a [`WaitQueue`] must
    /// synchronize with some additional state, such as an `AtomicBool` or
    /// counter. If a task first checks that state, and then chooses whether or
    /// not to wait on the `WaitQueue` based on that state, then a race
    /// condition may occur where the `WaitQueue` wakes waiters *between* when
    /// the task checked the external state and when it first polled its `Wait`
    /// future to wait on the queue. This method allows registering the `Wait`
    /// future with the queue *prior* to checking the external state, without
    /// actually sleeping, so that when the task does wait for the `Wait` future
    /// to complete, it will have received any wakeup that was sent between when
    /// the external state was checked and the `Wait` future was first polled.
    ///
    /// # Returns
    ///
    /// This method returns a [`Poll`]`<`[`WaitResult`]`>` which is `Ready` a wakeup was
    /// already received. This method returns [`Poll::Ready`] in the following
    /// cases:
    ///
    ///  1. The [`WaitQueue::wake()`] method was called between the creation of the
    ///     `Wait` and the call to this method.
    ///  2. This is the first call to `subscribe` or `poll` on this future, and the
    ///     `WaitQueue` was holding a stored wakeup from a previous call to
    ///     [`wake()`]. This method consumes the wakeup in that case.
    ///  3. The future has previously been `subscribe`d or polled, and it has since
    ///     then been marked ready by either consuming a wakeup from the
    ///     `WaitQueue`, or by a call to [`wake()`] or [`wake_all()`] that
    ///     removed it from the list of futures ready to receive wakeups.
    ///  4. The `WaitQueue` has been [`close`d](WaitQueue::close), in which case
    ///     this method returns `Poll::Ready(Err(Closed))`.
    ///
    /// If this method returns [`Poll::Ready`], any subsequent `poll`s of this
    /// `Wait` future will also immediately return [`Poll::Ready`].
    ///
    /// If the [`Wait`] future subscribed to wakeups from the queue, and
    /// has not been woken, this method returns [`Poll::Pending`].
    ///
    /// [`wake()`]: WaitQueue::wake
    /// [`wake_all()`]: WaitQueue::wake_all
    pub fn subscribe(self: Pin<&mut Self>) -> Poll<WaitResult<()>> {
        let this = self.project();
        this.waiter.poll_wait(this.queue, None)
    }
}

impl<Lock: ScopedRawMutex> Future for Wait<'_, Lock> {
    type Output = WaitResult<()>;

    fn poll(self: Pin<&mut Self>, cx: &mut Context<'_>) -> Poll<Self::Output> {
        let this = self.project();
        this.waiter.poll_wait(this.queue, Some(cx.waker()))
    }
}

#[pinned_drop]
impl<Lock: ScopedRawMutex> PinnedDrop for Wait<'_, Lock> {
    fn drop(mut self: Pin<&mut Self>) {
        let this = self.project();
        this.waiter.release(this.queue);
    }
}

// === impl QueueState ===

impl QueueState {
    const ONE_WAKE_ALL: usize = Self::WAKE_ALLS.first_bit();

    fn with_state(self, state: State) -> Self {
        self.with(Self::STATE, state)
    }
}

impl FromBits<usize> for State {
    const BITS: u32 = 2;
    type Error = core::convert::Infallible;

    fn try_from_bits(bits: usize) -> Result<Self, Self::Error> {
        Ok(match bits as u8 {
            bits if bits == Self::Empty as u8 => Self::Empty,
            bits if bits == Self::Waiting as u8 => Self::Waiting,
            bits if bits == Self::Woken as u8 => Self::Woken,
            bits if bits == Self::Closed as u8 => Self::Closed,
            _ => unsafe {
                unreachable_unchecked!("all potential 2-bit patterns should be covered!")
            },
        })
    }

    fn into_bits(self) -> usize {
        self.into_usize()
    }
}

impl State {
    const fn into_usize(self) -> usize {
        self as u8 as usize
    }
}

// === impl WaitOwned ===

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

    use alloc::sync::Arc;

    /// Future returned from [`WaitQueue::wait_owned()`].
    ///
    /// This is identical to the [`Wait`] future, except that it takes an
    /// [`Arc`] reference to the [`WaitQueue`], allowing the returned future to
    /// live for the `'static` lifetime.
    ///
    /// This future is fused, so once it has completed, any future calls to poll
    /// will immediately return [`Poll::Ready`].
    ///
    /// # Notes
    ///
    /// This future is `!Unpin`, as it is unsafe to [`core::mem::forget`] a
    /// `WaitOwned`  future once it has been polled. For instance, the following
    /// code must not compile:
    ///
    ///```compile_fail
    /// use maitake_sync::wait_queue::WaitOwned;
    ///
    /// // Calls to this function should only compile if `T` is `Unpin`.
    /// fn assert_unpin<T: Unpin>() {}
    ///
    /// assert_unpin::<WaitOwned<'_>>();
    /// ```
    #[derive(Debug)]
    #[pin_project(PinnedDrop)]
    pub struct WaitOwned<Lock: ScopedRawMutex = DefaultMutex> {
        /// The `WaitQueue` being waited on.
        queue: Arc<WaitQueue<Lock>>,

        /// Entry in the wait queue.
        #[pin]
        waiter: Waiter,
    }

    impl<Lock: ScopedRawMutex> WaitQueue<Lock> {
        /// Wait to be woken up by this queue, returning a future that's valid
        /// for the `'static` lifetime.
        ///
        /// This returns a [`WaitOwned`] future that will complete when the task
        /// is woken by a call to [`wake()`] or [`wake_all()`], or when the
        /// `WaitQueue` is [closed].
        ///
        /// This is identical to the [`wait()`] method, except that it takes a
        /// [`Arc`] reference to the [`WaitQueue`], allowing the returned future
        /// to live for the `'static` lifetime. See the documentation for
        /// [`wait()`] for details on how to use the future returned by this
        /// method.
        ///
        /// # Returns
        ///
        /// The [`Future`] returned by this method completes with one of the
        /// following [outputs](Future::Output):
        ///
        /// - [`Ok`]`(())` if the task was woken by a call to [`wake()`] or
        ///   [`wake_all()`].
        /// - [`Err`]`(`[`Closed`]`)` if the task was woken by the `WaitQueue`
        ///   being [closed].
        ///
        /// # Cancellation
        ///
        /// A `WaitQueue` fairly distributes wakeups to waiting tasks in the
        /// order that they started to wait. If a [`WaitOwned`] future is
        /// dropped, the task will forfeit its position in the queue.
        ///
        /// [`wake()`]: Self::wake
        /// [`wake_all()`]: Self::wake_all
        /// [`wait()`]: Self::wait
        /// [closed]: Self::close
        /// [`Closed`]: crate::Closed
        pub fn wait_owned(self: &Arc<Self>) -> WaitOwned<Lock> {
            let waiter = self.waiter();
            let queue = self.clone();
            WaitOwned { queue, waiter }
        }
    }

    // === impl WaitOwned ===

    impl<Lock: ScopedRawMutex> WaitOwned<Lock> {
        /// Returns `true` if this `WaitOwned` future is waiting for a
        /// notification from the provided [`WaitQueue`].
        ///
        /// # Examples
        ///
        /// ```
        /// use maitake_sync::WaitQueue;
        /// use std::sync::Arc;
        ///
        /// let queue1 = Arc::new(WaitQueue::new());
        /// let queue2 = Arc::new(WaitQueue::new());
        ///
        /// let wait = queue1.clone().wait_owned();
        /// assert!(wait.waits_on(&queue1));
        /// assert!(!wait.waits_on(&queue2));
        /// ```
        #[inline]
        #[must_use]
        pub fn waits_on(&self, queue: &WaitQueue<Lock>) -> bool {
            ptr::eq(&*self.queue, queue)
        }

        /// Returns `true` if `self` and `other` are waiting on a notification
        /// from the same [`WaitQueue`].
        ///
        /// # Examples
        ///
        /// Two [`WaitOwned`] futures waiting on the same [`WaitQueue`] return
        /// `true`:
        ///
        /// ```
        /// use maitake_sync::WaitQueue;
        /// use std::sync::Arc;
        ///
        /// let queue = Arc::new(WaitQueue::new());
        ///
        /// let wait1 = queue.clone().wait_owned();
        /// let wait2 = queue.clone().wait_owned();
        /// assert!(wait1.same_queue(&wait2));
        /// assert!(wait2.same_queue(&wait1));
        /// ```
        ///
        /// Two [`WaitOwned`] futures waiting on different [`WaitQueue`]s return
        /// `false`:
        ///
        /// ```
        /// use maitake_sync::WaitQueue;
        /// use std::sync::Arc;
        ///
        /// let queue1 = Arc::new(WaitQueue::new());
        /// let queue2 = Arc::new(WaitQueue::new());
        ///
        /// let wait1 = queue1.wait_owned();
        /// let wait2 = queue2.wait_owned();
        /// assert!(!wait1.same_queue(&wait2));
        /// assert!(!wait2.same_queue(&wait1));
        /// ```
        #[inline]
        #[must_use]
        pub fn same_queue(&self, other: &WaitOwned<Lock>) -> bool {
            Arc::ptr_eq(&self.queue, &other.queue)
        }

        /// Eagerly subscribe this future to wakeups from [`WaitQueue::wake()`].
        ///
        /// Polling a `WaitOwned` future adds that future to the list of waiters
        /// that may receive a wakeup from a `WaitQueue`. However, in some
        /// cases, it is desirable to subscribe to wakeups *prior* to actually
        /// waiting for one. This method should be used when it is necessary to
        /// ensure a `WaitOwned` future is in the list of waiters before the
        /// future is `poll`ed for the rst time.
        ///
        /// In general, this method is used in cases where a [`WaitQueue`] must
        /// synchronize with some additional state, such as an `AtomicBool` or
        /// counter. If a task first checks that state, and then chooses whether or
        /// not to wait on the `WaitQueue` based on that state, then a race
        /// condition may occur where the `WaitQueue` wakes waiters *between* when
        /// the task checked the external state and when it first polled its
        /// `WaitOwned` future to wait on the queue. This method allows
        /// registering the `WaitOwned`  future with the queue *prior* to
        /// checking the external state, without actually sleeping, so that when
        /// the task does wait for the `WaitOwned` future to complete, it will
        /// have received any wakeup that was sent between when the external
        /// state was checked and the `WaitOwned` future was first polled.
        ///
        /// # Returns
        ///
        /// This method returns a [`Poll`]`<`[`WaitResult`]`>` which is `Ready`
        /// a wakeup was already received. This method returns [`Poll::Ready`]
        /// in the following cases:
        ///
        ///  1. The [`WaitQueue::wake()`] method was called between the creation
        ///     of the `WaitOwned` future and the call to this method.
        ///  2. This is the first call to `subscribe` or `poll` on this future,
        ///     and the `WaitQueue` was holding a stored wakeup from a previous
        ///     call to [`wake()`]. This method consumes the wakeup in that case.
        ///  3. The future has previously been `subscribe`d or polled, and it
        ///     has since then been marked ready by either consuming a wakeup
        ///     from the `WaitQueue`, or by a call to [`wake()`] or
        ///     [`wake_all()`] that removed it from the list of futures ready to
        ///     receive wakeups.
        ///  4. The `WaitQueue` has been [`close`d](WaitQueue::close), in which
        ///     case this method returns `Poll::Ready(Err(Closed))`.
        ///
        /// If this method returns [`Poll::Ready`], any subsequent `poll`s of
        /// this `Wait` future will also immediately return [`Poll::Ready`].
        ///
        /// If the [`WaitOwned`] future subscribed to wakeups from the queue,
        /// and has not been woken, this method returns [`Poll::Pending`].
        ///
        /// [`wake()`]: WaitQueue::wake
        /// [`wake_all()`]: WaitQueue::wake_all
        pub fn subscribe(self: Pin<&mut Self>) -> Poll<WaitResult<()>> {
            let this = self.project();
            this.waiter.poll_wait(this.queue, None)
        }
    }

    impl<Lock: ScopedRawMutex> Future for WaitOwned<Lock> {
        type Output = WaitResult<()>;

        fn poll(self: Pin<&mut Self>, cx: &mut Context<'_>) -> Poll<Self::Output> {
            let this = self.project();
            this.waiter.poll_wait(&*this.queue, Some(cx.waker()))
        }
    }

    #[pinned_drop]
    impl<Lock: ScopedRawMutex> PinnedDrop for WaitOwned<Lock> {
        fn drop(mut self: Pin<&mut Self>) {
            let this = self.project();
            this.waiter.release(&*this.queue);
        }
    }
}