Kotlinにおける排他制御

私はAndroidアプリエンジニアですが、アプリの開発では排他処理を自前で実装しなければならない状況は少ないように思います。ただ、重要な概念であるため振り返りのためにもまとめておきます。

# 排他制御とは

そもそも排他制御とは何で、なぜ必要なのかにさらっと触れておきます。

通常プログラミングコードでは特定のコンテキストで順次実行する処理を記述します。
シングルスレッドな環境であれば何も気にする必要はないのですが、マルチスレッド環境では同時に別の処理が実行されているため、シングルスレッドでは発生し得ない問題に対処する必要があります。

例えば、特定の変数をインクリメントするという非常にシンプルな処理であっても

```kotlin
private var count: Int = 0

fun count() {
    count++
}
```

1. 値を参照
1. 1を加算
1. 変数に代入

のような複数ステップの命令のまとまりであり、これがほぼ同時に2つのスレッドが実行した場合

1. (スレッドA) 値を参照
1. (スレッドB) 値を参照
1. (スレッドB) 1を加算
1. (スレッドB) 変数に代入
1. (スレッドA) 1を加算
1. (スレッドA) 変数に代入

のように、あるスレッドが値を参照してから代入が完了するまでの間に、別のスレッドによって値が読み取られる場合があります。上記の場合、2回実行されたにもかかわらず、1しかインクリメントされないという想定と異なる動作をしてしまいます。

他の処理が入り込む余地のないワンステップの処理（**不可分な処理**とか**アトミックな処理**と表現されることがあります）でない限り、こういった可能性を考える必要があります。不可分な処理とは何か、は環境によっても異なり、例えば値の代入であっても不可分とは限らない場合もあります。

`ConcurrentModificationException` については遭遇したことがあるという人も多いでしょう。これはコレクションのイテレート中に内容が書き換えられた場合などで発生する例外です。コレクションの処理中にその内容が書き換えられてしまうと意図した結果にならないため例外が発生するようになっています。

このように、一連の処理中に他のスレッドによる操作が行われると破綻する可能性のある部分（クリティカルセクション）を保護するため、他のスレッドによる実行を制限する制御が排他制御です。

# Object と synchronized

排他処理のコードがないと説明しにくいので、先にそちらの紹介をします。
一番シンプルな排他制御の方法は synchronized を使った方法です。

Javaでは synchronized メソッドとして実装していたものです。

```java:Java
synchronized void foo() {
    // do something
}
```

Kotlinにはメソッドに対する修飾子としてsynchronizedはありませんが、代わりにアノテーションをつけることで同等の表現が可能です。

```kotlin:Kotlin
@Synchronized
fun foo() {
    // do something
}
```

また、メソッドの中でsynchronizedセクションを作ることもできます。
Javaでは以下のように記述します。

```java:Java
void foo() {
    synchronized (this) {
        // do something
    }
}
```

Kotlinでは以下のように記述します。

```kotlin:Kotlin
fun foo() {
    synchronized(this) {
        // do something
    }
}
```

Javaでは構文であり、Kotlinではメソッドであるという違いはありますが、ほぼ同じですね。
synchronizedメソッドやアノテーションとは違って、引数を渡す必要があります。

## 排他するとはどういうことか？

synchronizedメソッドやsynchronizedセクションの内側が排他されます。
さて、ここまでぼかしていましたが、排他するとはどういうことでしょうか？

簡単に言うと、リソースに所有権を設定し、そのリソースの所有権を獲得したコンテキストだけが、そのリソースにアクセスできるように制限する制御のことです。
所有権の獲得と他のコンテキストの実行制限は不可分に行われなければならないため、通常のプログラミングコードでは実装することはできず、専用のAPIを利用する必要があります。

synchronizedの仕組みでは、コンテキストとはスレッドであり、対象となるリソースの種別を表現するために利用されるのが、synchronizedセクションで引数として渡しているオブジェクトです。synchronizedメソッドの場合はそのメソッドが所属するインスタンス（this）です。
Javaでは任意クラスのインスタンスを、排他リソースを表現する同期オブジェクトとして扱うことができます。

synchronizedメソッド/セクションは、そのリソースの所有権獲得した一つのスレッドだけが実行できます。
実行時に所有権を持っていなければ獲得しようとし、すでに別のスレッドが所有権を持っている場合は、所有権が開放され、自身が獲得できるまで実行が停止します。

