1。
with ProcessPoolExecutor() as executor:
futures = [executor.submit(foo, user_number, user_id)
for user_number, user_id in enumerate(user_ids, start=1)]
for future in as_completed(futures):
future.result()
2。
pool = Pool()
results = [pool.apply_async(foo, args=(user_number, user_id))
for user_number, user_id in enumerate(user_ids, start=1)]
for result in results:
result.get()
pool.close()
pool.join()
3。
pool = Pool()
results = [pool.apply_async(foo, args=(user_number, user_id))
for user_number, user_id in enumerate(user_ids, start=1)]
try:
for result in results:
result.get()
finally:
pool.close()
pool.join()
[foo
做一些工作,然后引发Value错误。
[使用第一个示例时,我只有在所有期货完成后才能得到例外。在第二个示例中,当第一个作业失败时,我得到了异常。对于第三个示例,其行为类似于第一个示例
如何快速失败,仍然在退出前清理资源?
为什么会这样?根据文档,as_completed
会在期货完成后立即返回,调用future.result()
应该会引发异常。
Python版本为3.6.9
问题是Python无法安全取消已启动的作业。区别仅在于您告诉Python要做的事情:
情况1:future.result()
引发的异常是。然后,控制流跳出with
语句,并触发ProcessPoolExecutor.__exit__
。默认情况下,这会等待所有挂起的作业完成,因此挂起将一直挂起,直到出现这种情况。
情况2:遇到异常时,Python解释器立即退出。但这并不意味着您的作业已停止运行!您只是从不等待它们完成。
案例3:引发异常后,您调用pool.join()
,它与情况1的情况大致相同。执行等待作业完成,然后退出。
您可以使用此脚本检查情况2中到底发生了什么:
import signal
from multiprocessing import Pool
import time
def throw():
raise ValueError()
def foo():
def sigterm_handler(*args):
print('received sigterm')
raise SystemExit()
signal.signal(signal.SIGTERM, sigterm_handler)
while True:
print('still alive')
time.sleep(0.1)
pool = Pool()
results = [pool.apply_async(throw), pool.apply_async(foo)]
time.sleep(1)
for result in results:
result.get()
pool.close()
pool.join()
在OSX上,输出:
$ python mp_test.py
still alive
still alive
still alive
still alive
still alive
still alive
still alive
still alive
still alive
still alive
multiprocessing.pool.RemoteTraceback:
"""
Traceback (most recent call last):
File "/usr/local/Cellar/python/3.7.3/Frameworks/Python.framework/Versions/3.7/lib/python3.7/multiprocessing/pool.py", line 121, in worker
result = (True, func(*args, **kwds))
File "mp_test.py", line 8, in throw
raise ValueError()
ValueError
"""
The above exception was the direct cause of the following exception:
Traceback (most recent call last):
File "mp_test.py", line 27, in <module>
result.get()
File "/usr/local/Cellar/python/3.7.3/Frameworks/Python.framework/Versions/3.7/lib/python3.7/multiprocessing/pool.py", line 657, in get
raise self._value
ValueError
still alive
received sigterm
因此,当解释器退出时,工作人员会收到SIGTERM
信号(但是,行为可能取决于OS)。请注意,SIGINT
can将被忽略(例如,您在工作人员中使用的第三方库),因此不能保证在这种情况下您的工作人员实际上会退出。
现在,如果您有长期的工作并且确保可以安全地取消它们(例如,因为它们不执行任何I / O),则可以使用类似这样的方法来模拟大小写的行为2:
with concurrent.futures.ProcessPoolExecutor() as executor:
try:
futures = [executor.submit(foo, user_number, user_id)
for user_number, user_id in enumerate(user_ids, start=1)]
for future in concurrent.futures.as_completed(futures):
future.result()
except Exception:
# abort workers immediately if anything goes wrong
for process in executor._processes.values():
process.terminate()
raise
这会在遇到异常时将SIGTERM
发送到仍在运行的所有作业,然后引发异常(并等待所有进程完成,因此可以确保它们已停止)。同样,这不是一个正常的退出-当您在I / O期间中断时,它会并且会导致数据丢失或资源悬空。
Python docs说关于terminate
方法的这句话:
终止该过程。在Unix上,这是使用SIGTERM信号完成的。在Windows上使用TerminateProcess()。请注意,将不执行出口处理程序和finally子句等。
请注意,该进程的后代进程不会终止-它们只会变成孤立的。
警告
如果在关联的进程使用管道或队列时使用此方法,则该管道或队列可能会损坏,并可能无法被其他进程使用。同样,如果进程已获取锁定或信号量等,则终止该进程可能会导致其他进程死锁。