To pytanie jest motywowane moim kolejnym pytaniem: Jak czekać w cdef?

W sieci jest mnóstwo artykułów i postów na blogach asyncio, ale wszystkie są bardzo powierzchowne. Nie mogłem znaleźć żadnych informacji o tym, jak asynciofaktycznie jest zaimplementowany i co sprawia, że ​​operacje we / wy są asynchroniczne. Próbowałem odczytać kod źródłowy, ale to tysiące wierszy kodu C nie najwyższej klasy, z których wiele dotyczy obiektów pomocniczych, ale co najważniejsze, trudno jest połączyć składnię Pythona z kodem C, który by przetłumaczył w.

Dokumentacja własna Asycnio jest jeszcze mniej pomocna. Nie ma tam żadnych informacji o tym, jak to działa, tylko kilka wskazówek, jak z niego korzystać, które są czasem mylące / bardzo słabo napisane.

Jestem zaznajomiony z implementacją coroutines w Go i miałem nadzieję, że Python zrobił to samo. Gdyby tak było, kod, który wymyśliłem w powyższym poście, zadziałałby. Ponieważ tak się nie stało, próbuję teraz dowiedzieć się, dlaczego. Jak dotąd moje przypuszczenie jest następujące, proszę poprawić mnie tam, gdzie się mylę:

1. Definicje procedur formularza async def foo(): ...są w rzeczywistości interpretowane jako metody dziedziczenia klas coroutine.
2. Być może w async defrzeczywistości jest podzielony na wiele metod przez awaitinstrukcje, w których obiekt, na którym te metody są wywoływane, jest w stanie śledzić dotychczasowy postęp, jaki poczynił podczas wykonywania.
3. Jeśli powyższe jest prawdą, to w zasadzie wykonanie programu sprowadza się do wywołania metod obiektu coroutine przez jakiegoś globalnego menedżera (pętlę?).
4. Menedżer globalny jest w jakiś sposób (jak?) Świadomy, kiedy operacje we / wy są wykonywane przez kod Pythona (tylko?) I jest w stanie wybrać jedną z oczekujących metod standardowych do wykonania po tym, jak bieżąca metoda wykonawcza zrezygnowała z kontroli (trafienie w awaitinstrukcję ).

Innymi słowy, oto moja próba „desugeringu” jakiejś asyncioskładni w coś bardziej zrozumiałego:

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()

Jeśli moje przypuszczenie okaże się słuszne: mam problem. Jak właściwie przebiega I / O w tym scenariuszu? W osobnym wątku? Czy cały tłumacz jest zawieszony, a wejścia / wyjścia odbywają się poza tłumaczem? Co dokładnie oznacza I / O? Jeśli moja procedura w Pythonie nazywa się procedurą C open(), a ona z kolei wysłała przerwanie do jądra, zrzekając się kontroli nad nim, w jaki sposób interpreter Pythona wie o tym i jest w stanie kontynuować wykonywanie innego kodu, podczas gdy kod jądra wykonuje rzeczywiste operacje we / wy i dopóki budzi procedurę Pythona, która pierwotnie wysłała przerwanie? W jaki sposób interpreter języka Python może w zasadzie być tego świadomy?

Jak działa asyncio?

Zanim odpowiemy na to pytanie, musimy zrozumieć kilka podstawowych terminów, pomiń je, jeśli już znasz któreś z nich.

Generatory

Generatory to obiekty, które pozwalają nam zawiesić wykonanie funkcji Pythona. Generatory wybrane przez użytkownika są implementowane przy użyciu słowa kluczowego yield. Tworząc normalną funkcję zawierającą yieldsłowo kluczowe, zamieniamy tę funkcję w 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

Jak widać, wywołanie next()generatora powoduje, że interpreter ładuje ramkę testu i zwraca yieldwartość ed. Wywołanie next()ponownie powoduje ponowne załadowanie ramki do stosu interpretera i kontynuowanie yieldkolejnej wartości.

Za trzecim razem next()nasz generator był skończony i StopIterationzostał wyrzucony.

Komunikacja z generatorem

Mniej znaną cechą generatorów jest to, że można się z nimi komunikować na dwa sposoby: send()i 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

Po wywołaniu gen.send()wartość jest przekazywana jako wartość zwracana przez yieldsłowo kluczowe.

