Building An Async Runtime From Scratch In Python

We are trying to build a chat app that can eventually move toward handling roughly a million requests per second. Along the way, we are going to build our own event loop and our own async runtime from scratch in Python.

The way we will proceed is simple: first, we will build something. Then we will see the concrete problem with it. Only after we understand that problem will we introduce the async concept needed to solve it.

Blocking Chat Server

Let us start with a simple blocking chat server:

"""Cumulative Relay project: failure-first checkpoint 1.

Chat feature attempted
----------------------
Alice can remain joined while Bob connects and joins the same room.

Desired interaction:

    Alice -> JOIN alice blue
    Relay -> OK JOIN alice blue

    # Alice remains connected.
    Bob   -> JOIN bob blue
    Relay -> OK JOIN bob blue

Broken behavior in this checkpoint
----------------------------------
Relay directly runs Alice's complete blocking session. Bob's TCP connection may
succeed because the kernel queues it, but Bob receives no JOIN response until
Alice sends QUIT or disconnects.

Run Relay:

    python3 relay_from_scratch.py

Connect Alice from a second terminal:

    nc 127.0.0.1 8000

Then connect Bob from a third terminal using the same command. This file does
not contain async machinery yet; that machinery must be earned by this failure.
"""

from __future__ import annotations

import socket


HOST = "127.0.0.1"
PORT = 8000
MAXIMUM_CHUNK_BYTES = 1024


def create_listener(host: str, port: int) -> socket.socket:
    listener = socket.socket(socket.AF_INET, socket.SOCK_STREAM)
    listener.bind((host, port))
    listener.listen()
    return listener


def receive_lines(client_socket: socket.socket):
    buffer = b""

    while True:
        chunk = client_socket.recv(MAXIMUM_CHUNK_BYTES)

        if chunk == b"":
            return

        buffer += chunk

        while b"\n" in buffer:
            line, buffer = buffer.split(b"\n", 1)
            yield line.rstrip(b"\r")


def handle_client(
    client_socket: socket.socket,
    client_address: tuple[str, int],
) -> None:
    print("session started:", client_address)
    joined = False

    try:
        for line in receive_lines(client_socket):
            words = line.decode("utf-8").split()

            if not joined:
                if len(words) != 3 or words[0] != "JOIN":
                    client_socket.sendall(b"ERR expected JOIN <name> <room>\n")
                    continue

                _, name, room = words
                joined = True
                response = f"OK JOIN {name} {room}\n"
                client_socket.sendall(response.encode("utf-8"))
                continue

            if words == ["QUIT"]:
                client_socket.sendall(b"OK QUIT\n")
                return

            client_socket.sendall(b"ERR expected QUIT\n")
    finally:
        client_socket.close()
        print("session ended:", client_address)


def run_relay(host: str = HOST, port: int = PORT) -> None:
    listener = create_listener(host, port)
    print("Relay listening at:", listener.getsockname())

    try:
        while True:
            print("waiting for the next connection")
            client_socket, client_address = listener.accept()
            print("accepted:", client_address)

            # Defect under test: this ordinary call must finish before Relay can
            # return to listener.accept() and serve the next queued connection.
            handle_client(client_socket, client_address)
    finally:
        listener.close()


if __name__ == "__main__":
    run_relay()

The Problem

Suppose Alice connects and joins the chat. Relay then enters Alice’s session and waits for Alice’s next message inside recv(). That call blocks the only Python thread running the server.

Now Bob connects and tries to join. Bob’s TCP connection may succeed because the operating system can keep it in the listener’s pending connection queue. But our Python code cannot return to listener.accept() while it is still waiting for Alice. It therefore cannot accept Bob’s connection, read Bob’s JOIN command, or send Bob a response.

Bob keeps waiting until Alice sends QUIT or disconnects. Only then does handle_client() return, allowing the server to go back to listener.accept() and start handling Bob.

This is the first problem we need to solve: Alice should be able to remain connected while Bob connects and joins. We will solve this next.

Selector-Driven Callback Server

Now we solve the blocking problem.

The main trick here is that we use selector.select() and set all the sockets to non-blocking. There are two types of sockets. One is the listening socket, which listens for incoming connection requests. The other type is the connected socket, through which Alice, Bob, or anyone else communicates after connecting.

The selector watches both types of sockets at the same time. When selector.select() returns and tells us that something is ready, each selector registration can contain a data value. We use that data value to store the Python function that should be called for that socket.

For the listening socket, the stored callback is accept_client. It checks which new client has arrived by calling listener.accept(). If a new client is accepted, we create a separate connected socket for that client and register that new socket with the selector too. The listening socket stays registered because it must continue accepting future clients; it is not replaced by Alice's or Bob's socket.

For an already-connected socket, the stored callback is read_client. It checks what that particular client has sent and processes the bytes for that client. Alice's socket and Bob's socket therefore have different registrations and different session state, even though the same Python function handles both.

The important part is that the callback is delayed. This stores a function call for later:

data=lambda: accept_client(selector, listener)

It does not call accept_client while registering the socket. Later, after the selector reports that the listener is ready, the event loop does:

callback = selector_key.data
callback()

The connected-client registration works the same way:

data=lambda: read_client(selector, session)

So the flow is:

listener becomes ready
    → call accept_client
    → create and register Bob's connected socket

Bob's connected socket becomes ready
    → call read_client for Bob
    → read and process Bob's message

Here is the complete Relay code at this checkpoint:

"""Cumulative Relay project: selector-driven server.

Chat feature completed in this lesson
-------------------------------------
Alice can remain connected while Bob connects and joins. One operating-system
thread watches the listener, Alice, Bob, and every other connected client.

This checkpoint directly dispatches each selector event. It does not yet
contain coroutines, Futures, Tasks, or our own event-loop classes.

Run Relay:

    python3 relay_from_scratch.py

Connect Alice and Bob from separate terminals:

    nc 127.0.0.1 8000

Each client can send:

    JOIN <name> <room>
    QUIT
"""

from __future__ import annotations

import selectors
import socket


HOST = "127.0.0.1"
PORT = 8000
MAXIMUM_CHUNK_BYTES = 1024


class ClientSession:
    """State that must survive between separate socket-readability events."""

    def __init__(
        self,
        client_socket: socket.socket,
        client_address: tuple[str, int],
    ) -> None:
        self.client_socket = client_socket
        self.client_address = client_address
        self.buffer = b""
        self.joined = False


