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