Pertanyaan ini dimotivasi oleh pertanyaan saya yang lain: Bagaimana cara menunggu di cdef?

Ada banyak sekali artikel dan postingan blog di web tentang asyncio, tetapi semuanya sangat dangkal. Saya tidak dapat menemukan informasi apa pun tentang bagaimana asynciosebenarnya diterapkan, dan apa yang membuat I / O asinkron. Saya mencoba membaca kode sumber, tetapi itu ribuan baris bukan kode C kelas tertinggi, banyak di antaranya berkaitan dengan objek tambahan, tetapi yang paling penting, sulit untuk menghubungkan antara sintaks Python dan kode C apa yang akan diterjemahkan ke.

Dokumentasi Asycnio sendiri bahkan kurang membantu. Tidak ada informasi di sana tentang cara kerjanya, hanya beberapa pedoman tentang cara menggunakannya, yang terkadang juga menyesatkan / ditulis dengan sangat buruk.

Saya akrab dengan implementasi coroutine Go, dan agak berharap Python melakukan hal yang sama. Jika itu masalahnya, kode yang saya dapatkan di pos yang ditautkan di atas akan berfungsi. Karena tidak, saya sekarang mencoba mencari tahu mengapa. Tebakan terbaik saya sejauh ini adalah sebagai berikut, tolong perbaiki jika saya salah:

1. Definisi prosedur bentuk async def foo(): ...sebenarnya diartikan sebagai metode mewarisi kelas coroutine.
2. Mungkin, async defsebenarnya dipecah menjadi beberapa metode berdasarkan awaitpernyataan, di mana objek, tempat metode ini dipanggil, dapat melacak kemajuan yang dibuatnya melalui eksekusi sejauh ini.
3. Jika hal di atas benar, maka, pada dasarnya, eksekusi coroutine bermuara pada metode pemanggilan objek coroutine oleh beberapa manajer global (loop?).
4. Manajer global entah bagaimana (bagaimana?) Mengetahui kapan operasi I / O dilakukan oleh kode Python (hanya?) Dan dapat memilih salah satu metode coroutine yang tertunda untuk dieksekusi setelah metode eksekusi saat ini melepaskan kontrol (klik pada awaitpernyataan ).

Dengan kata lain, inilah upaya saya untuk "mendesain" beberapa asynciosintaks menjadi sesuatu yang lebih dapat dimengerti:

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

Jika tebakan saya terbukti benar: maka saya punya masalah. Bagaimana sebenarnya I / O terjadi dalam skenario ini? Di utas terpisah? Apakah seluruh juru bahasa ditangguhkan dan I / O terjadi di luar penerjemah? Apa sebenarnya yang dimaksud dengan I / O? Jika prosedur python saya disebut prosedur C open(), dan pada gilirannya mengirim interupsi ke kernel, melepaskan kontrol padanya, bagaimana penerjemah Python tahu tentang ini dan dapat terus menjalankan beberapa kode lain, sementara kode kernel melakukan I / O yang sebenarnya dan sampai itu membangunkan prosedur Python yang mengirim interupsi aslinya? Bagaimana interpreter Python pada prinsipnya menyadari hal ini terjadi?

Bagaimana cara kerja asyncio?

Sebelum menjawab pertanyaan ini kita perlu memahami beberapa istilah dasar, lewati ini jika Anda sudah mengetahuinya.

Generator

Generator adalah objek yang memungkinkan kita untuk menangguhkan eksekusi fungsi python. Generator yang dikurasi pengguna diimplementasikan menggunakan kata kunci yield. Dengan membuat fungsi normal yang berisi yieldkata kunci, kami mengubah fungsi itu menjadi 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

Seperti yang Anda lihat, memanggil next()generator menyebabkan interpreter memuat frame pengujian, dan mengembalikan nilai yielded. Memanggil next()lagi, menyebabkan frame memuat lagi ke tumpukan interpreter, dan melanjutkan ke yieldnilai lain.

Pada saat ketiga next()dipanggil, generator kami selesai, dan StopIterationterlempar.

Berkomunikasi dengan generator

Fitur generator yang kurang dikenal adalah kenyataan bahwa Anda dapat berkomunikasi dengan mereka menggunakan dua metode: send()dan 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

