Weitergabe eines Timeouts an den äußeren Geltungsbereich

Angenommen, wir möchten über einen asynchronen Iterator iterieren, aber für jedes Element höchstens max_time Sekunden warten. Wir könnten die Logik dafür natürlich in einem asynchronen Generator kapseln, damit die Aufrufstelle weiterhin eine einfache async for-Schleife verwenden kann.

async def iter_with_timeout(ait, max_time):
    try:
        while True:
            with timeout(max_time):
                yield await anext(ait)
    except StopAsyncIteration:
        return

async def fn():
    async for elem in iter_with_timeout(ait, max_time=1.0):
        await do_something_with(elem)

Leider gibt es in dieser Version einen Fehler: Der Timeout könnte ablaufen, nachdem der Generator ge-yieldet hat, aber bevor er wieder aufgenommen wird! In diesem Fall sehen wir eine CancelledError, die in der äußeren Aufgabe ausgelöst wird, wo sie nicht vom with timeout(max_time):-Statement abgefangen werden kann.

Die Korrektur ist ziemlich einfach: Holen Sie sich das nächste Element innerhalb des Timeout-Kontexts und yielden Sie es dann *außerhalb* dieses Kontexts.

async def correct_iter_with_timeout(ait, max_time):
    try:
        while True:
            with timeout(max_time):
                tmp = await anext(ait)
            yield tmp
    except StopAsyncIteration:
        return

Weitergabe von Hintergrundaufgaben (unterbricht Abbrechung und Fehlerbehandlung)

Timeouts sind nicht die einzige Schnittstelle, die einen Abbruchkontext umschließt – und wenn Sie Hintergrund-Worker-Aufgaben benötigen, können Sie die TaskGroup nicht einfach vor dem Yield schließen.

Als Beispiel betrachten wir einen Fan-In-Generator, den wir verwenden werden, um die Feeds von mehreren "Sensoren" zusammenzuführen. Wir werden unsere Mock-Sensoren auch mit einem kleinen Puffer einrichten, damit wir eine Fehlermeldung in der Hintergrundaufgabe auslösen, während die Ablaufsteuerung außerhalb des combined_iterators-Generators liegt.

import asyncio, itertools

async def mock_sensor(name):
    for n in itertools.count():
        await asyncio.sleep(0.1)
        if n == 1 and name == "b":  # 'presence detection'
            yield "PRESENT"
        elif n == 3 and name == "a":  # inject a simple bug
            print("oops, raising RuntimeError")
            raise RuntimeError
        else:
            yield f"{name}-{n}"  # non-presence sensor data

async def move_elements_to_queue(ait, queue):
    async for obj in ait:
        await queue.put(obj)

async def combined_iterators(*aits):
    """Combine async iterators by starting N tasks, each of
    which move elements from one iterable to a shared queue."""
    q = asyncio.Queue(maxsize=2)
    async with asyncio.TaskGroup() as tg:
        for ait in aits:
            tg.create_task(move_elements_to_queue(ait, q))
        while True:
            yield await q.get()

async def turn_on_lights_when_someone_gets_home():
    combined = combined_iterators(mock_sensor("a"), mock_sensor("b"))
    async for event in combined:
        print(event)
        if event == "PRESENT":
            break
    print("main task sleeping for a bit")
    await asyncio.sleep(1)  # do some other operation

asyncio.run(turn_on_lights_when_someone_gets_home())

Wenn wir diesen Code ausführen, sehen wir die erwartete Abfolge von Beobachtungen, dann eine 'detection' und dann, während die Hauptaufgabe schläft, lösen wir diesen RuntimeError im Hintergrund aus. Aber... wir beobachten den RuntimeError nicht wirklich, nicht einmal als __context__ einer anderen Ausnahme!

>> python3.11 demo.py
a-0
b-0
a-1
PRESENT
main task sleeping for a bit
oops, raising RuntimeError

