Rustの `Arc` を読む(5): Arcを読む

概要: Rustの `Arc` 型の実装は宝の宝庫です。そこで、これを隅から隅まで解説してみます。

第5回(最終回)「Arcを読む」では、 `Rc` と `Arc` の差分を読んでいきます。

- [第1回 Arc/Rcの基本](https://qiita.com/qnighy/items/4bbbb20e71cf4ae527b9)
- [第2回 Rcを読む基本編](https://qiita.com/qnighy/items/5b2fbf27e3ee36e57b8d)
- [第3回 Rcを読む発展編](https://qiita.com/qnighy/items/5e673b2d74e5b01a6c8d)
- [第4回 アトミック変数とメモリ順序](https://qiita.com/qnighy/items/b3b728adf5e4a3f1a841)
- **第5回** Arcを読む

# はじめに

第2回～第3回にかけて、 `Arc` のシングルスレッド版である `Rc` の実装を読んできました。残るは `Arc` 特有の部分です。

第1回/第2回 で説明した `Arc`/`Rc` の基本的な仕組みを仮定します。またアトミック変数とメモリ順序に言及するので、第4回の内容を適宜参照するといいでしょう。

# まとめ

- `Arc` はスレッドの休眠処理が不要でアトミック操作にしか依存しないため、直感に反して `alloc` クレートで定義されている。これは `Rc` が定義されているのと同じ場所である。
- `Arc` は共有ポイントである一方、たまたま参照カウントが1のときに所有権を取り出す機能もあるため、 `Arc<T>: Send` も `Arc<T>: Sync` も `T: Send + Sync` を要求する必要がある。
- 原則としてカウンタ操作は競合状態を防ぐためRead-Modify-Write (RMW) アトミック操作を使って行われる。
- 原則として、インクリメント時はRelaxed, デクリメント時はRelease操作として行い、カウンタが0になったときはAcquire fenceを呼ぶことで必要な同期が達成できる。
- `get_mut` を安全に実装するために、弱参照カウンタにはロック機構が実装されている。
- 公開インターフェースである `strong_count`/`weak_count` は内部処理と異なり `SeqCst` 順序が使われている。その実装は単なる内部カウンタのload/storeよりもやや興味深い実装になっている。

# ソースコード

以下の1.39.0の実装を読んでいきます。

https://github.com/rust-lang/rust/blob/1.39.0/src/liballoc/sync.rs

# 定義されている場所

少し意外かもしれませんが、 `Arc` は `Rc` と同じ `alloc` クレートに定義されています。

第3回でも述べたように、Rustの標準ライブラリは以下の3段階のクレートに分割されています。

- `core` ... CPUの基本機能のみに依存
- `alloc` ... `core` + malloc/freeに依存
- `std` ... `alloc` + OSの諸機能に依存

アトミック命令はOSの機能ではなくCPUの機能であり、基本的にどのCPUでも実装されているかエミュレート可能 (シングルコアのCPUの場合は通常の命令で代替可能) であるため、Rustでは基本的に利用可能とみなされます。ただmalloc/freeには依存するので、`Arc` は `std` ではなく `alloc` で定義されています。

なお、Mutexはスピンロックかどうかによって定義可能な場所が異なります。

- 標準ライブラリのMutexは、ロックが確保できるまでスレッドを休眠させることがあります。スレッドの休眠はOSのスケジューラーへの指示によって実現されるため、 `std` でのみ定義されます。
- `parking_lot` の定義するMutexも同様の理由で `std` のある環境でのみ利用可能です。
- `spin` の定義するMutexはスピンロックであり、スレッドの休眠をともなわないため、 `core` のある環境であればいつでも利用可能です。

# `Send`/`Sync`

第3回と同様、 [`Send`/`Sync` の実装](https://github.com/rust-lang/rust/blob/1.39.0/src/liballoc/sync.rs#L199-L202)を見てみましょう。

```rust
unsafe impl<T: ?Sized + Sync + Send> Send for Arc<T> {}
unsafe impl<T: ?Sized + Sync + Send> Sync for Arc<T> {}

unsafe impl<T: ?Sized + Sync + Send> Send for Weak<T> {}
unsafe impl<T: ?Sized + Sync + Send> Sync for Weak<T> {}
```

`Rc` と異なり、 `Arc` の実装自体はマルチスレッドを考慮して作られているので、中身のスレッド安全性に注意すればよいはずです。 `Send` は**所有権を他スレッドに送っても安全か**どうかで、 `Sync` は**共有権を他スレッドに与えても安全か**どうかでした。

- `Arc<T>` を受け取ったスレッドは中の `&T` にアクセスできます(← `T: Sync`)。また、 `try_unwrap` で `T` を取得できる可能性があります (← `T: Send`)。したがって、 `Arc<T>: Send` であるためには `T: Send + Sync` が必要です。
- `&Arc<T>` を受け取ったスレッドはこれをcloneして `Arc<T>` を得ることができます。したがって、 `Arc<T>: Sync` であるためには `T: Send + Sync` が必要です。
- `Weak<T>` を受け取ったスレッドはupgradeによって `Arc<T>` を取得できる可能性があります。したがって、 `Weak<T>: Send` であるためには `T: Send + Sync` が必要です。
- `&Weak<T>` を受け取ったスレッドはupgradeによって `Arc<T>` を取得できる可能性があります。したがって、 `Weak<T>: Sync` であるためには `T: Send + Sync` が必要です。

# 同期をとる必要があるのはどんなときか

以降ではメモリ順序指定の根拠を追っていきたいので、同期の必要性についてあらかじめ確認しておきます。この節は読み飛ばして後で必要に応じて戻ってくるのがいいかもしれません。

まず、Rustのメモリモデル (C++20のメモリモデル) では**アトミック操作はRelaxedでも各メモリ領域ごとには一貫した順序で行われる**ので、同期が必要なのは**複数のメモリ領域がかかわった場合**のみです。より具体的に言ってしまえば**strongとvalueの関係**、**weakとヒープ領域の関係**を保証したいときに同期が必要です。それは具体的には以下のケースです。

- valueへの読み取りアクセスは、valueのdrop処理とhappens before関係にある必要がある。そのためには……
  - **strongを減らす操作**をするときは、valueへのアクセスを終えた後にRelease同期を行う必要がある。
  - **strongを1以上から0にする操作**をするときは、dropを開始する前にAcquire同期を行う必要がある。
- ヒープ領域への全てのアクセスは、ヒープ領域の解放処理とhappens before関係にある必要がある。そのためには……
  - **weakを減らす操作**をするときは、ヒープ領域へのアクセスを終えた後にRelease同期を行う必要がある。
  - **weakを1以上から0にする操作**をするときは、ヒープ領域の解放を開始する前にAcquire同期を行う必要がある。
- valueへの読み取りアクセスは、後続するvalueへの書き込みアクセスとhappens before関係にある必要がある。そのためには……
  - **strongを減らす操作**をするときは、valueへのアクセスを終えた後にRelease同期を行う必要がある。
  - 書き込みアクセスのために**strongが1であることを確認する操作**をするときは、valueへの書き込みアクセスを開始する前にAcquire同期を行う必要がある。
- valueへの書き込みアクセスは、後続するvalueへの読み取りアクセスとhappens before関係にある必要がある。書き込みアクセスを終えた時点で専有状態が保証されているので、これは自然に成立する。

# 強参照のライフサイクル

第2回の流れに合わせて、まずは強参照のライフサイクルを追っていきます。

## 初期化

初期化時はまだシングルスレッドなので、特に興味深いことはありません。

## `Arc::clone`

https://github.com/rust-lang/rust/blob/1.39.0/src/liballoc/sync.rs#L902-L950

```rust
impl<T: ?Sized> Clone for Arc<T> {
    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)
    }
}

/// A soft limit on the amount of references that may be made to an `Arc`.
///
/// Going above this limit will abort your program (although not
/// necessarily) at _exactly_ `MAX_REFCOUNT + 1` references.
const MAX_REFCOUNT: usize = (isize::MAX) as usize;
```

まず、カウンタをインクリメントするために RMW (Read-Modify-Write) アトミック操作である `fetch_add` を使っています。これは第4回の冒頭で説明した例にそっくりですね。ここのReadとWriteを別々に行ってしまうと以下の状況が発生してしまいます。

- スレッド1がカウンタを読む。
- スレッド2がカウンタを読む。
- スレッド1がカウンタに書き込む。
- スレッド2がカウンタに書き込む。 (→スレッド1のインクリメント効果が消滅する)

これを防ぐには `fetch_add` で読み取りと書き込みを一気に行う必要があります。

一方メモリ順序としては最弱の `Relaxed` が指定されています。これを考えるには `Acquire` にしなくてよい理由と `Release` にしなくてよい理由をそれぞれ考えればよさそうです。

- `Acquire` をするときは、 `value` への先行する処理の終了を待ちたいときです。 `Arc::clone` が保証するのは `value` への共有アクセス権ですが、これを確保するために必要なのは「初期化が完了していること」で、これは言うまでもなく成立しています。したがってAcquire同期をする必要はありません。
- `Release` をするときは、 `value` への後続する処理を待たせたいときです。そのような処理としては「`value` への専有アクセス」「`value` のdrop」の2つがありますが、どちらに対する同期も `Arc::clone` のタイミングではなく、そうして作った `Arc` がdropされるタイミングで行う必要があり、 `Arc::clone` のタイミングでわざわざ行う必要はありません。

以上の理由から `Relaxed` で十分です。

さて、単に `fetch_add` をするだけだとカウンタがオーバーフローしてしまう可能性があります。そこでオーバーフローを防ぐ必要があります。

オーバーフローを防ぐにはCAS loopで以下のように実装する手もあります。

```rust
let mut old_counter = strong.load(Relaxed);
loop {
    if old_counter >= MAX_REFCOUNT {
        // abort
    }
    let c = strong.compare_and_swap(old_counter, old_counter + 1);
    if c == old_counter {
        break;
    } else {
        old_counter = c;
    }
}
```

しかし、このように実装すると余計な命令をたくさん発行する必要が出てしまいます。そこでRustでは `fetch_add` を使うかわりに、 `MAX_REFCOUNT` を以下のように**カウンタの最大値の約半分**に設定しています:

```rust
// usizeの最大値の約半分!
const MAX_REFCOUNT: usize = (isize::MAX) as usize;
```

その上で、以下のように「`fetch_add` した後に、カウンタ溢れをチェック」しています。

```rust
let old_size = self.inner().strong.fetch_add(1, Relaxed);

if old_size > MAX_REFCOUNT {
    unsafe {
        abort();
    }
}
```

この手法自体は競合状態があり、 `fetch_add` ～ `if` 間で留まるスレッドが大量にあった場合に `MAX_REFCOUNT` よりも少しばかり大きい値が観測される可能性があります。しかし、 `MAX_REFCOUNT` は `usize` の半分なので、カウンタが本当にオーバーフローするには**異常な量のスレッドが同時にこの位置に留まる**必要があります。そんなにスレッドがあったらそもそもメモリなり何なりが足りなくなること必至なので、**事実上発生しない**と考えて問題ないわけです。

## `Arc::drop`

https://github.com/rust-lang/rust/blob/1.39.0/src/liballoc/sync.rs#L1156-L1228

```rust
unsafe impl<#[may_dangle] T: ?Sized> Drop for Arc<T> {
    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();
        }
    }
}
```

`fetch_sub` でアトミックにデクリメントしています。アトミックでなければいけない理由はいつも通りです。

ここでは以下のように同期を行っています。

- `fetch_sub` 自身は `Release` 同期を行う。
- `fetch_sub` の結果(直前の値)が1だった場合はさらに `Acquire` 同期を行う。

これは `value` の読み取りと `value` のdropの間にhappens before関係を成立させるためです。 `value` の読み取りは `fetch_sub` の直前まで行われ、 `value` のdropは `fetch_sub` によって値が0になったあとに行われるので、この2つの `fetch_sub` の間で同期を取ればよさそうです。

具体例で考えてみます。スレッド1, 2, 3がそれぞれ `Arc` を1個ずつdropしたとします。そして、 `strong` のmodification orderが「スレッド3の `fetch_sub`」→「スレッド2の `fetch_sub`」→「スレッド1の `fetch_sub`」だったとします。このとき以下の操作があらわれます。

- スレッド1
  - `value` に読み取りアクセスする。
  - `strong.fetch_sub(1)` ... 1から0になった。
  - Acquireフェンス。
  - `value` のdropを開始する。
- スレッド2
  - `value` に読み取りアクセスする。
  - `strong.fetch_sub(1)` ... 2から1になった。
- スレッド3
  - `value` に読み取りアクセスする。
  - `strong.fetch_sub(1)` ... 3から2になった。

第4回で導入した記法を用いると、以下のようなグラフになります。

![図: `Arc::drop` の操作からなるグラフ](https://qiita-image-store.s3.ap-northeast-1.amazonaws.com/0/31409/6489c108-f1c0-07a5-e400-7f7040bde13a.png)

- スレッド1-スレッド1は同じスレッドなので(sequenced beforeなので) happens beforeが成立します。
- スレッド2-スレッド1はmemory-fence同期によりhappens beforeが成立します。
- スレッド3-スレッド1もmemory-fetch同期によりhappens beforeが成立します。このときrelease sequenceが使われていることに注意してください。

fenceを使うことで、特定の条件のときだけ同期を成立させていることも注目に値します。

# 弱参照のライフサイクル

弱参照のライフサイクルも同様です。と言いたいところですが実は1点重要な部分があります。

## `get_mut` 問題

`get_mut` は、 `&mut Arc<T>` から `&mut T` の取得を試みる関数でした。この関数は「`strong == 1` かつ `weak == 1`」と、**両方のカウンタに対する制約**をチェックする必要があります。これがマルチスレッド環境では問題になります。というのも、この2つのカウンタを**同時にアトミックに処理する方法がない**からです。(ごく一部のアーキテクチャではDouble CASと呼ばれる種類の命令があるほか、ワード長の倍のアトミックを使って解決することができる可能性もありますが、汎用的とはいえません)

### `strong` を先にチェックする場合

`strong` を先にチェックするとどうでしょうか。

```rust
if self.strong.load(Relaxed) != 1 {
    return None;
}
// ←競合条件
if self.weak.load(Relaxed) != 1 {
    return None;
}
fence(Acquire);
return Some(unsafe { &mut *self.value.get() });
```

この場合、strongをチェックしてからweakをチェックするまでの間に、以下が発生する可能性があります。

- 他のスレッドが持っていた `Weak<T>` が `Arc<T>` にアップグレードされる。
- さらに元となった `Weak<T>` がdropされ、弱参照カウンタが1になる。

こうすると、チェックは成功するにもかかわらず実際には他にも `Arc<T>` が存在する状態になってしまいます。

### `weak` を先にチェックする場合

では逆に `weak` を先にチェックするとどうでしょうか。

```rust
if self.weak.load(Relaxed) != 1 {
    return None;
}
// ←競合条件
if self.strong.load(Relaxed) != 1 {
    return None;
}
fence(Acquire);
return Some(unsafe { &mut *self.value.get() });
```

この場合、strongをチェックしてからweakをチェックするまでの間に、以下が発生する可能性があります。

- 他のスレッドが持っていた `Arc<T>` が `Weak<T>` にダウングレードされる。
- さらに元となった `Arc<T>` がdropされ、強参照カウンタが1になる。

こうすると、チェックは成功するにもかかわらず実際には他にも `Weak<T>` が存在する状態になってしまいます。

### `strong` をチェックして変更する場合

ReadではなくRMW操作を使えば競合状態を回避できるでしょうか。 `strong` を水増ししても仕方ないので、1だったら0にするのが良さそうです。

```rust
if self.strong.compare_and_swap(1, 0, Relaxed) != 1 {
    return None;
}
if self.weak.load(Relaxed) != 1 {
    self.strong.fetch_add(1, Relaxed);
    return None;
}
self.strong.fetch_add(1, Acquire);
return Some(unsafe { &mut *self.value.get() });
```

こうすると、競合条件によってチェックをすり抜けること自体はありません。 `strong` が1だったら一時的に0にしているので、その間に試みられたアップグレードは全て失敗します。

しかしこれは**ユーザーから観測される挙動**に問題があります。`Arc<T>` がまだ存在していてもアップグレードが気まぐれで失敗することになりますし、あるときアップグレードに失敗しても、その先ずっとアップグレードに失敗するとは限らないことになります。他のスレッドから見ると強参照が0なだけなので、「`get_mut` の最中」と「本当に `Arc<T>` がなくなった状態」の区別はつきません。そのため、リトライによって「気まぐれな失敗」を隠蔽することもできません。

### `weak` をチェックして変更する場合

`weak` をいじっても、ダウングレードはいつでも成功するので、競合状態を防ぐことはできません。

### ロックによる解決

以上のように、 `get_mut` 自身を工夫するだけでは安全にポインタの唯一性をチェックすることはできません。そこで `weak` の定義を見直し、 **`weak` のロック**を可能にします。このことは [`ArcInner` の定義](https://github.com/rust-lang/rust/blob/1.39.0/src/liballoc/sync.rs#L273-L282)に説明があります。

```rust
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,
}
```

**`usize` の最大値をロックされた状態として特別に扱うこと**になっています。そのため、 **weakを扱う処理はロック状態を念頭に置いて書かれる必要があります**。

## `Arc::downgrade`

さっそく `Arc::downgrade` から見ていきましょう。 https://github.com/rust-lang/rust/blob/1.39.0/src/liballoc/sync.rs#L600-L629

```rust
impl<T: ?Sized> Arc<T> {
    pub fn downgrade(this: &Self) -> Weak<T> {
        // This Relaxed is OK because we're checking the value in the CAS
        // below.
        let mut cur = this.inner().weak.load(Relaxed);

        loop {
            // check if the weak counter is currently "locked"; if so, spin.
            if cur == usize::MAX {
                cur = this.inner().weak.load(Relaxed);
                continue;
            }

            // NOTE: this code currently ignores the possibility of overflow
            // into usize::MAX; in general both Rc and Arc need to be adjusted
            // to deal with overflow.

            // Unlike with Clone(), we need this to be an Acquire read to
            // synchronize with the write coming from `is_unique`, so that the
            // events prior to that write happen before this read.
            match this.inner().weak.compare_exchange_weak(cur, cur + 1, Acquire, Relaxed) {
                Ok(_) => {
                    // Make sure we do not create a dangling Weak
                    debug_assert!(!is_dangling(this.ptr));
                    return Weak { ptr: this.ptr };
                }
                Err(old) => cur = old,
            }
        }
    }
}
```

ロックに対応するため、CAS loopとして実装されています。ロックされている場合にリトライすることを除くと以下と同等です。

```rust
this.inner().weak.fetch_add(1, Acquire);
// Make sure we do not create a dangling Weak
debug_assert!(!is_dangling(this.ptr));
return Weak { ptr: this.ptr };
```

ここで `Relaxed` ではなく `Acquire` が使われている理由は謎です。 `is_unique` がどうのと書いてありますが、 `is_unique` は同期に対して透過的に振る舞うはずです。そもそも同期が必要なのは他のメモリ領域 (`strong`, `value`) との順序を強制する必要がある場合ですが、ここでは `weak` をインクリメントするにあたって `strong`, `value` が特定の状態であることを保証する必要がありません。 (特に、すでに `strong > 0` であることは既存の `Arc` の存在から保証されており、ここで別途同期する必要はない)

NOTEに記載の通りオーバーフローがチェックされていないので、 `downgrade`→`forget` を繰り返し行うと問題が起きそうです。

## `Weak::clone`

`Weak::clone` は `Arc::downgrade` と近い処理をしていますが、実装はよりシンプルです。 https://github.com/rust-lang/rust/blob/1.39.0/src/liballoc/sync.rs#L1599-L1634

```rust
impl<T: ?Sized> Clone for Weak<T> {
    fn clone(&self) -> Weak<T> {
        let inner = if let Some(inner) = self.inner() {
            inner
        } else {
            return Weak { ptr: self.ptr };
        };
        // See comments in Arc::clone() for why this is relaxed.  This can use a
        // fetch_add (ignoring the lock) because the weak count is only locked
        // where are *no other* weak pointers in existence. (So we can't be
        // running this code in that case).
        let old_size = inner.weak.fetch_add(1, Relaxed);

        // See comments in Arc::clone() for why we do this (for mem::forget).
        if old_size > MAX_REFCOUNT {
            unsafe {
                abort();
            }
        }

        return Weak { ptr: self.ptr };
    }
}
```

最初の `if let` 分岐は第3回で説明済みなので省略します。

`Arc::downgrade` と異なり、ロックされていないことは保証されています。**`weak` カウンタのロックは `Weak` が存在しないときにしかかからない**からです。それ以外のポイントは `Arc::clone` のところで説明済みでほとんど考えることはありません。

## `Weak::upgrade`

`Weak` の存在意義といっても過言ではない `Weak::upgrade` の実装は以下のようになっています。 https://github.com/rust-lang/rust/blob/1.39.0/src/liballoc/sync.rs#L1429-L1485

```rust
impl<T: ?Sized> Weak<T> {
    pub fn upgrade(&self) -> Option<Arc<T>> {
        // We use a CAS loop to increment the strong count instead of a
        // fetch_add because once the count hits 0 it must never be above 0.
        let inner = self.inner()?;

        // Relaxed load because any write of 0 that we can observe
        // leaves the field in a permanently zero state (so a
        // "stale" read of 0 is fine), and any other value is
        // confirmed via the CAS below.
        let mut n = inner.strong.load(Relaxed);

        loop {
            if n == 0 {
                return None;
            }

            // See comments in `Arc::clone` for why we do this (for `mem::forget`).
            if n > MAX_REFCOUNT {
                unsafe {
                    abort();
                }
            }

            // Relaxed is valid for the same reason it is on Arc's Clone impl
            match inner.strong.compare_exchange_weak(n, n + 1, Relaxed, Relaxed) {
                Ok(_) => return Some(Arc::from_inner(self.ptr)), // null checked above
                Err(old) => n = old,
            }
        }
    }
}
```

ここでもCAS loopが使われていますが、これはweakのロックとは別の理由です。**強参照カウンタが0かどうかで分岐**する必要があるため、 `fetch_add` ではなくCAS loopを使う必要があります。それ以外は `Arc::clone` と同じです。

## `Weak::drop`

`Weak::drop` https://github.com/rust-lang/rust/blob/1.39.0/src/liballoc/sync.rs#L1658-L1706 も `Arc::drop` と同様です。

```rust
impl<T: ?Sized> Drop for Weak<T> {
    fn drop(&mut self) {
        // If we find out that we were the last weak pointer, then its time to
        // deallocate the data entirely. See the discussion in Arc::drop() about
        // the memory orderings
        //
        // It's not necessary to check for the locked state here, because the
        // weak count can only be locked if there was precisely one weak ref,
        // meaning that drop could only subsequently run ON that remaining weak
        // ref, which can only happen after the lock is released.
        let inner = if let Some(inner) = self.inner() {
            inner
        } else {
            return
        };

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

弱参照カウンタのロックは `Weak` がないときにしか起こらないので、ロック対策をする必要はありません。残りの挙動は `Arc::drop` と相似で、「強参照カウンタではなく弱参照カウンタが対象であること」「`value` ではなくヒープ領域が対象であること」だけが異なります。

# コピーオンライトのための処理

`Rc` と同様、 `Arc` にもコピーオンライトのための3つの関数 `try_unwrap`, `get_mut`, `make_mut` が存在します。

## `try_unwrap`

`try_unwrap` は `Arc::drop` と似ています。 https://github.com/rust-lang/rust/blob/1.39.0/src/liballoc/sync.rs#L348-L388

```rust
impl<T> Arc<T> {
    pub fn try_unwrap(this: Self) -> Result<T, Self> {
        // See `drop` for why all these atomics are like this
        if this.inner().strong.compare_exchange(1, 0, Release, Relaxed).is_err() {
            return Err(this);
        }

        atomic::fence(Acquire);

        unsafe {
            let elem = ptr::read(&this.ptr.as_ref().data);

            // Make a weak pointer to clean up the implicit strong-weak reference
            let _weak = Weak { ptr: this.ptr };
            mem::forget(this);

            Ok(elem)
        }
    }
}
```

`Arc::try_unwrap` と `Arc::drop` の違いは以下の2点だけです。

- 強参照カウントが1のときだけ動作する。
- `T` をdropせずに、所有権を取り出して返す。

この前者の条件を実現するために、 `strong.fetch_sub(1, Release)` のかわりに `strong.compare_exchange(1, 0, Release, Relaxed)` を使っています。

ところで、Release順序を保つ必要はあるのでしょうか？ `Arc::drop` がデクリメントをReleaseで行っているのは、 `value` への読み取りアクセスの終了を実際に `value` をdropするスレッドに同期するためでした。 `Arc::try_unwrap` では、デクリメントに成功したときは必ずそのスレッドで `value` の所有権を取り出すので、自分自身がデクリメントをReleaseで行う必要はなさそうです。わざわざ危険を冒してまでこの部分のメモリ順序を弱めるほどではないという判断かもしれません。

なお、Acquireフェンスのほうは相変らず必要です。他のスレッドの `Arc::drop` と同期する必要があるからです。

## `get_mut`

すでに説明したように `get_mut` の実装はややタフで、以下のようになっています。 https://github.com/rust-lang/rust/blob/1.39.0/src/liballoc/sync.rs#L1077-L1092 https://github.com/rust-lang/rust/blob/1.39.0/src/liballoc/sync.rs#L1131-L1153

```rust
impl<T: ?Sized> Arc<T> {
    pub fn get_mut(this: &mut Self) -> Option<&mut T> {
        if this.is_unique() {
            // This unsafety is ok because we're guaranteed that the pointer
            // returned is the *only* pointer that will ever be returned to T. Our
            // reference count is guaranteed to be 1 at this point, and we required
            // the Arc itself to be `mut`, so we're returning the only possible
            // reference to the inner data.
            unsafe {
                Some(Arc::get_mut_unchecked(this))
            }
        } else {
            None
        }
    }

    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
        }
    }
}
```

すでに説明したように `get_mut` が必要とする唯一性 (`strong == 1 && weak == 1`) をチェックするには弱参照カウントをロックする必要があります。この「ロック」の処理だけ切り出すと以下のようになります。

```rust
if weak.compare_exchange(1, usize::MAX, Acquire, Relaxed).is_ok() {
    // ...

    weak.store(1, Release);
}
```

これは典型的なスピンロックにおける `try_lock`～`unlock` のパターンになっていますね。

さて、このロックはアンロック時に `weak` にRMWではない単純書き込みを行っています。つまり、この操作によってそこまで続いてきたrelease sequenceが途切れることになります。これは問題ないのかという疑問が生じます。結論から言うと、その単純書き込み自身が `Release` 操作になっていること(+ロックがAcquireであること)で同期を肩代わりできるため、問題ありません。

そしてそのロック中では強参照カウンタをacquire loadしています。

さて、問題は `is_unique` がtrueを返したときに、ほかに `Arc` も `Weak` もないと確実に言えるかどうかです。これは以下のように順番に推論することでわかります。

- まず、weakのmodification orderを考えます。weakのロックからアンロックまでの間にweakが変更されることはないので、 `Weak` の生成/消滅に由来するweakのインクリメント/デクリメントは「ロック以前」「アンロック以後」に分類できます。ロック時に `weak == 1` を観測しているので、ロック以前に行われたインクリメントは、対応するデクリメントもロック以前に行われていることが保証されます。デクリメントは常にReleaseで行われているので、ロックと(直接的または間接的に)同期されます。この保証はアンロックが起きるタイミングまで延長できます。
- 次に、strongのmodification orderを考えます。ロックはありませんが、 `Arc` の生成/消滅に由来するstrongのインクリメント/デクリメントは「`is_unique` 内のロード以前」「`is_unique` 内のロード以後」に分類できます。 `is_unique` 内のロードで `strong == 1` を観測しているので、このタイミング以前に行われた(他の`Arc`のための)インクリメントは、対応するデクリメントもロード以前に行われていることが保証されます。デクリメントは常にReleaseで行われているので、当該ロードと同期されます。
- `is_unique` 内のロードは `weak` のロック～アンロックの間に発生しているので、このタイミングでは「この時点で存在しうる全ての他の `Arc`/`Weak` がdrop済み(正確には、dropは終わっていないかもしれないが、もはやヒープ領域へのアクセスはしない)」であることが保証されます。
- 他スレッドに `Arc`/`Weak` がないので、それ以降は明示的に何かしない限りはこの状況が続くことが保証されます。

ひとたびユニーク性が保証できたら、あとは `Rc::get_mut` の場合と同じ理屈で考えることができます。

## `make_mut`

`make_mut` は `try_unwrap` と `get_mut` の合わせ技的に振る舞うので、ロックは不要です。 https://github.com/rust-lang/rust/blob/1.39.0/src/liballoc/sync.rs#L995-L1047

```rust
impl<T: Clone> Arc<T> {
    pub fn make_mut(this: &mut Self) -> &mut T {
        // Note that we hold both a strong reference and a weak reference.
        // Thus, releasing our strong reference only will not, by itself, cause
        // the memory to be deallocated.
        //
        // Use Acquire to ensure that we see any writes to `weak` that happen
        // before release writes (i.e., decrements) to `strong`. Since we hold a
        // weak count, there's no chance the ArcInner itself could be
        // deallocated.
        if this.inner().strong.compare_exchange(1, 0, Acquire, Relaxed).is_err() {
            // Another strong pointer exists; clone
            *this = Arc::new((**this).clone());
        } else if this.inner().weak.load(Relaxed) != 1 {
            // Relaxed suffices in the above because this is fundamentally an
            // optimization: we are always racing with weak pointers being
            // dropped. Worst case, we end up allocated a new Arc unnecessarily.

            // We removed the last strong ref, but there are additional weak
            // refs remaining. We'll move the contents to a new Arc, and
            // invalidate the other weak refs.

            // Note that it is not possible for the read of `weak` to yield
            // usize::MAX (i.e., locked), since the weak count can only be
            // locked by a thread with a strong reference.

            // Materialize our own implicit weak pointer, so that it can clean
            // up the ArcInner as needed.
            let weak = Weak { ptr: this.ptr };

            // mark the data itself as already deallocated
            unsafe {
                // there is no data race in the implicit write caused by `read`
                // here (due to zeroing) because data is no longer accessed by
                // other threads (due to there being no more strong refs at this
                // point).
                let mut swap = Arc::new(ptr::read(&weak.ptr.as_ref().data));
                mem::swap(this, &mut swap);
                mem::forget(swap);
            }
        } else {
            // We were the sole reference of either kind; bump back up the
            // strong ref count.
            this.inner().strong.store(1, Release);
        }

        // As with `get_mut()`, the unsafety is ok because our reference was
        // either unique to begin with, or became one upon cloning the contents.
        unsafe {
            &mut this.ptr.as_mut().data
        }
    }
}
```

`make_mut` はまず `strong.compare_exchange(1, 0, Acquire, Relaxed)` を行うことで、 `try_unwrap` できる状況かどうかを調べています。できなかったら諦めてcloneします。

↑のCASが成功したら `try_unwrap` が可能ですが、 `get_mut` が可能ならそちらにフォールバックします。`get_mut` と異なり、**一度0を入れた強参照に1を書き戻すのは参照がユニークだったときだけ**なので、このように書いてもユーザーが観測する挙動としておかしなことにはなりません。

# `strong_count`/`weak_count`

`Arc`/`Rc` は強参照カウントや弱参照カウントを取得する関数を露出しています。

```rust
impl<T: ?Sized> Arc<T> {
    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 }
    }

    pub fn strong_count(this: &Self) -> usize {
        this.inner().strong.load(SeqCst)
    }
}

impl<T: ?Sized> Weak<T> {
    #[unstable(feature = "weak_counts", issue = "57977")]
    pub fn strong_count(&self) -> usize {
        if let Some(inner) = self.inner() {
            inner.strong.load(SeqCst)
        } else {
            0
        }
    }

    #[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)
            }
        })
    }
}
```

興味深いのは以下の4点です。

- メモリ順序として `SeqCst` を使っている点。アトミック操作を直接露出する観点から、できるだけ直感的に振る舞うよう `SeqCst` を選んだのかもしれない。
- `weak` がロックされていたときの処理。これらの関数は単なる読み取りしかないので、少し時系列をずらして「ロックされる直前の値」を返すことで、ロックの解放待ちを防ぐことができている。
- `Weak::weak_count` が `Option` を返す点。ぶら下がり `Weak` は、他の `Weak` 参照との同一性という観点から少々変わった挙動をするため、特定の整数を返すよりも `None` を返すのが妥当ということかもしれない。
- `Weak::weak_count` が必ずしも正確な値を返さないこと。 `weak` は内部的には強参照が1個以上あるときは1多い値を格納する仕組みになっていたが、このせいで「`Weak` の個数」を数えるのが難しい。 `weak` と `strong` を同時に読むのは難しいからである。 `Weak::upgrade` を呼べば一貫した値を取得することも可能そうだけど、そこまでするほどでもなさそう。

# まとめ

- `Arc` はスレッドの休眠処理が不要でアトミック操作にしか依存しないため、直感に反して `alloc` クレートで定義されている。これは `Rc` が定義されているのと同じ場所である。
- `Arc` は共有ポイントである一方、たまたま参照カウントが1のときに所有権を取り出す機能もあるため、 `Arc<T>: Send` も `Arc<T>: Sync` も `T: Send + Sync` を要求する必要がある。
- 原則としてカウンタ操作は競合状態を防ぐためRead-Modify-Write (RMW) アトミック操作を使って行われる。
- 原則として、インクリメント時はRelaxed, デクリメント時はRelease操作として行い、カウンタが0になったときはAcquire fenceを呼ぶことで必要な同期が達成できる。
- `get_mut` を安全に実装するために、弱参照カウンタにはロック機構が実装されている。
- 公開インターフェースである `strong_count`/`weak_count` は内部処理と異なり `SeqCst` 順序が使われている。その実装は単なる内部カウンタのload/storeよりもやや興味深い実装になっている。

以上で「Rustの `Arc` を読む」シリーズは終わりです。比較的短いソースから、予想以上に多くのことが学べたのではないでしょうか。(筆者自身も予想を超えてたくさん執筆したため驚いています。) これが皆さんの学習の助けになればさいわいです。