小编典典

适用于并行进程的Python多处理

python

如果对某些人来说这太简单了,我感到抱歉,但是我仍然不了解python的多处理技巧。我已经阅读了
http://docs.python.org/dev/library/multiprocessing

http://pymotw.com/2/multiprocessing/basics.html
以及Google给我的许多其他教程和示例…其中许多从这里也是。

好吧,我的情况是我必须计算许多numpy矩阵,然后需要将它们存储在单个numpy矩阵中。假设我要使用20个核心(或者我可以使用20个核心),但是由于资源使进程一直存活到“死亡”,因此我没有成功使用池资源。所以我想做这样的事情:

from multiprocessing import Process, Queue  
import numpy as np

def f(q,i):  
     q.put( np.zeros( (4,4) ) )

if __name__ == '__main__':   
     q = Queue()   
     for i in range(30):   
          p = Process(target=f, args=(q,))  
          p.start()  
          p.join()  
     result = q.get()  
     while q.empty() == False:
          result += q.get()  
     print result

但是然后看来这些进程不是并行运行,而是按顺序运行(如果我错了,请纠正我),而且我不知道它们在执行计算后是否会死亡(因此对于20多个进程,尽了自己的本分,让核心可以自由地用于其他过程)。另外,对于一个非常大的数字(比如说100.000),将所有这些矩阵(可能也真的很大)存储在队列中将占用大量内存,使代码无用,因为这种想法是将每个结果都放在每次迭代中在最终结果中,就像使用锁(及其锁(及其Acquisition()和release()方法))一样,但是如果此代码不用于并行处理,则锁也无济于事…

我希望有人可以帮助我。

提前致谢!


阅读 208

收藏
2020-12-20

共1个答案

小编典典

您是正确的,它们在您的示例中按顺序执行。

p.join()导致当前线程阻塞,直到完成执行。您可能想在for循环之外单独加入您的进程(例如,通过将它们存储在列表中,然后对其进行迭代),或使用类似numpy.Poolapply_async的回调函数。这样一来,您也可以直接将其添加到结果中,而无需保留对象。

例如:

def f(i):  
    return i*np.identity(4)

if __name__ == '__main__':
    p=Pool(5)
    result = np.zeros((4,4))
    def adder(value):
        global result
        result += value

    for i in range(30):
        p.apply_async(f, args=(i,), callback=adder)
    p.close()
    p.join()
    print result

最后关闭并再加入池可确保池的进程已完成,并且result对象的计算已完成。您也可以调查将其Pool.imap用作解决问题的方法。该特定解决方案如下所示:

if __name__ == '__main__':
    p=Pool(5)
    result = np.zeros((4,4))

    im = p.imap_unordered(f, range(30), chunksize=5)

    for x in im:
        result += x

    print result

这对于您的特定情况比较干净,但对于最终尝试执行的操作可能不是这样。

至于存储所有变化的结果,如果我理解您的问题,您可以将其添加到回调方法(如上)中或使用imap/
imap_unordered(同时存储结果,但每次一次)添加到结果中您会在构建时清除它)。这样一来,存储时间就不会超过添加到结果中所需的时间。

2020-12-20