Lesson 18: Framing & Concurrency

🧱 Why a Stream Needs Frames

Imagine someone reading you two sentences over the phone without pausing: "move x one move x two". You hear every word in order, but where does the first message end? TCP works the same way. It delivers a stream of bytes, reliably and in order, but it has no idea where one of your messages stops and the next begins.

This complete program sends two JSON messages and reads once. socket.socketpair() gives two already-connected sockets, handy for experiments.

import json
import socket

sender, receiver = socket.socketpair()      # two connected sockets, no server needed

sender.sendall(json.dumps({"type": "move", "x": 1}).encode("utf-8"))
sender.sendall(json.dumps({"type": "move", "x": 2}).encode("utf-8"))

data = receiver.recv(4096)                  # "give me what has arrived"
print("one recv() returned:", data)
try:
    print(json.loads(data))
except json.JSONDecodeError as exc:
    print("json.loads failed:", exc)

sender.close()
receiver.close()

On most machines both messages arrive in the one recv(), and json.loads fails with Extra data. On a slow network the opposite happens too: one message arrives split across two recv() calls. Either way, "one recv() is one message" is a bug that hides on your own computer and shows up for real players. The fix is to frame each message so the receiver knows exactly where it ends.

📏 Length-Prefixed Messages

The simplest reliable frame puts the message's length in front of it, like writing the page count on an envelope:

The receiver reads exactly 4 bytes, learns N, then reads exactly N bytes. Python's struct module turns numbers into fixed-size bytes and back. struct.Struct("!I") describes the header: ! means network byte order (big-endian, the same on every machine) and I means a 4-byte unsigned integer. Its pack(n) returns 4 bytes and its unpack(data) returns a one-item tuple.

Here is the whole framing layer as one module. Save it as framing.py: the Lobby Server & Client lesson imports it unchanged. Run it directly and it tests itself.

"""framing.py: length-prefixed JSON messages over a TCP stream.

The reference framing layer from Advanced Lesson 18 (Framing & Concurrency).
The lobby lab (Advanced Lesson 21) copies this file unchanged
next to its own program and does ``from framing import send_msg, recv_msg``.

Wire format of one message (a "frame"):

    +----------------------+---------------------------------+
    | 4-byte length N      | N bytes of UTF-8 JSON           |
    | unsigned, big-endian |                                 |
    +----------------------+---------------------------------+

Public API
----------
send_msg(sock, message)      JSON-encode ``message`` and send the whole frame (sendall).
recv_msg(sock)               Read one frame; return the decoded message, or None if the
                             peer closed the connection cleanly between frames.
recv_exact(sock, n)          Read exactly n bytes; None on a clean close before byte 1.
encode_msg(message)          The frame as bytes (header + body), without sending it.
FrameBuffer                  For non-blocking sockets / selectors: ``feed(data)`` returns
                             the list of complete messages, keeping any partial frame.
FramingError                 Raised for an oversized frame or a body that is not JSON.
MAX_MESSAGE_BYTES            Largest body accepted (1 MiB), so a bad header can't make
                             the receiver allocate gigabytes.

A blocking socket is assumed for send_msg / recv_msg. A peer that disconnects in
the middle of a frame raises ConnectionError.
"""
import json
import struct

HEADER = struct.Struct("!I")          # 4 bytes, network byte order, unsigned
MAX_MESSAGE_BYTES = 1024 * 1024       # 1 MiB


class FramingError(ValueError):
    """A frame that can't be a valid message: too large, or not UTF-8 JSON."""


def encode_msg(message):
    """Return one frame (length header + JSON body) for ``message``."""
    body = json.dumps(message, separators=(",", ":")).encode("utf-8")
    if len(body) > MAX_MESSAGE_BYTES:
        raise FramingError(f"message is {len(body)} bytes; the limit is {MAX_MESSAGE_BYTES}")
    return HEADER.pack(len(body)) + body


def decode_body(body):
    """Turn a frame body back into a message, or raise FramingError."""
    try:
        return json.loads(body.decode("utf-8"))
    except (UnicodeDecodeError, json.JSONDecodeError) as exc:
        raise FramingError(f"frame body is not UTF-8 JSON: {exc}") from None


def send_msg(sock, message):
    """Send one whole message. sendall() loops until every byte is handed to the OS."""
    sock.sendall(encode_msg(message))