def create_listener(host: str, port: int) -> socket.socket:
    listener = socket.socket(socket.AF_INET, socket.SOCK_STREAM)
    listener.setsockopt(socket.SOL_SOCKET, socket.SO_REUSEADDR, 1)
    listener.bind((host, port))
    listener.listen()
    listener.setblocking(False)
    return listener


def close_session(
    selector: selectors.BaseSelector,
    session: ClientSession,
) -> None:
    selector.unregister(session.client_socket)
    session.client_socket.close()
    print("session ended:", session.client_address)


def process_line(
    selector: selectors.BaseSelector,
    session: ClientSession,
    line: bytes,
) -> None:
    words = line.decode("utf-8").split()

    if not session.joined:
        if len(words) != 3 or words[0] != "JOIN":
            session.client_socket.sendall(b"ERR expected JOIN <name> <room>\n")
            return

        _, name, room = words
        session.joined = True
        response = f"OK JOIN {name} {room}\n"
        session.client_socket.sendall(response.encode("utf-8"))
        return

    if words == ["QUIT"]:
        session.client_socket.sendall(b"OK QUIT\n")
        close_session(selector, session)
        return

    session.client_socket.sendall(b"ERR expected QUIT\n")


def read_client(
    selector: selectors.BaseSelector,
    session: ClientSession,
) -> None:
    try:
        chunk = session.client_socket.recv(MAXIMUM_CHUNK_BYTES)
    except BlockingIOError:
        return

    if chunk == b"":
        close_session(selector, session)
        return

    session.buffer += chunk

    while b"\n" in session.buffer:
        line, session.buffer = session.buffer.split(b"\n", 1)
        process_line(selector, session, line.rstrip(b"\r"))

        if session.client_socket.fileno() == -1:
            return


def accept_client(
    selector: selectors.BaseSelector,
    listener: socket.socket,
) -> None:
    try:
        client_socket, client_address = listener.accept()
    except BlockingIOError:
        return

    client_socket.setblocking(False)
    session = ClientSession(client_socket, client_address)

    selector.register(
        client_socket,
        selectors.EVENT_READ,
        data=lambda: read_client(selector, session),
    )
    print("session started:", client_address)


def run_relay(host: str = HOST, port: int = PORT) -> None:
    selector = selectors.DefaultSelector()
    listener = create_listener(host, port)

    selector.register(
        listener,
        selectors.EVENT_READ,
        data=lambda: accept_client(selector, listener),
    )
    print("Relay listening at:", listener.getsockname())

    try:
        while True:
            ready_events = selector.select()

            for selector_key, _event_mask in ready_events:
                callback = selector_key.data
                callback()
    finally:
        selector.close()
        listener.close()


if __name__ == "__main__":
    run_relay()

With this change, Alice can remain connected and silent while Bob connects and receives his JOIN response. We have solved this problem with non-blocking sockets, a selector, and callbacks. Futures and coroutines are still not needed yet.

Non-Blocking Writes And Slow Clients

We have solved reading from many clients without blocking the event-loop thread, but writing has the same kind of problem.

Suppose Bob remains connected but stops reading from the network. Relay keeps broadcasting room messages to Bob. At first, sendall() appears to work because the operating system can temporarily store bytes in the socket's send and receive buffers. Those buffers have finite capacity, though. If Bob continues not to read, his receive buffer fills, Relay's corresponding kernel send buffer eventually fills, and a non-blocking sendall() raises BlockingIOError.

Making Bob's socket blocking would be worse. The only event-loop thread could become stuck waiting for Bob, preventing Relay from reading Alice, accepting Charlie, or serving anyone else.

The main idea is to give every ClientSession its own application-level outgoing buffer:

self.outgoing_buffer = bytearray()

Whenever Relay wants to write bytes to a client's socket, it does not immediately insist on sending everything. It calls queue_bytes(). That function appends the bytes to the session's outgoing_buffer, retrieves the socket's selector registration, and adds EVENT_WRITE to the events being watched:

def queue_bytes(
    selector: selectors.BaseSelector,
    rooms: dict[str, set[ClientSession]],
    session: ClientSession,
    outgoing: bytes,
) -> bool:
    session.outgoing_buffer.extend(outgoing)

    registration = selector.get_key(session.client_socket)
    selector.modify(
        session.client_socket,
        registration.events | selectors.EVENT_WRITE,
        data=registration.data,
    )
    return True

queue_bytes() stores the work; it does not perform the network write. After EVENT_WRITE is added, selector.select() can report that the socket is writable. It returns a selector_key identifying the socket and an event_mask describing whether that socket is readable, writable, or both:

ready_events = selector.select()

for selector_key, event_mask in ready_events:
    callback = selector_key.data
    callback(event_mask)

For a connected client, service_client() checks the mask. A write event calls write_client():

if event_mask & selectors.EVENT_WRITE:
    write_client(selector, rooms, session)

write_client() uses send() instead of sendall(). A non-blocking socket might accept only part of the buffer, and send() tells us exactly how many bytes were accepted:

bytes_sent = session.client_socket.send(session.outgoing_buffer)
del session.outgoing_buffer[:bytes_sent]

Suppose 100 bytes are queued but the kernel currently accepts only 40. send() returns 40, we delete those 40 bytes, and the remaining 60 stay in outgoing_buffer. Because the socket is still registered for EVENT_WRITE, the selector can notify us later and we try the remaining bytes again.

Once the outgoing buffer becomes empty, we remove EVENT_WRITE from that socket's registration:

registration.events & ~selectors.EVENT_WRITE

Most connected sockets are writable most of the time. If we kept watching an empty socket for write readiness, the selector could return it repeatedly and create a busy loop.

Limit The Outgoing Queue

The application-level buffer solves temporary backpressure, but an unlimited buffer creates another problem. If Bob never reads while messages continue arriving, Bob's outgoing_buffer can grow until Relay runs out of memory.

We therefore limit each client's queued output to 256 KiB:

MAXIMUM_OUTGOING_BYTES = 256 * 1024

Before adding another message, Relay checks the proposed total:

if len(session.outgoing_buffer) + len(outgoing) > MAXIMUM_OUTGOING_BYTES:
    print("closing slow client:", session.client_address)
    close_session(selector, rooms, session)
    return False

