Python多处理池在连接时挂起？

Question

我试图并行地在几个文件上运行一些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'>, ())

如果我甚至看到其中一个,进程父进程永远挂起,永远不会收获孩子并退出.

Answer 1

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

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

• 我并不疯狂,我正在努力做的事情本应该起作用
• 至少在python2中,如果不是不可能将"异常"反馈回父进程则非常困难.简单的工作,但许多其他人没有.

在我的例子中,我的worker函数正在启动一个segfaulting的子进程.这返回了CalledProcessError异常,该异常不是pickleable.由于某种原因,这使得父对象中的池对象出去吃午餐而不是从对join()的调用返回.

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

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()))

然后,我有我的初始map函数调用process_file_wrapped()而不是原始的.现在我的代码按预期工作.

Answer 2

实际上，可以使用functools.partial实例代替lambda需要腌制对象的情况。partial自Python 2.7（和Python 3）以来，对象是可腌制的。

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