রুবি কাঙ্কারেন্সি মাস্টার করুন: ইভেন্ট লুপ দিয়ে চ্যাট সার্ভার বিল্ড করুন

আমাদের কাঙ্কারেন্সি নিয়ে সিরিজের এই লাস্ট রুবি ম্যাজিক আর্টিকেলে স্বাগতম। আগের এডিশনগুলোতে আমরা একাধিক প্রসেস এবং একাধিক থ্রেড ব্যবহার করে একটি চ্যাট সার্ভার ইমপ্লিমেন্ট করেছিলাম। এইবার আমরা একই জিনিস একটি ইভেন্ট লুপ ব্যবহার করে করব।

রিক্যাপ

আমরা আগের আর্টিকেলগুলোতে ব্যবহৃত একই ক্লায়েন্ট এবং সার্ভার সেটআপ ব্যবহার করব। আমাদের লক্ষ্য একটি চ্যাট সিস্টেম বিল্ড করা যেটি এভাবে দেখে:

বেসিক সেটআপের আরও ডিটেইলসের জন্য আগের আর্টিকেলগুলো দেখুন। এই আর্টিকেলের উদাহরণগুলোর ফুল সোর্স কোড গিটহাবে উপলব্ধ, যাতে আপনি নিজে এক্সপেরিমেন্ট করতে পারেন।

ইভেন্ট লুপ ব্যবহার করে চ্যাট সার্ভার

আমাদের চ্যাট সার্ভারের জন্য ইভেন্ট লুপ ব্যবহার করতে হলে আপনাকে একটি ভিন্ন মেন্তাল মডেল রাখতে হবে থ্রেড বা প্রসেসের চেয়ে। ক্লাসিক অপ্রোച്ചে, একটি থ্রেড বা প্রসেস একক কানেকশন হ্যান্ডল করার জন্য দায়িত্বপূর্ণ। ইভেন্ট লুপ ব্যবহার করলে আপনার একটি সিঙ্গেল থ্রেড একটি সিঙ্গেল প্রসেসে থাকবে যা একাধিক কানেকশন হ্যান্ডল করে। চলুন এটা কীভাবে কাজ করে সেটা ব্রেক ডাউন করে দেখি।

ইভেন্ট লুপ

EventMachine বা NodeJS দ্বারা ব্যবহৃত একটি ইভেন্ট লুপ নিচের মতো কাজ করে। আমরা অপারেটিং সিস্টেমকে জানাই আমরা নির্দিষ্ট ইভেন্টগুলোর মধ্যে ইন্টারেস্টেড। উদাহরণস্বরূপ, যখন একটি সকেটের সাথে কানেকশন খুলা হয়। আমরা সেটি একটি ফাংশন কল করে করি যা কিছু IO অবজেক্টে ইন্টারেস্ট রেজিস্টার করে, যেন একটি কানেকশন বা সকেট।

যখন এই IO অবজেক্টে কিছু ঘটে, অপারেটিং সিস্টেম আমাদের প্রোগ্রামে একটি ইভেন্ট পাঠায়। আমরা এই ইভেন্টগুলো একটি কিউতে রাখি। ইভেন্ট লুপ কিউ থেকে এক এক করে ইভেন্ট পপ করে এবং সেগুলো হ্যান্ডল করে এক এক করে।

একটি অর্থে ইভেন্ট লুপ সত্যিকারের কাঙ্কারেন্ট নয়। এটা সিকোয়েনশিয়ালি খুব ছোট ছোট ব্যাচে কাজ করে কাঙ্কারেন্সির ইফেক্ট সিমুলেট করতে।

ইন্টারেস্ট রেজিস্টার করতে এবং অপারেটিং সিস্টেম IO ইভেন্ট আমাদের পাস করতে দিতে, আমাদের একটি C এক্সটেনশন লিখতে হতো, কারণ রুবি স্ট্যান্ডার্ড লাইব্রেরিতে সেটার জন্য কোনো API নেই। সেটি গভীরে যাওয়া এই আর্টিকেলের স্কোপের বাইরে, তাই আমরা IO.select ব্যবহার করব ইভেন্ট জেনারেট করতে। IO.select একটি IO অবজেক্টের অ্যারে নেয় মনিটর করার জন্য। এটা অপেক্ষা করে যতক্ষণ না অ্যারে থেকে এক বা একাধিক অবজেক্ট রেডি হয় পড়তে বা লিখতে, এবং শুধু সেই IO অবজেক্টগুলোর একটি অ্যারে রিটার্ন করে।

