Rust std::sync::Arc

[水文] Rust std::sync::Arc

1
2
3
4
5
6
#[cfg_attr(not(test), lang = "arc")]
#[stable(feature = "rust1", since = "1.0.0")]
pub struct Arc<T: ?Sized> {
    ptr: NonNull<ArcInner<T>>,
    phantom: PhantomData<T>,
}

内部是：

1
2
3
4
5
6
7
8
9
10
struct ArcInner<T: ?Sized> {
    strong: atomic::AtomicUsize,

    // the value usize::MAX acts as a sentinel for temporarily "locking" the
    // ability to upgrade weak pointers or downgrade strong ones; this is used
    // to avoid races in `make_mut` and `get_mut`.
    weak: atomic::AtomicUsize,

    data: T,
}

Comparing with std::rc::Rc

内部可变形

其实你应该记得 Rc 实现用了一堆 Cell 来表示 counter, 

1
2
3
4
5
6
7
8
9
10
11
12
#[cfg_attr(not(test), lang = "rc")]
#[stable(feature = "rust1", since = "1.0.0")]
pub struct Rc<T: ?Sized> {
    ptr: NonNull<RcBox<T>>,
    phantom: PhantomData<T>,
}

struct RcBox<T: ?Sized> {
    strong: Cell<usize>,
    weak: Cell<usize>,
    value: T,
}

这里内部可变形中， ArcInner 靠的是 Atomic 来实现

方法

如果是 new 的话：

1
2
3
4
5
6
7
8
9
10
11
12
#[inline]
#[stable(feature = "rust1", since = "1.0.0")]
pub fn new(data: T) -> Arc<T> {
    // Start the weak pointer count as 1 which is the weak pointer that's
    // held by all the strong pointers (kinda), see std/rc.rs for more info
    let x: Box<_> = box ArcInner {
        strong: atomic::AtomicUsize::new(1),
        weak: atomic::AtomicUsize::new(1),
        data,
    };
    Self::from_inner(Box::into_raw_non_null(x))
}

对于 clone 来说，注释非常详细：

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
#[inline]
fn clone(&self) -> Arc<T> {
    // Using a relaxed ordering is alright here, as knowledge of the
    // original reference prevents other threads from erroneously deleting
    // the object.
    //
    // As explained in the [Boost documentation][1], Increasing the
    // reference counter can always be done with memory_order_relaxed: New
    // references to an object can only be formed from an existing
    // reference, and passing an existing reference from one thread to
    // another must already provide any required synchronization.
    //
    // [1]: (www.boost.org/doc/libs/1_55_0/doc/html/atomic/usage_examples.html)
    let old_size = self.inner().strong.fetch_add(1, Relaxed);

    // However we need to guard against massive refcounts in case someone
    // is `mem::forget`ing Arcs. If we don't do this the count can overflow
    // and users will use-after free. We racily saturate to `isize::MAX` on
    // the assumption that there aren't ~2 billion threads incrementing
    // the reference count at once. This branch will never be taken in
    // any realistic program.
    //
    // We abort because such a program is incredibly degenerate, and we
    // don't care to support it.
    if old_size > MAX_REFCOUNT {
        unsafe {
            abort();
        }
    }

    Self::from_inner(self.ptr)
}
  1. Relaxed 的 fetch_add 是原子的
  2. 如果有 old_size > MAX_REFCOUNT ，这里认为你一般出了 mem::forget 的 奇怪 bug

对于 Drop 来说：

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
#[inline]
fn drop(&mut self) {
    // Because `fetch_sub` is already atomic, we do not need to synchronize
    // with other threads unless we are going to delete the object. This
    // same logic applies to the below `fetch_sub` to the `weak` count.
    if self.inner().strong.fetch_sub(1, Release) != 1 {
        return;
    }

    // This fence is needed to prevent reordering of use of the data and
    // deletion of the data.  Because it is marked `Release`, the decreasing
    // of the reference count synchronizes with this `Acquire` fence. This
    // means that use of the data happens before decreasing the reference
    // count, which happens before this fence, which happens before the
    // deletion of the data.
    //
    // As explained in the [Boost documentation][1],
    //
    // > It is important to enforce any possible access to the object in one
    // > thread (through an existing reference) to *happen before* deleting
    // > the object in a different thread. This is achieved by a "release"
    // > operation after dropping a reference (any access to the object
    // > through this reference must obviously happened before), and an
    // > "acquire" operation before deleting the object.
    //
    // In particular, while the contents of an Arc are usually immutable, it's
    // possible to have interior writes to something like a Mutex<T>. Since a
    // Mutex is not acquired when it is deleted, we can't rely on its
    // synchronization logic to make writes in thread A visible to a destructor
    // running in thread B.
    //
    // Also note that the Acquire fence here could probably be replaced with an
    // Acquire load, which could improve performance in highly-contended
    // situations. See [2].
    //
    // [1]: (www.boost.org/doc/libs/1_55_0/doc/html/atomic/usage_examples.html)
    // [2]: (https://github.com/rust-lang/rust/pull/41714)
    atomic::fence(Acquire);

    unsafe {
        self.drop_slow();
    }
}

