並行性の習得に関する以前のRubyMagicの記事では、Ruby開発者として利用できる並行性を実現する3つの方法を紹介しました。この記事は、各方法を深く掘り下げた3部構成のシリーズの最初の記事です。

最初に：マルチプロセス 。この方法では、マスタープロセスはそれ自体を複数のワーカープロセスにフォークします。マスターがワーカーを管理している間、ワーカープロセスが実際の作業を行います。

この記事の例で使用されている完全なソースコードはGitHubで入手できるため、自分で試すことができます。

チャットシステムを構築しましょう！

チャットシステムを構築することは、並行性に飛び込むための良い方法です。複数のクライアントとの接続を維持できるチャットシステムのサーバーコンポーネントが必要になります。これにより、1つのクライアントから受信したメッセージを、接続されている他のすべてのクライアントに配信できます。

チャットサーバーは左側のタブで実行されています。右側のタブで実行されている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)

この時点まで、コードは3つの同時実行モデルすべてで同じです。すべてのモデルのチャットサーバーは、次の2つのシナリオを処理する必要があります。

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

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

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

フォークプロセス

forkメソッドを呼び出すと、プロセスが存在するのとまったく同じ状態で現在のプロセスのコピーが作成されます。

フォークされたプロセスには独自のプロセスIDがあり、topなどのツールで個別に表示されます。 またはアクティビティモニター。これは次のようになります：

開始するプロセスはマスタープロセスと呼ばれ、マスタープロセスから分岐したプロセスはワーカープロセスと呼ばれます。

これらの新しくフォークされたワーカープロセスは本当に別個のプロセスであるため、それらとマスタープロセスの間でメモリを共有することはできません。彼らの間でコミュニケーションをとる何かが必要です。

Unixパイプ

プロセス間で通信するには、Unixパイプを使用します。 Unixパイプは、2つのプロセス間に双方向のバイトストリームを設定し、それを使用して1つのプロセスから別のプロセスにデータを送信できます。幸いなことに、Rubyはこれらのパイプの周りに優れたラッパーを提供しているため、ホイールを再発明する必要はありません。

次の例では、Rubyでパイプを設定し、読み取りと書き込みの終わりを設定し、fork マスタープロセス。 forkに渡されるブロック内のコード フォークされたプロセスで実行されています。元のプロセスは、このブロックの後も続行されます。次に、分岐したプロセスから元のプロセスにメッセージを書き込みます。

reader, writer = IO.pipe
 
fork do
  # This is running in the forked process.
  writer.puts 'Hello from the forked process'
end
 
# This is running in the original process, it will puts the
# message from the forked process.
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の動作の詳細を確認したい場合は、GitHubで確認してください。

新しい接続を受け入れる

着信接続を受け入れ、パイプを設定する必要があります。新しいライターはclient_writersにプッシュされます 配列。メインプロセスは、配列をループして、パイプに書き込むことで各ワーカープロセスにメッセージを送信できます。

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

loop do
  while socket = server.accept
    # Create a client reader and writer so that the master
    # process can write messages back to us.
    client_reader, client_writer = IO.pipe
 
    # Put the client writer on the list of writers so the
    # master process can write to them.
    client_writers.push(client_writer)
 
    # Fork child process, everything in the fork block
    # only runs in the child process.
    fork do
      # Handle connection
    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)
 
# Read incoming messages from the client.
while incoming = read_line_from(socket)
  master_writer.puts "#{nickname}: #{incoming}"
end
 
puts "#{Process.pid}: Disconnected #{nickname}"

機能するチャットシステム

これで、チャットシステム全体が機能します。しかし、ご覧のとおり、マルチプロセッシングを使用するプログラムの作成は非常に複雑で、多くのリソースを使用します。利点は、非常に堅牢であるということです。子プロセスの1つがクラッシュしても、システムの残りの部分は機能し続けます。サンプルコードを実行し、kill -9 <process-id>を実行することで、これを試すことができます。 プロセスの1つで（プロセスIDはサーバーのログ出力で確認できます）。

次の記事では、スレッドのみを使用して同じチャットシステムを実装するため、1つのプロセスと少ないメモリを使用して同じ機能でサーバーを実行できます。