テストがハングし続ける場所をテストするために 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 を使用してプログラムを起動すると、テストが常にハングします。
利用可能なスレッドを見ると、問題が明らかになります。提供されたテスト コード サンプルを使用すると、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
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"