Project — Simulating the Spanning Tree Protocol

What it is#

A simulator of IEEE 802.1D Spanning Tree Protocol. The input is a switched-LAN graph: switches connected by point-to-point links, each switch with a unique numeric ID, possibly containing physical loops. The output is a loop-free spanning tree, computed cooperatively by the switches themselves: every switch knows who the root is, which of its ports forwards (the root port and zero or more designated ports), and which ports block (everything else). Traffic to the root flows along the tree; the blocking ports stay idle until the tree changes.

The project teaches three things at once. First, distributed agreement under a deterministic rule: the lowest bridge ID wins the root election, ties broken by the lowest-cost path, then by sender ID. Second, port roles — root port, designated port, blocked port — and how each is chosen from the BPDUs a switch hears. Third, why STP exists in the first place: Ethernet has no TTL, so a single physical loop in a switched LAN broadcasts itself to death within milliseconds.

When to use it#

Build this when you want to:

• See why loops are catastrophic on Ethernet. Add a “no STP” mode to the simulator: every broadcast on a looped graph multiplies without bound. STP is the only thing keeping real LANs alive.
• Understand the BPDU exchange. A switch advertises a tuple (root_id, cost, sender_id). The rule for “is this BPDU better than mine?” is just lexicographic comparison. The whole protocol falls out from that.
• Practise distributed-systems reasoning. Every switch makes local decisions from local BPDUs and converges on a globally consistent tree without any central authority.
• Build a teaching prop. Print BPDUs as they exchange and you can walk a class through convergence step by step.

Skip this if you actually need a working switch. Real STP (and RSTP, MSTP) handles timers, topology change notifications, port states (disabled / blocking / listening / learning / forwarding), and edge ports — far more than this simulator. The project is for the core algorithm, not a production switch.

How it works#

The topology#

Four switches, IDs 1–4, with one loop. Every link costs 19 (the legacy STP cost for 100 Mb/s; modern RSTP would use 200000 for 100 Mb/s and 20000 for 1 Gb/s).

       1 ─────── 2
       │         │
       │         │
       4 ─────── 3
       │
       └── (also link 4-2, making a loop)

Adjacency:

S1 ─ S2
S2 ─ S3
S3 ─ S4
S4 ─ S1
S2 ─ S4   <-- the loop-closing link

Switch 1 has the lowest ID, so it will be elected root. The link S2-S4 is the one we expect to be blocked — it’s a “back link” that exists only to provide loop redundancy.

The simulator#

"""stp_sim.py — simulate IEEE 802.1D Spanning Tree Protocol.

Each switch is a thread that exchanges BPDUs with its neighbours on each link.
BPDU = (root_id, root_path_cost, sender_id). The lexicographically smallest
BPDU wins. A switch picks one root port (toward root), designates ports on
links where its own BPDU wins, and blocks the rest.
"""
import queue
import threading
import time
from dataclasses import dataclass, field

LINK_COST = 19            # legacy STP cost for 100 Mb/s
HELLO_INTERVAL = 1.0      # condensed from STP's real 2 s, for fast demos
CONVERGE_AFTER = 5.0      # seconds with no improvement = converged


@dataclass(order=True)
class BPDU:
    root_id: int
    cost: int
    sender_id: int


@dataclass
class Port:
    link_id: str          # unique link label, e.g. "S1-S2"
    peer_id: int
    inbox: queue.Queue
    peer_inbox: queue.Queue
    received: BPDU | None = None       # best BPDU heard on this port
    role: str = 'unknown'              # 'root' | 'designated' | 'blocked'