gen.throw()z drugiej strony umożliwia rzucanie wyjątków wewnątrz generatorów, z wyjątkiem wywoływanego w tym samym miejscu yield.

Zwracanie wartości z generatorów

Zwrócenie wartości z generatora powoduje umieszczenie wartości wewnątrz StopIterationwyjątku. Możemy później odzyskać wartość z wyjątku i wykorzystać ją do naszych potrzeb.

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

Oto nowe słowo kluczowe: yield from

Python 3.4 przyszedł z dodaniem nowego hasła: yield from. Co to słowo pozwala nam zrobić, to przekazać każdy next(), send()a throw()do generatora wewnętrznego, najbardziej zagnieżdżonych. Jeśli generator wewnętrzny zwraca wartość, jest to również wartość zwracana z 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

Napisałem artykuł, aby dalej rozwinąć ten temat.

Kładąc wszystko razem

Po wprowadzeniu nowego słowa kluczowego yield fromw Pythonie 3.4 mogliśmy teraz tworzyć generatory wewnątrz generatorów, które podobnie jak tunel przekazują dane tam iz powrotem z generatorów najbardziej wewnętrznych do najbardziej zewnętrznych. To zrodziło nowe znaczenie dla generatorów - coroutines .

Coroutines to funkcje, które można zatrzymać i wznowić podczas działania. W Pythonie są definiowane za pomocą async defsłowa kluczowego. Podobnie jak generatory, oni też korzystać z własnej formy yield from, która jest await. Przed wprowadzeniem asynci awaitwprowadzeniem w Pythonie 3.5 tworzyliśmy procedury w dokładnie taki sam sposób, jak generatory (z yield fromzamiast await).

async def inner():
    return 1

async def outer():
    await inner()

Jak każdy iterator lub generator, który implementuje __iter__()metodę, programy te implementują, __await__()co pozwala im kontynuować działanie za każdym razem, gdy await corozostanie wywołany.

Jest ładny sekwencja schemat wewnątrz docs Pythona , które należy sprawdzić.

W asyncio oprócz podstawowych funkcji mamy 2 ważne obiekty: zadania i przyszłość .

Futures

Futures to obiekty, które mają __await__()zaimplementowaną metodę, a ich zadaniem jest utrzymanie określonego stanu i wyniku. Stan może być jednym z następujących:

1. PENDING - future nie ma żadnego wyniku ani zestawu wyjątków.
2. ANULOWANE - przyszłość została anulowana przy użyciu fut.cancel()
3. FINISHED - przyszłość została zakończona przez zestaw wyników za pomocą fut.set_result()lub przez zestaw wyjątków za pomocąfut.set_exception()

Rezultatem, tak jak się domyślasz, może być obiekt Pythona, który zostanie zwrócony, lub wyjątek, który może zostać zgłoszony.

Inną ważną cechą futureobiektów jest to, że zawierają metodę o nazwie add_done_callback(). Ta metoda umożliwia wywoływanie funkcji zaraz po wykonaniu zadania - niezależnie od tego, czy zgłosiło wyjątek, czy zostało zakończone.

Zadania

Obiekty zadań są specjalnymi przyszłościami, które owijają się wokół programów i komunikują się z najbardziej wewnętrznymi i najbardziej zewnętrznymi programami. Za każdym razem, gdy program jest awaitprzyszłością, przyszłość jest przekazywana z powrotem do zadania (tak jak w yield from) i zadanie ją otrzymuje.

Następnie zadanie wiąże się z przyszłością. Robi to, odwołując add_done_callback()się do przyszłości. Odtąd, jeśli przyszłość kiedykolwiek zostanie wykonana, poprzez anulowanie, przekazanie wyjątku lub przekazanie w rezultacie obiektu Pythona, zostanie wywołane wywołanie zwrotne zadania i powróci do istnienia.

Asyncio

Ostatnie palące pytanie, na które musimy odpowiedzieć, brzmi: w jaki sposób wdrażane jest IO?

W głębi asyncio mamy pętlę zdarzeń. Pętla zdarzeń zadań. Zadaniem pętli zdarzeń jest wywoływanie zadań za każdym razem, gdy są one gotowe, i koordynowanie całego wysiłku w jednej działającej maszynie.

