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 marker::{Sync, Send};
  17 use mem;
  18 use clone::Clone;
  19 use time::Instant;
  20
  21 struct Inner {
  22     thread: Thread,
  23     woken: AtomicBool,
  24 }
  25
  26 unsafe impl Send for Inner {}
  27 unsafe impl Sync for Inner {}
  28
  29 #[derive(Clone)]
  30 pub struct SignalToken {
  31     inner: Arc<Inner>,
  32 }
  33
  34 pub struct WaitToken {
  35     inner: Arc<Inner>,
  36 }
  37
  38 impl !Send for WaitToken {}
  39
  40 impl !Sync for WaitToken {}
  41
  42 pub fn tokens() -> (WaitToken, SignalToken) {
  43     let inner = Arc::new(Inner {
  44         thread: thread::current(),
  45         woken: AtomicBool::new(false),
  46     });
  47     let wait_token = WaitToken {
  48         inner: inner.clone(),
  49     };
  50     let signal_token = SignalToken {
  51         inner: inner
  52     };
  53     (wait_token, signal_token)
  54 }
  55
  56 impl SignalToken {
  57     pub fn signal(&self) -> bool {
  58         let wake = !self.inner.woken.compare_and_swap(false, true, Ordering::SeqCst);
  59         if wake {
  60             self.inner.thread.unpark();
  61         }
  62         wake
  63     }
  64
  65     /// Convert to an unsafe usize value. Useful for storing in a pipe's state
  66     /// flag.
  67     #[inline]
  68     pub unsafe fn cast_to_usize(self) -> usize {
  69         mem::transmute(self.inner)
  70     }
  71
  72     /// Convert from an unsafe usize value. Useful for retrieving a pipe's state
  73     /// flag.
  74     #[inline]
  75     pub unsafe fn cast_from_usize(signal_ptr: usize) -> SignalToken {
  76         SignalToken { inner: mem::transmute(signal_ptr) }
  77     }
  78 }
  79
  80 impl WaitToken {
  81     pub fn wait(self) {
  82         while !self.inner.woken.load(Ordering::SeqCst) {
  83             thread::park()
  84         }
  85     }
  86
  87     /// Returns true if we wake up normally, false otherwise.
  88     pub fn wait_max_until(self, end: Instant) -> bool {
  89         while !self.inner.woken.load(Ordering::SeqCst) {
  90             let now = Instant::now();
  91             if now >= end {
  92                 return false;
  93             }
  94             thread::park_timeout(end - now)
  95         }
  96         true
  97     }
  98 }