src/libstd/sync/once.rs

   1 //! A "once initialization" primitive
   2 //!
   3 //! This primitive is meant to be used to run one-time initialization. An
   4 //! example use case would be for initializing an FFI library.
   5
   6 // A "once" is a relatively simple primitive, and it's also typically provided
   7 // by the OS as well (see `pthread_once` or `InitOnceExecuteOnce`). The OS
   8 // primitives, however, tend to have surprising restrictions, such as the Unix
   9 // one doesn't allow an argument to be passed to the function.
  10 //
  11 // As a result, we end up implementing it ourselves in the standard library.
  12 // This also gives us the opportunity to optimize the implementation a bit which
  13 // should help the fast path on call sites. Consequently, let's explain how this
  14 // primitive works now!
  15 //
  16 // So to recap, the guarantees of a Once are that it will call the
  17 // initialization closure at most once, and it will never return until the one
  18 // that's running has finished running. This means that we need some form of
  19 // blocking here while the custom callback is running at the very least.
  20 // Additionally, we add on the restriction of **poisoning**. Whenever an
  21 // initialization closure panics, the Once enters a "poisoned" state which means
  22 // that all future calls will immediately panic as well.
  23 //
  24 // So to implement this, one might first reach for a `Mutex`, but those cannot
  25 // be put into a `static`. It also gets a lot harder with poisoning to figure
  26 // out when the mutex needs to be deallocated because it's not after the closure
  27 // finishes, but after the first successful closure finishes.
  28 //
  29 // All in all, this is instead implemented with atomics and lock-free
  30 // operations! Whee! Each `Once` has one word of atomic state, and this state is
  31 // CAS'd on to determine what to do. There are four possible state of a `Once`:
  32 //
  33 // * Incomplete - no initialization has run yet, and no thread is currently
  34 //                using the Once.
  35 // * Poisoned - some thread has previously attempted to initialize the Once, but
  36 //              it panicked, so the Once is now poisoned. There are no other
  37 //              threads currently accessing this Once.
  38 // * Running - some thread is currently attempting to run initialization. It may
  39 //             succeed, so all future threads need to wait for it to finish.
  40 //             Note that this state is accompanied with a payload, described
  41 //             below.
  42 // * Complete - initialization has completed and all future calls should finish
  43 //              immediately.
  44 //
  45 // With 4 states we need 2 bits to encode this, and we use the remaining bits
  46 // in the word we have allocated as a queue of threads waiting for the thread
  47 // responsible for entering the RUNNING state. This queue is just a linked list
  48 // of Waiter nodes which is monotonically increasing in size. Each node is
  49 // allocated on the stack, and whenever the running closure finishes it will
  50 // consume the entire queue and notify all waiters they should try again.
  51 //
  52 // You'll find a few more details in the implementation, but that's the gist of
  53 // it!
  54
  55 use fmt;
  56 use marker;
  57 use ptr;
  58 use sync::atomic::{AtomicUsize, AtomicBool, Ordering};
  59 use thread::{self, Thread};
  60
  61 /// A synchronization primitive which can be used to run a one-time global
  62 /// initialization. Useful for one-time initialization for FFI or related
  63 /// functionality. This type can only be constructed with the [`ONCE_INIT`]
  64 /// value or the equivalent [`Once::new`] constructor.
  65 ///
  66 /// [`ONCE_INIT`]: constant.ONCE_INIT.html
  67 /// [`Once::new`]: struct.Once.html#method.new
  68 ///
  69 /// # Examples
  70 ///
  71 /// ```
  72 /// use std::sync::Once;
  73 ///
  74 /// static START: Once = Once::new();
  75 ///
  76 /// START.call_once(|| {
  77 ///     // run initialization here
  78 /// });
  79 /// ```
  80 #[stable(feature = "rust1", since = "1.0.0")]
  81 pub struct Once {
  82     // This `state` word is actually an encoded version of just a pointer to a
  83     // `Waiter`, so we add the `PhantomData` appropriately.
  84     state: AtomicUsize,
  85     _marker: marker::PhantomData<*mut Waiter>,
  86 }
  87
  88 // The `PhantomData` of a raw pointer removes these two auto traits, but we
  89 // enforce both below in the implementation so this should be safe to add.
  90 #[stable(feature = "rust1", since = "1.0.0")]
  91 unsafe impl Sync for Once {}
  92 #[stable(feature = "rust1", since = "1.0.0")]
  93 unsafe impl Send for Once {}
  94
  95 /// State yielded to [`call_once_force`]’s closure parameter. The state can be
  96 /// used to query the poison status of the [`Once`].
  97 ///
  98 /// [`call_once_force`]: struct.Once.html#method.call_once_force
  99 /// [`Once`]: struct.Once.html
 100 #[unstable(feature = "once_poison", issue = "33577")]
 101 #[derive(Debug)]
 102 pub struct OnceState {
 103     poisoned: bool,
 104 }
 105
 106 /// Initialization value for static [`Once`] values.
 107 ///
 108 /// [`Once`]: struct.Once.html
 109 ///
 110 /// # Examples
 111 ///
 112 /// ```
 113 /// use std::sync::{Once, ONCE_INIT};
 114 ///
 115 /// static START: Once = ONCE_INIT;
 116 /// ```
 117 #[stable(feature = "rust1", since = "1.0.0")]
 118 pub const ONCE_INIT: Once = Once::new();
 119
 120 // Four states that a Once can be in, encoded into the lower bits of `state` in
 121 // the Once structure.
 122 const INCOMPLETE: usize = 0x0;
 123 const POISONED: usize = 0x1;
 124 const RUNNING: usize = 0x2;
 125 const COMPLETE: usize = 0x3;
 126
 127 // Mask to learn about the state. All other bits are the queue of waiters if
 128 // this is in the RUNNING state.
 129 const STATE_MASK: usize = 0x3;
 130
 131 // Representation of a node in the linked list of waiters in the RUNNING state.
 132 struct Waiter {
 133     thread: Option<Thread>,
 134     signaled: AtomicBool,
 135     next: *mut Waiter,
 136 }
 137
 138 // Helper struct used to clean up after a closure call with a `Drop`
 139 // implementation to also run on panic.
 140 struct Finish<'a> {
 141     panicked: bool,
 142     me: &'a Once,
 143 }
 144
 145 impl Once {
 146     /// Creates a new `Once` value.
 147     #[stable(feature = "once_new", since = "1.2.0")]
 148     pub const fn new() -> Once {
 149         Once {
 150             state: AtomicUsize::new(INCOMPLETE),
 151             _marker: marker::PhantomData,
 152         }
 153     }
 154
 155     /// Performs an initialization routine once and only once. The given closure
 156     /// will be executed if this is the first time `call_once` has been called,
 157     /// and otherwise the routine will *not* be invoked.
 158     ///
 159     /// This method will block the calling thread if another initialization
 160     /// routine is currently running.
 161     ///
 162     /// When this function returns, it is guaranteed that some initialization
 163     /// has run and completed (it may not be the closure specified). It is also
 164     /// guaranteed that any memory writes performed by the executed closure can
 165     /// be reliably observed by other threads at this point (there is a
 166     /// happens-before relation between the closure and code executing after the
 167     /// return).
 168     ///
 169     /// If the given closure recursively invokes `call_once` on the same `Once`
 170     /// instance the exact behavior is not specified, allowed outcomes are
 171     /// a panic or a deadlock.
 172     ///
 173     /// # Examples
 174     ///
 175     /// ```
 176     /// use std::sync::Once;
 177     ///
 178     /// static mut VAL: usize = 0;
 179     /// static INIT: Once = Once::new();
 180     ///
 181     /// // Accessing a `static mut` is unsafe much of the time, but if we do so
 182     /// // in a synchronized fashion (e.g., write once or read all) then we're
 183     /// // good to go!
 184     /// //
 185     /// // This function will only call `expensive_computation` once, and will
 186     /// // otherwise always return the value returned from the first invocation.
 187     /// fn get_cached_val() -> usize {
 188     ///     unsafe {
 189     ///         INIT.call_once(|| {
 190     ///             VAL = expensive_computation();
 191     ///         });
 192     ///         VAL
 193     ///     }
 194     /// }
 195     ///
 196     /// fn expensive_computation() -> usize {
 197     ///     // ...
 198     /// # 2
 199     /// }
 200     /// ```
 201     ///
 202     /// # Panics
 203     ///
 204     /// The closure `f` will only be executed once if this is called
 205     /// concurrently amongst many threads. If that closure panics, however, then
 206     /// it will *poison* this `Once` instance, causing all future invocations of
 207     /// `call_once` to also panic.
 208     ///
 209     /// This is similar to [poisoning with mutexes][poison].
 210     ///
 211     /// [poison]: struct.Mutex.html#poisoning
 212     #[stable(feature = "rust1", since = "1.0.0")]
 213     pub fn call_once<F>(&self, f: F) where F: FnOnce() {
 214         // Fast path check
 215         if self.is_completed() {
 216             return;
 217         }
 218
 219         let mut f = Some(f);
 220         self.call_inner(false, &mut |_| f.take().unwrap()());
 221     }
 222
 223     /// Performs the same function as [`call_once`] except ignores poisoning.
 224     ///
 225     /// Unlike [`call_once`], if this `Once` has been poisoned (i.e., a previous
 226     /// call to `call_once` or `call_once_force` caused a panic), calling
 227     /// `call_once_force` will still invoke the closure `f` and will _not_
 228     /// result in an immediate panic. If `f` panics, the `Once` will remain
 229     /// in a poison state. If `f` does _not_ panic, the `Once` will no
 230     /// longer be in a poison state and all future calls to `call_once` or
 231     /// `call_one_force` will be no-ops.
 232     ///
 233     /// The closure `f` is yielded a [`OnceState`] structure which can be used
 234     /// to query the poison status of the `Once`.
 235     ///
 236     /// [`call_once`]: struct.Once.html#method.call_once
 237     /// [`OnceState`]: struct.OnceState.html
 238     ///
 239     /// # Examples
 240     ///
 241     /// ```
 242     /// #![feature(once_poison)]
 243     ///
 244     /// use std::sync::Once;
 245     /// use std::thread;
 246     ///
 247     /// static INIT: Once = Once::new();
 248     ///
 249     /// // poison the once
 250     /// let handle = thread::spawn(|| {
 251     ///     INIT.call_once(|| panic!());
 252     /// });
 253     /// assert!(handle.join().is_err());
 254     ///
 255     /// // poisoning propagates
 256     /// let handle = thread::spawn(|| {
 257     ///     INIT.call_once(|| {});
 258     /// });
 259     /// assert!(handle.join().is_err());
 260     ///
 261     /// // call_once_force will still run and reset the poisoned state
 262     /// INIT.call_once_force(|state| {
 263     ///     assert!(state.poisoned());
 264     /// });
 265     ///
 266     /// // once any success happens, we stop propagating the poison
 267     /// INIT.call_once(|| {});
 268     /// ```
 269     #[unstable(feature = "once_poison", issue = "33577")]
 270     pub fn call_once_force<F>(&self, f: F) where F: FnOnce(&OnceState) {
 271         // Fast path check
 272         if self.is_completed() {
 273             return;
 274         }
 275
 276         let mut f = Some(f);
 277         self.call_inner(true, &mut |p| {
 278             f.take().unwrap()(&OnceState { poisoned: p })
 279         });
 280     }
 281
 282     /// Returns `true` if some `call_once` call has completed
 283     /// successfully. Specifically, `is_completed` will return false in
 284     /// the following situations:
 285     ///   * `call_once` was not called at all,
 286     ///   * `call_once` was called, but has not yet completed,
 287     ///   * the `Once` instance is poisoned
 288     ///
 289     /// It is also possible that immediately after `is_completed`
 290     /// returns false, some other thread finishes executing
 291     /// `call_once`.
 292     ///
 293     /// # Examples
 294     ///
 295     /// ```
 296     /// #![feature(once_is_completed)]
 297     /// use std::sync::Once;
 298     ///
 299     /// static INIT: Once = Once::new();
 300     ///
 301     /// assert_eq!(INIT.is_completed(), false);
 302     /// INIT.call_once(|| {
 303     ///     assert_eq!(INIT.is_completed(), false);
 304     /// });
 305     /// assert_eq!(INIT.is_completed(), true);
 306     /// ```
 307     ///
 308     /// ```
 309     /// #![feature(once_is_completed)]
 310     /// use std::sync::Once;
 311     /// use std::thread;
 312     ///
 313     /// static INIT: Once = Once::new();
 314     ///
 315     /// assert_eq!(INIT.is_completed(), false);
 316     /// let handle = thread::spawn(|| {
 317     ///     INIT.call_once(|| panic!());
 318     /// });
 319     /// assert!(handle.join().is_err());
 320     /// assert_eq!(INIT.is_completed(), false);
 321     /// ```
 322     #[unstable(feature = "once_is_completed", issue = "54890")]
 323     #[inline]
 324     pub fn is_completed(&self) -> bool {
 325         // An `Acquire` load is enough because that makes all the initialization
 326         // operations visible to us, and, this being a fast path, weaker
 327         // ordering helps with performance. This `Acquire` synchronizes with
 328         // `SeqCst` operations on the slow path.
 329         self.state.load(Ordering::Acquire) == COMPLETE
 330     }
 331
 332     // This is a non-generic function to reduce the monomorphization cost of
 333     // using `call_once` (this isn't exactly a trivial or small implementation).
 334     //
 335     // Additionally, this is tagged with `#[cold]` as it should indeed be cold
 336     // and it helps let LLVM know that calls to this function should be off the
 337     // fast path. Essentially, this should help generate more straight line code
 338     // in LLVM.
 339     //
 340     // Finally, this takes an `FnMut` instead of a `FnOnce` because there's
 341     // currently no way to take an `FnOnce` and call it via virtual dispatch
 342     // without some allocation overhead.
 343     #[cold]
 344     fn call_inner(&self,
 345                   ignore_poisoning: bool,
 346                   init: &mut dyn FnMut(bool)) {
 347
 348         // This cold path uses SeqCst consistently because the
 349         // performance difference really does not matter there, and
 350         // SeqCst minimizes the chances of something going wrong.
 351         let mut state = self.state.load(Ordering::SeqCst);
 352
 353         'outer: loop {
 354             match state {
 355                 // If we're complete, then there's nothing to do, we just
 356                 // jettison out as we shouldn't run the closure.
 357                 COMPLETE => return,
 358
 359                 // If we're poisoned and we're not in a mode to ignore
 360                 // poisoning, then we panic here to propagate the poison.
 361                 POISONED if !ignore_poisoning => {
 362                     panic!("Once instance has previously been poisoned");
 363                 }
 364
 365                 // Otherwise if we see a poisoned or otherwise incomplete state
 366                 // we will attempt to move ourselves into the RUNNING state. If
 367                 // we succeed, then the queue of waiters starts at null (all 0
 368                 // bits).
 369                 POISONED |
 370                 INCOMPLETE => {
 371                     let old = self.state.compare_and_swap(state, RUNNING,
 372                                                           Ordering::SeqCst);
 373                     if old != state {
 374                         state = old;
 375                         continue
 376                     }
 377
 378                     // Run the initialization routine, letting it know if we're
 379                     // poisoned or not. The `Finish` struct is then dropped, and
 380                     // the `Drop` implementation here is responsible for waking
 381                     // up other waiters both in the normal return and panicking
 382                     // case.
 383                     let mut complete = Finish {
 384                         panicked: true,
 385                         me: self,
 386                     };
 387                     init(state == POISONED);
 388                     complete.panicked = false;
 389                     return
 390                 }
 391
 392                 // All other values we find should correspond to the RUNNING
 393                 // state with an encoded waiter list in the more significant
 394                 // bits. We attempt to enqueue ourselves by moving us to the
 395                 // head of the list and bail out if we ever see a state that's
 396                 // not RUNNING.
 397                 _ => {
 398                     assert!(state & STATE_MASK == RUNNING);
 399                     let mut node = Waiter {
 400                         thread: Some(thread::current()),
 401                         signaled: AtomicBool::new(false),
 402                         next: ptr::null_mut(),
 403                     };
 404                     let me = &mut node as *mut Waiter as usize;
 405                     assert!(me & STATE_MASK == 0);
 406
 407                     while state & STATE_MASK == RUNNING {
 408                         node.next = (state & !STATE_MASK) as *mut Waiter;
 409                         let old = self.state.compare_and_swap(state,
 410                                                               me | RUNNING,
 411                                                               Ordering::SeqCst);
 412                         if old != state {
 413                             state = old;
 414                             continue
 415                         }
 416
 417                         // Once we've enqueued ourselves, wait in a loop.
 418                         // Afterwards reload the state and continue with what we
 419                         // were doing from before.
 420                         while !node.signaled.load(Ordering::SeqCst) {
 421                             thread::park();
 422                         }
 423                         state = self.state.load(Ordering::SeqCst);
 424                         continue 'outer
 425                     }
 426                 }
 427             }
 428         }
 429     }
 430 }
 431
 432 #[stable(feature = "std_debug", since = "1.16.0")]
 433 impl fmt::Debug for Once {
 434     fn fmt(&self, f: &mut fmt::Formatter) -> fmt::Result {
 435         f.pad("Once { .. }")
 436     }
 437 }
 438
 439 impl Drop for Finish<'_> {
 440     fn drop(&mut self) {
 441         // Swap out our state with however we finished. We should only ever see
 442         // an old state which was RUNNING.
 443         let queue = if self.panicked {
 444             self.me.state.swap(POISONED, Ordering::SeqCst)
 445         } else {
 446             self.me.state.swap(COMPLETE, Ordering::SeqCst)
 447         };
 448         assert_eq!(queue & STATE_MASK, RUNNING);
 449
 450         // Decode the RUNNING to a list of waiters, then walk that entire list
 451         // and wake them up. Note that it is crucial that after we store `true`
 452         // in the node it can be free'd! As a result we load the `thread` to
 453         // signal ahead of time and then unpark it after the store.
 454         unsafe {
 455             let mut queue = (queue & !STATE_MASK) as *mut Waiter;
 456             while !queue.is_null() {
 457                 let next = (*queue).next;
 458                 let thread = (*queue).thread.take().unwrap();
 459                 (*queue).signaled.store(true, Ordering::SeqCst);
 460                 thread.unpark();
 461                 queue = next;
 462             }
 463         }
 464     }
 465 }
 466
 467 impl OnceState {
 468     /// Returns `true` if the associated [`Once`] was poisoned prior to the
 469     /// invocation of the closure passed to [`call_once_force`].
 470     ///
 471     /// [`call_once_force`]: struct.Once.html#method.call_once_force
 472     /// [`Once`]: struct.Once.html
 473     ///
 474     /// # Examples
 475     ///
 476     /// A poisoned `Once`:
 477     ///
 478     /// ```
 479     /// #![feature(once_poison)]
 480     ///
 481     /// use std::sync::Once;
 482     /// use std::thread;
 483     ///
 484     /// static INIT: Once = Once::new();
 485     ///
 486     /// // poison the once
 487     /// let handle = thread::spawn(|| {
 488     ///     INIT.call_once(|| panic!());
 489     /// });
 490     /// assert!(handle.join().is_err());
 491     ///
 492     /// INIT.call_once_force(|state| {
 493     ///     assert!(state.poisoned());
 494     /// });
 495     /// ```
 496     ///
 497     /// An unpoisoned `Once`:
 498     ///
 499     /// ```
 500     /// #![feature(once_poison)]
 501     ///
 502     /// use std::sync::Once;
 503     ///
 504     /// static INIT: Once = Once::new();
 505     ///
 506     /// INIT.call_once_force(|state| {
 507     ///     assert!(!state.poisoned());
 508     /// });
 509     #[unstable(feature = "once_poison", issue = "33577")]
 510     pub fn poisoned(&self) -> bool {
 511         self.poisoned
 512     }
 513 }
 514
 515 #[cfg(all(test, not(target_os = "emscripten")))]
 516 mod tests {
 517     use panic;
 518     use sync::mpsc::channel;
 519     use thread;
 520     use super::Once;
 521
 522     #[test]
 523     fn smoke_once() {
 524         static O: Once = Once::new();
 525         let mut a = 0;
 526         O.call_once(|| a += 1);
 527         assert_eq!(a, 1);
 528         O.call_once(|| a += 1);
 529         assert_eq!(a, 1);
 530     }
 531
 532     #[test]
 533     fn stampede_once() {
 534         static O: Once = Once::new();
 535         static mut RUN: bool = false;
 536
 537         let (tx, rx) = channel();
 538         for _ in 0..10 {
 539             let tx = tx.clone();
 540             thread::spawn(move|| {
 541                 for _ in 0..4 { thread::yield_now() }
 542                 unsafe {
 543                     O.call_once(|| {
 544                         assert!(!RUN);
 545                         RUN = true;
 546                     });
 547                     assert!(RUN);
 548                 }
 549                 tx.send(()).unwrap();
 550             });
 551         }
 552
 553         unsafe {
 554             O.call_once(|| {
 555                 assert!(!RUN);
 556                 RUN = true;
 557             });
 558             assert!(RUN);
 559         }
 560
 561         for _ in 0..10 {
 562             rx.recv().unwrap();
 563         }
 564     }
 565
 566     #[test]
 567     fn poison_bad() {
 568         static O: Once = Once::new();
 569
 570         // poison the once
 571         let t = panic::catch_unwind(|| {
 572             O.call_once(|| panic!());
 573         });
 574         assert!(t.is_err());
 575
 576         // poisoning propagates
 577         let t = panic::catch_unwind(|| {
 578             O.call_once(|| {});
 579         });
 580         assert!(t.is_err());
 581
 582         // we can subvert poisoning, however
 583         let mut called = false;
 584         O.call_once_force(|p| {
 585             called = true;
 586             assert!(p.poisoned())
 587         });
 588         assert!(called);
 589
 590         // once any success happens, we stop propagating the poison
 591         O.call_once(|| {});
 592     }
 593
 594     #[test]
 595     fn wait_for_force_to_finish() {
 596         static O: Once = Once::new();
 597
 598         // poison the once
 599         let t = panic::catch_unwind(|| {
 600             O.call_once(|| panic!());
 601         });
 602         assert!(t.is_err());
 603
 604         // make sure someone's waiting inside the once via a force
 605         let (tx1, rx1) = channel();
 606         let (tx2, rx2) = channel();
 607         let t1 = thread::spawn(move || {
 608             O.call_once_force(|p| {
 609                 assert!(p.poisoned());
 610                 tx1.send(()).unwrap();
 611                 rx2.recv().unwrap();
 612             });
 613         });
 614
 615         rx1.recv().unwrap();
 616
 617         // put another waiter on the once
 618         let t2 = thread::spawn(|| {
 619             let mut called = false;
 620             O.call_once(|| {
 621                 called = true;
 622             });
 623             assert!(!called);
 624         });
 625
 626         tx2.send(()).unwrap();
 627
 628         assert!(t1.join().is_ok());
 629         assert!(t2.join().is_ok());
 630
 631     }
 632 }