This gives Relay a clear backpressure policy:

brief slowdown
→ preserve the bytes and retry on EVENT_WRITE

sustained slowdown beyond 256 KiB
→ disconnect that client
→ keep the server's memory bounded

Here is the complete Relay code at this checkpoint. It intentionally stops before heartbeat support:

"""Cumulative Relay project: selector-driven server.

Chat feature completed in this lesson
-------------------------------------
Alice and Bob can join the same room and exchange messages. One operating-
system thread watches the listener and every connected client.

This checkpoint directly dispatches each selector event. It does not yet
contain coroutines, Futures, Tasks, or our own event-loop classes.

Run Relay:

    python3 relay_from_scratch.py

Connect Alice and Bob from separate terminals:

    nc 127.0.0.1 8000

Each client can send:

    JOIN <name> <room>
    MSG <text>
    QUIT
"""

from __future__ import annotations

import selectors
import socket


HOST = "127.0.0.1"
PORT = 8000
MAXIMUM_CHUNK_BYTES = 1024
MAXIMUM_OUTGOING_BYTES = 256 * 1024


class ClientSession:
    """State that must survive between separate socket-readability events."""

    def __init__(
        self,
        client_socket: socket.socket,
        client_address: tuple[str, int],
    ) -> None:
        self.client_socket = client_socket
        self.client_address = client_address
        self.buffer = b""
        self.outgoing_buffer = bytearray()
        self.joined = False
        self.name: str | None = None
        self.room: str | None = None
        self.close_after_writing = False


def create_listener(host: str, port: int) -> socket.socket:
    listener = socket.socket(socket.AF_INET, socket.SOCK_STREAM)
    listener.setsockopt(socket.SOL_SOCKET, socket.SO_REUSEADDR, 1)
    listener.bind((host, port))
    listener.listen()
    listener.setblocking(False)
    return listener


def close_session(
    selector: selectors.BaseSelector,
    rooms: dict[str, set[ClientSession]],
    session: ClientSession,
) -> None:
    if session.room is not None:
        room_members = rooms[session.room]
        room_members.remove(session)

        if not room_members:
            del rooms[session.room]

    selector.unregister(session.client_socket)
    session.client_socket.close()
    print("session ended:", session.client_address)


def queue_bytes(
    selector: selectors.BaseSelector,
    rooms: dict[str, set[ClientSession]],
    session: ClientSession,
    outgoing: bytes,
) -> bool:
    if len(session.outgoing_buffer) + len(outgoing) > MAXIMUM_OUTGOING_BYTES:
        print("closing slow client:", session.client_address)
        close_session(selector, rooms, session)
        return False

    session.outgoing_buffer.extend(outgoing)

    registration = selector.get_key(session.client_socket)
    selector.modify(
        session.client_socket,
        registration.events | selectors.EVENT_WRITE,
        data=registration.data,
    )
    return True


def process_line(
    selector: selectors.BaseSelector,
    rooms: dict[str, set[ClientSession]],
    session: ClientSession,
    line: bytes,
) -> None:
    words = line.decode("utf-8").split()

    if not session.joined:
        if len(words) != 3 or words[0] != "JOIN":
            queue_bytes(
                selector,
                rooms,
                session,
                b"ERR expected JOIN <name> <room>\n",
            )
            return

        _, name, room = words
        session.joined = True
        session.name = name
        session.room = room
        rooms.setdefault(room, set()).add(session)
        response = f"OK JOIN {name} {room}\n"
        queue_bytes(selector, rooms, session, response.encode("utf-8"))
        return

    if words == ["QUIT"]:
        if not queue_bytes(selector, rooms, session, b"OK QUIT\n"):
            return

        session.close_after_writing = True

        registration = selector.get_key(session.client_socket)
        selector.modify(
            session.client_socket,
            selectors.EVENT_WRITE,
            data=registration.data,
        )
        return

    if len(words) >= 2 and words[0] == "MSG":
        message_text = line.decode("utf-8").split(maxsplit=1)[1]
        outgoing = f"MESSAGE {session.name} {message_text}\n".encode("utf-8")

        for room_member in tuple(rooms[session.room]):
            queue_bytes(selector, rooms, room_member, outgoing)
        return

    queue_bytes(
        selector,
        rooms,
        session,
        b"ERR expected MSG <text> or QUIT\n",
    )


def read_client(
    selector: selectors.BaseSelector,
    rooms: dict[str, set[ClientSession]],
    session: ClientSession,
) -> None:
    try:
        chunk = session.client_socket.recv(MAXIMUM_CHUNK_BYTES)
    except BlockingIOError:
        return

    if chunk == b"":
        close_session(selector, rooms, session)
        return

    session.buffer += chunk

    while b"\n" in session.buffer:
        line, session.buffer = session.buffer.split(b"\n", 1)
        process_line(selector, rooms, session, line.rstrip(b"\r"))

        if session.client_socket.fileno() == -1 or session.close_after_writing:
            return


def write_client(
    selector: selectors.BaseSelector,
    rooms: dict[str, set[ClientSession]],
    session: ClientSession,
) -> None:
    try:
        bytes_sent = session.client_socket.send(session.outgoing_buffer)
    except BlockingIOError:
        return

    del session.outgoing_buffer[:bytes_sent]

    if session.outgoing_buffer:
        return

    if session.close_after_writing:
        close_session(selector, rooms, session)
        return

    registration = selector.get_key(session.client_socket)
    selector.modify(
        session.client_socket,
        registration.events & ~selectors.EVENT_WRITE,
        data=registration.data,
    )


def service_client(
    selector: selectors.BaseSelector,
    rooms: dict[str, set[ClientSession]],
    session: ClientSession,
    event_mask: int,
) -> None:
    if event_mask & selectors.EVENT_READ:
        read_client(selector, rooms, session)

    if session.client_socket.fileno() == -1:
        return

    if event_mask & selectors.EVENT_WRITE:
        write_client(selector, rooms, session)


def accept_client(
    selector: selectors.BaseSelector,
    listener: socket.socket,
    rooms: dict[str, set[ClientSession]],
) -> None:
    try:
        client_socket, client_address = listener.accept()
    except BlockingIOError:
        return

    client_socket.setblocking(False)
    session = ClientSession(client_socket, client_address)

    selector.register(
        client_socket,
        selectors.EVENT_READ,
        data=lambda event_mask: service_client(
            selector,
            rooms,
            session,
            event_mask,
        ),
    )
    print("session started:", client_address)