def recv_exact(sock, n):
    """Read exactly n bytes from a blocking socket.

    Returns None if the peer closed the connection before the first byte
    (a clean end of stream). Raises ConnectionError if it closed part-way.
    """
    chunks = []
    received = 0
    while received < n:
        chunk = sock.recv(n - received)        # may return fewer bytes than asked for
        if not chunk:                          # b"" means the peer closed its side
            if received == 0:
                return None
            raise ConnectionError(f"peer closed after {received} of {n} bytes")
        chunks.append(chunk)
        received += len(chunk)
    return b"".join(chunks)


def recv_msg(sock):
    """Read one message. Returns None when the peer closed cleanly between messages."""
    header = recv_exact(sock, HEADER.size)
    if header is None:
        return None
    (length,) = HEADER.unpack(header)
    if length > MAX_MESSAGE_BYTES:
        raise FramingError(f"frame claims {length} bytes; the limit is {MAX_MESSAGE_BYTES}")
    body = recv_exact(sock, length) if length else b""
    if body is None:
        raise ConnectionError("peer closed between a frame header and its body")
    return decode_body(body)


class FrameBuffer:
    """Reassemble frames from whatever byte chunks a non-blocking socket delivers.

        buf = FrameBuffer()
        for message in buf.feed(sock.recv(4096)):
            handle(message)
    """

    def __init__(self):
        self._data = bytearray()

    def feed(self, data):
        """Add received bytes; return every message that is now complete, in order."""
        self._data.extend(data)
        messages = []
        while len(self._data) >= HEADER.size:
            (length,) = HEADER.unpack_from(self._data)
            if length > MAX_MESSAGE_BYTES:
                raise FramingError(f"frame claims {length} bytes; the limit is {MAX_MESSAGE_BYTES}")
            end = HEADER.size + length
            if len(self._data) < end:
                break                                  # wait for the rest of this frame
            messages.append(decode_body(bytes(self._data[HEADER.size:end])))
            del self._data[:end]
        return messages

    def pending(self):
        """Number of buffered bytes that don't yet form a complete frame."""
        return len(self._data)


if __name__ == "__main__":
    # Self-test over a local socket pair (no network needed).
    import socket

    a, b = socket.socketpair()
    try:
        send_msg(a, {"type": "hello", "name": "alice"})
        send_msg(a, {"type": "move", "x": 1.5, "y": -2})
        assert recv_msg(b) == {"type": "hello", "name": "alice"}
        assert recv_msg(b) == {"type": "move", "x": 1.5, "y": -2}
        frames = encode_msg({"n": 1}) + encode_msg({"n": 2})
        buf = FrameBuffer()
        got = []
        for i in range(len(frames)):                   # one byte at a time: worst case
            got += buf.feed(frames[i:i + 1])
        assert got == [{"n": 1}, {"n": 2}] and buf.pending() == 0
        a.close()
        assert recv_msg(b) is None                     # clean close between frames
        print("framing.py self-test passed")
    finally:
        a.close()
        b.close()

Four details make it robust:

• sendall(), not send(). send() may hand only part of the data to the OS. sendall() loops until every byte is out, and sending header and body together in one call keeps a frame in one piece.
• recv_exact() loops. recv(4) may return 1, 2, 3 or 4 bytes. The loop keeps asking for "what's still missing" until it has exactly n.
• Clean close vs broken frame. b"" before any byte means the peer hung up between messages, so recv_msg() returns None. b"" in the middle of a frame is an error, so it raises ConnectionError.
• A size limit. A 4-byte header can claim up to about 4 GB. Checking MAX_MESSAGE_BYTES before reading the body stops a buggy or hostile client from making your server wait for, or allocate, a huge message.

FrameBuffer does the same job for sockets you don't want to block on: you feed it whatever bytes arrived and it hands back every message that is now complete. You will use it with selectors below.

🧵 Many Clients: Threads and Locks

accept() and recv() both wait. If the server handled one client inside its accept loop, the second player would wait in the queue until the first one left. The classic fix is one thread per client: the accept loop hands each new socket to its own handler thread and goes straight back to accept().

def accept_loop(self):
    while True:
        try:
            conn, _ = self.listener.accept()
        except OSError:                      # the listener was closed: shut down
            return
        threading.Thread(target=self.handle, args=(conn,), daemon=True).start()

daemon=True marks a thread that shouldn't keep the program alive: when the main program ends, daemon threads end with it, so a handler still waiting in recv() can't stop the server from exiting. (Common Questions says more.)

Now several threads touch the same data: the dictionary of connected players and the next free player ID. A step like this looks harmless:

pid = len(self.clients) + 1       # two threads can both read 1 here...
self.clients[pid] = conn           # ...and both register as player 1

