piedev PyUnit मुद्दा थ्रेड.जॉइन का उपयोग करते समय यह सुनिश्चित करने के लिए कि सभी धागे

Question

में शामिल हैं, मुझे परीक्षण के लिए pydev का उपयोग करते समय कोई समस्या हो रही है जहां मेरे परीक्षण लटकते रहते हैं। मैंने इस मुद्दे में खोला है और पता है कि मूल कारण क्या है। मैंने नीचे दिए गए कोड के नमूने प्रदान किए हैं जिनका उपयोग इस मुद्दे को पुन: पेश करने के लिए किया जा सकता है।

मैं मुख्य रूप से केंद्र 6.3, पायथन 2.7, ग्रहण जूनो, पायडेव 2.7.1 पर परीक्षण कर रहा हूं, हालांकि यह समस्या विंडोज 7 पर समान सेटअप के साथ भी होती है।

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

यह सुनिश्चित करने के लिए कि सभी थ्रेड मेरे process.py के अंत में समाप्त हो गए हैं मेरे पास कोड का एक अनुभाग है जो बाहर निकलने से पहले सभी धागे में शामिल होने का प्रयास करता है।

for t in threading.enumerate():    
    if t.getName() != 'MainThread': 
     t.join() 

यह सामान्य उत्पादन कोड में ठीक काम करता है।

समस्या pydev के साथ ग्रहण में Pyunit में परीक्षण चलते समय होता है। पाइथन में अतिरिक्त धागे जोड़े जा रहे हैं जिसके परिणामस्वरूप मेरे परीक्षण लटक रहे हैं।

यदि मैं रन ए -> पायथन रन का उपयोग करके अपना प्रोग्राम लॉन्च करता हूं, तो मेरा कोड अपेक्षित रूप से चलता है और ठीक से बाहर निकलता है। यदि मैं रन ए -> पायथन यूनिट-टेस्ट का उपयोग करके अपना प्रोग्राम लॉन्च करता हूं, तो परीक्षण हमेशा लटकता है।

यदि मैं देखता हूं कि थ्रेड उपलब्ध हैं तो समस्या स्पष्ट हो जाती है। परीक्षण कोड नमूना प्रदान की का उपयोग करना, मैं देख सकता हूँ जब सिर्फ एक अजगर रन के रूप में परीक्षण चल रहा है, तो निम्न सूत्र दिखाया जाता है (के रूप में उम्मीद)

Thread: <bound method _MainThread.getName of <_MainThread(MainThread, started 140268135126784)>> 
Thread: <bound method ThreadClass.getName of <ThreadClass(Thread A, started 140268006471424)>> 
Thread: <bound method ThreadClass.getName of <ThreadClass(Thread B, started 140267927631616)>> 

जब मैं एक इकाई परीक्षण

Thread: <bound method _MainThread.getName of <_MainThread(MainThread, started 139904571213568)>> 
Thread: <bound method WriterThread.getName of <WriterThread(pydevd.Writer, started daemon 139904379361024)>> 
Thread: <bound method ThreadClass.getName of <ThreadClass(Thread A, started 139904130479872)>> 
Thread: <bound method ThreadClass.getName of <ThreadClass(Thread B, started 139904119990016)>> 
Thread: <bound method PyDBCommandThread.getName of <PyDBCommandThread(pydevd.CommandThread, started daemon 139904358381312)>> 
Thread: <bound method ReaderThread.getName of <ReaderThread(pydevd.Reader, started daemon 139904368871168)>> 
Thread: <bound method ServerComm.getName of <ServerComm(Thread-4, started 139904345736960)>> 

के रूप में मेरे परीक्षण चलाने

पायथन द्वारा जोड़े गए अतिरिक्त धागे इस कोड को तोड़ने लगते हैं। जब मेरा कोड सर्वरकॉम या pydev.Writer में शामिल होने का प्रयास करता है तो यह लटकता है।

मैं जानता हूँ कि मैं नाम से इन धागों में शामिल नहीं करने का प्रयास कर सकता है, फिर भी उस तरह से मैं उत्पादन कोड बदल रहा हूँ इस से निपटने के लिए और मुझे लगता है कि समाधान पर भी उत्सुक नहीं हूँ। क्या कोई और इससे पहले आया है और एक अच्छा कामकाज मिला है? इसमें कोई भी मदद की सराहना की जाएगी। इस मुद्दे के लिए नीचे नमूना कोड है।

नमूना test_process.py

import sys 
import traceback 
import unittest 

class TestUBProcessManager(unittest.TestCase): 
    def setUp(self): 
     pass 

    def runTest(self): 
     globalsDict = {'sys':sys, '__name__':'__main__'} 

     haveException = False 
     try: 
      execfile('Process.py', globalsDict) 
     except Exception, detail: 
      haveException = True 
      traceback.print_exc() 

     self.assertFalse(haveException) 

if __name__ == '__main__': 
    unittest.main()  

नमूना Process.py

import threading                          
import time                           

class ThreadClass(threading.Thread):                     

    def __init__(self, threadName, threadRunCount,threadSleep):              
     threading.Thread.__init__(self)                    
     self.name = threadName;                      
     self.runCount = threadRunCount;                    
     self.sleepTime = threadSleep;                    

    def run(self):                         
     for i in range (1,self.runCount):                   
      print "\t",self.name, "Working";                   
      time.sleep(self.sleepTime);                    

class Launcher(object):                        
    def __init__(self):                        
     print "Init Threads";                      
     self.threadA = ThreadClass("Thread A",3,2)                 
     self.threadB = ThreadClass("Thread B",7,2)                 

    def launchProcess(self):                       
     print "Starting Threads";                     
     self.threadA.start();                      
     self.threadB.start();                      
     time.sleep(2);                        

if __name__ == '__main__':                       
    launcher = Launcher()                       
    try:                            
     launcher.launchProcess()                      
    finally:                           
     print "Available Threads Needed To Finish"                 
     for t in threading.enumerate():                    
      print "Thread: ",t.getName                    

     print "Attempt to join threads to ensure all threads are finished"           
     for t in threading.enumerate():                    
      print "About To Join : ",t.getName                  
      if t.getName() != 'MainThread':                   
        t.join()                       
    print "All Done"                         

Answer 1

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

if __name__ == '__main__':                       
    initialThreads = [];  
    for t in threading.enumerate(): 
     initialThreads.append(t); 

    launcher = Launcher()                       
    try:                            
     launcher.launchProcess()                      
    finally:                           
     print "Available Threads Needed To Finish"                 
     for t in threading.enumerate():                    
     print "Thread: ",t.getName                    

     print "Attempt to join threads to ensure all threads are finished"           
     for t in threading.enumerate():                    
      print "About To Join : ",t.getName                  
      if t not in initialThreads:                   
       t.join()                       
    print "All Done"