เชี่ยวชาญ Ruby Concurrency: สร้าง Multi-Threaded Chat System

ใน Ruby Magic edition ก่อนหน้า เราแสดงวิธี implement chat system ใช้ multiple processes ครั้งนี้เราจะแสดงว่าคุณทำอย่างเดียวกันได้ด้วย multiple threads

Quick Recap

หากอยากได้ full explanation ของ basic setup ดู previous article แต่เตือนให้รู้เร็ว: chat system ของเราเป็นแบบนี้:

เราใช้ client เดียวกับที่ใช้ก่อนหน้า:

# 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

Basic setup สำห server เหมือนเดิม:

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

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

Multi-threaded Chat Server

ตอนนี้เราไปถึงส่วนที่ต่างจาก multi-process implementation ใช้ Multi-threading เราทำ multiple things พร้อมกันได้ด้วยแค่ one Ruby process เราจะทำโดย spawn multiple threads ที่ทำงาน

Threads

Thread รัน independently execute code ภายใน process Multiple threads สามารถอยู่ใน process เดียวกันและ share memory ได้

<img src="/images/blog/2017-04/threads.png"> 

Storage บางอย่างจำเป็นเก็บ incoming chat messages เราจะใช้ plain Array แต่เราต้องการ Mutex ด้วย เพื่อให้แน่ใจว่าแค่ one thread เปลี่ยน messages พร้อมกัน (เราจะเห็น Mutex ทำงานยังไงในบิต)

mutex = Mutex.new
messages = []

ต่อไปเราเริ่ม loop ที่เรา accept incoming connections จาก chat clients เมื่อ connection established เราจะ spawn thread handle incoming และ outgoing messages จาก client connection นั้น

Thread.new call blocks จนกว่า server.accept จะคืนค่า และ yield block ต่อไปใน newly created thread โค้ดใน thread จึงดำเนินต่ออ่านบรรทัดแรกที่ส่งมาเก็บเป็น nickname สุดท้ายเริ่มส่งและอ่าน messages

loop do
  Thread.new(server.accept) do |socket|
    nickname = read_line_from(socket)
    # Send incoming message (coming up)
    # Read incoming messages (coming up)
  end
end

Mutex

Mutex เป็น object ให้ multiple threads coordinate ว่าใช้ shared resources อย่างไร เช่น array Thread หนึ่ง indicate ว่าต้องการ access และในช่วงเวลานั้น threads อื่นๆ เข้าถึง shared resource ไม่ได้

Server อ่าน incoming messages จาก socket ใช้ synchronize ได้ lock บน messages store เพื่อเพิ่ม message ลง Array อย่างปลอดภัย

# Read incoming messages
while incoming = read_line_from(socket)
  mutex.synchronize do
    messages.push(
      :time => Time.now,
      :nickname => nickname,
      :text => incoming
    )
  end
end

สุดท้าย Thread spawn ที่รัน continuously ใน loop เพื่อให้แน่ใจว่า all new messages ที่ server ได้รับ ถูกส่งไป client อีกครั้งได้ lock เพื่อรู้ว่า threads อื่นไม่ interfere หลังจบ tick ของ loop มัน sleep บิตแล้ว continue

# Send incoming message
Thread.new do
  sent_until = Time.now
  loop do
    messages_to_send = mutex.synchronize do
      get_messages_to_send(nickname, messages, sent_until).tap do
        sent_until = Time.now
      end
    end
    messages_to_send.each do |message|
      socket.puts "#{message[:nickname]}: #{message[:text]}"
    end
    sleep 0.2
  end
end

Global Interpreter Lock

คุณอาจได้ยินเรื่อง Ruby ไม่ทำ "real" threading ได้เพราะ Global Interpreter Lock (GIL) นี่ partially true GIL เป็น lock รอบ execution ของ Ruby code ทั้งหมด ป้องกัน Ruby process ใช้ multiple CPUs concurrently IO operations (เช่น network connections ที่เราใช้ใน article นี้) operate outside GIL หมายความว่าคุณ achieve decent concurrency ได้ในกรณีนี้

Concluding

ตอนนี้เรามี chat server รันภายใน single process ใช้ thread ต่อ connection นี่ใช้ resources น้อยกว่า multi-process implementation มาก หากอยากดูรายละเอียดโค้ดหรือลอง ทดสอบได้ที่ example code

ใน final article ใน series นี้ เราจะ implement chat server เดียวกันแค่ใช้ single thread และ event loop ทางทฤษฎีควรใช้ resources น้อยกว่า thread implementation!