並行処理の仕組みを徹底解説：Rubyで学ぶイベントループ

はじめに

本記事は「Ruby Magic」による並行処理（コンカレンシー）シリーズの最終回です。これまでの記事では、複数プロセスと複数スレッドを使ってチャットサーバーを実装してきました。今回は同じものをイベントループを使って実装します。

おさらい

前回までの記事と同じクライアントとサーバーの構成を使用します。目標は、複数のユーザーがリアルタイムにメッセージをやり取りできるチャットシステムを構築することです。

基本的なセットアップの詳細については、以前の記事をご覧ください。本記事のサンプルで使用している完全なソースコードはGitHubで公開されているので、実際に手元で動かしながら試すこともできます。

イベントループを使ったチャットサーバー

イベントループでチャットサーバーを構築するには、スレッドやプロセスを使う場合とは異なる考え方が必要です。従来のアプローチでは、1つのスレッドまたはプロセスが1つの接続を担当します。一方、イベントループでは、単一のスレッド・単一のプロセスで複数の接続を処理します。仕組みを分解して見ていきましょう。

イベントループとは

EventMachineやNode.jsなどで採用されているイベントループは、次のように動作します。まず、OSに対して「どのイベントに関心があるか」を伝えます。たとえば、ソケットへの接続が開かれたときなどです。これは、IOオブジェクト（接続やソケットなど）に対する関心を登録する関数を呼び出すことで実現します。

このIOオブジェクトで何かが起こると、OSからプログラムへイベントが送られてきます。これらのイベントはキューに入れられ、イベントループはキューからイベントを1つずつ取り出して順番に処理していきます。

ある意味で、イベントループは真の並行処理ではありません。ごく小さなバッチで逐次処理を行うことで、並行性をシミュレートしているのです。

関心の登録とIOイベントの受け渡しをOSに行わせるには、本来C拡張を書く必要があります。Ruby標準ライブラリにはそのためのAPIが存在しないためです。ただし、この記事ではそこまで踏み込みません。代わりにIO.selectを使ってイベントを生成します。IO.selectは監視対象のIOオブジェクトの配列を受け取り、そのうちの1つ以上が読み書き可能になるまで待機し、準備ができたIOオブジェクトだけを含む配列を返します。

接続に関するすべての処理はFiberとして実装します。以降、このコードを「ハンドラー」と呼びます。Fiberとは、一時停止と再開が可能なコードブロックのことです。Ruby VMが自動的に制御してくれるわけではないため、resumeとyieldを手動で呼び出す必要があります。IO.selectからの結果を使って、接続が読み書き可能になったタイミングでハンドラーに通知します。

以前のスレッド版・マルチプロセス版の例と同様に、クライアントと送信されたメッセージを追跡するためのストレージが必要です。ただし今回はMutexは不要です。イベントループはシングルスレッドで動作するため、オブジェクトが複数のスレッドから同時に書き換えられる心配がないからです。

client_handlers = {}
messages = []

クライアントハンドラーは次のFiberとして実装されています。ソケットが読み書き可能になるとイベントが発生し、Fiberがそれに応答します。状態が:readableの場合はソケットから1行読み取り、messages配列に追加します。状態が:writableの場合は、前回の書き込み以降に他のクライアントから受信したメッセージをクライアントへ書き込みます。イベントを処理した後はFiber.yieldを呼び出し、次のイベントまで一時停止して待機します。

def create_client_handler(nickname, socket)
  Fiber.new do
    last_write = Time.now
    loop do
      state = Fiber.yield
 
      if state == :readable
        # ソケットからメッセージを読み取る
        incoming = read_line_from(socket)
        # 問題なければ書き込みリストに追加
        $messages.push(
          :time => Time.now,
          :nickname => nickname,
          :text => incoming
        )
      elsif state == :writable
        # ソケットへメッセージを書き込む
        get_messages_to_send(last_write, nickname, $messages).each do |message|
          socket.puts "#{message[:nickname]}: #{message[:text]}"
        end
        last_write = Time.now
      end
    end
  end
end

