python - 检索使用 multiprocessing.Pool.map 启动的进程的退出代码

Question

我正在使用 python multiprocessing 模块来并行处理一些计算量大的任务。 显而易见的选择是使用 Pool 工作人员，然后使用 map 方法。

但是，进程可能会失败。例如，它们可能会被 oom-killer 悄悄杀死。因此，我希望能够检索使用 map 启动的进程的退出代码。

此外，出于日志记录的目的，我希望能够知道为执行可迭代对象中的每个值而启动的进程的 PID。

Answer 1

如果您正在使用 multiprocessing.Pool.map，您通常对池中子进程的退出代码不感兴趣，您感兴趣的是他们从工作项目中返回的值(value)是多少。这是因为在正常情况下，Pool 中的进程在您关闭/加入 池之前不会退出，因此没有退出代码检索直到所有工作完成，Pool 即将被销毁。因此，没有公共(public) API 来获取这些子进程的退出代码。

现在，您担心异常情况，在这种情况下，某些带外的东西会在其中一个子进程运行时将其杀死。如果遇到这样的问题，您可能会遇到一些奇怪的行为。事实上，在我的测试中，我杀死了 Pool 中的一个进程，而它正在作为 map 调用的一部分工作，map 从未完成，因为被杀死的进程没有完成。然而，Python 确实立即启动了一个新进程来替换我杀死的进程。

也就是说，您可以通过使用私有(private) _pool 属性直接访问池中的 multiprocessing.Process 对象来获取池中每个进程的 pid:

pool = multiprocessing.Pool()
for proc in pool._pool:
  print proc.pid

因此，您可以做一件事来尝试检测进程何时意外终止(假设您没有因此陷入阻塞调用)。您可以通过在调用 map_async 之前和之后检查池中的进程列表来执行此操作:

before = pool._pool[:]  # Make a copy of the list of Process objects in our pool
result = pool.map_async(func, iterable)  # Use map_async so we don't get stuck.
while not result.ready():  # Wait for the call to complete
    if any(proc.exitcode for proc in before):  # Abort if one of our original processes is dead.
        print "One of our processes has exited. Something probably went horribly wrong."
        break
    result.wait(timeout=1)
else:  # We'll enter this block if we don't reach `break` above.
    print result.get() # Actually fetch the result list here.

我们必须复制列表，因为当 Pool 中的进程死亡时，Python 会立即用新进程替换它，并从列表中删除死亡的进程。

这在我的测试中对我有用，但因为它依赖于 Pool 对象 (_pool) 的私有(private)属性，所以在生产代码中使用它是有风险的。我还建议，过分担心这种情况可能有点矫枉过正，因为这种情况不太可能发生，并且会使实现变得非常复杂。