Welche Art von Problemen (falls vorhanden) wäre es die Kombination von asyncio mit multiprocessing?

Als fast jeder ist sich dessen bewusst, wenn Sie zum ersten mal threading in Python gibt es das GIL macht das Leben Elend für die Menschen, die wirklich wollen, zu tun processing in parallel - oder geben es zumindest eine chance.

Ich bin derzeit auf der Suche an, die die Umsetzung so etwas wie das Reaktor-Muster. Effektiv möchte ich für eingehende socket-verbindungen auf einem thread-like, und wenn jemand versucht, zu verbinden, zu akzeptieren, dass die Verbindung und es weitergeben an einen anderen thread-wie für die Verarbeitung.

Bin ich (noch) nicht sicher, welche Art von laden ich vielleicht vor. Ich weiß, es ist derzeit setup-2MB-cap auf eingehende Nachrichten. Theoretisch könnten wir Tausende pro Sekunde (obwohl ich nicht weiß, wenn praktisch wir gesehen haben, etwas in der Art). Der Zeitaufwand für die Verarbeitung einer Nachricht nicht furchtbar wichtig, obwohl offensichtlich schneller wäre besser.

Ich war auf der Suche in den Reaktor-Muster und entwickelte ein kleines Beispiel für die Verwendung der multiprocessing Bibliothek, die (zumindest im Test) scheint zu funktionieren nur fine. Aber jetzt/bald haben wir die asyncio - Bibliothek zur Verfügung, die behandeln würden, die Ereignis-Schleife für mich.

Dort irgendetwas ist, was könnte mich beißen durch die Kombination asyncio und multiprocessing?

InformationsquelleAutor Wayne Werner | 2014-01-16

Sollten Sie in der Lage sein, um sicher zu kombinieren asyncio und multiprocessing ohne zu viel Mühe, obwohl Sie nicht benutzen sollte multiprocessing direkt. Die Todsünde der asyncio (und jedes andere event-loop-basierte asynchrone framework) ist die Blockierung der event-Schleife. Wenn Sie versuchen, verwenden Sie multiprocessing direkt, zu jeder Zeit, die Sie blockieren, zu warten, bis ein Kind-Prozess, Sie gehen zu blockieren, die die event-Schleife. Das ist natürlich schlecht.

Der einfachste Weg, um dies zu vermeiden, ist die Verwendung BaseEventLoop.run_in_executor zum ausführen einer Funktion in einer gleichzeitige.futures.ProcessPoolExecutor. ProcessPoolExecutor ist ein Prozess-pool umgesetzt mit multiprocessing.Process, aber asyncio hat eine eingebaute Unterstützung für das ausführen einer Funktion, ohne die Blockierung der event-Schleife. Hier ein einfaches Beispiel:

import time
import asyncio
from concurrent.futures import ProcessPoolExecutor

def blocking_func(x):
   time.sleep(x) # Pretend this is expensive calculations
   return x * 5

@asyncio.coroutine
def main():
    #pool = multiprocessing.Pool()
    #out = pool.apply(blocking_func, args=(10,)) # This blocks the event loop.
    executor = ProcessPoolExecutor()
    out = yield from loop.run_in_executor(executor, blocking_func, 10)  # This does not
    print(out)

if __name__ == "__main__":
    loop = asyncio.get_event_loop()
    loop.run_until_complete(main())

Den meisten Fällen ist dies allein die Funktion ist gut genug. Wenn Sie feststellen, selbst benötigen andere Konstrukte aus multiprocessing wie Queue, Event, Manager usw., es ist ein third-party-Bibliothek namens aioprocessing (full disclosure: ich schrieb es), bietet asyncio-kompatible Versionen aller multiprocessing Datenstrukturen. Hier ist ein Beispiel demonstrieren, dass:

import time
import asyncio
import aioprocessing
import multiprocessing

def func(queue, event, lock, items):
    with lock:
        event.set()
        for item in items:
            time.sleep(3)
            queue.put(item+5)
    queue.close()

@asyncio.coroutine
def example(queue, event, lock):
    l = [1,2,3,4,5]
    p = aioprocessing.AioProcess(target=func, args=(queue, event, lock, l)) 
    p.start()
    while True:
        result = yield from queue.coro_get()
        if result is None:
            break
        print("Got result {}".format(result))
    yield from p.coro_join()