Setelah memanggil gen.send(), nilai tersebut diteruskan sebagai nilai kembali dari yieldkata kunci.

gen.throw()di sisi lain, memungkinkan untuk melempar Pengecualian ke dalam generator, dengan pengecualian yang dimunculkan di tempat yang sama yielddipanggil.

Mengembalikan nilai dari generator

Mengembalikan nilai dari generator, menghasilkan nilai yang dimasukkan ke dalam StopIterationpengecualian. Nanti kita dapat memulihkan nilai dari pengecualian dan menggunakannya untuk kebutuhan kita.

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

Lihatlah, kata kunci baru: yield from

Python 3.4 datang dengan penambahan kata kunci baru: yield from. Apa yang memungkinkan kata kunci itu untuk kita lakukan, meneruskan apa saja next(), send()dan throw()menjadi generator bersarang paling dalam. Jika generator bagian dalam mengembalikan nilai, itu juga merupakan nilai pengembalian 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

Saya telah menulis artikel untuk menguraikan lebih lanjut tentang topik ini.

Menyatukan semuanya

Setelah memperkenalkan kata kunci baru yield fromdi Python 3.4, kami sekarang dapat membuat generator di dalam generator yang seperti terowongan, meneruskan data bolak-balik dari generator paling dalam ke paling luar. Ini telah melahirkan arti baru bagi generator - coroutine .

Coroutine adalah fungsi yang dapat dihentikan dan dilanjutkan saat dijalankan. Di Python, mereka didefinisikan menggunakan async defkata kunci. Sama seperti generator, mereka juga menggunakan bentuk mereka sendiri yield fromyang await. Sebelum asyncdan awaitdiperkenalkan dengan Python 3.5, kami membuat coroutine dengan cara yang sama persis dengan generator yang dibuat (dengan yield fromalih - alih await).

async def inner():
    return 1

async def outer():
    await inner()

Seperti setiap iterator atau generator yang mengimplementasikan __iter__()metode ini, implementasi coroutine __await__()yang memungkinkan mereka untuk melanjutkan setiap kali await corodipanggil.

Ada diagram urutan yang bagus di dalam dokumen Python yang harus Anda periksa.

Dalam asyncio, selain fungsi coroutine, kami memiliki 2 objek penting: tugas dan masa depan .

Futures

Futures adalah objek yang __await__()metodenya telah diterapkan, dan tugasnya adalah mempertahankan keadaan dan hasil tertentu. Negara bagian dapat menjadi salah satu dari berikut ini:

1. TERTUNDA - masa depan tidak memiliki hasil atau pengecualian apa pun.
2. DIBATALKAN - masa depan dibatalkan menggunakan fut.cancel()
3. SELESAI - masa depan telah selesai, baik dengan set hasil menggunakan fut.set_result()atau dengan set pengecualian menggunakanfut.set_exception()

Hasilnya, seperti yang Anda tebak, bisa berupa objek Python, yang akan dikembalikan, atau pengecualian yang mungkin dimunculkan.

Lain yang penting fitur dari futurebenda-benda, adalah bahwa mereka mengandung metode yang disebut add_done_callback(). Metode ini memungkinkan fungsi dipanggil segera setelah tugas selesai - baik itu memunculkan pengecualian atau selesai.

Tugas

Objek tugas adalah masa depan khusus, yang membungkus coroutine, dan berkomunikasi dengan coroutine paling dalam dan paling luar. Setiap kali coroutine awaitsa masa depan, masa depan diteruskan kembali ke tugas (seperti di yield from), dan tugas menerimanya.

Selanjutnya, tugas tersebut mengikat dirinya ke masa depan. Ia melakukannya dengan meneleponadd_done_callback() masa depan. Mulai sekarang, jika masa depan akan dilakukan, baik dengan dibatalkan, melewati pengecualian, atau meneruskan objek Python sebagai hasilnya, callback tugas akan dipanggil, dan akan bangkit kembali.

Asyncio

Pertanyaan terakhir yang harus kita jawab adalah - bagaimana IO diimplementasikan?

Jauh di dalam asyncio, kami memiliki loop acara. Perulangan tugas. Tugas loop acara adalah memanggil tugas setiap kali mereka siap dan mengoordinasikan semua upaya itu ke dalam satu mesin yang berfungsi.