Część IO pętli zdarzeń jest zbudowana na jednej kluczowej funkcji o nazwie select. Select to funkcja blokująca, zaimplementowana przez system operacyjny poniżej, która umożliwia czekanie w gniazdach na dane przychodzące lub wychodzące. Po odebraniu danych budzi się i zwraca gniazda, które odebrały dane, lub gniazda gotowe do zapisu.

Podczas próby odebrania lub wysłania danych przez gniazdo za pośrednictwem asyncio, to, co dzieje się poniżej, polega na tym, że gniazdo jest najpierw sprawdzane, czy zawiera dane, które można natychmiast odczytać lub wysłać. Jeśli jego .send()bufor jest pełny lub .recv()bufor jest pusty, gniazdo jest rejestrowane w selectfunkcji (po prostu dodając je do jednej z list rlistdla recvi wlistdla send), a odpowiednia funkcja jest awaitnowo utworzonym futureobiektem, powiązanym z tym gniazdem.

Gdy wszystkie dostępne zadania czekają na przyszłość, pętla zdarzeń wywołuje selecti czeka. Kiedy jedno z gniazd ma przychodzące dane lub jego sendbufor jest wyczerpany, asyncio sprawdza przyszły obiekt powiązany z tym gniazdem i ustawia go na gotowe.

Teraz dzieje się cała magia. Przyszłość jest gotowa do wykonania, zadanie, które dodało się wcześniej z add_done_callback(), .send()ożywa i wywołuje proces, który wznawia najbardziej wewnętrzną procedurę (z powodu awaitłańcucha) i odczytujesz nowo otrzymane dane z pobliskiego bufora. został rozlany do.

Łańcuch metod ponownie, w przypadku recv():

1. select.select czeka.
2. Zwracane jest gotowe gniazdo z danymi.
3. Dane z gniazda są przenoszone do bufora.
4. future.set_result() jest nazywany.
5. Zadanie, które dodało się za pomocą, add_done_callback()jest teraz obudzone.
6. Task .send()przywołuje program, który przechodzi do najbardziej wewnętrznego programu i budzi go.
7. Dane są odczytywane z bufora i zwracane naszemu skromnemu użytkownikowi.

Podsumowując, asyncio wykorzystuje możliwości generatora, które pozwalają na wstrzymywanie i wznawianie funkcji. Wykorzystuje yield frommożliwości, które pozwalają na przekazywanie danych tam iz powrotem z generatora najbardziej wewnętrznego do najbardziej zewnętrznego. Używa ich wszystkich w celu zatrzymania wykonywania funkcji podczas oczekiwania na zakończenie operacji we / wy (za pomocą selectfunkcji systemu operacyjnego ).

A co najlepsze? Podczas gdy jedna funkcja jest wstrzymana, inna może działać i przeplatać się z delikatną tkaniną, którą jest asyncio.

Mówienie o async/awaiti asyncioto nie to samo. Pierwsza to fundamentalna konstrukcja niskiego poziomu (coroutines), a druga to biblioteka korzystająca z tych konstrukcji. I odwrotnie, nie ma jednej ostatecznej odpowiedzi.

Poniżej znajduje się ogólny opis działania bibliotek async/awaiti asynciopodobnych. Oznacza to, że na górze mogą być inne sztuczki (są ...), ale nie mają one znaczenia, chyba że sam je zbudujesz. Różnica powinna być znikoma, chyba że wiesz już wystarczająco dużo, aby nie musieć zadawać takiego pytania.

1. Korekty a podprogramy w łupinie orzecha

Podobnie jak podprogramy (funkcje, procedury, ...), procedury (generatory, ...) są abstrakcją stosu wywołań i wskaźnika instrukcji: istnieje stos wykonujących się fragmentów kodu, a każdy znajduje się przy określonej instrukcji.

Rozróżnienie między defwersjami async defsłuży jedynie przejrzystości. Rzeczywista różnica jest w returnporównaniu yield. Z tego awaitlub yield fromweź różnicę z pojedynczych wywołań do całych stacków.

1.1. Podprogramy

Podprocedura reprezentuje nowy poziom stosu, na którym będą przechowywane zmienne lokalne, oraz pojedyncze przejście jego instrukcji, aby osiągnąć koniec. Rozważ podprogram taki jak ten:

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

