我从 Python GUI (PyGTK) 中启动一个进程（使用多处理）。该过程需要很长时间（约 20 分钟）才能完成。该过程完成后，我想对其进行清理（提取结果并加入该过程）。我如何知道该过程何时完成？
我的同事建议在父进程中使用一个繁忙的循环来检查子进程是否已完成。肯定有更好的方法。
在 Unix 中，当一个进程被分叉时，a signal handler is called from within the parent process when the child process has finished。但我在 Python 中看不到类似的东西。我错过了什么吗？
如何从父进程中观察到子进程的结束？ （当然，我不想调用 Process.join()，因为它会冻结 GUI 界面。）
这个问题不限于多处理：我对多线程也有完全相同的问题。
【问题讨论】：
标签： python multithreading user-interface parallel-processing multiprocessing
我认为作为使 python 多平台的一部分，像 SIGCHLD 这样的简单事情必须自己完成。同意，当您只想知道孩子何时完成时，这需要做更多的工作，但实际上并没有那么痛苦。考虑以下使用一个子进程来完成工作、两个 multiprocessing.Event 实例和一个线程来检查子进程是否完成的情况：
import threading
from multiprocessing import Process, Event
from time import sleep
def childsPlay(event):
print "Child started"
for i in range(3):
print "Child is playing..."
sleep(1)
print "Child done"
event.set()
def checkChild(event, killEvent):
event.wait()
print "Child checked, and is done playing"
if raw_input("Do again? y/n:") == "y":
event.clear()
t = threading.Thread(target=checkChild, args=(event, killEvent))
t.start()
p = Process(target=childsPlay, args=(event,))
p.start()
else:
cleanChild()
killEvent.set()
def cleanChild():
print "Cleaning up the child..."
if __name__ == '__main__':
event = Event()
killEvent = Event()
# process to do work
p = Process(target=childsPlay, args=(event,))
p.start()
# thread to check on child process
t = threading.Thread(target=checkChild, args=(event, killEvent))
t.start()
try:
while not killEvent.is_set():
print "GUI running..."
sleep(1)
except KeyboardInterrupt:
print "Quitting..."
exit(0)
finally:
print "Main done"
加入所有创建的进程和线程是一种很好的做法，因为它有助于指示何时创建僵尸（永不完成）进程/线程。我修改了上面的代码，创建了一个继承自 threading.Thread 的 ChildChecker 类。它的唯一目的是在一个单独的进程中启动一个作业，等待该进程完成，然后在一切完成时通知 GUI。加入 ChildChecker 也将加入它正在“检查”的过程。现在，如果进程在 5 秒后没有加入，线程将强制终止进程。输入“y”会创建一个运行“endlessChildsPlay”的子进程，该子进程必须演示强制终止。
import threading
from multiprocessing import Process, Event
from time import sleep
def childsPlay(event):
print "Child started"
for i in range(3):
print "Child is playing..."
sleep(1)
print "Child done"
event.set()
def endlessChildsPlay(event):
print "Endless child started"
while True:
print "Endless child is playing..."
sleep(1)
event.set()
print "Endless child done"
class ChildChecker(threading.Thread):
def __init__(self, killEvent):
super(ChildChecker, self).__init__()
self.killEvent = killEvent
self.event = Event()
self.process = Process(target=childsPlay, args=(self.event,))
def run(self):
self.process.start()
while not self.killEvent.is_set():
self.event.wait()
print "Child checked, and is done playing"
if raw_input("Do again? y/n:") == "y":
self.event.clear()
self.process = Process(target=endlessChildsPlay, args=(self.event,))
self.process.start()
else:
self.cleanChild()
self.killEvent.set()
def join(self):
print "Joining child process"
# Timeout on 5 seconds
self.process.join(5)
if self.process.is_alive():
print "Child did not join! Killing.."
self.process.terminate()
print "Joining ChildChecker thread"
super(ChildChecker, self).join()
def cleanChild(self):
print "Cleaning up the child..."
if __name__ == '__main__':
killEvent = Event()
# thread to check on child process
t = ChildChecker(killEvent)
t.start()
try:
while not killEvent.is_set():
print "GUI running..."
sleep(1)
except KeyboardInterrupt:
print "Quitting..."
exit(0)
finally:
t.join()
print "Main done"
【讨论】：
这个答案真的很简单！ （我只花了 几天 来解决这个问题。）
结合 PyGTK 的 idle_add()，你可以创建一个 AutoJoiningThread。总代码是微不足道的：
class AutoJoiningThread(threading.Thread):
def run(self):
threading.Thread.run(self)
gobject.idle_add(self.join)
如果您想做的不仅仅是加入（例如收集结果），那么您可以扩展上述类以在完成时发出信号，如下例所示：
import threading
import time
import sys
import gobject
gobject.threads_init()
class Child:
def __init__(self):
self.result = None
def play(self, count):
print "Child starting to play."
for i in range(count):
print "Child playing."
time.sleep(1)
print "Child finished playing."
self.result = 42
def get_result(self, obj):
print "The result was "+str(self.result)
class AutoJoiningThread(threading.Thread, gobject.GObject):
__gsignals__ = {
'finished': (gobject.SIGNAL_RUN_LAST,
gobject.TYPE_NONE,
())
}
def __init__(self, *args, **kwargs):
threading.Thread.__init__(self, *args, **kwargs)
gobject.GObject.__init__(self)
def run(self):
threading.Thread.run(self)
gobject.idle_add(self.join)
gobject.idle_add(self.emit, 'finished')
def join(self):
threading.Thread.join(self)
print "Called Thread.join()"
if __name__ == '__main__':
print "Creating child"
child = Child()
print "Creating thread"
thread = AutoJoiningThread(target=child.play,
args=(3,))
thread.connect('finished', child.get_result)
print "Starting thread"
thread.start()
print "Running mainloop (Ctrl+C to exit)"
mainloop = gobject.MainLoop()
try:
mainloop.run()
except KeyboardInterrupt:
print "Received KeyboardInterrupt. Quiting."
sys.exit()
print "God knows how we got here. Quiting."
sys.exit()
上述示例的输出将取决于线程执行的顺序，但类似于：
创造孩子 创建线程 开始线程 孩子开始玩。 孩子玩。 运行主循环（Ctrl+C 退出） 孩子玩。 孩子玩。 孩子玩完了。 调用 Thread.join() 结果是 42 ^C收到键盘中断。退出。
不可能以相同的方式创建 AutoJoiningProcess（因为我们不能跨两个不同的进程调用 idle_add()），但是我们可以使用 AutoJoiningThread 来获得我们想要的：
class AutoJoiningProcess(multiprocessing.Process):
def start(self):
thread = AutoJoiningThread(target=self.start_process)
thread.start() # automatically joins
def start_process(self):
multiprocessing.Process.start(self)
self.join()
为了演示 AutoJoiningProcess 这里是另一个例子：
import threading
import multiprocessing
import time
import sys
import gobject
gobject.threads_init()
class Child:
def __init__(self):
self.result = multiprocessing.Manager().list()
def play(self, count):
print "Child starting to play."
for i in range(count):
print "Child playing."
time.sleep(1)
print "Child finished playing."
self.result.append(42)
def get_result(self, obj):
print "The result was "+str(self.result)
class AutoJoiningThread(threading.Thread, gobject.GObject):
__gsignals__ = {
'finished': (gobject.SIGNAL_RUN_LAST,
gobject.TYPE_NONE,
())
}
def __init__(self, *args, **kwargs):
threading.Thread.__init__(self, *args, **kwargs)
gobject.GObject.__init__(self)
def run(self):
threading.Thread.run(self)
gobject.idle_add(self.join)
gobject.idle_add(self.emit, 'finished')
def join(self):
threading.Thread.join(self)
print "Called Thread.join()"
class AutoJoiningProcess(multiprocessing.Process, gobject.GObject):
__gsignals__ = {
'finished': (gobject.SIGNAL_RUN_LAST,
gobject.TYPE_NONE,
())
}
def __init__(self, *args, **kwargs):
multiprocessing.Process.__init__(self, *args, **kwargs)
gobject.GObject.__init__(self)
def start(self):
thread = AutoJoiningThread(target=self.start_process)
thread.start()
def start_process(self):
multiprocessing.Process.start(self)
self.join()
gobject.idle_add(self.emit, 'finished')
def join(self):
multiprocessing.Process.join(self)
print "Called Process.join()"
if __name__ == '__main__':
print "Creating child"
child = Child()
print "Creating thread"
process = AutoJoiningProcess(target=child.play,
args=(3,))
process.connect('finished',child.get_result)
print "Starting thread"
process.start()
print "Running mainloop (Ctrl+C to exit)"
mainloop = gobject.MainLoop()
try:
mainloop.run()
except KeyboardInterrupt:
print "Received KeyboardInterrupt. Quiting."
sys.exit()
print "God knows how we got here. Quiting."
sys.exit()
生成的输出将与上面的示例非常相似，只是这次我们同时加入了进程和伴随线程：
创造孩子 创建线程 开始线程 运行主循环（Ctrl+C 退出） 孩子开始玩。 孩子玩。 孩子玩。 孩子玩。 孩子玩完了。 调用 Process.join() 结果是 [42] 调用 Thread.join() ^C收到键盘中断。退出。
不幸的是：
因此，要使用这种方法，最好只从主循环/GUI 中创建线程/进程。
【讨论】：
您可以使用queue 与子进程通信。您可以在其上粘贴中间结果，或指示已达到里程碑的消息（用于进度条）或仅指示流程已准备好加入的消息。使用empty 轮询它既简单又快速。
如果你真的只想知道它是否完成，你可以观看你的进程的exitcode或投票is_alive()。
【讨论】：
在努力寻找自己问题的答案时，我偶然发现了 PyGTK 的 idle_add() function。这给了我以下可能性：
这似乎是重新创建 Unix 的 call-callback-when-child-process-is-done 的一种过于复杂的方法。
这一定是 Python 中 GUI 的一个非常常见的问题。肯定有解决这个问题的标准模式吗？
【讨论】：
看看子流程模块：
http://docs.python.org/library/subprocess.html
import subprocess
let pipe = subprocess.Popen("ls -l", stdout=subprocess.PIPE)
allText = pipe.stdout.read()
pipe.wait()
retVal = pipe.returncode
【讨论】：
pipe.poll 测试进程是否已退出。它将立即返回而不是阻塞。这可用于轮询一组子进程。