9.14. Sockety s asyncio¶

Blokující volání recv() zmrazí celý skript, dokud nedorazí byte. Blokující volání accept() obsluhuje vždy jen jednoho klienta. Obě tyto situace jsou přesně tím druhem „čekání na I/O“, k jehož řešení asyncio existuje. Kapitola o asyncio pokrývá smyčku událostí, korutiny a synchronizační primitiva; tato stránka pokrývá části specifické pro síť.

Modul asyncio zpřístupňuje síťování prostřednictvím malého počtu pomocníků, kteří přijímají a vracejí streamy – objekty vysoké úrovně, které obalují socket a nabízejí await-ovatelné verze čtení a zápisu. Podkladový socket tam stále je; aplikace se ho jen přímo nedotýká.

9.14.1. Klient s asyncio¶

asyncio.open_connection() je protějškem socket.socket.connect() v asyncio. Otevře TCP spojení a vrátí dva objekty streamu: reader a writer

import asyncio

async def client():
    reader, writer = await asyncio.open_connection("192.168.1.20", 9000)

    writer.write(b"hello\n")
    await writer.drain()                   # wait until bytes have been sent

    reply = await reader.readline()
    print("reply:", reply)

    writer.close()
    await writer.wait_closed()

asyncio.run(client())

Tři věci, kterých si všimnout:

Navázání spojení je jediné await místo blokujícího volání. Zatímco handshake probíhá, smyčka událostí může spouštět jiné korutiny.
write() vkládá byty do odchozího bufferu; drain() je ono await, které předá řízení smyčce, dokud tyto byty nejsou skutečně odeslány po síti.
readline() čte byty, dokud nedorazí znak nového řádku. Třída streamu zahrnuje také read() (přečte až N bytů) a readexactly() (přečte přesně N bytů), které řeší problém hranic zpráv v TCP, aniž byste museli rámovací smyčky psát ručně.

9.14.2. Server s asyncio¶

asyncio.start_server() je protějškem tance bind/listen/accept v asyncio. Přijímá callback, který se spustí jednou pro každé příchozí spojení, se stejnou dvojicí reader/writer, jakou používá klientská strana:

import asyncio

async def handle(reader, writer):
    addr = writer.get_extra_info("peername")
    print("connection from", addr)

    while True:
        data = await reader.read(1024)
        if not data:
            break
        writer.write(data)                 # echo back
        await writer.drain()

    writer.close()
    await writer.wait_closed()

async def main():
    server = await asyncio.start_server(handle, "0.0.0.0", 9000)
    print("listening on", server.sockets[0].getsockname())
    async with server:
        await server.serve_forever()

asyncio.run(main())

Každé přijaté spojení se stává vlastní úlohou spouštějící handle. Smyčka událostí mezi nimi přirozeně přepíná – jeden pomalý klient nemůže blokovat ostatní, protože zatímco čeká na await reader.read(...), smyčka může pokračovat v práci na každém dalším spojení. Přidání deseti souběžných klientů nevyžaduje deset vláken; všechny pohání tatáž jednovláknová smyčka událostí.

To je praktický důvod, proč se kamerové síťové aplikace napsané pro asyncio škálují mnohem lépe než ekvivalentní blokující kód: obrázek serveru na TCP sokety zvládal jednoho klienta naráz; tento zvládá mnoho klientů naráz bez jakéhokoli úsilí navíc.

9.14.3. Souběžná práce vedle síťování¶

Velkou výhodou je míchání síťování se zbytkem práce kamery ve stejné smyčce. Kamera může pořizovat snímek, provádět zpracování obrazu a obsluhovat síťový protokol, vše prokládaně:

import asyncio

async def capture_loop():
    while True:
        img = await camera.snapshot()
        # process img ...
        await asyncio.sleep_ms(100)

async def handle(reader, writer):
    ...

async def main():
    server = await asyncio.start_server(handle, "0.0.0.0", 9000)
    await asyncio.gather(
        server.serve_forever(),
        capture_loop(),
    )

asyncio.run(main())

asyncio.gather() spouští obě korutiny na stejné smyčce událostí. Zatímco kamera spí v sleep_ms() mezi snímky, server dostává prostor odbavovat síťový provoz. Zatímco server čeká na další byte, kamera dostává prostor pořizovat snímek. Oba postupují vpřed na jediném vláknu MicroPythonu.

9.14.4. UDP s asyncio¶

Modul asyncio nenabízí stejné streamy vysoké úrovně pro UDP – datagramy se nehodí do podoby čtení/zápisu streamu. Praktický přístup na kameře je umístit práci s UDP do vlastní korutiny, přepnout socket do neblokujícího režimu a mezi pokusy o čtení předávat řízení smyčce událostí:

import asyncio
import socket

async def udp_listener(port):
    s = socket.socket(socket.AF_INET, socket.SOCK_DGRAM)
    s.setblocking(False)
    s.bind(("0.0.0.0", port))

    while True:
        try:
            data, src = s.recvfrom(1024)
        except OSError:
            await asyncio.sleep_ms(10)
            continue
        print("got", data, "from", src)

Socket je nastaven jako neblokující pomocí s.setblocking(False), takže recvfrom() vyvolá OSError okamžitě, když žádný datagram nečeká, místo aby blokoval celou smyčku událostí. await asyncio.sleep_ms(10) v prázdné větvi předá řízení zpět smyčce událostí až do dalšího dotazu.

Odesílání má stejnou podobu: sendto() na neblokujícím socketu buď okamžitě uspěje, nebo vyvolá výjimku. Neexistuje žádné sendallto – UDP datagramy jsou atomické, takže každé odeslání je buď celý datagram, nebo žádný. Pokud je odesílací buffer plný, správným krokem u UDP je obvykle datagram zahodit a nechat ten další odejít při příštím průchodu smyčkou:

async def udp_telemetry(target_addr, period_ms):
    s = socket.socket(socket.AF_INET, socket.SOCK_DGRAM)
    s.setblocking(False)

    while True:
        payload = collect_telemetry()
        try:
            s.sendto(payload, target_addr)
        except OSError:
            pass                # buffer full -- skip this one

        await asyncio.sleep_ms(period_ms)

Selhávající větev je v praxi vzácná. UDP nemá řízení toku, takže sendto() téměř vždy uspěje na první pokus; except existuje hlavně proto, aby krátkodobá síťová porucha korutinu neshodila.

Sekce Asyncio pokrývá širší vzory pro míchání blokujícího I/O do programu s asyncio; stejné vzory platí přímo i pro UDP socket.

9.14.5. Časové limity a zrušení¶

Obalení síťového volání do asyncio.wait_for() mu nastaví časový limit:

try:
    reply = await asyncio.wait_for(reader.readline(), timeout=2.0)
except asyncio.TimeoutError:
    print("server is slow")

Korutinu, která trvá příliš dlouho, lze také cancel()-novat odjinud. Oba mechanismy jsou podrobně popsány v kapitole o koordinaci; platí beze změny pro streamy vracené asyncio.open_connection() a asyncio.start_server().

Úplnou referenci Stream (třída za readery a writery plus pomocníci, které tato stránka mimochodem použila) najdete v asyncio — asynchronní I/O plánovač.