40. Sockets four ways: blocking, threads, selectors, asyncio

40.1 A blocking echo server and client

# labs/echo_blocking.py
import socket

with socket.create_server(("127.0.0.1", 9000), reuse_port=False) as srv:   # socket+bind+listen
    print("listening on", srv.getsockname())
    while True:
        conn, addr = srv.accept()              # blocks until a client connects (3-way handshake done)
        with conn:
            print("client", addr)
            while data := conn.recv(4096):     # b"" means the peer closed (FIN)
                conn.sendall(data)             # sendall loops until every byte is written

# client
import socket
with socket.create_connection(("127.0.0.1", 9000), timeout=5) as s:
    s.sendall(b"hello\n")
    print(s.recv(4096))

socket.create_server and create_connection wrap the socket()/bind()/listen()/accept() and getaddrinfo()/connect() calls from Net Ch 19. Two facts that cause most socket bugs:

• TCP is a byte stream, not a message stream. One send(b"hello") can arrive as two recv calls (b"hel", b"lo"), and two sends can arrive as one. You must frame messages yourself (§40.5).
• send may write only part of the buffer. Use sendall.

This server handles one client at a time: while it serves client A, client B waits in the kernel's accept queue.

40.2 One thread per connection

# labs/echo_threads.py
import socket
import threading

def handle(conn: socket.socket, addr) -> None:
    with conn:
        conn.settimeout(30)                     # an idle client cannot hold a thread forever
        try:
            while data := conn.recv(4096):
                conn.sendall(data)
        except (TimeoutError, ConnectionResetError):
            pass

with socket.create_server(("127.0.0.1", 9001)) as srv:
    while True:
        conn, addr = srv.accept()
        threading.Thread(target=handle, args=(conn, addr), daemon=True).start()

Simple and fine for hundreds of clients. Each thread costs memory and the GIL serializes the Python parts, but since the threads spend their time blocked in recv, that rarely matters.

40.3 One thread, many sockets: selectors

# labs/echo_selectors.py
import selectors
import socket

sel = selectors.DefaultSelector()               # epoll on Linux, kqueue on macOS

def accept(srv: socket.socket) -> None:
    conn, addr = srv.accept()
    conn.setblocking(False)
    sel.register(conn, selectors.EVENT_READ, echo)

def echo(conn: socket.socket) -> None:
    try:
        data = conn.recv(4096)
    except ConnectionResetError:
        data = b""
    if data:
        conn.send(data)                         # simplified: a real server buffers unsent bytes
    else:
        sel.unregister(conn)
        conn.close()

srv = socket.create_server(("127.0.0.1", 9002))
srv.setblocking(False)
sel.register(srv, selectors.EVENT_READ, accept)
while True:
    for key, _ in sel.select():                 # sleep until some socket is ready
        key.data(key.fileobj)                   # call the registered callback

This is an event loop written by hand: one thread asks the kernel "which of these file descriptors are ready?" and handles only those. It is the C guide's select-based chat server in Python, and exactly what asyncio does internally.

40.4 asyncio streams

# labs/echo_async.py
import asyncio

async def handle(reader: asyncio.StreamReader, writer: asyncio.StreamWriter) -> None:
    addr = writer.get_extra_info("peername")
    try:
        while data := await asyncio.wait_for(reader.read(4096), timeout=30):
            writer.write(data)
            await writer.drain()                 # backpressure: wait if the send buffer is full
    except (TimeoutError, ConnectionResetError):
        pass
    finally:
        writer.close()
        await writer.wait_closed()

async def main() -> None:
    server = await asyncio.start_server(handle, "127.0.0.1", 9003)
    async with server:
        await server.serve_forever()

asyncio.run(main())

The same event loop as §40.3, with the callbacks turned back into straight-line code by await. Test it with 1000 concurrent clients:

import asyncio, time

async def client(i: int) -> bool:
    r, w = await asyncio.open_connection("127.0.0.1", 9003)
    w.write(f"msg {i}\n".encode()); await w.drain()
    ok = await r.readline() == f"msg {i}\n".encode()
    w.close(); await w.wait_closed()
    return ok

async def main():
    t = time.perf_counter()
    results = await asyncio.gather(*(client(i) for i in range(1000)))
    print(sum(results), "ok in", f"{time.perf_counter() - t:.2f}s")

asyncio.run(main())

(Raise the file-descriptor limit first if you go higher: ulimit -n 65535, OS Ch 43.)

40.5 Framing: length-prefixed messages

import socket
import struct

def send_msg(sock: socket.socket, payload: bytes) -> None:
    sock.sendall(struct.pack("!I", len(payload)) + payload)     # 4-byte big-endian length

def recv_exact(sock: socket.socket, n: int) -> bytes:
    buf = bytearray()
    while len(buf) < n:
        chunk = sock.recv(n - len(buf))
        if not chunk:
            raise ConnectionError("peer closed mid-message")
        buf += chunk
    return bytes(buf)

MAX = 16 * 2**20
def recv_msg(sock: socket.socket) -> bytes:
    (length,) = struct.unpack("!I", recv_exact(sock, 4))
    if length > MAX:                                            # never trust a length from the network
        raise ValueError(f"message too large: {length}")
    return recv_exact(sock, length)

a, b = socket.socketpair()
send_msg(a, b'{"op": "ping"}')
print(recv_msg(b))

The MAX check is a security control: without it, a peer that sends a length of 4 GB makes you allocate 4 GB. Every binary protocol parser needs it (Rust §53 covers the same idea).