Python Multiprocessing, Pool Karte - löschen Sie alle laufenden Prozesse, wenn man, kehrt das gewünschte Ergebnis

Question

angesichts der folgenden Python-Code:

import multiprocessing 

def unique(somelist): 
    return len(set(somelist)) == len(somelist) 


if __name__ == '__main__': 
    somelist = [[1,2,3,4,5,6,7,8,9,10,11,12,13,2], [1,2,3,4,5], [1,2,3,4,5,6,7,8,9,1], [0,1,5,1]] 

    pool = multiprocessing.Pool() 
    reslist = pool.map(unique, somelist) 
    pool.close() 
    pool.join() 
    print "Done!" 

    print reslist 

Nun stell dir vor, dass die Listen mit ganzen Zahlen in diesem Spielzeug Beispiel extrem lang sind, und Was ich hier erreichen möchte, ist folgendes: Wenn eine der Listen in der Sommeliste True zurückgibt, beende alle laufenden Prozesse.

Dies führt zu zwei Fragen (und wahrscheinlich mehr, was ich nicht kommen mit):

• Wie kann ich ‚lesen‘/‚hören‘ aus einem fertigen Prozess das Ergebnis, während andere Prozesse Rennen? Wenn z.B. ein Prozess beschäftigt sich mit [1,2,3,4,5] von der Somelist und ist vor allen anderen Prozessen abgeschlossen, wie kann ich das Ergebnis aus diesem Prozess in diesem Moment auslesen?

• Gegeben ist, dass es möglich ist, das Ergebnis eines abgeschlossenen Prozesses zu lesen, während andere laufen: Wie kann ich dieses Ergebnis als Bedingung verwenden, um alle anderen laufenden Prozesse zu beenden?

z.B. Wenn ein Prozess beendet ist und True zurückgegeben hat, wie kann ich dies als Bedingung verwenden, um alle anderen (noch) laufenden Prozesse zu beenden?

Vielen Dank im Voraus für alle Hinweise Dan

Answer 1

Verwenden pool.imap_unordered die Ergebnisse in beliebiger Reihenfolge sie kommen zu sehen.

reslist = pool.imap_unordered(unique, somelist) 
pool.close() 
for res in reslist: 
    if res: # or set other condition here 
     pool.terminate() 
     break 
pool.join() 

Sie können einen imap reslist in Ihrem Haupt-Prozess durchlaufen, während die Pool-Prozesse nach wie vor sind die Ergebnisse zu erzeugen.

Answer 2

Ohne ausgefallene IPC-Tricks (Inter-Process Communication) ist es am einfachsten, eine-Methode mit einer Callback-Funktion zu verwenden. Der Rückruf wird im Hauptprogramm ausgeführt (in einem Thread, der von multiprocessing erstellt wurde) und konsumiert jedes Ergebnis, sobald es verfügbar wird. Wenn der Rückruf ein gewünschtes Ergebnis sieht, kann er die Pool beenden. Zum Beispiel

import multiprocessing as mp 

def worker(i): 
    from time import sleep 
    sleep(i) 
    return i, (i == 5) 

def callback(t): 
    i, quit = t 
    result[i] = quit 
    if quit: 
     pool.terminate() 

if __name__ == "__main__": 
    N = 50 
    pool = mp.Pool() 
    result = [None] * N 
    for i in range(N): 
     pool.apply_async(func=worker, args=(i,), callback=callback) 
    pool.close() 
    pool.join() 
    print(result) 

die mit ziemlicher Sicherheit werden die folgenden Anzeigen (OS Scheduling Kapricen kann einen anderen Eingang erlauben oder zwei verzehrt werden):

[False, False, False, False, False, True, None, None, None, None, 
None, None, None, None, None, None, None, None, None, None, 
None, None, None, None, None, None, None, None, None, None, 
None, None, None, None, None, None, None, None, None, None, 
None, None, None, None, None, None, None, None, None, None]