並行性の詳細解説：マルチスレッドで実現するRubyチャットサーバー

前回のRuby Magicでは、複数のプロセスを使ってチャットシステムを実装する方法をご紹介しました。今回は、同じことをマルチスレッドで実現する方法を解説します。

おさらい

基本的なセットアップの詳細については、前回の記事をご覧ください。ここで簡単におさらいしておきましょう。今回作成するチャットシステムは次のような構成になっています。

クライアント側のコードは前回と同じものを使用します。

# client.rb
# $ ruby client.rb
require 'socket'
client = TCPSocket.open(ARGV[0], 2000)
 
Thread.new do
  while line = client.gets
    puts line.chop
  end
end
 
while input = STDIN.gets.chomp
  client.puts input
end

サーバーの基本的なセットアップも同様です。

# server_threads.rb
# $ ruby server_threads.rb
require 'socket'
 
puts 'Starting server on port 2000'
 
server = TCPServer.open(2000)

この記事の例で使用している完全なソースコードはGitHubで公開しています。ぜひご自身でも動かして試してみてください。

マルチスレッド対応チャットサーバー

ここからが、マルチプロセス実装との違いが出てくる部分です。マルチスレッドを使えば、たった1つのRubyプロセスで複数の処理を同時に実行できます。具体的には、作業を担当する複数のスレッドを生成することでこれを実現します。

スレッドとは

スレッドは、プロセス内で独立してコードを実行する仕組みです。1つのプロセスの中に複数のスレッドが存在でき、それらはメモリを共有できます。

受信したチャットメッセージを保存するための場所が必要になります。ここではシンプルなArrayを使用しますが、同時に複数のスレッドがメッセージを書き換えてしまわないよう、Mutex（ミューテックス）も併せて使います。Mutexの仕組みについては後ほど詳しく説明します。

mutex = Mutex.new
messages = []

次に、チャットクライアントからの接続を受け付けるループを開始します。接続が確立されたら、そのクライアントとのメッセージ送受信を担当するスレッドを生成します。

Thread.newはserver.acceptが値を返すまでブロックし、新しく作成されたスレッド内でブロックを実行します。スレッド内のコードはまず最初に送られてきた1行を読み込み、それをニックネームとして保存します。その後、メッセージの送信と読み取りを開始します。

loop do
  Thread.new(server.accept) do |socket|
    nickname = read_line_from(socket)
 
    # 受信メッセージの送信（後述）
 
    # 受信メッセージの読み取り（後述）
  end
end

Mutex（ミューテックス）とは

ミューテックスは、配列などの共有リソースへのアクセス方法を複数のスレッド間で調整するためのオブジェクトです。あるスレッドが「アクセスが必要」であることを示すと、その間は他のスレッドが共有リソースにアクセスできなくなります。

サーバーはソケットから受信メッセージを読み取ります。synchronizeを使ってメッセージストアのロックを取得することで、安全にメッセージをArrayへ追加できるようにしています。

# 受信メッセージの読み取り
while incoming = read_line_from(socket)
  mutex.synchronize do
    messages.push(
      :time => Time.now,
      :nickname => nickname,
      :text => incoming
    )
  end
end

最後に、サーバーが受け取った新しいメッセージをすべてクライアントへ送り届けるため、常時ループで動作するThreadを生成します。こちらもロックを取得することで、他のスレッドとの干渉を防いでいます。ループの1周が終わるごとに少しスリープし、その後処理を続けます。

# 受信メッセージの送信
Thread.new do
  sent_until = Time.now
  loop do
    messages_to_send = mutex.synchronize do
      get_messages_to_send(nickname, messages, sent_until).tap do
        sent_until = Time.now
      end
    end
    messages_to_send.each do |message|
      socket.puts "#{message[:nickname]}: #{message[:text]}"
    end
    sleep 0.2
  end
end

GIL（グローバルインタプリタロック）について

RubyにはGIL（Global Interpreter Lock）があるため「本当の」マルチスレッド処理ができない、という話を耳にしたことがあるかもしれません。これは部分的には正しい話です。GILはすべてのRubyコードの実行を囲むロックであり、1つのRubyプロセスが複数のCPUを同時に使用することを防ぎます。しかし、ネットワーク接続など本記事で扱ったようなIO操作はGILの外側で動作するため、このようなケースでは実際に十分な並行性を実現できます。

まとめ

これで、1つのプロセス内で接続ごとにスレッドを割り当てる形のチャットサーバーが完成しました。マルチプロセス実装と比べると、はるかに少ないリソースで動作します。コードの詳細を確認したい場合や、実際に試してみたい場合は、サンプルコードをご覧ください。

このシリーズの最終回となる次回の記事では、同じチャットサーバーをシングルスレッドとイベントループで実装します。理論上は、スレッド実装よりもさらに少ないリソースで動作するはずです。お楽しみに！