def run_relay(host: str = HOST, port: int = PORT) -> None:
    selector = selectors.DefaultSelector()
    listener = create_listener(host, port)
    # All client callbacks share this registry on the event-loop thread.
    rooms: dict[str, set[ClientSession]] = {}

    selector.register(
        listener,
        selectors.EVENT_READ,
        data=lambda _event_mask: accept_client(selector, listener, rooms),
    )
    print("Relay listening at:", listener.getsockname())

    try:
        while True:
            ready_events = selector.select()

            for selector_key, event_mask in ready_events:
                callback = selector_key.data
                callback(event_mask)
    finally:
        selector.close()
        listener.close()


if __name__ == "__main__":
    run_relay()

Heartbeats: Waking The Selector For Time

Relay now knows when a client sends bytes or closes its connection normally. But a connection does not always die cleanly. Bob's phone can lose power, his router can disappear, or a firewall can silently drop traffic without Relay immediately receiving a TCP close notification. Relay may continue holding Bob's socket, session, room membership, and buffers even though Bob is no longer reachable.

We need a way to ask whether Bob's connection is still alive. That is the role of a heartbeat.

Relay periodically sends an ordinary protocol message:

PING

A healthy client automatically responds:

PONG

The words are not special to TCP. They are commands in our own Relay protocol. PING means "are you still reachable?" and PONG means "yes, I am still here."

The Two Heartbeat Times

Every joined session stores:

self.next_ping_at = 0.0
self.pong_deadline: float | None = None

next_ping_at is the absolute time.monotonic() moment when Relay should send the next PING. After a client joins or answers with PONG, Relay sets:

session.next_ping_at = time.monotonic() + HEARTBEAT_INTERVAL_SECONDS

pong_deadline is different. None means Relay is not currently waiting for a heartbeat answer. After queuing PING, Relay sets:

session.pong_deadline = now + HEARTBEAT_TIMEOUT_SECONDS

The timeline is:

client joins or answers PONG
→ wait until next_ping_at
→ queue PING
→ wait until pong_deadline
→ PONG arrives: clear the deadline and schedule the next PING
→ no PONG: close the client

We use time.monotonic() instead of the wall clock because heartbeat durations should not change when the system clock is adjusted.

The Key Timer Idea: `selector.select(timeout)`

Before heartbeats, Relay called:

ready_events = selector.select()

With no timeout, the event-loop thread can sleep indefinitely until a socket becomes readable or writable. That is a problem because silence is not a socket event. A heartbeat deadline might pass while the thread remains asleep inside the selector.

Relay now calculates how many seconds remain until the earliest heartbeat action and passes that value to the selector:

heartbeat_timeout = seconds_until_next_heartbeat(rooms)
ready_events = selector.select(heartbeat_timeout)

This gives selector.select(timeout) two ways to return:

a socket becomes ready before the timeout
→ return its selector key and event mask immediately

the timeout expires with no socket activity
→ wake the thread and return an empty list: []

The empty list is important. There is no selector_key and no event_mask, so the callback loop runs zero times:

for selector_key, event_mask in ready_events:
    callback = selector_key.data
    callback(event_mask)

Execution then continues to:

run_heartbeat_checks(selector, rooms)

That is how an event loop combines I/O readiness with timers. The operating system selector still performs the sleeping, but the timeout limits how long it may sleep before Python must check time-based work.

Choosing The Selector Timeout

For each session, the next relevant timestamp is one of two values:

not waiting for PONG
→ next relevant time is next_ping_at

already waiting for PONG
→ next relevant time is pong_deadline

Relay collects those timestamps, chooses the earliest with min(deadlines), and subtracts the current monotonic time:

return max(0.0, min(deadlines) - time.monotonic())

If no joined sessions exist, there is no heartbeat deadline and the function returns None. selector.select(None) may then sleep indefinitely until actual socket activity.

If the earliest deadline has already passed, the subtraction is negative. max(0.0, ...) turns it into 0.0, making selector.select(0.0) return immediately so the overdue heartbeat work can run.

Running Heartbeat Checks

After the selector returns, Relay first processes every ready socket callback and then runs heartbeat checks:

for selector_key, event_mask in ready_events:
    callback = selector_key.data
    callback(event_mask)

run_heartbeat_checks(selector, rooms)

This order matters. If Bob's PONG has already arrived at the same moment as his deadline, the read callback processes it first and clears pong_deadline. The heartbeat check then sees the fresh state instead of closing Bob before reading an answer already waiting in the kernel.

The heartbeat checker has two states:

pong_deadline exists
→ PING was already sent
→ close the client if the deadline has passed
→ otherwise continue waiting

pong_deadline is None
→ no unanswered PING exists
→ if next_ping_at has arrived, queue PING and create pong_deadline

PING uses the same outgoing-buffer and EVENT_WRITE mechanism as every other server response. PONG uses the same EVENT_READ path as JOIN, MSG, and QUIT; process_line() recognizes it and schedules the next heartbeat.

This is the first real timer in Relay. We needed it because the event loop must sometimes wake when no socket has produced an event.

Here is the complete Relay code at the heartbeat checkpoint:

"""Cumulative Relay project: selector-driven server.

Chat feature completed in this lesson
-------------------------------------
Alice and Bob can join the same room and exchange messages. One operating-
system thread watches the listener and every connected client.

This checkpoint directly dispatches each selector event. It does not yet
contain coroutines, Futures, Tasks, or our own event-loop classes.

Run Relay:

    python3 relay_from_scratch.py

Connect Alice and Bob from separate terminals:

    nc 127.0.0.1 8000

Each client can send:

    JOIN <name> <room>
    MSG <text>
    QUIT
"""

from __future__ import annotations

import selectors
import socket
import time


HOST = "127.0.0.1"
PORT = 8000
MAXIMUM_CHUNK_BYTES = 1024
MAXIMUM_OUTGOING_BYTES = 256 * 1024
HEARTBEAT_INTERVAL_SECONDS = 10.0
HEARTBEAT_TIMEOUT_SECONDS = 5.0