:::note info
用語紹介
クリティカルセクションを守るアクセス権制御機構を **ロック(Lock)** や **ミューテックス(Mutex)** と呼びます。これらはクリティカルセクションを実行できるスレッドを一つに制限する機能を持っています。また、リソースへのアクセスを管理する機構として **セマフォ(Semaphore)** というものもあり、こちらは内部カウンタの数だけリソースへのアクセスを許可するものです。カウンタが1のセマフォは **バイナリセマフォ(Binary Semaphore)** とも呼ばれ、ロックやミューテックスと同等の機能になります。
:::

この排他制御の仕組みは複雑なので理解が曖昧な人も多いでしょう。いくつか例題を作ってみました。

### 例題1

以下のようなsynchronizedメソッドがあったとして、

```kotlin:Kotlin
class Foo {
    @Synchronized
    fun foo() {
        // do something
    }
}
```

Fooのインスタンスを2つ作って2つのスレッドからコールしたとき、この2つは排他されるでしょうか？

```kotlin:Kotlin
val foo1 = Foo()
val foo2 = Foo()
thread { foo1.foo() }
thread { foo2.foo() }
```

---

**答え：排他されない**

synchronizedメソッドの同期オブジェクトはそのインスタンスになりますので、この2つは排他されません。同時実行される可能性があります。

### 例題2

synchronizedメソッドを2つ持つクラスを考えます。

```kotlin:Kotlin
class Foo {
    @Synchronized
    fun foo1() {
        // do something
    }

    @Synchronized
    fun foo2() {
        // do something
    }
}
```

この2つのメソッドを2つのスレッドからコールしたとき、この2つは排他されるでしょうか？

```kotlin:Kotlin
val foo = Foo()
thread { foo.foo1() }
thread { foo.foo2() }
```

---

**答え：排他される**

メソッドは異なりますが、同期オブジェクトは共通ですので、排他されます。いずれかのメソッドの実行中はどちらのメソッドも他のスレッドから実行することはできません。

### 例題3

以下のように、synchronizedメソッドからsynchronizedメソッドを呼び出そうとした場合、どうなるでしょうか？
そのまま実行できるでしょうか？それとも、獲得済みのリソースを再獲得しようとして停止するでしょうか？

```kotlin:Kotlin
@Synchronized
fun foo1() {
    foo2()
}

@Synchronized
fun foo2() {
    // do something
}
```

---

**答え：実行される**

synchronizedによる排他は再帰的獲得可能であるため実行されます。

## 再帰的獲得可能

synchronizedによる排他ロックは再帰的獲得可能です。
所有権を獲得した状態であれば、そのまま同一の所有権が必要なセクションを呼び出すことが可能であるということです。そのため同一オブジェクトが対象であれば、synchronizedメソッドからsynchronizedメソッドを呼び出しても問題無く動作します。当然内側のセクションを抜けたときに所有権が開放されることも無く、一番外側のセクションを抜けたときに開放されるように制御されます。

注意してほしいのは、排他の仕組みすべてが再帰的獲得可能というわけではない点です。あくまでJava/Kotlinのsynchronizedでは再帰的獲得可能というだけです。
同様の使い方をした場合に、再度そのリソースの所有権を取りにいき、すでに自分が持っているため取得できず、開放されるまで実行が停止するが、自分が停止してしまっているので解放されることはない。と、デッドロック状態に陥るものもあります。

:::note warn
再帰的獲得可能であれば自己デッドロックを防ぐことはできますが、デッドロックが一切発生しないという意味ではありません。
以下の例で、スレッドAがfoo1をスレッドBがfoo2を同時に実行した場合、スレッドAがlock1を獲得、スレッドBがlock2を獲得し、スレッドAはlock1を持ったまま、lock2が開放されるのを待ち、スレッドBはlock2を持ったまま、lock1が開放されるのを待つことになります。これはいつまでも開放されないデッドロックとなります。
```kotlin:Kotlin
fun foo1() {
    synchronized (lock1) {
        // do something
        synchronized (lock2) {
            // do something
        }
    }
}

fun foo2() {
    synchronized (lock2) {
        // do something
        synchronized (lock1) {
            // do something
        }
    }
}
```
:::

## 条件待ちと再開通知

