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 fmt;
  12 use sync::atomic::{AtomicUsize, Ordering};
  13 use sync::{mutex, MutexGuard, PoisonError};
  14 use sys_common::condvar as sys;
  15 use sys_common::mutex as sys_mutex;
  16 use sys_common::poison::{self, LockResult};
  17 use time::Duration;
  18
  19 /// A type indicating whether a timed wait on a condition variable returned
  20 /// due to a time out or not.
  21 ///
  22 /// It is returned by the [`wait_timeout`] method.
  23 ///
  24 /// [`wait_timeout`]: struct.Condvar.html#method.wait_timeout
  25 #[derive(Debug, PartialEq, Eq, Copy, Clone)]
  26 #[stable(feature = "wait_timeout", since = "1.5.0")]
  27 pub struct WaitTimeoutResult(bool);
  28
  29 impl WaitTimeoutResult {
  30     /// Returns whether the wait was known to have timed out.
  31     ///
  32     /// # Examples
  33     ///
  34     /// This example spawns a thread which will update the boolean value and
  35     /// then wait 100 milliseconds before notifying the condvar.
  36     ///
  37     /// The main thread will wait with a timeout on the condvar and then leave
  38     /// once the boolean has been updated and notified.
  39     ///
  40     /// ```
  41     /// use std::sync::{Arc, Mutex, Condvar};
  42     /// use std::thread;
  43     /// use std::time::Duration;
  44     ///
  45     /// let pair = Arc::new((Mutex::new(false), Condvar::new()));
  46     /// let pair2 = pair.clone();
  47     ///
  48     /// thread::spawn(move|| {
  49     ///     let &(ref lock, ref cvar) = &*pair2;
  50     ///     let mut started = lock.lock().unwrap();
  51     ///     // We update the boolean value.
  52     ///     *started = true;
  53     ///     // Let's wait 20 milliseconds before notifying the condvar.
  54     ///     thread::sleep(Duration::from_millis(20));
  55     ///     cvar.notify_one();
  56     /// });
  57     ///
  58     /// // Wait for the thread to start up.
  59     /// let &(ref lock, ref cvar) = &*pair;
  60     /// let mut started = lock.lock().unwrap();
  61     /// loop {
  62     ///     // Let's put a timeout on the condvar's wait.
  63     ///     let result = cvar.wait_timeout(started, Duration::from_millis(10)).unwrap();
  64     ///     // 10 milliseconds have passed, or maybe the value changed!
  65     ///     started = result.0;
  66     ///     if *started == true {
  67     ///         // We received the notification and the value has been updated, we can leave.
  68     ///         break
  69     ///     }
  70     /// }
  71     /// ```
  72     #[stable(feature = "wait_timeout", since = "1.5.0")]
  73     pub fn timed_out(&self) -> bool {
  74         self.0
  75     }
  76 }
  77
  78 /// A Condition Variable
  79 ///
  80 /// Condition variables represent the ability to block a thread such that it
  81 /// consumes no CPU time while waiting for an event to occur. Condition
  82 /// variables are typically associated with a boolean predicate (a condition)
  83 /// and a mutex. The predicate is always verified inside of the mutex before
  84 /// determining that a thread must block.
  85 ///
  86 /// Functions in this module will block the current **thread** of execution and
  87 /// are bindings to system-provided condition variables where possible. Note
  88 /// that this module places one additional restriction over the system condition
  89 /// variables: each condvar can be used with precisely one mutex at runtime. Any
  90 /// attempt to use multiple mutexes on the same condition variable will result
  91 /// in a runtime panic. If this is not desired, then the unsafe primitives in
  92 /// `sys` do not have this restriction but may result in undefined behavior.
  93 ///
  94 /// # Examples
  95 ///
  96 /// ```
  97 /// use std::sync::{Arc, Mutex, Condvar};
  98 /// use std::thread;
  99 ///
 100 /// let pair = Arc::new((Mutex::new(false), Condvar::new()));
 101 /// let pair2 = pair.clone();
 102 ///
 103 /// // Inside of our lock, spawn a new thread, and then wait for it to start.
 104 /// thread::spawn(move|| {
 105 ///     let &(ref lock, ref cvar) = &*pair2;
 106 ///     let mut started = lock.lock().unwrap();
 107 ///     *started = true;
 108 ///     // We notify the condvar that the value has changed.
 109 ///     cvar.notify_one();
 110 /// });
 111 ///
 112 /// // Wait for the thread to start up.
 113 /// let &(ref lock, ref cvar) = &*pair;
 114 /// let mut started = lock.lock().unwrap();
 115 /// while !*started {
 116 ///     started = cvar.wait(started).unwrap();
 117 /// }
 118 /// ```
 119 #[stable(feature = "rust1", since = "1.0.0")]
 120 pub struct Condvar {
 121     inner: Box<sys::Condvar>,
 122     mutex: AtomicUsize,
 123 }
 124
 125 impl Condvar {
 126     /// Creates a new condition variable which is ready to be waited on and
 127     /// notified.
 128     ///
 129     /// # Examples
 130     ///
 131     /// ```
 132     /// use std::sync::Condvar;
 133     ///
 134     /// let condvar = Condvar::new();
 135     /// ```
 136     #[stable(feature = "rust1", since = "1.0.0")]
 137     pub fn new() -> Condvar {
 138         let mut c = Condvar {
 139             inner: box sys::Condvar::new(),
 140             mutex: AtomicUsize::new(0),
 141         };
 142         unsafe {
 143             c.inner.init();
 144         }
 145         c
 146     }
 147
 148     /// Blocks the current thread until this condition variable receives a
 149     /// notification.
 150     ///
 151     /// This function will atomically unlock the mutex specified (represented by
 152     /// `guard`) and block the current thread. This means that any calls
 153     /// to [`notify_one`] or [`notify_all`] which happen logically after the
 154     /// mutex is unlocked are candidates to wake this thread up. When this
 155     /// function call returns, the lock specified will have been re-acquired.
 156     ///
 157     /// Note that this function is susceptible to spurious wakeups. Condition
 158     /// variables normally have a boolean predicate associated with them, and
 159     /// the predicate must always be checked each time this function returns to
 160     /// protect against spurious wakeups.
 161     ///
 162     /// # Errors
 163     ///
 164     /// This function will return an error if the mutex being waited on is
 165     /// poisoned when this thread re-acquires the lock. For more information,
 166     /// see information about [poisoning] on the [`Mutex`] type.
 167     ///
 168     /// # Panics
 169     ///
 170     /// This function will [`panic!`] if it is used with more than one mutex
 171     /// over time. Each condition variable is dynamically bound to exactly one
 172     /// mutex to ensure defined behavior across platforms. If this functionality
 173     /// is not desired, then unsafe primitives in `sys` are provided.
 174     ///
 175     /// [`notify_one`]: #method.notify_one
 176     /// [`notify_all`]: #method.notify_all
 177     /// [poisoning]: ../sync/struct.Mutex.html#poisoning
 178     /// [`Mutex`]: ../sync/struct.Mutex.html
 179     /// [`panic!`]: ../../std/macro.panic.html
 180     ///
 181     /// # Examples
 182     ///
 183     /// ```
 184     /// use std::sync::{Arc, Mutex, Condvar};
 185     /// use std::thread;
 186     ///
 187     /// let pair = Arc::new((Mutex::new(false), Condvar::new()));
 188     /// let pair2 = pair.clone();
 189     ///
 190     /// thread::spawn(move|| {
 191     ///     let &(ref lock, ref cvar) = &*pair2;
 192     ///     let mut started = lock.lock().unwrap();
 193     ///     *started = true;
 194     ///     // We notify the condvar that the value has changed.
 195     ///     cvar.notify_one();
 196     /// });
 197     ///
 198     /// // Wait for the thread to start up.
 199     /// let &(ref lock, ref cvar) = &*pair;
 200     /// let mut started = lock.lock().unwrap();
 201     /// // As long as the value inside the `Mutex` is false, we wait.
 202     /// while !*started {
 203     ///     started = cvar.wait(started).unwrap();
 204     /// }
 205     /// ```
 206     #[stable(feature = "rust1", since = "1.0.0")]
 207     pub fn wait<'a, T>(&self, guard: MutexGuard<'a, T>)
 208                        -> LockResult<MutexGuard<'a, T>> {
 209         let poisoned = unsafe {
 210             let lock = mutex::guard_lock(&guard);
 211             self.verify(lock);
 212             self.inner.wait(lock);
 213             mutex::guard_poison(&guard).get()
 214         };
 215         if poisoned {
 216             Err(PoisonError::new(guard))
 217         } else {
 218             Ok(guard)
 219         }
 220     }
 221
 222     /// Waits on this condition variable for a notification, timing out after a
 223     /// specified duration.
 224     ///
 225     /// The semantics of this function are equivalent to [`wait`]
 226     /// except that the thread will be blocked for roughly no longer
 227     /// than `ms` milliseconds. This method should not be used for
 228     /// precise timing due to anomalies such as preemption or platform
 229     /// differences that may not cause the maximum amount of time
 230     /// waited to be precisely `ms`.
 231     ///
 232     /// Note that the best effort is made to ensure that the time waited is
 233     /// measured with a monotonic clock, and not affected by the changes made to
 234     /// the system time.
 235     ///
 236     /// The returned boolean is `false` only if the timeout is known
 237     /// to have elapsed.
 238     ///
 239     /// Like [`wait`], the lock specified will be re-acquired when this function
 240     /// returns, regardless of whether the timeout elapsed or not.
 241     ///
 242     /// [`wait`]: #method.wait
 243     ///
 244     /// # Examples
 245     ///
 246     /// ```
 247     /// use std::sync::{Arc, Mutex, Condvar};
 248     /// use std::thread;
 249     ///
 250     /// let pair = Arc::new((Mutex::new(false), Condvar::new()));
 251     /// let pair2 = pair.clone();
 252     ///
 253     /// thread::spawn(move|| {
 254     ///     let &(ref lock, ref cvar) = &*pair2;
 255     ///     let mut started = lock.lock().unwrap();
 256     ///     *started = true;
 257     ///     // We notify the condvar that the value has changed.
 258     ///     cvar.notify_one();
 259     /// });
 260     ///
 261     /// // Wait for the thread to start up.
 262     /// let &(ref lock, ref cvar) = &*pair;
 263     /// let mut started = lock.lock().unwrap();
 264     /// // As long as the value inside the `Mutex` is false, we wait.
 265     /// loop {
 266     ///     let result = cvar.wait_timeout_ms(started, 10).unwrap();
 267     ///     // 10 milliseconds have passed, or maybe the value changed!
 268     ///     started = result.0;
 269     ///     if *started == true {
 270     ///         // We received the notification and the value has been updated, we can leave.
 271     ///         break
 272     ///     }
 273     /// }
 274     /// ```
 275     #[stable(feature = "rust1", since = "1.0.0")]
 276     #[rustc_deprecated(since = "1.6.0", reason = "replaced by `std::sync::Condvar::wait_timeout`")]
 277     pub fn wait_timeout_ms<'a, T>(&self, guard: MutexGuard<'a, T>, ms: u32)
 278                                   -> LockResult<(MutexGuard<'a, T>, bool)> {
 279         let res = self.wait_timeout(guard, Duration::from_millis(ms as u64));
 280         poison::map_result(res, |(a, b)| {
 281             (a, !b.timed_out())
 282         })
 283     }
 284
 285     /// Waits on this condition variable for a notification, timing out after a
 286     /// specified duration.
 287     ///
 288     /// The semantics of this function are equivalent to [`wait`] except that
 289     /// the thread will be blocked for roughly no longer than `dur`. This
 290     /// method should not be used for precise timing due to anomalies such as
 291     /// preemption or platform differences that may not cause the maximum
 292     /// amount of time waited to be precisely `dur`.
 293     ///
 294     /// Note that the best effort is made to ensure that the time waited is
 295     /// measured with a monotonic clock, and not affected by the changes made to
 296     /// the system time.
 297     ///
 298     /// The returned [`WaitTimeoutResult`] value indicates if the timeout is
 299     /// known to have elapsed.
 300     ///
 301     /// Like [`wait`], the lock specified will be re-acquired when this function
 302     /// returns, regardless of whether the timeout elapsed or not.
 303     ///
 304     /// [`wait`]: #method.wait
 305     /// [`WaitTimeoutResult`]: struct.WaitTimeoutResult.html
 306     ///
 307     /// # Examples
 308     ///
 309     /// ```
 310     /// use std::sync::{Arc, Mutex, Condvar};
 311     /// use std::thread;
 312     /// use std::time::Duration;
 313     ///
 314     /// let pair = Arc::new((Mutex::new(false), Condvar::new()));
 315     /// let pair2 = pair.clone();
 316     ///
 317     /// thread::spawn(move|| {
 318     ///     let &(ref lock, ref cvar) = &*pair2;
 319     ///     let mut started = lock.lock().unwrap();
 320     ///     *started = true;
 321     ///     // We notify the condvar that the value has changed.
 322     ///     cvar.notify_one();
 323     /// });
 324     ///
 325     /// // wait for the thread to start up
 326     /// let &(ref lock, ref cvar) = &*pair;
 327     /// let mut started = lock.lock().unwrap();
 328     /// // as long as the value inside the `Mutex` is false, we wait
 329     /// loop {
 330     ///     let result = cvar.wait_timeout(started, Duration::from_millis(10)).unwrap();
 331     ///     // 10 milliseconds have passed, or maybe the value changed!
 332     ///     started = result.0;
 333     ///     if *started == true {
 334     ///         // We received the notification and the value has been updated, we can leave.
 335     ///         break
 336     ///     }
 337     /// }
 338     /// ```
 339     #[stable(feature = "wait_timeout", since = "1.5.0")]
 340     pub fn wait_timeout<'a, T>(&self, guard: MutexGuard<'a, T>,
 341                                dur: Duration)
 342                                -> LockResult<(MutexGuard<'a, T>, WaitTimeoutResult)> {
 343         let (poisoned, result) = unsafe {
 344             let lock = mutex::guard_lock(&guard);
 345             self.verify(lock);
 346             let success = self.inner.wait_timeout(lock, dur);
 347             (mutex::guard_poison(&guard).get(), WaitTimeoutResult(!success))
 348         };
 349         if poisoned {
 350             Err(PoisonError::new((guard, result)))
 351         } else {
 352             Ok((guard, result))
 353         }
 354     }
 355
 356     /// Wakes up one blocked thread on this condvar.
 357     ///
 358     /// If there is a blocked thread on this condition variable, then it will
 359     /// be woken up from its call to [`wait`] or [`wait_timeout`]. Calls to
 360     /// `notify_one` are not buffered in any way.
 361     ///
 362     /// To wake up all threads, see [`notify_all`].
 363     ///
 364     /// [`wait`]: #method.wait
 365     /// [`wait_timeout`]: #method.wait_timeout
 366     /// [`notify_all`]: #method.notify_all
 367     ///
 368     /// # Examples
 369     ///
 370     /// ```
 371     /// use std::sync::{Arc, Mutex, Condvar};
 372     /// use std::thread;
 373     ///
 374     /// let pair = Arc::new((Mutex::new(false), Condvar::new()));
 375     /// let pair2 = pair.clone();
 376     ///
 377     /// thread::spawn(move|| {
 378     ///     let &(ref lock, ref cvar) = &*pair2;
 379     ///     let mut started = lock.lock().unwrap();
 380     ///     *started = true;
 381     ///     // We notify the condvar that the value has changed.
 382     ///     cvar.notify_one();
 383     /// });
 384     ///
 385     /// // Wait for the thread to start up.
 386     /// let &(ref lock, ref cvar) = &*pair;
 387     /// let mut started = lock.lock().unwrap();
 388     /// // As long as the value inside the `Mutex` is false, we wait.
 389     /// while !*started {
 390     ///     started = cvar.wait(started).unwrap();
 391     /// }
 392     /// ```
 393     #[stable(feature = "rust1", since = "1.0.0")]
 394     pub fn notify_one(&self) {
 395         unsafe { self.inner.notify_one() }
 396     }
 397
 398     /// Wakes up all blocked threads on this condvar.
 399     ///
 400     /// This method will ensure that any current waiters on the condition
 401     /// variable are awoken. Calls to `notify_all()` are not buffered in any
 402     /// way.
 403     ///
 404     /// To wake up only one thread, see [`notify_one`].
 405     ///
 406     /// [`notify_one`]: #method.notify_one
 407     ///
 408     /// # Examples
 409     ///
 410     /// ```
 411     /// use std::sync::{Arc, Mutex, Condvar};
 412     /// use std::thread;
 413     ///
 414     /// let pair = Arc::new((Mutex::new(false), Condvar::new()));
 415     /// let pair2 = pair.clone();
 416     ///
 417     /// thread::spawn(move|| {
 418     ///     let &(ref lock, ref cvar) = &*pair2;
 419     ///     let mut started = lock.lock().unwrap();
 420     ///     *started = true;
 421     ///     // We notify the condvar that the value has changed.
 422     ///     cvar.notify_all();
 423     /// });
 424     ///
 425     /// // Wait for the thread to start up.
 426     /// let &(ref lock, ref cvar) = &*pair;
 427     /// let mut started = lock.lock().unwrap();
 428     /// // As long as the value inside the `Mutex` is false, we wait.
 429     /// while !*started {
 430     ///     started = cvar.wait(started).unwrap();
 431     /// }
 432     /// ```
 433     #[stable(feature = "rust1", since = "1.0.0")]
 434     pub fn notify_all(&self) {
 435         unsafe { self.inner.notify_all() }
 436     }
 437
 438     fn verify(&self, mutex: &sys_mutex::Mutex) {
 439         let addr = mutex as *const _ as usize;
 440         match self.mutex.compare_and_swap(0, addr, Ordering::SeqCst) {
 441             // If we got out 0, then we have successfully bound the mutex to
 442             // this cvar.
 443             0 => {}
 444
 445             // If we get out a value that's the same as `addr`, then someone
 446             // already beat us to the punch.
 447             n if n == addr => {}
 448
 449             // Anything else and we're using more than one mutex on this cvar,
 450             // which is currently disallowed.
 451             _ => panic!("attempted to use a condition variable with two \
 452                          mutexes"),
 453         }
 454     }
 455 }
 456
 457 #[stable(feature = "std_debug", since = "1.16.0")]
 458 impl fmt::Debug for Condvar {
 459     fn fmt(&self, f: &mut fmt::Formatter) -> fmt::Result {
 460         f.pad("Condvar { .. }")
 461     }
 462 }
 463
 464 #[stable(feature = "condvar_default", since = "1.10.0")]
 465 impl Default for Condvar {
 466     /// Creates a `Condvar` which is ready to be waited on and notified.
 467     fn default() -> Condvar {
 468         Condvar::new()
 469     }
 470 }
 471
 472 #[stable(feature = "rust1", since = "1.0.0")]
 473 impl Drop for Condvar {
 474     fn drop(&mut self) {
 475         unsafe { self.inner.destroy() }
 476     }
 477 }
 478
 479 #[cfg(test)]
 480 mod tests {
 481     use sync::mpsc::channel;
 482     use sync::{Condvar, Mutex, Arc};
 483     use thread;
 484     use time::Duration;
 485     use u32;
 486
 487     #[test]
 488     fn smoke() {
 489         let c = Condvar::new();
 490         c.notify_one();
 491         c.notify_all();
 492     }
 493
 494     #[test]
 495     #[cfg_attr(target_os = "emscripten", ignore)]
 496     fn notify_one() {
 497         let m = Arc::new(Mutex::new(()));
 498         let m2 = m.clone();
 499         let c = Arc::new(Condvar::new());
 500         let c2 = c.clone();
 501
 502         let g = m.lock().unwrap();
 503         let _t = thread::spawn(move|| {
 504             let _g = m2.lock().unwrap();
 505             c2.notify_one();
 506         });
 507         let g = c.wait(g).unwrap();
 508         drop(g);
 509     }
 510
 511     #[test]
 512     #[cfg_attr(target_os = "emscripten", ignore)]
 513     fn notify_all() {
 514         const N: usize = 10;
 515
 516         let data = Arc::new((Mutex::new(0), Condvar::new()));
 517         let (tx, rx) = channel();
 518         for _ in 0..N {
 519             let data = data.clone();
 520             let tx = tx.clone();
 521             thread::spawn(move|| {
 522                 let &(ref lock, ref cond) = &*data;
 523                 let mut cnt = lock.lock().unwrap();
 524                 *cnt += 1;
 525                 if *cnt == N {
 526                     tx.send(()).unwrap();
 527                 }
 528                 while *cnt != 0 {
 529                     cnt = cond.wait(cnt).unwrap();
 530                 }
 531                 tx.send(()).unwrap();
 532             });
 533         }
 534         drop(tx);
 535
 536         let &(ref lock, ref cond) = &*data;
 537         rx.recv().unwrap();
 538         let mut cnt = lock.lock().unwrap();
 539         *cnt = 0;
 540         cond.notify_all();
 541         drop(cnt);
 542
 543         for _ in 0..N {
 544             rx.recv().unwrap();
 545         }
 546     }
 547
 548     #[test]
 549     #[cfg_attr(target_os = "emscripten", ignore)]
 550     fn wait_timeout_ms() {
 551         let m = Arc::new(Mutex::new(()));
 552         let m2 = m.clone();
 553         let c = Arc::new(Condvar::new());
 554         let c2 = c.clone();
 555
 556         let g = m.lock().unwrap();
 557         let (g, _no_timeout) = c.wait_timeout(g, Duration::from_millis(1)).unwrap();
 558         // spurious wakeups mean this isn't necessarily true
 559         // assert!(!no_timeout);
 560         let _t = thread::spawn(move || {
 561             let _g = m2.lock().unwrap();
 562             c2.notify_one();
 563         });
 564         let (g, timeout_res) = c.wait_timeout(g, Duration::from_millis(u32::MAX as u64)).unwrap();
 565         assert!(!timeout_res.timed_out());
 566         drop(g);
 567     }
 568
 569     #[test]
 570     #[should_panic]
 571     #[cfg_attr(target_os = "emscripten", ignore)]
 572     fn two_mutexes() {
 573         let m = Arc::new(Mutex::new(()));
 574         let m2 = m.clone();
 575         let c = Arc::new(Condvar::new());
 576         let c2 = c.clone();
 577
 578         let mut g = m.lock().unwrap();
 579         let _t = thread::spawn(move|| {
 580             let _g = m2.lock().unwrap();
 581             c2.notify_one();
 582         });
 583         g = c.wait(g).unwrap();
 584         drop(g);
 585
 586         let m = Mutex::new(());
 587         let _ = c.wait(m.lock().unwrap()).unwrap();
 588     }
 589 }