library/std/src/sync/barrier.rs

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