排他機構と切り離せない、スレッド間の調停機能として、条件待ちと再開通知があります。
例えばメッセージキューのようなものを考えた場合、キューへの書き込み、キューからの読み出しは排他が必要です。加えて、キューが空の場合に失敗させるのでは無く、データが書き込まれるまで待たせたい、というケースがあります。これは前述のsychronizedの仕組みだけでは実現できません。
JavaのObject型にはこのためのメソッドが用意されています。Kotlinでは任意のクラスはObject型のサブクラスではなくなっていますが、任意のインスタンスをObject型にキャストすることで呼び出すことができますし、同期用にObject型のインスタンスを別途用意しても良いでしょう。

メッセージキューが必要であれば既存のクラスを使うべきですが、題材として簡単に実装してみます。

```kotlin:Kotlin
class MessageQueue {
    private val lock = Object()
    private val deque = ArrayDeque<String>()

    fun receive(): String {
        synchronized(lock) {
            while (deque.isEmpty()) {
                lock.wait()
            }
            return deque.removeFirst()
        }
    }

    fun post(message: String) {
        synchronized(lock) {
            deque.addLast(message)
            lock.notify()
        }
    }
}
```

この例では、別途Objectのインスタンスを作成しています。

`receive()`では、synchronizedで排他しておき、データがない場合は通知されるのを待つために、`wait()`をコールしています。`wait()`をコールするとスレッドは待ち状態となると同時にリソースの所有権を手放します。
`post()`では、synchronizedで排他しておき、データを追加します。その後、`notify()`をコールして、`wait()`で待っているスレッドがいれば通知して再開させます。

`notify()`は待ち状態のスレッド一つを再開させます。`notify()`を受けると`wait()`で待っていたスレッドは、再度所有権を獲得した状態で処理が再開され、データの有無を確認した後、また空であればもう一度`wait()`、空でなければ最初のデータを読み出す。という動作を行います。

`notify()`は必ずしも最初に待っていたスレッドが再開されるわけではなく、任意のスレッドとなります。他に`notifyAll()`という待ち状態のスレッドすべてを再開させるメソッドもあります。用途に応じて使い分けます。

例からも分かるように、`wait()`/`notify()`は所有権を取得した状態で実行する必要があります。
所有権が必要である理由は、所有権の開放と不可分に実行する必要があるためです。
不可分に実行できない場合、例えば、所有権開放と待ちの間に条件が成立し、条件が成立しているにもかかわらず条件待ちを続けてしまう可能性がでてきてしまいます。

:::note info
用語紹介
排他アクセス制御機構と条件待ち/通知に使用される条件変数機能を併せ持つものを**モニタ**と呼ぶこともあります。Java/Kotlinでの任意のインスタンスは**モニタ**としての機能を持っています。
:::

### 同期オブジェクトとしてthisを使うべきか、別途Objectを用意すべきか

synchronizedメソッドは`this`を対象に排他を行いますが、前項の例のように別途Objectを用意して利用することもできます。

`this`を対象にした場合、そのインスタンスを利用しているコードでも同一リソースに対する排他処理を実装できるということが重要です。例えば`Collections`の`SynchronizedCollection`では`this`を対象に各メソッドがsynchronized指定されています。この場合、set単体で利用する場合はそのままつかう一方、一連のまとまった処理の場合は、それらをこのcollectionのインスタンスを対象にsynchronizedするという使い方ができます。

以下の例の場合、`add`は`synchronized`で囲んでいませんが、`forEach`と排他されます。

```kotlin:Kotlin
val list = Collections.synchronizedList(mutableListOf<String>())

list.add("1")

synchronized(list) {
    list.forEach { 
        // do something
    }
}
```

一方で、排他処理を外部に影響を受けず実装したい場合は、別途Objectを用意し、外部から見えないようにするのがよいです。

Kotlinの場合は`wait()`/`notify()`メソッドへのアクセスを用意するため、`this`を使う場合もキャストしたものを保持させるのが良いかもしれません。

```kotlin:Kotlin
private val lock: Object = this as Object
```

# ReentrantLock

シンプルな排他処理はsynchronizedで実装することができますが、より高度な制御が必要な場合は、`java.util.concurrent` パッケージにある`ReentrantLock`を使うことができます。
synchronized による排他よりはこちらの方を使うべきなのか？というのはよくある議題ですが、synchronized では機能が不足している場合や、極めて多数のスレッドを扱うなど、性能がシビアに求められる場面でなければ synchronized で良いでしょう。

