std/sys/sync/thread_parking/
futex.rs

Help
1#![forbid(unsafe_op_in_unsafe_fn)]
2use crate::pin::Pin;
3use crate::sync::atomic::Ordering::{Acquire, Release};
4use crate::sys::futex::{self, futex_wait, futex_wake};
5use crate::time::Duration;
6
7type Futex = futex::SmallFutex;
8type State = futex::SmallPrimitive;
9
10const PARKED: State = State::MAX;
11const EMPTY: State = 0;
12const NOTIFIED: State = 1;
13
14pub 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().
38impl 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}
std/sys/sync/thread_parking/futex.rs

std/sys/sync/thread_parking/
futex.rs