Python can switch threads between any two steps, so two players joining at the same moment can get the same ID, and one of them silently replaces the other. A lock makes a group of steps happen as one: only one thread at a time can be inside a with lock: block. In Python that is threading.Lock(): create one lock per group of shared data, and wrap every read or write of that data in with self.lock:. A thread that reaches the block while another is inside simply waits its turn.

with self.lock:                    # the ID and the registration are one step
    pid = next(self.next_id)       # itertools.count(1): 1, 2, 3, ... never reused
    self.clients[pid] = (conn, send_lock, name)

Broadcasting needs two more rules:

1. Copy under the lock, send outside it. Sending can be slow. Take a snapshot of the client list while holding the lock, release it, then send, so one slow player can't freeze everyone who is trying to join or leave.
2. One sender per socket at a time. If two handler threads broadcast to the same player at the same moment, the bytes of their two frames can interleave and the player's stream is corrupted. Give each connection its own small send lock.

def broadcast(self, message):
    with self.lock:                              # copy the list while holding the lock...
        targets = list(self.clients.items())
    for pid, (conn, send_lock, _) in targets:    # ...then send without it
        try:
            with send_lock:                      # one sender per socket at a time
                send_msg(conn, message)
        except OSError:
            self.remove(pid)                     # that player is gone

🔀 One Thread, Many Sockets: selectors

Threads are easy to reason about per client, but every shared value needs a lock. The other classic design uses one thread and asks the operating system, "which of my sockets have data right now?" Python's selectors module is that question. Sockets are set to non-blocking, so a recv() returns immediately with whatever is there, and a FrameBuffer per client collects the bytes until a message is complete.

"""One thread, many clients: a framed echo server built on selectors (Advanced Lesson 18).

The server never blocks on one client. The selector tells it which sockets
have data waiting; each client gets its own FrameBuffer that collects bytes
until a whole length-prefixed message has arrived.
"""
import json
import selectors
import socket
import struct
import threading

HOST = "127.0.0.1"
HEADER = struct.Struct("!I")


def encode_msg(message):
    body = json.dumps(message).encode("utf-8")
    return HEADER.pack(len(body)) + body


class FrameBuffer:
    """Collect bytes from a non-blocking socket; hand back each complete message."""

    def __init__(self):
        self.data = bytearray()

    def feed(self, chunk):
        self.data.extend(chunk)
        messages = []
        while len(self.data) >= HEADER.size:
            (length,) = HEADER.unpack_from(self.data)
            end = HEADER.size + length
            if len(self.data) < end:
                break                                # the rest of this frame is still on its way
            messages.append(json.loads(self.data[HEADER.size:end].decode("utf-8")))
            del self.data[:end]
        return messages


def serve(listener, stop):
    sel = selectors.DefaultSelector()
    listener.setblocking(False)
    sel.register(listener, selectors.EVENT_READ, data=None)   # always registered: never empty
    while not stop.is_set():
        for key, _ in sel.select(timeout=0.1):
            if key.data is None:                     # the listener: a new client
                conn, _ = listener.accept()
                conn.setblocking(False)
                sel.register(conn, selectors.EVENT_READ, data=FrameBuffer())
                continue
            conn, buf = key.fileobj, key.data
            try:
                chunk = conn.recv(4096)
            except ConnectionError:
                chunk = b""
            if not chunk:                            # client closed: forget it
                sel.unregister(conn)
                conn.close()
                continue
            for msg in buf.feed(chunk):
                # Small replies to a local client: sendall on a non-blocking socket is fine
                # here. A real server queues output and waits for EVENT_WRITE instead.
                conn.sendall(encode_msg({"echo": msg}))
    for key in list(sel.get_map().values()):
        sel.unregister(key.fileobj)
        key.fileobj.close()
    sel.close()


def client(name, port, results):
    with socket.create_connection((HOST, port), timeout=5) as sock:
        frames = b"".join(encode_msg({"from": name, "n": i}) for i in range(3))
        sock.sendall(frames)                         # three messages in ONE send
        buf, replies = FrameBuffer(), []
        while len(replies) < 3:
            chunk = sock.recv(4096)
            if not chunk:
                break
            replies += buf.feed(chunk)
        results[name] = [r["echo"]["n"] for r in replies]


def main():
    listener = socket.socket(socket.AF_INET, socket.SOCK_STREAM)
    listener.bind((HOST, 0))
    listener.listen()
    stop = threading.Event()
    server = threading.Thread(target=serve, args=(listener, stop))
    server.start()
    results = {}
    clients = [threading.Thread(target=client, args=(name, listener.getsockname()[1], results))
               for name in ("alice", "bob", "carol")]
    for t in clients:
        t.start()
    for t in clients:
        t.join()
    stop.set()
    server.join()
    for name in sorted(results):
        print(f"{name}: echoes {results[name]}")


