Rails: ActiveRecordのupdate_countersで競合状態を防ぐ（翻訳）

参考: 週刊Railsウォッチ20210818前編『カウンタキャッシュをスレッドセーフに更新する』

Rails: ActiveRecordのupdate_countersで競合状態を防ぐ（翻訳）

Railsは大規模なフレームワークであり、特定の状況に対応できる便利ツールが多数組み込まれています。本シリーズでは、Railsの大規模なコードベースに隠れている、あまり知られていないツールを紹介します。

今回紹介するのはActiveRecordのupdate_countersメソッドです。また、マルチスレッドプログラムにありがちな「競合状態」の罠と、このメソッドで競合状態を防ぐ方法についても見ていきます。

🔗 スレッドについて

プログラミングでコードを並行して実行する場合、「プロセス」「スレッド」、最近のRubyなら「Fiber」や「Ractor」といったさまざまな方法がありますが、本記事ではその中でもRails開発者が目にすることの多い「スレッド」にのみ注目します。たとえばPumaはマルチスレッドのWebサーバーですし、Sidekiqもバックグラウンドジョブをマルチスレッドで処理します。

本記事ではスレッドおよびスレッド安全性についてはこれ以上立ち入りません。知っておいていただきたいのは、2つのスレッドが同一のデータを操作すると、データの同期が取れなくなりがちだという点です。これがいわゆる「競合状態（race condition）」です。

🔗 競合状態について

競合状態は、2つ以上のスレッドが「同じデータを」「同時に」操作している場合に発生し、スレッドが使うデータが最新でなくなってしまう可能性があります。2つのスレッドが互いに競争しているように見えることから「競合状態」と呼ばれます。データの最終的な状態は、どちらのスレッドが「レースに勝った」かによって影響され、データの最終的な状態が一定しなくなる可能性があります。競合状態でおそらく最悪なのは、再現が非常に難しいという点です。理由は、競合状態はスレッドがコードの特定の場所で特定の順序で実行された場合にのみ発生することが多いためです

🔗 コード例

銀行口座の残高更新は、競合状態を説明する一般的なシナリオのひとつです。以下のように、基本的なRailsアプリケーションの中に簡単なテストクラスを作成して様子を確認します。

class UnsafeTransaction
  def self.run
    account = Account.find(1)
    account.update!(balance: 0)

    threads = []
    4.times do
      threads << Thread.new do
        balance = account.reload.balance
        account.update!(balance: balance + 100)

        balance = account.reload.balance
        account.update!(balance: balance - 100)
      end
    end

    threads.map(&:join)

    account.reload.balance
  end
end

上のUnsafeTransactionクラスは、Accountモデルを探索するメソッドが1個あるだけのかなりシンプルなつくりです（AccountはBigDecimal型のbalance属性を持つ標準的なRailsモデルです）。ここではテストの再実行をシンプルにしたいので、残高をゼロに設定しています。

内側のループでは少々興味深いことが行われています。4つのスレッドを作成して口座の現在の残高を取得し、そこに100を加えたら（例：100ドルの入金）、すぐに100を引きます（例：100ドルの出金）。念のため、残高が最新になるよう入金と出金の両方でreloadを呼び出しています。

残りの行は事後処理です。Thread.joinはすべてのスレッドが終了するのを待ってから処理を先に進めるためのものです。続いてメソッドの末尾では最終的な残高を返します。

これを1つのスレッドで実行した場合（ループを 1.times do に変更した場合）なら、100万回実行しても、最終的な勘定残高が常にゼロになることが確信できます。しかし、これを2つ以上のスレッドに変更すると確実性が落ちてしまいます。

このテストをコンソールで1回実行すると、おそらく正しい答えが得られるでしょう。

UnsafeTransaction.run
#=> 0.0

しかしこのテストを繰り返し実行するとどうなるでしょうか。たとえば以下のように10回繰り返したとします。

(1..10).map { UnsafeTransaction.run }.map(&:to_f)
#=> [0.0, 300.0, 300.0, 100.0, 100.0, 100.0, 300.0, 300.0, 100.0, 300.0]

この構文に慣れていない方向けに念のため説明すると、(1..10).map {} はブロック内のコードを10回実行し、それぞれの実行結果を配列にします。最後の .map(&:to_f) は人間が読みやすい数値にするためのもので、BigDecimalの値は通常 0.1e3 のような指数表記で表示されます。

