1
  2
  3
  4
  5
  6
  7
  8
  9
 10
 11
 12
 13
 14
 15
 16
 17
 18
 19
 20
 21
 22
 23
 24
 25
 26
 27
 28
 29
 30
 31
 32
 33
 34
 35
 36
 37
 38
 39
 40
 41
 42
 43
 44
 45
 46
 47
 48
 49
 50
 51
 52
 53
 54
 55
 56
 57
 58
 59
 60
 61
 62
 63
 64
 65
 66
 67
 68
 69
 70
 71
 72
 73
 74
 75
 76
 77
 78
 79
 80
 81
 82
 83
 84
 85
 86
 87
 88
 89
 90
 91
 92
 93
 94
 95
 96
 97
 98
 99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
165
166
167
168
169
170
171
172
173
174
175
176
177
178
179
180
181
182
183
184
185
186
187
188
189
190
191
192
193
194
195
196
197
198
199
200
201
202
203
204
205
206
207
208
209
210
211
212
213
214
215
216
217
218
219
220
221
222
223
224
225
226
227
228
229
230
231
232
233
234
235
236
237
238
239
240
241
242
243
244
245
246
247
248
249
250
251
252
253
254
255
256
257
258
259
260
261
262
263
264
265
266
267
268
269
270
271
272
273
274
275
276
277
278
279
280
281
282
283
284
285
286
287
288
289
290
291
292
293
294
295
296
297
298
299
300
301
302
303
304
305
306
307
308
309
310
311
312
313
314
315
316
317
318
319
320
321
322
323
324
325
326
327
328
329
330
331
332
333
334
335
336
337
338
339
340
341
342
343
344
345
346
347
348
349
350
351
352
353
354
355
356
357
358
359
360
361
362
363
364
365
366
367
368
369
370
371
372
373
374
375
376
377
378
379
380
381
382
383
384
385
386
387
388
389
390
391
392
393
394
395
396
397
398
399
400
401
402
403
404
405
406
407
408
409
410
411
412
413
414
415
416
417
418
419
420
421
422
423
424
425
426
427
428
429
430
431
432
433
434
435
436
437
438
439
440
441
442
443
444
445
446
447
448
449
450
451
452
453
454
455
456
457
458
459
460
461
462
463
464
465
466
467
468
469
470
471
472
473
474
475
476
477
478
479
480
481
482
483
484
485
486
487
488
489
490
491
492
493
494
495
496
497
498
499
500
501
502
503
504
505
506
507
508
509
510
511
512
513
514
515
516
517
518
519
520
521
522
523
524
525
526
527
528
529
530
531
532
533
534
535
536
537
538
539
540
541
542
543
544
545
546
547
548
549
550
551
552
553
554
555
556
557
558
559
560
561
562
563
564
//! "一次初始化" 原语
//!
//! 该原语旨在用于运行一次性初始化。
//! 一个示例用例将是初始化 FFI 库。

