Diese Frage ist durch meine andere Frage motiviert: Wie kann man in cdef warten?

Es gibt Unmengen von Artikeln und Blog-Posts im Web asyncio, aber sie sind alle sehr oberflächlich. Ich konnte keine Informationen darüber finden, wie die asyncioImplementierung tatsächlich erfolgt und was die E / A asynchron macht. Ich habe versucht, den Quellcode zu lesen, aber es sind Tausende von Zeilen nicht des C-Codes der höchsten Klasse, von denen sich viele mit Hilfsobjekten befassen, aber vor allem ist es schwierig, eine Verbindung zwischen der Python-Syntax und dem zu übersetzenden C-Code herzustellen in.

Asycnios eigene Dokumentation ist noch weniger hilfreich. Es gibt dort keine Informationen darüber, wie es funktioniert, nur einige Richtlinien zur Verwendung, die manchmal auch irreführend / sehr schlecht geschrieben sind.

Ich bin mit der Implementierung von Coroutinen durch Go vertraut und hatte gehofft, dass Python dasselbe tut. Wenn dies der Fall wäre, hätte der Code, den ich in dem oben verlinkten Beitrag gefunden habe, funktioniert. Da dies nicht der Fall war, versuche ich jetzt herauszufinden, warum. Meine bisher beste Vermutung lautet wie folgt: Bitte korrigieren Sie mich, wo ich falsch liege:

1. Prozedurdefinitionen des Formulars async def foo(): ...werden tatsächlich als Methoden einer erbenden Klasse interpretiert coroutine.
2. Möglicherweise wird async defes tatsächlich durch awaitAnweisungen in mehrere Methoden aufgeteilt , wobei das Objekt, für das diese Methoden aufgerufen werden, den Fortschritt verfolgen kann, den es bisher durch die Ausführung erzielt hat.
3. Wenn das oben Gesagte zutrifft, läuft die Ausführung einer Coroutine im Wesentlichen darauf hinaus, Methoden eines Coroutine-Objekts durch einen globalen Manager (Schleife?) Aufzurufen.
4. Der globale Manager weiß irgendwie (wie?), Wann E / A-Operationen von Python-Code (nur?) Ausgeführt werden, und kann eine der ausstehenden Coroutine-Methoden auswählen, die ausgeführt werden sollen, nachdem die aktuelle ausführende Methode die Kontrolle aufgegeben hat (auf die awaitAnweisung klicken) ).

Mit anderen Worten, hier ist mein Versuch, eine asyncioSyntax in etwas Verständlicheres zu "entschärfen" :

async def coro(name):
    print('before', name)
    await asyncio.sleep()
    print('after', name)

asyncio.gather(coro('first'), coro('second'))

# translated from async def coro(name)
class Coro(coroutine):
    def before(self, name):
        print('before', name)

    def after(self, name):
        print('after', name)

    def __init__(self, name):
        self.name = name
        self.parts = self.before, self.after
        self.pos = 0

    def __call__():
        self.parts[self.pos](self.name)
        self.pos += 1

    def done(self):
        return self.pos == len(self.parts)


# translated from asyncio.gather()
class AsyncIOManager:

    def gather(*coros):
        while not every(c.done() for c in coros):
            coro = random.choice(coros)
            coro()

Sollte sich meine Vermutung als richtig erweisen, dann habe ich ein Problem. Wie geschieht E / A in diesem Szenario? In einem separaten Thread? Ist der gesamte Dolmetscher gesperrt und erfolgt die E / A außerhalb des Dolmetschers? Was genau ist mit E / A gemeint? Wenn meine Python-Prozedur die C-Prozedur aufgerufen open()hat und ihrerseits einen Interrupt an den Kernel gesendet hat und die Kontrolle an ihn abgegeben hat, woher weiß der Python-Interpreter davon und kann weiterhin anderen Code ausführen, während der Kernel-Code die eigentliche E / A ausführt und bis es weckt die Python-Prozedur, die den Interrupt ursprünglich gesendet hat? Wie kann sich der Python-Interpreter im Prinzip dessen bewusst sein?

Wie funktioniert Asyncio?

Bevor wir diese Frage beantworten, müssen wir einige Grundbegriffe verstehen. Überspringen Sie diese, wenn Sie bereits einen kennen.

Generatoren

Generatoren sind Objekte, mit denen wir die Ausführung einer Python-Funktion unterbrechen können. Vom Benutzer kuratierte Generatoren werden mithilfe des Schlüsselworts implementiert yield. Indem yieldwir eine normale Funktion erstellen, die das Schlüsselwort enthält, verwandeln wir diese Funktion in einen Generator:

>>> def test():
...     yield 1
...     yield 2
...
>>> gen = test()
>>> next(gen)
1
>>> next(gen)
2
>>> next(gen)
Traceback (most recent call last):
  File "<stdin>", line 1, in <module>
StopIteration

Wie Sie sehen können, next()lädt der Interpreter beim Aufrufen des Generators den Testrahmen und gibt den yielded-Wert zurück. Wenn Sie next()erneut aufrufen , wird der Frame erneut in den Interpreter-Stack geladen, und fahren yieldSie mit einem anderen Wert fort.

Beim dritten next()Aufruf war unser Generator fertig und StopIterationwurde geworfen.

Kommunikation mit einem Generator

Ein weniger bekanntes Merkmal von Generatoren ist die Tatsache, dass Sie mit ihnen auf zwei Arten kommunizieren können: send()und throw().

>>> def test():
...     val = yield 1
...     print(val)
...     yield 2
...     yield 3
...
>>> gen = test()
>>> next(gen)
1
>>> gen.send("abc")
abc
2
>>> gen.throw(Exception())
Traceback (most recent call last):
  File "<stdin>", line 1, in <module>
  File "<stdin>", line 4, in test
