Python multiprocessing mit einer Update-Warteschlange und eine output-queue

Wie kann ich ein Skript, ein Python-Multiprozess-mit zwei Warteschlangen wie diese?:

  1. man als Arbeits-Warteschlange, beginnt mit einigen Daten und, die, je nach den Bedingungen der Funktionen parallelisiert werden, erhält weitere Aufgaben on-the-fly,
  2. anderen, sammelt Ergebnisse und wird verwendet, um zu notieren Sie das Ergebnis nach der Verarbeitung abgeschlossen ist.

Ich im Grunde brauchen, um einige weitere Aufgaben in der aktiven Warteschlange, je nach dem, was ich in seiner ursprünglichen Elemente. Das Beispiel, das ich post unten ist albern (ich verwandeln könnte der Artikel, wie ich Sie gerne und setzen es direkt in die Ausgabe-Warteschlange), aber seine mechanik sind klar und spiegeln Teil des Konzepts, die ich brauche, sich zu entwickeln.

Hiermit mein Versuch:

import multiprocessing as mp

def worker(working_queue, output_queue):
    item = working_queue.get() #I take an item from the working queue
    if item % 2 == 0:
        output_queue.put(item**2) # If I like it, I do something with it and conserve the result.
    else:
        working_queue.put(item+1) # If there is something missing, I do something with it and leave the result in the working queue 

if __name__ == '__main__':
    static_input = range(100)    
    working_q = mp.Queue()
    output_q = mp.Queue()
    for i in static_input:
        working_q.put(i)
    processes = [mp.Process(target=worker,args=(working_q, output_q)) for i in range(mp.cpu_count())] #I am running as many processes as CPU my machine has (is this wise?).
    for proc in processes:
        proc.start()
    for proc in processes:
        proc.join()
    for result in iter(output_q.get, None):
        print result #alternatively, I would like to (c)pickle.dump this, but I am not sure if it is possible.

Dieser nicht zu Ende, noch drucken ohne Ergebnis.

Am Ende des gesamten Prozesses, würde ich mag, um sicherzustellen, dass die Arbeits-Warteschlange leer ist, und dass alle parallelen Funktionen fertig geschrieben haben, um die Ausgabe-Warteschlange vor der später iteriert wird, um die Ergebnisse. Haben Sie Vorschläge, wie man damit es funktioniert?