@asyncio.coroutine
def example2(queue, event, lock):
    yield from event.coro_wait()
    with (yield from lock):
        yield from queue.coro_put(78)
        yield from queue.coro_put(None) # Shut down the worker

if __name__ == "__main__":
    loop = asyncio.get_event_loop()
    queue = aioprocessing.AioQueue()
    lock = aioprocessing.AioLock()
    event = aioprocessing.AioEvent()
    tasks = [ 
        asyncio.async(example(queue, event, lock)),
        asyncio.async(example2(queue, event, lock)),
    ]   
    loop.run_until_complete(asyncio.wait(tasks))
    loop.close()

die io-event-Schleife ist im wesentlichen Prozess, wenn ich das will send/recv via socket in den child-Prozess, wie kann ich das tun? Ich fand ich kann nicht rufen Sie einfach main_proc_loop.ensure_future(send_socket_data...) weil Sie in einem anderen Prozess? was ist der beste Weg, es zu erreichen? über die Warteschlange?

InformationsquelleAutor dano

5

Ja, es gibt durchaus ein paar bits, die möglicherweise (oder auch nicht) beißen.
- Wenn Sie so etwas wie laufen asyncio es erwartet, um die Ausführung auf einem thread oder Prozess. Diese nicht (von selbst), arbeiten mit der parallelen Verarbeitung. Sie haben irgendwie zu verteilen, während die IO-Operationen (insbesondere jene, die auf sockets) in einem einzigen thread/Prozess.
- Während Sie Ihre Idee in die hand von einzelnen verbindungen zu einem anderen handler-Prozesses ist schön, es ist schwer zu implementieren. Das erste Hindernis ist, dass Sie brauchen eine Möglichkeit, ziehen Sie die Verbindung heraus, von asyncio ohne es zu schließen. Das nächste Hindernis ist, dass man nicht einfach senden Sie eine Datei-Deskriptor für einen anderen Prozess, es sei denn, Sie verwenden die Plattform-spezifischen (wahrscheinlich Linux) - code von einem C-Erweiterung.
- Beachten Sie, dass die multiprocessing Modul bekannt ist, erstellen Sie eine Anzahl von threads für die Kommunikation. Die meisten der Zeit, wenn Sie Kommunikations-Strukturen (z.B. Queues), ein thread erzeugt. Leider sind diese threads sind nicht völlig unsichtbar. Zum Beispiel, dass Sie nicht reißen Sie Sie sauber (wenn Sie beabsichtigen, kündigen Sie Ihr Programm), aber je nach Anzahl der Ressourcen-Auslastung, deutlich auf seine eigenen.
Wenn Sie wirklich wollen, zu behandeln einzelne verbindungen in die einzelnen Prozesse, schlage ich vor, zu prüfen, verschiedene Ansätze. Zum Beispiel kann man einen socket in den listening mode und dann gleichzeitig die Annahme von verbindungen aus mehreren worker-Prozessen parallel. Sobald ein Arbeiter ist die Verarbeitung einer Anforderung abgeschlossen ist, kann es noch akzeptieren, die nächste Verbindung, damit Sie noch weniger Ressourcen als forking-Prozess für jede Verbindung. Spamassassin und Apache (mpm-prefork) können diese worker-Modell zum Beispiel. Könnte es am Ende leichter und robuster, die vom Anwendungsfall abhängig. Insbesondere können Sie machen Sie Ihre Arbeiter sterben, nachdem er eine konfigurierte Anzahl von Anfragen und erneut gespawnt werden durch einen master-Prozess, wodurch ein Großteil der negativen Auswirkungen der Speicherverluste auf.
- Ich nehme an, meine Frage war ein bisschen zweideutig - als ich erwähnte, ich würde senden Sie es aus, um einen thread-wie ich meinte eigentlich, dass Sie unterschiedliche event-loops.
InformationsquelleAutor Helmut Grohne
1

Siehe PEP 3156, insbesondere den Abschnitt über Thread-Interaktion:

http://www.python.org/dev/peps/pep-3156/#thread-interaction

Diese Dokumente eindeutig die neue asyncio Methoden, die Sie möglicherweise verwenden, einschließlich run_in_executor(). Beachten Sie, dass der Testamentsvollstrecker ist definiert im gleichzeitigen.futures, schlage ich vor, Sie auch hier nachschauen.

InformationsquelleAutor Glenn

Schreibe einen Kommentar

Du musst angemeldet sein, um einen Kommentar abzugeben.