class Switch(threading.Thread):
    def __init__(self, sw_id: int) -> None:
        super().__init__(daemon=True)
        self.id = sw_id
        self.ports: dict[str, Port] = {}
        # initial belief: I am the root, cost 0 to myself
        self.best = BPDU(root_id=sw_id, cost=0, sender_id=sw_id)
        self.root_port: str | None = None
        self.running = True
        self.last_change = time.monotonic()
        self.lock = threading.Lock()

    def attach(self, link_id: str, peer: 'Switch', inbox: queue.Queue,
               peer_inbox: queue.Queue) -> None:
        self.ports[link_id] = Port(link_id, peer.id, inbox, peer_inbox)

    def send_bpdus(self) -> None:
        """Send my current best BPDU + LINK_COST out every port."""
        with self.lock:
            out = BPDU(self.best.root_id, self.best.cost, self.id)
        for port in self.ports.values():
            port.peer_inbox.put((port.link_id, out))

    def recompute(self) -> bool:
        """Look at all received BPDUs and recompute self.best + port roles.

        Returns True if anything changed.
        """
        with self.lock:
            previous_best = self.best
            previous_roles = {pid: p.role for pid, p in self.ports.items()}
            previous_rp = self.root_port

            # Step 1: find the best BPDU I've heard (including my own).
            candidates: list[tuple[BPDU, str | None]] = [
                (BPDU(self.id, 0, self.id), None)  # my own self-claim
            ]
            for pid, port in self.ports.items():
                if port.received is not None:
                    # cost is what the neighbour said + the cost of the link to them
                    advertised = BPDU(
                        port.received.root_id,
                        port.received.cost + LINK_COST,
                        port.received.sender_id,
                    )
                    candidates.append((advertised, pid))

            best_bpdu, best_port = min(candidates, key=lambda x: (x[0].root_id,
                                                                  x[0].cost,
                                                                  x[0].sender_id))
            self.best = BPDU(best_bpdu.root_id, best_bpdu.cost, self.id)
            self.root_port = best_port  # None means "I am the root"

            # Step 2: assign roles to every port.
            for pid, port in self.ports.items():
                if pid == self.root_port:
                    port.role = 'root'
                    continue
                # On each non-root port, am I the better advertiser on this link?
                # I would advertise (root_id, my_cost, my_id) outward.
                my_out = BPDU(self.best.root_id, self.best.cost, self.id)
                # The neighbour's last claim, as received, before adding link cost:
                neighbour = port.received
                if neighbour is None:
                    port.role = 'designated'  # never heard from neighbour yet
                    continue
                # On this link, the lower (root, cost, sender) advertises.
                if (my_out.root_id, my_out.cost, my_out.sender_id) < (
                    neighbour.root_id, neighbour.cost, neighbour.sender_id
                ):
                    port.role = 'designated'
                else:
                    port.role = 'blocked'

            changed = (
                previous_best != self.best
                or previous_rp != self.root_port
                or any(previous_roles.get(p) != self.ports[p].role for p in self.ports)
            )
            if changed:
                self.last_change = time.monotonic()
            return changed

    def absorb(self, link_id: str, bpdu: BPDU) -> None:
        with self.lock:
            port = self.ports.get(link_id)
            if port is None:
                return
            port.received = bpdu
        if self.recompute():
            self.send_bpdus()

    def run(self) -> None:
        next_hello = time.monotonic() + HELLO_INTERVAL
        self.send_bpdus()
        while self.running:
            timeout = max(0.0, next_hello - time.monotonic())
            try:
                link_id, bpdu = next(iter(self.ports.values())).inbox.get(timeout=0.05)
            except (queue.Empty, StopIteration):
                # Round-robin across all port inboxes (any one ready is fine).
                got = self._poll_all()
                if got is None:
                    if time.monotonic() >= next_hello:
                        self.send_bpdus()
                        next_hello = time.monotonic() + HELLO_INTERVAL
                    continue
                link_id, bpdu = got
            self.absorb(link_id, bpdu)

    def _poll_all(self) -> tuple[str, BPDU] | None:
        for port in self.ports.values():
            try:
                return port.inbox.get_nowait()
            except queue.Empty:
                continue
        return None

    def snapshot(self) -> dict:
        with self.lock:
            return {
                'id': self.id,
                'root': self.best.root_id,
                'cost': self.best.cost,
                'root_port': self.root_port,
                'ports': {pid: p.role for pid, p in self.ports.items()},
            }