Exception

Beim Aufruf gen.send()wird der Wert als Rückgabewert vom yieldSchlüsselwort übergeben.

gen.throw()Auf der anderen Seite können Ausnahmen in Generatoren ausgelöst werden, wobei die an derselben Stelle ausgelöste Ausnahme yieldaufgerufen wurde.

Rückgabe von Werten von Generatoren

Wenn Sie einen Wert von einem Generator zurückgeben, wird der Wert in die StopIterationAusnahme eingefügt. Wir können den Wert später aus der Ausnahme wiederherstellen und nach Bedarf verwenden.

>>> def test():
...     yield 1
...     return "abc"
...
>>> gen = test()
>>> next(gen)
1
>>> try:
...     next(gen)
... except StopIteration as exc:
...     print(exc.value)
...
abc

Siehe, ein neues Schlüsselwort: yield from

Python 3.4 wurde mit einem neuen Schlüsselwort versehen : yield from. Was das Schlüsselwort uns erlaubt , zu tun ist auf jedem passieren next(), send()und throw()in einen inneren am weitesten verschachtelten Generator. Wenn der innere Generator einen Wert zurückgibt, ist dies auch der Rückgabewert von yield from:

>>> def inner():
...     inner_result = yield 2
...     print('inner', inner_result)
...     return 3
...
>>> def outer():
...     yield 1
...     val = yield from inner()
...     print('outer', val)
...     yield 4
...
>>> gen = outer()
>>> next(gen)
1
>>> next(gen) # Goes inside inner() automatically
2
>>> gen.send("abc")
inner abc
outer 3
4

Ich habe einen Artikel geschrieben , um dieses Thema weiter zu erläutern.

Alles zusammenfügen

Mit der Einführung des neuen Schlüsselworts yield fromin Python 3.4 konnten wir nun Generatoren in Generatoren erstellen, die genau wie ein Tunnel die Daten vom innersten zum äußersten Generator hin und her übertragen. Dies hat eine neue Bedeutung für Generatoren hervorgebracht - Coroutinen .

Coroutinen sind Funktionen, die während der Ausführung gestoppt und fortgesetzt werden können. In Python werden sie mit dem async defSchlüsselwort definiert . Ähnlich wie Generatoren, verwenden sie auch ihre eigene Form yield fromdavon ist await. Vor asyncund awaitin Python 3.5 eingeführt, haben wir Coroutinen genauso erstellt, wie Generatoren erstellt wurden (mit yield fromstatt await).

async def inner():
    return 1

async def outer():
    await inner()

Wie jeder Iterator oder Generator, der die __iter__()Methode implementiert , werden Coroutinen implementiert, __await__()die es ihnen ermöglichen, bei jedem await coroAufruf fortzufahren .

In den Python-Dokumenten befindet sich ein schönes Sequenzdiagramm , das Sie überprüfen sollten.

In asyncio haben wir neben Coroutine-Funktionen zwei wichtige Objekte: Aufgaben und Zukunft .

Futures

Futures sind Objekte, bei denen die __await__()Methode implementiert ist und deren Aufgabe es ist, einen bestimmten Status und ein bestimmtes Ergebnis zu halten. Der Zustand kann einer der folgenden sein:

1. PENDING - Zukunft hat kein Ergebnis oder keine Ausnahme gesetzt.
2. ABGESAGT - Zukunft wurde mit abgebrochen fut.cancel()
3. FINISHED - future wurde entweder durch eine Ergebnismenge mit fut.set_result()oder durch eine Ausnahmesatz mit beendetfut.set_exception()

Das Ergebnis kann, wie Sie vermutet haben, entweder ein Python-Objekt sein, das zurückgegeben wird, oder eine Ausnahme, die möglicherweise ausgelöst wird.

Ein weiteres wichtiges Merkmal von futureObjekten ist, dass sie eine aufgerufene Methode enthalten add_done_callback(). Mit dieser Methode können Funktionen aufgerufen werden, sobald die Aufgabe erledigt ist - unabhängig davon, ob eine Ausnahme ausgelöst oder beendet wurde.

Aufgaben

Aufgabenobjekte sind spezielle Zukünfte, die sich um Coroutinen wickeln und mit den innersten und äußersten Coroutinen kommunizieren. Jedes Mal, wenn eine Coroutine eine awaitZukunft hat, wird die Zukunft vollständig an die Aufgabe zurückgegeben (genau wie in yield from), und die Aufgabe erhält sie.

Als nächstes bindet sich die Aufgabe an die Zukunft. Dies geschieht durch einen Aufruf add_done_callback()an die Zukunft. Von nun an wird der Rückruf der Aufgabe aufgerufen, wenn die Zukunft jemals durch Abbrechen, Übergeben einer Ausnahme oder Übergeben eines Python-Objekts erreicht wird.

Asyncio

Die letzte brennende Frage, die wir beantworten müssen, lautet: Wie wird das E / A implementiert?

Tief in Asyncio haben wir eine Ereignisschleife. Eine Ereignisschleife von Aufgaben. Die Aufgabe der Ereignisschleife besteht darin, Aufgaben jedes Mal aufzurufen, wenn sie bereit sind, und all diese Anstrengungen in einer einzigen Arbeitsmaschine zu koordinieren.

Der E / A-Teil der Ereignisschleife basiert auf einer einzelnen entscheidenden Funktion, die aufgerufen wird select. Select ist eine Blockierungsfunktion, die vom darunter liegenden Betriebssystem implementiert wird und das Warten auf eingehende oder ausgehende Daten auf Sockets ermöglicht. Wenn Daten empfangen werden, werden sie aktiviert und geben die Sockets zurück, die Daten empfangen haben, oder die Sockets, die zum Schreiben bereit sind.

