第11章 实现 Arc 和 Mutex
了解理论是很好的，但是理解某件事最好的方法是使用它。为了更好地理解原子和内部可变性，我们将实现标准库中的Arc和Mutex类型。
TODO：编写Mutex章节。
实现 Arc
在本节中，我们将实现一个更简单的std::sync::Arc。与我们之前做的Vec的实现类似，我们不会像标准库那样利用许多优化、内建指令或不稳定的代码。
这个实现大致上基于标准库的实现（技术上可以认为是取自 1.49 中的alloc::sync，因为它实际上是在那里实现的），但它目前不支持弱引用，因为它们使实现稍微复杂一些。
请注意，这一部分目前还处于 WIP 阶段。
布局
让我们开始为我们的Arc的实现做布局。
一个Arc<T>为T类型的值提供了线程安全的共享所有权，并在堆中分配。在 Rust 中，共享意味着不变性，所以我们不需要设计任何东西来管理对该值的访问，对吧？虽然像 Mutex 这样的内部可变性类型允许 Arc 的用户创建共享可变性，但 Arc 本身并不需要关注这些问题。
然而，有一个地方 Arc 需要关注可变：销毁。当 Arc 的所有所有者都销毁时，我们需要能够drop其内容并释放其分配。所以我们需要一种方法让所有者知道它是否是最后一个所有者，而最简单的方法就是对所有者进行计数——引用计数。
不幸的是，这种引用计数本质上是共享的可变状态，所以 Arc 需要 考虑同步问题。我们可以为此使用 Mutex，但那太过于杀鸡用牛刀了。相反，我们将使用 atomics。既然每个人都需要一个指向 T 的分配的指针，我们也可以把引用计数放在同一个分配中。
直观地说，它看起来就像这样：
#![allow(unused)] fn main() { use std::sync::atomic; pub struct Arc<T> { ptr: *mut ArcInner<T>, } pub struct ArcInner<T> { rc: atomic::AtomicUsize, data: T, } }
这可以编译通过，然而它是不正确的。首先，编译器会给我们太严格的可变性。例如，在期望使用Arc<&'a str>的地方不能使用Arc<&'static str>。更重要的是，它将给 drop checker 提供不正确的所有权信息，因为它将假定我们不拥有任何T类型的值。由于这是一个提供值的共享所有权的结构，在某些时候会有一个完全拥有其数据的结构实例。参见关于所有权和生命周期的章节，了解关于变异和 drop checker 的所有细节。
为了解决第一个问题，我们可以使用NonNull<T>。请注意，NonNull<T>是一个围绕原始指针的包装，并声明以下内容：
- 我们是
T的协变 - 我们的指针从不为空
为了解决第二个问题，我们可以包含一个包含ArcInner<T>的PhantomData标记。这将告诉 drop checker，我们对ArcInner<T>（它本身包含T）的值有一些所有权的概念。
通过这些改变，我们得到了最终的结构：
#![allow(unused)] fn main() { use std::marker::PhantomData; use std::ptr::NonNull; use std::sync::atomic::AtomicUsize; pub struct Arc<T> { ptr: NonNull<ArcInner<T>>, phantom: PhantomData<ArcInner<T>>, } pub struct ArcInner<T> { rc: AtomicUsize, data: T, } }
基本代码
现在我们已经确定了实现Arc的布局，让我们开始写一些基本代码。
构建 Arc
我们首先需要一种方法来构造一个Arc<T>。
这很简单，因为我们只需要把ArcInner<T>扔到一个 Box 里并得到一个NonNull<T>的指针。
impl<T> Arc<T> {
pub fn new(data: T) -> Arc<T> {
// 当前的指针就是第一个引用，因此初始时设置 count 为 1
let boxed = Box::new(ArcInner {
rc: AtomicUsize::new(1),
data,
});
Arc {
// 我们从 Box::into_raw 得到该指针，因此使用 `.unwrap()` 是完全可行的
ptr: NonNull::new(Box::into_raw(boxed)).unwrap(),
phantom: PhantomData,
}
}
}Send 和 Sync
由于我们正在构建并发原语，因此我们需要能够跨线程发送它。因此，我们可以实现Send和Sync标记特性。有关这些的更多信息，请参阅有关Send和Sync的部分。
这是没问题的，因为：
- 当且仅当你拥有唯一的 Arc 引用时，你才能获得其引用数据的可变引用（这仅发生在
Drop中） - 我们使用原子操作进行共享可变引用计数
unsafe impl<T: Sync + Send> Send for Arc<T> {}
unsafe impl<T: Sync + Send> Sync for Arc<T> {}我们需要约束T: Sync + Send，因为如果我们不提供这些约束，就有可能通过Arc跨越线程边界共享不安全的值，这有可能导致数据竞争或不可靠。
例如，如果没有这些约束，Arc<Rc<u32>>将是Sync + Send，这意味着你可以从Arc中克隆出Rc来跨线程发送（不需要创建一个全新的Rc），这将产生数据竞争，因为Rc不是线程安全的.
获取ArcInner
为了将NonNull<T>指针解引用为T，我们可以调用NonNull::as_ref。这是不安全的，与普通的as_ref函数不同，所以我们必须这样调用它。
unsafe { self.ptr.as_ref() }在这段代码中，我们将多次使用这个片段（通常与相关的let绑定）。
这种不安全是没问题的，因为当这个Arc存活的时候，我们可以保证内部指针是有效的。
Deref
好了。现在我们可以制作Arc了（很快就能正确地克隆和销毁它们），但是我们怎样才能获得里面的数据呢？
我们现在需要的是一个Deref的实现。
我们需要导入该 Trait：
use std::ops::Deref;这里是实现：
impl<T> Deref for Arc<T> {
type Target = T;
fn deref(&self) -> &T {
let inner = unsafe { self.ptr.as_ref() };
&inner.data
}
}看着很简单，对不？这只是解除了对ArcInner<T>的NonNull指针的引用，然后得到了对里面数据的引用。
代码
下面是本节的所有代码。
use std::ops::Deref;
impl<T> Arc<T> {
pub fn new(data: T) -> Arc<T> {
// 当前的指针就是第一个引用，因此初始时设置 count 为 1
let boxed = Box::new(ArcInner {
rc: AtomicUsize::new(1),
data,
});
Arc {
// 我们从 Box::into_raw 得到该指针，因此使用 `.unwrap()` 是完全可行的
ptr: NonNull::new(Box::into_raw(boxed)).unwrap(),
phantom: PhantomData,
}
}
}
unsafe impl<T: Sync + Send> Send for Arc<T> {}
unsafe impl<T: Sync + Send> Sync for Arc<T> {}
impl<T> Deref for Arc<T> {
type Target = T;
fn deref(&self) -> &T {
let inner = unsafe { self.ptr.as_ref() };
&inner.data
}
}克隆
现在我们已经有了一些基本的代码，我们需要一种方法来克隆Arc。
我们大致需要：
- 递增原子引用计数
- 从内部指针构建一个新的
Arc实例
首先，我们需要获得对ArcInner的访问。
let inner = unsafe { self.ptr.as_ref() };我们可以通过以下方式更新原子引用计数：
let old_rc = inner.rc.fetch_add(1, Ordering::???);但是我们在这里应该使用什么顺序？我们实际上没有任何代码在克隆时需要原子同步，因为我们在克隆时不修改内部值。因此，我们可以在这里使用 Relaxed 顺序，这意味着没有 happen-before 的关系，但却是原子性的。然而，当Drop Arc 时，我们需要在递减引用计数时进行原子同步。这在关于Arc的Drop实现部分中有更多描述。关于原子关系和 Relaxed ordering 的更多信息，请参见atomics 部分。
因此，代码变成了这样：
let old_rc = inner.rc.fetch_add(1, Ordering::Relaxed);我们需要增加一个导入来使用Ordering。
#![allow(unused)] fn main() { use std::sync::atomic::Ordering; }
然而，我们现在的这个实现有一个问题：如果有人决定mem::forget一堆 Arc 怎么办？到目前为止，我们所写的代码（以及将要写的代码）假设引用计数准确地描绘了内存中的 Arc 的数量，但在mem::forget的情况下，这是错误的。因此，当越来越多的 Arc 从这个 Arc 中克隆出来，而它们又没有被Drop和参考计数被递减时，我们就会溢出！这将导致释放后使用（use-after-free）。这是非常糟糕的事情！
为了处理这个问题，我们需要检查引用计数是否超过某个任意值（低于usize::MAX，因为我们把引用计数存储为AtomicUsize），并做一些防御。
标准库的实现决定，如果任何线程上的引用计数达到isize::MAX（大约是usize::MAX的一半），就直接中止程序（因为在正常代码中这是非常不可能的情况，如果它发生，程序可能是非常有问题的）。基于的假设是，不应该有大约 20 亿个线程（或者在一些 64 位机器上大约9万亿个）在同时增加引用计数。这就是我们要做的。
实现这种行为是非常简单的。
if old_rc >= isize::MAX as usize {
std::process::abort();
}然后，我们需要返回一个新的Arc的实例。
Self {
ptr: self.ptr,
phantom: PhantomData
}现在，让我们把这一切包在Clone的实现中。
use std::sync::atomic::Ordering;
impl<T> Clone for Arc<T> {
fn clone(&self) -> Arc<T> {
let inner = unsafe { self.ptr.as_ref() };
// 我们没有修改 Arc 中的数据，因此在这里不需要任何原子的同步操作，
// 使用 relax 这种排序方式也就完全可行了
let old_rc = inner.rc.fetch_add(1, Ordering::Relaxed);
if old_rc >= isize::MAX as usize {
std::process::abort();
}
Self {
ptr: self.ptr,
phantom: PhantomData,
}
}
}丢弃
我们现在需要一种方法来减少引用计数，并在计数足够低时丢弃数据，否则数据将永远存在于堆中。
为了做到这一点，我们可以实现Drop。
我们大致需要：
- 递减引用计数
- 如果数据只剩下一个引用，那么：
- 原子化地对数据进行屏障，以防止对数据的使用和删除进行重新排序
- 丢弃内部数据
首先，我们需要获得对ArcInner的访问：
let inner = unsafe { self.ptr.as_ref() };现在，我们需要递减引用计数。为了简化我们的代码，如果从fetch_sub返回的值（递减引用计数之前的值）不等于1，我们可以直接返回（我们不是数据的最后一个引用）。
if inner.rc.fetch_sub(1, Ordering::Release) != 1 {
return;
}然后我们需要创建一个原子屏障来防止重新排序使用数据和删除数据。正如标准库对Arc的实现中所述。
需要这个内存屏障来防止数据使用的重新排序和数据的删除。因为它被标记为
Release，引用计数的减少与Acquire屏障同步。这意味着数据的使用发生在减少引用计数之前，而减少引用计数发生在这个屏障之前，而屏障发生在数据的删除之前。（译者注：use < decrease < 屏障 < delete）正如Boost 文档中所解释的那样。
强制要求一个线程中对该对象的任何可能的访问（通过现有的引用）发生在不同线程中删除该对象之前是很重要的。这可以通过在丢弃一个引用后的“Release”操作来实现（任何通过该引用对对象的访问显然必须在之前发生），以及在删除对象前的“Acquire”操作。
特别是，虽然 Arc 的内容通常是不可改变的，但有可能对类似 Mutex 的东西进行内部可变。由于 Mutex 在被删除时不会被获取，我们不能依靠它的同步逻辑来使线程 A 的写操作对线程 B 的析构器可见。
还要注意的是，这里的 Acquire fence 可能可以用 Acquire load 来代替，这可以在高度竞争的情况下提高性能。 参见2。
为了做到这一点，我们可以这么做：
#![allow(unused)] fn main() { use std::sync::atomic::Ordering; use std::sync::atomic; atomic::fence(Ordering::Acquire); }
最后，我们可以 drop 数据本身。我们使用Box::from_raw来丢弃 Box 中的ArcInner<T>和它的数据。这需要一个*mut T而不是NonNull<T>，所以我们必须使用NonNull::as_ptr进行转换。
unsafe { Box::from_raw(self.ptr.as_ptr()); }这是安全的，因为我们知道我们拥有的是最后一个指向ArcInner的指针，而且其指针是有效的。
现在，让我们在Drop的实现中把这一切整合起来。
impl<T> Drop for Arc<T> {
fn drop(&mut self) {
let inner = unsafe { self.ptr.as_ref() };
if inner.rc.fetch_sub(1, Ordering::Release) != 1 {
return;
}
// 我们需要防止针对 inner 的使用和删除的重排序，
// 因此使用 fence 来进行保护是非常有必要的
atomic::fence(Ordering::Acquire);
// 安全保证：我们知道这是最后一个对 ArcInner 的引用，并且这个指针是有效的
unsafe { Box::from_raw(self.ptr.as_ptr()); }
}
}最终代码
这就是我们的最终代码，我在这里加了一些额外的注释并排序了一下 imports：
#![allow(unused)] fn main() { use std::marker::PhantomData; use std::ops::Deref; use std::ptr::NonNull; use std::sync::atomic::{self, AtomicUsize, Ordering}; pub struct Arc<T> { ptr: NonNull<ArcInner<T>>, phantom: PhantomData<ArcInner<T>>, } pub struct ArcInner<T> { rc: AtomicUsize, data: T, } impl<T> Arc<T> { pub fn new(data: T) -> Arc<T> { // 当前的指针就是第一个引用，因此初始时设置 count 为 1 let boxed = Box::new(ArcInner { rc: AtomicUsize::new(1), data, }); Arc { // 我们从 Box::into_raw 得到该指针，因此使用 `.unwrap()` 是完全可行的 ptr: NonNull::new(Box::into_raw(boxed)).unwrap(), phantom: PhantomData, } } } unsafe impl<T: Sync + Send> Send for Arc<T> {} unsafe impl<T: Sync + Send> Sync for Arc<T> {} impl<T> Deref for Arc<T> { type Target = T; fn deref(&self) -> &T { let inner = unsafe { self.ptr.as_ref() }; &inner.data } } impl<T> Clone for Arc<T> { fn clone(&self) -> Arc<T> { let inner = unsafe { self.ptr.as_ref() }; // 我们没有修改 Arc 中的数据，因此在这里不需要任何原子的同步操作， // 使用 relax 这种排序方式也就完全可行 let old_rc = inner.rc.fetch_add(1, Ordering::Relaxed); if old_rc >= isize::MAX as usize { std::process::abort(); } Self { ptr: self.ptr, phantom: PhantomData, } } } impl<T> Drop for Arc<T> { fn drop(&mut self) { let inner = unsafe { self.ptr.as_ref() }; if inner.rc.fetch_sub(1, Ordering::Release) != 1 { return; } // 我们需要防止针对 inner 的使用和删除的重排序 // 因此使用 fence 来进行保护是非常有必要 atomic::fence(Ordering::Acquire); // 安全保证：我们知道这是最后一个对 ArcInner 的引用，并且这个指针是有效的 unsafe { Box::from_raw(self.ptr.as_ptr()); } } } }
