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

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