if __name__ == "__main__":
    main()

• sel.select(timeout=0.1) returns the sockets that are ready. The data you registered with each one (here, its FrameBuffer) comes back as key.data.
• The listening socket stays registered for the server's whole life, so the selector is never empty. That also avoids a Windows-only trap: Windows' select() fails with WinError 10022 when it is given no sockets at all.
• stop = threading.Event() is a thread-safe on/off flag. The main thread calls stop.set() when the demo is over, and the server loop checks stop.is_set() between select() calls, which is why select() has a short timeout: the loop wakes up at least ten times a second to look at the flag.
• No locks are needed, because only one thread touches the server's data.
• The trade-off: nothing in the loop may block. A slow step (a big sendall() to a full socket, a file load) stalls every client. Production servers queue outgoing data and wait for EVENT_WRITE, which is more code. For a small game server, either design works; pick the one you can debug.

📨 What About UDP?

UDP already keeps message boundaries: one sendto() is one datagram, and one recvfrom() returns it whole. So UDP needs no length prefix. What it needs is one receive loop, not a thread per client or per datagram, plus a sequence number so the receiver can ignore anything older than what it already has:

newest = {}                                   # sender address -> newest sequence number seen
while running:
    data, addr = sock.recvfrom(2048)          # one whole datagram per call
    msg = json.loads(data.decode("utf-8"))
    if msg["seq"] <= newest.get(addr, 0):
        continue                              # late or duplicate: a newer one already arrived
    newest[addr] = msg["seq"]
    handle(addr, msg)

Starting a thread for every datagram, as some tutorials do, creates hundreds of threads a second in a real game and lets updates from one player be handled out of order.

🏋️ Practice Exercise: Framed Chat Server

"""Framed Chat Server: Advanced Lesson 18 practice exercise (solution).

One program, three kinds of thread, all on 127.0.0.1:
  * the accept loop hands each new connection to its own handler thread;
  * each handler reads length-prefixed JSON messages from one client;
  * two client threads, alice and bob, join, chat and leave.
The server gives every player a unique ID and broadcasts chat to everyone.
"""
import itertools
import json
import socket
import struct
import threading

HOST = "127.0.0.1"
HEADER = struct.Struct("!I")          # 4-byte unsigned length, network byte order
MAX_MESSAGE_BYTES = 64 * 1024


# ------------------------------------------------------------------ framing
def send_msg(sock, message):
    body = json.dumps(message).encode("utf-8")
    sock.sendall(HEADER.pack(len(body)) + body)     # header and body in one sendall


def recv_exact(sock, n):
    """Exactly n bytes, or None if the peer closed before the first byte."""
    data = b""
    while len(data) < n:
        chunk = sock.recv(n - len(data))
        if not chunk:
            if not data:
                return None
            raise ConnectionError(f"peer closed after {len(data)} of {n} bytes")
        data += chunk
    return data


def recv_msg(sock):
    header = recv_exact(sock, HEADER.size)
    if header is None:
        return None                                  # clean close between messages
    (length,) = HEADER.unpack(header)
    if length > MAX_MESSAGE_BYTES:
        raise ValueError(f"message of {length} bytes is too large")
    body = recv_exact(sock, length)
    if body is None:
        raise ConnectionError("peer closed between a header and its body")
    return json.loads(body.decode("utf-8"))


# ------------------------------------------------------------------ server
class ChatServer:
    def __init__(self):
        self.listener = socket.socket(socket.AF_INET, socket.SOCK_STREAM)
        self.listener.setsockopt(socket.SOL_SOCKET, socket.SO_REUSEADDR, 1)
        self.listener.bind((HOST, 0))
        self.listener.listen()
        self.port = self.listener.getsockname()[1]
        self.lock = threading.Lock()                 # guards clients and next_id
        self.clients = {}                            # player id -> (socket, send lock, name)
        self.next_id = itertools.count(1)
        self.assigned_ids = []

    def accept_loop(self):
        while True:
            try:
                conn, _ = self.listener.accept()
            except OSError:                          # listener closed: shut down
                return
            threading.Thread(target=self.handle, args=(conn,), daemon=True).start()

    def broadcast(self, message):
        with self.lock:                              # copy the list while holding the lock...
            targets = list(self.clients.items())
        for pid, (conn, send_lock, _) in targets:    # ...then send without it
            try:
                with send_lock:                      # one sender per socket at a time
                    send_msg(conn, message)
            except OSError:
                self.remove(pid)

    def remove(self, pid):
        with self.lock:
            entry = self.clients.pop(pid, None)
        if entry is not None:
            entry[0].close()

    def handle(self, conn):
        pid = None
        try:
            hello = recv_msg(conn)
            if not hello or hello.get("type") != "hello":
                return
            name = str(hello.get("name", "?"))[:16]
            send_lock = threading.Lock()
            with self.lock:                          # the ID and the registration are one step
                pid = next(self.next_id)
                self.assigned_ids.append(pid)
                self.clients[pid] = (conn, send_lock, name)
            with send_lock:
                send_msg(conn, {"type": "welcome", "id": pid})
            print(f"[server] {name} joined as player {pid}")
            while True:
                msg = recv_msg(conn)
                if msg is None:                      # client hung up
                    break
                if msg.get("type") == "chat":
                    self.broadcast({"type": "chat", "from": name, "text": str(msg.get("text", ""))})
        except (OSError, ValueError) as exc:
            print(f"[server] dropped a client: {exc}")
        finally:
            if pid is not None:
                self.remove(pid)
                print(f"[server] player {pid} left")
            else:
                conn.close()


