9.14. Socket con asyncio¶

Una chiamata bloccante recv() congela l’intero script finché non arriva un byte. Una chiamata bloccante accept() serve un solo client alla volta. Entrambe sono esattamente il tipo di situazione di «attesa su I/O» per gestire la quale esiste asyncio. Il capitolo su asyncio tratta il loop di eventi, le coroutine e le primitive di sincronizzazione; questa pagina tratta le parti specifiche della rete.

Il modulo asyncio espone la rete attraverso un piccolo numero di helper che accettano e restituiscono stream – oggetti di alto livello che incapsulano un socket e offrono versioni await-abili di lettura e scrittura. Il socket sottostante è ancora presente; l’applicazione semplicemente non lo tocca direttamente.

9.14.1. Un client con asyncio¶

asyncio.open_connection() è la controparte asyncio di socket.socket.connect(). Apre una connessione TCP e restituisce due oggetti stream: un reader e un 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())

Tre cose da notare:

L’impostazione della connessione è un singolo await invece di una chiamata bloccante. Mentre l’handshake è in corso, il loop di eventi è libero di eseguire altre coroutine.
write() mette i byte in un buffer di uscita; drain() è l”await che cede il controllo al loop finché quei byte non sono stati effettivamente inviati sulla rete.
readline() legge i byte finché non arriva un newline. La classe stream include anche read() (legge fino a N byte) e readexactly() (legge esattamente N byte), che risolvono il problema dei confini dei messaggi di TCP senza dover scrivere a mano i loop di framing.

9.14.2. Un server con asyncio¶

asyncio.start_server() è la controparte asyncio della sequenza bind/listen/accept. Accetta una callback che verrà eseguita una volta per ogni connessione in entrata, con la stessa coppia reader/writer usata dal lato client:

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

Ogni connessione accettata diventa il proprio task che esegue handle. Il loop di eventi smista tra di esse in modo naturale – un client lento non può bloccare gli altri, perché mentre è in attesa su await reader.read(...) il loop è libero di fare progressi su ogni altra connessione. Aggiungere dieci client concorrenti non richiede dieci thread; lo stesso loop di eventi a thread singolo li gestisce tutti.

Questo è il motivo pratico per cui le applicazioni di rete per camera scritte per asyncio scalano molto meglio del codice bloccante equivalente: lo scenario del server su Socket TCP era un-client-alla-volta; questo è molti-client-contemporaneamente senza alcuno sforzo aggiuntivo.

9.14.3. Lavoro concorrente insieme alla rete¶

Il grande vantaggio è combinare la rete con il resto del lavoro della camera nello stesso loop. La camera può catturare un frame, eseguire l’elaborazione delle immagini e servire un protocollo di rete, il tutto in modo interlacciato:

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() esegue le due coroutine sullo stesso loop di eventi. Mentre la camera dorme in sleep_ms() tra un frame e l’altro, il server può smistare il traffico di rete. Mentre il server attende il byte successivo, la camera può catturare. Entrambi fanno progressi su un singolo thread MicroPython.

9.14.4. UDP con asyncio¶

Il modulo asyncio non offre gli stessi stream di alto livello per UDP – i datagrammi non si adattano alla forma di lettura/scrittura di uno stream. L’approccio pratico sulla camera è mettere il lavoro UDP nella propria coroutine, commutare il socket in modalità non bloccante e cedere il controllo al loop di eventi tra i tentativi di lettura:

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)

Il socket è impostato come non bloccante con s.setblocking(False), in modo che recvfrom() sollevi OSError immediatamente quando nessun datagramma è in attesa, invece di bloccare l’intero loop di eventi. L”await asyncio.sleep_ms(10) nel ramo vuoto restituisce il controllo al loop di eventi fino al polling successivo.

L’invio segue la stessa forma: sendto() su un socket non bloccante o riesce immediatamente o solleva un’eccezione. Non esiste sendallto – i datagrammi UDP sono atomici, quindi ogni invio è un intero datagramma o nessuno. Se il buffer di invio è pieno, la mossa giusta per UDP è di solito scartare il datagramma e lasciare che il successivo esca al passaggio successivo nel loop:

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)

Il ramo di fallimento è raro nella pratica. UDP non ha controllo di flusso, quindi sendto() ha quasi sempre successo al primo tentativo; l”except esiste per lo più affinché un breve intoppo di rete non faccia crashare la coroutine.

La sezione Asyncio tratta i pattern più ampi per integrare l’I/O bloccante in un programma asyncio; gli stessi pattern si applicano direttamente a un socket UDP.

9.14.5. Timeout e cancellazione¶

Avvolgere una chiamata di rete in asyncio.wait_for() le impone una scadenza:

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

Una coroutine che impiega troppo tempo può anche essere cancellata con cancel() da altrove. Entrambi i meccanismi sono trattati in dettaglio nel capitolo sul coordinamento; si applicano invariati agli stream restituiti da asyncio.open_connection() e asyncio.start_server().

Per il riferimento completo di Stream (la classe alla base dei reader e dei writer, oltre agli helper che questa pagina ha usato di sfuggita), vedere asyncio — scheduler di I/O asincrono.