हम नामित पाइपों के साथ काम करने के लिए C ++ में एक क्रॉस-प्लेटफॉर्म लाइब्रेरी लिख रहे हैं

प्रागितिहास

एक बी 2 बी परियोजना को विकसित करने की प्रक्रिया में, 1 सी, ओरेकल, एमएस एसक्यूएल सर्वर जैसे विभिन्न प्रणालियों से हमारे आवेदन को एक्सेस करना आवश्यक हो गया। पहला "सार्वभौमिक" विकल्प जो दिमाग में आया, वह था वेब सेवाओं का उपयोग करना। लेकिन, दुर्भाग्य से, उपरोक्त प्रणालियों को इस मानक की थोड़ी अलग समझ है (उदाहरण के लिए, इस तथ्य को नहीं कि एक ही दैवज्ञ और 1C एक दूसरे को समझेंगे), इसके अलावा, हम भारी तीसरे पक्ष के पुस्तकालयों का उपयोग करके परियोजना को नहीं बढ़ाना चाहते थे (उस समय हम पहले से ही बढ़ावा देते थे। और Intel TBB, हम अभी भी Qt या gSoap को आकर्षित नहीं करना चाहते हैं)।
इसलिए, अपनी बाइक लिखने का निर्णय लिया गया।

उपरोक्त प्रणालियों में dll प्लगइन्स, तथाकथित बाहरी घटकों का उपयोग करके आंतरिक भाषाओं की क्षमताओं के विस्तार के लिए एक तंत्र है। बाहरी दुनिया के साथ संवाद करने के लिए उनका उपयोग करने का विचार नया नहीं है - हब पर इस विषय पर पहले से ही लेख थे:
habrahabr.ru/post/149956
habrahabr.ru/company/Centrobit/blog/165441
habrahabr.ru/post/163859
इसलिए, हमने समस्या को हल करने के लिए बाहरी घटकों का उपयोग करने का निर्णय लिया। तब हमें तय करना चाहिए कि किस तरह से, और यहां तक ​​कि अधिमानतः प्लेटफ़ॉर्म (जो कि विभिन्न प्लेटफार्मों के लिए घटक और एप्लिकेशन को सरल पुनर्संयोजन द्वारा इकट्ठा किया जा सकता है), हम घटक को हमारे आवेदन के साथ जोड़ देंगे। इस विषय पर तैयार किए गए IPC समाधान (Qt में QLocalSocket, ACE में नामांकित पाइप्स) को ढूंढना आसान नहीं होगा, लेकिन फिर, उपयोग की जाने वाली पुस्तकालयों की संख्या को कम करने के कारणों के लिए, अपनी खुद की छोटी बाइक लिखने का निर्णय लिया गया।
Boost :: interprocess :: message_queue आदर्श समाधान होगा, लेकिन यह एक साधारण कतार नहीं थी, बल्कि एक प्राथमिकता कतार थी, जिसने कुछ हद तक प्रदर्शन को कम कर दिया (उच्च प्रदर्शन की इच्छा भी एक कारण था कि हमने वेब सेवाओं से इनकार कर दिया)। इसके अलावा, हमें किसी प्रकार के क्लाइंट-सर्वर समाधान की आवश्यकता थी।
हमने तुरंत लूपबैक के माध्यम से सॉकेट्स को मना कर दिया - उनके पास खिड़कियों में बहुत अधिक ओवरहेड है (देखें habrahabr.ru/post/81067 देखें)। इसलिए, हमने यूनिक्स में नामित पाइप विंडो और डोमेन सॉकेट की ओर देखना शुरू किया।

भाग 1. हम टीके की रचना करते हैं

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