Wenn Sie versuchen, Daten über einen Socket über Asyncio zu empfangen oder zu senden, geschieht im Folgenden tatsächlich, dass der Socket zuerst überprüft wird, ob Daten vorhanden sind, die sofort gelesen oder gesendet werden können. Wenn der .send()Puffer voll ist oder der .recv()Puffer leer ist, wird der Socket für die selectFunktion registriert (indem er einfach rlistfür recvund wlistfür zu einer der Listen hinzugefügt wird send), und die entsprechende Funktion ist ein awaitneu erstelltes futureObjekt, das an diesen Socket gebunden ist.

Wenn alle verfügbaren Aufgaben auf Futures warten, ruft die Ereignisschleife auf selectund wartet. Wenn auf einem der Sockets eingehende Daten vorhanden sind oder der sendPuffer leer ist, sucht Asyncio nach dem zukünftigen Objekt, das an diesen Socket gebunden ist, und setzt es auf Fertig.

Jetzt passiert die ganze Magie. Die Zukunft ist erledigt, die Aufgabe, die sich zuvor mit hinzugefügt hat add_done_callback(), wird wieder lebendig und ruft .send()die Coroutine auf, die die innerste Coroutine (aufgrund der awaitKette) wieder aufnimmt, und Sie lesen die neu empfangenen Daten aus einem nahe gelegenen Puffer wurde verschüttet.

Wieder eine Methodenkette bei recv():

1. select.select wartet.
2. Ein fertiger Socket mit Daten wird zurückgegeben.
3. Daten vom Socket werden in einen Puffer verschoben.
4. future.set_result() wird genannt.
5. Die Aufgabe, mit der sie sich hinzugefügt hat, add_done_callback()wird jetzt aktiviert.
6. Task ruft .send()die Coroutine auf, die bis in die innerste Coroutine reicht und diese aufweckt.
7. Daten werden aus dem Puffer gelesen und an unseren bescheidenen Benutzer zurückgegeben.

Zusammenfassend lässt sich sagen, dass Asyncio Generatorfunktionen verwendet, mit denen Funktionen angehalten und fortgesetzt werden können. Es verwendet yield fromFunktionen, mit denen Daten vom innersten zum äußersten Generator hin und her übertragen werden können. Es nutzt alle diejenigen , um Halt Funktion Ausführung , während es für IO , um eine vollständige warten (durch das Betriebssystem mit selectFunktion).

Und das Beste von allem? Während eine Funktion angehalten wird, kann eine andere ausgeführt werden und sich mit dem empfindlichen Stoff verschachteln, der asynchron ist.

Sprechen async/awaitund asyncioist nicht dasselbe. Das erste ist ein grundlegendes Konstrukt auf niedriger Ebene (Coroutinen), während das letztere eine Bibliothek ist, die diese Konstrukte verwendet. Umgekehrt gibt es keine einzige endgültige Antwort.

Das Folgende ist eine allgemeine Beschreibung, wie async/awaitund asyncio-ähnlichen Bibliotheken arbeiten. Das heißt, es gibt vielleicht noch andere Tricks (es gibt ...), aber sie spielen keine Rolle, es sei denn, Sie bauen sie selbst. Der Unterschied sollte vernachlässigbar sein, es sei denn, Sie wissen bereits genug, um eine solche Frage nicht stellen zu müssen.

1. Coroutinen versus Subroutinen in einer Nussschale

Genau wie Unterprogramme (Funktionen, Prozeduren, ...) sind Coroutinen (Generatoren, ...) eine Abstraktion von Aufrufstapel und Anweisungszeiger: Es gibt einen Stapel ausführender Codeteile, und jedes befindet sich an einem bestimmten Befehl.

Die Unterscheidung zwischen defversus async defdient lediglich der Klarheit. Der tatsächliche Unterschied ist returnversus yield. Daraus awaitoder yield fromnehmen Sie die Differenz von einzelnen Anrufen ganzen Stapel.

1.1. Unterprogramme

Eine Unterroutine repräsentiert eine neue Stapelebene zum Speichern lokaler Variablen und ein einzelnes Durchlaufen ihrer Anweisungen, um ein Ende zu erreichen. Stellen Sie sich ein Unterprogramm wie das folgende vor:

def subfoo(bar):
     qux = 3
     return qux * bar

Wenn Sie es ausführen, bedeutet das

1. Ordnen Sie Stapelspeicherplatz für barund zuqux
2. Führen Sie die erste Anweisung rekursiv aus und springen Sie zur nächsten Anweisung
3. einmal ein return, drücken Sie den Wert auf den anrufenden Stapel
4. Löschen Sie den Stapel (1.) und den Anweisungszeiger (2.).

Insbesondere bedeutet 4., dass eine Unterroutine immer im selben Zustand beginnt. Alles, was nur für die Funktion selbst gilt, geht nach Abschluss verloren. Eine Funktion kann nicht wieder aufgenommen werden, auch wenn danach Anweisungen vorliegen return.

root -\
  :    \- subfoo --\
  :/--<---return --/
  |
  V

1.2. Coroutinen als persistente Unterprogramme

Eine Coroutine ist wie eine Subroutine, kann jedoch beendet werden, ohne ihren Zustand zu zerstören. Stellen Sie sich eine Coroutine wie diese vor:

 def cofoo(bar):
      qux = yield bar  # yield marks a break point
      return qux

Wenn Sie es ausführen, bedeutet das

