src/libstd/sync/semaphore.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 ops::Drop;
  12 use sync::{Mutex, Condvar};
  13
  14 /// A counting, blocking, semaphore.
  15 ///
  16 /// Semaphores are a form of atomic counter where access is only granted if the
  17 /// counter is a positive value. Each acquisition will block the calling thread
  18 /// until the counter is positive, and each release will increment the counter
  19 /// and unblock any threads if necessary.
  20 ///
  21 /// # Example
  22 ///
  23 /// ```
  24 /// use std::sync::Semaphore;
  25 ///
  26 /// // Create a semaphore that represents 5 resources
  27 /// let sem = Semaphore::new(5);
  28 ///
  29 /// // Acquire one of the resources
  30 /// sem.acquire();
  31 ///
  32 /// // Acquire one of the resources for a limited period of time
  33 /// {
  34 ///     let _guard = sem.access();
  35 ///     // ...
  36 /// } // resources is released here
  37 ///
  38 /// // Release our initially acquired resource
  39 /// sem.release();
  40 /// ```
  41 pub struct Semaphore {
  42     lock: Mutex<int>,
  43     cvar: Condvar,
  44 }
  45
  46 /// An RAII guard which will release a resource acquired from a semaphore when
  47 /// dropped.
  48 pub struct SemaphoreGuard<'a> {
  49     sem: &'a Semaphore,
  50 }
  51
  52 impl Semaphore {
  53     /// Creates a new semaphore with the initial count specified.
  54     ///
  55     /// The count specified can be thought of as a number of resources, and a
  56     /// call to `acquire` or `access` will block until at least one resource is
  57     /// available. It is valid to initialize a semaphore with a negative count.
  58     pub fn new(count: int) -> Semaphore {
  59         Semaphore {
  60             lock: Mutex::new(count),
  61             cvar: Condvar::new(),
  62         }
  63     }
  64
  65     /// Acquires a resource of this semaphore, blocking the current thread until
  66     /// it can do so.
  67     ///
  68     /// This method will block until the internal count of the semaphore is at
  69     /// least 1.
  70     pub fn acquire(&self) {
  71         let mut count = self.lock.lock().unwrap();
  72         while *count <= 0 {
  73             count = self.cvar.wait(count).unwrap();
  74         }
  75         *count -= 1;
  76     }
  77
  78     /// Release a resource from this semaphore.
  79     ///
  80     /// This will increment the number of resources in this semaphore by 1 and
  81     /// will notify any pending waiters in `acquire` or `access` if necessary.
  82     pub fn release(&self) {
  83         *self.lock.lock().unwrap() += 1;
  84         self.cvar.notify_one();
  85     }
  86
  87     /// Acquires a resource of this semaphore, returning an RAII guard to
  88     /// release the semaphore when dropped.
  89     ///
  90     /// This function is semantically equivalent to an `acquire` followed by a
  91     /// `release` when the guard returned is dropped.
  92     pub fn access(&self) -> SemaphoreGuard {
  93         self.acquire();
  94         SemaphoreGuard { sem: self }
  95     }
  96 }
  97
  98 #[unsafe_destructor]
  99 impl<'a> Drop for SemaphoreGuard<'a> {
 100     fn drop(&mut self) {
 101         self.sem.release();
 102     }
 103 }
 104
 105 #[cfg(test)]
 106 mod tests {
 107     use prelude::v1::*;
 108
 109     use sync::Arc;
 110     use super::Semaphore;
 111     use sync::mpsc::channel;
 112     use thread::Thread;
 113
 114     #[test]
 115     fn test_sem_acquire_release() {
 116         let s = Semaphore::new(1);
 117         s.acquire();
 118         s.release();
 119         s.acquire();
 120     }
 121
 122     #[test]
 123     fn test_sem_basic() {
 124         let s = Semaphore::new(1);
 125         let _g = s.access();
 126     }
 127
 128     #[test]
 129     fn test_sem_as_mutex() {
 130         let s = Arc::new(Semaphore::new(1));
 131         let s2 = s.clone();
 132         let _t = Thread::spawn(move|| {
 133             let _g = s2.access();
 134         });
 135         let _g = s.access();
 136     }
 137
 138     #[test]
 139     fn test_sem_as_cvar() {
 140         /* Child waits and parent signals */
 141         let (tx, rx) = channel();
 142         let s = Arc::new(Semaphore::new(0));
 143         let s2 = s.clone();
 144         let _t = Thread::spawn(move|| {
 145             s2.acquire();
 146             tx.send(()).unwrap();
 147         });
 148         s.release();
 149         let _ = rx.recv();
 150
 151         /* Parent waits and child signals */
 152         let (tx, rx) = channel();
 153         let s = Arc::new(Semaphore::new(0));
 154         let s2 = s.clone();
 155         let _t = Thread::spawn(move|| {
 156             s2.release();
 157             let _ = rx.recv();
 158         });
 159         s.acquire();
 160         tx.send(()).unwrap();
 161     }
 162
 163     #[test]
 164     fn test_sem_multi_resource() {
 165         // Parent and child both get in the critical section at the same
 166         // time, and shake hands.
 167         let s = Arc::new(Semaphore::new(2));
 168         let s2 = s.clone();
 169         let (tx1, rx1) = channel();
 170         let (tx2, rx2) = channel();
 171         let _t = Thread::spawn(move|| {
 172             let _g = s2.access();
 173             let _ = rx2.recv();
 174             tx1.send(()).unwrap();
 175         });
 176         let _g = s.access();
 177         tx2.send(()).unwrap();
 178         rx1.recv().unwrap();
 179     }
 180
 181     #[test]
 182     fn test_sem_runtime_friendly_blocking() {
 183         let s = Arc::new(Semaphore::new(1));
 184         let s2 = s.clone();
 185         let (tx, rx) = channel();
 186         {
 187             let _g = s.access();
 188             Thread::spawn(move|| {
 189                 tx.send(()).unwrap();
 190                 drop(s2.access());
 191                 tx.send(()).unwrap();
 192             }).detach();
 193             rx.recv().unwrap(); // wait for child to come alive
 194         }
 195         rx.recv().unwrap(); // wait for child to be done
 196     }
 197 }