ใน Ruby Magic article ก่อนหน้าถึง Mastering Concurrency เราให้ introduction กับ 3 methods ของการ achieve concurrency ที่ available ให้เราในฐานะ Ruby developers บทความนี้เป็น first ใน three-part series ที่เรา deep dive แต่ละ method

First up: Multi-process ด้วย method นี้ master process fork ตัวเองเป็น multiple worker processes Worker process ทำ actual work ขณะ master manages workers

Full source code ที่ใช้ในตัวอย่างใน article นี้ available on GitHub ลองทดลองเองได้

Let's Build a Chat System!

การสร้าง chat system เป็น good way เพื่อ dive into concurrency เราต้องการ server component ของ chat system ที่สามารถ maintain connections กับ multiple clients นี่จะ allow เรา distribute messages ที่รับจาก client หนึ่งไปยัง all other connected clients

Chat server ของเรารันใน left tab มี two chat clients รันใน right tabs ข้อความที่ส่งโดย client จะถูก received โดย all other clients

The Chat Client

Article นี้ focus บน chat server แต่เพื่อ communicate กับมัน เราต้องการ chat client ก่อน โค้ดต่อไปจะเป็น very simple client ของเรา (complete example หาได้บน 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

Client เปิด TCP connection ไป server รันบน port 2000 เมื่อ connected มัน spawn thread ที่จะ puts อะไรที่ server ส่ง ทำให้ chat visible ใน terminal output สุดท้ายมี while loop ส่งบรรทัดที่คุณพิมพ์ไป server ซึ่งจะส่งไป all other connected clients

The Chat Server

ในตัวอย่างนี้ client connects ไป chat server เพื่อ communicate กับ other clients สำหรับ all three concurrency approaches เราจะใช้ TCP server เดียวกันจาก Ruby's standard library

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

จนถึงจุดนี้ code เหมือนกันสำหรับ all three concurrency models Chat server ในทุก model จะต้อง handle 2 scenarios:

Accept new connections จาก clients
Receive messages จาก clients และส่งไป all other clients

A Multi-process Chat Server

เพื่อ handle 2 scenarios นี้ด้วย multi-process chat server เราจะ spawn process ต่อ client connection Process นี้จะ handle all messages being sent และ received สำหรับ client นั้น เราสร้าง processes เหล่านี้โดย fork original server process

Forking Processes

เมื่อคุณเรียก fork method มันสร้าง copy ของ current process ด้วย exact same state ที่ process อยู่

Forked process มี process id ของตัวเอง และจะ visible แยกกันใน tool อย่าง top หรือ Activity Monitor เป็นแบบนี้:

Process ที่คุณเริ่มเรียก master process และ processes ที่ fork ออกจาก master process เรียก worker processes

เนื่องจาก newly forked worker processes เหล่านี้เป็น truly separate processes เราไม่สามารถ share memory ระหว่างพวกเขาและ master process ได้ เราต้องการสิ่งหนึ่ง communicate ระหว่างพวกเขา

Unix Pipes

เพื่อ communicate ระหว่าง processes เราจะใช้ Unix pipes Unix pipe ตั้งค่า two-way stream of bytes ระหว่าง two processes และคุณใช้ส่ง data จาก process หนึ่งไปอีก process ได้ โชคดีที่ Ruby ให้ nice wrapper รอบ pipes นี้ ดังนั้นเราไม่ต้อง re-invent the wheel

ในตัวอย่างต่อไปเราตั้งค่า pipe ใน Ruby –พร้อม reading และ writing end– และเรา fork master process โค้ดภายใน block ที่ส่งให้ fork รันใน forked process Original process continue หลัง block นี้ เราจึงเขียน message ไป original process จาก forked process

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

ใช้ pipes เรา communicate ระหว่าง separate processes ได้แม้ processes isolated จากกันสมบูรณ์

The Chat Server's Implementation

First เราตั้งค่า array เก็บ track ของ pipes สำหรับ all clients และ "writers" (writing end ของ pipe) เพื่อ communicate กับ clients แล้วเราทำให้ all incoming messages จาก clients ส่งไป all other clients

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

คุณหา implementation ของ write_incoming_messages_to_child_processes บน GitHub ได้ หากอยากดูรายละเอียดการ operate

Accepting New Connections

เราต้อง accept incoming connections และ setup pipes New writer จะ pushed onto client_writers array Main process จะ loop through array และส่ง message ไปแต่ละ worker process โดยเขียนไป pipe ของมัน

เราแล้ว fork master process และ code ภายใน forked worker process จะ handle client connection

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

Handling Client Connections

เราต้อง handle client connection ด้วย

Forked process เริ่มด้วยการได้ nickname จาก client (client ส่ง nickname โดย default) หลังจากนั้นมันเริ่ม thread ใน write_incoming_messages_to_client ที่ listen สำหรับ messages จาก main process

สุดท้าย forked process เริ่ม loop listen สำหรับ incoming messages และส่งไป master process Master process ทำให้ other worker processes รับ message

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}"

A Working Chat System

ตอนนี้ whole chat system works! แต่เห็นได้ว่าเขียนโปรแกรมใช้ multiprocessing ค่อนข้าง complex และใช้ resources เยอะ Upside คือมัน very robust หาก child process หนึ่ง crash ระบบส่วนอื่นก็ทำงานต่อได้ ลองดูโดยรัน example code และรัน kill -9 บน process หนึ่ง (หา process id ได้ใน server's log output)

ใน next article เราจะ implement chat system เดียวกันแค่ใช้ threads ดังนั้นเราจะรัน server พร้อม features เดียวกันใช้แค่ one process และ memory น้อยกว่า

Deep Dive Ruby Concurrency: Mastering Multi-Process Techniques