Python多处理池在加入时挂起？

Posted 2023-02-16

【中文标题】Python多处理池在加入时挂起？

【英文标题】：Python multiprocessing pool hangs at join?

【发布时间】：2013-02-25 04:32:52

【问题描述】：

我正在尝试在多个文件上并行运行一些 python 代码。构造基本上是：

def process_file(filename, foo, bar, baz=biz):
    # do stuff that may fail and cause exception

if __name__ == '__main__':
    # setup code setting parameters foo, bar, and biz

    psize = multiprocessing.cpu_count()*2
    pool = multiprocessing.Pool(processes=psize)

    map(lambda x: pool.apply_async(process_file, (x, foo, bar), dict(baz=biz)), sys.argv[1:])
    pool.close()
    pool.join()

我以前使用 pool.map 来做类似的事情并且效果很好，但我似乎不能在这里使用它，因为 pool.map 不允许（似乎）允许我传递额外的参数（并且使用 lambda 来做这件事是行不通的，因为 lambda 不能被编组）。

所以现在我正在尝试直接使用 apply_async() 来让事情正常工作。我的问题是代码似乎挂起并且永远不会退出。一些文件因异常而失败，但我不明白为什么会导致连接失败/挂起？有趣的是，如果没有一个文件因异常而失败，它确实会干净地退出。

我错过了什么？

编辑：当函数（以及工作程序）失败时，我看到了这个异常：

Exception in thread Thread-3:
Traceback (most recent call last):
  File "/usr/lib/python2.7/threading.py", line 552, in __bootstrap_inner
    self.run()
  File "/usr/lib/python2.7/threading.py", line 505, in run
    self.__target(*self.__args, **self.__kwargs)
  File "/usr/lib/python2.7/multiprocessing/pool.py", line 376, in _handle_results
    task = get()
TypeError: ('__init__() takes at least 3 arguments (1 given)', <class 'subprocess.CalledProcessError'>, ())

如果我看到其中之一，进程父进程将永远挂起，永远不会收获子进程并退出。

【问题讨论】：

您的代码似乎运行良好，即使我在process_file 中抛出随机异常。因此，这可能与您在 process_file 中实际所做的事情有关，这导致了问题。

嗯。什么版本的python？我是2.7的。实际程序中的 process_file 相当复杂，大量使用 PIL、NetworkX、poly2tri 和其他库。我知道至少有两个地方我知道在某些情况下可能导致异常的错误，但我需要简单地忽略这些错误并继续前进。我很困惑为什么它永远不会为我退出，而是为你工作。

2.7.2，这是我测试过的：gist.github.com/robertklep/5125319

这看起来确实是一个合理的测试用例，它在我的系统上也运行良好。现在我完全迷路了。

我也刚看到这个：bugs.python.org/issue9400

【参考方案1】：

很抱歉回答我自己的问题，但我至少找到了一种解决方法，所以如果其他人有类似的问题，我想在这里发布。我会接受任何更好的答案。

我相信问题的根源是 http://bugs.python.org/issue9400 。这告诉了我两件事：

我没疯，我真正想做的应该是有效的 至少在 python2 中，即使不是不可能，也很难将“异常”返回父进程。简单的方法有效，但许多其他方法无效。

就我而言，我的工作函数正在启动一个存在段错误的子进程。这返回了 CalledProcessError 异常，这是不可腌制的。出于某种原因，这使得父对象中的池对象出去吃午饭，而不是从对 join() 的调用中返回。

在我的特殊情况下，我不在乎异常是什么。最多我想记录它并继续前进。为此，我只需将我的***工作函数包装在 try/except 子句中。如果工作进程抛出任何异常，它会在尝试返回父进程之前被捕获，记录下来，然后工作进程正常退出，因为它不再尝试发送异常。见下文：

def process_file_wrapped(filenamen, foo, bar, baz=biz):
    try:
        process_file(filename, foo, bar, baz=biz)
    except:
        print('%s: %s' % (filename, traceback.format_exc()))

然后，我有我的初始映射函数调用 process_file_wrapped() 而不是原来的。现在我的代码按预期工作。

【讨论】：

您无需为回答自己的问题而道歉。这个页面现在记录了一个解决方法的真正问题。这很好。

顺便说一句，另一种解决方案可能是只获取异常的错误消息并使用基“异常”类引发，我认为它是可腌制的。

我还是 StackExchange 的新手，我不确定礼仪。鉴于上述 cmets 中的 @robertklep 的 sn-p 与普通的 Exception() 一起工作，我怀疑这也可以。但最重要的是，您必须捕获任何异常并返回一个已知可以工作的异常。

我很幸运能找到这篇文章。 That python bug 2014 年此时的状态仍然是“Open”。

【参考方案2】：

在需要腌制对象的情况下，您实际上可以使用functools.partial 实例而不是lambda。 partial 对象从 Python 2.7（和 Python 3）开始就可以腌制了。

pool.map(functools.partial(process_file, x, foo, bar, baz=biz), sys.argv[1:])

【讨论】：

嗯。我以前没有使用过functools。谢谢（你的）信息。我仍然怀疑这仍然会遇到同样的异常传播问题。

可能；我说不出来。您确实提到您之前在 pool.map 方面取得了成功，所以也许这会有所帮助。

我在完全不同的上下文中使用了 pool.map，在这种情况下事情不太可能导致异常。我应该在问题中更清楚地说明这一点。

functools.partial() 将允许您在更多场景中使用Pool.map()，但不影响异常问题。我刚刚遇到了与CalledProcessError 挂池相同的问题。捕捉、记录和提高RuntimeError 对我有用。

【参考方案3】：

不管怎样，当pool.map 挂起时，我遇到了类似的错误（不一样）。 我的用例允许我使用pool.terminate 来解决它（在更改之前确保你的用例也一样）。

我在调用terminate 之前使用了pool.map，所以我知道一切都已完成，来自docs：

map() 内置函数的并行等效项（尽管它仅支持一个可迭代参数）。它会一直阻塞，直到结果准备好。

如果这是您的用例，这可能是一种修补方法。