Mori Atsushi

Kotlin CoroutinesをUIスレッド( Dispatchers.Main )のみで使っている場合、単一のリソースを複数Coroutinesで扱った場合でも問題が発生することはほとんどありません。

一方で、バックグラウンドスレッドを含めた複数スレッド( Dispatchers.Default, Dispathcers.IO ) で単一のリソースを共有した場合は、競合が発生し予期せぬ不具合を引き起こす可能性があります。

今回はその危険性と解決方法について紹介します。

競合が起きるとき

例えば以下のように10回インクリメントするコードを考えます。

var i = 0
runBlocking {
    repeat(10) {
        launch(Dispatchers.Default) {
            i += 1
        }
    }
}
println(i)

インクリメントを行う処理は Dispatchers.Default で launch することでマルチスレッドで行います。

これを実行すると、大抵の場合以下のように出力されるでしょう。

10

正しく動いているように思います。

では次に、10万までインクリメントしてみます。

var i = 0
runBlocking {
-     repeat(10) {
+     repeat(100_000) {
        launch(Dispatchers.Default) {
            i += 1
        }
    }
}
println(i)

結果は環境によって異なると思いますが、10万より少ないケースが多いでしょう。

99622

なぜ、インクリメントが漏れているのでしょうか？

それは、複数のスレッドが同時に値を取得、更新しようとして失敗しているためです。

1. スレッド1が変数 i から値 n を取得する
2. スレッド2が変数 i から値 n を取得する
3. スレッド1が変数 i を n + 1 に更新する
4. スレッド2が変数 i を n + 1 に更新する

上記は2つのスレッドで合計2回処理が行われてるのにも関わらず、1しかインクリメントされません。

Mutexを使って解決する

上記の１つ目の解決策は、Mutex を使うものです。

これは withLock で囲ったブロックが同時に1つしか処理されないことを保証してくれます。

同時に実行されそうになった場合、後に実行を開始したほうを先に実行が開始されたものが終わるまで待機させてくれます。

var i = 0
+ val mutex = Mutex()
runBlocking {
    repeat(100_000) {
        launch(Dispatchers.Default) {
+             mutex.withLock {
                i += 1
+             }
        }
    }
}
println(i)

このように正確な結果が得られるようになりました。

100000

StateFlowを使う

StateFlow には update というスレッド安全なメソッドが生えており、以下のように利用することが出来ます。

- var i = 0
+ val i = MutableStateFlow(0)
runBlocking {
     repeat(100_000) {
        launch(Dispatchers.Default) {
-            i += 1
+            i.update { it + 1 }
        }
    }
}
- println(i)
+ println(i.value)

競合が発生した場合、update は再試行することでスレッド安全を実現しているため、update 内に副作用を持たせてはいけません。

マルチスレッドは使わない

当然ですが、マルチスレッドを使わなければ競合は発生しません。

UIスレッドは１つしか無いため、Dispatchers.Main のみを使っていれば基本的にこのような問題は起きません。

var i = 0
runBlocking {
     repeat(10) {
     repeat(100_000) {
-         launch(Dispatchers.Default) {
+         launch(Dispatchers.Main) {
            i += 1
        }
    }
}
println(i)

値を取得してから書き換えるまでに delay 等のsuspend functionを挟むと当然競合が発生するので注意してください。

var i = 0
runBlocking {
     repeat(10) {
     repeat(100_000) {
        launch(Dispatchers.Main) {
            val current = i
            delay(100)
            i = current + 1
        }
    }
}
println(i)

また、バックグラウンドスレッドを使う場合でも、リソースを単一のスレッド内のみで扱えば問題は起こりません。

runBlocking {
    launch(Dispatchers.Default) {
        var i = 0
        repeat(100_000) {
            i += 1
        }
        println(i)
    }
}

マルチスレッドで共通リソースを扱うのは難しく、どうしても考慮漏れが発生することが多々あります。

また、間違った書き方をしていても正しく動くことも多く、その間違いを認識しづらいのも問題の１つでしょう。

共有のリソースを複数スレッドから扱うのは最低限、もしくは全く行わないようにするべきだと考えています。

Kotlin Coroutinesの解説本をZennにて販売しています。より詳しく学びたい方は、こちらも合わせて確認してみて下さい。

詳解 Kotlin Coroutines [2021] | Zenn