गैर-अवरुद्ध कतारें: थ्रेड्स के बीच मैसेजिंग

इस तरह के मॉड्यूल को लिखने के विचार ने हमारे कॉर्पोरेट उत्पाद के पीएलएम को जन्म दिया। हमारे डिजाइन प्रलेखन का निरीक्षण करते हुए, हमें बताया गया कि हमारे कोड को किसी भी मामले में उस कार्य को ब्लॉक नहीं करना चाहिए जिससे इसे कहा जाता है, और आमतौर पर जितना संभव हो उतना कम समय लगता है, यह प्रणाली के निर्माण की ख़ासियत है , घड़ी कुत्ते, आदि। कार्य में कुछ संदेशों को एक कार्य से दूसरे में स्थानांतरित करना शामिल है, जैसे कुछ लॉग, उन्नत निदान। प्राप्तकर्ता कार्य को फ़ाइल में परिणाम लिखने के लिए बनाया गया था, क्योंकि यह स्पष्ट है कि फ़ाइल में लिखने के बारे में मूल कार्य का कोई प्रश्न नहीं हो सकता है। और यद्यपि एक स्रोत (निर्माता) और एक उपभोक्ता है, और यहां तक ​​कि म्यूटेक्स या सेमाफोर की उपस्थिति मूल कार्य को प्रभावित नहीं करेगी, उन्हें पूरी तरह से त्यागने का निर्णय लिया गया था। फिर, भविष्य में, कार्य को कई अन्य फेरबदल के लिए विस्तारित करना संभव था, और इसलिए ऐसी स्थिति जब एक कार्य दूसरे के लिए इंतजार करता है, हालांकि यह सीमित है (और स्रोत कोड अपने सूचना संदेशों का आदान-प्रदान करने के लिए सेमाफोर का उपयोग करता है), बहुत अवांछनीय है।

प्रारंभ में, यह एक स्थिर परिपत्र बफर बनाने वाला था, जहां प्रत्येक तत्व में थोड़ा सा होता है जो यह निर्धारित करता है कि स्रोत या उपभोक्ता इसका है। एल्गोरिथ्म बेहद सरल है, स्रोत अपना डेटा उस सेल पर लिखता है जहां बिट शून्य है, और फिर परिवर्तन को "प्रकाशित" करता है, इस बिट को एक यूनिट लिखता है और अगले तत्व पर आगे बढ़ता है। दूसरी ओर, उपभोक्ता, जिसका तत्व यह बिट एक के बराबर है, संदेश पढ़ता है, और फिर बिट शून्य पर रीसेट हो जाता है। कोई दौड़ की स्थिति नहीं, सब कुछ ठीक होने लगता है। लेकिन बहुत पहले यातायात परीक्षण से पता चला कि एक स्रोत सैद्धांतिक रूप से एक स्लाइस में लगभग 30-40 हजार तत्वों का उत्पादन कर सकता है। वास्तव में, निश्चित रूप से, यह कम होगा, क्योंकि वह अभी भी कुछ करता है, इन पंक्तियों के उत्पादन को छोड़कर, लेकिन बफर के आकार को निर्धारित करना संभव नहीं है, जो पर्याप्त होगा। इसके कारणों में से एक फ़ाइल में लिखने की अस्थिर गति भी है - कुछ प्रणालियों पर हार्ड ड्राइव के बजाय सीएफ कार्ड हैं। और मैं वास्तव में संदेश खोना नहीं चाहूंगा।

इंटरनेट पर अफवाह फैलाने के बाद, मैं निम्नलिखित समाधान पर आया, जिसे मैंने अपने कार्य में लागू किया: drdobbs.com/altecture-and-design/210604448
एल्गोरिथ्म को पर्याप्त विवरण में वर्णित किया गया है; मैं इसे यहां नहीं दोहराऊंगा।
दो बदलाव मैंने किए:
1) मुझे समझ नहीं आया कि स्रोत तत्वों को क्यों जारी करता है, और उपभोक्ता नहीं। उपभोक्ता द्वारा तत्वों की रिहाई भी एक दौड़ की स्थिति नहीं बनाती है (वैसे, मैं इस वाक्यांश का अनुवाद कैसे कर सकता हूं?)। यह उपभोक्ता से लोड का हिस्सा निकाल देता है और उपयोग की गई मेमोरी को कम कर देता है, क्योंकि भस्म तत्व तुरंत हटा दिए जाते हैं।
2) इसके अलावा, यातायात परीक्षण, या बल्कि प्रोफाइलर ने खुलासा किया कि मॉलोक एक अपेक्षाकृत महंगा ऑपरेशन है। चूँकि स्रोत संदेशों का अधिकतम आकार ज्ञात है, इसलिए एक ऑपरेशन में स्मृति आवंटन को तुरंत 8 तत्वों में समूहित करने का निर्णय लिया गया था। इससे गति में दोगुनी से अधिक वृद्धि हुई, विशेष रूप से, प्रोसेसर लोड को आधा करके, जिसे हम मूल कार्य में जोड़ते हैं।

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