1. Ordnen Sie Stapelspeicherplatz für barund zuqux
2. Führen Sie die erste Anweisung rekursiv aus und springen Sie zur nächsten Anweisung
   1. einmal auf einer yieldschieben seinen Wert an den anrufenden Stapel , aber Speichern des Stapels und Befehlszeiger
   2. Stellen Sie nach dem Aufruf yieldden Stapel- und Anweisungszeiger wieder her und drücken Sie die Argumente anqux
3. einmal ein return, drücken Sie den Wert auf den anrufenden Stapel
4. Löschen Sie den Stapel (1.) und den Anweisungszeiger (2.).

Beachten Sie die Hinzufügung von 2.1 und 2.2 - eine Coroutine kann an vordefinierten Punkten ausgesetzt und wieder aufgenommen werden. Dies ähnelt dem Anhalten einer Unterroutine beim Aufrufen einer anderen Unterroutine. Der Unterschied besteht darin, dass die aktive Coroutine nicht streng an ihren aufrufenden Stapel gebunden ist. Stattdessen ist eine suspendierte Coroutine Teil eines separaten, isolierten Stapels.

root -\
  :    \- cofoo --\
  :/--<+--yield --/
  |    :
  V    :

Dies bedeutet, dass suspendierte Coroutinen frei gelagert oder zwischen Stapeln bewegt werden können. Jeder Aufrufstapel, der Zugriff auf eine Coroutine hat, kann diese fortsetzen.

1.3. Durchlaufen des Aufrufstapels

Bisher geht unsere Coroutine nur mit den Call-Stack runter yield. Eine Unterroutine kann den Aufrufstapel mit und nach unten und oben gehen . Der Vollständigkeit halber benötigen Coroutinen auch einen Mechanismus, um den Aufrufstapel zu erhöhen. Stellen Sie sich eine Coroutine wie diese vor:return()

def wrap():
    yield 'before'
    yield from cofoo()
    yield 'after'

Wenn Sie es ausführen, bedeutet dies, dass der Stapel und der Anweisungszeiger weiterhin wie eine Unterroutine zugewiesen werden. Wenn es angehalten wird, ist das immer noch wie das Speichern einer Unterroutine.

Allerdings yield fromtut beides . Es verschiebt Stapel und Befehlszeiger wrap und läuft cofoo. Beachten Sie, dass wrapbis zum cofoovollständigen Abschluss ausgesetzt bleibt . Immer wenn cofooangehalten oder etwas gesendet wird, cofooist es direkt mit dem aufrufenden Stack verbunden.

1.4. Coroutinen ganz nach unten

yield fromErmöglicht, wie festgelegt, das Verbinden von zwei Bereichen über einen anderen Zwischenbereich. Bei rekursiver Anwendung bedeutet dies, dass die Oberseite des Stapels mit der Unterseite des Stapels verbunden werden kann.

root -\
  :    \-> coro_a -yield-from-> coro_b --\
  :/ <-+------------------------yield ---/
  |    :
  :\ --+-- coro_a.send----------yield ---\
  :                             coro_b <-/

Beachten Sie das rootund coro_bwissen nicht voneinander. Dies macht Coroutinen viel sauberer als Rückrufe: Coroutinen bauen immer noch auf einer 1: 1-Beziehung auf, wie Subroutinen. Coroutinen setzen ihren gesamten vorhandenen Ausführungsstapel bis zu einem regulären Aufrufpunkt aus und setzen ihn fort.

Insbesondere rootkönnte eine beliebige Anzahl von Coroutinen wieder aufgenommen werden. Es kann jedoch niemals mehr als eine gleichzeitig wieder aufnehmen. Coroutinen derselben Wurzel sind gleichzeitig, aber nicht parallel!

1.5. Pythons asyncundawait

Die Erklärung hat bisher ausdrücklich die verwendet yieldund yield fromVokabular von Generatoren - die zugrunde liegende Funktionalität ist das gleiche. Die neue Python3.5 Syntax asyncund awaitbesteht in erster Linie aus Gründen der Übersichtlichkeit.

def foo():  # subroutine?
     return None

def foo():  # coroutine?
     yield from foofoo()  # generator? coroutine?

async def foo():  # coroutine!
     await foofoo()  # coroutine!
     return None

Die Anweisungen async forund async withwerden benötigt, da Sie die yield from/awaitKette mit den Anweisungen bare forund withAnweisungen unterbrechen würden.

2. Anatomie einer einfachen Ereignisschleife

An sich hat eine Coroutine kein Konzept, einer anderen Coroutine die Kontrolle zu geben . Es kann nur dem Aufrufer am unteren Rand eines Coroutine-Stapels die Kontrolle geben. Dieser Anrufer kann dann zu einer anderen Coroutine wechseln und diese ausführen.

Dieser Wurzelknoten mehrerer Coroutinen ist üblicherweise eine Ereignisschleife : Bei Suspendierung liefert eine Coroutine ein Ereignis, bei dem sie fortgesetzt werden soll. Die Ereignisschleife kann wiederum effizient auf das Auftreten dieser Ereignisse warten. Auf diese Weise kann entschieden werden, welche Coroutine als Nächstes ausgeführt werden soll oder wie gewartet werden soll, bevor die Wiederaufnahme fortgesetzt wird.

Ein solches Design impliziert, dass es eine Reihe vordefinierter Ereignisse gibt, die die Schleife versteht. Mehrere Coroutinen awaiteinander, bis schließlich ein Ereignis awaited ist. Dieses Ereignis kann durch Steuerung direkt mit der Ereignisschleife kommunizieren yield.

loop -\
  :    \-> coroutine --await--> event --\
  :/ <-+----------------------- yield --/
  |    :
  |    :  # loop waits for event to happen
  |    :
  :\ --+-- send(reply) -------- yield --\
  :        coroutine <--yield-- event <-/