Bagian IO dari event loop dibangun di atas satu fungsi penting yang dipanggil select. Select adalah fungsi pemblokiran, diimplementasikan oleh sistem operasi di bawahnya, yang memungkinkan menunggu di soket untuk data masuk atau keluar. Setelah data diterima, ia bangun, dan mengembalikan soket yang menerima data, atau soket yang siap untuk ditulis.

Ketika Anda mencoba menerima atau mengirim data melalui soket melalui asyncio, yang sebenarnya terjadi di bawah ini adalah soket diperiksa terlebih dahulu jika ada data yang dapat segera dibaca atau dikirim. Jika .send()buffernya penuh, atau .recv()buffernya kosong, soket didaftarkan ke selectfungsi (hanya dengan menambahkannya ke salah satu daftar, rlistuntuk recvdan wlistuntuk send) dan fungsi yang sesuai yang awaitbaru dibuatfuture , terikat ke soket itu.

Ketika semua tugas yang tersedia menunggu masa depan, event loop memanggil selectdan menunggu. Ketika salah satu soket memiliki data yang masuk, atau sendbuffernya terkuras, asyncio memeriksa objek masa depan yang terkait dengan soket itu, dan menyetelnya ke selesai.

Sekarang semua keajaiban terjadi. Masa depan diatur untuk selesai, tugas yang ditambahkan sebelumnya dengan add_done_callback()bangkit kembali, dan memanggil .send()coroutine yang melanjutkan coroutine paling dalam (karena awaitrantai) dan Anda membaca data yang baru diterima dari buffer terdekat itu tumpah ke.

Rantai metode lagi, jika recv():

1. select.select menunggu.
2. Soket yang siap, dengan data dikembalikan.
3. Data dari soket dipindahkan ke buffer.
4. future.set_result() disebut.
5. Tugas yang ditambahkan dengan sendirinya add_done_callback()sekarang sudah aktif.
6. Tugas memanggil .send()coroutine yang masuk ke coroutine paling dalam dan membangunkannya.
7. Data sedang dibaca dari buffer dan dikembalikan ke pengguna kami yang sederhana.

Singkatnya, asyncio menggunakan kapabilitas generator, yang memungkinkan menjeda dan melanjutkan fungsi. Ini menggunakan yield fromkemampuan yang memungkinkan melewatkan data bolak-balik dari generator paling dalam ke paling luar. Ia menggunakan semua itu untuk menghentikan eksekusi fungsi sambil menunggu IO selesai (dengan menggunakan selectfungsi OS ).

Dan yang terbaik dari semuanya? Saat satu fungsi dijeda, fungsi lainnya mungkin berjalan dan menyatu dengan kain halus, yaitu asyncio.

Berbicara tentang async/awaitdan asynciobukanlah hal yang sama. Yang pertama adalah konstruksi dasar tingkat rendah (coroutine) sedangkan yang berikutnya adalah pustaka yang menggunakan konstruksi ini. Sebaliknya, tidak ada jawaban akhir tunggal.

Berikut ini adalah gambaran umum tentang bagaimana async/awaitdan asyncio-seperti perpustakaan bekerja. Artinya, mungkin ada trik lain di atas (ada ...) tetapi trik itu tidak penting kecuali Anda membuatnya sendiri. Perbedaannya harus dapat diabaikan kecuali Anda sudah cukup tahu untuk tidak perlu mengajukan pertanyaan seperti itu.

1. Coroutine versus subrutin dalam kulit kacang

Sama seperti subrutin (fungsi, prosedur, ...), coroutine (generator, ...) adalah abstraksi tumpukan panggilan dan penunjuk instruksi: ada setumpuk potongan kode yang sedang dieksekusi, dan masing-masing berada pada instruksi tertentu.

Perbedaan defversus async defhanya untuk kejelasan. Perbedaan sebenarnya adalah returnversus yield. Dari sini, awaitatau yield fromambil perbedaan dari panggilan individu ke seluruh tumpukan.

1.1. Subrutin

Sebuah subrutin mewakili tingkat tumpukan baru untuk menampung variabel lokal, dan satu traversal instruksi untuk mencapai tujuan. Pertimbangkan subrutin seperti ini:

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

Saat Anda menjalankannya, itu artinya

