Asynchronous Generators

Ein Python-*Generator* ist jede Funktion, die einen oder mehrere yield-Ausdrücke enthält

def func():            # a function
    return

def genfunc():         # a generator function
    yield

Wir schlagen vor, denselben Ansatz zu verwenden, um *asynchrone Generatoren* zu definieren

async def coro():      # a coroutine function
    await smth()

async def asyncgen():  # an asynchronous generator function
    await smth()
    yield 42

Das Ergebnis des Aufrufs einer *asynchronen Generatorfunktion* ist ein *asynchrones Generatorobjekt*, das das in PEP 492 definierte asynchrone Iterationsprotokoll implementiert.

Es ist ein SyntaxError, eine nicht leere return-Anweisung in einem asynchronen Generator zu haben.

Unterstützung für das asynchrone Iterationsprotokoll

Das Protokoll erfordert die Implementierung von zwei speziellen Methoden

1. Eine __aiter__-Methode, die einen *asynchronen Iterator* zurückgibt.
2. Eine __anext__-Methode, die ein *awaitable*-Objekt zurückgibt, das die StopIteration-Ausnahme verwendet, um Werte zu "yielden", und die StopAsyncIteration-Ausnahme, um das Ende der Iteration zu signalisieren.

Asynchrone Generatoren definieren beide dieser Methoden. Lassen Sie uns manuell über einen einfachen asynchronen Generator iterieren

async def genfunc():
    yield 1
    yield 2

gen = genfunc()

assert gen.__aiter__() is gen

assert await gen.__anext__() == 1
assert await gen.__anext__() == 2

await gen.__anext__()  # This line will raise StopAsyncIteration.

Finalisierung

PEP 492 erfordert eine Event-Schleife oder einen Scheduler, um Koroutinen auszuführen. Da asynchrone Generatoren aus Koroutinen verwendet werden sollen, benötigen sie ebenfalls eine Event-Schleife, um sie auszuführen und zu finalisieren.

Asynchrone Generatoren können try..finally-Blöcke sowie async with haben. Es ist wichtig, eine Garantie zu geben, dass Generatoren, selbst wenn sie teilweise iteriert und dann vom Garbage Collector aufgeräumt werden, sicher finalisiert werden können. Zum Beispiel

async def square_series(con, to):
    async with con.transaction():
        cursor = con.cursor(
            'SELECT generate_series(0, $1) AS i', to)
        async for row in cursor:
            yield row['i'] ** 2

async for i in square_series(con, 1000):
    if i == 100:
        break

Der obige Code definiert einen asynchronen Generator, der async with verwendet, um über einen Datenbankcursor in einer Transaktion zu iterieren. Der Generator wird dann mit async for iteriert, was die Iteration an einem bestimmten Punkt unterbricht.

Der Generator square_series() wird dann vom Garbage Collector aufgeräumt, und ohne einen Mechanismus zum asynchronen Schließen des Generators könnte der Python-Interpreter nichts tun.

Um dieses Problem zu lösen, schlagen wir Folgendes vor

1. Implementierung einer aclose-Methode für asynchrone Generatoren, die ein spezielles *awaitable* zurückgibt. Wenn dies awaited wird, wirft es eine GeneratorExit in den pausierten Generator und iteriert darüber, bis entweder eine GeneratorExit oder eine StopAsyncIteration auftritt.

   Dies ähnelt stark dem, was die close()-Methode für reguläre Python-Generatoren tut, nur dass eine Event-Schleife zur Ausführung von aclose() benötigt wird.

2. Auslösen einer RuntimeError, wenn ein asynchroner Generator einen yield-Ausdruck in seinem finally-Block ausführt (die Verwendung von await ist jedoch in Ordnung)

   async def gen():
       try:
           yield
       finally:
           await asyncio.sleep(1)   # Can use 'await'.
   
           yield                    # Cannot use 'yield',
                                    # this line will trigger a
                                    # RuntimeError.
   

3. Hinzufügen von zwei neuen Methoden zum sys-Modul: set_asyncgen_hooks() und get_asyncgen_hooks().

