swmr_cell/lib.rs
1//! # SWMR Version-Based Single Object
2//!
3//! This crate provides a single-writer, multi-reader (SWMR) cell that supports
4//! concurrent wait-free reads and lock-free writes using version-based garbage collection.
5//!
6//! ## Core Concepts
7//!
8//! - **Single Object**: The `swmr_cell` library manages a single versioned object per `SwmrCell`.
9//! - **Version**: The version counter represents the state of the object. Each write increments the version.
10//! - **Pinning**: Readers pin the current version when they start reading, preventing the writer from reclaiming that version (and any older versions still visible to other readers) until they are done.
11//!
12//! ## Typical Usage
13//!
14//! ```rust
15//! use swmr_cell::SwmrCell;
16//!
17//! // 1. Create a new SWMR cell with an initial value
18//! let mut cell = SwmrCell::new(42i32);
19//!
20//! // 2. Create a local reader for this thread (or pass to another thread)
21//! let local = cell.local_reader();
22//!
23//! // 3. Pin and read the value by dereferencing the guard
24//! let guard = local.pin();
25//! assert_eq!(*guard, 42);
26//! drop(guard);
27//!
28//! // 4. Writer updates the value
29//! cell.store(100i32);
30//!
31//! // 5. Read the new value
32//! let guard = local.pin();
33//! assert_eq!(*guard, 100);
34//! drop(guard);
35//!
36//! // 6. Manually collect garbage (optional, happens automatically too)
37//! cell.collect();
38//! ```
39#![cfg_attr(not(feature = "std"), no_std)]
40
41#[cfg(not(feature = "std"))]
42extern crate alloc;
43
44#[cfg(all(not(feature = "std"), test))]
45extern crate std;
46
47mod shim;
48
49#[cfg(test)]
50mod tests;
51use crate::shim::{
52 Arc, AtomicPtr, AtomicUsize, Box, Cell, Mutex, Ordering, Vec, VecDeque, heavy_barrier,
53 light_barrier,
54};
55use core::{fmt, marker::PhantomData, ops::Deref};
56
57/// Default threshold for automatic garbage reclamation (count of retired nodes).
58/// 自动垃圾回收的默认阈值(已退休节点的数量)。
59pub(crate) const AUTO_RECLAIM_THRESHOLD: usize = 16;
60
61/// Represents a reader that is not currently pinned to any version.
62/// 表示当前未被钉住到任何版本的读者。
63pub(crate) const INACTIVE_VERSION: usize = usize::MAX;
64
65/// A single-writer, multi-reader cell with version-based garbage collection.
66///
67/// `SwmrCell` provides safe concurrent access where one writer can update the value
68/// and multiple readers can read it concurrently. Readers access the value by
69/// creating a `LocalReader` and pinning it.
70///
71/// 单写多读单元,带有基于版本的垃圾回收。
72///
73/// `SwmrCell` 提供安全的并发访问,其中一个写入者可以更新值,
74/// 多个读者可以并发读取它。读者通过创建 `LocalReader` 并 pin 来访问值。
75pub struct SwmrCell<T: 'static, const RP: bool = false> {
76 shared: Arc<SharedState<T, RP>>,
77 garbage: GarbageSet<T>,
78 auto_reclaim_threshold: Option<usize>,
79}
80
81impl<T: 'static> SwmrCell<T, false> {
82 /// Create a new SWMR cell with default settings and the given initial value.
83 ///
84 /// 使用默认设置和给定的初始值创建一个新的 SWMR 单元。
85 #[inline]
86 pub fn new(data: T) -> Self {
87 Self::builder().build(data)
88 }
89
90 /// Returns a builder for configuring the SWMR cell.
91 ///
92 /// 返回用于配置 SWMR 单元的构建器。
93 #[inline]
94 pub fn builder() -> SwmrCellBuilder<T, false> {
95 SwmrCellBuilder::default()
96 }
97}
98
99impl<T: 'static, const RP: bool> SwmrCell<T, RP> {
100 /// Create a new `LocalReader` for reading.
101 ///
102 /// Each thread should create its own `LocalReader` and reuse it.
103 /// `LocalReader` is `!Sync` and should not be shared between threads.
104 ///
105 /// 创建一个新的 `LocalReader` 用于读取。
106 /// 每个线程应该创建自己的 `LocalReader` 并重复使用。
107 /// `LocalReader` 是 `!Sync` 的,不应在线程之间共享。
108 #[inline]
109 pub fn local_reader(&self) -> LocalReader<T, RP> {
110 LocalReader::new(self.shared.clone())
111 }
112
113 /// Create a new `SwmrReaderFactory` that can be shared across threads.
114 ///
115 /// `SwmrReaderFactory` is `Sync` + `Clone` and acts as a factory for `LocalReader`s.
116 /// This is useful for distributing reader creation capability to other threads.
117 ///
118 /// 创建一个新的 `SwmrReaderFactory`,可以在线程之间共享。
119 /// `SwmrReaderFactory` 是 `Sync` + `Clone` 的,充当 `LocalReader` 的工厂。
120 /// 这对于将读者创建能力分发给其他线程很有用。
121 #[inline]
122 pub fn reader_factory(&self) -> SwmrReaderFactory<T, RP> {
123 SwmrReaderFactory {
124 shared: self.shared.clone(),
125 }
126 }
127
128 /// Store a new value, making it visible to readers.
129 /// The old value is retired and will be garbage collected.
130 ///
131 /// This operation increments the global version.
132 ///
133 /// 存储新值,使其对读者可见。
134 /// 旧值已退休,将被垃圾回收。
135 /// 此操作会增加全局版本。
136 pub fn store(&mut self, data: T) {
137 let new_ptr = Box::into_raw(Box::new(data));
138 let old_ptr = self.shared.ptr.swap(new_ptr, Ordering::Release);
139
140 // Increment global version.
141 // The old value belongs to the previous version (the one before this increment).
142 // 增加全局版本。
143 // 旧值属于前一个版本(此次增加之前的那个)。
144 let old_version = self.shared.global_version.fetch_add(1, Ordering::AcqRel);
145
146 if !old_ptr.is_null() {
147 // Safe because we just swapped it out and we own the writer
148 unsafe {
149 self.garbage.add(Box::from_raw(old_ptr), old_version);
150 }
151 }
152
153 // Auto-reclaim
154 if let Some(threshold) = self.auto_reclaim_threshold
155 && self.garbage.len() > threshold
156 {
157 self.collect();
158 }
159 }
160
161 /// Get a reference to the previously stored value, if any.
162 ///
163 /// Returns `None` if no previous value exists (i.e., only the initial value has been stored).
164 ///
165 /// **Note**: The previous value is guaranteed not to be garbage collected because
166 /// `collect()` uses `safety_limit = current_version - 2`, which always preserves
167 /// the most recently retired value (version = current_version - 1).
168 ///
169 /// This is useful for comparing the current value with the previous one,
170 /// or for implementing undo/rollback logic.
171 ///
172 /// 获取上一个存储值的引用(如果存在)。
173 ///
174 /// 如果不存在上一个值(即只存储了初始值),则返回 `None`。
175 ///
176 /// **注意**:上一个值保证不会被垃圾回收,因为 `collect()` 使用 `safety_limit = current_version - 2`,
177 /// 这始终保留最近退休的值(版本 = current_version - 1)。
178 ///
179 /// 这对于将当前值与上一个值进行比较,或实现撤销/回滚逻辑很有用。
180 ///
181 /// # Example
182 ///
183 /// ```rust
184 /// use swmr_cell::SwmrCell;
185 ///
186 /// let mut cell = SwmrCell::new(1);
187 /// assert!(cell.previous().is_none()); // No previous value yet
188 ///
189 /// cell.store(2);
190 /// assert_eq!(cell.previous(), Some(&1)); // Previous value is 1
191 ///
192 /// cell.store(3);
193 /// assert_eq!(cell.previous(), Some(&2)); // Previous value is 2
194 /// ```
195 #[inline]
196 pub fn previous(&self) -> Option<&T> {
197 self.garbage.back()
198 }
199
200 /// Get a reference to the current value (writer-only, no pinning required).
201 ///
202 /// This is only accessible from the writer thread since `SwmrCell` is `!Sync`.
203 ///
204 /// 获取当前值的引用(仅写者可用,无需 pin)。
205 /// 这只能从写者线程访问,因为 `SwmrCell` 是 `!Sync` 的。
206 #[inline]
207 pub fn get(&self) -> &T {
208 // Safety: We own the writer, and the current pointer is always valid.
209 // 安全性:我们拥有写者,当前指针始终有效。
210 unsafe { &*self.shared.ptr.load(Ordering::Acquire) }
211 }
212
213 /// Update the value using a closure.
214 ///
215 /// The closure receives the current value and should return the new value.
216 /// This is equivalent to `cell.store(f(cell.get().clone()))` but more ergonomic.
217 ///
218 /// 使用闭包更新值。
219 /// 闭包接收当前值并应返回新值。
220 /// 这相当于 `cell.store(f(cell.get().clone()))` 但更符合人体工程学。
221 #[inline]
222 pub fn update<F>(&mut self, f: F)
223 where
224 F: FnOnce(&T) -> T,
225 {
226 let new_value = f(self.get());
227 self.store(new_value);
228 }
229
230 /// Get the current global version.
231 ///
232 /// The version is incremented each time `store()` or `replace()` is called.
233 ///
234 /// 获取当前全局版本。
235 /// 每次调用 `store()` 或 `replace()` 时版本会增加。
236 #[inline]
237 pub fn version(&self) -> usize {
238 self.shared.global_version.load(Ordering::Acquire)
239 }
240
241 /// Get the number of retired objects waiting for garbage collection.
242 ///
243 /// 获取等待垃圾回收的已退休对象数量。
244 #[inline]
245 pub fn garbage_count(&self) -> usize {
246 self.garbage.len()
247 }
248
249 /// Manually trigger garbage collection.
250 /// 手动触发垃圾回收。
251 pub fn collect(&mut self) {
252 // In this design, we don't necessarily advance the version just for collection.
253 // But we need to find min_active_version.
254
255 let current_version = self.shared.global_version.load(Ordering::Acquire);
256
257 // Safety limit ensures we never reclaim the most recent retired value (previous).
258 // The most recent retired value has version = current_version - 1.
259 // With safety_limit = current_version - 2, we only reclaim versions < current_version - 2,
260 // so the previous value (version = current_version - 1) is always preserved.
261 // 安全限制确保我们永远不会回收最近退休的值(previous)。
262 // 最近退休的值的版本 = current_version - 1。
263 // 使用 safety_limit = current_version - 2,我们只回收版本 < current_version - 2 的,
264 // 因此上一个值(版本 = current_version - 1)始终被保留。
265 let safety_limit = current_version.saturating_sub(2);
266
267 let mut min_active = current_version;
268
269 // Force memory visibility of any preceding stores and serialize reader streams.
270 // This ensures we see any active readers that have completed their light_barrier.
271 heavy_barrier::<RP>();
272
273 let mut shared_readers = self.shared.readers.lock();
274
275 for arc_slot in shared_readers.iter() {
276 let version = arc_slot.active_version.load(Ordering::Acquire);
277 if version != INACTIVE_VERSION {
278 min_active = min_active.min(version);
279 }
280 }
281
282 // Clean up dead reader slots (strong_count == 1 means only SharedState holds it)
283 // 清理死读者槽(strong_count == 1 表示只有 SharedState 持有它)
284 shared_readers.retain(|arc_slot| Arc::strong_count(arc_slot) > 1);
285
286 drop(shared_readers);
287
288 let reclaim_threshold = min_active.min(safety_limit);
289
290 self.shared
291 .min_active_version
292 .store(reclaim_threshold, Ordering::Release);
293
294 self.garbage.collect(reclaim_threshold, current_version);
295 }
296}
297
298/// A handle for creating `LocalReader`s that can be shared across threads.
299///
300/// Unlike `LocalReader`, which is `!Sync` and bound to a single thread,
301/// `SwmrReaderFactory` is `Sync` and `Clone`. It holds a reference to the shared state
302/// but does not register a reader slot until `local_reader()` is called.
303///
304/// 可以跨线程共享的用于创建 `LocalReader` 的句柄。
305///
306/// 与 `!Sync` 且绑定到单个线程的 `LocalReader` 不同,
307/// `SwmrReaderFactory` 是 `Sync` 和 `Clone` 的。它持有对共享状态的引用,
308/// 但直到调用 `local_reader()` 时才注册读者槽。
309pub struct SwmrReaderFactory<T: 'static, const RP: bool = false> {
310 shared: Arc<SharedState<T, RP>>,
311}
312
313impl<T: 'static, const RP: bool> SwmrReaderFactory<T, RP> {
314 /// Create a new `LocalReader` for the current thread.
315 ///
316 /// 为当前线程创建一个新的 `LocalReader`。
317 #[inline]
318 pub fn local_reader(&self) -> LocalReader<T, RP> {
319 LocalReader::new(self.shared.clone())
320 }
321}
322
323impl<T: 'static, const RP: bool> Clone for SwmrReaderFactory<T, RP> {
324 #[inline]
325 fn clone(&self) -> Self {
326 Self {
327 shared: self.shared.clone(),
328 }
329 }
330}
331
332impl<T: 'static, const RP: bool> fmt::Debug for SwmrReaderFactory<T, RP> {
333 fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result {
334 f.debug_struct("SwmrReaderFactory")
335 .field("read_preferred", &RP)
336 .finish()
337 }
338}
339
340/// A builder for configuring and creating a SWMR cell.
341///
342/// 用于配置和创建 SWMR 单元的构建器。
343pub struct SwmrCellBuilder<T, const RP: bool = false> {
344 auto_reclaim_threshold: Option<usize>,
345 marker: PhantomData<T>,
346}
347
348impl<T: 'static, const RP: bool> SwmrCellBuilder<T, RP> {
349 /// Sets the threshold for automatic garbage reclamation.
350 ///
351 /// When the number of retired objects exceeds this threshold,
352 /// garbage collection is triggered automatically during `store`.
353 ///
354 /// Set to `None` to disable automatic reclamation.
355 /// Default is `Some(64)`.
356 ///
357 /// 设置自动垃圾回收的阈值。
358 /// 当已退休对象的数量超过此阈值时,将在 `store` 期间自动触发垃圾回收。
359 /// 设置为 `None` 以禁用自动回收。
360 /// 默认为 `Some(64)`。
361 #[inline]
362 pub fn auto_reclaim_threshold(mut self, threshold: Option<usize>) -> Self {
363 self.auto_reclaim_threshold = threshold;
364 self
365 }
366
367 /// Creates a new SWMR cell with the configured settings and initial value.
368 ///
369 /// 使用配置的设置和初始值创建一个新的 SWMR 单元。
370 pub fn build(self, data: T) -> SwmrCell<T, RP> {
371 let shared = Arc::new(SharedState {
372 global_version: AtomicUsize::new(0),
373 min_active_version: AtomicUsize::new(0),
374 ptr: AtomicPtr::new(Box::into_raw(Box::new(data))),
375 readers: Mutex::new(Vec::new()),
376 });
377
378 SwmrCell {
379 shared,
380 garbage: GarbageSet::new(),
381 auto_reclaim_threshold: self.auto_reclaim_threshold,
382 }
383 }
384}
385
386impl<T: 'static, const RP: bool> Default for SwmrCellBuilder<T, RP> {
387 fn default() -> Self {
388 SwmrCellBuilder {
389 auto_reclaim_threshold: Some(AUTO_RECLAIM_THRESHOLD),
390 marker: PhantomData,
391 }
392 }
393}
394
395impl<T: 'static> SwmrCellBuilder<T, false> {
396 /// Enable read-preferred mode.
397 ///
398 /// 启用读优先模式。
399 #[inline]
400 pub fn read_preferred(self) -> SwmrCellBuilder<T, true> {
401 SwmrCellBuilder {
402 auto_reclaim_threshold: self.auto_reclaim_threshold,
403 marker: PhantomData,
404 }
405 }
406}
407
408/// Manages retired objects and their reclamation.
409///
410/// This struct encapsulates the logic for:
411/// - Storing retired objects in version-ordered queue.
412/// - Reclaiming objects when they are safe to delete.
413///
414/// 管理已退休对象及其回收。
415///
416/// 此结构体封装了以下逻辑:
417/// - 将已退休对象存储在按版本排序的队列中。
418/// - 当对象可以安全删除时进行回收。
419struct GarbageSet<T> {
420 /// Queue of garbage items, ordered by version.
421 /// Each element is (version, node).
422 queue: VecDeque<(usize, Box<T>)>,
423}
424
425impl<T> GarbageSet<T> {
426 /// Create a new empty garbage set.
427 /// 创建一个新的空垃圾集合。
428 fn new() -> Self {
429 Self {
430 queue: VecDeque::new(),
431 }
432 }
433
434 /// Get the total number of retired objects.
435 /// 获取已退休对象的总数。
436 #[inline]
437 fn len(&self) -> usize {
438 self.queue.len()
439 }
440
441 /// Get a reference to the most recently retired object (the previous value).
442 /// 获取最近退休对象(上一个值)的引用。
443 #[inline]
444 fn back(&self) -> Option<&T> {
445 self.queue.back().map(|(_, boxed)| boxed.as_ref())
446 }
447
448 /// Add a retired node to the set for the current version.
449 ///
450 /// 将已退休节点添加到当前版本的集合中。
451 #[inline]
452 fn add(&mut self, node: Box<T>, current_version: usize) {
453 self.queue.push_back((current_version, node));
454 }
455
456 /// Reclaim garbage that is safe to delete.
457 ///
458 /// Garbage from versions older than `min_active_version` is dropped.
459 ///
460 /// 回收可以安全删除的垃圾。
461 ///
462 /// 来自比 `min_active_version` 更旧的版本的垃圾将被 drop。
463 #[inline]
464 fn collect(&mut self, min_active_version: usize, _current_version: usize) {
465 // We reclaim everything that is strictly older than min_active_version.
466 // If min_active_version == current_version, then everything (all < current_version) is reclaimed.
467 while let Some((version, _)) = self.queue.front() {
468 if *version >= min_active_version {
469 break;
470 }
471 self.queue.pop_front(); // Box<T> is dropped here
472 }
473 }
474}
475
476/// A slot allocated for a reader thread to record its active version.
477///
478/// Cache-aligned to prevent false sharing between readers.
479///
480/// 为读者线程分配的槽,用于记录其活跃版本。
481/// 缓存对齐以防止读者之间的伪共享。
482#[derive(Debug)]
483#[repr(align(64))]
484pub(crate) struct ReaderSlot {
485 /// The version currently being accessed by the reader, or INACTIVE_VERSION.
486 /// 读者当前访问的版本,或 INACTIVE_VERSION。
487 pub(crate) active_version: AtomicUsize,
488}
489
490/// Global shared state for the version GC domain.
491///
492/// Contains the global version, the minimum active version, the data pointer, and the list of reader slots.
493///
494/// version GC 域的全局共享状态。
495/// 包含全局版本、最小活跃版本、数据指针和读者槽列表。
496#[repr(align(64))]
497pub(crate) struct SharedState<T: 'static, const RP: bool = false> {
498 /// The global monotonic version counter.
499 /// 全局单调版本计数器。
500 pub(crate) global_version: AtomicUsize,
501 /// The minimum version among all active readers (cached for performance).
502 /// 所有活跃读者中的最小版本(为性能而缓存)。
503 pub(crate) min_active_version: AtomicUsize,
504 /// The current data pointer.
505 /// 当前数据指针。
506 pub(crate) ptr: AtomicPtr<T>,
507 /// List of all registered reader slots. Protected by a Mutex.
508 /// 所有注册读者槽的列表。由 Mutex 保护。
509 pub(crate) readers: Mutex<Vec<Arc<ReaderSlot>>>,
510}
511
512impl<T: 'static, const RP: bool> Drop for SharedState<T, RP> {
513 fn drop(&mut self) {
514 // Drop the current value held by ptr to avoid leaking it.
515 // Drop ptr 持有的当前值,以避免泄漏。
516 let ptr = self.ptr.load(Ordering::Acquire);
517 if !ptr.is_null() {
518 unsafe {
519 drop(Box::from_raw(ptr));
520 }
521 }
522 }
523}
524
525/// A reader thread's local version state.
526///
527/// Each reader thread should create exactly one `LocalReader` via `SwmrCell::local_reader()`.
528/// It is `!Sync` (due to `Cell`) and must be stored per-thread.
529///
530/// The `LocalReader` is used to:
531/// - Pin the thread to the current version via `pin()`.
532/// - Obtain a `PinGuard` that protects access to values and can be dereferenced.
533///
534/// **Thread Safety**: `LocalReader` is not `Sync` and must be used by only one thread.
535///
536/// 读者线程的本地版本状态。
537/// 每个读者线程应该通过 `SwmrCell::local_reader()` 创建恰好一个 `LocalReader`。
538/// 它是 `!Sync` 的(因为 `Cell`),必须在每个线程中存储。
539/// `LocalReader` 用于:
540/// - 通过 `pin()` 将线程钉住到当前版本。
541/// - 获取保护对值访问的 `PinGuard`,可以解引用来读取值。
542/// **线程安全性**:`LocalReader` 不是 `Sync` 的,必须仅由一个线程使用。
543pub struct LocalReader<T: 'static, const RP: bool = false> {
544 slot: Arc<ReaderSlot>,
545 shared: Arc<SharedState<T, RP>>,
546 pin_count: Cell<usize>,
547}
548
549impl<T: 'static, const RP: bool> LocalReader<T, RP> {
550 fn new(shared: Arc<SharedState<T, RP>>) -> Self {
551 let slot = Arc::new(ReaderSlot {
552 active_version: AtomicUsize::new(INACTIVE_VERSION),
553 });
554
555 // Register the reader immediately in the shared readers list
556 shared.readers.lock().push(Arc::clone(&slot));
557
558 LocalReader {
559 slot,
560 shared,
561 pin_count: Cell::new(0),
562 }
563 }
564
565 /// Pin this thread to the current version.
566 ///
567 /// Returns a `PinGuard` that keeps the thread pinned for its lifetime.
568 /// The guard can be dereferenced to access the current value.
569 ///
570 /// **Reentrancy**: This method is reentrant. Multiple calls can be nested, and the thread
571 /// remains pinned until all returned guards are dropped. You can also clone a guard to create
572 /// additional references: `let guard2 = guard1.clone();`
573 ///
574 /// **Example**:
575 /// ```ignore
576 /// let local = cell.local_reader();
577 /// let guard1 = local.pin();
578 /// let value = *guard1; // Dereference to read
579 /// let guard2 = local.pin(); // Reentrant call
580 /// let guard3 = guard1.clone(); // Clone for nested scope
581 /// // Thread remains pinned until all three guards are dropped
582 /// ```
583 ///
584 /// While pinned, the thread is considered "active" at a particular version,
585 /// and the garbage collector will not reclaim data from that version.
586 ///
587 /// 将此线程钉住到当前版本。
588 ///
589 /// 返回一个 `PinGuard`,在其生命周期内保持线程被钉住。
590 /// 可以解引用该守卫来访问当前值。
591 ///
592 /// **可重入性**:此方法是可重入的。多个调用可以嵌套,线程在所有返回的守卫被 drop 之前保持被钉住。
593 /// 你也可以克隆一个守卫来创建额外的引用:`let guard2 = guard1.clone();`
594 ///
595 /// **示例**:
596 /// ```ignore
597 /// let local = cell.local_reader();
598 /// let guard1 = local.pin();
599 /// let value = *guard1; // 解引用来读取
600 /// let guard2 = local.pin(); // 可重入调用
601 /// let guard3 = guard1.clone(); // 克隆用于嵌套作用域
602 /// // 线程保持被钉住直到所有三个守卫被 drop
603 /// ```
604 ///
605 /// 当被钉住时,线程被认为在特定版本"活跃",垃圾回收器不会回收该版本的数据。
606 /// Check if this reader is currently pinned.
607 ///
608 /// 检查此读者当前是否被 pin。
609 #[inline]
610 pub fn is_pinned(&self) -> bool {
611 self.pin_count.get() > 0
612 }
613
614 /// Get the current global version.
615 ///
616 /// Note: This returns the global version, not the pinned version.
617 /// To get the pinned version, use `PinGuard::version()`.
618 ///
619 /// 获取当前全局版本。
620 /// 注意:这返回全局版本,而不是 pin 的版本。
621 /// 要获取 pin 的版本,请使用 `PinGuard::version()`。
622 #[inline]
623 pub fn version(&self) -> usize {
624 self.shared.global_version.load(Ordering::Acquire)
625 }
626
627 #[inline]
628 pub fn pin(&self) -> PinGuard<'_, T, RP> {
629 let pin_count = self.pin_count.get();
630
631 // Reentrant pin: the version is already protected by the outer pin.
632 // Just increment count and reuse the existing pinned pointer.
633 // 可重入 pin:版本已经被外层 pin 保护。
634 // 只需增加计数并复用现有的 pinned 指针。
635 if pin_count > 0 {
636 self.pin_count.set(pin_count + 1);
637
638 // Load the pointer that corresponds to our already-pinned version.
639 // Since we're reentrant, we should see the same or newer pointer.
640 // 加载与我们已 pin 版本对应的指针。
641 // 由于是可重入的,我们应该看到相同或更新的指针。
642 let ptr = self.shared.ptr.load(Ordering::Acquire);
643 let version = self.slot.active_version.load(Ordering::Acquire);
644
645 return PinGuard {
646 local: self,
647 ptr,
648 version,
649 };
650 }
651
652 // First pin: need to acquire a version and validate it.
653 // 首次 pin:需要获取版本并验证。
654 loop {
655 let current_version = self.shared.global_version.load(Ordering::Acquire);
656
657 self.slot
658 .active_version
659 .store(current_version, Ordering::Release);
660
661 // Light barrier coupled with Writer's Heavy barrier prevents Store-Load reordering.
662 light_barrier::<RP>();
663
664 // Check if our version is still valid (not yet reclaimed).
665 // 检查我们的版本是否仍然有效(尚未被回收)。
666 let min_active = self.shared.min_active_version.load(Ordering::Acquire);
667
668 if current_version >= min_active {
669 break;
670 }
671
672 // Version was reclaimed between our read and store.
673 // Retry with a fresh version.
674 // 版本在我们读取和存储之间被回收了。
675 // 用新版本重试。
676 core::hint::spin_loop();
677 }
678
679 self.pin_count.set(1);
680
681 // Capture the pointer and version at pin time for snapshot semantics.
682 // 在 pin 时捕获指针和版本以实现快照语义。
683 let ptr = self.shared.ptr.load(Ordering::Acquire);
684 let version = self.slot.active_version.load(Ordering::Acquire);
685
686 PinGuard {
687 local: self,
688 ptr,
689 version,
690 }
691 }
692
693 /// Create a new `SwmrReaderFactory` from this `LocalReader`.
694 ///
695 /// `SwmrReaderFactory` is `Sync` + `Clone` and acts as a factory for `LocalReader`s.
696 /// This is equivalent to calling `swmr_cell.reader_factory()`, but using the `LocalReader`'s reference to the shared state.
697 ///
698 /// 从此 `LocalReader` 创建一个新的 `SwmrReaderFactory`。
699 /// `SwmrReaderFactory` 是 `Sync` + `Clone` 的,充当 `LocalReader` 的工厂。
700 /// 这相当于调用 `swmr_cell.reader_factory()`,但使用 `LocalReader` 对共享状态的引用。
701 #[inline]
702 pub fn reader_factory(&self) -> SwmrReaderFactory<T, RP> {
703 SwmrReaderFactory {
704 shared: self.shared.clone(),
705 }
706 }
707
708 /// Convert this `LocalReader` into a `SwmrReaderFactory`.
709 ///
710 /// This consumes the `LocalReader` and returns a `SwmrReaderFactory`
711 /// that can be sent to another thread to create new `LocalReader`s.
712 ///
713 /// 将此 `LocalReader` 转换为 `SwmrReaderFactory`。
714 /// 这会消耗 `LocalReader` 并返回一个 `SwmrReaderFactory`,
715 /// 该 `SwmrReaderFactory` 可以发送到另一个线程以创建新 `LocalReader`。
716 #[inline]
717 pub fn into_swmr(self) -> SwmrReaderFactory<T, RP> {
718 SwmrReaderFactory {
719 shared: self.shared.clone(),
720 }
721 }
722}
723
724impl<T: 'static, const RP: bool> Clone for LocalReader<T, RP> {
725 #[inline]
726 fn clone(&self) -> Self {
727 Self::new(self.shared.clone())
728 }
729}
730
731/// A guard that keeps the current thread pinned to a version.
732///
733/// `PinGuard` is obtained by calling `LocalReader::pin()`.
734/// It implements `Deref<Target = T>` to allow reading the current value.
735/// It is `!Send` and `!Sync` because it references a `!Sync` `LocalReader`.
736/// Its lifetime is bound to the `LocalReader` it came from.
737///
738/// While a `PinGuard` is held, the thread is considered "active" at a particular version,
739/// and the garbage collector will not reclaim data from that version.
740///
741/// `PinGuard` supports internal cloning via reference counting (increments the pin count),
742/// allowing nested pinning. The thread remains pinned until all cloned guards are dropped.
743///
744/// **Safety**: The `PinGuard` is the mechanism that ensures safe concurrent access to
745/// shared values. Readers must always hold a valid `PinGuard` when accessing
746/// shared data.
747///
748/// 一个保持当前线程被钉住到一个版本的守卫。
749/// `PinGuard` 通过调用 `LocalReader::pin()` 获得。
750/// 它实现了 `Deref<Target = T>`,允许读取当前值。
751/// 它是 `!Send` 和 `!Sync` 的,因为它引用了一个 `!Sync` 的 `LocalReader`。
752/// 它的生命周期被绑定到它来自的 `LocalReader`。
753/// 当 `PinGuard` 被持有时,线程被认为在特定版本"活跃",
754/// 垃圾回收器不会回收该版本的数据。
755/// `PinGuard` 支持通过引用计数的内部克隆(增加 pin 计数),允许嵌套 pinning。
756/// 线程保持被钉住直到所有克隆的守卫被 drop。
757/// **安全性**:`PinGuard` 是确保对值安全并发访问的机制。
758/// 读者在访问共享数据时必须始终持有有效的 `PinGuard`。
759#[must_use]
760pub struct PinGuard<'a, T: 'static, const RP: bool = false> {
761 local: &'a LocalReader<T, RP>,
762 /// The pointer captured at pin time for snapshot semantics.
763 /// 在 pin 时捕获的指针,用于快照语义。
764 ptr: *const T,
765 /// The version at pin time.
766 /// pin 时的版本。
767 version: usize,
768}
769
770impl<T: 'static, const RP: bool> PinGuard<'_, T, RP> {
771 /// Get the version that this guard is pinned to.
772 ///
773 /// 获取此守卫被 pin 到的版本。
774 #[inline]
775 pub fn version(&self) -> usize {
776 self.version
777 }
778}
779
780impl<'a, T, const RP: bool> Deref for PinGuard<'a, T, RP> {
781 type Target = T;
782
783 /// Dereference to access the pinned value.
784 ///
785 /// Returns a reference to the value that was current when this guard was created.
786 /// This provides snapshot semantics - the value won't change during the guard's lifetime.
787 ///
788 /// 解引用以访问被 pin 的值。
789 ///
790 /// 返回对创建此守卫时当前值的引用。
791 /// 这提供了快照语义 - 在守卫的生命周期内值不会改变。
792 #[inline]
793 fn deref(&self) -> &T {
794 // Safety: pin() guarantees pinned_version >= min_active,
795 // and the pointer was captured at pin time.
796 // The value is valid as long as guard is held.
797 // 安全性:pin() 保证 pinned_version >= min_active,
798 // 并且指针在 pin 时被捕获。
799 // 只要 guard 被持有,值就是有效的。
800 unsafe { &*self.ptr }
801 }
802}
803
804impl<'a, T, const RP: bool> Clone for PinGuard<'a, T, RP> {
805 /// Clone this guard to create a nested pin.
806 ///
807 /// Cloning increments the pin count, and the thread remains pinned until all cloned guards
808 /// are dropped. This allows multiple scopes to hold pins simultaneously.
809 ///
810 /// 克隆此守卫以创建嵌套 pin。
811 ///
812 /// 克隆会增加 pin 计数,线程保持被钉住直到所有克隆的守卫被 drop。
813 /// 这允许多个作用域同时持有 pin。
814 #[inline]
815 fn clone(&self) -> Self {
816 let pin_count = self.local.pin_count.get();
817
818 assert!(
819 pin_count > 0,
820 "BUG: Cloning a PinGuard in an unpinned state (pin_count = 0). \
821 This indicates incorrect API usage or a library bug."
822 );
823
824 self.local.pin_count.set(pin_count + 1);
825
826 PinGuard {
827 local: self.local,
828 ptr: self.ptr,
829 version: self.version,
830 }
831 }
832}
833
834impl<'a, T, const RP: bool> Drop for PinGuard<'a, T, RP> {
835 #[inline]
836 fn drop(&mut self) {
837 let pin_count = self.local.pin_count.get();
838
839 assert!(
840 pin_count > 0,
841 "BUG: Dropping a PinGuard in an unpinned state (pin_count = 0). \
842 This indicates incorrect API usage or a library bug."
843 );
844
845 if pin_count == 1 {
846 self.local
847 .slot
848 .active_version
849 .store(INACTIVE_VERSION, Ordering::Release);
850 }
851
852 self.local.pin_count.set(pin_count - 1);
853 }
854}
855
856impl<T: 'static, const RP: bool> AsRef<T> for PinGuard<'_, T, RP> {
857 #[inline]
858 fn as_ref(&self) -> &T {
859 self.deref()
860 }
861}
862
863// ============================================================================
864// Standard Trait Implementations
865// 标准 trait 实现
866// ============================================================================
867
868impl<T: Default + 'static, const RP: bool> Default for SwmrCell<T, RP> {
869 /// Create a new SWMR cell with the default value.
870 ///
871 /// 使用默认值创建一个新的 SWMR 单元。
872 #[inline]
873 fn default() -> Self {
874 SwmrCell::from(T::default())
875 }
876}
877
878impl<T: 'static, const RP: bool> From<T> for SwmrCell<T, RP> {
879 /// Create a new SWMR cell from a value.
880 ///
881 /// 从一个值创建一个新的 SWMR 单元。
882 #[inline]
883 fn from(value: T) -> Self {
884 SwmrCellBuilder::default().build(value)
885 }
886}
887
888impl<T: fmt::Debug + 'static, const RP: bool> fmt::Debug for SwmrCell<T, RP> {
889 #[inline]
890 fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result {
891 f.debug_struct("SwmrCell")
892 .field("value", self.get())
893 .field("version", &self.version())
894 .field("garbage_count", &self.garbage_count())
895 .field("reclaim_threshold", &self.auto_reclaim_threshold)
896 .field("is_read-preferred", &RP)
897 .finish()
898 }
899}
900
901impl<T: 'static, const RP: bool> fmt::Debug for LocalReader<T, RP> {
902 #[inline]
903 fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result {
904 f.debug_struct("LocalReader")
905 .field("is_pinned", &self.is_pinned())
906 .field("version", &self.version())
907 .field("is_read-preferred", &RP)
908 .finish()
909 }
910}
911
912impl<T: fmt::Debug + 'static, const RP: bool> fmt::Debug for PinGuard<'_, T, RP> {
913 #[inline]
914 fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result {
915 f.debug_struct("PinGuard")
916 .field("value", &self.deref())
917 .field("version", &self.version)
918 .field("is_read-preferred", &RP)
919 .finish()
920 }
921}