1. mengalokasikan ruang tumpukan untuk bardanqux
2. jalankan pernyataan pertama secara rekursif dan lompat ke pernyataan berikutnya
3. sekali per return, dorong nilainya ke stack pemanggil
4. bersihkan tumpukan (1.) dan penunjuk instruksi (2.)

Khususnya, 4. berarti bahwa subrutin selalu dimulai pada keadaan yang sama. Segala sesuatu yang eksklusif untuk fungsi itu sendiri akan hilang setelah selesai. Suatu fungsi tidak dapat dilanjutkan, meskipun ada instruksi setelahnya return.

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

1.2. Coroutine sebagai subrutin yang persisten

Coroutine seperti subrutin, tetapi dapat keluar tanpa merusak statusnya. Pertimbangkan coroutine seperti ini:

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

Saat Anda menjalankannya, itu artinya

1. mengalokasikan ruang tumpukan untuk bardanqux
2. jalankan pernyataan pertama secara rekursif dan lompat ke pernyataan berikutnya
   1. sekali pada a yield, dorong nilainya ke stack pemanggil tetapi simpan stack dan penunjuk instruksi
   2. sekali memanggil yield, pulihkan tumpukan dan penunjuk instruksi dan dorong argumen kequx
3. sekali per return, dorong nilainya ke stack pemanggil
4. bersihkan tumpukan (1.) dan penunjuk instruksi (2.)

Perhatikan penambahan 2.1 dan 2.2 - coroutine dapat ditangguhkan dan dilanjutkan pada poin yang telah ditentukan. Ini mirip dengan bagaimana subrutin ditangguhkan selama pemanggilan subrutin lain. Perbedaannya adalah bahwa coroutine aktif tidak terikat secara ketat ke stack pemanggilnya. Sebaliknya, coroutine yang ditangguhkan adalah bagian dari tumpukan terpisah dan terisolasi.

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

Ini berarti coroutine yang ditangguhkan dapat disimpan atau dipindahkan dengan bebas di antara tumpukan. Setiap tumpukan panggilan yang memiliki akses ke coroutine dapat memutuskan untuk melanjutkannya.

1.3. Melintasi tumpukan panggilan

Sejauh ini, coroutine kami hanya menggunakan call stack yield. Sebuah subrutin bisa turun dan naik tumpukan panggilan dengan returndan (). Untuk kelengkapan, coroutine juga membutuhkan mekanisme untuk menaikkan tumpukan panggilan. Pertimbangkan coroutine seperti ini:

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

Ketika Anda menjalankannya, itu berarti masih mengalokasikan stack dan penunjuk instruksi seperti subrutin. Saat dihentikan, itu masih seperti menyimpan subrutin.

Namun, yield fromlakukan keduanya . Ini menangguhkan tumpukan dan penunjuk instruksi wrap dan berjalan cofoo. Perhatikan bahwa wraptetap ditangguhkan sampai cofooselesai sepenuhnya. Setiap kali cofoomenangguhkan atau sesuatu dikirim, cofoolangsung terhubung ke stack panggilan.

1.4. Coroutine sampai ke bawah

Seperti yang sudah mapan, yield frommemungkinkan untuk menghubungkan dua cakupan di satu lingkup perantara lainnya. Ketika diterapkan secara rekursif, itu berarti bagian atas tumpukan dapat dihubungkan ke bagian bawah tumpukan.

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

Perhatikan itu rootdan coro_btidak tahu tentang satu sama lain. Ini membuat coroutine jauh lebih bersih daripada callback: coroutine masih dibangun di atas relasi 1: 1 seperti subrutin. Coroutine menangguhkan dan melanjutkan seluruh tumpukan eksekusi yang ada hingga titik panggilan biasa.

Khususnya, rootdapat memiliki jumlah coroutine yang berubah-ubah untuk dilanjutkan. Namun, itu tidak pernah bisa melanjutkan lebih dari satu pada waktu yang sama. Coroutine dari root yang sama bersifat bersamaan tetapi tidak paralel!

1.5. Python asyncdanawait

Penjelasannya sejauh ini secara eksplisit menggunakan kosakata yielddan yield fromgenerator - fungsionalitas yang mendasarinya sama. Sintaks Python3.5 baru asyncdan awaitada terutama untuk kejelasan.