• फ़ंक्शंस को दो पैरामीटर लेना चाहिए शून्य * पॉइंटर को लिखने के लिए या जिसमें से पढ़ने के लिए और डेटा के आकार को पढ़ने / भेजने के लिए
• किसी भी त्रुटि के मामले में (उदाहरण के लिए, प्राप्त बाइट्स की संख्या प्राप्त या भेजी गई पैरामीटर के बराबर नहीं है), फ़ंक्शन को अपवाद std :: runtime_error फेंकना चाहिए (हां, अपवाद को थोड़ा कम करने के प्रदर्शन को कम करता है, त्रुटि कोड का उपयोग किया जा सकता है)
• एक वर्ग के ढांचे के भीतर, सर्वर चैनल (स्वीकृति) और क्लाइंट चैनल के साथ काम भी लागू किया जाता है। सर्वर चैनल बनाने के मामले में, पठन-लेखन कार्य नहीं करना चाहिए, लेकिन उन्हें अपवाद फेंक देना चाहिए

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

#pragma once #include <stdexcept> class INamedPipe { protected: std::string _name; bool _server; INamedPipe(){}; virtual void internalReadBytes(void* buf,size_t size)=0; virtual void internalWriteBytes(const void* buf,size_t size)=0; virtual void internalFlush()=0; public: INamedPipe(const std::string prefix, const std::string& name, bool server):_name(prefix),_server(server) { _name.append(name); } void ConnectOrOpen() { if(_server) open(); else connect(); } virtual void ReadBytes(void* buf,size_t size) { if(!_server) { if(size<1) throw std::out_of_range("Size is 0 or less"); internalReadBytes(buf,size); } else throw std::runtime_error("This operation is not supported on server pipe"); } virtual void WriteBytes(const void* buf,size_t size) { if(!_server) { if(size<1) throw std::out_of_range("Size is 0 or less"); internalWriteBytes(buf,size); 

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

भाग 2. WinNamedPipe का एक विंडोज़ कार्यान्वयन लिखना