Der Schlüssel ist, dass die Coroutine-Suspendierung die direkte Kommunikation zwischen Ereignisschleife und Ereignissen ermöglicht. Der Zwischen-Coroutine-Stapel erfordert keine Kenntnisse darüber, welche Schleife ihn ausführt oder wie Ereignisse funktionieren.

2.1.1. Ereignisse in der Zeit

Das am einfachsten zu behandelnde Ereignis ist das Erreichen eines Zeitpunkts. Dies ist ebenfalls ein grundlegender Block von Thread-Code: Ein Thread wiederholt sleeps, bis eine Bedingung erfüllt ist. Eine reguläre sleepBlockierung der Ausführung von selbst - wir möchten, dass andere Coroutinen nicht blockiert werden. Stattdessen möchten wir der Ereignisschleife mitteilen, wann sie den aktuellen Coroutine-Stapel wieder aufnehmen soll.

2.1.2. Ein Ereignis definieren

Ein Ereignis ist einfach ein Wert, den wir identifizieren können - sei es über eine Aufzählung, einen Typ oder eine andere Identität. Wir können dies mit einer einfachen Klasse definieren, die unsere Zielzeit speichert. Zusätzlich zum Speichern der Ereignisinformationen können wir awaiteiner Klasse direkt erlauben .

class AsyncSleep:
    """Event to sleep until a point in time"""
    def __init__(self, until: float):
        self.until = until

    # used whenever someone ``await``s an instance of this Event
    def __await__(self):
        # yield this Event to the loop
        yield self

    def __repr__(self):
        return '%s(until=%.1f)' % (self.__class__.__name__, self.until)

Diese Klasse speichert nur das Ereignis - sie sagt nicht aus, wie sie tatsächlich behandelt werden soll.

Die einzige Besonderheit ist __await__- es ist das, wonach das awaitSchlüsselwort sucht. Praktisch ist es ein Iterator, aber nicht für die reguläre Iterationsmaschinerie verfügbar.

2.2.1. Warten auf ein Ereignis

Wie reagieren Coroutinen nach einem Ereignis darauf? Wir sollten in der Lage sein, das Äquivalent von sleepdurch awaitunsere Veranstaltung auszudrücken . Um besser zu sehen, was los ist, warten wir zweimal die halbe Zeit:

import time

async def asleep(duration: float):
    """await that ``duration`` seconds pass"""
    await AsyncSleep(time.time() + duration / 2)
    await AsyncSleep(time.time() + duration / 2)

Wir können diese Coroutine direkt instanziieren und ausführen. Ähnlich wie bei einem Generator wird coroutine.senddie Coroutine mit verwendet, bis ein yieldErgebnis erzielt wird .

coroutine = asleep(100)
while True:
    print(coroutine.send(None))
    time.sleep(0.1)

Dies gibt uns zwei AsyncSleepEreignisse und dann ein, StopIterationwenn die Coroutine fertig ist. Beachten Sie, dass die einzige Verzögerung von time.sleepin der Schleife ist! Jeder AsyncSleepspeichert nur einen Versatz von der aktuellen Zeit.

2.2.2. Ereignis + Schlaf

Zu diesem Zeitpunkt stehen uns zwei separate Mechanismen zur Verfügung:

• AsyncSleep Ereignisse, die innerhalb einer Coroutine ausgelöst werden können
• time.sleep das kann warten, ohne die Coroutinen zu beeinträchtigen

Bemerkenswerterweise sind diese beiden orthogonal: Keiner beeinflusst oder löst den anderen aus. Infolgedessen können wir unsere eigene Strategie entwickeln sleep, um die Verzögerung eines zu bewältigen AsyncSleep.

2.3. Eine naive Ereignisschleife

Wenn wir mehrere Coroutinen haben, kann jeder uns sagen, wann er geweckt werden möchte. Wir können dann warten, bis der erste von ihnen wieder aufgenommen werden möchte, dann auf den nachfolgenden und so weiter. Insbesondere kümmern wir uns an jedem Punkt nur darum, welcher der nächste ist .

Dies ermöglicht eine einfache Planung:

1. Sortieren Sie die Coroutinen nach der gewünschten Weckzeit
2. Wählen Sie die erste, die aufwachen möchte
3. Warten Sie bis zu diesem Zeitpunkt
4. Führen Sie diese Coroutine aus
5. Wiederholen von 1.

Eine triviale Implementierung erfordert keine fortgeschrittenen Konzepte. A listermöglicht das Sortieren von Coroutinen nach Datum. Warten ist eine regelmäßige time.sleep. Das Ausführen von Coroutinen funktioniert wie zuvor mit coroutine.send.

def run(*coroutines):
    """Cooperatively run all ``coroutines`` until completion"""
    # store wake-up-time and coroutines
    waiting = [(0, coroutine) for coroutine in coroutines]
    while waiting:
        # 2. pick the first coroutine that wants to wake up
        until, coroutine = waiting.pop(0)
        # 3. wait until this point in time
        time.sleep(max(0.0, until - time.time()))
        # 4. run this coroutine
        try:
            command = coroutine.send(None)
        except StopIteration:
            continue
        # 1. sort coroutines by their desired suspension
        if isinstance(command, AsyncSleep):
            waiting.append((command.until, coroutine))
            waiting.sort(key=lambda item: item[0])

Dies hat natürlich viel Raum für Verbesserungen. Wir können einen Heap für die Warteschlange oder eine Versandtabelle für Ereignisse verwenden. Wir könnten auch Rückgabewerte von der StopIterationabrufen und sie der Coroutine zuweisen. Das Grundprinzip bleibt jedoch dasselbe.

2.4. Genossenschaftliches Warten

Das AsyncSleepEreignis und die runEreignisschleife sind eine voll funktionsfähige Implementierung von zeitgesteuerten Ereignissen.