基本的な使い方は以下のようになります。

```kotlin:Kotlin
val lock = ReentrantLock()

lock.lock()
try {
    // do something
} finally {
    lock.unlock()
}
```

synchronizedと異なり、`lock()`/`unlock()`を明示的に呼び出します。
Lockの開放条件が複雑な場合でも実装しやすい一方、開放漏れがあると致命的な不具合となってしまうため、Javaではtry〜finallyを使って確実に開放されるよう実装するのが通例でした。Kotlinでは`withLock`という拡張関数が用意されています。これを使うことで、ボイラープレートコードを減らし抜け漏れを防止することができます。

```kotlin:Kotlin
lock.withLock {
    // do something
}
```

ReentrantLockには誰が所有権を持っているかなど、状態を確認するメソッドや、`tryLock()`という所有権を獲得できない場合に停止するのでは無く、失敗で抜けるメソッドなど、より細やかに同期制御するための機能が豊富に用意されています。

ReentrantLockはsynchronizedと同様に再帰的獲得可能で、そのスレッドがlockを獲得している状態であれば、`lock()`は成功します。そして、`unlock()`を`lock()`と同じ回数呼び出したときに所有権を開放します。

## 条件待ちと再開通知

`ReentrantLock`で条件待ちを行う場合、`newCondition()`で条件変数を作成しこれを利用します。synchronizedの仕組みでは同期オブジェクトとしてのインスタンスに直接メソッドがあったため、条件は一つしか使えませんでしたが、`ReentrantLock`では複数の条件変数を扱うことができます。
例えば、synchronziedで説明したメッセージキューですが、キューの容量が決められている場合、送信時にも容量の上限に達している場合は、空きができるまで待機させることもしたくなります。この場合、待つ条件が空の場合とは異なるため、別の条件変数が必要になります。

```kotlin:Kotlin
class MessageQueue {
    private val lock = ReentrantLock()
    private val emptyCondition = lock.newCondition()
    private val fullCondition = lock.newCondition()

    private val deque = ArrayDeque<String>()
    private val maxsize = 10

    fun receive(): String {
        lock.withLock {
            while (deque.isEmpty()) {
                emptyCondition.await()
            }
            val result = deque.removeFirst()
            fullCondition.signal()
            return result
        }
    }

    fun post(message: String) {
        lock.withLock {
            while (deque.size >= maxsize) {
                fullCondition.await()
            }
            deque.addLast(message)
            emptyCondition.signal()
        }
    }
}
```

## ReentrantReadWriteLock

メッセージキューは読み出しによってキューからデータを削除するため副作用（オブジェクトの変化）がありますが、副作用のない読み出し処理も多いです。
副作用のない読み出し処理を行う場合において、高頻度に複数のスレッドから読み出しアクセスされるリソースを排他する場合は、読み込みと書き込み両方を同じように排他したのでは無駄が発生します。
書き込み処理は、他の書き込み処理も、読み出し処理も行われていない状態で実行する必要がありますが、読み出し処理は、書き込み処理とは排他する必要がありますが、読み出し処理同士の排他は不要です。

そのような場合に使える排他の仕組みとして `ReentrantReadWriteLock` というものも用意されています。`readLock()`/`writeLock()`でそれぞれ用のLockを取得し他処理を実装するのですが、Kotlinではこれ用の拡張関数も用意されており、シンプルに記述することができます

```kotlin:Kotlin
val lock = ReentrantReadWriteLock()
lock.read {
    // 読み出し処理
}
lock.write {
    // 書き込み処理
}
```

# コルーチンの排他処理

前項までで説明した排他の仕組みはスレッドの仕組みです。コルーチンではどうでしょう？
コルーチンを実行しているのはスレッドであるため、スレッドの排他の仕組みも問題無く使えます。
しかし、一部不便な面もあります。
一つは、排他処理はブロック動作であり、例えばリソースの所有権を獲得できるまでの間、スレッドが停止します。コルーチンの中断ではないため、シングルスレッドのディスパッチャでは全体が停止してしまいます。IOディスパッチャなどマルチスレッドなディスパッチャを使う必要があります。
もう一つは、クリティカルセクションは中断できない点です。コルーチンの中にクリティカルセクションを作ることはできても、そこからsuspendメソッドを呼び出すことができません。（現在のKotlinコンパイラではエラーとなります）