// "once" 是一个相对简单的原语,通常也由操作系统提供 (请参见 `pthread_once` 或 `InitOnceExecuteOnce`)。
// 但是,操作系统原语往往具有令人惊讶的限制,例如 Unix 不允许将参数传递给函数。
//
// 结果,我们最终在标准库中自己实现了它。
// 这也为我们提供了优化实现的机会,这将有助于在调用站点上实现快速路径。因此,让我们解释一下该原语现在是如何工作的!
//
// 因此,回顾一下,一次的保证是它将最多调用一次初始化闭包,并且直到运行的那个完成运行后它才会返回。
// 这意味着,至少在自定义回调运行时,我们需要在此处进行某种形式的阻塞。
// 此外,我们增加了中毒的限制。
// 每当初始化闭包 panics 时,`Once` 都会进入 "poisoned" 状态,这意味着所有 future 调用也将立即变为 panic。
//
// 因此,要实现此目的,可能首先需要购买 `Mutex`,但不能将其放入 `static`。
// 通过中毒来确定何时需要释放互斥锁也会变得更加困难,因为它不是在闭包结束之后,而是在第一个成功的闭包结束之后。
//
// 总而言之,这是通过原子和无锁操作实现的! 呜!每个 `Once` 都有一个原子状态的字,此状态已通过 CAS 确定要做什么。
// `Once` 有四种可能的状态:
//
// * 未完成 - 尚未运行任何初始化,并且当前没有线程在使用 `Once`。
// * 中毒 - 一些线程之前尝试初始化 `Once`,但是它 panic 了,所以 `Once` 现在是中毒的。当前没有其他线程访问此一次。
// * 正在运行 - 某些线程当前正在尝试运行初始化。它可能会成功,因此所有 future 线程都需要等待它完成。
//             请注意,这种状态伴随着一个有效载荷,如下所述。
// * 完成 - 初始化已完成,所有 future 调用应立即完成。
//
// 对于 4 种状态,我们需要 2 位来对此进行编码,并且我们将已分配的字中的其余位用作线程队列,等待负责进入 RUNNING 状态的线程。
//
// 该队列只是一个 Waiter 节点的链表,其大小单调增加。
// 每个节点都分配在栈上,并且每当运行的闭包完成时,它将消耗整个队列,并通知所有等待者他们应再次尝试。
//
// 您会在实现中找到更多细节,但这就是要点!
//
// 原子顺序:
// 当运行 `Once` 时,我们处理多个原子:
// `Once.state_and_queue` 和未知数量的 `Waiter.signaled`。
// * `state_and_queue` 使用 (1) 作为状态标志,(2) 用于同步 `Once` 的结果,(3) 用于同步 `Waiter` 节点。
//     - 在 `call_inner` 函数的末尾,我们必须确保已获取 `Once` 的结果。
//     因此,可能是唯一要加载完成的负载,每个负载都必须至少具有获取顺序,这意味着它们全部都是三个。
//     - `WaiterQueue::Drop` 是唯一可以存储 COMPLETED 的地方,并且必须使用 Release 排序才能使结果可用。
//     - `wait` 将 `Waiter` 节点作为 `state_and_queue` 中的指针插入,并且需要使用 Release 排序使节点可用。
//     `compare_exchange` 中的负载可以放宽,因为它只需要比较原子,而不必读取其他数据。
//     - `WaiterQueue::Drop` 必须看到 `Waiter` 节点,因此它必须使用 Acquire ordering 加载 `state_and_queue`。
//     - 只有一个存储区,其中 `state_and_queue` 仅用作状态标志,而不必同步数据:在 `call_inner` 中将状态从 INCOMPLETE 切换为 RUNNING。
// 这个 store 可以是 Relaxed,但是 read 必须是 Acquire,因为上面提到的要求。
// * `Waiter.signaled` 既用作标志,又用于保护 `Waiter` 中具有内部可变性的字段。
// `Waiter.thread` 在 `WaiterQueue::Drop` 中更改,然后将 `signaled` 设置为发布顺序。
//   `wait` 用 Acquire 加载 `signaled` 后,如果发现它是正确的,则需要查看更改以正确放置 `Waiter` 结构体。
// * 在一个地方,两个原子 `Once.state_and_queue` 和 `Waiter.signaled` 在一起,并且可能由编译器或处理器重新排序。
// 因为两者都使用 Acquire 排序,所以不允许这样的重新排序,因此不需要 SeqCst。
//
//
//
//
//
//
//
//
//
//
//
//
//
//
//
//
//
//
//
//
//
//
//
//
//
//
//
//
//
//
//
//
//

#[cfg(all(test, not(target_os = "emscripten")))]
mod tests;

use crate::cell::Cell;
use crate::fmt;
use crate::marker;
use crate::sync::atomic::{AtomicBool, AtomicUsize, Ordering};
use crate::thread::{self, Thread};

/// 同步原语,可用于运行一次性初始化。
/// 对于 FFI 或相关功能的一次性初始化很有用。
/// 该类型只能用 [`Once::new()`] 构造。
///
/// # Examples
///
/// ```
/// use std::sync::Once;
///
/// static START: Once = Once::new();
///
/// START.call_once(|| {
///     // 在这里运行初始化
/// });
/// ```
#[stable(feature = "rust1", since = "1.0.0")]
pub struct Once {
    // `state_and_queue` 实际上是一个指向带有额外状态位的 `Waiter` 的指针,因此我们适当地添加了 `PhantomData`。
    //
    state_and_queue: AtomicUsize,
    _marker: marker::PhantomData<*const Waiter>,
}