def foo():  # subroutine?
     return None

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

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

The async fordan async withpernyataan diperlukan karena Anda akan memutus yield from/awaitrantai dengan telanjang fordan withpernyataan.

2. Anatomi lingkaran peristiwa sederhana

Dengan sendirinya, coroutine tidak memiliki konsep untuk menyerahkan kendali ke coroutine lain . Itu hanya dapat menghasilkan kontrol ke pemanggil di bagian bawah tumpukan coroutine. Penelepon ini kemudian dapat beralih ke coroutine lain dan menjalankannya.

Node akar dari beberapa coroutine ini biasanya merupakan loop peristiwa : saat ditangguhkan, coroutine menghasilkan peristiwa yang ingin dilanjutkan. Pada gilirannya, loop peristiwa mampu menunggu peristiwa ini terjadi secara efisien. Ini memungkinkannya untuk memutuskan coroutine mana yang akan dijalankan berikutnya, atau bagaimana menunggu sebelum melanjutkan.

Desain seperti itu menyiratkan bahwa ada satu set peristiwa yang telah ditentukan sebelumnya yang dipahami oleh loop. Beberapa coroutine awaitsatu sama lain, sampai akhirnya ada event await. Peristiwa ini dapat berkomunikasi secara langsung dengan perulangan peristiwa dengan yieldkontrol.

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

Kuncinya adalah penangguhan coroutine memungkinkan loop peristiwa dan peristiwa untuk berkomunikasi secara langsung. Tumpukan coroutine menengah tidak memerlukan pengetahuan apa pun tentang loop mana yang menjalankannya, atau cara kerja peristiwa.

2.1.1. Acara tepat waktu

Peristiwa paling sederhana untuk ditangani adalah mencapai suatu titik waktu. Ini adalah blok fundamental dari kode berulir juga: utas berulang kali sleepsampai kondisi benar. Namun, sleepeksekusi blok biasa dengan sendirinya - kami ingin coroutine lain tidak diblokir. Sebagai gantinya, kami ingin memberi tahu event loop kapan harus melanjutkan tumpukan coroutine saat ini.

2.1.2. Mendefinisikan Acara

Peristiwa hanyalah nilai yang dapat kita identifikasi - baik itu melalui enum, jenis atau identitas lainnya. Kita dapat mendefinisikan ini dengan kelas sederhana yang menyimpan waktu target kita. Selain menyimpan informasi acara, kami dapat mengizinkan ke awaitkelas secara langsung.

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)

Kelas ini hanya menyimpan acara - tidak mengatakan bagaimana sebenarnya menanganinya.

Satu-satunya fitur khusus adalah __await__- itulah yang awaitdicari kata kunci. Secara praktis, ini adalah iterator tetapi tidak tersedia untuk mesin iterasi biasa.

2.2.1. Menunggu acara

Sekarang kita punya acara, bagaimana reaksi coroutine terhadapnya? Kita harus bisa mengungkapkan padanannya sleepdengan awaitacara kita. Untuk lebih melihat apa yang sedang terjadi, kami menunggu dua kali untuk separuh waktu:

import time

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

Kita dapat langsung membuat instance dan menjalankan coroutine ini. Mirip dengan generator, menggunakan coroutine.sendmenjalankan coroutine sampai yieldhasilnya.

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

Ini memberi kita dua AsyncSleepperistiwa dan kemudian StopIterationsaat coroutine selesai. Perhatikan bahwa satu-satunya penundaan adalah dari time.sleepdalam loop! Masing-masing AsyncSleephanya menyimpan offset dari waktu saat ini.

2.2.2. Acara + Tidur

Pada titik ini, kami memiliki dua mekanisme terpisah yang kami miliki:

• AsyncSleep Peristiwa yang dapat dihasilkan dari dalam coroutine
• time.sleep yang dapat menunggu tanpa memengaruhi coroutine

Khususnya, keduanya ortogonal: tidak satu pun yang memengaruhi atau memicu yang lain. Alhasil, kita bisa memikirkan strategi kita sendiri sleepuntuk mengatasi keterlambatan sebuah AsyncSleep.

2.3. Perulangan peristiwa yang naif

