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