以前の「Ruby Magic」の並行性入門記事では、Ruby開発者が利用できる3つの並行性実現手法について紹介しました。本記事は、それぞれの手法を詳しく掘り下げていく3部作シリーズの第1弾です。

まず取り上げるのはマルチプロセスです。この手法では、マスタープロセスが自身を複数のワーカープロセスにフォークします。実際の処理はワーカープロセスが担当し、マスタープロセスはそれらのワーカーを管理します。

本記事の例で使用しているソースコード一式はGitHubで公開されているので、ぜひ自分でも試してみてください。

チャットシステムを作ってみよう!

並行性を学ぶには、チャットシステムの構築がうってつけです。ここでは、複数のクライアントとの接続を維持できるチャットシステムのサーバー側コンポーネントが必要になります。これにより、あるクライアントから受信したメッセージを、接続中の他のすべてのクライアントへ配信できるようになります。

チャットサーバーは左側のタブで稼働しており、右側のタブでは2つのチャットクライアントが動いています。あるクライアントが送信したメッセージは、他のすべてのクライアントで受信されます。

チャットクライアント

本記事の主役はチャットサーバーですが、通信するためにはまずチャットクライアントが必要です。以下のコードは非常にシンプルなクライアントです(より完全な例はGitHubで公開しています)。

# 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

このクライアントは、ポート2000で待ち受けるサーバーへのTCP接続を開きます。接続すると、サーバーから送られてきた内容をputsで出力するスレッドを生成するため、ターミナル上でチャットの様子を確認できます。最後に、入力された行をサーバーへ送信するwhileループがあり、サーバーはそのメッセージを他の接続済みクライアント全員に配信します。

チャットサーバー

この例では、クライアントがチャットサーバーに接続して他のクライアントとやり取りします。3つの並行性モデルすべてにおいて、Ruby標準ライブラリの同じTCPサーバーを使用します。

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

ここまでのコードは、どの並行性モデルでも共通です。各モデルのチャットサーバーは、次の2つのシナリオに対応する必要があります。

1. クライアントからの新しい接続を受け入れる。
2. クライアントからメッセージを受信し、他のすべてのクライアントへ送信する。

マルチプロセス版チャットサーバー

マルチプロセスのチャットサーバーでこれら2つのシナリオに対応するには、クライアント接続ごとにプロセスを1つ生成します。このプロセスが、そのクライアントに関するメッセージの送受信をすべて処理します。プロセスは、元のサーバープロセスをフォークすることで作成できます。

プロセスのフォーク

forkメソッドを呼び出すと、現在のプロセスの状態をそのまま引き継いだコピーが作成されます。

フォークされたプロセスは独自のプロセスIDを持ち、topやアクティビティモニタなどのツールでは別々のプロセスとして表示されます。イメージは次のような感じです。

最初に起動したプロセスを「マスタープロセス」と呼び、マスタープロセスからフォークされたプロセスを「ワーカープロセス」と呼びます。

フォークされたワーカープロセスは完全に独立したプロセスであるため、マスタープロセスとの間でメモリを共有することはできません。何らかの手段でプロセス間の通信を行う必要があります。

Unixパイプ

プロセス間の通信にはUnixパイプを使用します。Unixパイプは2つのプロセス間に双方向のバイトストリームを確立し、一方のプロセスからもう一方へデータを送るために使えます。幸いなことに、Rubyにはパイプの便利なラッパーが用意されているので、車輪の再発明をする必要はありません。

次の例では、読み込み側と書き込み側を持つパイプをRubyでセットアップし、マスタープロセスをforkします。forkに渡されたブロック内のコードはフォークされたプロセスで実行され、元のプロセスはブロックの後から処理を続けます。そして、フォーク先のプロセスから元のプロセスへメッセージを書き込んでいます。

reader, writer = IO.pipe
 
fork do
  # ここはフォークされたプロセスで実行される
  writer.puts 'Hello from the forked process'
end
 
# こちらは元のプロセスで実行され、
# フォーク先からのメッセージを出力する
puts reader.gets

パイプを使えば、互いに完全に分離されたプロセス同士でも通信が可能になります。

チャットサーバーの実装

まず、すべてのクライアントのパイプとその「ライター」(パイプの書き込み側)を追跡するための配列を用意します。これによりクライアントと通信できるようになります。次に、クライアントから届いたメッセージを他のすべてのクライアントへ転送する仕組みを作ります。

client_writers = []
master_reader, master_writer = IO.pipe
 
write_incoming_messages_to_child_processes(master_reader, client_writers)

write_incoming_messages_to_child_processesの実装詳細はGitHubで確認できます。

新しい接続の受け入れ

次に、着信接続を受け入れ、パイプをセットアップします。新しいライターはclient_writers配列に追加されます。メインプロセスはこの配列をループして、各ワーカープロセスのパイプに書き込むことでメッセージを送信できます。

続いてマスタープロセスをフォークし、フォークされたワーカープロセス内のコードがクライアント接続を処理します。

loop do
  while socket = server.accept
    # クライアント用のリーダーとライターを作成し、
    # マスタープロセスがメッセージを返せるようにする
    client_reader, client_writer = IO.pipe
 
    # ライターをリストに追加して、
    # マスタープロセスから書き込めるようにする
    client_writers.push(client_writer)
 
    # 子プロセスをフォーク。forkブロック内の処理は
    # 子プロセスでのみ実行される
    fork do
      # 接続を処理する
    end
  end
end

クライアント接続の処理

さらに、クライアント接続自体の処理も必要です。

フォークされたプロセスは、まずクライアントからニックネームを取得します(クライアントはデフォルトでニックネームを送信します)。その後、write_incoming_messages_to_client内でスレッドを開始し、メインプロセスからのメッセージを待ち受けます。

最後に、フォークされたプロセスはループに入り、着信メッセージを待ち受けてマスタープロセスへ送信します。マスタープロセスが、他のワーカープロセスへのメッセージ配信を担います。

nickname = read_line_from(socket)
puts "#{Process.pid}: Accepted connection from #{nickname}"
 
write_incoming_messages_to_client(nickname, client_reader, socket)
 
# クライアントからの着信メッセージを読み取る
while incoming = read_line_from(socket)
  master_writer.puts "#{nickname}: #{incoming}"
end
 
puts "#{Process.pid}: Disconnected #{nickname}"

動作するチャットシステムの完成

これでチャットシステム全体が動くようになりました!ただし、ご覧のとおりマルチプロセスを使ったプログラムはかなり複雑で、多くのリソースを消費します。その一方で、非常に堅牢という大きなメリットがあります。子プロセスの1つがクラッシュしても、システムの残りの部分は動作し続けます。サンプルコードを実行して、いずれかのプロセスに対してkill -9 <process-id>を実行してみてください(プロセスIDはサーバーのログ出力で確認できます)。

次回の記事では、同じチャットシステムをスレッドだけで実装します。そうすれば、単一のプロセスで少ないメモリ使用量ながら同じ機能を持つサーバーを動かせるようになります。