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

   1 use crate::sync::atomic::AtomicI32;
   2 use crate::sync::atomic::Ordering::{Acquire, Relaxed, Release};
   3 use crate::time::Duration;
   4
   5 const PARKED: i32 = -1;
   6 const EMPTY: i32 = 0;
   7 const NOTIFIED: i32 = 1;
   8
   9 pub struct Parker {
  10     state: AtomicI32,
  11 }
  12
  13 impl Parker {
  14     #[inline]
  15     pub const fn new() -> Self {
  16         Parker { state: AtomicI32::new(EMPTY) }
  17     }
  18
  19     // Assumes this is only called by the thread that owns the Parker,
  20     // which means that `self.state != PARKED`.
  21     pub unsafe fn park(&self) {
  22         // Change NOTIFIED=>EMPTY or EMPTY=>PARKED, and directly return in the
  23         // first case.
  24         if self.state.fetch_sub(1, Acquire) == NOTIFIED {
  25             return;
  26         }
  27         loop {
  28             // Wait for something to happen, assuming it's still set to PARKED.
  29             futex_wait(&self.state, PARKED, None);
  30             // Change NOTIFIED=>EMPTY and return in that case.
  31             if self.state.compare_and_swap(NOTIFIED, EMPTY, Acquire) == NOTIFIED {
  32                 return;
  33             } else {
  34                 // Spurious wake up. We loop to try again.
  35             }
  36         }
  37     }
  38
  39     // Assumes this is only called by the thread that owns the Parker,
  40     // which means that `self.state != PARKED`.
  41     pub unsafe fn park_timeout(&self, timeout: Duration) {
  42         // Change NOTIFIED=>EMPTY or EMPTY=>PARKED, and directly return in the
  43         // first case.
  44         if self.state.fetch_sub(1, Acquire) == NOTIFIED {
  45             return;
  46         }
  47         // Wait for something to happen, assuming it's still set to PARKED.
  48         futex_wait(&self.state, PARKED, Some(timeout));
  49         // This is not just a store, because we need to establish a
  50         // release-acquire ordering with unpark().
  51         if self.state.swap(EMPTY, Acquire) == NOTIFIED {
  52             // Woke up because of unpark().
  53         } else {
  54             // Timeout or spurious wake up.
  55             // We return either way, because we can't easily tell if it was the
  56             // timeout or not.
  57         }
  58     }
  59
  60     #[inline]
  61     pub fn unpark(&self) {
  62         // Change PARKED=>NOTIFIED, EMPTY=>NOTIFIED, or NOTIFIED=>NOTIFIED, and
  63         // wake the thread in the first case.
  64         //
  65         // Note that even NOTIFIED=>NOTIFIED results in a write. This is on
  66         // purpose, to make sure every unpark() has a release-acquire ordering
  67         // with park().
  68         if self.state.swap(NOTIFIED, Release) == PARKED {
  69             futex_wake(&self.state);
  70         }
  71     }
  72 }
  73
  74 fn futex_wait(futex: &AtomicI32, expected: i32, timeout: Option<Duration>) {
  75     let timespec;
  76     let timespec_ptr = match timeout {
  77         Some(timeout) => {
  78             timespec = libc::timespec {
  79                 tv_sec: timeout.as_secs() as _,
  80                 tv_nsec: timeout.subsec_nanos() as _,
  81             };
  82             &timespec as *const libc::timespec
  83         }
  84         None => crate::ptr::null(),
  85     };
  86     unsafe {
  87         libc::syscall(
  88             libc::SYS_futex,
  89             futex as *const AtomicI32,
  90             libc::FUTEX_WAIT | libc::FUTEX_PRIVATE_FLAG,
  91             expected,
  92             timespec_ptr,
  93         );
  94     }
  95 }
  96
  97 fn futex_wake(futex: &AtomicI32) {
  98     unsafe {
  99         libc::syscall(
 100             libc::SYS_futex,
 101             futex as *const AtomicI32,
 102             libc::FUTEX_WAKE | libc::FUTEX_PRIVATE_FLAG,
 103             1,
 104         );
 105     }
 106 }