// 裸指针的 `PhantomData` 删除了这两个自动 traits,但是我们在实现中强制执行以下两项,因此添加起来应该很安全。
//
#[stable(feature = "rust1", since = "1.0.0")]
unsafe impl Sync for Once {}
#[stable(feature = "rust1", since = "1.0.0")]
unsafe impl Send for Once {}

/// 状态产生于 [`Once::call_once_force () `] 的闭包参数。
/// 该状态可用于查询 [`Once`] 的中毒状态。
#[stable(feature = "once_poison", since = "1.51.0")]
#[derive(Debug)]
pub struct OnceState {
    poisoned: bool,
    set_state_on_drop_to: Cell<usize>,
}

/// 静态 [`Once`] 值的初始化值。
///
/// # Examples
///
/// ```
/// use std::sync::{Once, ONCE_INIT};
///
/// static START: Once = ONCE_INIT;
/// ```
#[stable(feature = "rust1", since = "1.0.0")]
#[rustc_deprecated(
    since = "1.38.0",
    reason = "the `new` function is now preferred",
    suggestion = "Once::new()"
)]
pub const ONCE_INIT: Once = Once::new();

// 一次可以进入的四个状态,被编码为一次结构体中 `state_and_queue` 的低位。
//
const INCOMPLETE: usize = 0x0;
const POISONED: usize = 0x1;
const RUNNING: usize = 0x2;
const COMPLETE: usize = 0x3;

// 隐含的了解状态。
// 如果其他所有位都处于 RUNNING 状态,则它们都是等待者的队列。
const STATE_MASK: usize = 0x3;

// 处于 RUNNING 状态时使用的等待者的链表中的节点表示。
// Note: `Waiter` 不能持有指向下一个线程的可变指针,因为 `wait` 会同时向其 `Waiter` 节点发出变量引用,并保留共享的引用以检查 `signaled`。
// 相反,我们使用共享引用,并使用内部可变性。
//
//
//
#[repr(align(4))] // 确保两个低位可以自由用作状态位。
struct Waiter {
    thread: Cell<Option<Thread>>,
    signaled: AtomicBool,
    next: *const Waiter,
}

// 等待者链表的头。
// 每个节点都是等待线程栈上的结构体。
// 丢弃时将唤醒等待者,例如在 panic 上。
struct WaiterQueue<'a> {
    state_and_queue: &'a AtomicUsize,
    set_state_on_drop_to: usize,
}

impl Once {
    /// 创建一个新的 `Once` 值。
    #[inline]
    #[stable(feature = "once_new", since = "1.2.0")]
    #[rustc_const_stable(feature = "const_once_new", since = "1.32.0")]
    #[must_use]
    pub const fn new() -> Once {
        Once { state_and_queue: AtomicUsize::new(INCOMPLETE), _marker: marker::PhantomData }
    }

