python-multiprocessing - Prozess hängt mitmachen für große Warteschlange

Ich bin mit python 2.7.3 und mir ist aufgefallen das Folgendes merkwürdige Verhalten. Betrachten Sie dieses minimal-Beispiel:

from multiprocessing import Process, Queue

def foo(qin, qout):
    while True:
        bar = qin.get()
        if bar is None:
            break
        qout.put({'bar': bar})

if __name__ == '__main__':
    import sys

    qin = Queue()
    qout = Queue()
    worker = Process(target=foo,args=(qin,qout))
    worker.start()

    for i in range(100000):
        print i
        sys.stdout.flush()
        qin.put(i**2)

    qin.put(None)
    worker.join()

Wenn ich den loop über die 10.000 oder mehr, mein Skript hängt sich auf worker.join(). Es funktioniert gut, wenn die Schleife geht nur bis 1.000.

Irgendwelche Ideen?

InformationsquelleAutor user545424 | 2014-02-08
  1. 34

    Den qout Warteschlange im Teilprozess voll ist. Die Daten, die Sie in Sie setzen aus foo() passt nicht in den Puffer des OS ' s pipes intern verwendet, so dass der Teilprozess Blöcke versuchen, passen mehr Daten. Aber der parent-Prozess ist nicht das Lesen dieser Daten: es ist einfach blockiert zu, warten für den Unterprozess zu beenden. Dies ist ein typischer deadlock.

    • Es wäre toll, wenn Sie auch angeboten werden, eine code-Lösung für das problem. I. e. wie, um den Puffer zu leeren, so dass der Teilprozess nicht blockiert.
    InformationsquelleAutor Armin Rigo
  2. 4

    Muss es eine Grenze für die Größe von Warteschlangen. Betrachten Sie die folgende änderung:

    from multiprocessing import Process, Queue

def foo(qin,qout):
    while True:
        bar = qin.get()
        if bar is None:
            break
        #qout.put({'bar':bar})

if __name__=='__main__':
    import sys

    qin=Queue()
    qout=Queue()   ## POSITION 1
    for i in range(100):
        #qout=Queue()   ## POSITION 2
        worker=Process(target=foo,args=(qin,))
        worker.start()
        for j in range(1000):
            x=i*100+j
            print x
            sys.stdout.flush()
            qin.put(x**2)

        qin.put(None)
        worker.join()

    print 'Done!'

    Dieser Werke, wie Sie sind (mit qout.put Zeile auskommentiert). Wenn Sie versuchen, speichern Sie alle 100000 Ergebnisse, dann qout zu groß wird: wenn ich entfernen Sie die Kommentarzeichen aus der qout.put({'bar':bar}) im foo, und lassen die definition von qout in POSITION 1, wird der code hängt sich auf. Wenn ich jedoch verschieben qout definition auf POSITION 2, dann beendet sich das Skript.

    Also kurz gesagt, Sie müssen vorsichtig sein, dass weder qin noch qout zu groß wird. (Siehe auch: Multiprocessing Queue maxsize-limit ist 32767)

    InformationsquelleAutor amd
  3. 3

    Ich hatte das gleiche problem auf python3 wenn er versucht wird, um strings in eine Warteschlange mit Gesamtgröße über 5000 cahrs.

    In meinem Projekt gab es einen host-Prozess, setzt eine Warteschlange und startet Unterprozess, dann verbindet. Afrer join host-Prozess liest form der Warteschlange. Wenn Teilprozess producess zu viel Daten, host-hungs auf join. Ich fixierte diese mit der folgenden Funktion zu warten für den Teilprozess in der host-Prozess:

    def yield_from_process(q, p):
    while p.is_alive():
        p.join(timeout=1)
        while True:
            try:
                yield q.get(block=False)
            except Empty:
                break

    Lese ich aus der Warteschlange aus, sobald es füllt also es wird nie sehr groß

    InformationsquelleAutor Rugnar
  4. 0

    Ich versuche zu .get() eine asynchrone worker, nachdem der pool geschlossen hatte

    Einrückung Fehler außerhalb der with-block

    ich hatte dieses

    with multiprocessing.Pool() as pool:
    async_results = list()
    for job in jobs:
        async_results.append(
            pool.apply_async(
                _worker_func,
                (job,),
            )
        )
# wrong
for async_result in async_results:
    yield async_result.get()

    ich brauchte in diesem

    with multiprocessing.Pool() as pool:
    async_results = list()
    for job in jobs:
        async_results.append(
            pool.apply_async(
                _worker_func,
                (job,),
            )
        )
    # right
    for async_result in async_results:
        yield async_result.get()
    InformationsquelleAutor ThorSummoner
Schreibe einen Kommentar