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