Die Idee hinter sys.set_asyncgen_hooks() ist es, Event-Schleifen zu ermöglichen, die Iteration und Finalisierung von asynchronen Generatoren abzufangen, so dass der Endbenutzer sich nicht um das Finalisierungsproblem kümmern muss und alles einfach funktioniert.

sys.set_asyncgen_hooks() akzeptiert zwei Argumente

• firstiter: ein Callable, das aufgerufen wird, wenn ein asynchroner Generator zum ersten Mal iteriert wird.
• finalizer: ein Callable, das aufgerufen wird, wenn ein asynchroner Generator kurz vor dem Garbage Collection steht.

Wenn ein asynchroner Generator zum ersten Mal iteriert wird, speichert er eine Referenz auf den aktuellen *Finalizer*.

Wenn ein asynchroner Generator kurz vor dem Garbage Collection steht, ruft er seinen gespeicherten *Finalizer* auf. Es wird davon ausgegangen, dass der Finalizer einen aclose()-Aufruf mit der Schleife plant, die zum Zeitpunkt des Beginns der Iteration aktiv war.

Hier ist zum Beispiel, wie asyncio modifiziert wird, um eine sichere Finalisierung von asynchronen Generatoren zu ermöglichen

# asyncio/base_events.py

class BaseEventLoop:

    def run_forever(self):
        ...
        old_hooks = sys.get_asyncgen_hooks()
        sys.set_asyncgen_hooks(finalizer=self._finalize_asyncgen)
        try:
            ...
        finally:
            sys.set_asyncgen_hooks(*old_hooks)
            ...

    def _finalize_asyncgen(self, gen):
        self.create_task(gen.aclose())

Das zweite Argument, firstiter, ermöglicht es Event-Schleifen, eine schwache Menge von asynchronen Generatoren zu verwalten, die unter ihrer Kontrolle instanziiert wurden. Dies ermöglicht die Implementierung von "Shutdown"-Mechanismen, um alle offenen Generatoren sicher zu finalisieren und die Event-Schleife zu schließen.

sys.set_asyncgen_hooks() ist threadspezifisch, sodass mehrere parallel laufende Event-Schleifen es sicher verwenden können.

sys.get_asyncgen_hooks() gibt eine Namedtuple-ähnliche Struktur mit den Feldern firstiter und finalizer zurück.

asyncio

Die asyncio Event-Schleife verwendet die sys.set_asyncgen_hooks() API, um eine schwache Menge aller geplanten asynchronen Generatoren zu verwalten und ihre aclose() Koroutinen-Methoden zu planen, wenn die Generatoren vom Garbage Collector aufgeräumt werden sollen.

Um sicherzustellen, dass asyncio-Programme alle geplanten asynchronen Generatoren zuverlässig finalisieren können, schlagen wir die Hinzufügung einer neuen Event-Schleifen-Koroutinenmethode loop.shutdown_asyncgens() vor. Diese Methode plant die Schließung aller aktuell offenen asynchronen Generatoren mit einem aclose()-Aufruf.

Nachdem die Methode loop.shutdown_asyncgens() aufgerufen wurde, gibt die Event-Schleife eine Warnung aus, wann immer ein neuer asynchroner Generator zum ersten Mal iteriert wird. Die Idee ist, dass nach der Aufforderung zur Beendigung aller asynchronen Generatoren das Programm keinen Code ausführen sollte, der über neue asynchrone Generatoren iteriert.

Ein Beispiel dafür, wie die Koroutine shutdown_asyncgens verwendet werden sollte

try:
    loop.run_forever()
finally:
    loop.run_until_complete(loop.shutdown_asyncgens())
    loop.close()

Asynchrones Generatorobjekt

Das Objekt ist dem Standard-Python-Generatorobjekt nachempfunden. Im Wesentlichen ist das Verhalten asynchroner Generatoren so konzipiert, dass es das Verhalten synchroner Generatoren nachbildet, wobei der einzige Unterschied darin besteht, dass die API asynchron ist.

Die folgenden Methoden und Eigenschaften sind definiert

