रूबी कंकरेंसी में महारत: इवेंट लूप के साथ चैट सर्वर बनाना

कंकरेंसी पर हमारी श्रृंखला में आखिरी रूबी मैजिक लेख में आपका स्वागत है। पिछले संस्करणों में हमने कई प्रोसेस और कई थ्रेड का उपयोग करके एक चैट सर्वर लागू किया था। इस बार हम उसी चीज़ को इवेंट लूप का उपयोग करके करने जा रहे हैं।

रिकैप

हम पहले के लेखों में उपयोग किए गए उसी क्लाइंट और सर्वर सेटअप का उपयोग करने जा रहे हैं। हमारा लक्ष्य एक चैट सिस्टम बनाना है जो ऐसा दिखता है:

बुनियादी सेटअप पर अधिक विवरण के लिए पिछले लेख देखें। इस लेख में उदाहरणों में उपयोग किया गया पूरा सोर्स कोड GitHub पर उपलब्ध है, ताकि आप स्वयं इसके साथ प्रयोग कर सकें।

इवेंट लूप का उपयोग करके चैट सर्वर

हमारे चैट सर्वर के लिए इवेंट लूप का उपयोग करने के लिए थ्रेड या प्रोसेस का उपयोग करने की तुलना में एक अलग मानसिक मॉडल की आवश्यकता होती है। क्लासिक दृष्टिकोण में, एक थ्रेड या प्रोसेस एक कनेक्शन को संभालने के लिए जिम्मेदार होता है। इवेंट लूप का उपयोग करके आपके पास एक ही थ्रेड में एक ही प्रोसेस होता है जो कई कनेक्शन को संभालता है। आइए देखें कि यह कैसे काम करता है इसे तोड़कर।

इवेंट लूप

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
        # सॉकेट से संदेश पढ़ें
        incoming = read_line_from(socket)
        # सब ठीक है, इसे लिखने के लिए सूची में जोड़ें
        $messages.push(
          :time => Time.now,
          :nickname => nickname,
          :text => incoming
        )
      elsif state == :writable
        # सॉकेट पर संदेश लिखें
        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
  # चरण 1: आने वाले कनेक्शन स्वीकार करें
  accept_incoming_connections
  # चरण 2: पढ़ने या लिखने के लिए तैयार कनेक्शन प्राप्त करें
  get_ready_connections
  # चरण 3: पठनीय कनेक्शन से पढ़ें
  read_from_readable_connections
  # चरण 4: लेखन योग्य कनेक्शन पर लिखें
  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
  # फिलहाल कोई नया आने वाला कनेक्शन नहीं
end

चरण 2: पढ़ने या लिखने के लिए तैयार कनेक्शन प्राप्त करें

अगला, हम OS से पूछते हैं कि कब कनेक्शन तैयार है। हम पढ़ने, लिखने और त्रुटि प्रबंधन के लिए client_handlers स्टोर की कुंजियों को पास करते हैं। ये कुंजियाँ चरण 1 में स्वीकार किए गए सॉकेट ऑब्जेक्ट हैं। हम इसके होने के लिए 10 मिलीसेकंड इंतजार करते हैं।

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

चरण 3: पठनीय कनेक्शन से पढ़ें

यदि हमारे कोई कनेक्शन पठनीय हैं, तो हम क्लाइंट हैंडलर को ट्रिगर करेंगे और उन्हें readable स्थिति के साथ फिर से शुरू करेंगे। हम इन क्लाइंट हैंडलर को देख सकते हैं क्योंकि IO.select द्वारा लौटाया गया Socket ऑब्जेक्ट हैंडलर स्टोर की कुंजी के रूप में उपयोग किया जाता है।

if readable
  readable.each do |ready_socket|
    # स्टोरेज से क्लाइंट प्राप्त करें
    client = $client_handlers[ready_socket]
    client.resume(:readable)
  end
end

चरण 4: लेखन योग्य कनेक्शन पर लिखें

यदि हमारे कोई कनेक्शन लिखने योग्य हैं, तो हम क्लाइंट हैंडलर को ट्रिगर करेंगे और उन्हें writable स्थिति के साथ फिर से शुरू करेंगे।

if writable
  writable.each do |ready_socket|
    # स्टोरेज से क्लाइंट प्राप्त करें
    client = $client_handlers[ready_socket]
    next unless client
    client.resume(:writable)
  end
end

इन चार चरणों का उपयोग करके एक लूप में जो हैंडलर बनाता है, और इन हैंडलर पर सही समय पर readable और writable कॉल करके, हमने एक पूरी तरह कार्यात्मक इवेंटेड चैट सर्वर बनाया है। प्रति कनेक्शन बहुत कम ओवरहेड है, और हम इसे बड़ी संख्या में समवर्ती क्लाइंट तक स्केल कर सकते हैं।

यह दृष्टिकोण तब तक बहुत अच्छा काम करता है जब तक हम लूप के प्रति टिक काम की मात्रा को छोटा रखते हैं। यह विशेष रूप से गणना से जुड़े काम के लिए महत्वपूर्ण है, क्योंकि एक इवेंट लूप एक ही थ्रेड में चलता है और इस प्रकार केवल एक CPU का उपयोग कर सकता है। उत्पादन प्रणालियों में अक्सर एक इवेंट लूप चलाने वाली कई प्रोसेस होती हैं ताकि इस सीमा को पार किया जा सके।

निष्कर्ष

इस सब के बाद आप पूछ सकते हैं, इन तीन विधियों में से मुझे किसका उपयोग करना चाहिए?

• अधिकांश ऐप्स के लिए, थ्रेडिंग समझ में आता है। यह काम करने का सबसे सरल दृष्टिकोण है।
• यदि आप लंबे समय तक चलने वाले स्ट्रीम के साथ अत्यधिक समवर्ती ऐप्स चलाते हैं, तो इवेंट लूप आपको स्केल करने की अनुमति देते हैं।
• यदि आप अपनी प्रोसेस के क्रैश होने की उम्मीद करते हैं, तो अच्छे पुराने मल्टी-प्रोसेस के लिए जाएं, क्योंकि यह सबसे मजबूत दृष्टिकोण है।

यह हमारी कंकरेंसी पर श्रृंखला का समापन करता है। यदि आप एक पूर्ण रिकैप चाहते हैं तो मूल कंकरेंसी लेख के साथ-साथ कई प्रोसेस और कई थ्रेड का उपयोग करने पर विस्तृत लेख देखें।