To znaczy, kiedy go uruchamiasz

1. przydziel miejsce na stosie dla bariqux
2. rekurencyjnie wykonuje pierwszą instrukcję i przeskakuje do następnej
3. raz na raz return, umieść jego wartość na stosie wywołań
4. wyczyść stos (1.) i wskaźnik instrukcji (2.)

Warto zauważyć, że 4. oznacza, że ​​podprogram zawsze zaczyna się w tym samym stanie. Wszystko, co dotyczy samej funkcji, zostaje utracone po zakończeniu. Funkcji nie można wznowić, nawet jeśli są po niej instrukcje return.

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

1.2. Korekty jako trwałe podprogramy

Program jest podobny do podprogramu, ale może wyjść bez niszczenia jego stanu. Rozważmy taki program:

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

To znaczy, kiedy go uruchamiasz

1. przydziel miejsce na stosie dla bariqux
2. rekurencyjnie wykonuje pierwszą instrukcję i przeskakuje do następnej
   1. raz na yield, umieść jego wartość na stosie wywołań, ale zapisz stos i wskaźnik instrukcji
   2. po wywołaniu do yield, przywróć stos i wskaźnik instrukcji i wypchnij argumenty doqux
3. raz na raz return, umieść jego wartość na stosie wywołań
4. wyczyść stos (1.) i wskaźnik instrukcji (2.)

Zwróć uwagę na dodanie 2.1 i 2.2 - program można zawiesić i wznowić we wcześniej określonych punktach. Jest to podobne do zawieszenia podprogramu podczas wywoływania innego podprogramu. Różnica polega na tym, że aktywny coroutine nie jest ściśle powiązany ze swoim stosem wywołań. Zamiast tego zawieszony program jest częścią oddzielnego, izolowanego stosu.

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

Oznacza to, że zawieszone programy można dowolnie przechowywać lub przenosić między stosami. Każdy stos wywołań, który ma dostęp do programu, może zdecydować o jego wznowieniu.

1.3. Przechodzenie przez stos wywołań

Jak dotąd, nasz coroutine idzie w dół stosu wywołań tylko z yield. Podprogram może zejść w dół i w górę stosu wywołań za pomocą returni (). W celu zapewnienia kompletności, procedury potrzebują również mechanizmu, aby przejść w górę stosu wywołań. Rozważmy taki program:

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

Po uruchomieniu oznacza to, że nadal alokuje stos i wskaźnik instrukcji jak podprogram. Kiedy się zawiesza, nadal przypomina to zapisywanie podprogramu.

Jednak yield fromrobi jedno i drugie . Zawiesza stos i wskaźnik instrukcji wrap i działa cofoo. Zauważ, że wrappozostaje zawieszony do cofoocałkowitego zakończenia. Zawsze, gdy cofoozawiesza się lub coś jest wysyłane, cofoojest bezpośrednio podłączane do stosu wywołań.

1.4. Korekty w dół

Jak ustalono, yield fromumożliwia połączenie dwóch zakresów w innym pośrednim. W przypadku zastosowania rekurencyjnego oznacza to, że góra stosu może być połączona z dnem stosu.

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

Zwróć na to uwagę rooti coro_bnie wiedzcie o sobie. To sprawia, że ​​programy są znacznie czystsze niż wywołania zwrotne: programy nadal są zbudowane na relacji 1: 1, podobnie jak podprogramy. Koordynatorzy zawieszają i wznawiają cały istniejący stos wykonania aż do zwykłego punktu wywołania.

W szczególności rootmoże mieć dowolną liczbę programów do wznowienia. Jednak nigdy nie może wznowić więcej niż jednego w tym samym czasie. Korekty tego samego rdzenia są współbieżne, ale nie równoległe!

1.5. Pythona asynciawait

Wyjaśnieniem tej pory jednoznacznie używany yieldi yield fromsłownictwo generatorów - funkcjonalność bazowym jest taka sama. Nowa składnia Pythona 3.5 asynci awaitistnieje głównie dla przejrzystości.

def foo():  # subroutine?
     return None

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

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

Instrukcje async fori async withsą potrzebne, ponieważ przerwałbyś yield from/awaitłańcuch instrukcjami gołymi fori with.

2. Anatomia prostej pętli zdarzeń

