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