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