Sam w sobie program nie ma koncepcji poddania kontroli innemu programowi. Może przekazać kontrolę tylko wywołującemu na dole stosu coroutine. Ten wywołujący może następnie przełączyć się na inny program i uruchomić go.

Ten węzeł główny kilku programów jest zwykle pętlą zdarzeń : w przypadku zawieszenia, program generuje zdarzenie, od którego chce wznowić. Z kolei pętla zdarzeń jest w stanie skutecznie czekać na wystąpienie tych zdarzeń. Dzięki temu może zdecydować, który program ma zostać uruchomiony jako następny lub jak czekać przed wznowieniem.

Taki projekt oznacza, że ​​istnieje zestaw predefiniowanych zdarzeń, które rozumie pętla. Kilka programów współpracuje awaitze sobą, aż w końcu odbywa się wydarzenie await. To zdarzenie może komunikować się bezpośrednio z pętlą zdarzeń przez yieldsterowanie.

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

Kluczem jest to, że standardowe zawieszenie umożliwia bezpośrednią komunikację pętli zdarzeń i zdarzeń. Pośredni stos programu coroutine nie wymaga żadnej wiedzy o tym, która pętla go uruchamia ani jak działają zdarzenia.

2.1.1. Wydarzenia w czasie

Najprostszym zdarzeniem do obsłużenia jest osiągnięcie punktu w czasie. Jest to również podstawowy blok kodu z wątkami: wątek jest powtarzany, sleepaż warunek zostanie spełniony. Jednak zwykłe sleepwykonywanie bloków samo w sobie - chcemy, aby inne programy nie były blokowane. Zamiast tego chcemy powiedzieć pętli zdarzeń, kiedy powinna wznowić bieżący stos programu.

2.1.2. Definiowanie wydarzenia

Zdarzenie to po prostu wartość, którą możemy zidentyfikować - czy to poprzez wyliczenie, typ czy inną tożsamość. Możemy to zdefiniować za pomocą prostej klasy, która przechowuje nasz docelowy czas. Oprócz przechowywania informacji o wydarzeniach, możemy pozwolić awaitbezpośrednio na zajęcia.

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)

Ta klasa tylko przechowuje zdarzenie - nie mówi, jak właściwie je obsłużyć.

Jedyną specjalną cechą jest __await__to, czego awaitszuka słowo kluczowe. W praktyce jest to iterator, ale nie jest dostępny dla zwykłych maszyn iteracyjnych.

2.2.1. Oczekiwanie na wydarzenie

Teraz, gdy mamy wydarzenie, jak reagują na to programy? Powinniśmy być w stanie wyrazić odpowiednikiem sleepprzez awaiting naszą imprezę. Aby lepiej zobaczyć, co się dzieje, przez połowę czasu czekamy dwa razy:

import time

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

Możemy bezpośrednio utworzyć instancję i uruchomić tę procedurę. Podobnie jak w przypadku generatora, użycie coroutine.sendpowoduje uruchomienie programu aż do uzyskania yieldwyniku.

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

To daje nam dwa AsyncSleepzdarzenia, a następnie StopIterationmoment zakończenia programu. Zwróć uwagę, że jedyne opóźnienie pochodzi z time.sleeppętli! Każdy AsyncSleepzapisuje tylko przesunięcie od bieżącego czasu.

2.2.2. Wydarzenie + uśpienie

W tym momencie mamy do dyspozycji dwa odrębne mechanizmy:

• AsyncSleep Zdarzenia, które można wywołać z wnętrza programu
• time.sleep które mogą czekać bez wpływu na programy

Warto zauważyć, że te dwa są ortogonalne: żaden z nich nie wpływa ani nie uruchamia drugiego. W rezultacie możemy opracować własną strategię, sleepaby sprostać opóźnieniu związanemu z plikiem AsyncSleep.

2.3. Naiwna pętla wydarzeń

Jeśli mamy kilka programów, każdy może nam powiedzieć, kiedy chce się obudzić. Możemy wtedy zaczekać, aż pierwszy z nich będzie chciał wznowić, potem następny i tak dalej. Warto zauważyć, że w każdym punkcie zależy nam tylko na tym, który z nich będzie następny .

To sprawia, że ​​planowanie jest proste:

