Python:如何在多处理池中使用值和数组

Question

对于使用Process 的multiprocessing，我可以通过设置args 参数来使用Value, Array。

multiprocessing 和 Pool，我如何使用 Value, Array。 文档中没有关于如何执行此操作的任何内容。

from multiprocessing import Process, Value, Array

def f(n, a):
    n.value = 3.1415927
    for i in range(len(a)):
        a[i] = -a[i]

if __name__ == '__main__':
    num = Value('d', 0.0)
    arr = Array('i', range(10))

    p = Process(target=f, args=(num, arr))
    p.start()
    p.join()

    print(num.value)
    print(arr[:])

我正在尝试在下面的代码片段中使用 Value, Array。

import multiprocessing


def do_calc(data):
    #  access num or 
    #  work to update arr
    newdata =data * 2
    return newdata

def start_process():
    print 'Starting', multiprocessing.current_process().name

if __name__ == '__main__':
    num             = Value('d', 0.0)
    arr             = Array('i', range(10))  
    inputs          = list(range(10))
    print 'Input   :', inputs

    pool_size       = multiprocessing.cpu_count() * 2
    pool            = multiprocessing.Pool(processes=pool_size,initializer=start_process, )
    pool_outputs    = pool.map(do_calc, inputs)
    pool.close() # no more tasks
    pool.join()  # wrap up current tasks

    print 'Pool    :', pool_outputs

Answer 1

我从来不知道这样做的“原因”，但是 multiprocessing (mp) 对传递给大多数 Pool 的函数使用不同的 pickler/unpickler 机制> 方法。因此，由 mp.Value、mp.Array、mp.Lock 等创建的对象，...作为参数传递给此类方法，尽管它们可以作为参数传递给 mp.Process 和给可选的初始化器 mp.Pool() 的函数。由于后者，这有效:

import multiprocessing as mp

def init(aa, vv):
    global a, v
    a = aa
    v = vv

def worker(i):
    a[i] = v.value * i

if __name__ == "__main__":
    N = 10
    a = mp.Array('i', [0]*N)
    v = mp.Value('i', 3)
    p = mp.Pool(initializer=init, initargs=(a, v))
    p.map(worker, range(N))
    print(a[:])

然后打印

[0, 3, 6, 9, 12, 15, 18, 21, 24, 27]

这是我所知道的使它跨平台工作的唯一方法。

在 Linux-y 平台上(mp 通过 fork() 创建新进程)，您可以改为创建您的 mp.Array 和mp.Value(等)对象作为模块全局变量之前你做mp.Pool()。 fork() 创建的进程在 mp.Pool() 执行时继承模块全局地址空间中的任何内容。

但这在不支持 fork() 的平台(阅读“Windows”)上根本不起作用。