কানেকশন 관련된 সব কোড একটি Fiber-এ ইমপ্লিমেন্ট করা হয়েছে: আমরা এখন থেকে এই কোডকে "হ্যান্ডলার" বলব। একটি Fiber হলো একটি কোড ব্লক যা পজ এবং রিজিউম করা যায়। রুবি VM এটা অটোমেটিক্যালি করে না, তাই আমাদের ম্যানুয়ালি রিজিউম এবং ইয়িল্ড করতে হবে। আমরা IO.select-এর ইনপুট ব্যবহার করব আমাদের হ্যান্ডলারদের জানাতে তাদের কানেকশনগুলো রেডি পড়তে বা লিখতে।

পূর্বের থ্রেডেড এবং মাল্টি-প্রসেস উদাহরণগুলোর মতো, আমাদের কিছু স্টোরেজ দরকার ক্লায়েন্ট এবং পাঠানো মেসেজ ট্র্যাক রাখতে। এইবার আমাদের Mutex দরকার নেই। আমাদের ইভেন্ট লুপ একটি সিঙ্গেল থ্রেডে রান করছে, তাই ডিফারেন্ট থ্রেড দ্বারা একই সময় অবজেক্ট মিউটেড হওয়ার রিস্ক নেই।

client_handlers = {}
messages = []

ক্লায়েন্ট হ্যান্ডলার নিচের Fiber-এ ইমপ্লিমেন্ট করা হয়েছে। যখন সকেট থেকে পড়া বা লিখা যায়, একটি ইভেন্ট ট্রিগার হয় যেন Fiber রেসপন্স দেয়। যখন স্টেট :readable হয়, সেটি সকেট থেকে একটি লাইন পড়ে এবং messages অ্যারেতে পুশ করে। যখন স্টেট :writable হয়, সেটি লাস্ট রাইটের পর থেকে অন্যান্য ক্লায়েন্ট থেকে প্রাপ্ত যেকোনো মেসেজ ক্লায়েন্টে লিখে। একটি ইভেন্ট হ্যান্ডল করার পর সেটি Fiber.yield কল করে, যাতে সেটি পজ করে এবং পরবর্তী ইভেন্টের জন্য অপেক্ষা করে।

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

তাহলে কীভাবে আমরা Fiber-কে সঠিক সময়ে রিড বা রাইট করতে ট্রিগার করব যখন Socket রেডি? আমরা একটি ইভেন্ট লুপ ব্যবহার করব যেন চারটি স্টেপ থাকে:

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

লক্ষ্য করুন এখানে কোনো ম্যাজিক নেই। এটা একটি নরমাল রুবি লুপ।

স্টেপ ১: ইনকামিং কানেকশন এক্সেপ্ট করা

দেখুন আমাদের কি কোনো নতুন ইনকামিং কানেকশন আছে। আমরা accept_nonblock ব্যবহার করব, যেটি একটি ক্লায়েন্ট কানেক্ট হওয়ার জন্য অপেক্ষা করবে না। বদলে এটা একটি এরর রেইজ করবে যদি কোনো নতুন ক্লায়েন্ট নেই, এবং যদি সেটি হয় তবে আমরা ক্যাচ করব এবং পরবর্তী স্টেপে যাব। যদি নতুন ক্লায়েন্ট থাকে, আমরা তার জন্য হ্যান্ডলার তৈরি করব এবং সেটি clients স্টোরে রাখব। আমরা সকেট অবজেক্টকে সেই Hash-এর কী হিসেবে ব্যবহার করব যাতে পরে ক্লায়েন্ট হ্যান্ডলার খুঁজে পেতে পারি।

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

