MERY のサーバーサイドエンジニアの @saidie です。

MERY では画像アップロードや記事投稿による検索インデックス作成などなど、ユーザからのリクエスト起因で起こる時間のかかる処理の多くを非同期に行うことでレスポンスタイムの向上に努めています。また、重複した非同期処理が並行して走ることによるスループットの劣化を分散ロックを用いた排他制御で緩和する取り組みなども行っています。

MERY は Ruby on Rails を用いて開発されており、非同期処理には Ruby 製のフレームワークである Sidekiq を採用しています。この記事では Sidekiq と Redis による分散ロックを使って、同一の非同期処理が(あまり)重複しないような MERY の非同期処理システムについてご紹介します。

Sidekiq

Sidekiq はクライアントサーバモデルのマルチスレッド型のジョブキューシステムで、Redis をキューとして使っています。クライアントがジョブをエンキューすると、サーバ側で動いているワーカスレッドが順番にジョブを処理していきます。

このようなジョブキューシステムでは処理のスループットが一般的な性能指標となっています。というのも、仮にクライアント側のエンキュー速度が高く、サーバ側の処理が追いつかなくなってしまうと、キューにジョブが溜まり続けてしまいます。結果として、エンキューから処理完了までの時間が長くなり UX を大きく損なったり、最悪の場合は Redis のメモリを使いきって障害に繋がる可能性もあります。そのため、事前にエンキューの頻度とジョブの性質などを考慮しておくことが重要です。

ジョブのユニーク性

ジョブキューシステムを用いた非同期処理の設計においては、ジョブの粒度を比較的小さくすることでシステムをできるだけ疎に保つ傾向があるように思われます。そのため、ワーカー同士は基本的にお互いの存在を知らずに処理を行う形になることが多いです。このような設計だと、異なるワーカーが同じ処理を同時に実行する可能性があり、重複したジョブが頻繁にエンキューされる場合などはスループットが著しく落ちることがあります。そのため、重複したジョブを処理しないような仕組み、言い換えるとジョブのユニーク性を保つ仕組みが必要なケースがあります。

ジョブのユニーク性と一言で言っても、そこにはいくつかの文脈におけるユニーク性が存在します。このあたりについては、Sidekiq でジョブのユニーク性を担保するための gem である sidekiq-unique-jobs の README である程度まとめられています。6 つほどあるので、それらを図にしてみたのが以下になります。

詳しくは上述の README を参照して欲しいのですが、簡単に説明すると、ジョブにはデキュー前の「待ち」とデキュー後の「実行」の二つの状態があり、これらの状態の組み合わせに対して異なるユニーク性が定義されています (Until timeout を除いて; エンキューから所定の時間が過ぎるまでユニーク)。それぞれ特性が異なるため、処理内容次第で適宜使い分ける必要があります。ここで重要なことは、「待ち」と「実行」状態のユニーク性の担保はそれぞれクライアントとサーバが行うという点になります。よって、前者のユニーク性担保をする場合、エンキューがブロックされる可能性が出てきます。

MERY でジョブのユニーク性を担保する仕組み

MERY では、とある特定のデータの更新が起こった時に、その更新内容を非同期で別のデータストアに反映する仕組みが必要になったことがあります。平常時は何も問題ないのですが、バルクでデータ更新をする際にジョブが重複していくつも入ることが予想されました。そのため、上図のように重複を考えずナイーブに実装をしてしまうと意味のない更新処理でワーカーが専有されスループットが落ちてしまいます。

また、単に現在処理中のジョブと同一のジョブを捨てるという実装をしてしまうと、処理中に起こった更新の内容を取りこぼしてしまうこととなります (上図の「ダメな実装」)。そのため、ジョブ実行中にそれと同一のジョブを全て捨ててしまうのではなく、一つだけ取っておく必要があります。これは前述の "Until & while executing" に相当するユニーク性となります。

MERY では sidekiq-unique-jobs とは異なる方法でこのユニーク性の担保を行っています。

まず、実行用と待機用の二つの分散ロック*1を用意し、各ワーカーは以下のような順序でロックの取得を試みます。

1. 待機用ロックの取得を試みる
   • 取得に失敗したらジョブを終了
2. 実行用ロックの取得を試みる
   • 取得に失敗したら sleep したのち複数回リトライ
     • 最大リトライ回数に達したら終了
3. 待機用ロックを解放する
4. ジョブの処理を行う
5. 実行用ロックを解放する

このようにすることで、あるジョブを実行するワーカーを一台に制限し、かつ実行中も最大一つのジョブを受け入れることができるようになります。具体的なコードはこんな感じになります。

require 'securerandom'

def lock(redis, wait_key, exec_key)
  token = SecureRandom.uuid

  waiting = redis.set(wait_key, true, nx: true, ex: 60)
  while waiting
    if redis.set(exec_key, token, nx: true, ex: 60)
      redis.del(wait_key)
      waiting = false
      yield
      break
    end
    redis.expire(wait_key, 60)
    sleep(0.1)
  end
rescue
  redis.del(wait_key) if waiting
  raise
ensure
  redis.del(exec_key) if redis.get(exec_key) == token
end

コードをある程度シンプルにしたかったので一部簡略化しており、このままでは実用にはなりませんが (失効時間がハードコードされていたり、競合状態の問題が残っていたり)、上述の Redis の分散ロックを二重で取って処理を実行 (yield にしています) するというロジックをそのまま実装したものになります。以下がこのメソッドを使ったサンプルコードになります。

require 'redis'

redis = Redis.new

threads = []
threads << Thread.new do
  sleep 3
  lock(redis, 'wait key', 'exec key') { puts 'hoge' }
end
threads << Thread.new do
  sleep 1
  lock(redis, 'wait key', 'exec key') { sleep 2; puts 'fuga' }
end
threads << Thread.new do
  sleep 2
  lock(redis, 'wait key', 'exec key') { sleep 2; puts 'piyo' }
end

threads.each(&:join)

2 番目のスレッドが最初に実行用ロックを取得し、次に 3 番目のスレッドが待機用ロックを取得し、1 番目のスレッドは待機用ロックを取得できず即座に処理が終了します。結果は

fuga
piyo

となり、期待通りの挙動を実現することができました！

そもそも sidekiq-unique-jobs gem を使わずに、あえて独自実装を行った大きな理由の一つは、MERY での実装時に "Until & while executing" がサポートされていなかったことなのですが、サポートされた現在でもこちらの方式にはサーバ側でユニーク性の担保が完結するという利点があります。そのため、クライアント側でエンキューがブロックする可能性がなく、またクライアント側にその gem を導入する必要もなくなります。

まとめ

Sidekiq と Redis による分散ロックで重複ジョブを取り除いた MERY の取り組みについての話でした。

そもそも重複ジョブが多数投入されても札束で殴る (ワーカをスケールアウトする) というモダンな手法でなんとかなる話ではあるのですが、インフラコスト削減という建前の中に Redis で遊びたいという筆者の個人的な趣味をそっと包み込んだというところもないとは言い切れないです。 ただし、パフォーマンスを追求するために一種のロック機構を組み込むというのは止むを得ないと思っていて、その時 Redis を分散ロックとして使うのは理にかなった選択だと考えています。

株式会社ペロリでは女の子のかわいいのために堅牢な分散システム基盤の設計・開発・運用をしたいエンジニアを大募集しています。 www.wantedly.com