library/std/src/sync/mpsc/blocking.rs

   1 //! Generic support for building blocking abstractions.
   2
   3 use crate::sync::atomic::{AtomicBool, Ordering};
   4 use crate::sync::Arc;
   5 use crate::thread::{self, Thread};
   6 use crate::time::Instant;
   7
   8 struct Inner {
   9     thread: Thread,
  10     woken: AtomicBool,
  11 }
  12
  13 unsafe impl Send for Inner {}
  14 unsafe impl Sync for Inner {}
  15
  16 #[derive(Clone)]
  17 pub struct SignalToken {
  18     inner: Arc<Inner>,
  19 }
  20
  21 pub struct WaitToken {
  22     inner: Arc<Inner>,
  23 }
  24
  25 impl !Send for WaitToken {}
  26
  27 impl !Sync for WaitToken {}
  28
  29 pub fn tokens() -> (WaitToken, SignalToken) {
  30     let inner = Arc::new(Inner { thread: thread::current(), woken: AtomicBool::new(false) });
  31     let wait_token = WaitToken { inner: inner.clone() };
  32     let signal_token = SignalToken { inner };
  33     (wait_token, signal_token)
  34 }
  35
  36 impl SignalToken {
  37     pub fn signal(&self) -> bool {
  38         let wake = self
  39             .inner
  40             .woken
  41             .compare_exchange(false, true, Ordering::SeqCst, Ordering::SeqCst)
  42             .is_ok();
  43         if wake {
  44             self.inner.thread.unpark();
  45         }
  46         wake
  47     }
  48
  49     /// Converts to an unsafe raw pointer. Useful for storing in a pipe's state
  50     /// flag.
  51     #[inline]
  52     pub unsafe fn to_raw(self) -> *mut u8 {
  53         Arc::into_raw(self.inner) as *mut u8
  54     }
  55
  56     /// Converts from an unsafe raw pointer. Useful for retrieving a pipe's state
  57     /// flag.
  58     #[inline]
  59     pub unsafe fn from_raw(signal_ptr: *mut u8) -> SignalToken {
  60         SignalToken { inner: Arc::from_raw(signal_ptr as *mut Inner) }
  61     }
  62 }
  63
  64 impl WaitToken {
  65     pub fn wait(self) {
  66         while !self.inner.woken.load(Ordering::SeqCst) {
  67             thread::park()
  68         }
  69     }
  70
  71     /// Returns `true` if we wake up normally.
  72     pub fn wait_max_until(self, end: Instant) -> bool {
  73         while !self.inner.woken.load(Ordering::SeqCst) {
  74             let now = Instant::now();
  75             if now >= end {
  76                 return false;
  77             }
  78             thread::park_timeout(end - now)
  79         }
  80         true
  81     }
  82 }