use crate::fmt; use crate::sync::{Mutex, Condvar}; /// A barrier enables multiple threads to synchronize the beginning /// of some computation. /// /// # Examples /// /// ``` /// use std::sync::{Arc, Barrier}; /// use std::thread; /// /// let mut handles = Vec::with_capacity(10); /// let barrier = Arc::new(Barrier::new(10)); /// for _ in 0..10 { /// let c = barrier.clone(); /// // The same messages will be printed together. /// // You will NOT see any interleaving. /// handles.push(thread::spawn(move|| { /// println!("before wait"); /// c.wait(); /// println!("after wait"); /// })); /// } /// // Wait for other threads to finish. /// for handle in handles { /// handle.join().unwrap(); /// } /// ``` #[stable(feature = "rust1", since = "1.0.0")] pub struct Barrier { lock: Mutex, cvar: Condvar, num_threads: usize, } // The inner state of a double barrier struct BarrierState { count: usize, generation_id: usize, } /// A `BarrierWaitResult` is returned by [`wait`] when all threads in the [`Barrier`] /// have rendezvoused. /// /// [`wait`]: struct.Barrier.html#method.wait /// [`Barrier`]: struct.Barrier.html /// /// # Examples /// /// ``` /// use std::sync::Barrier; /// /// let barrier = Barrier::new(1); /// let barrier_wait_result = barrier.wait(); /// ``` #[stable(feature = "rust1", since = "1.0.0")] pub struct BarrierWaitResult(bool); #[stable(feature = "std_debug", since = "1.16.0")] impl fmt::Debug for Barrier { fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result { f.pad("Barrier { .. }") } } impl Barrier { /// Creates a new barrier that can block a given number of threads. /// /// A barrier will block `n`-1 threads which call [`wait`] and then wake up /// all threads at once when the `n`th thread calls [`wait`]. /// /// [`wait`]: #method.wait /// /// # Examples /// /// ``` /// use std::sync::Barrier; /// /// let barrier = Barrier::new(10); /// ``` #[stable(feature = "rust1", since = "1.0.0")] pub fn new(n: usize) -> Barrier { Barrier { lock: Mutex::new(BarrierState { count: 0, generation_id: 0, }), cvar: Condvar::new(), num_threads: n, } } /// Blocks the current thread until all threads have rendezvoused here. /// /// Barriers are re-usable after all threads have rendezvoused once, and can /// be used continuously. /// /// A single (arbitrary) thread will receive a [`BarrierWaitResult`] that /// returns `true` from [`is_leader`] when returning from this function, and /// all other threads will receive a result that will return `false` from /// [`is_leader`]. /// /// [`BarrierWaitResult`]: struct.BarrierWaitResult.html /// [`is_leader`]: struct.BarrierWaitResult.html#method.is_leader /// /// # Examples /// /// ``` /// use std::sync::{Arc, Barrier}; /// use std::thread; /// /// let mut handles = Vec::with_capacity(10); /// let barrier = Arc::new(Barrier::new(10)); /// for _ in 0..10 { /// let c = barrier.clone(); /// // The same messages will be printed together. /// // You will NOT see any interleaving. /// handles.push(thread::spawn(move|| { /// println!("before wait"); /// c.wait(); /// println!("after wait"); /// })); /// } /// // Wait for other threads to finish. /// for handle in handles { /// handle.join().unwrap(); /// } /// ``` #[stable(feature = "rust1", since = "1.0.0")] pub fn wait(&self) -> BarrierWaitResult { let mut lock = self.lock.lock().unwrap(); let local_gen = lock.generation_id; lock.count += 1; if lock.count < self.num_threads { // We need a while loop to guard against spurious wakeups. // http://en.wikipedia.org/wiki/Spurious_wakeup while local_gen == lock.generation_id && lock.count < self.num_threads { lock = self.cvar.wait(lock).unwrap(); } BarrierWaitResult(false) } else { lock.count = 0; lock.generation_id = lock.generation_id.wrapping_add(1); self.cvar.notify_all(); BarrierWaitResult(true) } } } #[stable(feature = "std_debug", since = "1.16.0")] impl fmt::Debug for BarrierWaitResult { fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result { f.debug_struct("BarrierWaitResult") .field("is_leader", &self.is_leader()) .finish() } } impl BarrierWaitResult { /// Returns `true` if this thread from [`wait`] is the "leader thread". /// /// Only one thread will have `true` returned from their result, all other /// threads will have `false` returned. /// /// [`wait`]: struct.Barrier.html#method.wait /// /// # Examples /// /// ``` /// use std::sync::Barrier; /// /// let barrier = Barrier::new(1); /// let barrier_wait_result = barrier.wait(); /// println!("{:?}", barrier_wait_result.is_leader()); /// ``` #[stable(feature = "rust1", since = "1.0.0")] pub fn is_leader(&self) -> bool { self.0 } } #[cfg(test)] mod tests { use crate::sync::{Arc, Barrier}; use crate::sync::mpsc::{channel, TryRecvError}; use crate::thread; #[test] #[cfg_attr(target_os = "emscripten", ignore)] fn test_barrier() { const N: usize = 10; let barrier = Arc::new(Barrier::new(N)); let (tx, rx) = channel(); for _ in 0..N - 1 { let c = barrier.clone(); let tx = tx.clone(); thread::spawn(move|| { tx.send(c.wait().is_leader()).unwrap(); }); } // At this point, all spawned threads should be blocked, // so we shouldn't get anything from the port assert!(match rx.try_recv() { Err(TryRecvError::Empty) => true, _ => false, }); let mut leader_found = barrier.wait().is_leader(); // Now, the barrier is cleared and we should get data. for _ in 0..N - 1 { if rx.recv().unwrap() { assert!(!leader_found); leader_found = true; } } assert!(leader_found); } }