क्या कनेक्शन.send() ब्लॉक बना सकता है? (conn1 से, conn2 = multiprocessing.Pipe())

Question

मैं डिबगिंग एप्लिकेशन हूं जो 2 सेंसर से जानकारी एकत्र करता है: एक वेबकैम और एक माइक्रोफोन।

सामान्य वास्तुकला काफी सरल है:

• मुख्य प्रक्रिया बच्चे प्रक्रियाओं (प्रत्येक के लिए एक) के लिए पाइप के माध्यम से संदेश (, शुरू रोक, get_data) भेजता है।
• बच्चे प्रक्रियाओं डेटा इकट्ठा और (उपयोगकर्ता, मुख्य प्रक्रिया से बच्चे की प्रक्रिया से मुख्य प्रक्रिया) आदेशों पर कार्रवाई करने के मुख्य प्रक्रिया

बाल & मुख्य प्रक्रियाओं अनंत लूप में हैं को भेजें।

यह वैश्विक रूप से काम करता है लेकिन मुझे बाल प्रक्रियाओं को रोकने में परेशानी है।

मैं कोड लॉग इन किया है और यह 2 बातें होती हैं लगता है:

1. 'स्टॉप' संदेश भेजा है, लेकिन पाइप के माध्यम से नहीं मिलता है।
2. बाल प्रक्रिया डेटा और conn.send (डेटा) ब्लॉक भेजना जारी रखती है।

व्यवहार स्पष्ट रूप से कनेक्शन की स्थिति से जुड़ा हुआ है, क्योंकि बाल प्रक्रियाएं जो कुछ भी नहीं भेजती हैं, इस व्यवहार में नहीं है। फिर भी, मैं नहीं देखता कि वर्तमान आर्किटेक्चर को डीबग/संशोधित कैसे करें जो तर्कसंगत लगता है।

तो, इस अवरोधक व्यवहार और इससे कैसे बचें इसका कारण क्या है?

def do(self): 
     while self.cnx.poll(): 
      msg = self.cnx.recv() 
      self.queue.append(msg) 
     #== 
     if not self.queue: 
      func_name = 'default_action' 
      self.queue.append([func_name, ]) 
     #== 
     msg    = self.queue.pop() 
     func_name, args = msg[0], msg[1:] 
     #== 
     res = self.target.__getattribute__(func_name)(*args) 
     #== 
     running = func_name != 'stop' 
     #== 
     if res and self.send: 
      assert running 
      self.output_queue.append(res[0]) 
     if self.output_queue and running: 
      self.cnx.send(self.output_queue.popleft()) 
     #== 
     return running 

अद्यतन:

इस कोड को जो बच्चे प्रक्रिया में अनंत लूप के प्रत्येक यात्रा के लिए मार डाला है ऐसा लगता है कि पाइप दोनों छोर पर एक साथ नहीं लिखा जा सकता है। यह काम करता है, तो करने के लिए ऊपर दिए गए कोड के अंतिम कुछ पंक्तियों में परिवर्तन:

 if self.output_queue and running: 
      if not self.cnx.poll(): 
       self.cnx.send(self.output_queue.popleft()) 

सवाल खुला हालांकि के रूप में पाइप डिफ़ॉल्ट रूप से पूर्ण द्वैध के रूप में प्रलेखित हैं रहता है और इस व्यवहार बिल्कुल दर्ज नहीं किया गया है। मैंने कुछ गलत समझा होगा। कृपया मुझे उजागर करें!

अपडेट 2: बस स्पष्ट होने के लिए, इस स्थिति में कोई कनेक्शन बंद नहीं है। घटनाओं के अनुक्रम का वर्णन करने के लिए:

• मुख्य प्रक्रिया एक पृष्ठभूमि में ("रोक") भेजता
• मुख्य प्रक्रिया एक (अनंत) पाश है कि जब बंद हो जाता है में प्रवेश (यह संदेश भेजने से पहले कनेक्शन खाली) बाल प्रक्रिया समाप्त हो जाती है।
• इस बीच, बच्चे की प्रक्रिया भेजने में अवरुद्ध है और कभी संदेश नहीं मिलता है।

Answer 1

एक पूर्ण डुप्लेक्स multiprocessing.Pipesocketpair() के रूप में लागू किया गया है। .send को कॉल करना सॉकेट से बात करते समय सभी सामान्य कारणों से अवरुद्ध हो सकता है।आपके विवरण के आधार पर मुझे लगता है कि यह संभावना है कि आपके Pipe के पाठक ने पढ़ना छोड़ दिया है और कर्नेल में बफर में डेटा बनाया गया है जहां आपके .send ब्लॉक हैं।

आप स्पष्ट रूप से .close प्राप्त ओर आप शायद त्रुटि किसी तरह मिल जाएगा तो (हालांकि संभवतः SIGPIPE रूप में अच्छी तरह, यकीन नहीं) आप .send करने का प्रयास करते। यदि आपका प्राप्त कनेक्शन गुंजाइश से बाहर जा रहा था तो यह शायद स्वचालित रूप से हो जाएगा। आप प्राप्त करने के पक्ष में संदर्भ (प्रत्यक्ष या अप्रत्यक्ष) को संग्रहीत न करने के लिए और अधिक सावधान रहकर समस्या को ठीक करने में सक्षम हो सकते हैं, इसलिए जब यह धागा दूर हो जाता है तो इसे हटा दिया जाता है।

.send अवरुद्ध तुच्छ डेमो:

import multiprocessing 

a, b = multiprocessing.Pipe() 

while True: 
    print "send!" 
    a.send("hello world") 

अब ध्यान दें कि कुछ समय बाद यह मुद्रण इस्तीफा "भेज!"