স্টেপ ২: রিড বা রাইটের জন্য রেডি কানেকশন পাওয়া

পরবর্তী, আমরা OSকে অনুরোধ করব আমাদের জানাতে যখন একটি কানেকশন রেডি। আমরা client_handlers স্টোরের কীগুলো পাস করব পড়তে, লিখতে এবং এরর হ্যান্ডলিং-এর জন্য। এই কীগুলো সেটি সকেট অবজেক্ট যেগুলো আমরা স্টেপ ১-এ এক্সেপ্ট করেছিলাম। আমরা ১০ মিলিসেকেন্ড অপেক্ষা করব সেটি হয়ার জন্য।

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

স্টেপ ৩: রিডেবল কানেকশন থেকে পড়া

আমাদের কোনো কানেকশন রিডেবল হলে, আমরা ক্লায়েন্ট হ্যান্ডলার ট্রিগার করব এবং সেগুলো readable স্টেটসহ রিজিউম করব। আমরা এই ক্লায়েন্ট হ্যান্ডলার লুকআপ করতে পারি কারণ IO.select দ্বারা রিটার্ন করা Socket অবজেক্ট হ্যান্ডলার স্টোরের কী হিসেবে ব্যবহৃত হয়।

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

স্টেপ ৪: রাইটেবল কানেকশনে লিখা

আমাদের কোনো কানেকশন রাইটেবল হলে, আমরা ক্লায়েন্ট হ্যান্ডলার ট্রিগার করব এবং সেগুলো 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

এই চারটি স্টেপ একটি লুপে ব্যবহার করে যেটি হ্যান্ডলার তৈরি করে, এবং সঠিক সময়ে হ্যান্ডলারদের readable এবং writable কল করে, আমরা একটি ফুলি ফাংশনাল ইভেন্টেড চ্যাট সার্ভার তৈরি করেছি। কানেকশন প্রতি খুব কম ওভারহেড, এবং আমরা এটা বড় সংখ্যক কাঙ্কারেন্ট ক্লায়েন্ট পর্যন্ত স্কেল আপ করতে পারি।

এই অপ্রোচ খুব ভালো কাজ করে যতক্ষণ না আমরা লুপের প্রতি টিকের কাজ ছোট রাখি। এটা স্পেশিয়ালি গুরুত্বপূর্ণ কাজের জন্য যেটি ক্যালকুলেশন ইনভল্ভ করে, যেহেতু একটি ইভেন্ট লুপ একটি সিঙ্গেল থ্রেডে রান করে এবং তাই একটি সিঙ্গেল সিপিইউই ইউটিলাইজ করতে পারে। প্রোডাকশন সিস্টেমে প্রায়শই একাধিক প্রসেস থাকে একটি ইভেন্ট লুপ রান করে এই লিমিটেশন ওয়ার্কঅরাউন্ড করতে।

উপসংহার

এই সবের পর আপনি জিজ্ঞাসা করতে পারেন, এই তিনটি মেথডের কোনটি ব্যবহার করব?

• অधিকাংশ অ্যাপের জন্য, থ্রেডিং সেন্সIBLE। এটা সবচেয়ে সিম্পল অপ্রোচ কাজ করার জন্য।
• আপনি যদি হাইলি কাঙ্কারেন্ট অ্যাপ লং-রানিং স্ট্রিমসহ রান করেন, ইভেন্ট লুপ আপনাকে স্কেল করতে দেবে।
• আপনি যদি এক্সপেক্ট করেন আপনার প্রসেস ক্র্যাশ করবে, পুরনো মাল্টি-প্রসেসের দিকে যান, কারণ এটা সবচেয়ে রবাস্ট অপ্রোচ।

আমাদের কাঙ্কারেন্সি সিরিজের সমাপ্তি এখানে। আপনি যদি ফুল রিক্যাপ চান, অরিজিনাল মাস্টারিং কাঙ্কারেন্সি আর্টিকেল এবং একাধিক প্রসেস এবং একাধিক থ্রেড নিয়ে ডিটেইলড আর্টিকেলগুলো চেক করুন।