class ClientSession:
    """State that must survive between separate socket-readability events."""

    def __init__(
        self,
        client_socket: socket.socket,
        client_address: tuple[str, int],
    ) -> None:
        self.client_socket = client_socket
        self.client_address = client_address
        self.buffer = b""
        self.outgoing_buffer = bytearray()
        self.joined = False
        self.name: str | None = None
        self.room: str | None = None
        self.close_after_writing = False
        self.next_ping_at = 0.0
        self.pong_deadline: float | None = None


def create_listener(host: str, port: int) -> socket.socket:
    listener = socket.socket(socket.AF_INET, socket.SOCK_STREAM)
    listener.setsockopt(socket.SOL_SOCKET, socket.SO_REUSEADDR, 1)
    listener.bind((host, port))
    listener.listen()
    listener.setblocking(False)
    return listener


def close_session(
    selector: selectors.BaseSelector,
    rooms: dict[str, set[ClientSession]],
    session: ClientSession,
) -> None:
    if session.room is not None:
        room_members = rooms[session.room]
        room_members.remove(session)

        if not room_members:
            del rooms[session.room]

    selector.unregister(session.client_socket)
    session.client_socket.close()
    print("session ended:", session.client_address)


def queue_bytes(
    selector: selectors.BaseSelector,
    rooms: dict[str, set[ClientSession]],
    session: ClientSession,
    outgoing: bytes,
) -> bool:
    if len(session.outgoing_buffer) + len(outgoing) > MAXIMUM_OUTGOING_BYTES:
        print("closing slow client:", session.client_address)
        close_session(selector, rooms, session)
        return False

    session.outgoing_buffer.extend(outgoing)

    registration = selector.get_key(session.client_socket)
    selector.modify(
        session.client_socket,
        registration.events | selectors.EVENT_WRITE,
        data=registration.data,
    )
    return True


def process_line(
    selector: selectors.BaseSelector,
    rooms: dict[str, set[ClientSession]],
    session: ClientSession,
    line: bytes,
) -> None:
    words = line.decode("utf-8").split()

    if not session.joined:
        if len(words) != 3 or words[0] != "JOIN":
            queue_bytes(
                selector,
                rooms,
                session,
                b"ERR expected JOIN <name> <room>\n",
            )
            return

        _, name, room = words
        session.joined = True
        session.name = name
        session.room = room
        session.next_ping_at = time.monotonic() + HEARTBEAT_INTERVAL_SECONDS
        rooms.setdefault(room, set()).add(session)
        response = f"OK JOIN {name} {room}\n"
        queue_bytes(selector, rooms, session, response.encode("utf-8"))
        return

    if words == ["QUIT"]:
        if not queue_bytes(selector, rooms, session, b"OK QUIT\n"):
            return

        session.close_after_writing = True

        registration = selector.get_key(session.client_socket)
        selector.modify(
            session.client_socket,
            selectors.EVENT_WRITE,
            data=registration.data,
        )
        return

    if words == ["PONG"]:
        session.pong_deadline = None
        session.next_ping_at = time.monotonic() + HEARTBEAT_INTERVAL_SECONDS
        return

    if len(words) >= 2 and words[0] == "MSG":
        message_text = line.decode("utf-8").split(maxsplit=1)[1]
        outgoing = f"MESSAGE {session.name} {message_text}\n".encode("utf-8")

        for room_member in tuple(rooms[session.room]):
            queue_bytes(selector, rooms, room_member, outgoing)
        return

    queue_bytes(
        selector,
        rooms,
        session,
        b"ERR expected MSG <text> or QUIT\n",
    )


def read_client(
    selector: selectors.BaseSelector,
    rooms: dict[str, set[ClientSession]],
    session: ClientSession,
) -> None:
    try:
        chunk = session.client_socket.recv(MAXIMUM_CHUNK_BYTES)
    except BlockingIOError:
        return

    if chunk == b"":
        close_session(selector, rooms, session)
        return

    session.buffer += chunk

    while b"\n" in session.buffer:
        line, session.buffer = session.buffer.split(b"\n", 1)
        process_line(selector, rooms, session, line.rstrip(b"\r"))

        if session.client_socket.fileno() == -1 or session.close_after_writing:
            return


def write_client(
    selector: selectors.BaseSelector,
    rooms: dict[str, set[ClientSession]],
    session: ClientSession,
) -> None:
    try:
        bytes_sent = session.client_socket.send(session.outgoing_buffer)
    except BlockingIOError:
        return

    del session.outgoing_buffer[:bytes_sent]

    if session.outgoing_buffer:
        return

    if session.close_after_writing:
        close_session(selector, rooms, session)
        return

    registration = selector.get_key(session.client_socket)
    selector.modify(
        session.client_socket,
        registration.events & ~selectors.EVENT_WRITE,
        data=registration.data,
    )


def service_client(
    selector: selectors.BaseSelector,
    rooms: dict[str, set[ClientSession]],
    session: ClientSession,
    event_mask: int,
) -> None:
    if event_mask & selectors.EVENT_READ:
        read_client(selector, rooms, session)

    if session.client_socket.fileno() == -1:
        return

    if event_mask & selectors.EVENT_WRITE:
        write_client(selector, rooms, session)


def accept_client(
    selector: selectors.BaseSelector,
    listener: socket.socket,
    rooms: dict[str, set[ClientSession]],
) -> None:
    try:
        client_socket, client_address = listener.accept()
    except BlockingIOError:
        return

    client_socket.setblocking(False)
    session = ClientSession(client_socket, client_address)

    selector.register(
        client_socket,
        selectors.EVENT_READ,
        data=lambda event_mask: service_client(
            selector,
            rooms,
            session,
            event_mask,
        ),
    )
    print("session started:", client_address)


def joined_sessions(
    rooms: dict[str, set[ClientSession]],
) -> tuple[ClientSession, ...]:
    return tuple(
        session
        for room_members in rooms.values()
        for session in room_members
    )


def seconds_until_next_heartbeat(
    rooms: dict[str, set[ClientSession]],
) -> float | None:
    deadlines = []

    for session in joined_sessions(rooms):
        if session.pong_deadline is None:
            deadlines.append(session.next_ping_at)
        else:
            deadlines.append(session.pong_deadline)

    if not deadlines:
        return None

    return max(0.0, min(deadlines) - time.monotonic())


