Pub/Sub: Fan-out over TCP

Before the code: messaging and replication, in plain English

Picture a group chat. One person publishes a message. The chat app knows who is in the room and forwards the message to each of them. That is publish and subscribe, and it is the simplest piece of Redis that is not a key-value store. Replication is the same idea pointed at a backup machine: the leader sends every write to a follower, who quietly replays them. Same plumbing, different audience.

Fan-out and follow-the-leader

A publisher writes once. The server forwards to every subscriber. A replica is just another subscriber, listening to the full write log.

Quiz: Quiz

Loading practice…

Pub/Sub is the smallest piece of Redis that is not a key-value store. A registry from channel to set of subscribers. PUBLISH iterates and writes the message to every socket. The kernel handles the delivery.

07-pubsub/server.py
python
# channel -> set of client sockets subscribed to it
SUBSCRIBERS: dict[str, set[socket.socket]] = defaultdict(set)

# A coarse lock guarding SUBSCRIBERS. Held only during membership
# changes and during the snapshot read in PUBLISH.
SUB_LOCK = threading.Lock()

SUBSCRIBERS is a dict from channel name to set of client sockets. Pub/Sub is literally fanning out a payload to every socket in a set.

07-pubsub/server.py
python
def cmd_publish(args, *_):
    channel, message = args[0], args[1]
    with SUB_LOCK:
        subs = list(SUBSCRIBERS.get(channel, ()))
    delivered = 0
    payload = encode_array(["message", channel, message])
    for s in subs:
        try:
            s.sendall(payload); delivered += 1
        except OSError:
            with SUB_LOCK:
                SUBSCRIBERS.get(channel, set()).discard(s)
    return encode_integer(delivered)

Take a snapshot of the subscribers under the lock, then release the lock before sending. Holds the lock for microseconds, not for the duration of the write.

Pub/Sub demos require at least two clients: one subscribed, one publishing. The blocking single-client server cannot do that. One thread per client is the simplest fix. The event loop in the final step is the proper solution.

Quiz: Quiz

Loading practice…