Wie verwende ich Python Multiprocessing Pool.map, um numpy Array in einer for-Schleife zu füllen

Question

Ich möchte ein 2D-numpy Array innerhalb einer for-Schleife füllen und die Berechnung mit Multiprocessing befestigen.

import numpy 
from multiprocessing import Pool 


array_2D = numpy.zeros((20,10)) 
pool = Pool(processes = 4) 

def fill_array(start_val): 
    return range(start_val,start_val+10) 

list_start_vals = range(40,60) 
for line in xrange(20): 
    array_2D[line,:] = pool.map(fill_array,list_start_vals) 
pool.close() 

print array_2D 

Die Wirkung, die sie von der Ausführung ist, dass Python 4 Subprozesse läuft und belegt 4 CPU-Kerne aber die Ausführung Passt nicht Finish und das Array wird nicht gedruckt. Wenn ich versuche, das Array auf die Festplatte zu schreiben, passiert nichts.

Kann mir jemand sagen, warum?

Answer 1

Das Problem liegt an der Ausführung der pool.map in for-Schleife, das Ergebnis der map() -Methode ist funktionell gleichwertig mit der integrierten map(), außer dass einzelne Aufgaben parallel ausgeführt werden. so in Ihrem Fall die pool.map (fill_array, list_start_vals) 20 Mal aufgerufen wird, und starten Sie läuft parallel sollten

für jede Iteration der for-Schleife, Below-Code arbeiten

Code:

#!/usr/bin/python 

import numpy 
from multiprocessing import Pool 

def fill_array(start_val): 
    return range(start_val,start_val+10) 

if __name__ == "__main__": 
    array_2D = numpy.zeros((20,10)) 
    pool = Pool(processes = 4)  
    list_start_vals = range(40,60) 

    # running the pool.map in a for loop is wrong 
    #for line in xrange(20): 
    # array_2D[line,:] = pool.map(fill_array,list_start_vals) 

    # get the result of pool.map (list of values returned by fill_array) 
    # in a pool_result list 
    pool_result = pool.map(fill_array,list_start_vals) 

    # the pool is processing its inputs in parallel, close() and join() 
    #can be used to synchronize the main process 
    #with the task processes to ensure proper cleanup. 
    pool.close() 
    pool.join() 

    # Now assign the pool_result to your numpy 
    for line,result in enumerate(pool_result): 
     array_2D[line,:] = result 

    print array_2D 

Answer 2

Die folgende Arbeiten. Zuerst ist es eine gute Idee, den Hauptteil Ihres Codes in einem Hauptblock zu schützen, um seltsame Nebenwirkungen zu vermeiden. Das Ergebnis von poo.map() ist eine Liste, die die Auswertungen für jeden Wert im Iterator list_start_vals enthält, sodass Sie zuvor keine array_2D erstellen müssen.

import numpy as np 
from multiprocessing import Pool 

def fill_array(start_val): 
    return list(range(start_val, start_val+10)) 

if __name__=='__main__': 
    pool = Pool(processes=4) 
    list_start_vals = range(40, 60) 
    array_2D = np.array(pool.map(fill_array, list_start_vals)) 
    pool.close() # ATTENTION HERE 
    print array_2D 

vielleicht werden Sie Schwierigkeiten haben, pool.close() verwenden, aus den Kommentaren von @hpaulj Sie können nur diese Zeile im Falle entfernen Sie Probleme haben ...

Answer 3

Wenn Sie immer noch das Array füllen verwenden möchten, können Sie kann pool.apply_async anstelle von pool.map verwenden. Arbeiten von zu Saullo Antwort:

import numpy as np 
from multiprocessing import Pool 

def fill_array(start_val): 
    return range(start_val, start_val+10) 

if __name__=='__main__': 
    pool = Pool(processes=4) 
    list_start_vals = range(40, 60) 
    array_2D = np.zeros((20,10)) 
    for line, val in enumerate(list_start_vals): 
     result = pool.apply_async(fill_array, [val]) 
     array_2D[line,:] = result.get() 
    pool.close() 
    print array_2D 

Dies läuft etwas langsamer als die map. Aber es erzeugt keinen Laufzeitfehler wie mein Test der Kartenversion: Exception RuntimeError: RuntimeError('cannot join current thread',) in <Finalize object, dead> ignored