async def sleepy(identifier: str = "coroutine", count=5):
    for i in range(count):
        print(identifier, 'step', i + 1, 'at %.2f' % time.time())
        await asleep(0.1)

run(*(sleepy("coroutine %d" % j) for j in range(5)))

Dies schaltet kooperativ zwischen jeder der fünf Coroutinen um und unterbricht jede für 0,1 Sekunden. Obwohl die Ereignisschleife synchron ist, führt sie die Arbeit in 0,5 Sekunden statt in 2,5 Sekunden aus. Jede Coroutine hält den Zustand und handelt unabhängig.

3. E / A-Ereignisschleife

Eine Ereignisschleife, die unterstützt, sleepeignet sich zum Abrufen . Das Warten auf E / A in einem Dateihandle kann jedoch effizienter durchgeführt werden: Das Betriebssystem implementiert E / A und weiß somit, welche Handles bereit sind. Im Idealfall sollte eine Ereignisschleife ein explizites "Bereit für E / A" -Ereignis unterstützen.

3.1. Der selectAnruf

Python verfügt bereits über eine Schnittstelle zum Abfragen des Betriebssystems nach Lese-E / A-Handles. Beim Aufruf mit Handles zum Lesen oder Schreiben werden die Handles zurückgegeben , die zum Lesen oder Schreiben bereit sind :

readable, writeable, _ = select.select(rlist, wlist, xlist, timeout)

Zum Beispiel können wir openeine Datei zum Schreiben erstellen und warten, bis sie fertig ist:

write_target = open('/tmp/foo')
readable, writeable, _ = select.select([], [write_target], [])

Sobald select zurückkehrt, writeableenthält unsere geöffnete Datei.

3.2. Grundlegendes E / A-Ereignis

Ähnlich wie bei der AsyncSleepAnforderung müssen wir ein Ereignis für E / A definieren. Mit der zugrunde liegenden selectLogik muss sich das Ereignis auf ein lesbares Objekt beziehen - beispielsweise eine openDatei. Außerdem speichern wir, wie viele Daten gelesen werden sollen.

class AsyncRead:
    def __init__(self, file, amount=1):
        self.file = file
        self.amount = amount
        self._buffer = ''

    def __await__(self):
        while len(self._buffer) < self.amount:
            yield self
            # we only get here if ``read`` should not block
            self._buffer += self.file.read(1)
        return self._buffer

    def __repr__(self):
        return '%s(file=%s, amount=%d, progress=%d)' % (
            self.__class__.__name__, self.file, self.amount, len(self._buffer)
        )

Wie bei AsyncSleepspeichern wir meist nur die Daten, die für den zugrunde liegenden Systemaufruf erforderlich sind. Dieses Mal kann __await__es mehrmals fortgesetzt werden - bis unser Wunsch amountgelesen wurde. Darüber hinaus erhalten wir returndas E / A-Ergebnis, anstatt nur fortzufahren.

3.3. Erweitern einer Ereignisschleife mit Lese-E / A.

Die Basis für unsere Ereignisschleife ist immer noch die runzuvor definierte. Zuerst müssen wir die Leseanforderungen verfolgen. Dies ist kein sortierter Zeitplan mehr, wir ordnen nur Leseanforderungen Coroutinen zu.

# new
waiting_read = {}  # type: Dict[file, coroutine]

Da select.selectein Timeout-Parameter benötigt wird, können wir ihn anstelle von verwenden time.sleep.

# old
time.sleep(max(0.0, until - time.time()))
# new
readable, _, _ = select.select(list(reads), [], [])

Dies gibt uns alle lesbaren Dateien - falls vorhanden, führen wir die entsprechende Coroutine aus. Wenn es keine gibt, haben wir lange genug darauf gewartet, dass unsere aktuelle Coroutine ausgeführt wird.

# new - reschedule waiting coroutine, run readable coroutine
if readable:
    waiting.append((until, coroutine))
    waiting.sort()
    coroutine = waiting_read[readable[0]]

Schließlich müssen wir tatsächlich auf Leseanfragen warten.

# new
if isinstance(command, AsyncSleep):
    ...
elif isinstance(command, AsyncRead):
    ...

3.4. Etwas zusammensetzen

Das Obige war eine kleine Vereinfachung. Wir müssen etwas wechseln, um schlafende Coroutinen nicht zu verhungern, wenn wir immer lesen können. Wir müssen damit umgehen, nichts zu lesen oder zu warten. Das Endergebnis passt jedoch immer noch in 30 LOC.

def run(*coroutines):
    """Cooperatively run all ``coroutines`` until completion"""
    waiting_read = {}  # type: Dict[file, coroutine]
    waiting = [(0, coroutine) for coroutine in coroutines]
    while waiting or waiting_read:
        # 2. wait until the next coroutine may run or read ...
        try:
            until, coroutine = waiting.pop(0)
        except IndexError:
            until, coroutine = float('inf'), None
            readable, _, _ = select.select(list(waiting_read), [], [])
        else:
            readable, _, _ = select.select(list(waiting_read), [], [], max(0.0, until - time.time()))
        # ... and select the appropriate one
        if readable and time.time() < until:
            if until and coroutine:
                waiting.append((until, coroutine))
                waiting.sort()
            coroutine = waiting_read.pop(readable[0])
        # 3. run this coroutine
        try:
            command = coroutine.send(None)
        except StopIteration:
            continue
        # 1. sort coroutines by their desired suspension ...
        if isinstance(command, AsyncSleep):
            waiting.append((command.until, coroutine))
            waiting.sort(key=lambda item: item[0])
        # ... or register reads
        elif isinstance(command, AsyncRead):
            waiting_read[command.file] = coroutine