考虑一下这段代码：

deletion of the data. Because it is marked Release, the decreasing of the reference count synchronizes with this Acquire fence. This means that use of the data happens before decreasing the reference count, which happens before this fence, which happens before the deletion of the data.

1
atomic::fence(Acquire);
  1. self.inner().strong.fetch_sub(1, Release) 本身是原子的
  2. atomic::fence(Acquire); 保证 self.drop_slow 不会被重排序到 fetch_sub 之前

然后 drop_slow 也包含了一段这样的逻辑：

1
2
3
4
5
6
7
8
9
10
11
12
// Non-inlined part of `drop`.
#[inline(never)]
unsafe fn drop_slow(&mut self) {
    // Destroy the data at this time, even though we may not free the box
    // allocation itself (there may still be weak pointers lying around).
    ptr::drop_in_place(&mut self.ptr.as_mut().data);

    if self.inner().weak.fetch_sub(1, Release) == 1 {
        atomic::fence(Acquire);
        Global.dealloc(self.ptr.cast(), Layout::for_value(self.ptr.as_ref()))
    }
}

fetch_sub 也有同样逻辑。详细内容可以参考这个解答：

count

Arc 实现了：

1
2
3
4
5
6
7
8
9
10
11
12
13
14
  #[inline]
  #[stable(feature = "arc_counts", since = "1.15.0")]
  pub fn weak_count(this: &Self) -> usize {
      let cnt = this.inner().weak.load(SeqCst);
      // If the weak count is currently locked, the value of the
      // count was 0 just before taking the lock.
      if cnt == usize::MAX { 0 } else { cnt - 1 }
  }
  
#[inline]
  #[stable(feature = "arc_counts", since = "1.15.0")]
  pub fn strong_count(this: &Self) -> usize {
      this.inner().strong.load(SeqCst)
  }

其实可以看看 Weak 实现的

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
#[unstable(feature = "weak_counts", issue = "57977")]
pub fn weak_count(&self) -> Option<usize> {
    // Due to the implicit weak pointer added when any strong pointers are
    // around, we cannot implement `weak_count` correctly since it
    // necessarily requires accessing the strong count and weak count in an
    // unsynchronized fashion. So this version is a bit racy.
    self.inner().map(|inner| {
        let strong = inner.strong.load(SeqCst);
        let weak = inner.weak.load(SeqCst);
        if strong == 0 {
            // If the last `Arc` has *just* been dropped, it might not yet
            // have removed the implicit weak count, so the value we get
            // here might be 1 too high.
            weak
        } else {
            // As long as there's still at least 1 `Arc` around, subtract
            // the implicit weak pointer.
            // Note that the last `Arc` might get dropped between the 2
            // loads we do above, removing the implicit weak pointer. This
            // means that the value might be 1 too low here. In order to not
            // return 0 here (which would happen if we're the only weak
            // pointer), we guard against that specifically.
            cmp::max(1, weak - 1)
        }
    })
}

这里的 SeqCst 可以看看：https://stackoverflow.com/questions/12340773/how-do-memory-order-seq-cst-and-memory-order-acq-rel-differ

而 is_unique 又不同了：

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
/// Determine whether this is the unique reference (including weak refs) to
/// the underlying data.
///
/// Note that this requires locking the weak ref count.
fn is_unique(&mut self) -> bool {
    // lock the weak pointer count if we appear to be the sole weak pointer
    // holder.
    //
    // The acquire label here ensures a happens-before relationship with any
    // writes to `strong` (in particular in `Weak::upgrade`) prior to decrements
    // of the `weak` count (via `Weak::drop`, which uses release).  If the upgraded
    // weak ref was never dropped, the CAS here will fail so we do not care to synchronize.
    if self.inner().weak.compare_exchange(1, usize::MAX, Acquire, Relaxed).is_ok() {
        // This needs to be an `Acquire` to synchronize with the decrement of the `strong`
        // counter in `drop` -- the only access that happens when any but the last reference
        // is being dropped.
        let unique = self.inner().strong.load(Acquire) == 1;

        // The release write here synchronizes with a read in `downgrade`,
        // effectively preventing the above read of `strong` from happening
        // after the write.
        self.inner().weak.store(1, Release); // release the lock
        unique
    } else {
        false
    }
}