Jika kami memiliki beberapa coroutine, masing-masing dapat memberi tahu kami kapan ingin dibangunkan. Kemudian kita bisa menunggu sampai yang pertama ingin dilanjutkan, lalu yang setelahnya, dan seterusnya. Khususnya, di setiap titik kami hanya peduli tentang mana yang berikutnya .

Ini membuat penjadwalan menjadi mudah:

1. urutkan coroutine berdasarkan waktu bangun yang diinginkan
2. pilih yang pertama ingin bangun
3. tunggu sampai saat ini
4. jalankan coroutine ini
5. ulangi dari 1.

Implementasi sepele tidak membutuhkan konsep lanjutan. A listmemungkinkan untuk mengurutkan coroutine berdasarkan tanggal. Menunggu adalah hal yang biasa time.sleep. Menjalankan coroutine berfungsi seperti sebelumnya dengan 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])

Tentu saja, ini memiliki banyak ruang untuk perbaikan. Kita bisa menggunakan heap untuk antrian tunggu atau tabel pengiriman untuk acara. Kita juga bisa mengambil nilai kembali dariStopIteration dan menetapkannya ke coroutine. Namun, prinsip dasarnya tetap sama.

2.4. Koperasi Menunggu

The AsyncSleepevent dan runevent loop adalah implementasi sepenuhnya bekerja peristiwa waktunya.

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

Ini secara kooperatif beralih di antara masing-masing dari lima coroutine, menangguhkan masing-masing selama 0,1 detik. Meskipun event loop sinkron, event loop masih mengeksekusi pekerjaan dalam 0,5 detik, bukan 2,5 detik. Setiap coroutine memegang status dan bertindak secara independen.

3. Perulangan peristiwa I / O

Perulangan peristiwa yang mendukung sleepcocok untuk polling . Namun, menunggu I / O pada pegangan file dapat dilakukan dengan lebih efisien: sistem operasi mengimplementasikan I / O dan dengan demikian mengetahui pegangan mana yang siap. Idealnya, event loop harus mendukung event eksplisit "ready for I / O".

3.1. The selectpanggilan

Python sudah memiliki antarmuka untuk meminta OS membaca pegangan I / O. Saat dipanggil dengan tuas untuk membaca atau menulis, ia mengembalikan tuas siap untuk membaca atau menulis:

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

Misalnya, kita dapat openmembuat file untuk ditulis dan menunggu sampai siap:

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

Setelah memilih kembali, writeableberisi file terbuka kami.

3.2. Acara I / O dasar

Mirip dengan AsyncSleeppermintaan tersebut, kita perlu mendefinisikan acara untuk I / O. Dengan selectlogika yang mendasarinya , acara tersebut harus mengacu pada objek yang dapat dibaca - misalnya openfile. Selain itu, kami menyimpan berapa banyak data untuk dibaca.

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

Seperti AsyncSleepkita kebanyakan hanya menyimpan data yang diperlukan untuk panggilan sistem yang mendasarinya. Kali ini, __await__dapat dilanjutkan beberapa kali - sampai keinginan kita amounttelah terbaca. Selain itu, kami mendapatkan returnhasil I / O, bukan hanya melanjutkan.

3.3. Menambahkan event loop dengan read I / O

Basis untuk loop acara kami masih runditentukan sebelumnya. Pertama, kita perlu melacak permintaan baca. Ini bukan lagi jadwal yang diurutkan, kami hanya memetakan permintaan baca ke coroutine.

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

Karena select.selectmengambil parameter waktu tunggu, kita dapat menggunakannya sebagai pengganti time.sleep.

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

Ini memberi kita semua file yang dapat dibaca - jika ada, kita menjalankan coroutine yang sesuai. Jika tidak ada, kita harus menunggu cukup lama agar coroutine kita saat ini berjalan.

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

Akhirnya, kami harus benar-benar mendengarkan permintaan baca.

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

3.4. Menyatukannya

Di atas adalah sedikit penyederhanaan. Kita perlu melakukan beberapa peralihan ke pola tidur tidak kelaparan jika kita selalu bisa membaca. Kita perlu menangani tidak ada untuk dibaca atau tidak ada yang menunggu. Namun, hasil akhirnya masih cocok dengan 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. Koperasi I / O

