Beitrag vom 25. September 2022, für den Neubau übersetzt und fachlich ergänzt. „Horizontal“ und „vertikal“ sind meine Bilder, keine formalen Kategorien. Die Originalbeispiele bleiben als historische Fragmente erhalten; sie sind nicht vollständig oder aktuell getestet.

In diesem Beitrag versuche ich, mein mentales Modell von Parallelität beim Threading und bei asynchronem Code zu erklären. Außerdem zeige ich, wie sich beides kombinieren lässt.

Disclaimer

Das ist keine akademische Definition von Parallelität in Python. Es geht um ein grundlegendes Verständnis und darum, typische Denkfehler beim Einstieg in asynchronen Code zu vermeiden.

Synchron: vertikal

Synchroner Kontrollfluss läuft innerhalb eines Aufrufs Schritt für Schritt. Schleifen, Aufrufe und Verzweigungen ändern dabei natürlich den Ablauf. Für das Bild hier nenne ich diese Leserichtung vertikal.

# synchron code
def foo():
    print("starting here")
    a = 0
    a += 1
    b = (a + 1) / 2
    # this for loop jumps at a well-defined place.
    for x in range(10):
        b += x
    print("leaving funtion")
    return b

Von oben nach unten, mit einem Sprung an der Schleife. Nichts Neues – aber ein brauchbarer Ausgangspunkt für den nächsten Gedanken.

Threading: horizontal

Mehrere Threads können unabhängig voneinander Fortschritt machen. Auf mehreren Kernen kann tatsächlich gleichzeitig gearbeitet werden; auf einem Kern wechseln sich ausführbare Threads ab. Wann ein Wechsel passiert, lässt sich aus der Funktion heraus normalerweise nicht genau vorhersagen. Die Möglichkeiten für parallelen Python-Code hängen auch von Interpreter und GIL-Konfiguration ab. Das nächste Beispiel zeigt meine Vorstellung von horizontaler Ausführung.

Hinweis: Dieses und weitere Fragmente brauchen passende Imports; bar() muss aufgerufen werden. Die zeitliche Reihenfolge der Ausgabe ist nicht garantiert.

# parallel code using threading
def bar():

    # this function will be executed virtual parallel. 
    # The context switch and therefore the execution of a line of code 
    # is non-deterministic from perspective of the function.
    def foo(result):
        b = 0
        for x in range(10):
            b = b + x
            time.sleep(0.1)  # here we cannot tell that we halt here for some time. 
        result.append(b)

    results = []
    #  create 2 threads
    threads = [threading.Thread(target=foo, args=(results,)) for _ in range(2)]
    # start 2 threads "parallel"
    [t.start() for t in threads]
    # wait for them
    [t.join() for t in threads]
    print("result:", results)
    # prints result: [45, 45]

Ein häufiges Muster sind Producer und Consumer: Einer legt Arbeit in eine Queue, der andere nimmt sie entgegen. Eine threadsichere Queue übernimmt die Synchronisierung ihrer eigenen Operationen.

# this is "sync" consumer producer pattern were the parallelism
# will be added by creating more producer oder consumer
# instead a queue one also could use a mutable buffer like a list but needs then a semaphore.
# The queue does this all for us.
def producer(q:queue.Queue):
    for work in range(100):
        time.sleep(0.001)  # extra heavy work
        q.put(work)
    q.put(None)  # poison pill
    print("producer done")


def consumer(q: queue.Queue):
    res = 0
    while True:
        item = q.get()
        if item is None:
            break
        else:
            res = res + item
    print("consumer result", res)


def main():
    q = queue.Queue()
    threads = [threading.Thread(target=target, args=(q,)) for target in [producer, consumer]]
    [t.start() for t in threads]
    [t.join() for t in threads]
    print("bye")

Mit mehreren Consumern reicht ein einziges Endsignal nicht automatisch für alle. Threads und Kontextwechsel haben Kosten; ob sie ins Gewicht fallen, hängt vom Programm ab. Nebenläufigkeit bringt außerdem mögliche Fallstricke mit sich.

Asynchron: vertikal

