1 | #![forbid (unsafe_op_in_unsafe_fn)] |
2 | use crate::pin::Pin; |
3 | use crate::sync::atomic::Ordering::{Acquire, Release}; |
4 | use crate::sys::futex::{self, futex_wait, futex_wake}; |
5 | use crate::time::Duration; |
6 | |
7 | type Futex = futex::SmallFutex; |
8 | type State = futex::SmallPrimitive; |
9 | |
10 | const PARKED: State = State::MAX; |
11 | const EMPTY: State = 0; |
12 | const NOTIFIED: State = 1; |
13 | |
14 | pub struct Parker { |
15 | state: Futex, |
16 | } |
17 | |
18 | // Notes about memory ordering: |
19 | // |
20 | // Memory ordering is only relevant for the relative ordering of operations |
21 | // between different variables. Even Ordering::Relaxed guarantees a |
22 | // monotonic/consistent order when looking at just a single atomic variable. |
23 | // |
24 | // So, since this parker is just a single atomic variable, we only need to look |
25 | // at the ordering guarantees we need to provide to the 'outside world'. |
26 | // |
27 | // The only memory ordering guarantee that parking and unparking provide, is |
28 | // that things which happened before unpark() are visible on the thread |
29 | // returning from park() afterwards. Otherwise, it was effectively unparked |
30 | // before unpark() was called while still consuming the 'token'. |
31 | // |
32 | // In other words, unpark() needs to synchronize with the part of park() that |
33 | // consumes the token and returns. |
34 | // |
35 | // This is done with a release-acquire synchronization, by using |
36 | // Ordering::Release when writing NOTIFIED (the 'token') in unpark(), and using |
37 | // Ordering::Acquire when checking for this state in park(). |
38 | impl Parker { |
39 | /// Constructs the futex parker. The UNIX parker implementation |
40 | /// requires this to happen in-place. |
41 | pub unsafe fn new_in_place(parker: *mut Parker) { |
42 | unsafe { parker.write(Self { state: Futex::new(EMPTY) }) }; |
43 | } |
44 | |
45 | // Assumes this is only called by the thread that owns the Parker, |
46 | // which means that `self.state != PARKED`. |
47 | pub unsafe fn park(self: Pin<&Self>) { |
48 | // Change NOTIFIED=>EMPTY or EMPTY=>PARKED, and directly return in the |
49 | // first case. |
50 | if self.state.fetch_sub(1, Acquire) == NOTIFIED { |
51 | return; |
52 | } |
53 | loop { |
54 | // Wait for something to happen, assuming it's still set to PARKED. |
55 | futex_wait(&self.state, PARKED, None); |
56 | // Change NOTIFIED=>EMPTY and return in that case. |
57 | if self.state.compare_exchange(NOTIFIED, EMPTY, Acquire, Acquire).is_ok() { |
58 | return; |
59 | } else { |
60 | // Spurious wake up. We loop to try again. |
61 | } |
62 | } |
63 | } |
64 | |
65 | // Assumes this is only called by the thread that owns the Parker, |
66 | // which means that `self.state != PARKED`. This implementation doesn't |
67 | // require `Pin`, but other implementations do. |
68 | pub unsafe fn park_timeout(self: Pin<&Self>, timeout: Duration) { |
69 | // Change NOTIFIED=>EMPTY or EMPTY=>PARKED, and directly return in the |
70 | // first case. |
71 | if self.state.fetch_sub(1, Acquire) == NOTIFIED { |
72 | return; |
73 | } |
74 | // Wait for something to happen, assuming it's still set to PARKED. |
75 | futex_wait(&self.state, PARKED, Some(timeout)); |
76 | // This is not just a store, because we need to establish a |
77 | // release-acquire ordering with unpark(). |
78 | if self.state.swap(EMPTY, Acquire) == NOTIFIED { |
79 | // Woke up because of unpark(). |
80 | } else { |
81 | // Timeout or spurious wake up. |
82 | // We return either way, because we can't easily tell if it was the |
83 | // timeout or not. |
84 | } |
85 | } |
86 | |
87 | // This implementation doesn't require `Pin`, but other implementations do. |
88 | #[inline ] |
89 | pub fn unpark(self: Pin<&Self>) { |
90 | // Change PARKED=>NOTIFIED, EMPTY=>NOTIFIED, or NOTIFIED=>NOTIFIED, and |
91 | // wake the thread in the first case. |
92 | // |
93 | // Note that even NOTIFIED=>NOTIFIED results in a write. This is on |
94 | // purpose, to make sure every unpark() has a release-acquire ordering |
95 | // with park(). |
96 | if self.state.swap(NOTIFIED, Release) == PARKED { |
97 | futex_wake(&self.state); |
98 | } |
99 | } |
100 | } |
101 | |