Trong bài Ruby Magic trước về Mastering Concurrency, chúng tôi giới thiệu 3 methods achieve concurrency available cho Ruby developers. Bài này đầu tiên trong 3-part series deep dive từng method.

Đầu tiên: Multi-process. Với method này, master process fork thành multiple worker processes. Worker process làm actual work, master manage workers.

Full source code dùng trong examples bài này available trên GitHub, bạn tự experiment.

Xây Dựng Chat System!

Xây chat system là cách tốt dive vào concurrency. Ta cần server component chat system maintain connections với multiple clients. Cho phép distribute messages nhận từ one client đến all other connected clients.

Chat server chạy left tab. Hai chat clients chạy right tabs. Message gửi bởi client sẽ nhận bởi all other clients.

Chat Client

Bài focus vào chat server, nhưng communicate với nó cần chat client trước. Code sau là client rất simple. (Complete example trên 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 mở TCP connection đến server port 2000. Khi connected, spawn thread puts gì server gửi, visible chat terminal output. Cuối cùng, while loop gửi line bạn type đến server, gửi đến all other connected clients.

Chat Server

Trong example client connect chat server communicate với other clients. Với 3 concurrency approaches ta dùng same TCP server từ Ruby standard library.

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

Đến đây code same cho 3 concurrency models. Chat server mọi model handle 2 scenarios:

1. Accept new connections từ clients.
2. Receive messages từ clients gửi đến all other clients.

Multi-process Chat Server

Handle 2 scenarios với multi-process chat server, ta spawn process per client connection. Process này handle all messages sent/received cho client đó. Tạo processes bằng cách fork original server process.

Forking Processes

Khi call fork method, nó tạo copy của current process với exact same state process đang ở.

Forked process có own process id, visible separately trong tool như top hoặc Activity Monitor. Trông như:

Process start gọi master process, processes forked từ master gọi worker processes.

Vì newly forked worker processes thực sự separate processes, ta không share memory giữa chúng và master process. Cần cái gì communicate giữa chúng.

Unix Pipes

Communicate giữa processes dùng Unix pipes. Unix pipe setup two-way byte stream giữa 2 processes, dùng send data từ process này sang kia. May mắn Ruby offer nice wrapper quanh pipes không reinvent wheel.

Ví dụ sau setup pipe Ruby – với reading và writing end – và fork master process. Code trong block pass cho fork chạy trong forked process. Original process continue sau block. Ta write message từ forked process về original 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

Dùng pipes communicate giữa separate processes dù processes hoàn toàn isolated.

Implementation Chat Server

Đầu setup array track pipes all clients và "writers" (writing end pipe), để communicate với clients. Sau đảm bảo all incoming messages từ clients gửi đến 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 trên GitHub nếu muốn xem details.

Accepting New Connections

Cần accept incoming connections setup pipes. New writer push lên array client_writers. Main process loop through array gửi message đến each worker process bằng cách write to its pipe.

Sau fork master process, code trong 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

Cũng cần handle client connection.

Forked process start lấy nickname từ client (client gửi nickname mặc định). Sau start thread trong write_incoming_messages_to_client listen messages từ main process.

Cuối cùng, forked process start loop listen incoming messages gửi đến master process. Master process đảm bảo other worker processes nhận 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}"

Working Chat System

Bây giờ whole chat system works! Nhưng như thấy, viết program dùng multiprocessing phức tạp và dùng nhiều resources. Upside: rất robust. Nếu child process crash, rest system just keep working. Thử chạy example code và kill -9 <process-id> trên một process (tìm process id trong server log output).

Next article implement same chat system chỉ dùng threads, chạy server same features dùng one process và less memory.