このコードでは現在の残高に100を加えた直後に100を引いているので、最終的な結果は常に0.0になるはずだったことを思い出しましょう。この「100.0」と「300.0」のような数値のぶれは、競合状態が発生している証拠です。

🔗 コメント付きのコード例

今度は問題のコードに注目しながら観察してみましょう。わかりやすくするため、balanceの変更点を分離してみます。

threads << Thread.new do
  # ここで他のスレッドが実行されるかもしれない
  balance = account.reload.balance
  # ここかもしれない
  balance += 100
  # ここかもしれない
  account.update!(balance: balance)
  # ここかもしれない

  balance = account.reload.balance
  # ここかもしれない
  balance -= 100
  # ここかもしれない
  account.update!(balance: balance)
  # ここかもしれない
end

上のコードのコメントで示したように、このスレッドの処理中に他のスレッドが実行される可能性があります。例えば、スレッド1が残高を読み込んだ後、スレッド2が実行され、update!を呼び出す時点では既にデータが古くなっていてもおかしくないでしょう。つまり、スレッド1とスレッド2とデータベースのどれにもデータがあるにもかかわらず、それらのデータが互いに同期しなくなっているということです。

ここで取り上げた例は、簡単に分解できるようにあえて素朴な書き方にしています。しかし現実の世界では競合状態の診断は難しく、特に信頼性の高い形では再現できないことが普通です。

🔗 解決方法

競合状態を防ぐ方法はいくつかありますが、そのほとんどは「一度に1つのエンティティだけがデータを変更するようにする」という考え方に基づいています。

🔗 オプション1: ミューテックス

最もシンプルなオプションは「相互排他ロック（mutual exclusion lock）」で、一般にミューテックス（mutex）と呼ばれています。ミューテックスは「鍵が1つしかない錠」とみなせます。鍵を持っているスレッドはミューテックスにあるものなら何でも実行できますが、他のスレッドは鍵を手に入れるまで待たなければなりません。

コード例にミューテックスを適用すると次のようになります。

class MutexTransaction
  def self.run
    account = Account.find(1)
    account.update!(balance: 0)

    mutex = Mutex.new

    threads = []
    4.times do
      threads << Thread.new do
        mutex.lock
        balance = account.reload.balance
        account.update!(balance: balance + 100)
        mutex.unlock

        mutex.lock
        balance = account.reload.balance
        account.update!(balance: balance - 100)
        mutex.unlock
      end
    end

    threads.map(&:join)

    account.reload.balance
  end
end

上のコードでは、accountに読み書きするたびにまずmutex.lockを呼び出し、読み書きが終わったらmutex.unlock を呼び出して他のスレッドにも処理が回るようにしています。ブロックの冒頭でmutex.lock を呼び出し、末尾で mutex.unlockを呼び出すことも一応可能ですが、スレッドがコンカレントに実行されなくなり、スレッドを使う理由が多少損なわれてしまいます。パフォーマンスを考えれば、スレッドができるだけ多くのコードを並行に実行できるよう、mutex内のコード行数はできるだけ少なくしておくのがベストです。

今はわかりやすさのためlockとunlockを用いましたが、RubyのMutexクラスにはsynchronizeという便利メソッドもあります。処理をブロックとしてsynchronizeに渡せるので、以下のように書けます。

mutex.synchronize do
  balance = ...
  ...
end

RubyのMutexには必要な機能が備わっています。しかしおそらくご想像のとおり、Railsアプリケーションでは特定のデータベース行をロックする必要が生じることがよくあります。そしてActive Recordはそうしたシナリオに対応しています。

🔗 オプション2: Active Recordのロック機能

Active Recordには何種類かのロック機構が用意されていますが、本記事ではすべてを深掘りすることはしません。ここでは、更新したい行をロックするlock!メソッドを用いるにとどめます。

class LockedTransaction
  def self.run
    account = Account.find(1)
    account.update!(balance: 0)

    threads = []
    4.times do
      threads << Thread.new do
        Account.transaction do
          account = account.reload
          account.lock!
          account.update!(balance: account.balance + 100)
        end

        Account.transaction do
          account = account.reload
          account.lock!
          account.update!(balance: account.balance - 100)
        end
      end
    end

    threads.map(&:join)

    account.reload.balance
  end
end

ミューテックスはスレッドの特定のコードセクションを「ロック」しますが、Active Recordのlock!は特定のデータベース行をロックします。つまり、同じコードを複数のアカウントで並行して実行できること（多数のバックグラウンドジョブの中で実行できるなど）を意味し、待たされるのは同じレコードにアクセスする必要のあるスレッドだけです。

