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