3.5. Genossenschaftliche E / A.

Die AsyncSleep, AsyncReadund runImplementierungen sind jetzt voll funktionsfähig zu schlafen und / oder zu lesen. Wie für sleepykönnen wir einen Helfer definieren, um das Lesen zu testen:

async def ready(path, amount=1024*32):
    print('read', path, 'at', '%d' % time.time())
    with open(path, 'rb') as file:
        result = return await AsyncRead(file, amount)
    print('done', path, 'at', '%d' % time.time())
    print('got', len(result), 'B')

run(sleepy('background', 5), ready('/dev/urandom'))

Wenn wir dies ausführen, können wir sehen, dass unsere E / A mit der Warteaufgabe verschachtelt ist:

id background round 1
read /dev/urandom at 1530721148
id background round 2
id background round 3
id background round 4
id background round 5
done /dev/urandom at 1530721148
got 1024 B

4. Nicht blockierende E / A.

Während E / A für Dateien das Konzept vermittelt, ist es für eine Bibliothek wie diese nicht wirklich geeignet asyncio: Der selectAufruf wird immer für Dateien und beides zurückgegeben openund readkann auf unbestimmte Zeit blockiert werden . Dies blockiert alle Coroutinen einer Ereignisschleife - was schlecht ist. Bibliotheken wie aiofilesThreads und Synchronisation verwenden, um nicht blockierende E / A und Ereignisse in der Datei zu fälschen.

Sockets ermöglichen jedoch nicht blockierende E / A - und ihre inhärente Latenz macht sie viel kritischer. Bei Verwendung in einer Ereignisschleife kann das Warten auf Daten und das erneute Versuchen abgeschlossen werden, ohne dass etwas blockiert wird.

4.1. Nicht blockierendes E / A-Ereignis

Ähnlich wie bei uns AsyncReadkönnen wir ein Suspend-and-Read-Ereignis für Sockets definieren. Anstatt eine Datei zu nehmen, nehmen wir einen Socket - der nicht blockierend sein darf. Auch unsere __await__Verwendungen socket.recvanstelle von file.read.

class AsyncRecv:
    def __init__(self, connection, amount=1, read_buffer=1024):
        assert not connection.getblocking(), 'connection must be non-blocking for async recv'
        self.connection = connection
        self.amount = amount
        self.read_buffer = read_buffer
        self._buffer = b''

    def __await__(self):
        while len(self._buffer) < self.amount:
            try:
                self._buffer += self.connection.recv(self.read_buffer)
            except BlockingIOError:
                yield self
        return self._buffer

    def __repr__(self):
        return '%s(file=%s, amount=%d, progress=%d)' % (
            self.__class__.__name__, self.connection, self.amount, len(self._buffer)
        )

Im Gegensatz zu AsyncRead, __await__blockiert nicht-führt wirklich I / O. Wenn Daten verfügbar sind, werden sie immer gelesen. Wenn keine Daten verfügbar sind, werden diese immer angehalten. Das heißt, die Ereignisschleife wird nur blockiert, während wir nützliche Arbeit leisten.

4.2. Entsperren der Ereignisschleife

An der Ereignisschleife ändert sich nicht viel. Das Ereignis, auf das gewartet werden soll, ist immer noch dasselbe wie für Dateien - ein Dateideskriptor, der als bereit markiert ist select.

# old
elif isinstance(command, AsyncRead):
    waiting_read[command.file] = coroutine
# new
elif isinstance(command, AsyncRead):
    waiting_read[command.file] = coroutine
elif isinstance(command, AsyncRecv):
    waiting_read[command.connection] = coroutine

An diesem Punkt sollte es offensichtlich sein, dass AsyncReadund AsyncRecvsind die gleiche Art von Ereignis. Wir könnten sie leicht zu einem Ereignis mit einer austauschbaren E / A-Komponente umgestalten . Tatsächlich trennen die Ereignisschleife, Coroutinen und Ereignisse einen Scheduler, einen beliebigen Zwischencode und die tatsächliche E / A sauber voneinander .

4.3. Die hässliche Seite der nicht blockierenden E / A.

Im Prinzip sollten Sie an dieser Stelle die Logik von readas recvfor wiederholen AsyncRecv. Dies ist jetzt jedoch viel hässlicher - Sie müssen mit frühen Rückgaben umgehen, wenn Funktionen im Kernel blockieren, aber Ihnen die Kontrolle geben. Zum Beispiel ist das Öffnen einer Verbindung viel länger als das Öffnen einer Datei:

# file
file = open(path, 'rb')
# non-blocking socket
connection = socket.socket()
connection.setblocking(False)
# open without blocking - retry on failure
try:
    connection.connect((url, port))
except BlockingIOError:
    pass

Kurz gesagt, es bleiben ein paar Dutzend Zeilen für die Ausnahmebehandlung. Die Ereignisse und die Ereignisschleife funktionieren bereits zu diesem Zeitpunkt.

id background round 1
read localhost:25000 at 1530783569
read /dev/urandom at 1530783569
done localhost:25000 at 1530783569 got 32768 B
id background round 2
id background round 3
id background round 4
done /dev/urandom at 1530783569 got 4096 B
id background round 5

Nachtrag

Beispielcode bei github

Ihre coro Desugaring ist konzeptionell korrekt, aber etwas unvollständig.

awaitwird nicht unbedingt angehalten, sondern nur, wenn ein blockierender Anruf auftritt. Woher weiß es, dass ein Anruf blockiert wird? Dies wird durch den erwarteten Code entschieden. Zum Beispiel könnte eine erwartete Implementierung des Socket-Lesens entschärft werden für:

