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 fmt;
  12 use sync::{Mutex, Condvar};
  13
  14 /// A barrier enables multiple threads to synchronize the beginning
  15 /// of some computation.
  16 ///
  17 /// ```
  18 /// use std::sync::{Arc, Barrier};
  19 /// use std::thread;
  20 ///
  21 /// let mut handles = Vec::with_capacity(10);
  22 /// let barrier = Arc::new(Barrier::new(10));
  23 /// for _ in 0..10 {
  24 ///     let c = barrier.clone();
  25 ///     // The same messages will be printed together.
  26 ///     // You will NOT see any interleaving.
  27 ///     handles.push(thread::spawn(move|| {
  28 ///         println!("before wait");
  29 ///         c.wait();
  30 ///         println!("after wait");
  31 ///     }));
  32 /// }
  33 /// // Wait for other threads to finish.
  34 /// for handle in handles {
  35 ///     handle.join().unwrap();
  36 /// }
  37 /// ```
  38 #[stable(feature = "rust1", since = "1.0.0")]
  39 pub struct Barrier {
  40     lock: Mutex<BarrierState>,
  41     cvar: Condvar,
  42     num_threads: usize,
  43 }
  44
  45 // The inner state of a double barrier
  46 struct BarrierState {
  47     count: usize,
  48     generation_id: usize,
  49 }
  50
  51 /// A result returned from wait.
  52 ///
  53 /// Currently this opaque structure only has one method, `.is_leader()`. Only
  54 /// one thread will receive a result that will return `true` from this function.
  55 #[stable(feature = "rust1", since = "1.0.0")]
  56 pub struct BarrierWaitResult(bool);
  57
  58 #[stable(feature = "std_debug", since = "1.15.0")]
  59 impl fmt::Debug for Barrier {
  60     fn fmt(&self, f: &mut fmt::Formatter) -> fmt::Result {
  61         f.pad("Barrier { .. }")
  62     }
  63 }
  64
  65 impl Barrier {
  66     /// Creates a new barrier that can block a given number of threads.
  67     ///
  68     /// A barrier will block `n`-1 threads which call `wait` and then wake up
  69     /// all threads at once when the `n`th thread calls `wait`.
  70     #[stable(feature = "rust1", since = "1.0.0")]
  71     pub fn new(n: usize) -> Barrier {
  72         Barrier {
  73             lock: Mutex::new(BarrierState {
  74                 count: 0,
  75                 generation_id: 0,
  76             }),
  77             cvar: Condvar::new(),
  78             num_threads: n,
  79         }
  80     }
  81
  82     /// Blocks the current thread until all threads have rendezvoused here.
  83     ///
  84     /// Barriers are re-usable after all threads have rendezvoused once, and can
  85     /// be used continuously.
  86     ///
  87     /// A single (arbitrary) thread will receive a `BarrierWaitResult` that
  88     /// returns `true` from `is_leader` when returning from this function, and
  89     /// all other threads will receive a result that will return `false` from
  90     /// `is_leader`
  91     #[stable(feature = "rust1", since = "1.0.0")]
  92     pub fn wait(&self) -> BarrierWaitResult {
  93         let mut lock = self.lock.lock().unwrap();
  94         let local_gen = lock.generation_id;
  95         lock.count += 1;
  96         if lock.count < self.num_threads {
  97             // We need a while loop to guard against spurious wakeups.
  98             // http://en.wikipedia.org/wiki/Spurious_wakeup
  99             while local_gen == lock.generation_id &&
 100                   lock.count < self.num_threads {
 101                 lock = self.cvar.wait(lock).unwrap();
 102             }
 103             BarrierWaitResult(false)
 104         } else {
 105             lock.count = 0;
 106             lock.generation_id += 1;
 107             self.cvar.notify_all();
 108             BarrierWaitResult(true)
 109         }
 110     }
 111 }
 112
 113 #[stable(feature = "std_debug", since = "1.15.0")]
 114 impl fmt::Debug for BarrierWaitResult {
 115     fn fmt(&self, f: &mut fmt::Formatter) -> fmt::Result {
 116         f.debug_struct("BarrierWaitResult")
 117             .field("is_leader", &self.is_leader())
 118             .finish()
 119     }
 120 }
 121
 122 impl BarrierWaitResult {
 123     /// Returns whether this thread from `wait` is the "leader thread".
 124     ///
 125     /// Only one thread will have `true` returned from their result, all other
 126     /// threads will have `false` returned.
 127     #[stable(feature = "rust1", since = "1.0.0")]
 128     pub fn is_leader(&self) -> bool { self.0 }
 129 }
 130
 131 #[cfg(test)]
 132 mod tests {
 133     use sync::{Arc, Barrier};
 134     use sync::mpsc::{channel, TryRecvError};
 135     use thread;
 136
 137     #[test]
 138     #[cfg_attr(target_os = "emscripten", ignore)]
 139     fn test_barrier() {
 140         const N: usize = 10;
 141
 142         let barrier = Arc::new(Barrier::new(N));
 143         let (tx, rx) = channel();
 144
 145         for _ in 0..N - 1 {
 146             let c = barrier.clone();
 147             let tx = tx.clone();
 148             thread::spawn(move|| {
 149                 tx.send(c.wait().is_leader()).unwrap();
 150             });
 151         }
 152
 153         // At this point, all spawned threads should be blocked,
 154         // so we shouldn't get anything from the port
 155         assert!(match rx.try_recv() {
 156             Err(TryRecvError::Empty) => true,
 157             _ => false,
 158         });
 159
 160         let mut leader_found = barrier.wait().is_leader();
 161
 162         // Now, the barrier is cleared and we should get data.
 163         for _ in 0..N - 1 {
 164             if rx.recv().unwrap() {
 165                 assert!(!leader_found);
 166                 leader_found = true;
 167             }
 168         }
 169         assert!(leader_found);
 170     }
 171 }