C # में एक नॉन-ब्लॉकिंग क्यू लागू करना

पहला कदम कतार के तत्व का वर्णन करना है।

         कक्षा क
         {
           सार्वजनिक वस्तु संदेश;
           सार्वजनिक queItem अगला;

           सार्वजनिक queItem (वस्तु संदेश = अशक्त)
           {
             this.message = message;
             अगला = अशक्त;
           }
         }

और वास्तव में खुद की बारी:

       क्लास लॉकलेस क्यू / थ्रेड-टू-थ्रेड लॉकलेस क्यू
       {
         पहले यह;
         queItem अंतिम;
       }

यहां, पहले उपभोक्ता के अंतर्गत आता है, और अंतिम तत्व में अगला स्रोत के अंतर्गत आता है। न तो पहले और न ही अंतिम शून्य होना चाहिए, इसलिए निर्माणकर्ता "पहले से ही उपभोग किए गए" राज्य में एक खाली तत्व बनाता है।

         सार्वजनिक लॉकलेस क्यू ()
         {
           पहली = नई क्वीटेम ();
           अंतिम = पहला;
         }

कतार में जोड़ने के आगे के तरीके और, तदनुसार, इससे निकालना।

         सार्वजनिक शून्य उपज (वस्तु संदेश)
         {
           last.next = new queItem (संदेश);
           last = last.next;
         }

         सार्वजनिक बूल खपत (वस्तु संदेश बाहर)
         {
           अगर (पहला == अंतिम || पहला.नेक्स्ट == अशक्त)
           {
             संदेश = अशक्त;
             झूठे लौटना;
           }
           message = first.next.message;
           first = first.next;
           सच लौटना;
         }
       }

परिणामी वर्ग अकेले अच्छी तरह से स्थापित नहीं है, क्योंकि कॉनक्यूरेक्चुए वर्ग पहले से ही 4.0 में शामिल है, जो न केवल पूरी तरह से सुरक्षित है, बल्कि परिणामी वर्ग के विपरीत, आपको एक ही समय में कतार से कई थ्रेड्स जोड़ने और निकालने की अनुमति देता है। और यह आपको अवरुद्ध विकल्प की तुलना में कतार से 1.5-3 गुना तेजी से काम करने की अनुमति देता है। pruflink
लॉग कलेक्टर के लिए, समवर्ती कतार वर्ग पर्याप्त से अधिक है। हालांकि, मैंने अपने आवेदन के लिए कार्य का विस्तार किया, और समवर्ती क्यू मुझे सूट नहीं किया, यह पता नहीं है।

इंटर-थ्रेड मैसेजिंग

प्रत्येक थ्रेड को उसके नाम से, किसी अन्य थ्रेड को संदेश भेजने में सक्षम होना चाहिए। मेरे मामले में, ये tcp सॉकेट हैंडलर (क्लाइंट या सर्वर) और स्वयं थ्रेड हैंडलर हैं। यह कैसे पता करें कि किस प्रोसेसर को भेजा जाना चाहिए, मैं इसे इस नोट के बाहर छोड़ देता हूं।

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

इसलिए, प्रत्येक प्रतिभागी धागे के लिए, आपको दो कतार बनाने की आवश्यकता है, एक में यह संदेश भेजेगा और दूसरे से पढ़ेगा। इन दो कतारों के लिए प्रॉक्सी कंटेनर:

     कक्षा धागा
     {
       सार्वजनिक स्ट्रिंग tName;
       सार्वजनिक int tid;
       locklessQueue outgoing = new locklessQueue ();  // मैसेंजर से नोड तक 
       लॉकलेस क्यू आवक = नया लॉकलेस क्यू ();  // नोड से मैसेंजर तक

       सार्वजनिक थ्रेडकोड ​​(स्ट्रिंग tName, int tid)
       {
         this.tid = tid;
         this.tName = tName;
       }

       सार्वजनिक शून्य गणना (संदेशवाहक संदेश) // जिसे नोड कहा जाता है
       {
         इनकमिंग।प्रोड्यूस (संदेश);
       }

       सार्वजनिक बूल विच्छेद (संदेशवाहक संदेश के बाहर) // नोड द्वारा कहा जाता है
       {
         वस्तु संदेश;
         bool result = outgoing.consume (out msg);
         संदेश = संदेशवाहक के रूप में संदेश;
         वापसी परिणाम;
       }

       सार्वजनिक शून्य संचारित (संदेशवाहक संदेश) // मैसेंजर द्वारा बुलाया
       {
         आउटगोइंग.प्रोड्यूस (संदेश);
       }

       सार्वजनिक बूल पुनर्प्राप्त (दूत संदेश से बाहर) // मैसेंजर द्वारा बुलाया
       {
         वस्तु संदेश;
         बूल परिणाम = इनकमिंग.कॉन्यूम (संदेश के बाहर);
         संदेश = संदेशवाहक के रूप में संदेश;
         वापसी परिणाम;
       }
     }

