並行性の詳細：イベントループ

並行性に関するシリーズの最後のRubyMagicの記事へようこそ。以前のエディションでは、複数のプロセスと複数のスレッドを使用してチャットサーバーを実装しました。今回は、イベントループを使用して同じことを行います。

要約

以前の記事で使用したのと同じクライアントと同じサーバー設定を使用します。私たちの目的は、次のようなチャットシステムを構築することです。

基本的な設定の詳細については、以前の記事を参照してください。この記事の例で使用されている完全なソースコードはGitHubで入手できるため、自分で試すことができます。

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

チャットサーバーにイベントループを使用するには、スレッドやプロセスを使用する場合とは異なるメンタルモデルが必要です。従来のアプローチでは、スレッドまたはプロセスが単一の接続の処理を担当します。イベントループを使用すると、複数の接続を処理する単一のプロセスに単一のスレッドがあります。分解して、これがどのように機能するかを見てみましょう。

イベントループ

たとえば、EventMachineまたはNodeJSで使用されるイベントループは次のように機能します。まず、特定のイベントに関心があるオペレーティングシステムに通知します。たとえば、ソケットへの接続が開かれたとき。これを行うには、接続やソケットなどのIOオブジェクトに関心を登録する関数を呼び出します。

このIOオブジェクトで何かが発生すると、オペレーティングシステムはプログラムにイベントを送信します。これらのイベントをキューに入れます。イベントループは、イベントをリストから外し続け、それらを1つずつ処理します。

ある意味で、イベントループは真に同時ではありません。効果をシミュレートするために、非常に小さなバッチで順番に機能します。

インタレストを登録し、オペレーティングシステムにIOイベントを渡させるには、Ruby標準ライブラリにそのためのAPIが存在しないため、C拡張機能を作成する必要があります。これについてはこの記事の範囲外なので、IO.selectを使用します。 代わりにイベントを生成します。 IO.select IOの配列を取ります 監視するオブジェクト。配列からの1つ以上のオブジェクトが読み取りまたは書き込みの準備ができるまで待機し、それらのIOのみを含む配列を返します。 オブジェクト。

接続に関連するすべてを処理するコードは、Fiberとして実装されます。 ：これからはこのコードを「ハンドラー」と呼びます。 Fiber 一時停止および再開できるコードブロックです。 Ruby VMはこれを自動的に実行しないため、手動で再開して譲歩する必要があります。 IO.selectからの入力を使用します 接続の読み取りまたは書き込みの準備ができたときにハンドラーに通知します。

以前の投稿のスレッド化されたマルチプロセスの例のように、クライアントと送信されたメッセージを追跡するためのストレージが必要です。 Mutexは必要ありません この時。イベントループは単一のスレッドで実行されているため、異なるスレッドによってオブジェクトが同時に変更されるリスクはありません。

client_handlers = {}
messages = []

クライアントハンドラーは、次のFiberに実装されています。 。ソケットの読み取りまたは書き込みが可能になると、Fiberのイベントがトリガーされます。 応答します。状態が:readableの場合 ソケットから行を読み取り、これを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
        # Read a message from the socket
        incoming = read_line_from(socket)
        # All good, add it to the list to write
        $messages.push(
          :time => Time.now,
          :nickname => nickname,
          :text => incoming
        )
      elsif state == :writable
        # Write messages to the socket
        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をトリガーするのでしょうか。 Socketが適切なタイミングで読み取りまたは書き込みを行う 準備はいいですか？ 4つのステップを持つイベントループを使用します：

loop do
  # Step 1: Accept incoming connections
  accept_incoming_connections
 
  # Step 2: Get connections that are ready for reading or writing
  get_ready_connections
 
  # Step 3: Read from readable connections
  read_from_readable_connections
 
  # Step 4: Write to writable connections
  write_to_writable_connections
end

ここには魔法がないことに注意してください。これは通常のRubyループです。

ステップ1：着信接続を受け入れる

新しい着信接続があるかどうかを確認します。 accept_nonblockを使用します 、クライアントが接続するのを待ちません。新しいクライアントがない場合は代わりにエラーが発生し、そのエラーが発生した場合はそれをキャッチして次のステップに進みます。新しいクライアントがある場合は、そのハンドラーを作成し、それをclientsに配置します。 お店。そのHashのキーとしてソケットオブジェクトを使用します 後でクライアントハンドラーを見つけることができます。

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
  # No new incoming connections at the moment
end

ステップ2：読み取りまたは書き込みの準備ができている接続を取得する

次に、接続の準備ができたらOSに通知するように依頼します。 client_handlersのキーを渡します 読み取り、書き込み、およびエラー処理のために保存します。これらのキーは、手順1で受け入れたソケットオブジェクトです。これが発生するまで10ミリ秒待機します。

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

ステップ3：読み取り可能な接続から読み取る

接続のいずれかが読み取り可能である場合、クライアントハンドラーをトリガーし、readableで再開します。 州。 Socketであるため、これらのクライアントハンドラーを検索できます。 IO.selectによって返されるオブジェクト ハンドラストアのキーとして使用されます。

if readable
  readable.each do |ready_socket|
    # Get the client from storage
    client = $client_handlers[ready_socket]
 
    client.resume(:readable)
  end
end

ステップ4：書き込み可能な接続への書き込み

接続のいずれかが書き込み可能である場合、クライアントハンドラーをトリガーし、writableで再開します。 状態。

if writable
  writable.each do |ready_socket|
    # Get the client from storage
    client = $client_handlers[ready_socket]
    next unless client
 
    client.resume(:writable)
  end
end

ハンドラーを作成するループでこれらの4つのステップを使用し、readableを呼び出すことによって およびwritable これらのハンドラーで適切なタイミングで、完全に機能するイベントチャットサーバーを作成しました。接続ごとのオーバーヘッドはほとんどなく、これを多数の同時クライアントにスケールアップできます。

このアプローチは、ループのティックあたりの作業量を少なく保つ限り、非常にうまく機能します。イベントループは単一のスレッドで実行され、単一のCPUしか使用できないため、これは計算を伴う作業では特に重要です。本番システムでは、この制限を回避するためにイベントループを実行する複数のプロセスが存在することがよくあります。

おわりに

結局、あなたはこれらの3つの方法のどれを使うべきかと尋ねるかもしれません。

• ほとんどのアプリでは、スレッド化は理にかなっています。これは、作業するのに最も簡単なアプローチです。
• 長時間実行されるストリームを使用して同時実行性の高いアプリを実行している場合、イベントループを使用して拡張できます。
• プロセスがクラッシュすることが予想される場合は、最も堅牢なアプローチである古き良きマルチプロセスを選択してください。

これで、並行性に関するシリーズは終わりです。完全な要約が必要な場合は、元のマスタリング同時実行の記事と、複数のプロセスと複数のスレッドの使用に関する詳細な記事を確認してください。