Active Recordにはトランザクションとロックを一度に行える便利な#with_lockメソッドも用意されているので、上のコードは次のようによりシンプルに書き変えられます。

account = account.reload
account.with_lock do
  account.update!(account.balance + 100)
end
...

🔗 オプション3: アトミックなメソッド

「アトミック」なメソッドや関数は、実行中に停止できません。たとえば、Rubyでよく使われる+=は単一の操作に見えますが、実はアトミックではありません。

value += 10

# 上は以下と同等
value = value + 10

# 以下の冗長なコードも同等
temp_value = value + 10
value = temp_value

このvalue + 10を評価して結果をvalueに書き戻すまでの間にスレッドが突然「スリープ」すると、競合状態に陥る可能性があります。

しかしRubyがこの操作中にスレッドのスリープを許さないという状況を想像してみましょう。操作中にスレッドが決してスリープしない（コンピュータが実行を別のスレッドに切り替えない）と言い切れるなら、これは「アトミック」な操作と言えるでしょう。

ある種の言語では、まさにこうしたスレッド安全性のためにプリミティブ値のアトミック版が用意されています（AtomicIntegerやAtomicFloatなど）。しかし、Railsでそうした「アトミックな」操作が利用できないというわけではありません。Active Recordのupdate_countersがそうした例のひとつです。

update_countersメソッドは主にカウンタキャッシュの状態を最新に保つのが目的ですが、アプリケーションで使っていけない理由はありません。カウンタキャッシュについて詳しくは、以下の私の過去記事をどうぞ。

参考: How ActiveRecord Uses Caching To Avoid Unnecessary Trips To The Database - Honeybadger Developer Blog

update_countersメソッドの利用法は驚くほどシンプルです。

class CounterTransaction
  def self.run
    account = Account.find(1)
    account.update!(balance: 0)

    threads = []
    4.times do
      threads << Thread.new do
        Account.update_counters(account.id, balance: 100)

        Account.update_counters(account.id, balance: -100)
      end
    end

    threads.map(&:join)

    account.reload.balance
  end
end

ご覧のとおり、ミューテックスもロックも使わない、たった2行のRubyコードです。

update_countersはレコードIDを第1引数にとり、次に変更するカラム(balance:)と変更内容(100や-100)を指定します。これでうまくいく理由は、データベース読み書きのサイクルが単一のSQL呼び出しで行われるからです。つまりRubyのスレッドは処理を中断できません。たとえスリープが発生しても、実際の計算はデータベースで行われているので問題ありません。

実際に生成されるSQLは以下のようになります（少なくとも私のPC上のPostgreSQLでは）。

Account Update All (1.7ms)  UPDATE "accounts" SET "balance" = COALESCE("balance", 0) + $1 WHERE "accounts"."id" = $2  [["balance", "100.0"], ["id", 1]]

この方法はパフォーマンスも大きく向上します。計算がデータベース上で行われるため、最新の値を取得するためにreloadする必要がないのですから当然です。

ただしこの方法では処理が生SQLで実行されるため、Railsのモデルがバイパスされるという代償を伴います。つまり、モデルのバリデーションやコールバックが一切実行されません（updated_atタイムスタンプも変更されなくなります）¹。

まとめ

競合状態はハイゼンバグ（Heisenbug）の申し子と言ってもよいでしょう。競合状態は発生しやすいうえに再現できないことが多く、事前に予見するのも困難です。少なくともRubyとRailsは、これらの問題が見つかったときに解決する便利なツールを提供しています。

一般的なRubyのコードならMutexライブラリがよいオプションです。ほとんどの開発者が「スレッドセーフ」という言葉でおそらく最初に思い浮かべるのがこれでしょう。

Railsの場合はデータをActive Recordから取得することの方が多く、lock!またはwith_lockならデータベースの特定行だけをロックできるので、こちらの方がミューテックスよりも簡単です。

ここで率直に申し上げますが、私が現実にupdate_countersメソッドを使うことはあまりなさそうです。update_countersの振る舞いは開発者にもそれほど知られていませんし、コードが特に明確になるというわけでもありません。スレッド安全性の問題を踏んだときはActive Recordのlock!またはwith_lockでロックする方がコードの意図がより明確に伝わります。

しかし、シンプルな「足し算や引き算」を多数実行し、かつスピードアップが必要な場合なら、update_countersを知っておくと重宝するでしょう。

関連記事

Rails向け高機能カウンタキャッシュ gem「counter_culture」README（翻訳）