    /// 仅执行一次初始化例程。如果这是第一次调用 `call_once`,则将执行给定的闭包,否则将 *不* 调用例程。
    ///
    /// 如果当前正在运行另一个初始化例程,则此方法将阻止调用线程。
    ///
    /// 当这个函数返回时,保证一些初始化已经运行并完成 (它可能不是指定的闭包)。
    /// 还可以确保此时其他线程可以可靠地观察到由执行的闭包执行的任何内存写操作 (闭包与返回之后执行的代码之间存在先于发生的关系)。
    ///
    ///
    /// 如果给定的闭包在同一 [`Once`] 实例上递归调用 `call_once`,则未指定确切的行为,则允许的结果为 panic 或死锁。
    ///
    /// # Examples
    ///
    /// ```
    /// use std::sync::Once;
    ///
    /// static mut VAL: usize = 0;
    /// static INIT: Once = Once::new();
    ///
    /// // 在很多情况下,访问 `static mut` 是不安全的,但是如果我们以同步方式进行操作 (例如,一次写入或全部读取),那么我们就可以开始了!
    /////
    /////
    /// // 此函数将只调用一次 `expensive_computation`,否则将始终返回从第一次调用返回的值。
    /////
    /////
    /// fn get_cached_val() -> usize {
    ///     unsafe {
    ///         INIT.call_once(|| {
    ///             VAL = expensive_computation();
    ///         });
    ///         VAL
    ///     }
    /// }
    ///
    /// fn expensive_computation() -> usize {
    ///     // ...
    /// # 2
    /// }
    /// ```
    ///
    /// # Panics
    ///
    /// 如果在多个线程中同时调用闭包 `f`,则将仅执行一次。
    /// 但是,如果该闭包 panics 将会毒化该 [`Once`] 实例,从而导致 `call_once` 的所有 future 调用也变为 panic。
    ///
    /// 这类似于 [互斥体中毒][poison]。
    ///
    /// [poison]: struct.Mutex.html#poisoning
    ///
    ///
    ///
    ///
    ///
    ///
    ///
    ///
    ///
    ///
    #[stable(feature = "rust1", since = "1.0.0")]
    pub fn call_once<F>(&self, f: F)
    where
        F: FnOnce(),
    {
        // 快速路径检查
        if self.is_completed() {
            return;
        }

        let mut f = Some(f);
        self.call_inner(false, &mut |_| f.take().unwrap()());
    }

    /// 执行与 [`call_once()`] 相同的函数,只是忽略中毒。
    ///
    /// 与 [`call_once()`] 不同,如果此 [`Once`] 已中毒 (例如,先前对 [`call_once()`] 或 [`call_once_force()`] 的调用导致 panic),则调用 [`call_once_force()`] 仍将调用闭包 `f`,并且 _not_ 会立即导致 panic。
    /// 如果 `f` panics,则 [`Once`] 将保持中毒状态。
    /// 如果 `f` 执行 _not_ panic,则 [`Once`] 将不再处于中毒状态,并且所有对 [`call_once()`] 或 [`call_once_force()`] 的 future 调用都将变为无操作。
    ///
    /// 闭包 `f` 产生 [`OnceState`] 结构体,可用于查询 [`Once`] 的中毒状态。
    ///
    /// [`call_once()`]: Once::call_once
    /// [`call_once_force()`]: Once::call_once_force
    ///
    /// # Examples
    ///
    /// ```
    /// use std::sync::Once;
    /// use std::thread;
    ///
    /// static INIT: Once = Once::new();
    ///
    /// // 中毒一次
    /// let handle = thread::spawn(|| {
    ///     INIT.call_once(|| panic!());
    /// });
    /// assert!(handle.join().is_err());
    ///
    /// // 中毒传播
    /// let handle = thread::spawn(|| {
    ///     INIT.call_once(|| {});
    /// });
    /// assert!(handle.join().is_err());
    ///
    /// // call_once_force 仍将运行并重置中毒状态
    /// INIT.call_once_force(|state| {
    ///     assert!(state.is_poisoned());
    /// });
    ///
    /// // 一旦成功,我们就停止传播毒药
    /// INIT.call_once(|| {});
    /// ```
    ///
    ///
    ///
    ///
    ///
    #[stable(feature = "once_poison", since = "1.51.0")]
    pub fn call_once_force<F>(&self, f: F)
    where
        F: FnOnce(&OnceState),
    {
        // 快速路径检查
        if self.is_completed() {
            return;
        }

        let mut f = Some(f);
        self.call_inner(true, &mut |p| f.take().unwrap()(p));
    }

