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

Question

Я пытаюсь запустить многопроцессорную функцию и как-то хранить ее вывод в массиве, но не могу найти способ сделать это. Это то, что у меня до сих пор:

resultsAr = [] 
inputData = [a,b,c,d]#THIS ARRAY CONTAINS DATA 

def function(data): 
    values = [some_Number_1, some_Number_2] 
    resultsAr.append(values)   
    print ('Values = ', values) #THIS WORKS - THE CORRECT VALUES ARE PRINTED 
    print ('resultsAr = ', resultsAr) #WORKS AS WELL 

def parallel_run(function, inputData): #a function that runs in multiple processors  
    cpu_no = 4 
    if len(inputData) < cpu_no: 
     cpu_no = len(inputData) 
    p = multiprocessing.Pool(cpu_no) 
    p.starmap_async(function, inputData, chunksize=1) 
    p.close() 
    p.join() 
    print ('resultsAr = ', resultsAr) # THIS PRINTS OUT AN EMPTY ARRAY! 

parallel_run(function, inputData) 

Answer 1

Что происходит, что Питон раскручивается дочерние процессы для вас в вашем бассейне, а затем он копирует входные данные в течение этого процесса для вас. У дочернего процесса теперь есть копия оригинала, ваш function мутирует эту копию. Вот почему мутации появляются в ваших заявках на печать в function. Однако вы не изменяете его в родительском процессе, поэтому, когда выполнение возвращается к этому процессу, исходный resultsAr не был изменен.

канонический способ сделать то, что вы описываете здесь просто использовать Pool.map вызов, например, так:

resultsAr = [] 
inputData = [a,b,c,d] 

def function(data): 
    values = [some_Number_1, some_Number_2] 
    return values 

def parallel_run(function, inputData): 
    cpu_no = 4 
    if len(inputData) < cpu_no: 
     cpu_no = len(inputData) 
    p = multiprocessing.Pool(cpu_no) 
    # this needs to be declared global to mutate 
    # the resultsAr defined in the enclosing scope 
    global resultsAr 
    resultsAr = p.map(function, inputData, chunksize=1) 
    p.close() 
    p.join() 
    print ('resultsAr = ', resultsAr) 

Кроме того, пример вы дали должен бросить исключение, так как вы передаете параметр (элементы inputData) функции, которая не принимает аргументов (function). Я предполагаю, что вы действительно хотите параметр в function в конце концов.

Answer 2

Основываясь на ответе выше, я обнаружил, что pool.starmap_async извлекает экземпляр non-iterable. Эта проблема была решена путем добавления выражения get к коду, например:

resultsAr = [] 
inputData = [a,b,c,d] 

def function(data): 
    values = [some_Number_1, some_Number_2] 
    return values 

def parallel_run(function, inputData): 
    cpu_no = 4 
    if len(inputData) < cpu_no: 
     cpu_no = len(inputData) 
    p = multiprocessing.Pool(cpu_no) 

    global resultsAr 
    resultsAr = p.starmap_async(function, inputData, chunksize=1) 
    real_result = resultsAr.get() 

    p.close() 
    p.join() 
    print ('real_result = ', real_result)