def run_heartbeat_checks(
    selector: selectors.BaseSelector,
    rooms: dict[str, set[ClientSession]],
) -> None:
    now = time.monotonic()

    for session in joined_sessions(rooms):
        if session.pong_deadline is not None:
            if now >= session.pong_deadline:
                print("closing unresponsive client:", session.client_address)
                close_session(selector, rooms, session)
            continue

        if now >= session.next_ping_at:
            if queue_bytes(selector, rooms, session, b"PING\n"):
                session.pong_deadline = now + HEARTBEAT_TIMEOUT_SECONDS


def run_relay(host: str = HOST, port: int = PORT) -> None:
    selector = selectors.DefaultSelector()
    listener = create_listener(host, port)
    # All client callbacks share this registry on the event-loop thread.
    rooms: dict[str, set[ClientSession]] = {}

    selector.register(
        listener,
        selectors.EVENT_READ,
        data=lambda _event_mask: accept_client(selector, listener, rooms),
    )
    print("Relay listening at:", listener.getsockname())

    try:
        while True:
            heartbeat_timeout = seconds_until_next_heartbeat(rooms)
            ready_events = selector.select(heartbeat_timeout)

            for selector_key, event_mask in ready_events:
                callback = selector_key.data
                callback(event_mask)

            run_heartbeat_checks(selector, rooms)
    finally:
        selector.close()
        listener.close()


if __name__ == "__main__":
    run_relay()

Persistent Room History

We are now adding a HISTORY feature. Relay keeps the latest room messages so that a client who joins later can request what was said before they arrived.

The interaction looks like this:

Alice → MSG first message
Bob   → MSG second message

Charlie joins afterward.

Charlie → HISTORY
Relay   → HISTORY_BEGIN 2
Relay   → MESSAGE alice first message
Relay   → MESSAGE bob second message
Relay   → HISTORY_END

The first version stored history in a Python dictionary, but that history disappeared whenever the Relay process restarted. The current checkpoint stores messages in SQLite instead. Each MSG appends the already-formatted message bytes to the database, and HISTORY loads the latest 50 messages for that room in their original order.

Because each insert is committed, the messages survive a complete Relay restart. This checkpoint intentionally uses ordinary synchronous SQLite calls; it does not yet introduce worker threads, Futures, Tasks, or coroutines.

Here is the complete Relay code at the persistent-history checkpoint:

"""Cumulative Relay project: selector-driven server.

Chat feature completed in this lesson
-------------------------------------
Alice and Bob can join the same room and exchange messages. One operating-
system thread watches the listener and every connected client.

This checkpoint directly dispatches each selector event. It does not yet
contain coroutines, Futures, Tasks, or our own event-loop classes.

Run Relay:

    python3 relay_from_scratch.py

Connect Alice and Bob from separate terminals:

    nc 127.0.0.1 8000

Each client can send:

    JOIN <name> <room>
    MSG <text>
    HISTORY
    QUIT
"""

from __future__ import annotations

import selectors
import socket
import sqlite3
import time
from pathlib import Path


HOST = "127.0.0.1"
PORT = 8000
MAXIMUM_CHUNK_BYTES = 1024
MAXIMUM_OUTGOING_BYTES = 256 * 1024
MAXIMUM_HISTORY_MESSAGES = 50
HEARTBEAT_INTERVAL_SECONDS = 10.0
HEARTBEAT_TIMEOUT_SECONDS = 5.0
HISTORY_DATABASE_PATH = Path(__file__).with_name("relay_history.sqlite3")


class ClientSession:
    """State that must survive between separate socket-readability events."""

    def __init__(
        self,
        client_socket: socket.socket,
        client_address: tuple[str, int],
    ) -> None:
        self.client_socket = client_socket
        self.client_address = client_address
        self.buffer = b""
        self.outgoing_buffer = bytearray()
        self.joined = False
        self.name: str | None = None
        self.room: str | None = None
        self.close_after_writing = False
        self.next_ping_at = 0.0
        self.pong_deadline: float | None = None


class HistoryStore:
    def __init__(self, database_path: str | Path) -> None:
        self.connection = sqlite3.connect(database_path)
        self.connection.execute(
            """
            CREATE TABLE IF NOT EXISTS messages (
                id INTEGER PRIMARY KEY AUTOINCREMENT,
                room TEXT NOT NULL,
                payload BLOB NOT NULL
            )
            """
        )
        self.connection.commit()

    def append(self, room: str, payload: bytes) -> None:
        self.connection.execute(
            "INSERT INTO messages (room, payload) VALUES (?, ?)",
            (room, payload),
        )
        self.connection.commit()

    def recent(self, room: str, limit: int) -> list[bytes]:
        rows = self.connection.execute(
            """
            SELECT payload
            FROM messages
            WHERE room = ?
            ORDER BY id DESC
            LIMIT ?
            """,
            (room, limit),
        ).fetchall()
        return [bytes(row[0]) for row in reversed(rows)]

    def close(self) -> None:
        self.connection.close()


def create_listener(host: str, port: int) -> socket.socket:
    listener = socket.socket(socket.AF_INET, socket.SOCK_STREAM)
    listener.setsockopt(socket.SOL_SOCKET, socket.SO_REUSEADDR, 1)
    listener.bind((host, port))
    listener.listen()
    listener.setblocking(False)
    return listener


def close_session(
    selector: selectors.BaseSelector,
    rooms: dict[str, set[ClientSession]],
    session: ClientSession,
) -> None:
    if session.room is not None:
        room_members = rooms[session.room]
        room_members.remove(session)

        if not room_members:
            del rooms[session.room]

    selector.unregister(session.client_socket)
    session.client_socket.close()
    print("session ended:", session.client_address)


def queue_bytes(
    selector: selectors.BaseSelector,
    rooms: dict[str, set[ClientSession]],
    session: ClientSession,
    outgoing: bytes,
) -> bool:
    if len(session.outgoing_buffer) + len(outgoing) > MAXIMUM_OUTGOING_BYTES:
        print("closing slow client:", session.client_address)
        close_session(selector, rooms, session)
        return False

    session.outgoing_buffer.extend(outgoing)

    registration = selector.get_key(session.client_socket)
    selector.modify(
        session.client_socket,
        registration.events | selectors.EVENT_WRITE,
        data=registration.data,
    )
    return True


