src/libstd/sync/mpsc/blocking.rs

   1 // Copyright 2014 The Rust Project Developers. See the COPYRIGHT
   2 // file at the top-level directory of this distribution and at
   3 // http://rust-lang.org/COPYRIGHT.
   4 //
   5 // Licensed under the Apache License, Version 2.0 <LICENSE-APACHE or
   6 // http://www.apache.org/licenses/LICENSE-2.0> or the MIT license
   7 // <LICENSE-MIT or http://opensource.org/licenses/MIT>, at your
   8 // option. This file may not be copied, modified, or distributed
   9 // except according to those terms.
  10
  11 //! Generic support for building blocking abstractions.
  12
  13 use thread::{self, Thread};
  14 use sync::atomic::{AtomicBool, Ordering};
  15 use sync::Arc;
  16 use mem;
  17 use time::Instant;
  18
  19 struct Inner {
  20     thread: Thread,
  21     woken: AtomicBool,
  22 }
  23
  24 unsafe impl Send for Inner {}
  25 unsafe impl Sync for Inner {}
  26
  27 #[derive(Clone)]
  28 pub struct SignalToken {
  29     inner: Arc<Inner>,
  30 }
  31
  32 pub struct WaitToken {
  33     inner: Arc<Inner>,
  34 }
  35
  36 impl !Send for WaitToken {}
  37
  38 impl !Sync for WaitToken {}
  39
  40 pub fn tokens() -> (WaitToken, SignalToken) {
  41     let inner = Arc::new(Inner {
  42         thread: thread::current(),
  43         woken: AtomicBool::new(false),
  44     });
  45     let wait_token = WaitToken {
  46         inner: inner.clone(),
  47     };
  48     let signal_token = SignalToken {
  49         inner: inner
  50     };
  51     (wait_token, signal_token)
  52 }
  53
  54 impl SignalToken {
  55     pub fn signal(&self) -> bool {
  56         let wake = !self.inner.woken.compare_and_swap(false, true, Ordering::SeqCst);
  57         if wake {
  58             self.inner.thread.unpark();
  59         }
  60         wake
  61     }
  62
  63     /// Convert to an unsafe usize value. Useful for storing in a pipe's state
  64     /// flag.
  65     #[inline]
  66     pub unsafe fn cast_to_usize(self) -> usize {
  67         mem::transmute(self.inner)
  68     }
  69
  70     /// Convert from an unsafe usize value. Useful for retrieving a pipe's state
  71     /// flag.
  72     #[inline]
  73     pub unsafe fn cast_from_usize(signal_ptr: usize) -> SignalToken {
  74         SignalToken { inner: mem::transmute(signal_ptr) }
  75     }
  76 }
  77
  78 impl WaitToken {
  79     pub fn wait(self) {
  80         while !self.inner.woken.load(Ordering::SeqCst) {
  81             thread::park()
  82         }
  83     }
  84
  85     /// Returns true if we wake up normally, false otherwise.
  86     pub fn wait_max_until(self, end: Instant) -> bool {
  87         while !self.inner.woken.load(Ordering::SeqCst) {
  88             let now = Instant::now();
  89             if now >= end {
  90                 return false;
  91             }
  92             thread::park_timeout(end - now)
  93         }
  94         true
  95     }
  96 }