Rubyで非同期スレッドをテストする方法：トリガー・収集・検証のパターン

スレッドや非同期環境は、最初は少し扱いにくいものです。やり取りを整理するための適切なメンタルモデルがなければ、思わぬ結果に悩まされ、トラブルに陥りやすくなります。さらに、適切なツールやテストパターンがないと、非同期コードのテスト自体が難しくなってしまいます。

スレッドを「人」に見立て、共有オブジェクトを「所有できる物」として考えると、マルチスレッドシステムの動作を整理しやすくなります。この記事では、具体的な例を通じて、Rubyの非同期コードのテスト方法を学んでいきます。

RailsやRack、あるいはWebブラウザをフロントエンドとするアプリケーションを使っているなら、あなたはすでに非同期環境の中にいます。Rackの#callは常に非同期で呼び出されるため、意識していなくても、すでにマルチスレッドのコンポーネントを利用している可能性が高いのです。

テストの基本：トリガー、収集、検証

非同期のコールバックAPIのテストは、「トリガー（発火）」「収集」「検証」という3つのステップのパターンに従うことで、同期処理のように扱えるようになります。各スレッドを独立した個人として捉え、オブジェクトは一度に一人だけが所有できる「物」として考えてみましょう。

ここでは、バットマンと彼の7着のスーツを例にします。街を救いに出かける直前に、洗濯中のスーツがすべてアルフレッドの手元にあるかどうかを把握することの重要性は、実践的で分かりやすい例だからです。

実例：バットケーブの洗濯日

この例では、アルフレッドがバットマンのスーツを洗濯する場面を扱います。SuitWashSchedulerは、洗濯イベントごとにコールバックを呼び出すスケジューラです。開始から1秒後に最初のコールバックを実行し、以降1秒間隔で合計7回コールバックを行います。ここでの「トリガー」は、SuitWashSchedulerの生成です。

class SuitWashScheduler
  def initialize(cnt)
    Thread.new {
      cnt.times {
        sleep(1.0)
        yield
      }
    }
  end
end

結果の収集：スレッドセーフが鍵

結果の収集では、競合状態（レースコンディション）を避けるために、スレッドセーフであることが必須です。複数のスレッド間で共有されるオブジェクトは、必ず保護しなければなりません。「保護」とは、オブジェクトの所有者を管理する仕組みのことです。所有者だけがそのオブジェクトを変更・参照できます。スーツは、戦いに使うためバットマンの元にあるか、洗濯のためアルフレッドの元にあるか、どちらか一方だけなのです。

メタファーで言えば、バットマンやアルフレッドといったスレッドは、短時間だけ所有権を持ち、その後すぐ手放すのがマナーです。通常、所有者の管理にはMutexを使用します。SuitWashSchedulerのコールバックは、カウンタをインクリメントする際に結果カウンタの所有権を持ちます。そして、SuitWashSchedulerのスレッド内で実行されるコールバックは、カウンタが目標値に達したときに「すべての結果を受け取った」ことを通知します。

まずはグローバル変数をいくつか用意するところから始めます。実際のアプリケーションでは、これらのグローバル変数はクラス属性やオブジェクト属性に置き換えられます。

$main_thread = Thread.current
$mu = Mutex.new
$count = 0
$target = 7

管理変数と所有権

$main_threadと$muは、スレッドの管理とテスト完了の待機に使用し、$targetと$countはテスト結果の追跡に使用します。これはシンプルなテストなので、結果の収集と検証も簡潔に保つ必要があります。

テストは、SuitWashSchedulerの新しいインスタンスを作成することで開始されます。イニシャライザには反復回数として$targetを渡します。この場合、洗濯が必要な7着のスーツがそれに該当します。

渡されたブロックはSuitWashSchedulerのスレッド内で実行され、各反復で$countがインクリメントされて表示されます。

先を見越して考えると、メインのテストスレッドも$countをチェックすることになるため、こちらも$countの所有権が必要になります。つまり、$countの所有権を取得する手段が求められるのです。そこで登場するのが、所有権のトークンとなるMutexインスタンス$muです。SuitWashScheduler.newに渡すブロック内では、$mu.synchronizeブロックによって、$countの更新と結果確認に必要な最小限の時間だけ所有権を取得します。

SuitWashScheduler.new($target) {
  $mu.synchronize {
    $count += 1
    puts $count
    $main_thread.wakeup if $target <= $count
  }
}

検証：すべてのスーツは洗い上がったか？

メインスレッドに戻ると、今度はテストの完了を待つ必要があります。バットマンが7着すべてのスーツの完成を待たなければならないのと同じです。ここでチェックすべき条件は2つあります。1つ目は、テストが期待通りに$countを更新すること。もう1つは、待ちきれなくなったバットマン（メインスレッド）がタイムアウトすることです。

$countが$targetに達したかどうかを確認する前に、$countの所有権を取得する必要があります。SuitWashScheduler側のブロックと同様に、$mu.synchronizeを呼び出します。

しかし、ここで疑問が生じます。メインスレッドをロックしてしまったら、SuitWashSchedulerのスレッドはどうやって$countを変更できるのでしょうか？　幸い、この問題を解決する巧妙な仕組みがあります。Mutexクラスには#sleepメソッドがあり、これは所有権を一時的に手放し、タイムアウトするか起床させられるまで待機します。タイムアウトまたはメインスレッドへの#wakeup呼び出しによって起床すると、$muは処理を続行する前に再度所有権の取得を試みます。所有権を確保できたら、結果をチェックし、テストの合否を判定できるのです。

$mu.synchronize {
  $mu.sleep($target + 1)
  if $target != $count
    puts 'FAILED'
  else
    puts 'Passed! All suits are washed and clean'
  end
}

さらに理解を深めたい方は、複数のスケジューラを作成して、Mutexがどのように$countの変更の衝突を防ぐか観察してみると、例がより面白くなります。まるで、バットマンが一部のスーツをアルフレッドに、残りをクリーニング店に依頼するようなものです。その際は、$targetの判定が期待されるyieldの合計に対して行われるよう、ロジックを調整することを忘れないでください。

まとめ

適切なメンタルモデルがあれば、スレッドや非同期環境の扱いは格段に楽になります。この記事では、スレッドを「人」に、一度に一人（一つのスレッド）しか所有できない共有オブジェクトを「物」（スーツ）に見立てました。このような抽象化で考えることで、理解と記憶が容易になると私たちは考えています。

この例が非同期処理の仕組みを覚える助けになることを願いつつ、スーツを全部洗濯に出している間にバットマンが出動せざるを得なくなるという光景だけは、あまり長く頭に残らないことを祈っています。

ちなみに、ブログ上のバットマンのメタファーに飽きた方は、ぜひお知らせください。