InformationsquelleAutor Jaqo | 2014-02-06
  1. 2

    Haben Sie einen Tippfehler in der Zeile schafft, dass die Prozesse. Es sollte mp.Process, nicht mp.process. Dies ist, was die Ursache der Ausnahme, die Sie erhalten.

    Außerdem sind Sie nicht Schleifen Sie in Ihre Mitarbeiter, so dass Sie tatsächlich verbrauchen nur ein einzelnes Element aus der queue und beendet sich dann. Ohne zu wissen, mehr über die erforderliche Logik, es ist nicht einfach, um spezifische Beratung, aber Sie werden wahrscheinlich wollen, schließen Sie den Körper Ihres worker Funktion innerhalb einer while True Schleife und fügen Sie eine Bedingung in der Körper zu beenden, wenn die Arbeit getan ist.

    Bitte beachten Sie, dass, wenn Sie nicht hinzufügen eine Bedingung, die explizit die Ausfahrt aus der Schleife Ihrer Mitarbeiter einfach stall, für immer, wenn die Warteschlange leer ist. Sie sollten erwägen, mit Hilfe der so genannten poison pill Technik zu signalisieren, die Arbeiter, die Sie möglicherweise verlassen. Sie finden ein Beispiel und einige nützliche Diskussion in der PyMOTW Artikel auf Die Kommunikation Zwischen Prozessen.

    Als für die Anzahl der Prozesse zu verwenden, müssen Sie die benchmark-etwas zu finden, was für Sie arbeitet, aber im Allgemeinen, ein Prozess pro core ist ein guter Ausgangspunkt, wenn Sie Ihre Arbeitsauslastung CPU-gebunden. Wenn Ihre Arbeitsbelastung ist IO gebunden, die Sie haben könnten bessere Ergebnisse mit einer höheren Anzahl von Beschäftigten.

    • Danke! Ich werde Bearbeiten Sie dieses in Frage.
    • Sie sind herzlich willkommen! Bitte beachten Sie, dass, während Sie wo zu tun, dass ich bearbeitet meine Antwort zum start der Adressierung der Rest deiner Frage.
    • Gerade lese ich den restlichen Teil deiner Antwort. Ich werde versuchen, eine while True Schleife. Ich Frage mich, wenn der Prozess abgeschlossen ist, wenn es keine weiteren Elemente in die Warteschlange zu arbeiten. Ich würde gerne so etwas wie eine Warteschlange der Länge, aber die Dokumentation-Staaten ist nicht zuverlässig.
    • Einige weitere Informationen für Sie!
    • Vielen Dank! Ich bin bereits daran, deine Vorschläge =)
    • Ich versuche an der Stelle so genannte "poison pill" am Ende der ersten Aufgaben einer JoinableQueue. Aber als diese Warteschlange wird immer mehr Aufgaben entlang der multiprocesing, die die Pille bekommt, bevor Sie einige Aufgaben, die bearbeitet werden müssen. Sollte ich besser prüfen, ob ein 'sperren', wenn auch dies kann Auswirkungen auf Leistung?
    • Sie scheinen das hinzufügen von mehr Arbeit aus den Beschäftigten selbst. Wenn das der Fall ist, ist es der Arbeiter, der post sollte ein poison pill, wenn es feststellt, dass es nicht mehr echte Arbeit zu veröffentlichen. Auf diese Weise, Ihre Arbeiter werden sterben, eins nach dem anderen.
    • In der Tat, meine Arbeiter sind queuing-weitere Aufgaben je nach dem, was Sie finden zusammen Ihre arbeiten. Ich habe versucht zu implementieren, Ihren Vorschlag über die Herstellung jeder Arbeiter post-die poison pill, wenn keine Arbeit mehr vorhanden ist. Habe ich nur verwendet, die unzuverlässig sind .empty () - Methode, auf der worst-case-Szenario würde ich einen single-core-Arbeit am Ende auf die Aufgaben, die nicht verwirklicht wurden, als pending, die von dem rest der Arbeitnehmer. Ich habe meine Ergebnisse ausgedruckt, aber die "In" - Zeile in der Konsole nicht mehr angezeigt. Wissen Sie, warum? Ich aktualisierte den code in der folgenden anser und stellen Sie Ihr als die richtige Antwort. Vielen Dank für deine Hilfe.
    • Code zum kopieren und einfügen wäre schön, zusammen mit den Erläuterungen.

    InformationsquelleAutor tawmas
  2. 3

    Den folgenden code erreicht die erwarteten Ergebnisse. Es folgt die Anregungen von @tawmas.

    Dieser code ermöglicht die Nutzung mehrerer Kerne in einem Prozess, der erfordert, dass die Warteschlange, die Daten-feeds, um die Arbeitnehmer können aktualisiert werden, indem Sie während der Verarbeitung:

    import multiprocessing as mp
def worker(working_queue, output_queue):
    while True:
        if working_queue.empty() == True:
            break #this is the so-called 'poison pill'    
        else:
            picked = working_queue.get()
            if picked % 2 == 0: 
                    output_queue.put(picked)
            else:
                working_queue.put(picked+1)
    return

if __name__ == '__main__':
    static_input = xrange(100)    
    working_q = mp.Queue()
    output_q = mp.Queue()
    results_bank = []
    for i in static_input:
        working_q.put(i)
    processes = [mp.Process(target=worker,args=(working_q, output_q)) for i in range(mp.cpu_count())]
    for proc in processes:
        proc.start()
    for proc in processes:
        proc.join()
    results_bank = []
    while True:
       if output_q.empty() == True:
           break
       results_bank.append(output_q.get_nowait())
    print len(results_bank) # length of this list should be equal to static_input, which is the range used to populate the input queue. In other words, this tells whether all the items placed for processing were actually processed.
    results_bank.sort()
    print results_bank
    • Print-loop stecken bleibt ewig warten, sobald die Ergebnis-queue leer ist. Sie verwenden sollten, get_nowait und ausdrücklich fangen die Leere exception zu beenden sauber.
    • Danke nochmal für Eure Hilfe. Ich bin dabei eine try: \n print result \n except Empty: \n break \n Dies ist der Druck der insgesamt erwarteten Ergebnisse, aber die Ausgabe in der Konsole ist noch beschweren sich über die Ausnahme. Ich glaube ich bin nicht richtig damit umgehen.
    • Sie brauchen, um von der Warteschlange in Ihrem Versuch.
    • Liebe @tawmas Ihre Hilfe war sehr nützlich. Nochmals vielen Dank. Ich könnte es nicht schaffen, geben Sie die Ausnahme. Stattdessen habe ich eine while True Schleife und überprüft, wenn die Warteschlange leer war. Diese sollte zuverlässig sein, da nur ein cpu-Kern bedeutet diese Aufgabe. Ich bearbeitet den code in die obige Antwort mit der Lösung, die Sie half zu finden.
    InformationsquelleAutor
Schreibe einen Kommentar