1. posortuj programy według pożądanego czasu przebudzenia
2. wybierz pierwszą, która chce się obudzić
3. poczekaj do tego momentu
4. uruchom ten program
5. powtórz od 1.

Banalna implementacja nie wymaga żadnych zaawansowanych koncepcji. A listumożliwia sortowanie programów według daty. Czekanie jest normalne time.sleep. Uruchamianie programów działa tak samo jak wcześniej z 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])

Oczywiście jest to dużo miejsca na ulepszenia. Możemy użyć sterty dla kolejki oczekiwania lub tabeli wysyłkowej dla zdarzeń. Moglibyśmy również pobrać wartości zwracane z programu StopIterationi przypisać je do programu. Jednak podstawowa zasada pozostaje ta sama.

2.4. Czekanie w spółdzielni

AsyncSleepWydarzenie i runpętla zdarzenie to wdrożenie w pełni robocze czasowe zdarzeń.

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)))

To wspólnie przełącza się między każdym z pięciu programów, zawieszając każdy na 0,1 sekundy. Mimo że pętla zdarzeń jest synchroniczna, nadal wykonuje pracę w 0,5 sekundy zamiast 2,5 sekundy. Każdy program zachowuje stan i działa niezależnie.

3. Pętla zdarzeń we / wy

Pętla zdarzeń, która obsługuje, sleepjest odpowiednia do sondowania . Jednak oczekiwanie na I / O na uchwycie pliku może być wykonane bardziej wydajnie: system operacyjny implementuje I / O i wie, które uchwyty są gotowe. W idealnym przypadku pętla zdarzeń powinna obsługiwać jawne zdarzenie „gotowe do wejścia / wyjścia”.

3.1. selectwezwanie

Python ma już interfejs do wysyłania zapytań do systemu operacyjnego w celu odczytania uchwytów we / wy. Gdy jest wywoływana z uchwytami do odczytu lub zapisu, zwraca uchwyty gotowe do odczytu lub zapisu:

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

Na przykład możemy openplik do zapisu i poczekać aż będzie gotowy:

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

Po wybraniu zwraca, writeablezawiera nasz otwarty plik.

3.2. Podstawowe zdarzenie we / wy

Podobnie jak w przypadku AsyncSleepżądania, musimy zdefiniować zdarzenie dla I / O. Zgodnie z selectlogiką bazową zdarzenie musi odnosić się do czytelnego obiektu - powiedzmy do openpliku. Ponadto przechowujemy, ile danych do odczytania.

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)
        )

Podobnie jak w przypadku AsyncSleep, przechowujemy głównie dane wymagane dla podstawowego wywołania systemowego. Tym razem __await__możliwe jest wielokrotne wznawianie - aż do przeczytania naszego pożądanego amount. Ponadto otrzymujemy returnwynik I / O zamiast po prostu wznawiać.

3.3. Rozszerzanie pętli zdarzeń o odczyt we / wy

Podstawą naszej pętli zdarzeń jest nadal runzdefiniowana wcześniej. Najpierw musimy śledzić żądania odczytu. To nie jest już uporządkowany harmonogram, tylko mapujemy żądania odczytu do korektorów.

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

Ponieważ select.selectprzyjmuje parametr timeout, możemy go użyć zamiast time.sleep.

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

W ten sposób otrzymujemy wszystkie czytelne pliki - jeśli takie istnieją, uruchamiamy odpowiedni program. Jeśli ich nie ma, czekaliśmy wystarczająco długo na uruchomienie naszego obecnego programu.

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

Wreszcie, musimy faktycznie nasłuchiwać żądań odczytu.

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

3.4. Składając to razem

Powyższe było trochę uproszczeniem. Musimy trochę przejść, żeby nie głodować śpiących programów, jeśli zawsze potrafimy czytać. Musimy sobie poradzić, nie mając nic do czytania lub nie czekając. Jednak wynik końcowy nadal mieści się w 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. Współpraca we / wy

Te AsyncSleep, AsyncReadi runimplementacje są teraz w pełni funkcjonalny do snu i / lub przeczyta. Tak samo jak w przypadku sleepy, możemy zdefiniować pomocnika do testowania czytania:

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'))

Uruchamiając to, widzimy, że nasze I / O są przeplatane z oczekującym zadaniem:

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. Nieblokujące we / wy

