Thành Thạo Ruby Concurrency: Xây Dựng Chat Server Với Event Loops

Chào mừng bài Ruby Magic cuối trong series về concurrency. Các edition trước implement chat server dùng multiple processes và multiple threads. Lần này làm cùng việc dùng event loop.

Recap

Ta dùng same client và server setup từ earlier articles. Mục tiêu build chat system trông như này:

Xem previous articles để details basic setup. Full source code dùng trong examples bài này available trên GitHub, bạn tự experiment.

Chat Server Dùng Event Loop

Dùng event loop cho chat server yêu cầu mental model khác threads/processes. Classic approach: thread/process chịu trách nhiệm handle single connection. Dùng event loop: single thread trong single process handle multiple connections. Xem cách hoạt động bằng cách break down.

Event Loop

Event loop dùng bởi EventMachine hoặc NodeJS hoạt động như sau. Ta bắt đầu inform OS ta quan tâm certain events. Ví dụ, khi connection đến socket mở. Làm bằng cách call function register interest trên IO object, như connection hoặc socket.

Khi có chuyện trên IO object, OS gửi event cho program. Ta đặt events lên queue. Event loop keep popping events off list và handle one by one.

Với ý nghĩa nào đó, event loop không truly concurrent. Nó work sequentially trong very small batches để simulate effect.

Để register interest và OS pass IO events cho ta, ta phải viết C extension, vì không có API cho đó trong Ruby standard library. Diving vào đó ngoài scope bài này, ta dùng IO.select thay thế generate events. IO.select nhận array IO objects monitor. Nó wait cho đến khi một hoặc nhiều objects từ array ready for reading/writing, và return array chỉ những IO objects đó.

Code lo一切 related đến connection implement as Fiber: ta gọi code này "handler". Fiber là code block có thể pause và resume. Ruby VM không làm tự động, nên ta resume và yield manually. Ta dùng input từ IO.select inform handlers khi connections ready đọc/ghi.

Giống threaded và multi-process examples trước, ta cần storage track clients và messages sent. Không cần Mutex lần này. Event loop chạy single thread, nên không risk objects bị mutate cùng lúc bởi different threads.

client_handlers = {} messages = []

Client handler implement trong Fiber sau. Khi socket readable/writable, event trigger mà Fiber respond. Khi state :readable đọc line từ socket push lên array messages. Khi state :writable ghi messages nhận từ other clients since last write cho client. Sau handle event gọi Fiber.yield, pause wait next event.

def create_client_handler(nickname, socket) Fiber.new do last_write = Time.now loop do state = Fiber.yield if state == :readable # Read a message from the socket incoming = read_line_from(socket) # All good, add it to the list to write $messages.push( :time => Time.now, :nickname => nickname, :text => incoming ) elsif state == :writable # Write messages to the socket get_messages_to_send(last_write, nickname, $messages).each do |message| socket.puts "#{message[:nickname]}: #{message[:text]}" end last_write = Time.now end end end end

Làm sao trigger Fiber read/write đúng lúc Socket ready? Ta dùng event loop 4 steps:

loop do # Step 1: Accept incoming connections accept_incoming_connections # Step 2: Get connections that are ready for reading or writing get_ready_connections # Step 3: Read from readable connections read_from_readable_connections # Step 4: Write to writable connections write_to_writable_connections end

Chú ý không có magic. Đây normal Ruby loop.

Step 1: Accept Incoming Connections

Xem có new incoming connections. Dùng accept_nonblock, không wait client connect. Thay vào đó raise error nếu no new client, catch error goto next step. Nếu có new client, create handler cho nó đặt lên clients store. Dùng socket object làm key của Hash để tìm client handler sau.

begin socket = server.accept_nonblock nickname = socket.gets.chomp $client_handlers[socket] = create_client_handler(nickname, socket) puts "Accepted connection from #{nickname}" rescue IO::WaitReadable, Errno::EINTR # No new incoming connections at the moment end

Step 2: Get Connections Ready Reading/Writing

Tiếp, hỏi OS inform khi connection ready. Pass keys của client_handlers store cho reading, writing, error handling. Keys là socket objects accepted step 1. Wait 10 milliseconds.

readable, writable = IO.select( $client_handlers.keys, $client_handlers.keys, $client_handlers.keys, 0.01 )

Step 3: Read From Readable Connections

Nếu connections readable, trigger client handlers resume với state readable. Look up client handlers vì Socket object return bởi IO.select dùng làm key của handlers store.

if readable readable.each do |ready_socket| # Get the client from storage client = $client_handlers[ready_socket] client.resume(:readable) end end

Step 4: Write To Writable Connections

Nếu connections writable, trigger client handlers resume với state writable.

if writable writable.each do |ready_socket| # Get the client from storage client = $client_handlers[ready_socket] next unless client client.resume(:writable) end end

Bằng cách dùng 4 steps trong loop tạo handlers, gọi readable và writable trên handlers đúng lúc, ta tạo fully functional evented chat server. Overhead rất nhỏ per connection, scale up large concurrent clients.

Approach này work tốt miễn giữ work per tick loop nhỏ. Đặc biệt quan trọng cho work involved calculations, vì event loop chạy single thread chỉ utilize single CPU. Production systems thường multiple processes chạy event loop workaround limitation này.

Kết Luận

Sau tất cả, bạn hỏi: method nào nên dùng?

• Hầu hết apps, threading makes sense. Simplest approach work với.
• Nếu chạy highly concurrent apps với long-running streams, event loops allow scale.
• Nếu expect processes crash, chọn good old multi-process, robust nhất.

Kết thúc series concurrency. Muốn full recap xem original mastering concurrency article và detailed articles dùng multiple processes và multiple threads.