Pydev PyUnitの問題でthread.joinを使用してすべてのスレッドが結合されていることを確認しました

Question

pydevを使用してテストを停止しているときに問題が発生しました。私は問題を掘り下げ、根本原因が何であるか知っています。問題を再現するために使用できる以下のコードのサンプルを提供しました。

私はCentOS 6.3、python 2.7、eclipse juno、pydev 2.7.1で主にテストしていますが、同様の設定でWindows 7でも問題が発生します。

私はサーバーのさまざまなスレッド（すべてサードパーティのライブラリ内にあるため、システムのその側を辞任できません）のプロセスランチャーとして動作するPythonスクリプトを持っています。

私のprocess.pyの最後にすべてのスレッドが終了していることを確認するために、終了する前にすべてのスレッドを結合しようとするコードセクションがあります。

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

これは正常な生産コードで正常に動作します。

pydevを使用してEclipseでPyUnitでテストを実行すると問題が発生します。余分なスレッドがPythonに追加されているので、テストが止まってしまいます。

Run As - > Python Runを使用してプログラムを起動すると、コードが期待どおりに実行され、問題なく終了します。 Run As - > Python unit-testを使用してプログラムを起動すると、テストは常にハングします。

whatsスレッドが利用可能であるかどうかを確認すると、問題が明確になります。テストコードのサンプルが提供使うだけでPythonの実行などのテストを実行しているとき、私は見ることができ、次のスレッドが（予想通り）

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)>> 

としての私のテストを実行示しています

pythonによって追加された余分なスレッドは、このコードを破るように見えます。私のコードがServerCommまたは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他の誰かがこの問題accross来るだけで包み

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"