def process_line(
    selector: selectors.BaseSelector,
    rooms: dict[str, set[ClientSession]],
    history_store: HistoryStore,
    session: ClientSession,
    line: bytes,
) -> None:
    words = line.decode("utf-8").split()

    if not session.joined:
        if len(words) != 3 or words[0] != "JOIN":
            queue_bytes(
                selector,
                rooms,
                session,
                b"ERR expected JOIN <name> <room>\n",
            )
            return

        _, name, room = words
        session.joined = True
        session.name = name
        session.room = room
        session.next_ping_at = time.monotonic() + HEARTBEAT_INTERVAL_SECONDS
        rooms.setdefault(room, set()).add(session)
        response = f"OK JOIN {name} {room}\n"
        queue_bytes(selector, rooms, session, response.encode("utf-8"))
        return

    if words == ["QUIT"]:
        if not queue_bytes(selector, rooms, session, b"OK QUIT\n"):
            return

        session.close_after_writing = True

        registration = selector.get_key(session.client_socket)
        selector.modify(
            session.client_socket,
            selectors.EVENT_WRITE,
            data=registration.data,
        )
        return

    if words == ["PONG"]:
        session.pong_deadline = None
        session.next_ping_at = time.monotonic() + HEARTBEAT_INTERVAL_SECONDS
        return

    if words == ["HISTORY"]:
        room_history = history_store.recent(
            session.room,
            MAXIMUM_HISTORY_MESSAGES,
        )
        response = (
            f"HISTORY_BEGIN {len(room_history)}\n".encode("utf-8")
            + b"".join(room_history)
            + b"HISTORY_END\n"
        )
        queue_bytes(selector, rooms, session, response)
        return

    if len(words) >= 2 and words[0] == "MSG":
        message_text = line.decode("utf-8").split(maxsplit=1)[1]
        outgoing = f"MESSAGE {session.name} {message_text}\n".encode("utf-8")

        history_store.append(session.room, outgoing)

        for room_member in tuple(rooms[session.room]):
            queue_bytes(selector, rooms, room_member, outgoing)
        return

    queue_bytes(
        selector,
        rooms,
        session,
        b"ERR expected MSG <text>, HISTORY, or QUIT\n",
    )


def read_client(
    selector: selectors.BaseSelector,
    rooms: dict[str, set[ClientSession]],
    history_store: HistoryStore,
    session: ClientSession,
) -> None:
    try:
        chunk = session.client_socket.recv(MAXIMUM_CHUNK_BYTES)
    except BlockingIOError:
        return

    if chunk == b"":
        close_session(selector, rooms, session)
        return

    session.buffer += chunk

    while b"\n" in session.buffer:
        line, session.buffer = session.buffer.split(b"\n", 1)
        process_line(selector, rooms, history_store, session, line.rstrip(b"\r"))

        if session.client_socket.fileno() == -1 or session.close_after_writing:
            return


def write_client(
    selector: selectors.BaseSelector,
    rooms: dict[str, set[ClientSession]],
    session: ClientSession,
) -> None:
    try:
        bytes_sent = session.client_socket.send(session.outgoing_buffer)
    except BlockingIOError:
        return

    del session.outgoing_buffer[:bytes_sent]

    if session.outgoing_buffer:
        return

    if session.close_after_writing:
        close_session(selector, rooms, session)
        return

    registration = selector.get_key(session.client_socket)
    selector.modify(
        session.client_socket,
        registration.events & ~selectors.EVENT_WRITE,
        data=registration.data,
    )


def service_client(
    selector: selectors.BaseSelector,
    rooms: dict[str, set[ClientSession]],
    history_store: HistoryStore,
    session: ClientSession,
    event_mask: int,
) -> None:
    if event_mask & selectors.EVENT_READ:
        read_client(selector, rooms, history_store, session)

    if session.client_socket.fileno() == -1:
        return

    if event_mask & selectors.EVENT_WRITE:
        write_client(selector, rooms, session)


def accept_client(
    selector: selectors.BaseSelector,
    listener: socket.socket,
    rooms: dict[str, set[ClientSession]],
    history_store: HistoryStore,
) -> None:
    try:
        client_socket, client_address = listener.accept()
    except BlockingIOError:
        return

    client_socket.setblocking(False)
    session = ClientSession(client_socket, client_address)

    selector.register(
        client_socket,
        selectors.EVENT_READ,
        data=lambda event_mask: service_client(
            selector,
            rooms,
            history_store,
            session,
            event_mask,
        ),
    )
    print("session started:", client_address)


def joined_sessions(
    rooms: dict[str, set[ClientSession]],
) -> tuple[ClientSession, ...]:
    return tuple(
        session
        for room_members in rooms.values()
        for session in room_members
    )


def seconds_until_next_heartbeat(
    rooms: dict[str, set[ClientSession]],
) -> float | None:
    deadlines = []

    for session in joined_sessions(rooms):
        if session.pong_deadline is None:
            deadlines.append(session.next_ping_at)
        else:
            deadlines.append(session.pong_deadline)

    if not deadlines:
        return None

    return max(0.0, min(deadlines) - time.monotonic())


def run_heartbeat_checks(
    selector: selectors.BaseSelector,
    rooms: dict[str, set[ClientSession]],
) -> None:
    now = time.monotonic()

    for session in joined_sessions(rooms):
        if session.pong_deadline is not None:
            if now >= session.pong_deadline:
                print("closing unresponsive client:", session.client_address)
                close_session(selector, rooms, session)
            continue

        if now >= session.next_ping_at:
            if queue_bytes(selector, rooms, session, b"PING\n"):
                session.pong_deadline = now + HEARTBEAT_TIMEOUT_SECONDS


def run_relay(
    host: str = HOST,
    port: int = PORT,
    history_database_path: str | Path = HISTORY_DATABASE_PATH,
) -> None:
    selector = selectors.DefaultSelector()
    listener = create_listener(host, port)
    history_store = HistoryStore(history_database_path)
    # All client callbacks share this registry on the event-loop thread.
    rooms: dict[str, set[ClientSession]] = {}

    selector.register(
        listener,
        selectors.EVENT_READ,
        data=lambda _event_mask: accept_client(
            selector,
            listener,
            rooms,
            history_store,
        ),
    )
    print("Relay listening at:", listener.getsockname())

    try:
        while True:
            heartbeat_timeout = seconds_until_next_heartbeat(rooms)
            ready_events = selector.select(heartbeat_timeout)

            for selector_key, event_mask in ready_events:
                callback = selector_key.data
                callback(event_mask)

            run_heartbeat_checks(selector, rooms)
    finally:
        selector.close()
        listener.close()
        history_store.close()