suspendメソッドを含む処理を排他するには別の仕組みが必要です。それが`Mutex`です。

## Mutex

`Mutex`はCoroutineで使える排他制御の仕組みで、`ReentrantLock`と同様に、`lock()`/`unlock()`をコールするか

```kotlin:Kotlin
val mutex = Mutex()

mutex.lock()
try {
    // do something
} finally {
    mutex.unlock()
}
```

`withLock`拡張関数を使って実装します。

```kotlin:Kotlin
mutex.withLock {
    // do something
}
```

`Mutex`において所有権を獲得するコンテキストはコルーチンです。
`lock`や`withLock`はsuspendメソッドであり、suspendブロックの中でコールする必要があります。
所有権が獲得できるまで処理はブロックでは無く中断します。中断なので、シングルスレッドなディスパッチャで使っても全体が止まってしまうことはありません。
ロック中にsuspendメソッドを呼び出すこともできます。中断が発生しても所有権を保持したままなので、中断を含むセクションを排他することができます。

### Mutexは再帰的獲得不可

注意すべきなのは、`Mutex`は**再帰的獲得不可**であるという点です。mutexのlockを持った状態で、同じmutexのlockを獲得しようとするとデッドロックが発生します。
ownerを引数に指定することはできますが、これは主にデバッグ用で、すでにそのownerによって獲得済みの場合、`IllegalStateException`が発生します。

## 条件待ちと再開通知

残念ながら、コルーチンには条件待ちと再開通知の仕組みはないそうです。
必要な場合は`Flow`や`Channel`を使って擬似的な仕組みをつくるか、ブロック動作にしてスレッドの排他の仕組みを利用するしかありません。

## 「コルーチン = Mutexで排他」ではない

「コルーチンの中で排他が必要なセクション = Mutexを使う」ではありません。
中断を含まないセクションはスレッドの排他の仕組みを使うことができますし、場合によっては`withContext`で実行コンテキストを限定するだけで良い場合もあります。
そこで必要とされている排他の要件は何かをよく考えて、適切な方法を選択しましょう。

# Volatile

排他の仕組みとはちょっと違いますが、複数のスレッドからアクセスする仕組みとして重要なのが `volatile` です。
Javaは変数宣言の修飾子でKotlinではアノテーションで指定します。

```java:Java
private volatile int count = 0;
```

```kotlin:Kotlin
@Volatile 
private var count = 0
```

JVMではスレッドから共通のリソースへのアクセスにキャッシュが用いられることがあります。つまり、あるスレッドで値を書き換えても、別のスレッドからはしばらくの間書き換え前の値が読み出されてしまう可能性があります。
Volatileを指定すると、値の変化が他のスレッドにも即座に反映されることが保証されます。

また、一般にプログラムコードはアウトオブオーダー実行、すなわち、そのコンテキスト内の破綻がない範囲でより高速に処理できるよう、命令の順序を入れ替えながら実行されます。この並べ替えはシングルスレッドでは問題ありませんが、複数のスレッドからアクセスされるリソースまでは考慮されていません。
volatileはメモリバリア（メモリフェンス）としても機能し、volatile変数へのアクセスの前後で命令の順序が入れ替えられないことも保証します。

注意が必要なのは、可視性と実行順序を保証するだけであり、操作のアトミック性が保証される訳ではない点です。それらが必要な場合は、排他を利用します。

volatileはある意味、高速実行のための最適化処理を抑止する機能を持っているため、volatileではない変数を扱うよりもコストは高いですが、排他の仕組みほどはコストは高くありません。
排他までは不要であるが、可視性や順序を保証したい場合にvolatileを使います。

---

排他の仕組みは非常に難しく、基本これらを使わなくて良いように実装する。必要な場合も一から作るのでは無く、あらかじめ用意されたSDKのクラスがあるならそれを利用するのがよいです。
しかし、理解していないと再現性が低い割に重篤な不具合を発生させてしまいかねませんので、しっかりと理解したいところです。

正直自分の理解も合っているか自信がないので、誤りなどあればご指摘いただけると助かります。