The AsyncSleep, AsyncReaddan runimplementasi yang sekarang sepenuhnya fungsional untuk tidur dan / atau membaca. Sama seperti sleepy, kita bisa mendefinisikan helper untuk menguji bacaan:

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

Dengan menjalankan ini, kita dapat melihat bahwa I / O kita diselingi dengan tugas menunggu:

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. I / O Non-Pemblokiran

Sementara I / O pada file mendapatkan konsepnya, itu tidak benar-benar cocok untuk perpustakaan seperti asyncio: selectpanggilan selalu kembali untuk file , dan keduanya opendan readdapat memblokir tanpa batas . Ini memblokir semua coroutine dari sebuah event loop - yang buruk. Pustaka seperti aiofilesmenggunakan utas dan sinkronisasi ke I / O non-pemblokiran palsu dan acara di file.

Namun, soket memungkinkan untuk non-pemblokiran I / O - dan latensi yang melekat membuatnya jauh lebih kritis. Saat digunakan dalam event loop, menunggu data dan mencoba kembali dapat digabungkan tanpa memblokir apa pun.

4.1. Peristiwa I / O Non-Pemblokiran

Mirip dengan kami AsyncRead, kami dapat menentukan acara suspend-and-read untuk soket. Alih-alih mengambil file, kami mengambil soket - yang tidak boleh memblokir. Juga, kami __await__menggunakan, socket.recvbukan 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)
        )

Sebaliknya AsyncRead, __await__menjalankan I / O yang benar-benar tidak memblokir. Saat data tersedia, data selalu terbaca. Jika tidak ada data yang tersedia, itu selalu ditangguhkan. Itu berarti event loop hanya diblokir saat kami melakukan pekerjaan yang berguna.

4.2. Buka blokir event loop

Sejauh menyangkut loop acara, tidak banyak yang berubah. Peristiwa yang akan didengarkan masih sama dengan untuk file - deskriptor file yang ditandai siap oleh 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

Pada titik ini, harus jelas bahwa AsyncReaddan AsyncRecvmerupakan jenis peristiwa yang sama. Kami dapat dengan mudah merefaktornya menjadi satu acara dengan komponen I / O yang dapat ditukar. Akibatnya, event loop, coroutine, dan event dengan rapi memisahkan penjadwal, kode perantara arbitrer dan I / O aktual.

4.3. Sisi jelek dari I / O non-pemblokiran

Pada prinsipnya, apa yang harus Anda lakukan saat ini adalah mereplikasi logika readas a recvfor AsyncRecv. Namun, ini jauh lebih buruk sekarang - Anda harus menangani pengembalian awal ketika fungsi memblokir di dalam kernel, tetapi kontrol hasil kepada Anda. Misalnya, membuka koneksi versus membuka file jauh lebih lama:

# 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

Singkat cerita, yang tersisa adalah beberapa lusin baris penanganan Exception. Peristiwa dan loop peristiwa sudah berfungsi pada saat ini.

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

Tambahan

Contoh kode di github

coroDesugaring Anda secara konseptual benar, tetapi sedikit tidak lengkap.

awaittidak menangguhkan tanpa syarat, tetapi hanya jika menemui panggilan pemblokiran. Bagaimana cara mengetahui bahwa panggilan diblokir? Ini ditentukan oleh kode yang sedang menunggu. Misalnya, implementasi socket read yang menunggu dapat diinginkan untuk:

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

Dalam asyncio nyata, kode yang setara mengubah status a Futurealih - alih mengembalikan nilai ajaib, tetapi konsepnya sama. Jika diadaptasi dengan tepat ke objek seperti generator, kode di atas dapat diubah await.

Di sisi penelepon, jika coroutine Anda berisi:

data = await read(sock, 1024)

Ini menggambarkan sesuatu yang dekat dengan:

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

Orang yang akrab dengan generator cenderung menggambarkan hal di atas yield fromyang melakukan suspensi secara otomatis.

Rantai suspensi berlanjut hingga ke loop kejadian, yang memperhatikan bahwa coroutine ditangguhkan, menghapusnya dari set yang dapat dijalankan, dan melanjutkan untuk mengeksekusi coroutine yang dapat dijalankan, jika ada. Jika tidak ada coroutine yang dapat dijalankan, loop akan menunggu select()hingga deskriptor file yang diinginkan oleh coroutine siap untuk IO. (Event loop mempertahankan pemetaan file-deskriptor-to-coroutine.)

