Python：多处理、悲情等等

Posted 2023-02-16

【中文标题】Python：多处理、悲情等等

【英文标题】：Python: multiprocessing, pathos and what not

【发布时间】：2015-09-25 04:03:06

【问题描述】：

我必须提前道歉，因为这个问题很笼统，可能不够清楚。问题是：如何并行运行一个 Python 函数，该函数本身使用一个进程池来处理某些子任务并执行大量繁重的 I/O 操作？它甚至是一个有效的任务吗？

我将尝试提供更多信息。我有一个程序，比如test_reduce()，我需要并行运行。我尝试了几种方法来做到这一点（见下文），但我似乎缺乏一些知识来理解为什么所有方法都失败了。

这个test_reduce() 过程做了很多 的事情。其中一些与问题更相关（我在下面列出）：

它使用multiprocessing模块

（原文如此！）

，即pool.Pool实例， 它使用 MongoDB 连接， 它严重依赖 numpy 和 scikit-learn 库， 它使用回调和 lambda， 它使用dill 库来腌制一些东西。

首先我尝试使用multiprocessing.dummy.Pool（这似乎是一个线程池）。我不知道这个池有什么特别之处，为什么它是，嗯，“虚拟”；整个事情都奏效了，我得到了我的结果。 问题是 CPU 负载。对于test_reduce() 的并行部分，所有内核都是 100%；对于同步部分，大部分时间约为 40-50%。我不能说这种“并行”执行的整体速度有任何提高。

然后我尝试使用multiprocessing.pool.Pool 实例来map 这个过程来处理我的数据。它失败了：

File "/usr/lib/python2.7/multiprocessing/pool.py", line 251, in map
    return self.map_async(func, iterable, chunksize).get()
  File "/usr/lib/python2.7/multiprocessing/pool.py", line 558, in get
    raise self._value
cPickle.PicklingError: Can't pickle <type 'thread.lock'>: attribute lookup thread.lock failed

我猜测应该归咎于 cPickle，并找到了使用更高级的pickler dill 的pathos 库。然而它也失败了：

File "/local/lib/python2.7/site-packages/dill/dill.py", line 199, in load
    obj = pik.load()
  File "/usr/lib/python2.7/pickle.py", line 858, in load
    dispatch[key](self)
  File "/usr/lib/python2.7/pickle.py", line 1083, in load_newobj
    obj = cls.__new__(cls, *args)
TypeError: object.__new__(generator) is not safe, use generator.__new__()

现在，this 错误是我完全不明白的。当它在池中工作时，我的程序没有输出到stdout，所以很难猜测发生了什么。我唯一知道的是，test_reduce() 在不使用多处理时会成功运行。

那么，你将如何并行运行如此繁重和复杂的东西？

【问题讨论】：

您是否必须并行运行它，因为您不想冻结 GUI？我就是在这种情况下，为了运行一些繁重的东西，我使用 QT QProcess，它类似于 subprocess 库。它通常比使用线程简单。

我猜想多个线程不能同时使用pickle（或其他文件访问方法）访问同一个文件。作为可能的解决方案，您可以为在每个线程上腌制的输出文件使用不同的名称（文件名源自当前线程号）。最后，您可以运行一个脚本来读取和合并所有单独的 pickle 文件。

@EdSmith 恐怕，在我进行任何酸洗之前很久就失败了。我会说它是multiprocessing（它广泛使用酸洗）。

@tmoreau 不，很遗憾没有。我正在训练一些复杂的分类模型。

当有很多同步时，以 ~50% 的速度使用所有内核对我来说听起来不错。

【参考方案1】：

所以，感谢@MikeMcKerns 的回答，我找到了如何使用pathos 库完成工作。我需要摆脱所有pymongo 游标，它们（作为生成器）不能被dill 腌制；这样做解决了问题，我设法并行运行我的代码。