लागू करने से पहले, हमने Google उदाहरणों का निर्णय लिया। पहला कदम था MSDN ( msdn.microsoft.com/en-us/library/windows/desktop/aa365592 , msdn.microsoft.com/en-us/library/windows/bktop/aa365588 ) और रूसी में प्रलेखन - ट्यूटोरियल । Frolov-lib.ru/books/bsp/v27/ch2_3.htm इन लेखों की मदद से, WinNamedPipe वर्ग का पहला संस्करण लिखा गया था। कुछ समय बाद, हमने मेटाट्रेडर ( www.mql5.com/en/articles/115 ) के लिए एक बाहरी घटक बनाने के उदाहरण का उपयोग करके पाइप के साथ काम करने के लिए एक अच्छा लेख लिखा, जिसे हम पढ़ने की सलाह भी देते हैं।
WinNamedPipe कंस्ट्रक्टर में एक उपसर्ग पहले से ही सेट किया गया है - MSDN आवश्यकताओं के अनुसार, नामित पाइप का पूर्ण पथ "\\ computer_name \ पाइप \ channel_name" के रूप में होना चाहिए और स्थानीय मशीन पर नामांकित पाइप के साथ काम करने के मामले में, यह "\\। \ पाइप \ पाइप" (हाँ) के रूप में होना चाहिए। , विंडोज में पाइप नाम से आपको नेटवर्क पर जानकारी का आदान-प्रदान करने की अनुमति मिलती है)।
क्लास में एक कंस्ट्रक्टर भी होता है जो टाइप HANDLE का इनपुट लेता है - एक ऐसा नंबर जो ऑपरेटिंग सिस्टम में कुछ स्ट्रक्चर या ऑब्जेक्ट की पहचान करता है (इस कंस्ट्रक्टर का उपयोग करके हम WinNamedPipe क्लास का एक नया उदाहरण बना सकते हैं, जो किसी मौजूदा नाम वाले पाइप के HANDLE को ही जानता है)। इस निर्माता का उपयोग WaitForConnection () फ़ंक्शन में किया जाएगा (नीचे देखें)
खुला () कार्यान्वयन निम्नानुसार है:

 void WinNamedPipe::open(){ _hPipe = CreateNamedPipe( (LPSTR)_name.data(), //    PIPE_ACCESS_DUPLEX, //       PIPE_TYPE_BYTE | //  PIPE_WAIT, //   PIPE_UNLIMITED_INSTANCES, //     BUFFER_PIPE_SIZE, //     BUFFER_PIPE_SIZE, //     0, // -  (0=) NULL); //     –    if (_hPipe == INVALID_HANDLE_VALUE) { THROW_LAST_ERROR("CreateNamedPipe failed"); } 

ध्यान दें कि हमने फ़ाइल के आरंभ में बफर आकार BUFFER_PIPE_SIZE निर्धारित किया है। वास्तव में, इसका आकार संचारित डेटा के अधिकतम आकार को प्रभावित नहीं करता है - हमारी परियोजना में, हमने सफलतापूर्वक स्थानांतरित किए गए टुकड़े जो कि बफर से कई गुना बड़े थे।
विशेष रूप से पहले पैरामीटर पर ध्यान दिया जाना चाहिए: (LPSTR) _name.data ()
चूँकि हमारी कक्षा में चैनल का नाम std :: string में संग्रहीत है, WinAPI में इसके सही स्थानांतरण के लिए, प्रोजेक्ट को प्रॉपर्टीज में विज़ुअल स्टूडियो "मल्टी-बाइट कैरेक्टर सेट का उपयोग करें" में सेट किया जाना चाहिए (अधिक जानकारी के लिए habrahabr.ru/ostost/164193 देखें) ।
CreateNamedPipe फ़ंक्शन मापदंडों का विस्तृत विवरण MSDN ( msdn.microsoft.com/en-us/library/windows/desktop/aa365150 और www.frolov-lib.ru/books/bsp/v27/ch2_3.htm ) पर पाया जा सकता है।
चैनल का क्लाइंट कनेक्शन कनेक्ट () फ़ंक्शन का उपयोग करके किया जाता है, जिसमें निम्नलिखित कार्यान्वयन हैं:

 void WinNamedPipe::connect() for(;;) { WaitNamedPipe((LPSTR)_name.data(), NMPWAIT_WAIT_FOREVER); _hPipe = CreateFile( (LPSTR)_name.data(), //   GENERIC_READ | //      GENERIC_WRITE, 0, NULL, //      OPEN_EXISTING, //    0, NULL); // Break if the pipe handle is valid or error!=232 if (_hPipe != INVALID_HANDLE_VALUE||GetLastError() != ERROR_PIPE_BUSY) break; } if (_hPipe == INVALID_HANDLE_VALUE) THROW_LAST_ERROR("Could not open pipe"); DWORD dwMode = PIPE_TYPE_BYTE; BOOL fSuccess = SetNamedPipeHandleState( _hPipe, //     &dwMode, //  - NULL, //       NULL); //    if ( ! fSuccess) { THROW_LAST_ERROR("SetNamedPipeHandleState failed"); } } 

सबसे पहले, WaitNamedPipe फ़ंक्शन का उपयोग करके, हम "अनन्त" सर्वर चैनल के एक निशुल्क उदाहरण के लिए प्रतीक्षा करते हैं, फिर CreateFile का उपयोग करके हम इसे कनेक्ट करते हैं। उसके बाद, हम जांचें कि क्या कनेक्शन सफल था। यदि चैनल ऑब्जेक्ट का पॉइंटर वैध है (या ERROR_PIPE_BUSY के अलावा कोई त्रुटि हुई है - यह इंगित करने में त्रुटि है कि कोई निशुल्क चैनल नहीं है), हम लूप से बाहर निकलते हैं और चैनल के साथ ऑपरेशन के वांछित मोड को कॉन्फ़िगर करते हैं। अन्यथा, यह चक्र तब तक दोहराता है जब तक कि यह सामान्य रूप से जुड़ नहीं जाता है या ERROR_PIPE_BUSY के अलावा कोई त्रुटि होती है। यदि दिए गए नाम के साथ पाइप नाम का सर्वर मौजूद नहीं है, तो WaitNamedPipe तुरंत समाप्त हो जाएगा, GetLastError त्रुटि कोड 2 लौटा देगा (सिस्टम निर्दिष्ट फ़ाइल नहीं ढूँढ सकता) और कनेक्ट () एक अपवाद को फेंक देगा।
हमारे विचार के अनुसार, WaitForConnection () को एक नया WinNamedPipe के लिए एक पॉइंटर लौटना चाहिए, जिससे क्लाइंट पहले से जुड़ा हुआ है और जिस पर रीड-राइट फ़ंक्शन को लागू किया जा सकता है। हालाँकि, विंडोज में, क्लाइंट प्राप्त करने के लिए मानक तंत्र को एक अलग तरीके से लागू किया जाता है - कनेक्टनैमपाइप फ़ंक्शन (बनाया गया चैनल हैन्डल) बस स्ट्रीम को ब्लॉक करता है जब तक कि क्लाइंट चैनल के दूसरे छोर पर दिखाई न दे। फिर, अन्य ग्राहकों को न खोने के लिए, आपको एक नया श्रवण चैनल बनाने और इस फ़ंक्शन को पास करने की आवश्यकता है, आदि। इसलिए, हमारे विचार को लागू करने के लिए, हमें फिर से WaitForConnection से ओपन () कॉल करना होगा।
WaitForConnection () फ़ंक्शन निम्नानुसार है:

 WinNamedPipe* WinNamedPipe::WaitForConnection() { if(_server) { DWORD error; if (ConnectNamedPipe(_hPipe, NULL)||(error=GetLastError())==ERROR_PIPE_CONNECTED) { HANDLE client=_hPipe; open(); return new WinNamedPipe(client); } else { THROW_LAST_ERROR("WaitForConnection failed"); } } else { throw std::runtime_error("WaitForConnection is not supported on server pipe\n"); } } 

हम एक महत्वपूर्ण बिंदु पर ध्यान देते हैं - यदि बहुत सारे क्लाइंट चैनल हैं, तो एक स्थिति संभव है जब क्लाइंट कनेक्टनेमपाइप को कॉल करने से पहले ही चैनल से कनेक्ट होता है। इस स्थिति में, GetLastError त्रुटि कोड 535 वापस कर देगा (लेकिन यह एक त्रुटि नहीं होगी)
अतिरिक्त टिप्पणियों के बिना पढ़ने और लिखने के कार्य सरल और सीधे दोनों हैं:

 void WinNamedPipe::internalReadBytes(void* buf,size_t size) { DWORD cbBytesRead = 0; BOOL fSuccess = FALSE; //     fSuccess = ReadFile( _hPipe, //    buf, // -   size, //   &cbBytesRead, //   ,       NULL); //     if (!fSuccess || cbBytesRead == 0 ||cbBytesRead!=size) { if (GetLastError() == ERROR_BROKEN_PIPE) { THROW_LAST_ERROR("pipe disconnected"); } else { THROW_LAST_ERROR("read failed"); } } } void WinNamedPipe::internalWriteBytes(const void* buf,size_t size) { DWORD cbWritten; BOOL fSuccess = FALSE; fSuccess = WriteFile( _hPipe, //    buf, //        size, //     &cbWritten, //         NULL); //     if (!fSuccess || size != cbWritten) { THROW_LAST_ERROR("WriteFile failed"); } } 

बंद () फ़ंक्शन पर अधिक विस्तार से चर्चा की जानी चाहिए। सर्वर चैनल को बंद करने और इसके बारे में इसके दूसरे छोर पर क्लाइंट को सूचित करने के लिए, डिस्कनेक्टनामाइडपाइप (हैन्डल पाइप) का उपयोग करें। उसके बाद, आप उस पर नए ग्राहकों से फिर से उम्मीद कर सकते हैं, या आप ऑपरेटिंग सिस्टम को बता सकते हैं कि इसे बंदहैंडल (हैन्डल पाइप) का उपयोग करने की आवश्यकता नहीं है। CloseHandle को क्लाइंट चैनल पर भी कॉल किया जा सकता है, हालांकि, इस विधि को केवल एक बार कॉल किया जाना चाहिए - या तो एक कनेक्टेड क्लाइंट के साथ सर्वर पाइप पर, या क्लाइंट पर। इसलिए, यह निर्णय लिया गया कि हम केवल चैनल के सर्वर-साइड इंस्टेंस पर DisconnectNamedPipe और CloseHandle को कॉल करेंगे, और क्लाइंट क्लोज़ पर हम एक डमी विधि होगी।

 void WinNamedPipe::Close() { if(_server||_server_with_client) { DisconnectNamedPipe(_hPipe); CloseHandle(_hPipe); //May throw an exception if a debugger is attached to the process and handle is not valid } } 

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

 #define THROW_LAST_ERROR(e){ \ int error=GetLastError(); \ std::stringstream err; \ err<<e<<", GLE="<<error; \ throw std::runtime_error(err.str().data()); \ } 

भाग 3. एक यूनिक्स कार्यान्वयन लिखना - PosixNamedPipe

लाइब्रेरी के पॉज़िक्स-संगत हिस्से को लागू करने के लिए, हमने यूनिक्स डोमेन सॉकेट (यूनिक्स में सॉकेट्स के साथ काम करने के बारे में और अधिक जानकारी यहाँ पाई जा सकती है - wiki.linuxformat.ru/index.php/LXF83 : Unix_API) का उपयोग करने का निर्णय लिया।
खिड़कियों के विपरीत, एक स्थानीय सॉकेट के लिए एक पॉइंटर फ़ाइल फ़ाइल सिस्टम में कहीं भी स्थित हो सकती है। हमने पथ / tmp / एक उपसर्ग के रूप में पारित करने का निर्णय लिया। यह ध्यान दिया जाना चाहिए कि यदि एक खाली स्ट्रिंग को उपसर्ग के रूप में पारित किया जाता है, तो क्लाइंट सॉकेट से कनेक्ट नहीं हो पाएगा और एक फाइल लौटेगा जिसमें त्रुटि नहीं मिली। इसके अलावा, अगर स्थानीय सॉकेट के नाम वाली फ़ाइल स्थानीय सॉकेट बनाने वाले फ़ंक्शन से पहले मौजूद है, तो कोई भी क्लाइंट सर्वर से कनेक्ट नहीं कर पाएगा (कनेक्शन ने त्रुटि को अस्वीकार कर दिया)। इसलिए, सॉकेट बनाने से पहले, आपको इस फ़ाइल के अस्तित्व की जांच करनी चाहिए और इसे हटाने का प्रयास करना चाहिए। यहां से एक निश्चित पथ उपसर्ग के साथ विचार आया - किसी भी तरह से संभावित भेद्यता को समाप्त करना आवश्यक है जो सिस्टम में कहीं भी फ़ाइल को अधिलेखित करना संभव बनाता है (और एक्सेस अधिकार का उपयोग करना और बेसनेम फ़ंक्शन का उपयोग करना अच्छा होगा)। यह भी इस तथ्य का तात्पर्य है कि एक ही नाम के साथ PosixNamedPipe वर्ग के एक से अधिक सर्वर-साइड इंस्टेंसेस नहीं होने चाहिए - सही संचालन के लिए, प्रत्येक नाम में कक्षा का केवल एक उदाहरण होना चाहिए।
खुले () फ़ंक्शन के लिए कोड निम्नानुसार है:

 void PosixNamedPipe::open() { sock= socket(AF_UNIX, SOCK_STREAM, 0); if(sock == -1) { THROW_ERROR("Create_socket failed: "); } unlink(_name.c_str()); desc.sun_family = AF_UNIX; strcpy(desc.sun_path, _name.c_str()); if (bind(sock, (sockaddr*)&desc, sizeof(struct sockaddr_un)) == -1) { THROW_ERROR("Connection failed(bind): "); } if (listen(sock,SOMAXCONN) == -1) { THROW_ERROR("Connection failed(listen): "); } } 

कनेक्ट () फ़ंक्शन लगभग समान है:

 void PosixNamedPipe::connect() { sock= socket(AF_UNIX, SOCK_STREAM, 0); if(sock == -1) { THROW_ERROR("Create_socket failed: "); } desc.sun_family = AF_UNIX; strcpy(desc.sun_path, _name.c_str()); if (::connect(sock, (sockaddr*)&desc, sizeof(struct sockaddr_un)) == -1) { THROW_ERROR("Connection failed(connect): "); } } 

हमारे TK के अनुसार WaitForConnection () लिखने के लिए, कोड के विंडोज़ संस्करण के विपरीत, हमें यहाँ भी बिगाड़ना नहीं चाहिए।

 PosixNamedPipe* PosixNamedPipe::WaitForConnection() { int client=accept(sock,NULL,NULL); if(client!=-1) return new PosixNamedPipe(client); else { THROW_ERROR("Accept error: "); } } 

इसमें कहा जाने वाला कंस्ट्रक्टर भी आम है:

 PosixNamedPipe::PosixNamedPipe(int pipe) { sock=pipe; _server=false; memset(&desc, 0, sizeof(struct sockaddr_un)); } 

यह ध्यान दिया जाना चाहिए कि बाइंड और :: कनेक्ट फ़ंक्शंस को सही ढंग से काम करने के लिए, यह अपने खेतों को शुरू करने से पहले sockaddr_un संरचना के उदाहरण को शून्य करने की सिफारिश की जाती है, और इसलिए क्लास के कंस्ट्रक्टर्स के लिए मेमसेट लाइन (& desc, 0, sizeof (स्ट्रक्चर sockaddr_un) को जोड़ा गया था।
सॉकेट के लिए पढ़ने / लिखने के कार्य भी काफी सरल हैं:

 void PosixNamedPipe::internalReadBytes(void* buf,size_t size) { size_t ret=-1; if ((recv(sock, buf, size, MSG_WAITALL)) == -1) { THROW_ERROR("Error while reading: "); } } void PosixNamedPipe::internalWriteBytes(const void* buf,size_t size) { size_t ret=-1; if ((ret = send(sock, buf, size, 0)) == -1||ret!=size) { THROW_ERROR("Error while sending: "); } } 

चैनल क्लोजर फ़ंक्शन में, बस मामले में, जब क्लास का सर्वर इंस्टेंस बंद हो जाता है, तो फ़ाइल को हटाना संभव है (और क्लोजर फ़ंक्शन को स्वचालित रूप से डिस्ट्रक्टर में भी कहा जाता है):

 void PosixNamedPipe::Close() { if(_server) unlink(desc.sun_path); close(sock); } PosixNamedPipe::~PosixNamedPipe() { this->Close(); } 

भाग 4. हम प्रसंस्करण ग्राहकों के लिए एक बहुस्तरीय सर्वर लिखते हैं

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

 //NamedPipe.h #pragma once #ifdef _WIN32 #include "WinNamedPipe.h" typedef WinNamedPipe NamedPipe; #else #include <unistd.h> #include "PosixNamedPipe.h" typedef PosixNamedPipe NamedPipe; #endif 

अब किसी भी प्रोजेक्ट में हम "NamedPipe.h" #include बनाते हैं और हम कोड का नाम NamedPipe * क्लाइंट = नया NamedPipe ("NamedPipe.h", 0) लिख सकते हैं, बिना इस बात की परवाह किए कि हमारा एप्लिकेशन किस ऑपरेटिंग सिस्टम के तहत बनाया जाएगा।
हमारी परियोजना में बाहरी घटकों से आने वाले अनुरोधों को संसाधित करने के लिए, हमने थ्रेड पूल पैटर्न के आधार पर एक साधारण बहु-थ्रेडेड सर्वर क्लास को बढ़ावा दिया है :: थ्रेड और इंटेल टीबीबी लाइब्रेरी का उपयोग करके। मुख्य विचार यह था: एक एकल थ्रेड में, एक WaitForConnection () कॉल एक अंतहीन लूप में किया जाता है, और कनेक्ट किए गए क्लाइंट के साथ NamedPipe ऑब्जेक्ट्स के लिए पॉइंटर्स को tbb :: concurrent_queue क्यू में जोड़ा जाता है, जिसमें से स्ट्रीम सीधे पढ़ने और लिखने और आने वाले संदेशों को संसाधित करते हैं और उन्हें इकट्ठा करते हैं।
फ़ंक्शन के लिए कोड जो आने वाले कनेक्शन को स्वीकार करता है वह इस तरह दिखता है:

 void NamedPipeServer::run() { NamedPipe* client; _pipe=new NamedPipe(_name, 1); try { _pipe->ConnectOrOpen(); } catch(const std::exception& e) { } while(_active) { try { client=_pipe->WaitForConnection(); _clients.push(client); } catch(const std::exception& e) { } } delete _pipe; } 

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

 void NamedPipeServer::workerProc() { NamedPipe* client; while(_active) if(_clients.try_pop(client)) { handleClient(client); } else boost::this_thread::sleep(boost::posix_time::milliseconds(100)); } 

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

 void NamedPipeServer::Start() { _active=true; startWorkers(); } void NamedPipeServer::startWorkers() { for (size_t i = 0; i < _thread_pool_size; ++i) { boost::shared_ptr<boost::thread> thread(new boost::thread(boost::bind(&NamedPipeServer::workerProc, this))); _threads.push_back(thread); } boost::shared_ptr<boost::thread> dispatcher(new boost::thread(boost::bind(&NamedPipeServer::run,this))); _threads.push_back(dispatcher); } 

ध्यान दें कि प्रारंभ () विधि निष्पादन थ्रेड को ब्लॉक नहीं करती है। सभी सर्वर थ्रेड के पूर्ण होने तक इसे ब्लॉक करने के लिए, JoinWorkers विधि का उपयोग करें:

 void NamedPipeServer::JoinWorkers() { size_t size=_threads.size(); for (std::size_t i = 0; i < size; ++i) _threads[i]->join(); for (std::size_t i = 0; i < size; ++i) _threads[i].reset(); _threads.clear(); } 

व्यवहार में, हमारे आवेदन में, यह माना जाता है कि अनुप्रयोग बंद होने तक चलने वाले सर्वर काम करते हैं। इसने नामांकित पाइप सर्वर थ्रेड्स को रोकने की कोशिश करने की परेशानी से हमें बचाया।
सर्वर थ्रेड्स को रोकने का प्रयास करने के लिए, आप लिख सकते हैं, उदाहरण के लिए, स्टॉप () विधि:

 void NamedPipeServer::Stop() { _active=false; this->JoinWorkers(); } 

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

 WinNamedPipe* WinNamedPipe::WaitForConnection(unsigned int timeout) { if(_server) { OVERLAPPED lpOverlapped = {0}; lpOverlapped.hEvent = CreateEvent(0,1,1,0); if(ConnectNamedPipe(_hPipe, &lpOverlapped)==0) { if(GetLastError()==ERROR_PIPE_CONNECTED) if (!SetEvent(lpOverlapped.hEvent)) THROW_LAST_ERROR("AsyncWaitForConnection failed"); int result = WaitForSingleObject(lpOverlapped.hEvent,timeout); if (WAIT_OBJECT_0 == result) { HANDLE client=_hPipe; open(); return new WinNamedPipe(client); } else { return NULL; } } else { THROW_LAST_ERROR("AsyncWaitForConnection failed"); } } else { throw std::runtime_error("WaitForConnection is not supported on client pipe\n"); } } 

इस विधि को सही तरीके से काम करने के लिए, आपको कॉल को CreateNamedPipe फ़ंक्शन को खुले में बदलना होगा () इस प्रकार है:

 _hPipe = CreateNamedPipe( (LPSTR)_name.data(), PIPE_ACCESS_DUPLEX | FILE_FLAG_OVERLAPPED, //        PIPE_TYPE_BYTE | PIPE_WAIT, PIPE_UNLIMITED_INSTANCES, BUFFER_PIPE_SIZE, BUFFER_PIPE_SIZE, 0, NULL); // default security attribute 

लिनक्स के लिए कार्यान्वयन इस तरह दिखता है:

 PosixNamedPipe* PosixNamedPipe::WaitForConnection(unsigned int timeout) { int nsock; int retour; fd_set readf; fd_set writef; struct timeval to; FD_ZERO(&readf); FD_ZERO(&writef); FD_SET(sock, &readf); FD_SET(sock, &writef); to.tv_usec = timeout*1000; retour = select(sock+1, &readf, &writef, 0, &to); if (retour == 0) { return NULL; } if ( (FD_ISSET(sock, &readf)) || (FD_ISSET(sock,&writef))) { nsock = accept(sock, NULL, NULL); return new PosixNamedPipe(nsock); } else { throw std::runtime_error("invalid socket descriptor!\n"); } } 

अंत में, हम नोट करते हैं कि सर्वर क्लास डिस्ट्रक्टर को सही तरीके से काम करने के लिए स्टॉप () विधि को कॉल करना चाहिए और फिर पूरे _clients की कतार को साफ़ करना चाहिए अगर वहाँ अनचाहे क्लाइंट हैं। चूंकि सर्वर में क्लाइंट कतार में संकेत होते हैं, इसलिए इसे मैन्युअल रूप से साफ़ करना होगा:

 NamedPipeServer::~NamedPipeServer(void) { this->Stop(); while(!_clients.empty()) { if(_clients.try_pop(_pipe)) delete _pipe; } } 

भाग 5. पुस्तकालय का उपयोग करने का एक उदाहरण।

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

 void spamming_thread() { std::vector<std::string> words; words.push_back(std::string("one ")); words.push_back(std::string("two ")); words.push_back(std::string("three ")); words.push_back(std::string("four ")); words.push_back(std::string("five ")); words.push_back(std::string("six ")); words.push_back(std::string("seven ")); words.push_back(std::string("eight ")); words.push_back(std::string("nine ")); words.push_back(std::string("ten ")); NamedPipe client("NamedPipeTester",0); try { client.ConnectOrOpen(); for(int i=0;i<words.size();++i) { std::cout<<"sending "<<words[i]<<"\n"; size_t size=words[i].size(); client.WriteBytes(&size,sizeof(size)); client.WriteBytes(words[i].data(),size); } client.Close(); } catch(const std::runtime_error &e) { std::cout<<"Exception: "<<e.what()<<"\n"; } } 

अब एक सर्वर लिखते हैं जो कनेक्शन स्वीकार करता है:

 #pragma once #include "../NamedPipeServer.h" class SimpleServer: public NamedPipeServer { protected: void handleClient(NamedPipe* client); public: SimpleServer(const std::string &name):NamedPipeServer(name,1){}; ~SimpleServer(){}; }; void SimpleServer::handleClient(NamedPipe* client) { for(int i=0;i<10;++i) { try { size_t size; client->ReadBytes(&size,sizeof(size)); if(size>0) { char* message=new char[size]; client->ReadBytes(message,size); //Using std::cout is bad in multi-threading apps //std::string msg(message,size); //std::cout<<"Message from pipe: "<<msg<<"\n"; delete[] message; } } catch(const std::exception& e) { std::cout<<"Exception!:"<<e.what()<<"\n"; } } client->Close(); } 

इसे चलाएं, आधा मिनट रुकें और रोकने की कोशिश करें:

  SimpleServer* s=new SimpleServer("NamedPipeTester"); s->Start(); boost::this_thread::sleep(boost::posix_time::milliseconds(30000)); delete s; system("pause"); 

निष्कर्ष।

हमारी परियोजना में छह महीने के लिए इस पुस्तकालय का सफलतापूर्वक उपयोग किया गया है। एक पुस्तकालय के लिए पुस्तकालय के कोड को सही और पूरक बनाया गया है। फिर भी, चूंकि लेख नौसिखिए प्रोग्रामर द्वारा लिखा गया था, इसलिए कोड में त्रुटियां संभव हैं। लेखक रचनात्मक आलोचना, टिप्पणियों और सुधार के लिए सुझावों और मिली त्रुटियों के लिए आभारी होंगे।
यह लेख व्यापक होने का इरादा नहीं है। यह ऐसी चीजों को गैर-अवरुद्ध पढ़ना और लिखना नहीं मानता है, नामित पाइपों के लिए एक्सेस अधिकार सेट करता है और बाइट रीड-राइट मोड / मैसेजिंग मोड का विकल्प है।
आप लेख के लिए सभी स्रोत कोड github.com/xpavlov/libNamedPipe पर पा सकते हैं