src/libsync/one.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 //! A "once initialization" primitive
  12 //!
  13 //! This primitive is meant to be used to run one-time initialization. An
  14 //! example use case would be for initializing an FFI library.
  15
  16 use core::prelude::*;
  17
  18 use core::int;
  19 use core::atomics;
  20
  21 use mutex::{StaticMutex, MUTEX_INIT};
  22
  23 /// A synchronization primitive which can be used to run a one-time global
  24 /// initialization. Useful for one-time initialization for FFI or related
  25 /// functionality. This type can only be constructed with the `ONCE_INIT`
  26 /// value.
  27 ///
  28 /// # Example
  29 ///
  30 /// ```rust
  31 /// use sync::one::{Once, ONCE_INIT};
  32 ///
  33 /// static mut START: Once = ONCE_INIT;
  34 ///
  35 /// unsafe {
  36 ///     START.doit(|| {
  37 ///         // run initialization here
  38 ///     });
  39 /// }
  40 /// ```
  41 pub struct Once {
  42     mutex: StaticMutex,
  43     cnt: atomics::AtomicInt,
  44     lock_cnt: atomics::AtomicInt,
  45 }
  46
  47 /// Initialization value for static `Once` values.
  48 pub static ONCE_INIT: Once = Once {
  49     mutex: MUTEX_INIT,
  50     cnt: atomics::INIT_ATOMIC_INT,
  51     lock_cnt: atomics::INIT_ATOMIC_INT,
  52 };
  53
  54 impl Once {
  55     /// Perform an initialization routine once and only once. The given closure
  56     /// will be executed if this is the first time `doit` has been called, and
  57     /// otherwise the routine will *not* be invoked.
  58     ///
  59     /// This method will block the calling task if another initialization
  60     /// routine is currently running.
  61     ///
  62     /// When this function returns, it is guaranteed that some initialization
  63     /// has run and completed (it may not be the closure specified).
  64     pub fn doit(&self, f: ||) {
  65         // Optimize common path: load is much cheaper than fetch_add.
  66         if self.cnt.load(atomics::SeqCst) < 0 {
  67             return
  68         }
  69
  70         // Implementation-wise, this would seem like a fairly trivial primitive.
  71         // The stickler part is where our mutexes currently require an
  72         // allocation, and usage of a `Once` shouldn't leak this allocation.
  73         //
  74         // This means that there must be a deterministic destroyer of the mutex
  75         // contained within (because it's not needed after the initialization
  76         // has run).
  77         //
  78         // The general scheme here is to gate all future threads once
  79         // initialization has completed with a "very negative" count, and to
  80         // allow through threads to lock the mutex if they see a non negative
  81         // count. For all threads grabbing the mutex, exactly one of them should
  82         // be responsible for unlocking the mutex, and this should only be done
  83         // once everyone else is done with the mutex.
  84         //
  85         // This atomicity is achieved by swapping a very negative value into the
  86         // shared count when the initialization routine has completed. This will
  87         // read the number of threads which will at some point attempt to
  88         // acquire the mutex. This count is then squirreled away in a separate
  89         // variable, and the last person on the way out of the mutex is then
  90         // responsible for destroying the mutex.
  91         //
  92         // It is crucial that the negative value is swapped in *after* the
  93         // initialization routine has completed because otherwise new threads
  94         // calling `doit` will return immediately before the initialization has
  95         // completed.
  96
  97         let prev = self.cnt.fetch_add(1, atomics::SeqCst);
  98         if prev < 0 {
  99             // Make sure we never overflow, we'll never have int::MIN
 100             // simultaneous calls to `doit` to make this value go back to 0
 101             self.cnt.store(int::MIN, atomics::SeqCst);
 102             return
 103         }
 104
 105         // If the count is negative, then someone else finished the job,
 106         // otherwise we run the job and record how many people will try to grab
 107         // this lock
 108         let guard = self.mutex.lock();
 109         if self.cnt.load(atomics::SeqCst) > 0 {
 110             f();
 111             let prev = self.cnt.swap(int::MIN, atomics::SeqCst);
 112             self.lock_cnt.store(prev, atomics::SeqCst);
 113         }
 114         drop(guard);
 115
 116         // Last one out cleans up after everyone else, no leaks!
 117         if self.lock_cnt.fetch_add(-1, atomics::SeqCst) == 1 {
 118             unsafe { self.mutex.destroy() }
 119         }
 120     }
 121 }
 122
 123 #[cfg(test)]
 124 mod test {
 125     use std::prelude::*;
 126     use std::task;
 127     use super::{ONCE_INIT, Once};
 128
 129     #[test]
 130     fn smoke_once() {
 131         static mut o: Once = ONCE_INIT;
 132         let mut a = 0;
 133         unsafe { o.doit(|| a += 1); }
 134         assert_eq!(a, 1);
 135         unsafe { o.doit(|| a += 1); }
 136         assert_eq!(a, 1);
 137     }
 138
 139     #[test]
 140     fn stampede_once() {
 141         static mut o: Once = ONCE_INIT;
 142         static mut run: bool = false;
 143
 144         let (tx, rx) = channel();
 145         for _ in range(0, 10) {
 146             let tx = tx.clone();
 147             spawn(proc() {
 148                 for _ in range(0, 4) { task::deschedule() }
 149                 unsafe {
 150                     o.doit(|| {
 151                         assert!(!run);
 152                         run = true;
 153                     });
 154                     assert!(run);
 155                 }
 156                 tx.send(());
 157             });
 158         }
 159
 160         unsafe {
 161             o.doit(|| {
 162                 assert!(!run);
 163                 run = true;
 164             });
 165             assert!(run);
 166         }
 167
 168         for _ in range(0, 10) {
 169             rx.recv();
 170         }
 171     }
 172 }