library/std/src/sys_common/thread_parker/generic.rs

   1 //! Parker implementaiton based on a Mutex and Condvar.
   2
   3 use crate::sync::atomic::AtomicUsize;
   4 use crate::sync::atomic::Ordering::SeqCst;
   5 use crate::sync::{Condvar, Mutex};
   6 use crate::time::Duration;
   7
   8 const EMPTY: usize = 0;
   9 const PARKED: usize = 1;
  10 const NOTIFIED: usize = 2;
  11
  12 pub struct Parker {
  13     state: AtomicUsize,
  14     lock: Mutex<()>,
  15     cvar: Condvar,
  16 }
  17
  18 impl Parker {
  19     pub fn new() -> Self {
  20         Parker { state: AtomicUsize::new(EMPTY), lock: Mutex::new(()), cvar: Condvar::new() }
  21     }
  22
  23     // This implementaiton doesn't require `unsafe`, but other implementations
  24     // may assume this is only called by the thread that owns the Parker.
  25     pub unsafe fn park(&self) {
  26         // If we were previously notified then we consume this notification and
  27         // return quickly.
  28         if self.state.compare_exchange(NOTIFIED, EMPTY, SeqCst, SeqCst).is_ok() {
  29             return;
  30         }
  31
  32         // Otherwise we need to coordinate going to sleep
  33         let mut m = self.lock.lock().unwrap();
  34         match self.state.compare_exchange(EMPTY, PARKED, SeqCst, SeqCst) {
  35             Ok(_) => {}
  36             Err(NOTIFIED) => {
  37                 // We must read here, even though we know it will be `NOTIFIED`.
  38                 // This is because `unpark` may have been called again since we read
  39                 // `NOTIFIED` in the `compare_exchange` above. We must perform an
  40                 // acquire operation that synchronizes with that `unpark` to observe
  41                 // any writes it made before the call to unpark. To do that we must
  42                 // read from the write it made to `state`.
  43                 let old = self.state.swap(EMPTY, SeqCst);
  44                 assert_eq!(old, NOTIFIED, "park state changed unexpectedly");
  45                 return;
  46             } // should consume this notification, so prohibit spurious wakeups in next park.
  47             Err(_) => panic!("inconsistent park state"),
  48         }
  49         loop {
  50             m = self.cvar.wait(m).unwrap();
  51             match self.state.compare_exchange(NOTIFIED, EMPTY, SeqCst, SeqCst) {
  52                 Ok(_) => return, // got a notification
  53                 Err(_) => {}     // spurious wakeup, go back to sleep
  54             }
  55         }
  56     }
  57
  58     // This implementaiton doesn't require `unsafe`, but other implementations
  59     // may assume this is only called by the thread that owns the Parker.
  60     pub unsafe fn park_timeout(&self, dur: Duration) {
  61         // Like `park` above we have a fast path for an already-notified thread, and
  62         // afterwards we start coordinating for a sleep.
  63         // return quickly.
  64         if self.state.compare_exchange(NOTIFIED, EMPTY, SeqCst, SeqCst).is_ok() {
  65             return;
  66         }
  67         let m = self.lock.lock().unwrap();
  68         match self.state.compare_exchange(EMPTY, PARKED, SeqCst, SeqCst) {
  69             Ok(_) => {}
  70             Err(NOTIFIED) => {
  71                 // We must read again here, see `park`.
  72                 let old = self.state.swap(EMPTY, SeqCst);
  73                 assert_eq!(old, NOTIFIED, "park state changed unexpectedly");
  74                 return;
  75             } // should consume this notification, so prohibit spurious wakeups in next park.
  76             Err(_) => panic!("inconsistent park_timeout state"),
  77         }
  78
  79         // Wait with a timeout, and if we spuriously wake up or otherwise wake up
  80         // from a notification we just want to unconditionally set the state back to
  81         // empty, either consuming a notification or un-flagging ourselves as
  82         // parked.
  83         let (_m, _result) = self.cvar.wait_timeout(m, dur).unwrap();
  84         match self.state.swap(EMPTY, SeqCst) {
  85             NOTIFIED => {} // got a notification, hurray!
  86             PARKED => {}   // no notification, alas
  87             n => panic!("inconsistent park_timeout state: {}", n),
  88         }
  89     }
  90
  91     pub fn unpark(&self) {
  92         // To ensure the unparked thread will observe any writes we made
  93         // before this call, we must perform a release operation that `park`
  94         // can synchronize with. To do that we must write `NOTIFIED` even if
  95         // `state` is already `NOTIFIED`. That is why this must be a swap
  96         // rather than a compare-and-swap that returns if it reads `NOTIFIED`
  97         // on failure.
  98         match self.state.swap(NOTIFIED, SeqCst) {
  99             EMPTY => return,    // no one was waiting
 100             NOTIFIED => return, // already unparked
 101             PARKED => {}        // gotta go wake someone up
 102             _ => panic!("inconsistent state in unpark"),
 103         }
 104
 105         // There is a period between when the parked thread sets `state` to
 106         // `PARKED` (or last checked `state` in the case of a spurious wake
 107         // up) and when it actually waits on `cvar`. If we were to notify
 108         // during this period it would be ignored and then when the parked
 109         // thread went to sleep it would never wake up. Fortunately, it has
 110         // `lock` locked at this stage so we can acquire `lock` to wait until
 111         // it is ready to receive the notification.
 112         //
 113         // Releasing `lock` before the call to `notify_one` means that when the
 114         // parked thread wakes it doesn't get woken only to have to wait for us
 115         // to release `lock`.
 116         drop(self.lock.lock().unwrap());
 117         self.cvar.notify_one()
 118     }
 119 }