Chociaż operacje we / wy na plikach są zrozumiałe, nie są one odpowiednie dla takich bibliotek, jak asyncio: selectwywołanie zawsze zwraca pliki i oba openi readmogą blokować się na czas nieokreślony . To blokuje wszystkie procedury pętli zdarzeń - co jest złe. Biblioteki lubiąaiofiles używają wątków i synchronizacji do fałszywego nieblokującego wejścia / wyjścia i zdarzeń w pliku.

Jednak gniazda pozwalają na nieblokujące wejścia / wyjścia - a ich nieodłączne opóźnienie sprawia, że ​​jest to znacznie bardziej krytyczne. W przypadku użycia w pętli zdarzeń oczekiwanie na dane i ponawianie próby można zawinąć bez blokowania czegokolwiek.

4.1. Nieblokujące zdarzenie we / wy

Podobnie jak w naszym przypadku AsyncRead, możemy zdefiniować zdarzenie wstrzymania i odczytu dla gniazd. Zamiast pobierać plik, bierzemy gniazdo - które musi być nieblokujące. Ponadto nasze __await__zastosowania socket.recvzamiast 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)
        )

W przeciwieństwie do AsyncRead, __await__wykonuje naprawdę nieblokujące operacje we / wy. Kiedy dane są dostępne, zawsze czyta. Gdy żadne dane nie są dostępne, zawiesza się zawsze . Oznacza to, że pętla zdarzeń jest blokowana tylko wtedy, gdy wykonujemy pożyteczną pracę.

4.2. Odblokowanie pętli zdarzeń

Jeśli chodzi o pętlę zdarzeń, nic się nie zmienia. Zdarzenie do nasłuchiwania jest nadal takie samo jak w przypadku plików - deskryptor pliku oznaczony jako gotowy 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

W tym momencie powinno być oczywiste, że AsyncReadi AsyncRecvsą to tego samego rodzaju wydarzenia. Moglibyśmy łatwo przekształcić je w jedno zdarzenie z wymiennym komponentem I / O. W efekcie pętla zdarzeń, procedury i zdarzenia wyraźnie oddzielają program planujący, dowolny kod pośredni i rzeczywiste operacje wejścia / wyjścia.

4.3. Brzydka strona nieblokujących wejść / wyjść

Zasadniczo to, co powinieneś zrobić w tym momencie, to powtórzyć logikę readas a recvfor AsyncRecv. Jednak teraz jest to o wiele bardziej brzydkie - musisz obsługiwać wczesne zwroty, gdy funkcje blokują się wewnątrz jądra, ale dają ci kontrolę. Na przykład otwarcie połączenia w porównaniu z otwarciem pliku jest znacznie dłuższe:

# 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

Krótko mówiąc, pozostaje kilkadziesiąt wierszy obsługi wyjątków. W tym momencie zdarzenia i pętla zdarzeń już działają.

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

Uzupełnienie

Przykładowy kod na github

Twoje corousuwanie cukru jest koncepcyjnie poprawne, ale nieco niekompletne.

awaitnie zawiesza się bezwarunkowo, ale tylko wtedy, gdy napotka połączenie blokujące. Skąd wie, że połączenie jest blokowane? Decyduje o tym oczekiwany kod. Na przykład oczekiwana implementacja odczytu gniazda może zostać pozbawiona cukru:

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

W prawdziwym asyncio kod równoważny modyfikuje stan a Futurezamiast zwracać wartości magiczne, ale koncepcja jest taka sama. Po odpowiednim dostosowaniu do obiektu podobnego do generatora powyższy kod można awaitedytować.

Po stronie dzwoniącego, gdy Twój program zawiera:

data = await read(sock, 1024)

Rozpada się w coś bliskiego:

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

Osoby zaznajomione z generatorami mają tendencję do opisywania powyższego w kategoriach, w yield fromktórych zawieszenie następuje automatycznie.

Łańcuch zawieszenia jest kontynuowany aż do pętli zdarzeń, która zauważa, że ​​program jest zawieszony, usuwa go z zestawu, który można uruchomić, i wykonuje programy, które można uruchomić, jeśli takie istnieją. Jeśli nie można uruchomić żadnego programu, pętla czeka, select()aż deskryptor pliku, którym program jest zainteresowany, stanie się gotowy do operacji we / wy. (Pętla zdarzeń zachowuje mapowanie deskryptora pliku do programu).

