Python: Как использовать значение и массив в многопроцессорном пуле

Question

Для multiprocessing с Process, я могу использовать Value, Array, установив args param.

С 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) используют различный Pickler/Unpickler механизмов для функций, передаваемых в большинстве Pool методов. Это следствие того, что объекты, созданные с помощью таких вещей, как mp.Value, mp.Array, mp.Lock, ..., не могут быть переданы в качестве аргументов таких методов, хотя они могут быть переданы в качестве аргументов mp.Processи к дополнительному initializer функции 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-у платформ (где mp создает новые процессы через fork()), вы можете вместо этого создать свой mp.Array и mp.Value (и т.д.) объектов, как модуль Глобалы в любое время, прежде чем вы mp.Pool(). Процессы, созданные fork(), наследуют все, что находится в глобальном адресном пространстве модуля во время выполнения mp.Pool().

Но это вообще не работает на платформах (см. «Windows»), которые не поддерживают fork().