Traceback (most recent call last):
  File "demo.py", line 39, in <module>
    asyncio.run(turn_on_lights_when_someone_gets_home())
  ...
  File "demo.py", line 37, in turn_on_lights_when_someone_gets_home
    await asyncio.sleep(1)  # do some other operation
  File ".../python3.11/asyncio/tasks.py", line 649, in sleep
    return await future
asyncio.exceptions.CancelledError

Auch hier ist das Problem, dass wir innerhalb eines Abbruchkontexts yieldet haben; diesmal der Kontext, den eine TaskGroup verwendet, um Geschwisteraufgaben abzubrechen, wenn eine der Kindaufgaben eine Ausnahme auslöst. Die CancelledError, die für die Geschwisteraufgabe bestimmt war, wurde stattdessen in die *äußere* Aufgabe injiziert, und so hatten wir keine Chance, eine ExceptionGroup(..., [RuntimeError()]) zu erstellen und auszulösen.

Um dies zu beheben, müssen wir unseren asynchronen Generator in einen asynchronen Kontextmanager umwandeln, der einen asynchronen Iterable liefert – in diesem Fall einen Generator, der die Warteschlange umschließt; in Zukunft vielleicht die Warteschlange selbst.

async def queue_as_aiterable(queue):
    # async generators that don't `yield` inside a cancel scope are fine!
    while True:
        try:
            yield await queue.get()
        except asyncio.QueueShutDown:
            return

@asynccontextmanager  #  yield-in-cancel-scope is OK in a context manager
async def combined_iterators(*aits):
    q = asyncio.Queue(maxsize=2)
    async with asyncio.TaskGroup() as tg:
        for ait in aits:
            tg.create_task(move_elements_to_queue(ait, q))
        yield queue_as_aiterable(q)

async def turn_on_lights_when_someone_gets_home():
    ...
    async with combined_iterators(...) as ait:
        async for event in ait:
            ...

In einem benutzerdefinierten Kontextmanager

Das Yielden innerhalb eines Abbruchkontexts kann sicher sein, wenn und nur wenn Sie den Generator zur Implementierung eines Kontextmanagers verwenden [3] – in diesem Fall werden alle weitergegebenen Ausnahmen an die erwartete Aufgabe weitergeleitet.

Wir haben auch die Linter-Regel ASYNC101 in flake8-async implementiert, die vor dem Yielden in bekannten Abbruchkontexten warnt. Könnte Benutzeraufklärung ausreichen, um diese Probleme zu vermeiden? Leider nicht: benutzerdefinierte Kontextmanager können auch einen Abbruchkontext umschließen, und es ist nicht machbar, alle solchen Fälle zu erkennen oder zu linten.

Dies tritt regelmäßig in der Praxis auf, da "einige Hintergrundaufgaben für die Dauer dieses Kontexts ausführen" ein sehr gängiges Muster in der strukturierten Nebenläufigkeit ist. Wir haben dies oben in combined_iterators() gesehen; und diesen Fehler in mehreren Implementierungen des WebSocket-Protokolls gesehen.

async def get_messages(websocket_url):
    # The websocket protocol requires background tasks to manage the socket heartbeat
    async with open_websocket(websocket_url) as ws:  # contains a TaskGroup!
        while True:
            yield await ws.get_message()

async with open_websocket(websocket_url) as ws:
    async for message in get_messages(ws):
        ...

Implementierung – Verfolgung von Frames

Der neue Kontextmanager sys.prevent_yields erfordert Unterstützung des Interpreters. Für jeden Frame verfolgen wir die Ein- und Ausgänge dieses Kontextmanagers.

Wir sind nicht besonders an die genaue Darstellung gebunden; wir werden sie als Stapel (was klare Fehlermeldungen unterstützen würde) besprechen, aber kompaktere Darstellungen wie Integer-Paare würden ebenfalls funktionieren.

