কনকারেন্সি ডিপ ডাইভ:ইভেন্ট লুপ

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

রিক্যাপ

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

বেসিক সেটআপ সম্পর্কে আরও বিস্তারিত জানার জন্য অনুগ্রহ করে পূর্ববর্তী নিবন্ধগুলি দেখুন। এই নিবন্ধের উদাহরণগুলিতে ব্যবহৃত সম্পূর্ণ উত্স কোডটি GitHub-এ উপলব্ধ, তাই আপনি নিজেই এটি পরীক্ষা করতে পারেন৷

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

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

ইভেন্ট লুপ

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

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

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

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

যে কোডটি একটি সংযোগের সাথে সম্পর্কিত সমস্ত কিছুর যত্ন নেয় সেটি একটি Fiber হিসাবে প্রয়োগ করা হয় :আমরা এখন থেকে এই কোডটিকে "হ্যান্ডলার" বলব। একটি Fiber একটি কোড ব্লক যা বিরতি এবং পুনরায় চালু করা যেতে পারে। রুবি ভিএম স্বয়ংক্রিয়ভাবে এটি করে না, তাই আমাদের পুনরায় শুরু করতে হবে এবং ম্যানুয়ালি ফলন করতে হবে। আমরা 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

লক্ষ্য করুন এখানে কোন জাদু নেই। এটি একটি সাধারণ রুবি লুপ৷

ধাপ 1:ইনকামিং সংযোগ গ্রহণ করুন

আমাদের কোন নতুন ইনকামিং সংযোগ আছে কিনা দেখুন. আমরা 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

ধাপ 2:পড়ার বা লেখার জন্য প্রস্তুত সংযোগগুলি পান

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

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

ধাপ 3:পঠনযোগ্য সংযোগ থেকে পড়ুন

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

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

পদক্ষেপ 4:লেখার যোগ্য সংযোগগুলিতে লিখুন

যদি আমাদের কোনো সংযোগ লেখার যোগ্য হয়, আমরা ক্লায়েন্ট হ্যান্ডলারদের ট্রিগার করব এবং একটি 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 সঠিক সময়ে এই হ্যান্ডলারগুলিতে, আমরা একটি সম্পূর্ণ কার্যকরী ইভেন্টেড চ্যাট সার্ভার তৈরি করেছি। সংযোগ প্রতি খুব কম ওভারহেড আছে, এবং আমরা এটিকে অনেক সমসাময়িক ক্লায়েন্ট পর্যন্ত স্কেল করতে পারি।

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

সমাপ্তিতে

এত কিছুর পর আপনি হয়তো জিজ্ঞেস করতে পারেন, এই তিনটি পদ্ধতির মধ্যে কোনটি ব্যবহার করব?

• বেশিরভাগ অ্যাপের জন্য, থ্রেডিং অর্থপূর্ণ। এটি কাজ করার সবচেয়ে সহজ পদ্ধতি।
• যদি আপনি দীর্ঘ-চলমান স্ট্রিমগুলির সাথে উচ্চ সমসাময়িক অ্যাপগুলি চালান, ইভেন্ট লুপগুলি আপনাকে স্কেল করার অনুমতি দেয়৷
• আপনি যদি আপনার প্রক্রিয়াগুলি ক্র্যাশ হওয়ার আশা করেন, তাহলে ভাল পুরানো মাল্টি-প্রসেসে যান, কারণ এটি সবচেয়ে শক্তিশালী পদ্ধতি।

এটি একযোগে আমাদের সিরিজের সমাপ্তি ঘটায়। আপনি যদি একটি সম্পূর্ণ রিক্যাপ চান তবে মূল মাস্টারিং কনকারেন্সি নিবন্ধের পাশাপাশি একাধিক প্রক্রিয়া এবং একাধিক থ্রেড ব্যবহার করার বিষয়ে বিস্তারিত নিবন্ধগুলি দেখুন৷