    /// 如果某些 [`call_once()`] 调用已成功完成,则返回 `true`。具体而言,在以下情况下,`is_completed` 将返回 false:
    ///   * [`call_once()`] 根本没有被调用,
    ///   * [`call_once()`] 已被调用,但尚未完成,
    ///   * [`Once`] 实例中毒
    ///
    /// 此函数返回 `false` 并不意味着 [`Once`] 尚未执行。
    /// 例如,它可能是在 `is_completed` 开始执行到返回之间的时间执行的,在这种情况下,`false` 的返回值将是陈旧的 (但仍然是允许的)。
    ///
    ///
    /// [`call_once()`]: Once::call_once
    ///
    /// # Examples
    ///
    /// ```
    /// use std::sync::Once;
    ///
    /// static INIT: Once = Once::new();
    ///
    /// assert_eq!(INIT.is_completed(), false);
    /// INIT.call_once(|| {
    ///     assert_eq!(INIT.is_completed(), false);
    /// });
    /// assert_eq!(INIT.is_completed(), true);
    /// ```
    ///
    /// ```
    /// use std::sync::Once;
    /// use std::thread;
    ///
    /// static INIT: Once = Once::new();
    ///
    /// assert_eq!(INIT.is_completed(), false);
    /// let handle = thread::spawn(|| {
    ///     INIT.call_once(|| panic!());
    /// });
    /// assert!(handle.join().is_err());
    /// assert_eq!(INIT.is_completed(), false);
    /// ```
    ///
    ///
    ///
    #[stable(feature = "once_is_completed", since = "1.43.0")]
    #[inline]
    pub fn is_completed(&self) -> bool {
        // `Acquire` 负载就足够了,因为这使我们可以看到所有初始化操作,并且这是一条捷径,较弱的排序有助于提高性能。
        // 该 `Acquire` 与慢速路径上的 `Release` 操作同步。
        //
        //
        self.state_and_queue.load(Ordering::Acquire) == COMPLETE
    }

    // 这是一个非通用类型的函数,可以减少使用 `call_once` 的单态化成本 (这并非微不足道或很小的实现)。
    //
    // 另外,它用 `#[cold]` 标记,因为它确实应该是冷的,它有助于让 LLVM 知道对该函数的调用应该不在快速路径之内。
    // 从本质上讲,这应该有助于在 LLVM 中生成更多直线代码。
    //
    // 最后,这需要 `FnMut` 而不是 `FnOnce`,因为当前没有办法通过虚拟调度来获取 `FnOnce` 并调用它,而没有一些分配开销。
    //
    //
    //
    //
    //
    #[cold]
    fn call_inner(&self, ignore_poisoning: bool, init: &mut dyn FnMut(&OnceState)) {
        let mut state_and_queue = self.state_and_queue.load(Ordering::Acquire);
        loop {
            match state_and_queue {
                COMPLETE => break,
                POISONED if !ignore_poisoning => {
                    // panic 会传播 poison。
                    panic!("Once instance has previously been poisoned");
                }
                POISONED | INCOMPLETE => {
                    // 尝试将该线程注册为一个正在运行的线程。
                    let exchange_result = self.state_and_queue.compare_exchange(
                        state_and_queue,
                        RUNNING,
                        Ordering::Acquire,
                        Ordering::Acquire,
                    );
                    if let Err(old) = exchange_result {
                        state_and_queue = old;
                        continue;
                    }
                    // `waiter_queue` 将管理其他等待线程,并在丢弃时唤醒它们。
                    //
                    let mut waiter_queue = WaiterQueue {
                        state_and_queue: &self.state_and_queue,
                        set_state_on_drop_to: POISONED,
                    };
                    // 运行初始化函数,让它知道我们是否中毒。
                    //
                    let init_state = OnceState {
                        poisoned: state_and_queue == POISONED,
                        set_state_on_drop_to: Cell::new(COMPLETE),
                    };
                    init(&init_state);
                    waiter_queue.set_state_on_drop_to = init_state.set_state_on_drop_to.get();
                    break;
                }
                _ => {
                    // 所有其他值都必须为 RUNNING,并且可能有一个指向更高有效位的侍者队列的指针。
                    //
                    assert!(state_and_queue & STATE_MASK == RUNNING);
                    wait(&self.state_and_queue, state_and_queue);
                    state_and_queue = self.state_and_queue.load(Ordering::Acquire);
                }
            }
        }
    }
}