• Beim Betreten eines neu erstellten oder wieder aufgenommenen Frames, initialisieren Sie leere Stapel von Einträgen und Ausgängen.
• Beim Zurückkehren aus einem Frame, verschmelzen Sie diese Stapel in den Stapel des Elternframes.
• Beim Yielden
  • wenn entries != [] und nicht frame.allow_yield_flag, lösen Sie eine RuntimeError aus, anstatt zu yielden (das neue Verhalten, das dieses PEP vorschlägt)
  • andernfalls verschmelzen Sie Stapel in den Elternframe wie bei einer Rückgabe.

Da es sich hier um das Yielden von Frames *innerhalb* einer Aufgabe handelt, nicht um das Umschalten zwischen Aufgaben, sollten syntaktisches yield und yield from betroffen sein, aber await-Ausdrücke sollten nicht.

Wir können den Overhead reduzieren, indem wir diese Metadaten in einem einzigen Stapel pro Thread für alle Stack-Frames speichern, die keine Generatoren sind.

Bearbeitete Beispiele

Kein-Yield-Beispiel

In diesem Beispiel sehen wir mehrere Runden des Stapel-Zusammenführens, während wir von sys.prevent_yields über den benutzerdefinierten ContextManager zurück zum ursprünglichen Frame entrollen. Der Grund für die Verhinderung von Yields wird der Kürze halber nicht gezeigt; er ist Teil des "1 enter" -Zustands.

Ohne yield lösen wir keine Fehler aus, und da die Anzahl der Ein- und Ausgänge die Frame-Rückgaben wie üblich ausgleicht, ohne weitere Nachverfolgung.

Versuche-zu-Yield-Beispiel

In diesem Beispiel versucht der Frame, innerhalb des sys.prevent_yields-Kontexts zu yielden. Dies wird vom Interpreter erkannt, der eine RuntimeError auslöst, anstatt den Frame auszusetzen.

Erlaubt-zu-Yield-Beispiel

In diesem Beispiel hat ein Dekorator den Frame als Yield-erlaubt markiert. Dies könnte @contextlib.contextmanager oder ein ähnlicher Dekorator sein.

Wenn der Frame Yields erlaubt sind, wird der Ein-/Ausgangsstapel vor dem Aussetzen in den Stapel des Elternframes gemerged. Wenn der Frame wieder aufgenommen wird, ist sein Stapel leer. Schließlich, wenn der Frame beendet wird, wird der Ausgang in den Stapel des Elternframes gemerged und dieser neu ausbalanciert.

Dies stellt sicher, dass der Elternframe korrekt jeglichen verbleibenden sys.prevent_yields-Zustand erbt, während der Frame sicher ausgesetzt und wieder aufgenommen werden kann.

Verhalten, wenn sys.prevent_yields missbraucht wird

Obwohl unklug, ist es möglich, sys.prevent_yields.__enter__ und .__exit__ in einer Reihenfolge aufzurufen, die keiner gültigen Verschachtelung entspricht, oder auf andere Weise einen ungültigen Frame-Zustand zu erhalten.

Es gibt zwei Möglichkeiten, wie sys.prevent_yields.__exit__ einen ungültigen Zustand erkennen kann. Erstens, wenn Yields nicht verhindert werden, können wir einfach eine Ausnahme auslösen, ohne den Zustand zu ändern. Zweitens, wenn ein unerwarteter Eintrag am oberen Rand des Stapels liegt, schlagen wir vor, diesen Eintrag zu entfernen und eine Ausnahme auszulösen – dies stellt sicher, dass fehlerhafte Aufrufe den Stapel immer noch leeren, während sie gleichzeitig deutlich machen, dass etwas nicht stimmt.

(und wenn wir z.B. eine Integer- statt einer Stapel-basierten Darstellung wählen, sind solche Zustände möglicherweise gar nicht von korrekter Verschachtelung unterscheidbar, in welchem Fall die Frage nicht aufkommt)

Wie weit verbreitet ist dieser Fehler?

