猿问

为什么在 python 多处理中关闭池之前不能使用 join()

我有一个类,它有一个方法可以进行一些并行计算,并且经常被调用。因此,我希望我的池在类的构造函数中初始化一次,而不是每次调用此方法时都创建一个新池。在此方法中,我想使用 apply_async() 为所有工作进程启动一个任务,然后等待(阻塞)并聚合每个任务的结果。我的代码如下所示:


class Foo:

     def __init__(self, ...):

         # ...

         self.pool = mp.Pool(mp.cpu_count())


     def do_parallel_calculations(self, ...):

         for _ in range(mp.cpu_count()):

              self.pool.apply_async(calc_func, args=(...), callback=aggregate_result)

         

         # wait for results to be aggregated to a global var by the callback

         self.pool.join()  # <-- ValueError: Pool is still running

         

         # do something with the aggregated result of all worker processes

但是,当我运行此命令时,我在 self.pool.join() 中收到错误消息:“ValueError:池仍在运行”。现在,在所有示例中,我都看到 self.pool.close() 在 self.pool.join() 之前被调用,我认为这就是我收到此错误的原因,但我不想关闭我的池,因为我想要它在那里下次调用此方法时!我不能不使用 self.pool.join(),因为我需要一种方法来等待所有进程完成,并且我不想浪费地手动旋转,例如使用“while not global_flag: pass”。


我可以做什么来实现我想要做的事情?为什么多重处理不允许我加入仍然开放的池?这似乎是一件完全合理的事情。


天涯尽头无女友
浏览 92回答 2
2回答

一只斗牛犬

让我们用一个真实的例子来具体说明这一点:import multiprocessing as mpdef calc_func(x):&nbsp; &nbsp; return x * xclass Foo:&nbsp; &nbsp; def __init__(self):&nbsp; &nbsp; &nbsp; &nbsp; self.pool = mp.Pool(mp.cpu_count())&nbsp; &nbsp; def do_parallel_calculations(self, values):&nbsp; &nbsp; &nbsp; &nbsp; results = []&nbsp; &nbsp; &nbsp; &nbsp; for value in values:&nbsp; &nbsp; &nbsp; &nbsp; &nbsp; &nbsp; results.append(self.pool.apply_async(calc_func, args=(value,)))&nbsp; &nbsp; &nbsp; &nbsp; for result in results:&nbsp; &nbsp; &nbsp; &nbsp; &nbsp; &nbsp; print(result.get())if __name__ == '__main__':&nbsp; &nbsp; foo = Foo()&nbsp; &nbsp; foo.do_parallel_calculations([1,2,3])

千巷猫影

我想我设法通过在 apply_async() 返回的 AsyncResult 对象上调用 get() 来做到这一点。所以代码就变成了:def do_parallel_calculations(self, ...):&nbsp; &nbsp; &nbsp;results = []&nbsp; &nbsp; &nbsp;for _ in range(mp.cpu_count()):&nbsp; &nbsp; &nbsp; &nbsp; &nbsp; results.append(self.pool.apply_async(calc_func, args=(...)))&nbsp; &nbsp; &nbsp;aggregated_result = 0&nbsp; &nbsp; &nbsp;for result in results:&nbsp; &nbsp; &nbsp; &nbsp; &nbsp; aggregated_result += result.get()其中 calc_func() 返回单个任务结果,不需要回调和全局变量。这并不理想,因为我以任意顺序等待它们,而不是按照它们实际完成的顺序(最有效的方法是减少结果),但由于我只有 4 个核心,所以几乎不会被注意到。
随时随地看视频慕课网APP

相关分类

Python
我要回答