Dalam contoh di atas, setelah select()memberi tahu loop peristiwa yang sockdapat dibaca, itu akan ditambahkan kembali coroke set yang dapat dijalankan, sehingga akan dilanjutkan dari titik penangguhan.

Dengan kata lain:

1. Semuanya terjadi di utas yang sama secara default.

2. Perulangan peristiwa bertanggung jawab untuk menjadwalkan coroutine dan membangunkannya ketika apa pun yang mereka tunggu (biasanya panggilan IO yang biasanya diblokir, atau batas waktu) menjadi siap.

Untuk wawasan tentang loop acara yang mendorong coroutine, saya merekomendasikan ceramah oleh Dave Beazley ini, di mana dia mendemonstrasikan pengkodean loop acara dari awal di depan audiens langsung.

Itu semua bermuara pada dua tantangan utama yang ditangani asyncio:

• Bagaimana cara melakukan beberapa I / O dalam satu thread?
• Bagaimana cara menerapkan multitasking kooperatif?

Jawaban untuk poin pertama telah ada sejak lama dan disebut loop seleksi . Dalam python, ini diimplementasikan dalam modul penyeleksi .

Pertanyaan kedua terkait dengan konsep coroutine , yaitu fungsi yang dapat menghentikan eksekusinya dan dikembalikan nanti. Dalam python, coroutine diimplementasikan menggunakan generator dan pernyataan yield from . Itulah yang bersembunyi di balik sintaks async / await .

Lebih banyak sumber daya dalam jawaban ini .

EDIT: Mengatasi komentar Anda tentang goroutine:

Persamaan terdekat dengan goroutine di asyncio sebenarnya bukanlah coroutine tetapi sebuah tugas (lihat perbedaannya dalam dokumentasi ). Dalam python, coroutine (atau generator) tidak tahu apa-apa tentang konsep event loop atau I / O. Ini hanyalah sebuah fungsi yang dapat menghentikan penggunaan eksekusinya yieldsambil mempertahankan statusnya saat ini, sehingga dapat dipulihkan nanti. The yield fromsintaks memungkinkan untuk chaining mereka dengan cara yang transparan.

Sekarang, dalam tugas asyncio, coroutine di bagian paling bawah rantai selalu menghasilkan masa depan . Masa depan ini kemudian menggelembung ke putaran peristiwa, dan diintegrasikan ke dalam mesin bagian dalam. Saat masa depan disetel untuk dilakukan oleh beberapa callback internal lainnya, loop peristiwa dapat memulihkan tugas dengan mengirim masa depan kembali ke rantai coroutine.

EDIT: Mengatasi beberapa pertanyaan di posting Anda:

Bagaimana sebenarnya I / O terjadi dalam skenario ini? Di utas terpisah? Apakah seluruh juru bahasa ditangguhkan dan I / O terjadi di luar penerjemah?

Tidak, tidak ada yang terjadi di utas. I / O selalu dikelola oleh event loop, kebanyakan melalui deskriptor file. Namun pendaftaran deskriptor file tersebut biasanya disembunyikan oleh coroutine tingkat tinggi, membuat pekerjaan kotor untuk Anda.

Apa sebenarnya yang dimaksud dengan I / O? Jika prosedur python saya disebut C open () prosedur, dan pada gilirannya mengirim interupsi ke kernel, melepaskan kendali padanya, bagaimana penerjemah Python tahu tentang ini dan dapat terus menjalankan beberapa kode lain, sementara kode kernel melakukan I / yang sebenarnya O dan sampai membangunkan prosedur Python yang mengirim interupsi aslinya? Bagaimana interpreter Python pada prinsipnya menyadari hal ini terjadi?

I / O adalah panggilan pemblokiran apa pun. Dalam asyncio, semua operasi I / O harus melalui event loop, karena seperti yang Anda katakan, event loop tidak memiliki cara untuk mengetahui bahwa panggilan pemblokiran sedang dilakukan dalam beberapa kode sinkron. Itu berarti Anda tidak seharusnya menggunakan sinkronisasi opendalam konteks coroutine. Sebagai gantinya, gunakan pustaka khusus seperti aiofiles yang menyediakan versi asinkron open.