python - 引入多处理队列时执行时间增加

Question

我正在尝试使用 Python 的多处理程序包，特别是使用 Process 函数来测量我“并行化”的一段代码。

我有两个函数要并行运行:function1 和 function2。 function1 不返回值，而 function2 返回值。 function2 的返回值是一个相当大的类实例。

这是我现有的使用队列并行化和获取返回值的代码:

import multiprocessing as mpc
...
def Wrapper(self,...):
  jobs = []
  q = mpc.Queue()
  p1 = mpc.Process(target=self.function1,args=(timestep,))
  jobs.append(p1)

  p2 = mpc.Process(target=self.function2,args=(timestep,arg1,arg2,arg3,...,q))
  jobs.append(p2)

  for j in jobs:
    j.start()
  result = q.get()

  for j in jobs:
    j.join()

所以，这就是我遇到的问题。如果我删除对 result = q.get() 的调用，则执行 Wrapper 函数所需的时间会显着减少，因为它不会从 function2 返回类，但是我显然没有从函数中得到我需要的数据。如果我把它放回去，运行时间会显着增加，从而表明并行化实际上比顺序执行这两个函数花费的时间更长。

以下是 Wrapper 的一些平均执行时间，供引用:

• 顺序代码(即，function1(timestep)，res = function2(timestep,a1,a2,a3,...,None)): 10秒

• 不使用队列的并行代码:8 秒

• 队列的并行代码:60 秒

我使用此代码的目的是展示并行化一段代码如何缩短执行不必要的并行函数所需的时间。作为引用，我正在使用 cProfile 包，生成我的代码的配置文件，并查看 Wrapper 运行所需的时间。

我开始对整个过程感到沮丧。它旨在基本上加快我添加到内部开发的现有自定义框架中的部分程序的速度，但是我无法实际证明我没有增加太多开销。

如果我查看程序的整体执行时间，并行代码运行得更快。然而，当我深入挖掘时，我的并行化代码开始花费更长的时间。

现在，我的想法是 Queue 正在执行某种深度复制操作，但是我找不到说明该事实的引用，所以我假设它正在返回一个浅拷贝，这对我来说应该' 需要这样的开销。

Answer 1

当您将一个对象传递到multiprocessing.Queue 时，它需要在put 端进行pickle，然后pickle 的字节必须被刷新到管道。在 get 端，需要从管道中读取 pickle 字节，然后将它们解 pickle 回 Python 对象。所以实际上，multiprocessing.Queue 执行的操作比深度复制还要慢。

您看到的开销几乎可以肯定是解封大型对象所需开销的结果。这是 Python 真正挣扎的并行编程领域 - 如果您正在执行 CPU 绑定(bind)操作(因此不能使用线程来获得并行性)并且需要共享状态，您将付出性能损失。如果您共享大型对象，惩罚也可能很大。 Python 中的并行性是通过并行化某些 CPU 绑定(bind)操作获得的性能提升与必须在进程之间共享状态而获得的性能惩罚之间的权衡。因此，您的目标需要是最小化共享状态的数量，并最大化并行化的工作量。

不幸的是，完成此操作后，进一步减轻性能影响的选项就有些受限了。您可以尝试将您的类转换为 ctypes 对象，这样您就可以使用 multiprocessing.sharedctypes在共享内存中创建对象。这应该比通过 Queue 返回对象更快，但是您必须处理 ctypes 的所有限制。

另一个想法是在 multiprocessing.Manager 中创建您的对象服务器。如果这样做，您的实际对象将存在于服务器进程中，并且您的父进程和子进程都将通过 Proxy 访问该对象。 .但是，这会使对象的每次读/写变慢，因此最终它可能不会比您现在的 Queue 实现更好。

这些替代方案都不是很好，并且可能都不适用于您的用例，在这种情况下，Python 可能不是解决此特定问题的最佳语言。不要误会我的意思；我喜欢 Python 并尽可能使用它，但这是它真正挣扎的领域。