src/libstd/sync/barrier.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 use kinds::{Send, Sync};
  12 use sync::{Mutex, Condvar};
  13
  14 /// A barrier enables multiple tasks to synchronize the beginning
  15 /// of some computation.
  16 ///
  17 /// ```rust
  18 /// use std::sync::{Arc, Barrier};
  19 /// use std::thread::Thread;
  20 ///
  21 /// let barrier = Arc::new(Barrier::new(10));
  22 /// for _ in range(0u, 10) {
  23 ///     let c = barrier.clone();
  24 ///     // The same messages will be printed together.
  25 ///     // You will NOT see any interleaving.
  26 ///     Thread::spawn(move|| {
  27 ///         println!("before wait");
  28 ///         c.wait();
  29 ///         println!("after wait");
  30 ///     }).detach();
  31 /// }
  32 /// ```
  33 pub struct Barrier {
  34     lock: Mutex<BarrierState>,
  35     cvar: Condvar,
  36     num_threads: uint,
  37 }
  38
  39 unsafe impl Send for Barrier {}
  40 unsafe impl Sync for Barrier {}
  41
  42 // The inner state of a double barrier
  43 struct BarrierState {
  44     count: uint,
  45     generation_id: uint,
  46 }
  47
  48 unsafe impl Send for BarrierState {}
  49 unsafe impl Sync for BarrierState {}
  50
  51 impl Barrier {
  52     /// Create a new barrier that can block a given number of threads.
  53     ///
  54     /// A barrier will block `n`-1 threads which call `wait` and then wake up
  55     /// all threads at once when the `n`th thread calls `wait`.
  56     pub fn new(n: uint) -> Barrier {
  57         Barrier {
  58             lock: Mutex::new(BarrierState {
  59                 count: 0,
  60                 generation_id: 0,
  61             }),
  62             cvar: Condvar::new(),
  63             num_threads: n,
  64         }
  65     }
  66
  67     /// Block the current thread until all threads has rendezvoused here.
  68     ///
  69     /// Barriers are re-usable after all threads have rendezvoused once, and can
  70     /// be used continuously.
  71     pub fn wait(&self) {
  72         let mut lock = self.lock.lock().unwrap();
  73         let local_gen = lock.generation_id;
  74         lock.count += 1;
  75         if lock.count < self.num_threads {
  76             // We need a while loop to guard against spurious wakeups.
  77             // http://en.wikipedia.org/wiki/Spurious_wakeup
  78             while local_gen == lock.generation_id &&
  79                   lock.count < self.num_threads {
  80                 lock = self.cvar.wait(lock).unwrap();
  81             }
  82         } else {
  83             lock.count = 0;
  84             lock.generation_id += 1;
  85             self.cvar.notify_all();
  86         }
  87     }
  88 }
  89
  90 #[cfg(test)]
  91 mod tests {
  92     use prelude::v1::*;
  93
  94     use sync::{Arc, Barrier};
  95     use sync::mpsc::{channel, TryRecvError};
  96     use thread::Thread;
  97
  98     #[test]
  99     fn test_barrier() {
 100         let barrier = Arc::new(Barrier::new(10));
 101         let (tx, rx) = channel();
 102
 103         for _ in range(0u, 9) {
 104             let c = barrier.clone();
 105             let tx = tx.clone();
 106             Thread::spawn(move|| {
 107                 c.wait();
 108                 tx.send(true).unwrap();
 109             }).detach();
 110         }
 111
 112         // At this point, all spawned tasks should be blocked,
 113         // so we shouldn't get anything from the port
 114         assert!(match rx.try_recv() {
 115             Err(TryRecvError::Empty) => true,
 116             _ => false,
 117         });
 118
 119         barrier.wait();
 120         // Now, the barrier is cleared and we should get data.
 121         for _ in range(0u, 9) {
 122             rx.recv().unwrap();
 123         }
 124     }
 125 }