# ------------------------------------------------------------------ clients
def client(name, lines, port, ready, results):
    with socket.create_connection((HOST, port), timeout=5) as sock:
        send_msg(sock, {"type": "hello", "name": name})
        welcome = recv_msg(sock)
        ready.wait()                                 # everyone is connected before chatting
        for text in lines:
            send_msg(sock, {"type": "chat", "text": text})
        chats = []
        while len(chats) < 6:                        # alice and bob send three lines each
            msg = recv_msg(sock)
            if msg is None:
                break
            if msg["type"] == "chat":
                chats.append((msg["from"], msg["text"]))
        results[name] = (welcome["id"], chats)


def framing_works():
    """True once send_msg / recv_exact / recv_msg (TODOs 1-3) pass a quick check."""
    a, b = socket.socketpair()
    a.settimeout(1)
    b.settimeout(1)
    try:
        send_msg(a, {"n": 1})
        send_msg(a, {"n": 2})
        return recv_msg(b) == {"n": 1} and recv_msg(b) == {"n": 2}
    except (OSError, ValueError):
        return False
    finally:
        a.close()
        b.close()


def main():
    if not framing_works():
        print("Framing check failed: finish TODOs 1-3, then run again.")
        return
    server = ChatServer()
    threading.Thread(target=server.accept_loop, daemon=True).start()
    ready = threading.Barrier(2)
    results = {}
    scripts = {"alice": ["hi", "anyone here?", "gg"], "bob": ["hello", "ready", "gg wp"]}
    threads = [threading.Thread(target=client, args=(name, lines, server.port, ready, results))
               for name, lines in scripts.items()]
    for t in threads:
        t.start()
    for t in threads:
        t.join()
    server.listener.close()

    for name, (pid, chats) in sorted(results.items()):
        for sender, lines in scripts.items():
            got = [text for who, text in chats if who == sender]
            order = "in order" if got == lines else f"OUT OF ORDER {got}"
            print(f"[{name}] got {len(got)} lines from {sender}, {order}")
    print(f"Assigned IDs: {sorted(server.assigned_ids)} (all unique: "
          f"{len(set(server.assigned_ids)) == len(server.assigned_ids)})")


if __name__ == "__main__":
    main()

📓 Learning Journal

📝 Summary

TCP delivers bytes, not messages, so you gave every message a 4-byte length header, sent it with sendall(), and read it back with a loop that asks for exactly the bytes still missing. That framing layer, saved as framing.py, is what the Lobby Server & Client lesson builds on. You then served many clients: one thread each, with a lock around shared data so IDs stay unique, a copy-then-send broadcast, and a send lock per socket. Finally you saw the single-threaded alternative, selectors with a FrameBuffer per client.

🔭 Looking Ahead

You can now move messages safely. Next, in Client Prediction & Reconciliation, you use them for a real-time game: the server owns the world, and the client hides latency by predicting its own movement and fixing its guesses when the server replies.

❓ Common Questions

🎯 Quick Quiz

🌟 Going Further

• Private messages: add a {"type": "whisper", "to": 2, "text": ...} message that the server sends to one player only, and an error reply when that ID doesn't exist.
• Selector chat: rewrite the chat server with selectors and FrameBuffer, and compare the two versions' line counts and the number of locks.
• Stress test: start 50 client threads that join, send 20 lines each and leave. Check that every ID is unique and no line is lost.
• Read the docs: struct, selectors and threading Lock objects.
• Coming up in Game Dev III: Advanced: the Lobby Server & Client lesson builds rooms, ready checks and a pygame client on this same framing.py.