def link(a: Switch, b: Switch, name: str) -> None:
    qa, qb = queue.Queue(), queue.Queue()
    a.attach(name, b, inbox=qa, peer_inbox=qb)
    b.attach(name, a, inbox=qb, peer_inbox=qa)


def build_loop() -> dict[int, Switch]:
    s = {i: Switch(i) for i in (1, 2, 3, 4)}
    link(s[1], s[2], 'L12')
    link(s[2], s[3], 'L23')
    link(s[3], s[4], 'L34')
    link(s[4], s[1], 'L41')
    link(s[2], s[4], 'L24')   # the loop-closing back link
    for sw in s.values():
        sw.start()
    return s


def print_state(switches: dict[int, Switch]) -> None:
    for sid in sorted(switches):
        snap = switches[sid].snapshot()
        ports = ' '.join(f'{p}={r[0].upper()}' for p, r in sorted(snap['ports'].items()))
        print(f"  S{sid}: root=S{snap['root']} cost={snap['cost']} "
              f"rp={snap['root_port']} | {ports}")


if __name__ == '__main__':
    switches = build_loop()
    time.sleep(5)
    print('converged spanning tree:')
    print_state(switches)

Expected output (port roles abbreviated: R = root, D = designated, B = blocked):

converged spanning tree:
  S1: root=S1 cost=0  rp=None  | L12=D L41=D
  S2: root=S1 cost=19 rp=L12   | L12=R L23=D L24=D
  S3: root=S1 cost=38 rp=L23   | L23=R L34=D
  S4: root=S1 cost=19 rp=L41   | L24=B L34=B L41=R

The tree: S1 is the root. S2 reaches root via L12 (cost 19). S4 reaches root via L41 (cost 19). S3 reaches root via L23 (cost 38). The L24 link is blocked at S4’s side because S2 has a lower cost-to-root (19 vs 19, then lower sender ID wins on the link); the L34 link is blocked at S4’s side for the same kind of reason. Either way, the loop is gone.

The rule in one sentence#

A BPDU advertises three things: who the root is, what it costs the sender to reach root, and who the sender is. Compare lexicographically — (root_id, cost, sender_id) ascending. The smallest wins. That single rule decides every port role on every switch.

Tracing convergence#

At t = 0 every switch thinks it is the root. S1 emits (1, 0, 1), S2 emits (2, 0, 2), etc. On each link, the better BPDU wins:

1. S2 receives (1, 0, 1) from S1 + LINK_COST = (1, 19, 1). That beats its own (2, 0, 2). S2 updates best = (1, 19, 2) and re-advertises.
2. S4 receives (1, 0, 1) from S1 + LINK_COST = (1, 19, 1). Same effect: best = (1, 19, 4).
3. S3 receives (1, 19, 2) + 19 = (1, 38, 2) from S2 and (1, 19, 4) + 19 = (1, 38, 4) from S4. Tied on cost, ties broken by sender — S2 wins (2 < 4). best = (1, 38, 3) via S2.
4. S2 and S4 both hear each other claim (1, 19, x). Each sees the other’s claim has equal (root, cost) but a different sender. On link L24 only one side can be designated: the lower sender ID (S2) wins, so S4’s port on L24 becomes blocked.
5. S3 and S4 likewise compare on link L34: both advertise (1, 19+19=38, x) outward, S3 has the lower ID, so S4’s L34 becomes blocked.

The whole thing converges in three or four round-trips, each one a HELLO_INTERVAL apart.

Variants#

