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