fn wait(state_and_queue: &AtomicUsize, mut current_state: usize) {
    // Note: 以下代码经过精心编写,以避免对 `node` 创建别名。
    //
    loop {
        // 如果状态不再运行,请不要将此线程排队,否则我们不会被唤醒。
        //
        if current_state & STATE_MASK != RUNNING {
            return;
        }

        // 为我们当前的线程创建节点。
        let node = Waiter {
            thread: Cell::new(Some(thread::current())),
            signaled: AtomicBool::new(false),
            next: (current_state & !STATE_MASK) as *const Waiter,
        };
        let me = &node as *const Waiter as usize;

        // 尝试在链表顶部的节点中滑动,确保其他线程不只是替换链表的头部。
        //
        let exchange_result = state_and_queue.compare_exchange(
            current_state,
            me | RUNNING,
            Ordering::Release,
            Ordering::Relaxed,
        );
        if let Err(old) = exchange_result {
            current_state = old;
            continue;
        }

        // 我们已经入队,现在让我们等待。
        // 重要的是不要在收到信号之前返回,否则我们将丢弃 `Waiter` 节点,并将 hole 留在链表中 (并悬垂引用)。
        // 提醒自己,直到收到信号为止,以防虚假唤醒。
        //
        //
        while !node.signaled.load(Ordering::Acquire) {
            // 如果管理线程恰好在我们可以停泊自己之前发出信号并释放了我们,则结果可能是该线程永远不会停泊。
            // 幸运的是,`park` 保证如果它在未停放的线程上获得 `unpark`,则它不会 park。
            //
            //
            thread::park();
        }
        break;
    }
}

#[stable(feature = "std_debug", since = "1.16.0")]
impl fmt::Debug for Once {
    fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result {
        f.debug_struct("Once").finish_non_exhaustive()
    }
}

impl Drop for WaiterQueue<'_> {
    fn drop(&mut self) {
        // 与我们交换状态,但是我们完成了。
        let state_and_queue =
            self.state_and_queue.swap(self.set_state_on_drop_to, Ordering::AcqRel);

        // 我们应该只看到一个正在运行的旧状态。
        assert_eq!(state_and_queue & STATE_MASK, RUNNING);

        // 遍历整个 waiters 的链表并唤醒他们 (按照 lifo 顺序,最后注册要首先唤醒)。
        //
        unsafe {
            // 设置 `node.signaled = true` 之后,如果恰好有虚假唤醒,其他线程可能会释放 `node`。
            //
            // 因此,我们必须取出 `thread` 字段,然后首先将指针复制到 `next`。
            //
            let mut queue = (state_and_queue & !STATE_MASK) as *const Waiter;
            while !queue.is_null() {
                let next = (*queue).next;
                let thread = (*queue).thread.take().unwrap();
                (*queue).signaled.store(true, Ordering::Release);
                // ^- FIXME (maybe): 这是 issue #55005 的另一种情况 `store()` 潜在地悬挂了对 `signaled` 的引用。
                //
                queue = next;
                thread.unpark();
            }
        }
    }
}

impl OnceState {
    /// 如果关联的 [`Once`] 在调用传递给 [`Once::call_once_force()`] 的闭包之前中毒,则返回 `true`。
    ///
    ///
    /// # Examples
    ///
    /// 中毒的 [`Once`]:
    ///
    /// ```
    /// use std::sync::Once;
    /// use std::thread;
    ///
    /// static INIT: Once = Once::new();
    ///
    /// // 中毒一次
    /// let handle = thread::spawn(|| {
    ///     INIT.call_once(|| panic!());
    /// });
    /// assert!(handle.join().is_err());
    ///
    /// INIT.call_once_force(|state| {
    ///     assert!(state.is_poisoned());
    /// });
    /// ```
    ///
    /// 无毒的 [`Once`]:
    ///
    /// ```
    /// use std::sync::Once;
    ///
    /// static INIT: Once = Once::new();
    ///
    /// INIT.call_once_force(|state| {
    ///     assert!(!state.is_poisoned());
    /// });
    #[stable(feature = "once_poison", since = "1.51.0")]
    pub fn is_poisoned(&self) -> bool {
        self.poisoned
    }

    /// 中毒关联的 [`Once`],而不显式 panic。
    // NOTE: 当前仅针对 `lazy` 模块公开
    pub(crate) fn poison(&self) {
        self.set_state_on_drop_to.set(POISONED);
    }
}