if __name__ == "__main__":
    run_relay()

Moving Blocking Sqlite Work To Another Thread

The persistent-history feature creates a new problem. SQLite is synchronous. When Relay calls:

history_store.append(session.room, outgoing)

the thread that called it stays inside SQLite until the insert and commit finish or fail.

In our current architecture, that caller is the only event-loop thread. Suppose Bob sends a message and SQLite is temporarily blocked by a database lock or slow disk operation. Relay enters Bob's callback and waits inside SQLite. During that time, Steve may connect or send a message, and his socket may already be ready in the operating system, but our Python thread cannot return to selector.select() to process it.

The path is:

Bob's socket becomes readable
→ event-loop thread calls Bob's callback
→ callback calls SQLite
→ SQLite blocks
→ event-loop thread cannot return to the selector
→ Steve waits too

Making the client sockets non-blocking does not help here. It only changes the behavior of socket operations such as accept(), recv(), and send(). It does not make SQLite non-blocking.

We therefore give synchronous SQLite work to one dedicated database-worker thread:

event-loop thread
→ handles listener and client sockets
→ submits database jobs
→ never waits inside SQLite

history-worker thread
→ owns the SQLite connection
→ executes database jobs one at a time
→ may block without freezing network handling

Sending Work To The Database Thread

The two threads share a thread-safe jobs queue:

self.jobs: queue.Queue[
    tuple[HistoryOperation, HistoryCompletion] | None
] = queue.Queue()

One queue item contains:

operation
→ the SQLite work to execute

completion callback
→ what the event-loop thread should do with the outcome

The event-loop thread submits the pair:

def submit(
    self,
    operation: HistoryOperation,
    completed: HistoryCompletion,
) -> None:
    self.jobs.put((operation, completed))

The database thread spends its idle time waiting here:

job = self.jobs.get()

Queue.get() sleeps when the queue is empty. When the event-loop thread calls jobs.put(...), the queue itself wakes the worker. We do not send a socket byte in this direction.

The worker then runs the synchronous operation:

operation, completed = job

try:
    result = operation(history_store)
except Exception as error:
    self.completed_jobs.put((completed, None, error))
else:
    self.completed_jobs.put((completed, result, None))

SQLite can block this worker thread, but the event-loop thread is already free to return to the selector and serve Steve.

Returning The Result To The Event Loop

The reverse direction has a different problem. When the worker finishes, it puts the real outcome in completed_jobs:

self.completed_jobs.put((completed, result, error))

But the event-loop thread may currently be asleep inside:

selector.select(heartbeat_timeout)

Changing a Python queue does not make any selector registration readable. The selector understands operating-system objects such as sockets, not Python queues.

We therefore create two private sockets that are already connected:

self.notification_reader, self.notification_writer = socket.socketpair()
self.notification_reader.setblocking(False)
self.notification_writer.setblocking(False)

The event loop watches notification_reader:

selector.register(
    history_worker.notification_reader,
    selectors.EVENT_READ,
    data=lambda _event_mask: history_worker.process_completed_jobs(),
)

After the worker places a result in completed_jobs, it sends one notification byte:

try:
    self.notification_writer.send(b"\0")
except BlockingIOError:
    pass

b"\0" is not empty bytes. It contains one byte whose numeric value is zero. Its contents do not matter. It is only a doorbell:

"A database operation has finished. Look in completed_jobs."

Writing that byte makes notification_reader readable, so selector.select() wakes and returns its registration. The event-loop thread drains the notification bytes and then takes the actual outcomes from completed_jobs:

def process_completed_jobs(self) -> None:
    while True:
        try:
            received_bytes = self.notification_reader.recv(4096)
        except BlockingIOError:
            break

        if received_bytes == b"":
            break

    while True:
        try:
            completed, result, error = self.completed_jobs.get_nowait()
        except queue.Empty:
            return

        completed(result, error)

The completion callback runs on the event-loop thread. It can therefore safely work with Relay's selector, rooms, sessions, and client output buffers.

What Callback Actually Runs Here?

The callback stored on the notification socket is:

data=lambda _event_mask: history_worker.process_completed_jobs()

When the worker sends the notification byte, the notification socket becomes readable. The selector returns this registration, and the event-loop thread calls the stored lambda. The lambda runs process_completed_jobs(). That function discards the notification bytes, removes all finished database outcomes from completed_jobs, and calls completed(result, error) for each one.

completed is not one fixed function. It is the callback submitted with that particular database job. For a HISTORY job, it calls finish_loading_history(). On success, that function builds HISTORY_BEGIN, the stored messages, and HISTORY_END, then queues those bytes for the requesting client. If loading failed, it queues ERR history unavailable instead.

For a MSG job, the callback calls finish_storing_message(). If SQLite stored the message successfully, that function queues the message for every client currently in the room. If storing failed, it reports the error to the sender and does not broadcast the message. Therefore, Relay broadcasts a message only after it has been stored successfully.

The two directions are intentionally different:

event-loop thread → worker thread
jobs.put(...)
→ Queue wakes worker sleeping in jobs.get()

worker thread → event-loop thread
completed_jobs.put(...)
notification_writer.send(b"\0")
→ selector wakes because notification_reader is readable

The notification byte never carries the database result. The queue carries the result; the byte only wakes the thread that knows how to process it.

This gives Relay one event-loop thread for all network clients and one dedicated SQLite thread. If Bob's database operation blocks, only the history worker waits. Steve's network events can still be processed immediately.

Building An Async Runtime From Scratch In Python

Building An Async Runtime From Scratch In Python

Blocking Chat Server

The Problem

Selector-Driven Callback Server

Non-Blocking Writes And Slow Clients

Limit The Outgoing Queue

Heartbeats: Waking The Selector For Time

The Two Heartbeat Times

The Key Timer Idea: selector.select(timeout)

Choosing The Selector Timeout

Running Heartbeat Checks

Persistent Room History

Moving Blocking Sqlite Work To Another Thread

Sending Work To The Database Thread

Returning The Result To The Event Loop

What Callback Actually Runs Here?

The Key Timer Idea: `selector.select(timeout)`