Ipyparallel langsame Ausführung mit Scatter/Gather

Question

Kontext: Ich habe ein Array, das ich über meine Motoren verstreut habe (4 Motoren zu diesem Zeitpunkt), möchte eine Funktion auf jeden Punkt im Array für eine beliebige Anzahl von Iterationen anwenden und sammeln Sie das resultierende Array von den Engines und führen Sie eine Analyse durch.

Zum Beispiel habe ich das Array von Datenpunkten, die gestreut werden, und die Anzahl der Iterationen auf jeder Datenpunkt:

data_points = range(16) 
iterations = 10 
dview.scatter('points', data_points) 

I einen Benutzer bereitgestellte Funktion haben als solche, die zu den Motoren gedrückt wird:

def user_supplied_function(point): 
    return randint(0, point) 

dview.push(dict(function_one = user_supplied_function)) 

Eine Liste für meine Ergebnisse und die parallele Ausführung:

result_list = [] 
for i in range(iterations): 
    %px engine_result = [function_one(j) for j in points] 
    result_list.append(dview.gather('engine_result')) 

Problem: Das funktioniert, und ich bekomme das Ergebnis, das ich von den Engines will, aber wie die Anzahl der Iterationen wächst, dauert die Ausführung der Schleife länger und länger. Bis zu dem Punkt, an dem 1000 Iterationen bei 50 Punkten mehr als 15 Sekunden dauern. Während eine sequentielle Version dieser Aufgabe weniger als eine Sekunde benötigt.

Irgendeine Idee, was könnte das verursachen? Könnte es der Overhead von der Nachricht sein, die von gather() geht? Wenn ja, kann jemand Lösungen vorschlagen?

Answer 1

Ich habe es herausgefunden. Es war der Overhead von gather() und .append() immerhin. Die einfachste Lösung ist gather(), nachdem die Engines ihre Arbeit beendet haben, im Gegensatz zu jeder Iteration.

Lösung

%autopx 
engine_result = [] 
for i in xrange(iterations): 
    engine_result += [[function_one(j) for j in points]] 
%autopx 
result_list = list(dview.gather('engine_result')) 

Dies wird jedoch die Ergebnisse in einer schlecht formatierte Liste von Listen, in denen die Ergebnisse von jedem Motor nebeneinander platziert werden, anstatt durch Iteration Nummer bestellt. Die folgenden Befehle verteilen die Listen und reduzieren die Unterlisten für jede Iteration.

gathered_list = [None] * iterations 
gathered_list = [[result_list[j * iterations + i] for j in xrange(len(result_list)/iterations)] for i in xrange(iterations)] 
gathered_list = [reduce(lambda x, y: x.extend(y) or x, z) for z in gathered_list]