Wir haben hier keine soliden Zahlen, glauben aber, dass viele Projekte in der Wildnis betroffen sind. Seitdem ich auf der Arbeit innerhalb einer Woche auf einen moderaten und einen kritischen Fehler gestoßen bin, der auf das Aussetzen eines Abbruchkontexts zurückzuführen ist, habe ich statische Analyse mit einigem Erfolg eingesetzt. Drei Personen, mit denen Zac auf der PyCon sprach, erkannten die Symptome und kamen zu dem Schluss, dass sie wahrscheinlich betroffen waren.

TODO: Führen Sie die Linter-Regel ASYNC101 in Ökosystem-Projekten aus, z.B. den aio-libs-Paketen, und verschaffen Sie sich einen Eindruck von der Häufigkeit in weit verbreiteten PyPI-Paketen? Dies würde helfen, die Break-/Deprecationspfade für Code der Standardbibliothek zu informieren.

PEP 533, deterministische Bereinigung für Iteratoren

PEP 533 schlägt die Ergänzung von __[a]iterclose__ zum Iterator-Protokoll vor, im Wesentlichen das Umschließen jeder (asynchronen) for-Schleife mit with [a]closing(ait). Dies wäre zwar nützlich, um die rechtzeitige und deterministische Bereinigung von Ressourcen sicherzustellen, die von Iteratoren gehalten werden, und löst das Problem, das es zu lösen versucht, aber es adressiert nicht vollständig die Probleme, die dieses PEP motivieren.

Selbst mit PEP 533 würden fehlgeleitete Abbrüche immer noch an die falsche Aufgabe geliefert und könnten verheerende Auswirkungen haben, bevor der Iterator geschlossen wird. Außerdem adressiert es nicht das grundlegende Problem der strukturierten Nebenläufigkeit mit TaskGroup, bei dem das Aussetzen eines Frames, der eine TaskGroup besitzt, mit dem Modell, dass Kindaufgaben vollständig in ihrem Elternframe gekapselt sind, unvereinbar ist.

Asynchrone Generatoren ganz verwerfen

Auf dem Sprachgipfel 2024 schlugen mehrere Teilnehmer stattdessen vor, asynchrone Generatoren *in toto* zu verwerfen. Unglücklicherweise verwenden zwar die in der Praxis häufigen Fälle asynchrone Generatoren, aber Trio-Code kann das gleiche Problem mit Standard-Generatoren auslösen.

# We use Trio for this example, because while `asyncio.timeout()` is async,
# Trio's CancelScope type and timeout context managers are synchronous.
import trio

def abandon_each_iteration_after(max_seconds):
    # This is of course broken, but I can imagine someone trying it...
    while True:
        with trio.move_on_after(max_seconds):
            yield

@trio.run
async def main():
    for _ in abandon_each_iteration_after(max_seconds=1):
        await trio.sleep(3)

Wenn es den fraglichen Fehler nicht gäbe, würde dieser Code ziemlich idiomatisch aussehen – aber nach etwa einer Sekunde löst er anstatt zur nächsten Iteration überzugehen eine Ausnahme aus.

Traceback (most recent call last):
  File "demo.py", line 10, in <module>
    async def main():
  File "trio/_core/_run.py", line 2297, in run
    raise runner.main_task_outcome.error
  File "demo.py", line 12, in main
    await trio.sleep(3)
  File "trio/_timeouts.py", line 87, in sleep
    await sleep_until(trio.current_time() + seconds)
  ...
  File "trio/_core/_run.py", line 1450, in raise_cancel
    raise Cancelled._create()
trio.Cancelled: Cancelled

Darüber hinaus gibt es einige synchrone Kontextmanager, die keine Abbruchkontexte umfassen, die verwandte Probleme aufweisen, wie z.B. der oben erwähnte decimal.localcontext. Während die Behebung des folgenden Beispiels kein Ziel dieses PEPs ist, zeigt es, dass Yield-innerhalb-mit-Probleme nicht ausschließlich asynchronen Generatoren vorbehalten sind.

import decimal

def why_would_you_do_this():
    with decimal.localcontext(decimal.Context(prec=1)):
        yield

