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