def read(sock, n):
    # sock must be in non-blocking mode
    try:
        return sock.recv(n)
    except EWOULDBLOCK:
        event_loop.add_reader(sock.fileno, current_task())
        return SUSPEND

In echtem Asyncio der entsprechende Code den Status von a, Futureanstatt magische Werte zurückzugeben, aber das Konzept ist dasselbe. Bei entsprechender Anpassung an ein generatorähnliches Objekt kann der obige Code bearbeitet werden await.

Auf der Anruferseite, wenn Ihre Coroutine enthält:

data = await read(sock, 1024)

Es entgiftet in etwas in der Nähe von:

data = read(sock, 1024)
if data is SUSPEND:
    return SUSPEND
self.pos += 1
self.parts[self.pos](...)

Leute, die mit Generatoren vertraut sind, neigen dazu, das Obige in Bezug auf zu beschreiben yield from die Federung automatisch erfolgt.

Die Aufhängungskette wird bis zur Ereignisschleife fortgesetzt, die feststellt, dass die Coroutine angehalten ist, sie aus dem ausführbaren Satz entfernt und gegebenenfalls ausführbare Coroutinen ausführt. Wenn keine Coroutinen ausgeführt werden können, wartet die Schleife, select()bis einer der Dateideskriptoren, an denen eine Coroutine interessiert ist, für die E / A bereit ist. (Die Ereignisschleife verwaltet eine Zuordnung von Dateideskriptor zu Coroutine.)

Im obigen Beispiel wird die lesbare select()Ereignisschleife sockerneut hinzugefügt , sobald sie lesbar istcoro zum ausführbaren Satz , sodass sie ab dem Punkt der Unterbrechung fortgesetzt wird.

Mit anderen Worten:

1. Standardmäßig geschieht alles im selben Thread.

2. Die Ereignisschleife ist dafür verantwortlich, die Coroutinen zu planen und aufzuwecken, wenn alles, worauf sie gewartet haben (normalerweise ein normalerweise blockierender E / A-Aufruf oder eine Zeitüberschreitung), bereit ist.

Für einen Einblick in Coroutine-treibende Event-Loops empfehle ich diesen Vortrag von Dave Beazley, in dem er vor einem Live-Publikum das Codieren eines Event-Loops von Grund auf demonstriert.

Alles läuft auf die beiden Hauptherausforderungen hinaus, mit denen sich Asyncio befasst:

• Wie führe ich mehrere E / A in einem einzigen Thread durch?
• Wie implementiere ich kooperatives Multitasking?

Die Antwort auf den ersten Punkt gibt es schon lange und wird als Auswahlschleife bezeichnet . In Python ist es im Selektormodul implementiert .

Die zweite Frage bezieht sich auf das Konzept der Coroutine , dh Funktionen, die ihre Ausführung stoppen und später wiederhergestellt werden können. In Python werden Coroutinen mithilfe von Generatoren und dem Ertrag aus der Anweisung implementiert . Das ist es, was sich hinter dem versteckt asynchronen / wartenden Syntax verbirgt .

Weitere Ressourcen in dieser Antwort .

BEARBEITEN: Adressierung Ihres Kommentars zu Goroutinen:

Das nächste Äquivalent zu einer Goroutine in Asyncio ist eigentlich keine Coroutine, sondern eine Aufgabe (siehe den Unterschied in der Dokumentation ). In Python weiß eine Coroutine (oder ein Generator) nichts über die Konzepte der Ereignisschleife oder der E / A. Es ist einfach eine Funktion, die die Ausführung unter yieldBeibehaltung des aktuellen Status stoppen kann , sodass sie später wiederhergestellt werden kann. Die yield fromSyntax ermöglicht eine transparente Verkettung.

Innerhalb einer Asyncio-Aufgabe ergibt die Coroutine ganz unten in der Kette immer eine Zukunft . Diese Zukunft sprudelt dann in die Ereignisschleife und wird in die innere Maschinerie integriert. Wenn die Zukunft durch einen anderen inneren Rückruf festgelegt wird, kann die Ereignisschleife die Aufgabe wiederherstellen, indem die Zukunft zurück in die Coroutine-Kette gesendet wird.

BEARBEITEN: Beantworten einiger Fragen in Ihrem Beitrag:

Wie geschieht E / A in diesem Szenario? In einem separaten Thread? Ist der gesamte Dolmetscher gesperrt und erfolgt die E / A außerhalb des Dolmetschers?

Nein, in einem Thread passiert nichts. E / A wird immer von der Ereignisschleife verwaltet, hauptsächlich über Dateideskriptoren. Die Registrierung dieser Dateideskriptoren wird jedoch normalerweise von hochrangigen Coroutinen ausgeblendet, sodass die Drecksarbeit für Sie erledigt wird.

Was genau ist mit E / A gemeint? Wenn meine Python-Prozedur C open () -Prozedur heißt und ihrerseits einen Interrupt an den Kernel sendet und die Kontrolle an ihn abgibt, woher weiß der Python-Interpreter davon und kann weiterhin anderen Code ausführen, während der Kernel-Code das eigentliche I / ausführt? O und bis die Python-Prozedur aktiviert wird, die den Interrupt ursprünglich gesendet hat? Wie kann sich der Python-Interpreter im Prinzip dessen bewusst sein?

Eine E / A ist ein blockierender Anruf. In asyncio sollten alle E / A-Vorgänge die Ereignisschleife durchlaufen, da die Ereignisschleife, wie Sie sagten, nicht erkennen kann, dass ein blockierender Aufruf in einem synchronen Code ausgeführt wird. Das heißt, Sie sollten keine Synchronisation openim Kontext einer Coroutine verwenden. Verwenden Sie stattdessen eine dedizierte Bibliothek wie aiofiles, die eine asynchrone Version von bereitstellt open.