では、ソケットの準備ができた適切なタイミングで、どうやってFiberに読み書きを促せばよいのでしょうか。ここで登場するのが、次の4つのステップからなるイベントループです。

loop do
  # ステップ1: 新しい接続を受け入れる
  accept_incoming_connections
 
  # ステップ2: 読み書き可能になった接続を取得する
  get_ready_connections
 
  # ステップ3: 読み取り可能な接続から読み込む
  read_from_readable_connections
 
  # ステップ4: 書き込み可能な接続へ書き込む
  write_to_writable_connections
end

お気づきのとおり、ここには魔法のような仕掛けはありません。ごく普通のRubyのループです。

ステップ1: 新しい接続を受け入れる

新しい着信接続があるかどうかを確認します。ここではaccept_nonblockを使用します。このメソッドはクライアントの接続を待ってブロックすることがなく、新しいクライアントがいない場合はエラーを発生させます。そのエラーを捕捉して次のステップへ進みます。新しいクライアントがあれば、そのハンドラーを作成してclientsストアに格納します。後でハンドラーを検索できるよう、ソケットオブジェクトをハッシュのキーとして使用します。

begin
  socket = server.accept_nonblock
  nickname = socket.gets.chomp
  $client_handlers[socket] = create_client_handler(nickname, socket)
  puts "Accepted connection from #{nickname}"
rescue IO::WaitReadable, Errno::EINTR
  # 現時点で新しい着信接続はない
end

ステップ2: 読み書き可能になった接続を取得する

次に、接続の準備ができたときにOSから通知してもらうよう依頼します。読み取り・書き込み・エラー処理のそれぞれについて、client_handlersストアのキーを渡します。これらのキーはステップ1で受け入れたソケットオブジェクトです。ここでは10ミリ秒待機します。

readable, writable = IO.select(
  $client_handlers.keys,
  $client_handlers.keys,
  $client_handlers.keys,
  0.01
)

ステップ3: 読み取り可能な接続から読み込む

読み取り可能な接続があれば、対応するクライアントハンドラーを:readable状態で再開（resume）します。IO.selectが返すSocketオブジェクトがハンドラーストアのキーとして使われているため、該当するハンドラーを簡単に見つけられます。

if readable
  readable.each do |ready_socket|
    # ストレージからクライアントを取得
    client = $client_handlers[ready_socket]
 
    client.resume(:readable)
  end
end

ステップ4: 書き込み可能な接続へ書き込む

書き込み可能な接続があれば、対応するクライアントハンドラーを:writable状態で再開します。

if writable
  writable.each do |ready_socket|
    # ストレージからクライアントを取得
    client = $client_handlers[ready_socket]
    next unless client
 
    client.resume(:writable)
  end
end

この4つのステップをループで回し、ハンドラーを作成しながら適切なタイミングでreadableとwritableを呼び出すことで、完全に機能するイベント駆動型のチャットサーバーが完成します。接続あたりのオーバーヘッドは非常に小さく、大量の同時クライアントにもスケールできます。

このアプローチは、ループの1回転（ティック）あたりの処理量を小さく保つ限り、非常にうまく機能します。特に計算を伴う処理では重要なポイントです。イベントループはシングルスレッドで動作するため、利用できるCPUは1つだけだからです。本番環境のシステムでは、この制限を回避するために、複数のプロセスでそれぞれイベントループを実行する構成がよく採用されます。

まとめ：どの手法を選ぶべきか

ここまで読んで、「結局どの方法を使えばいいのか？」と思った方もいるでしょう。以下が選択の目安です。

• ほとんどのアプリケーション：スレッドが最適です。最もシンプルに扱えるアプローチです。
• 高並行性かつ長時間のストリームを扱うアプリ：イベントループならスケールさせられます。
• プロセスのクラッシュが想定される場合：堅牢性を重視するなら、昔ながらのマルチプロセスが最良の選択です。

以上で並行処理シリーズは完結です。全体のおさらいとしてはオリジナルの「mastering concurrency」記事を、詳細については複数プロセス編・複数スレッド編の各記事も併せてご覧ください。