Bei asyncio können Coroutinen an geeigneten await-Punkten die Kontrolle abgeben, sodass der Event Loop andere Aufgaben ausführt. Eine einzelne Coroutine liest sich weiterhin von oben nach unten. Darum nenne ich das in meinem Modell vertikal.

# a simple asynchron function will execute vertical line by line 
# allowing at defined places to halt(wait) and resume for
# other code to be executed
async def foo():
    print("starting here")
    a = 1
    await asyncio.sleep(1)  # at this place other code can be executed
    print("slept for 1 second")
    a += 2
    await asyncio.sleep(0.5)  # at this place other code can be executed
    print("leaving")
    return a

Zum Vergleich folgt das ursprüngliche Beispiel mit time.sleep. Korrektur: Es blockiert den Thread des Event Loops. Das Freigeben des GIL während time.sleep ermöglicht gegebenenfalls anderen Threads Fortschritt, aber nicht anderen Tasks auf demselben blockierten Event Loop. Das async def macht einen blockierenden Aufruf nicht nicht-blockierend.

async def foo():
    print("starting here")
    a = 1
    time.sleep(1)  # at this place other code can be executed
    print("slept for 1 second")
    a += 2
    time.sleep(0.5)  # at this place other code can be executed
    print("leaving")
    return a

Beides zusammen

Ein Event Loop kann mehrere Tasks verwalten: Jede Request-Coroutine wartet an ihren I/O-Stellen, während andere Tasks laufen. Das ist Nebenläufigkeit; für Python-Code innerhalb eines einzelnen Event-Loop-Threads folgt daraus nicht automatisch parallele CPU-Ausführung.

Referenzen auf Hintergrund-Tasks sollten gehalten werden: Der Event Loop hält schwache Referenzen, weshalb ein Task unter Umständen schon vor seinem Ende eingesammelt werden kann. Die Details stehen in der Python-Dokumentation.

async def request(page:int,  client: aiohttp.ClientSession):
        response = await client.get(f"https://books.toscrape.com/catalogue/page-{page}.html")
        if response.status < 300:
            body = await response.text()
            print(body[:18].strip())


async def main():
    client = aiohttp.ClientSession()
    loop = asyncio.get_event_loop()
    time_taken = time.time()
    async with client:
        # hold reference here
        pending = [loop.create_task(request(page, client)) for page in range(10)]
        # wait for them and release when all complete (default of wait())
        done, pending = await asyncio.wait(pending, return_when=asyncio.ALL_COMPLETED)
    print(f"done in {time.time() - time_taken:.2f}")
    await client.close()
    await asyncio.sleep(0.5)

if __name__ == '__main__':
    asyncio.run(main(), debug=True)
    # prints 
    # <!DOCTYPE html>
    # ...
    # done in 1.02

Das Beispiel startet zehn Requests nahezu gleichzeitig. Die angezeigten Zeiten stammen aus dem ursprünglichen Lauf und sind keine Benchmarks. Das Fragment setzt unter anderem Imports voraus und verwendet historische APIs; die Zielseite und deren Seitennummerierung können sich ändern.

Im folgenden Gegenbeispiel wartet die List Comprehension auf jeden Request, bevor sie den nächsten startet. Die Funktion ist zwar asynchron, die Requests laufen hier aber nacheinander:

async def request(page: int, client: aiohttp.ClientSession):
    response = await client.get(f"https://books.toscrape.com/catalogue/page-{page}.html")
    if response.status < 300:
        body = await response.text()
        print(body[:18].strip())


async def main():
    client = aiohttp.ClientSession()
    loop = asyncio.get_event_loop()
    time_taken = time.time()
    async with client:
        results = [await request(page, client) for page in range(10)]
    print(f"done in {time.time() - time_taken:.2f}")
    await client.close()
    await asyncio.sleep(0.5)


if __name__ == '__main__':
    asyncio.run(main(), debug=True)
    # prints
    # <!DOCTYPE html>
    # ...
    # done in 2.50

Am Ende hilft nur: Dokumentation lesen, kleine Experimente machen und bei Performance-Fragen messen. Mehr zu asyncio und Prozessen steht im Folgebeitrag.