W powyższym przykładzie, gdy select()powie pętli zdarzenia, która sockjest czytelna, zostanie ona ponownie dodana corodo zestawu, który można uruchomić, więc będzie kontynuowana od punktu zawieszenia.

Innymi słowy:

1. Domyślnie wszystko dzieje się w tym samym wątku.

2. Pętla zdarzeń jest odpowiedzialna za planowanie korektorów i budzenie ich, gdy cokolwiek czekały (zwykle wywołanie IO, które normalnie by się blokowało lub przekroczono limit czasu), staje się gotowe.

Aby dowiedzieć się więcej na temat pętli wydarzeń związanych z jazdą regularnie, polecam wykład Dave'a Beazleya, w którym demonstruje on kodowanie pętli wydarzenia od zera przed publicznością na żywo.

Wszystko sprowadza się do dwóch głównych wyzwań, którym zajmuje się asyncio:

• Jak wykonać wiele operacji we / wy w jednym wątku?
• Jak wdrożyć współpracę wielozadaniową?

Odpowiedź na pierwszy punkt istnieje już od dłuższego czasu i nazywa się pętlą wyboru . W Pythonie jest zaimplementowany w module selektorów .

Drugie pytanie wiąże się z koncepcją coroutine , czyli funkcji, które mogą zatrzymać ich wykonanie i zostać przywrócone później. W Pythonie programy są implementowane przy użyciu generatorów i instrukcji yield from . To właśnie kryje się za składnią async / await .

Więcej zasobów w tej odpowiedzi .

EDYCJA: adresowanie twojego komentarza na temat goroutines:

Najbliższy odpowiednik gorutyny w asyncio w rzeczywistości nie jest coroutine, ale zadaniem (zobacz różnicę w dokumentacji ). W Pythonie program (lub generator) nie wie nic o pojęciach pętli zdarzeń lub I / O. Jest to po prostu funkcja, która może zatrzymać swoje wykonanie yield, zachowując swój bieżący stan, dzięki czemu można ją później przywrócić. yield fromSkładnia umożliwia łączenia ich w przejrzysty sposób.

Teraz, w ramach zadania asyncio, program na samym dole łańcucha zawsze kończy się określeniem przyszłości . Ta przyszłość pędzi do pętli zdarzeń i zostaje zintegrowana z wewnętrzną maszynerią. Gdy przyszłość jest ustawiona na wykonanie przez inne wewnętrzne wywołanie zwrotne, pętla zdarzeń może przywrócić zadanie, wysyłając przyszłość z powrotem do łańcucha coroutine.

EDYCJA: Odpowiadanie na niektóre pytania w Twoim poście:

Jak właściwie przebiega I / O w tym scenariuszu? W osobnym wątku? Czy cały tłumacz jest zawieszony, a wejścia / wyjścia odbywają się poza tłumaczem?

Nie, nic się nie dzieje w wątku. We / wy jest zawsze zarządzane przez pętlę zdarzeń, głównie poprzez deskryptory plików. Jednak rejestracja tych deskryptorów plików jest zwykle ukryta przez programy wysokiego poziomu, co czyni dla ciebie brudną robotę.

Co dokładnie oznacza I / O? Jeśli moja procedura w Pythonie nazywała się procedurą C open (), a ona z kolei wysłała przerwanie do jądra, zrzekając się nad nim kontroli, w jaki sposób interpreter Pythona wie o tym i jest w stanie kontynuować wykonywanie innego kodu, podczas gdy kod jądra wykonuje rzeczywisty I / O i dopóki nie obudzi procedury Pythona, która pierwotnie wysłała przerwanie? W jaki sposób interpreter języka Python może w zasadzie być tego świadomy?

I / O to każde wywołanie blokujące. W asyncio wszystkie operacje we / wy powinny przechodzić przez pętlę zdarzeń, ponieważ, jak powiedziałeś, pętla zdarzeń nie ma sposobu, aby być świadomym, że wywołanie blokujące jest wykonywane w pewnym kodzie synchronicznym. Oznacza to, że nie powinieneś używać synchronicznego openw kontekście coroutine. Zamiast tego użyj dedykowanej biblioteki, takiej jak aiofiles, która zapewnia asynchroniczną wersję open.