जैसा कि आप देख सकते हैं, टाइप मैसेंजर की एक वस्तु का प्रतिनिधित्व कतार में किया जाता है, जिसे निम्न वर्ग द्वारा दर्शाया जाता है:

     वर्ग दूत
     {
       सार्वजनिक स्ट्रिंग से;
       सार्वजनिक स्ट्रिंग को;
       सार्वजनिक वस्तु संदेश;

       सार्वजनिक संदेशवाहक (स्ट्रिंग से स्ट्रिंग, ऑब्जेक्ट संदेश)
       {
         this.from = from;
         this.to = to;
         this.message = message;
       }
     }

मैंने मैसेंजर.सेंड (...) लिखकर कोड में कहीं से भी संदेश भेजने में सक्षम होने के लिए मुख्य वर्ग को स्थिर बनाया;

   सार्वजनिक स्थिर वर्ग दूत
   {
     स्थिर शब्दकोश <int, threadNode> byTID = new myDictionary <int, threadNode> ();
     स्थिर शब्दकोश <string, threadNode> byRegName = new myDictionary <string, threadNode> ();
     स्थिर म्यूटेक्स regMutex = नया म्यूटेक्स ();  // केवल एक कार्य को एक बार में पंजीकृत करने की अनुमति है

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

     स्टेटिक बूल मैसेंजरफंक्शन ()
     {
       bool acted = false;
       संदेशवाहक आइटम;
       थ्रेडनोड डीएसटी;
       शब्दकोश <string, threadNode> tmp = byRegName;
       foreach (tmp.Values ​​में थ्रेड नोड नोड)
       {
         अगर (tmp! = byRegName)
           तोड़;
         अगर (नोड। क्रेवव (आउट आइटम))
           अगर (मद! = अशक्त)
           {
             कृतकृत्य = सत्य;
             अगर (tmp.TryGetValue (item.to, out dst))
             {
               dst.transmit (आइटम);
               भेजा = सत्य;
             }
           // और त्यागें
           }
       }
       वापसी ने अभिनय किया;
     }

संदेशवाहक में प्रवाह को दर्ज करने के लिए, निम्नलिखित फ़ंक्शन का उपयोग किया जाता है, जो वर्तमान में अवरुद्ध है:

     
     स्थिर सार्वजनिक शून्य रजिस्टर (स्ट्रिंग tName)
     {
       if (tName == null || tName == "")
         वापसी;
       int tid = Thread.CurrentThread.ManagedThreadId;
       mydirectory <int, threadNode> newbyTID = new myDictionary <int, threadNode> ();
       myDictionary <string, threadNode> newbyRegName = new myDictionary <string, threadNode> ();
       threadNode newnode = new threadNode (tName, tid);
       newbyTID.Add (tid, newnode);
       newbyRegName.Add (tName, newnode);
       regMutex.WaitOne ();
       foreach (byNID में थ्रेड नोड नोड)
       {
         newbyTID.Add (node.tid, node);
         newbyRegName.Add (node.tName, नोड);
       }
       byTID = newbyTID;
       byRegName = newbyRegName;
       regMutex.ReleaseMutex ();
     }

इसके अलावा, स्ट्रीम के अंत में डेरेगिस्टर के समान फ़ंक्शन का उपयोग किया जाता है, मैं इसके कोड को छोड़ दूंगा। वह सब कुछ बना रहता है जो धाराओं में संदेश भेजने और प्राप्त करने का कार्य है।

     स्थिर सार्वजनिक शून्य भेजें (स्ट्रिंग गंतव्य, ऑब्जेक्ट संदेश)
     {
       int tid = Thread.CurrentThread.ManagedThreadId;
       थ्रेडनोड नोड;
       अगर (byTID.TryGetValue (tid, out नोड))
         node.enqueue (नया मैसेंजर इटेम (नोड। नाम, गंतव्य, संदेश));
     }

     स्थिर सार्वजनिक बूल प्राप्त (वस्तु संदेश, स्ट्रिंग प्रेषक बाहर)
     {
       int tid = Thread.CurrentThread.ManagedThreadId;
       थ्रेडनोड नोड;
       अगर (byTID.TryGetValue (tid, out नोड))
       {
         प्रेषक = अशक्त;
         संदेश = अशक्त;
         झूठे लौटना;
       }
       अन्यथा
       {
       संदेशवाहक आइटम;
       बूल परिणाम = नोड.डीक्यू (आइटम से बाहर);
       अगर (परिणाम || आइटम == अशक्त)
       {
         प्रेषक = अशक्त;
         संदेश = अशक्त;
       }
       अन्यथा
       {
         message = item.message;
         प्रेषक = item.from;
       }
       वापसी परिणाम;
       }
     }

अंतिम फ़ंक्शन का उपयोग हैंडलर थ्रेड्स लूप में किया जाता है, संदेश को संसाधित करता है, या, यदि यह प्राप्त नहीं होता है, तो थ्रेड को निष्पादित करना। सो ()।