Sub Process in its own Thread
I'm wondering if the following class is sound. I'm using it to launch a bunch of simulators for each test in my test environment.
class SubProcessInOwnThread(threading.Thread): def __init__(self, arguments, currentWorkingDirectory): self.arguments = arguments self.currentWorkingDirectory = currentWorkingDirectory threading.Thread.__init__(self) self.isTerminated = False def run(self): try: self.subProcess = subprocess.Popen(self.arguments, cwd=self.currentWorkingDirectory) self.subProcess.wait() finally: self.isTerminated = True def kill(self): while not self.isTerminated: try: self.subProcess.kill() except: time.sleep(0.1)
Some senarios:
# Normal subProcessThreadArguments = ["cmd.exe"] subProcessThread = SubProcessInOwnThread(subProcessThreadArguments,r"C:\\") subProcessThread.start() time.sleep(5) subProcessThread.kill() # Process killed very quickly subProcessThreadArguments = ["cmd.exe"] subProcessThr开发者_C百科ead = SubProcessInOwnThread(subProcessThreadArguments,r"C:\\") subProcessThread.start() subProcessThread.kill() # Incorrect configuration subProcessThreadArguments = ["cmdsfgfg.exe"] subProcessThread = SubProcessInOwnThread(subProcessThreadArguments,r"C:\\") subProcessThread.start() time.sleep(5) subProcessThread.kill()
So I can create simulators like this:
subProcessThreadArguments1 = ["sim1.exe"] subProcessThread1 = SubProcessInOwnThread(subProcessThreadArguments1,r"C:\\") subProcessThread1.start() subProcessThreadArguments2 = ["sim2.exe"] subProcessThread2 = SubProcessInOwnThread(subProcessThreadArguments2,r"C:\\") subProcessThread2.start() # do test... subProcessThread1.kill() subProcessThread2.kill()
I'd be interested in any improvents. Should I consider the use of the with keyword? If so, what would the benifits be?
Thanks!
I don't see the point of having a separate thread being stuck in wait()
here. Working directly on the subprocess would work like
class SubProcessWithoutThread(object):
def __init__(self, arguments, currentWorkingDirectory):
self.arguments = arguments
self.currentWorkingDirectory = currentWorkingDirectory
self.isTerminated = False
def start(self):
self.subProcess = subprocess.Popen(self.arguments, cwd=self.currentWorkingDirectory)
def kill(self):
while self.subProcess.poll() is None:
try:
self.subProcess.kill()
except:
time.sleep(0.1)
__enter__ = start
def __exit__(self, *x):
self.kill()
(untested)
I have added the methods for a context manager, but I cannot see how that would help you as it would be quite a bunch of with
statements which you would have to create, including the necessary indentation.
But maybe I have got your intention wrong...
精彩评论