समवर्ती डीप डाइव:इवेंट लूप्स

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

रिकैप

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

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

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

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

इवेंट लूप

उदाहरण के लिए EventMachine या NodeJS द्वारा उपयोग किया जाने वाला इवेंट लूप निम्नानुसार काम करता है। हम कुछ घटनाओं में रुचि रखने वाले ऑपरेटिंग सिस्टम को सूचित करने के साथ शुरू करते हैं। उदाहरण के लिए, जब सॉकेट से कनेक्शन खोला जाता है। हम ऐसा फ़ंक्शन को कॉल करके करते हैं जो किसी IO ऑब्जेक्ट, जैसे कनेक्शन या सॉकेट पर रुचि दर्ज करता है।

जब इस IO ऑब्जेक्ट पर कुछ होता है, तो ऑपरेटिंग सिस्टम हमारे प्रोग्राम को एक ईवेंट भेजता है। हम इन घटनाओं को एक कतार में रखते हैं। ईवेंट लूप सूची से ईवेंट को पॉप अप करता रहता है और उन्हें एक-एक करके हैंडल करता है।

एक मायने में एक घटना पाश वास्तव में समवर्ती नहीं है। यह प्रभाव को अनुकरण करने के लिए बहुत छोटे बैचों में क्रमिक रूप से काम करता है।

रुचि दर्ज करने के लिए और ऑपरेटिंग सिस्टम को आईओ इवेंट पास करने के लिए हमें एक सी एक्सटेंशन लिखना होगा, क्योंकि रूबी मानक पुस्तकालय में इसके लिए कोई एपीआई मौजूद नहीं है। इसमें गोता लगाना इस लेख के दायरे से बाहर है, इसलिए हम 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:ऐसे कनेक्शन प्राप्त करें जो पढ़ने या लिखने के लिए तैयार हों

इसके बाद, हम ओएस से कनेक्शन तैयार होने पर हमें सूचित करने के लिए कहते हैं। हम 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 इन हैंडलर्स पर सही समय पर, हमने पूरी तरह कार्यात्मक इवेंट चैट सर्वर बनाया है। प्रति कनेक्शन बहुत कम ओवरहेड है, और हम इसे बड़ी संख्या में समवर्ती क्लाइंट तक बढ़ा सकते हैं।

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

समाप्त हो रहा है

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

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

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