เชี่ยวชาญ Ruby Concurrency: สร้าง Chat Server ด้วย Event Loops

ยินดีต้อนรับสู่ Ruby Magic article สุดท้ายใน series เกี่ยว concurrency ใน edittions ก่อนหน้าเรา implement chat server ใช้ multiple processes และ multiple threads ครั้งนี้เราจะทำอย่างเดียวกันด้วย event loop

Recap

เราจะใช้ client และ server setup เดียวกับ articles ก่อนหน้า เป้าคือสร้าง chat system ที่เป็นแบบนี้:

ดู previous articles เพื่อรายละเอียด basic setup โค้ดเต็มที่ใช้ในตัวอย่างใน article นี้ available on GitHub ลองทดลองเองได้

Chat Server ใช้ Event Loop

การใช้ event loop สำหรับ chat server ต้องการ mental model ต่างจาก threads หรือ processes ใน classic approach thread หรือ process หนึ่ง responsible สำหรับ handle single connection การใช้ event loop คุณมี single thread ใน single process ที่ handle multiple connections มาดูยังไงโดยแยกส่วน

Event Loop

Event loop ที่ EventMachine หรือ NodeJS ใช้ทำงานดังนี้ เราเริ่มด้วยการ inform operating system เราสนใจ certain events เช่นเมื่อ connection ไป socket เปิด เราทำได้โดย call function ที่ registers interest บน IO object บางอย่าง เช่น connection หรือ socket

เมื่อมีอะไรเกิดบน IO object นี้ operating system ส่ง event ไป program ของเรา เราวาง events เหล่านี้บน queue Event loop ทำ popping events ออกจาก list และ handle ทีละอัน

ในความรู้สึก event loop ไม่ได้ truly concurrent มันทำงาน sequentially ใน very small batches เพื่อ simulate effect

เพื่อ register interest และให้ OS ส่ง IO events ให้เรา เราต้องเขียน C extension เพราะไม่มี API สำหรับนั้นใน Ruby standard library การ diving ลงไปอยู่นอก scope ของ article นี้ ดังนั้นเราจะใช้ IO.select แทน generate events IO.select รับ array ของ IO objects ไป monitor มัน wait จนกว่าหนึ่งหรือมากกว่าจาก array จะ ready สำหรับ reading หรือ writing และคืน array พร้อมแค่那些 IO objects

Code ที่จัดการทุกอย่าง related to connection implement เป็น Fiber เราจะเรียก code นี้ว่า "handler" ตั้งแต่ตอนนี้ Fiber คือ code block ที่ pause และ resume ได้ Ruby VM ไม่ทำ automatic ดังนั้นเราต้อง resume และ yield เอง เราจะใช้ input จาก IO.select inform handlers เมื่อ connections ของพวกเขา ready สำหรับ reading หรือ writing

เหมือนใน threaded และ multi-process examples จาก posts ก่อนหน้า เราต้องการ storage เก็บ track ของ clients และ messages ที่ส่ง เราไม่ต้องการ Mutex ครั้งนี้ Event loop ของเรารันใน single thread ดังนั้นไม่มี risk ของ objects being mutated พร้อมกันโดย different threads

client_handlers = {}
messages = []

Client handler implement ใน Fiber ต่อไปนี้ เมื่อ socket อ่านหรือเขียนได้ event จะ triggered ซึ่ง Fiber responds เมื่อ state เป็น :readable มันอ่าน line จาก socket และ push ไป messages array เมื่อ state เป็น :writable มันเขียน messages ที่รับจาก clients อื่นๆ ตั้งแต่ last write ไป client หลัง handle event มันเรียก Fiber.yield ดังนั้นมันจะ pause และรอ 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

แต่เรา trigger Fiber ให้ read หรือ write เวลาที่ถูกต้องเมื่อ Socket ready ยังไง? เราใช้ 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

สังเกตว่าไม่มี magic ที่นี่ นี่คือ normal Ruby loop

Step 1: Accept Incoming Connections

ดูว่ามี incoming connections ใหม่ไหม เราใช้ accept_nonblock ซึ่งจะไม่ wait for client to connect มันจะ raise error แทนถ้าไม่มี client ใหม่ และถ้า error เกิดเราจับมันและไป step ถัดไป หากมี client ใหม่เราสร้าง handler สำหรับมันและวางบน clients store เราจะใช้ socket object เป็น key ของ Hash นั้น เพื่อหา client handler ภายหลัง

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 for Reading or Writing

ต่อไป เราขอ OS inform เราเมื่อ connection ready เราส่ง keys ของ client_handlers store สำหรับ reading, writing และ error handling Keys เหล่านี้คือ socket objects ที่เรา accept ใน 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

ถ้ามี connections ใด readable เราจะ trigger client handlers และ resume พวกเขาด้วย state readable เรา lookup client handlers ได้เพราะ Socket object ที่คืนโดย IO.select ใช้เป็น key ของ 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

ถ้ามี connections ใด writable เราจะ trigger client handlers และ resume พวกเขาด้วย 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

ด้วยการใช้ 4 steps นี้ใน loop ที่สร้าง handlers และเรียก readable และ writable บน handlers เหล่านี้เวลาที่ถูกต้อง เราสร้าง fully functional evented chat server ได้ มี overhead น้อยมาต่อ connection และเราสามารถ scale ขึ้นไป large number of concurrent clients ได้

Approach นี้ทำงานดีตราบใดที่เราเก็บ amount of work ต่อ tick ของ loop ให้น้อย โดยเฉพาะ work ที่เกี่ยวกับ calculations เนื่องจาก event loop รันใน single thread จึงใช้ single CPU ได้อย่างเดียว ใน production systems มักมี multiple processes รัน event loop เพื่อ work around limitation นี้

Concluding

หลังจากทั้งหมดนี้คุณอาจถาม วิธีไหนใน 3 วิธีนี้ควรใช้?

• สำหรับ most apps threading makes sense มันคือ simplest approach ทำงานได้
• ถ้าคุณรัน highly concurrent apps พร้อม long-running streams event loops ให้คุณ scale
• ถ้าคุณคาดว่า processes จะ crash ไปกับ good old multi-process เพราะมันคือ most robust approach

นี่คือสรุป series เกี่ยว concurrency หากอยาก full recap ดู original mastering concurrency article รวมถึง detailed articles เรื่อง multiple processes และ multiple threads ได้เลย