src/libstd/sync/condvar.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 prelude::*;
  12
  13 use sync::atomic::{mod, AtomicUint};
  14 use sync::{mutex, StaticMutexGuard};
  15 use sys_common::condvar as sys;
  16 use sys_common::mutex as sys_mutex;
  17 use time::Duration;
  18
  19 /// A Condition Variable
  20 ///
  21 /// Condition variables represent the ability to block a thread such that it
  22 /// consumes no CPU time while waiting for an event to occur. Condition
  23 /// variables are typically associated with a boolean predicate (a condition)
  24 /// and a mutex. The predicate is always verified inside of the mutex before
  25 /// determining that thread must block.
  26 ///
  27 /// Functions in this module will block the current **thread** of execution and
  28 /// are bindings to system-provided condition variables where possible. Note
  29 /// that this module places one additional restriction over the system condition
  30 /// variables: each condvar can be used with precisely one mutex at runtime. Any
  31 /// attempt to use multiple mutexes on the same condition variable will result
  32 /// in a runtime panic. If this is not desired, then the unsafe primitives in
  33 /// `sys` do not have this restriction but may result in undefined behavior.
  34 ///
  35 /// # Example
  36 ///
  37 /// ```
  38 /// use std::sync::{Arc, Mutex, Condvar};
  39 ///
  40 /// let pair = Arc::new((Mutex::new(false), Condvar::new()));
  41 /// let pair2 = pair.clone();
  42 ///
  43 /// // Inside of our lock, spawn a new thread, and then wait for it to start
  44 /// spawn(proc() {
  45 ///     let &(ref lock, ref cvar) = &*pair2;
  46 ///     let mut started = lock.lock();
  47 ///     *started = true;
  48 ///     cvar.notify_one();
  49 /// });
  50 ///
  51 /// // wait for the thread to start up
  52 /// let &(ref lock, ref cvar) = &*pair;
  53 /// let started = lock.lock();
  54 /// while !*started {
  55 ///     cvar.wait(&started);
  56 /// }
  57 /// ```
  58 pub struct Condvar { inner: Box<StaticCondvar> }
  59
  60 /// Statically allocated condition variables.
  61 ///
  62 /// This structure is identical to `Condvar` except that it is suitable for use
  63 /// in static initializers for other structures.
  64 ///
  65 /// # Example
  66 ///
  67 /// ```
  68 /// use std::sync::{StaticCondvar, CONDVAR_INIT};
  69 ///
  70 /// static CVAR: StaticCondvar = CONDVAR_INIT;
  71 /// ```
  72 pub struct StaticCondvar {
  73     inner: sys::Condvar,
  74     mutex: AtomicUint,
  75 }
  76
  77 /// Constant initializer for a statically allocated condition variable.
  78 pub const CONDVAR_INIT: StaticCondvar = StaticCondvar {
  79     inner: sys::CONDVAR_INIT,
  80     mutex: atomic::INIT_ATOMIC_UINT,
  81 };
  82
  83 /// A trait for vaules which can be passed to the waiting methods of condition
  84 /// variables. This is implemented by the mutex guards in this module.
  85 ///
  86 /// Note that this trait should likely not be implemented manually unless you
  87 /// really know what you're doing.
  88 pub trait AsMutexGuard {
  89     #[allow(missing_docs)]
  90     unsafe fn as_mutex_guard(&self) -> &StaticMutexGuard;
  91 }
  92
  93 impl Condvar {
  94     /// Creates a new condition variable which is ready to be waited on and
  95     /// notified.
  96     pub fn new() -> Condvar {
  97         Condvar {
  98             inner: box StaticCondvar {
  99                 inner: unsafe { sys::Condvar::new() },
 100                 mutex: AtomicUint::new(0),
 101             }
 102         }
 103     }
 104
 105     /// Block the current thread until this condition variable receives a
 106     /// notification.
 107     ///
 108     /// This function will atomically unlock the mutex specified (represented by
 109     /// `guard`) and block the current thread. This means that any calls to
 110     /// `notify_*()` which happen logically after the mutex is unlocked are
 111     /// candidates to wake this thread up. When this function call returns, the
 112     /// lock specified will have been re-acquired.
 113     ///
 114     /// Note that this function is susceptible to spurious wakeups. Condition
 115     /// variables normally have a boolean predicate associated with them, and
 116     /// the predicate must always be checked each time this function returns to
 117     /// protect against spurious wakeups.
 118     ///
 119     /// # Panics
 120     ///
 121     /// This function will `panic!()` if it is used with more than one mutex
 122     /// over time. Each condition variable is dynamically bound to exactly one
 123     /// mutex to ensure defined behavior across platforms. If this functionality
 124     /// is not desired, then unsafe primitives in `sys` are provided.
 125     pub fn wait<T: AsMutexGuard>(&self, mutex_guard: &T) {
 126         unsafe {
 127             let me: &'static Condvar = &*(self as *const _);
 128             me.inner.wait(mutex_guard)
 129         }
 130     }
 131
 132     /// Wait on this condition variable for a notification, timing out after a
 133     /// specified duration.
 134     ///
 135     /// The semantics of this function are equivalent to `wait()` except that
 136     /// the thread will be blocked for roughly no longer than `dur`. This method
 137     /// should not be used for precise timing due to anomalies such as
 138     /// preemption or platform differences that may not cause the maximum amount
 139     /// of time waited to be precisely `dur`.
 140     ///
 141     /// If the wait timed out, then `false` will be returned. Otherwise if a
 142     /// notification was received then `true` will be returned.
 143     ///
 144     /// Like `wait`, the lock specified will be re-acquired when this function
 145     /// returns, regardless of whether the timeout elapsed or not.
 146     // Note that this method is *not* public, and this is quite intentional
 147     // because we're not quite sure about the semantics of relative vs absolute
 148     // durations or how the timing guarantees play into what the system APIs
 149     // provide. There are also additional concerns about the unix-specific
 150     // implementation which may need to be addressed.
 151     #[allow(dead_code)]
 152     fn wait_timeout<T: AsMutexGuard>(&self, mutex_guard: &T,
 153                                      dur: Duration) -> bool {
 154         unsafe {
 155             let me: &'static Condvar = &*(self as *const _);
 156             me.inner.wait_timeout(mutex_guard, dur)
 157         }
 158     }
 159
 160     /// Wake up one blocked thread on this condvar.
 161     ///
 162     /// If there is a blocked thread on this condition variable, then it will
 163     /// be woken up from its call to `wait` or `wait_timeout`. Calls to
 164     /// `notify_one` are not buffered in any way.
 165     ///
 166     /// To wake up all threads, see `notify_one()`.
 167     pub fn notify_one(&self) { unsafe { self.inner.inner.notify_one() } }
 168
 169     /// Wake up all blocked threads on this condvar.
 170     ///
 171     /// This method will ensure that any current waiters on the condition
 172     /// variable are awoken. Calls to `notify_all()` are not buffered in any
 173     /// way.
 174     ///
 175     /// To wake up only one thread, see `notify_one()`.
 176     pub fn notify_all(&self) { unsafe { self.inner.inner.notify_all() } }
 177 }
 178
 179 impl Drop for Condvar {
 180     fn drop(&mut self) {
 181         unsafe { self.inner.inner.destroy() }
 182     }
 183 }
 184
 185 impl StaticCondvar {
 186     /// Block the current thread until this condition variable receives a
 187     /// notification.
 188     ///
 189     /// See `Condvar::wait`.
 190     pub fn wait<T: AsMutexGuard>(&'static self, mutex_guard: &T) {
 191         unsafe {
 192             let lock = mutex_guard.as_mutex_guard();
 193             let sys = mutex::guard_lock(lock);
 194             self.verify(sys);
 195             self.inner.wait(sys);
 196             (*mutex::guard_poison(lock)).check("mutex");
 197         }
 198     }
 199
 200     /// Wait on this condition variable for a notification, timing out after a
 201     /// specified duration.
 202     ///
 203     /// See `Condvar::wait_timeout`.
 204     #[allow(dead_code)] // may want to stabilize this later, see wait_timeout above
 205     fn wait_timeout<T: AsMutexGuard>(&'static self, mutex_guard: &T,
 206                                      dur: Duration) -> bool {
 207         unsafe {
 208             let lock = mutex_guard.as_mutex_guard();
 209             let sys = mutex::guard_lock(lock);
 210             self.verify(sys);
 211             let ret = self.inner.wait_timeout(sys, dur);
 212             (*mutex::guard_poison(lock)).check("mutex");
 213             return ret;
 214         }
 215     }
 216
 217     /// Wake up one blocked thread on this condvar.
 218     ///
 219     /// See `Condvar::notify_one`.
 220     pub fn notify_one(&'static self) { unsafe { self.inner.notify_one() } }
 221
 222     /// Wake up all blocked threads on this condvar.
 223     ///
 224     /// See `Condvar::notify_all`.
 225     pub fn notify_all(&'static self) { unsafe { self.inner.notify_all() } }
 226
 227     /// Deallocate all resources associated with this static condvar.
 228     ///
 229     /// This method is unsafe to call as there is no guarantee that there are no
 230     /// active users of the condvar, and this also doesn't prevent any future
 231     /// users of the condvar. This method is required to be called to not leak
 232     /// memory on all platforms.
 233     pub unsafe fn destroy(&'static self) {
 234         self.inner.destroy()
 235     }
 236
 237     fn verify(&self, mutex: &sys_mutex::Mutex) {
 238         let addr = mutex as *const _ as uint;
 239         match self.mutex.compare_and_swap(0, addr, atomic::SeqCst) {
 240             // If we got out 0, then we have successfully bound the mutex to
 241             // this cvar.
 242             0 => {}
 243
 244             // If we get out a value that's the same as `addr`, then someone
 245             // already beat us to the punch.
 246             n if n == addr => {}
 247
 248             // Anything else and we're using more than one mutex on this cvar,
 249             // which is currently disallowed.
 250             _ => panic!("attempted to use a condition variable with two \
 251                          mutexes"),
 252         }
 253     }
 254 }
 255
 256 #[cfg(test)]
 257 mod tests {
 258     use prelude::*;
 259
 260     use time::Duration;
 261     use super::{StaticCondvar, CONDVAR_INIT};
 262     use sync::{StaticMutex, MUTEX_INIT, Condvar, Mutex, Arc};
 263
 264     #[test]
 265     fn smoke() {
 266         let c = Condvar::new();
 267         c.notify_one();
 268         c.notify_all();
 269     }
 270
 271     #[test]
 272     fn static_smoke() {
 273         static C: StaticCondvar = CONDVAR_INIT;
 274         C.notify_one();
 275         C.notify_all();
 276         unsafe { C.destroy(); }
 277     }
 278
 279     #[test]
 280     fn notify_one() {
 281         static C: StaticCondvar = CONDVAR_INIT;
 282         static M: StaticMutex = MUTEX_INIT;
 283
 284         let g = M.lock();
 285         spawn(proc() {
 286             let _g = M.lock();
 287             C.notify_one();
 288         });
 289         C.wait(&g);
 290         drop(g);
 291         unsafe { C.destroy(); M.destroy(); }
 292     }
 293
 294     #[test]
 295     fn notify_all() {
 296         const N: uint = 10;
 297
 298         let data = Arc::new((Mutex::new(0), Condvar::new()));
 299         let (tx, rx) = channel();
 300         for _ in range(0, N) {
 301             let data = data.clone();
 302             let tx = tx.clone();
 303             spawn(proc() {
 304                 let &(ref lock, ref cond) = &*data;
 305                 let mut cnt = lock.lock();
 306                 *cnt += 1;
 307                 if *cnt == N {
 308                     tx.send(());
 309                 }
 310                 while *cnt != 0 {
 311                     cond.wait(&cnt);
 312                 }
 313                 tx.send(());
 314             });
 315         }
 316         drop(tx);
 317
 318         let &(ref lock, ref cond) = &*data;
 319         rx.recv();
 320         let mut cnt = lock.lock();
 321         *cnt = 0;
 322         cond.notify_all();
 323         drop(cnt);
 324
 325         for _ in range(0, N) {
 326             rx.recv();
 327         }
 328     }
 329
 330     #[test]
 331     fn wait_timeout() {
 332         static C: StaticCondvar = CONDVAR_INIT;
 333         static M: StaticMutex = MUTEX_INIT;
 334
 335         let g = M.lock();
 336         assert!(!C.wait_timeout(&g, Duration::nanoseconds(1000)));
 337         spawn(proc() {
 338             let _g = M.lock();
 339             C.notify_one();
 340         });
 341         assert!(C.wait_timeout(&g, Duration::days(1)));
 342         drop(g);
 343         unsafe { C.destroy(); M.destroy(); }
 344     }
 345
 346     #[test]
 347     #[should_fail]
 348     fn two_mutexes() {
 349         static M1: StaticMutex = MUTEX_INIT;
 350         static M2: StaticMutex = MUTEX_INIT;
 351         static C: StaticCondvar = CONDVAR_INIT;
 352
 353         let g = M1.lock();
 354         spawn(proc() {
 355             let _g = M1.lock();
 356             C.notify_one();
 357         });
 358         C.wait(&g);
 359         drop(g);
 360
 361         C.wait(&M2.lock());
 362
 363     }
 364 }
 365