相关疑难解决方法(0)

假设我有一个大内存numpy数组,我有一个函数func,它接受这个巨大的数组作为输入(连同一些其他参数).func具有不同参数可以并行运行.例如:

def func(arr, param):
    # do stuff to arr, param

# build array arr

pool = Pool(processes = 6)
results = [pool.apply_async(func, [arr, param]) for param in all_params]
output = [res.get() for res in results]

如果我使用多处理库,那么这个巨型数组将被多次复制到不同的进程中.

有没有办法让不同的进程共享同一个数组？此数组对象是只读的,永远不会被修改.

更复杂的是,如果arr不是一个数组,而是一个任意的python对象,有没有办法分享它？

[EDITED]

我读了答案,但我仍然有点困惑.由于fork()是copy-on-write,因此在python多处理库中生成新进程时不应调用任何额外的成本.但是下面的代码表明存在巨大的开销:

from multiprocessing import Pool, Manager
import numpy as np; 
import time

def f(arr):
    return len(arr)

t = time.time()
arr = np.arange(10000000)
print "construct array = ", time.time() - t;


pool = Pool(processes = 6)

t = …

我有一个类似大型dict的对象,需要在许多工作进程之间共享.每个工作者读取对象中信息的随机子集,并使用它进行一些计算.我想避免复制大对象,因为我的机器很快耗尽了内存.

我正在玩这个SO问题的代码,我修改了一下使用固定大小的进程池,这更适合我的用例.然而,这似乎打破了它.

from multiprocessing import Process, Pool
from multiprocessing.managers import BaseManager

class numeri(object):
    def __init__(self):
        self.nl = []

    def getLen(self):
        return len(self.nl)

    def stampa(self):
        print self.nl

    def appendi(self, x):
        self.nl.append(x)

    def svuota(self):
        for i in range(len(self.nl)):
            del self.nl[0]

class numManager(BaseManager):
    pass

def produce(listaNumeri):
    print 'producing', id(listaNumeri)
    return id(listaNumeri)

def main():
    numManager.register('numeri', numeri, exposed=['getLen', 'appendi',
                        'svuota', 'stampa'])
    mymanager = numManager()
    mymanager.start()
    listaNumeri = mymanager.numeri()
    print id(listaNumeri)

    print '------------ Process'
    for i in range(5):
        producer = Process(target=produce, args=(listaNumeri,)) …

我需要在python中的几个进程之间共享一个对象及其方法.我正在尝试使用Managers(在模块多处理中)但它崩溃了.这是生产者 - 消费者的一个愚蠢的例子,其中两个进程之间的共享对象只是一个包含四种方法的数字列表.

from multiprocessing import Process, Condition, Lock  
from multiprocessing.managers import BaseManager  
import time, os  

lock = Lock()  
waitC = Condition(lock)  
waitP = Condition(lock)  

class numeri(object):  
    def __init__(self):  
        self.nl = []  

    def getLen(self):  
        return len(self.nl)  

    def stampa(self):  
        print self.nl  

    def appendi(self, x):  
        self.nl.append(x)  

    def svuota(self):  
        for i in range(len(self.nl)):  
            del self.nl[0]  

class numManager(BaseManager):  
    pass  

numManager.register('numeri', numeri, exposed = ['getLen', 'appendi', 'svuota', 'stampa'])  

def consume(waitC, waitP, listaNumeri):  
    lock.acquire()  
    if (listaNumeri.getLen() == 0):  
        waitC.wait()  
    listaNumeri.stampa()  
    listaNumeri.svuota()  
    waitP.notify()  
    lock.release()  

def produce(waitC, waitP, …