library/std/src/thread/parker/futex.rs

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