• Topology change. Add a Switch.unplug(link_id) method that removes a port and forces a recompute. The neighbour eventually times out the BPDU it was hearing. Watch the tree re-form.
• Per-port cost configuration. Real STP lets you bias the tree by setting link cost. Add port.cost and use it in advertised.cost = port.received.cost + port.cost. Now you can experimentally shift the root or change which link is blocked.
• RSTP-style edge / point-to-point ports. Edge ports (connected to hosts, not switches) skip the listening/learning states. Point-to-point ports between switches use a fast handshake. Net effect: convergence in milliseconds rather than 30+ seconds.
• MSTP. Multiple Spanning Tree Protocol: one tree per VLAN group. Run N independent instances with different priorities for the same root candidate.
• BPDU guard. A port that suddenly receives a BPDU when it shouldn’t (e.g. an edge port) gets disabled. Useful demo of how operators trade convergence flexibility for security.
• TCN (topology change notification). When a port changes state, flood a TCN BPDU up to the root, which then flushes MAC-address tables. Simulating this teaches why a topology change triggers a brief flooding storm in real LANs.

Trade-offs#

Other tensions worth weighing:

• Convergence speed vs stability. Classic STP takes 30–50 s on a topology change; RSTP takes under a second. RSTP achieves this with proposal/agreement handshakes — more protocol complexity in exchange for fast recovery.
• Cost values. Legacy STP costs (19 for 100 Mb/s, 4 for 1 Gb/s, 2 for 10 Gb/s) saturate at 16 bits and can’t distinguish 10 Gb/s from 100 Gb/s. RSTP uses 20-bit costs and re-scales.
• Bridge ID structure. Real STP bridge ID = 16-bit priority + 48-bit MAC. Priority is what operators tune (default 32768; lower wins). The simulator collapses this to a single integer.
• What “blocked” means. Blocked ports still receive BPDUs (otherwise they’d never learn the tree changed). They just don’t forward data frames. The simulator doesn’t carry data frames at all; in a real switch this distinction is built into the port state.

Common pitfalls#

• Comparing BPDUs by root_id alone. Once everyone agrees on the root, the interesting comparisons are on cost and then sender_id. Skip those tie-breakers and ports flap forever.
• Forgetting to add LINK_COST. A BPDU received on a port advertises the sender’s cost to root. Your switch’s cost via that port is received.cost + LINK_COST. Miss this and every port thinks it’s one hop closer to root than it really is.
• Re-advertising your received BPDU instead of your new best. A switch’s outgoing BPDU is (best.root, best.cost, self.id) — self.id, not the original sender. Get this wrong and the tree looks plausible but cycles are not actually broken.
• Treating “no BPDU heard yet” as “I’m worse here”. On a fresh link, you should advertise designated until you hear something better. The simulator’s if neighbour is None: port.role = 'designated' handles this.
• Not re-broadcasting on change. A switch that updates best but doesn’t push the new BPDU out leaves its neighbours with stale beliefs. Always emit on change, plus periodically.
• Picking a root port from a blocked candidate. The candidate list for root port should include every received BPDU; blocking is a consequence of the role decision, not a precondition. Filter blocked ports out of the candidate list and you’ll pick wrong on some topologies.
• Ignoring sender ID in the tie-breaker. Two equal-cost paths to root are common. The lower sender wins on each link — without this rule, both sides think they’re designated and you have a loop again.
• Letting BPDUs go stale forever. In a real protocol, BPDUs have a max-age (default 20 s). If you don’t time them out, a vanished neighbour’s stale BPDU keeps influencing decisions. The simulator above relies on switches not crashing; add a per-port age field for realism.
• Skipping the lock when threads share state. The recv thread mutates port.received while the recompute path reads it. Without the lock you’ll see intermittent role flapping in the printout.
• Trusting BPDU(order=True) to compare three fields lexicographically. dataclass(order=True) does compare in declaration order, which is the correct STP order here. But if you reorder the fields for readability, comparisons silently change meaning. Pin the order with an explicit __lt__ or use a tuple for comparison.

Related building blocks#

• Spanning Tree Protocol
• Ethernet
• MAC Addressing & CSMA
• Project — A Distance-Vector (RIP) Routing Simulator
• Project — Dijkstra’s Shortest-Path for Link-State Routing