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