1. agen.__aiter__(): Gibt agen zurück.
2. agen.__anext__(): Gibt ein *awaitable* zurück, das eine asynchrone Generator-Iteration durchführt, wenn es awaited wird.
3. agen.asend(val): Gibt ein *awaitable* zurück, das das Objekt val in den agen-Generator pusht. Wenn agen noch nicht iteriert wurde, muss val None sein.

   Beispiel

   async def gen():
       await asyncio.sleep(0.1)
       v = yield 42
       print(v)
       await asyncio.sleep(0.2)
   
   g = gen()
   
   await g.asend(None)      # Will return 42 after sleeping
                            # for 0.1 seconds.
   
   await g.asend('hello')   # Will print 'hello' and
                            # raise StopAsyncIteration
                            # (after sleeping for 0.2 seconds.)
   

4. agen.athrow(typ, [val, [tb]]): Gibt ein *awaitable* zurück, das eine Ausnahme in den agen-Generator wirft.

   Beispiel

   async def gen():
       try:
           await asyncio.sleep(0.1)
           yield 'hello'
       except ZeroDivisionError:
           await asyncio.sleep(0.2)
           yield 'world'
   
   g = gen()
   v = await g.asend(None)
   print(v)                # Will print 'hello' after
                           # sleeping for 0.1 seconds.
   
   v = await g.athrow(ZeroDivisionError)
   print(v)                # Will print 'world' after
                           # sleeping 0.2 seconds.
   

5. agen.aclose(): Gibt ein *awaitable* zurück, das eine GeneratorExit-Ausnahme in den Generator wirft. Das *awaitable* kann entweder einen ausgegebenen Wert zurückgeben, wenn agen die Ausnahme behandelt hat, oder agen wird geschlossen und die Ausnahme wird an den Aufrufer weitergegeben.
6. agen.__name__ und agen.__qualname__: lesbare und beschreibbare Namen und qualifizierte Namensattribute.
7. agen.ag_await: Das Objekt, auf das agen gerade *wartet*, oder None. Dies ist ähnlich wie das derzeit verfügbare gi_yieldfrom für Generatoren und cr_await für Koroutinen.
8. agen.ag_frame, agen.ag_running und agen.ag_code: ähnlich wie die entsprechenden Attribute von Standardgeneratoren definiert.

StopIteration und StopAsyncIteration werden nicht aus asynchronen Generatoren weitergegeben und durch eine RuntimeError ersetzt.

Implementierungsdetails

Das asynchrone Generatorobjekt (PyAsyncGenObject) teilt sich das Struktur-Layout mit PyGenObject. Zusätzlich führt die Referenzimplementierung drei neue Objekte ein

1. PyAsyncGenASend: das awaitable-Objekt, das die Methoden __anext__ und asend() implementiert.
2. PyAsyncGenAThrow: das awaitable-Objekt, das die Methoden athrow() und aclose() implementiert.
3. _PyAsyncGenWrappedValue: jedes direkt von einem asynchronen Generator ausgegebene Objekt wird implizit in diese Struktur verpackt. So kann die Generatorimplementierung Objekte, die über das reguläre Iterationsprotokoll ausgegeben werden, von Objekten trennen, die über das asynchrone Iterationsprotokoll ausgegeben werden.

PyAsyncGenASend und PyAsyncGenAThrow sind awaitable (sie haben __await__-Methoden, die self zurückgeben) und sind koroutinenähnliche Objekte (die die Methoden __iter__, __next__, send() und throw() implementieren). Im Wesentlichen steuern sie, wie asynchrone Generatoren iteriert werden

Neue Funktionen und Typen der Standardbibliothek

1. types.AsyncGeneratorType – Typ des asynchronen Generatorobjekts.
2. sys.set_asyncgen_hooks() und sys.get_asyncgen_hooks() Methoden zur Einrichtung von Finalizern und Iterations-Interceptors für asynchrone Generatoren in Event-Schleifen.
3. inspect.isasyncgen() und inspect.isasyncgenfunction() Introspektionsfunktionen.
4. Neue Methode für die asyncio Event-Schleife: loop.shutdown_asyncgens().
5. Neue abstrakte Basisklasse collections.abc.AsyncGenerator.