one = decimal.Decimal(1)
print(one / 3)  # 0.3333333333333333333333333333
next(gen := why_would_you_do_this())
print(one / 3)  # 0.3

Obwohl ich gute Erfahrungen in asynchronem Python ohne asynchrone Generatoren gemacht habe [5], würde ich lieber das Problem lösen, als sie aus der Sprache zu entfernen.

Können wir Ausnahmen nicht einfach an die richtige Stelle liefern?

Wenn wir PEP 568 (Generator-Sensitivität für Kontextvariablen; siehe auch PEP 550) implementieren würden, wäre es möglich, Ausnahmen von Timeouts zu behandeln: Die Ereignisschleife könnte verhindern, eine CancelledError auszulösen, bis der Generator-Frame, der den Kontextmanager enthält, auf dem Stapel liegt – entweder wenn der Generator wieder aufgenommen wird oder wenn er finalisiert wird.

Dies kann beliebig lange dauern; selbst wenn wir PEP 533 implementieren würden, um eine rechtzeitige Bereinigung beim Beenden von (asynchronen) for-Schleifen sicherzustellen, ist es immer noch möglich, einen Generator manuell mit next/send anzutreiben.

Dies löst jedoch nicht das andere Problem mit TaskGroup. Das Modell für Generatoren ist, dass man einen Stack-Frame in suspendiertem Zustand versetzt und ihn dann als inertes Wert behandeln kann, das gespeichert, verschoben und vielleicht an einem beliebigen Ort verworfen oder wiederbelebt werden kann. Das Modell für strukturierte Nebenläufigkeit ist, dass Ihr Stack zu einem Baum wird, wobei Kindaufgaben innerhalb eines Elternframes gekapselt sind. Sie erweitern das grundlegende strukturierte Programmiermodell in unterschiedliche und leider inkompatible Richtungen.

Angenommen zum Beispiel, dass das Aussetzen eines Frames, der eine offene TaskGroup enthält, auch alle Kindaufgaben aussetzen würde. Dies würde die "abwärts gerichtete" strukturierte Nebenläufigkeit wahren, insofern als dass Kinder gekapselt bleiben – allerdings auf Kosten des Deadlocks unserer beiden motivierenden Beispiele und vieler realweltlicher Codes. Es wäre jedoch immer noch möglich, den Generator in einer anderen Aufgabe wieder aufzunehmen, was die "aufwärts gerichtete" Invariante der strukturierten Nebenläufigkeit verletzen würde.

Wir glauben nicht, dass es sich lohnt, so viel Maschinerie hinzuzufügen, um Abbruchkontexte zu handhaben, während Task Groups weiterhin kaputt bleiben.

Alternative Implementierung – Bytecode inspizieren

Jelle Zijlstra hat eine Alternative skizziert, bei der sys.prevent_yields den Bytecode von Aufrufern inspiziert, bis er sicher ist, dass zwischen dem aufrufenden Instruktionszeiger und dem nächsten Kontextende kein Yield liegt. Wir erwarten, dass die Unterstützung für syntaktisch verschachtelte Kontextmanager ziemlich einfach hinzugefügt werden könnte.

Es ist jedoch noch unklar, wie dies funktionieren würde, wenn benutzerdefinierte Kontextmanager sys.prevent_yields umschließen. Schlimmer noch, dieser Ansatz ignoriert explizite Aufrufe von __enter__() und __exit__(), was bedeutet, dass sich das Kontextmanagement-Protokoll je nachdem, ob die with-Anweisung verwendet wurde, unterscheiden würde.

Die "nur bezahlen, wenn Sie es benutzen"-Performance-Kosten sind sehr attraktiv. Die Inspektion von Frame-Objekten ist jedoch für Kern-Kontrollflusskonstrukte unerschwinglich teuer und verursacht durch Deoptimierung verlangsamte Programme. Auf der anderen Seite führt die Hinzufügung von Interpreter-Unterstützung für bessere Leistung zu den gleichen "unabhängig von der Nutzung bezahlten" Semantiken wie unsere bevorzugte Lösung oben.