我对futures模块感到陌生,但我的任务可以从并行化中受益;但我似乎无法弄清楚如何设置线程函数和进程函数。希望有人能提供任何帮助。
我正在运行粒子群优化(PSO)算法。不想过多介绍PSO本身,以下是代码的基本布局:
有一个Particle
类,其中有一个getFitness(self)
方法(计算某些指标并将其存储在self.fitness
中)。PSO模拟有多个粒子实例(对于一些模拟来说很容易超过10个;对于一些模拟来说可能达到数百或数千个)。
每隔一段时间,我都需要计算粒子的适应度。目前,我通过循环来完成这个过程:
for p in listOfParticles:
p.getFitness(args)
然而,我注意到每个粒子的适应度可以独立计算,这使得适应度计算成为并行化的首选。实际上,我可以使用 map(lambda p: p.getFitness(args), listOfParticles)
。
现在,我可以轻松地使用 futures.ProcessPoolExecutor
来完成这个任务:
with futures.ProcessPoolExecutor() as e:
e.map(lambda p: p.getFitness(args), listOfParticles)
由于调用 p.getFitness
的副作用存储在每个粒子本身中,因此我不必担心从 futures.ProcessPoolExecutor()
获得返回值。
到目前为止,一切都好。但现在我注意到 ProcessPoolExecutor
创建新进程,这意味着它会复制内存,这很慢。我想共享内存,所以我应该使用线程。这很好,直到我意识到在每个进程中运行多个线程的几个进程可能会更快,因为多个线程仍然只在我的 8 核机器的一个处理器上运行。
这里是我的问题所在:
根据我看到的示例,ThreadPoolExecutor
在 list
上操作。同样地,ProcessPoolExecutor
也是如此。因此,在 ProcessPoolExecutor
中不能进行任何迭代操作以将其分配给 ThreadPoolExecutor
,因为 ThreadPoolExecutor
然后只能处理单个对象(请参见我下面发布的尝试)。
另一方面,我不能自己对 listOfParticles
进行切片,因为我希望 ThreadPoolExecutor
做出自己的魔法来确定需要多少个线程。
那么,最重要的问题(终于来了):
我应该如何构建代码才能有效地使用进程和线程并行化以下内容:
for p in listOfParticles:
p.getFitness()
这就是我一直在尝试的,但我不敢尝试运行它,因为我知道它不会起作用:
>>> def threadize(func, L, mw):
... with futures.ThreadpoolExecutor(max_workers=mw) as executor:
... for i in L:
... executor.submit(func, i)
...
>>> def processize(func, L, mw):
... with futures.ProcessPoolExecutor() as executor:
... executor.map(lambda i: threadize(func, i, mw), L)
...
我希望能够得到您对如何解决这个问题的任何想法,甚至是关于如何改善我的方法的建议。
如果有影响的话,我正在使用python3.3.2版本。
getFitness()
运行什么类型的代码?在CPython中线程的问题在于它们只适合于I/O绑定任务,因为CPython有一个全局解释器锁(“GIL”),它只允许一个线程同时运行。例如,如果getFitness()
运行CPU绑定的Python代码,则GIL将使线程比不使用线程更慢(线程只会增加上下文切换的额外开销)。但是,如果getFitness()
运行释放了GIL的扩展模块函数,则线程可能会有所帮助(例如,许多numpy
函数会释放GIL)。 - Tim PetersgetFitness
将粒子编码的信息解释为神经网络的起始语义,运行生成的神经网络,并计算输出误差(这个输出误差就是适应度 - 实际上是它的倒数)。因此,我认为这个函数更多地受到 CPU 的限制而不是 I/O 的限制(我从头开始做了所有的神经网络工作,它们都是类列表和它们的乘积)。因此,在这种情况下,线程可能并不会太有帮助,但我仍然希望能够在适用的问题中使用 ThreadPool 和 ProcessPool。 - inspectorG4dget