Abwärtskompatibilität

Der Vorschlag ist vollständig abwärtskompatibel.

In Python 3.5 ist es ein SyntaxError, eine Funktion mit async def zu definieren, die einen yield-Ausdruck enthält. Daher ist es sicher, asynchrone Generatoren in 3.6 einzuführen.

Reguläre Generatoren

Für reguläre Generatoren gibt es keine Leistungseinbußen. Der folgende Mikro-Benchmark läuft auf CPython mit und ohne asynchrone Generatoren mit derselben Geschwindigkeit

def gen():
    i = 0
    while i < 100000000:
        yield i
        i += 1

list(gen())

Verbesserungen gegenüber asynchronen Iteratoren

Der folgende Mikro-Benchmark zeigt, dass asynchrone Generatoren etwa **2,3x schneller** sind als asynchrone Iteratoren, die in reinem Python implementiert sind

N = 10 ** 7

async def agen():
    for i in range(N):
        yield i

class AIter:
    def __init__(self):
        self.i = 0

    def __aiter__(self):
        return self

    async def __anext__(self):
        i = self.i
        if i >= N:
            raise StopAsyncIteration
        self.i += 1
        return i

aiter() und anext() Builtins

Ursprünglich definierte PEP 492 __aiter__ als eine Methode, die ein *awaitable*-Objekt zurückgeben sollte, was zu einem asynchronen Iterator führte.

In CPython 3.5.2 wurde __aiter__ jedoch neu definiert, um direkt asynchrone Iteratoren zurückzugeben. Um die Abwärtskompatibilität zu wahren, wurde beschlossen, dass Python 3.6 beide Wege unterstützt: __aiter__ kann immer noch ein *awaitable* mit einer DeprecationWarning zurückgeben.

Aufgrund dieser doppelten Natur von __aiter__ in Python 3.6 können wir keine synchrone Implementierung des Built-in aiter() hinzufügen. Daher wird vorgeschlagen, bis Python 3.7 zu warten.

Asynchrone Listen-/Dict-/Set-Comprehensions

Die Syntax für asynchrone Comprehensions steht in keinem Zusammenhang mit der Mechanik asynchroner Generatoren und sollte in einer separaten PEP behandelt werden.

Asynchrones yield from

Obwohl es theoretisch möglich ist, die Unterstützung für yield from für asynchrone Generatoren zu implementieren, würde dies eine ernsthafte Neugestaltung der Generatorimplementierung erfordern.

yield from ist auch für asynchrone Generatoren weniger kritisch, da kein Mechanismus zur Implementierung eines weiteren Koroutinenprotokolls aufbauend auf Koroutinen bereitgestellt werden muss. Und um asynchrone Generatoren zu komponieren, kann eine einfache async for-Schleife verwendet werden

async def g1():
    yield 1
    yield 2

async def g2():
    async for v in g1():
        yield v

Warum die Methoden asend() und athrow() notwendig sind

Sie ermöglichen die Implementierung von Konzepten, die contextlib.contextmanager ähneln, mithilfe asynchroner Generatoren. Zum Beispiel ist es mit dem vorgeschlagenen Design möglich, das folgende Muster zu implementieren

@async_context_manager
async def ctx():
    await open()
    try:
        yield
    finally:
        await close()

async with ctx():
    await ...

Ein weiterer Grund ist, dass es möglich ist, Daten in asynchrone Generatoren zu pushen und Ausnahmen über das Objekt zu werfen, das von __anext__ zurückgegeben wird, aber es ist schwierig, dies korrekt zu tun. Das Hinzufügen von expliziten asend() und athrow() ebnet einen sicheren Weg, dies zu erreichen.

In Bezug auf die Implementierung ist asend() eine leicht generischere Version von __anext__, und athrow() ist aclose() sehr ähnlich. Daher fügt die Definition dieser Methoden für asynchrone Generatoren keine zusätzliche Komplexität hinzu.