Chapter 48 — Priority Over Arrival

The halt that arrives last

Her behavior engine wants the arm to sweep: five angles from 0 to 180 degrees, a fraction of a second apart. Three moves in, the distance sensor sees a hand at twelve centimetres and asks for a halt. That halt is the newest command in the system, and by arrival it sits at the back of the line behind two moves that are about to run. Dispatch in arrival order and the sweep finishes into the obstacle, after which the halt executes politely, stopping something that already happened.

The list you reach for first almost works. Append every incoming action, sort before each dispatch, pop the front. Sorting a list you keep appending to redoes the same ordering work on every dispatch, and the wasted comparisons are the smaller problem: a plain list carries no lock. The sensor thread appends while the dispatch loop pops, and which of those two wins is decided by where the interpreter happens to switch threads.

So the rule for every command that reaches hardware from here on: an action carries how urgent it is and says nothing about when it runs; the queue owns dispatch order, and the record the queue stores is arranged so that it never has to compare two actions to decide. The second half of that reads like an implementation detail. It is the worked failure at the end of the chapter.

◆ Note — the heap underneath, and why not just sort

queue.PriorityQueue keeps its items in a binary heap: a tree flattened into a list, kept just ordered enough that the smallest item is always at the root. Inserting or removing costs work proportional to the height of that tree, which for a thousand pending actions is ten comparisons instead of a thousand-element sort. The standard library also exposes the bare heap as heapq, with the same operations and no lock; PriorityQueue is that heap plus the mutex that lets a sensor thread and a dispatch loop touch it at the same instant.

Urgency in the first field

▣ Build · stage 1 — prove that arrival is ignored

# labs/action_controller.py
import queue
from dataclasses import dataclass, field


@dataclass
class Action:
    name: str
    params: dict = field(default_factory=dict)
    priority: int = 1


pending: queue.PriorityQueue = queue.PriorityQueue()
arriving = [
    Action("led.set", {"state": "on"}, priority=1),
    Action("servo.set", {"angle": 90}, priority=2),
    Action("halt", {}, priority=5),
]
for order, action in enumerate(arriving):
    pending.put((-action.priority, order, action))

while not pending.empty():
    neg_priority, order, action = pending.get()
    print(f"dispatch p{-neg_priority} (arrived {order}): {action.name} {action.params}")

$ uv run python -m labs.action_controller
dispatch p5 (arrived 2): halt {}
dispatch p2 (arrived 1): servo.set {'angle': 90}
dispatch p1 (arrived 0): led.set {'state': 'on'}

A heap hands back its smallest item, and the urgent action needs to come out first, so the stored key is -priority. The halt's -5 is smaller than the move's -2, which is smaller than the lamp's -1, and the arrival index prints in descending order purely to show that nothing consulted it. The middle field carries that index for now; picking the right value to put there is the one decision in this chapter that a reasonable person gets wrong.

▣ Build · stage 2 — a controller, a registry, and a counter

import itertools
from typing import Callable

PRIORITY = {"halt": 5, "safety": 4, "speech": 3, "motion": 2, "ambient": 1}


class ActionController:
    def __init__(self) -> None:
        self._queue: queue.PriorityQueue = queue.PriorityQueue()
        self._handlers: dict[str, Callable[[dict], dict | None]] = {}
        self._seq = itertools.count()

    def register(self, name: str, handler: Callable[[dict], dict | None]) -> None:
        self._handlers[name] = handler

    def enqueue(self, action: Action) -> None:
        self._queue.put((-action.priority, next(self._seq), action))

    def process_one(self, timeout: float = 0.1) -> bool:
        try:
            _, _, action = self._queue.get(timeout=timeout)
        except queue.Empty:
            return False
        handler = self._handlers.get(action.name)
        if handler is None:
            print(f"[SKIP] p{action.priority} {action.name}: no handler registered")
            return True
        print(f"[ACT ] p{action.priority} {action.name} {action.params}")
        handler(action.params)
        return True


if __name__ == "__main__":
    ctrl = ActionController()
    ctrl.register("servo.set", lambda p: print(f"       arm to {p['angle']} deg"))
    ctrl.register("led.set", lambda p: print(f"       led {p['state']}"))
    ctrl.enqueue(Action("led.set", {"state": "on"}, PRIORITY["ambient"]))
    ctrl.enqueue(Action("servo.set", {"angle": 90}, PRIORITY["motion"]))
    ctrl.enqueue(Action("servo.set", {"angle": 20}, PRIORITY["motion"]))
    ctrl.enqueue(Action("gripper.close", {}, PRIORITY["motion"]))
    while ctrl.process_one():
        pass

$ uv run python -m labs.action_controller
[ACT ] p2 servo.set {'angle': 90}
       arm to 90 deg
[ACT ] p2 servo.set {'angle': 20}
       arm to 20 deg
[SKIP] p2 gripper.close: no handler registered
[ACT ] p1 led.set {'state': 'on'}
       led on

Three things earn their lines here. PRIORITY is a table of names, because a 5 buried in a call site tells the next reader nothing while PRIORITY["halt"] tells them everything. The handler registry is the dispatch idea from chapter 17 pointed at hardware: the controller looks a callable up by name and never learns what a servo is, so any subsystem can claim a verb by registering for it, and a missing one says so out loud instead of vanishing. And the two same-priority moves came out in the order they went in, because the counter from itertools.count only ever increases, so the earlier of two tied actions holds the smaller second field. Also note timeout=0.1: a background dispatch thread that blocks for a tenth of a second waiting for work is a thread that costs nothing while idle.

▣ Build · stage 3 — every dispatch leaves a record

import json
import time
from pathlib import Path

ACTION_LOG = Path("glados/data/action_log.json")


    # inside ActionController: self.log: list[dict] = [] in __init__

    def process_one(self, timeout: float = 0.1) -> bool:
        try:
            _, _, action = self._queue.get(timeout=timeout)
        except queue.Empty:
            return False
        handler = self._handlers.get(action.name)
        if handler is None:
            print(f"[SKIP] p{action.priority} {action.name}: no handler registered")
            return True
        print(f"[ACT ] p{action.priority} {action.name} {action.params}")
        started = time.perf_counter()
        result = handler(action.params)
        self.log.append({
            "at": round(time.time(), 3),
            "action": action.name,
            "params": action.params,
            "priority": action.priority,
            "ms": round((time.perf_counter() - started) * 1000, 1),
            "reply": result,
        })
        return True

    def save_log(self, path: Path) -> None:
        path.parent.mkdir(parents=True, exist_ok=True)
        with open(path, "w") as f:
            json.dump(self.log, f, indent=2)

$ uv run python -m labs.action_controller   # last entry, your timestamp will differ
{
  "at": 1787409659.816,
  "action": "led.set",
  "params": {
    "state": "on"
  },
  "priority": 1,
  "ms": 0.0,
  "reply": {
    "status": "ok",
    "state": "on"
  }
}

This list is the only place that will ever hold the true order in which her body did things. Enqueue order is not it, and neither is the behavior engine's view, since urgency reshuffles both. Each entry is appended after the handler returns, so it records the reply the hardware sent back and how long the whole exchange took; the 0.0 above is a lambda returning instantly, and it stops being zero the moment a real servo is on the other end. The write happens once, at the end, because opening and serializing a file inside a dispatch loop would make logging the slowest thing the controller does.

On the bench, with a sensor allowed to vote

▣ Build · stage 4 — the sweep, and the hand that stops it

# labs/hardware_actions.py
import threading
import time

from labs.action_controller import ACTION_LOG, PRIORITY, Action, ActionController
from labs.hardware_link import command, open_link

STOP_CM = 15.0
LINK_LOCK = threading.Lock()


def talk(link, payload: dict) -> dict:
    with LINK_LOCK:                      # the queue owns a lock; the cable does not
        return command(link, payload)


def register_handlers(ctrl: ActionController, link) -> None:
    def move(params: dict) -> dict:
        reply = talk(link, {"command": "servo.set", "params": {"angle": params["angle"]}})
        time.sleep(0.4)                  # the horn takes longer to arrive than the wire does
        return reply

    def light(params: dict) -> dict:
        return talk(link, {"command": "led.set", "params": {"state": params["state"]}})

    def halt(params: dict) -> dict:
        dropped = ctrl.cancel("servo.set")
        print(f"       cancelled {dropped} pending move(s): {params['reason']}")
        return talk(link, {"command": "led.set", "params": {"state": "on"}})

    ctrl.register("servo.set", move)
    ctrl.register("led.set", light)
    ctrl.register("halt", halt)


def watch_for_obstacles(ctrl: ActionController, link, stop: threading.Event) -> None:
    while not stop.is_set():
        cm = talk(link, {"command": "distance.read"}).get("cm")
        if cm is not None and cm < STOP_CM:
            ctrl.enqueue(Action("halt", {"reason": f"obstacle at {cm:.0f} cm"},
                                PRIORITY["halt"]))
            return
        time.sleep(0.1)


def main() -> None:
    ctrl = ActionController()
    stop = threading.Event()
    with open_link() as link:
        register_handlers(ctrl, link)
        watcher = threading.Thread(target=watch_for_obstacles,
                                   args=(ctrl, link, stop), daemon=True)
        watcher.start()
        ctrl.enqueue(Action("led.set", {"state": "on"}, PRIORITY["ambient"]))
        for angle in (0, 45, 90, 135, 180):
            ctrl.enqueue(Action("servo.set", {"angle": angle}, PRIORITY["motion"]))
        while ctrl.process_one():
            pass
        stop.set()
    ctrl.save_log(ACTION_LOG)
    print(f"logged {len(ctrl.log)} dispatches to {ACTION_LOG}")


if __name__ == "__main__":
    main()

$ uv run python -m labs.hardware_actions   # measured on the bench — yours will vary
[ACT ] p2 servo.set {'angle': 0}
[ACT ] p2 servo.set {'angle': 45}
[ACT ] p2 servo.set {'angle': 90}
[ACT ] p5 halt {'reason': 'obstacle at 12 cm'}
       cancelled 2 pending move(s): obstacle at 12 cm
[ACT ] p1 led.set {'state': 'on'}
logged 5 dispatches to glados/data/action_log.json

Put your hand in front of the sensor during the sweep and the arm stops between angles, which is the entire chapter in one gesture. Read the run from the top: three moves went out, the watcher thread saw twelve centimetres and enqueued a halt, and the halt was dispatched next even though 135 and 180 had been waiting far longer. The halt handler then emptied the queue of moves, so the sweep does not resume when the obstacle leaves. The lamp was queued before everything and dispatched after everything, exactly as an ambient action should be. Your distances, the moment the hand lands, and therefore which move is interrupted will all be different.

◆ Note — why cancel drains the whole queue

A heap has no removal operation: it can tell you its smallest item cheaply and nothing else about its contents. So cancel(name) pulls every item out with get_nowait(), counts the ones whose action matches, and puts the survivors back. The order is unharmed, because each survivor returns with the same key it had. This is a fine way to drop three pending moves and a poor way to filter a queue ten thousand deep, at which point the usual answer is to leave the entries in place and mark them cancelled, letting process_one discard them as it reaches them.

Why this works: one comparison, decided left to right

Python compares tuples element by element and stops at the first pair that differs, so a tuple is a way to state several ordering rules at once and rank them. Each field in (-priority, seq, action) is one rule. Compare -5 against -2 and the answer arrives in the first field with nothing else examined. Compare two moves both stored at -2 and the first field ties, so the counter decides, and the earlier arrival wins. The third field is the payload, and it is placed where a comparison can never reach it.

That last sentence is the design. A dataclass defines no < unless you ask for one, so the heap cannot order two Action objects and will say so the first time it tries. The counter guarantees the try never happens: two calls to next() on the same itertools.count can never return the same number, which makes every second field distinct and every comparison decidable by field two at the latest. Wall-clock time looks like it would do the same job and does not, for reasons the failure below makes concrete.

One honest limit, and it matters on a machine with a motor. A priority queue orders work that is waiting. It has no power over work already running. When the halt arrived, the move to 90 degrees was still executing inside its handler, and the queue could do nothing but let it finish; the sweep stopped one step later than a reader might assume from the word "halt". Everything a handler does happens at the priority of whatever was dispatched before it, so the real safety property comes from keeping handlers short. Exercise 2 turns that observation into a move that can be interrupted mid-travel.

⚠ Worked failure — the queue that crashed on the second move

The counter in the middle field looks like ceremony, so an early version stores just the priority and the action. It survives every test with distinct priorities, then the behavior engine emits two moves at once:

    def enqueue(self, action: Action) -> None:
        self._queue.put((-action.priority, action))     # BUG: nothing breaks the tie

$ uv run python -m labs.hardware_actions
Traceback (most recent call last):
  File ".../labs/hardware_actions.py", line 57, in main
    ctrl.enqueue(Action("servo.set", {"angle": angle}, PRIORITY["motion"]))
  File ".../labs/action_controller.py", line 31, in enqueue
    self._queue.put((-action.priority, action))
  File ".../queue.py", line 150, in put
    self._put(item)
  File ".../queue.py", line 236, in _put
    heappush(self.queue, item)
TypeError: '<' not supported between instances of 'Action' and 'Action'

Read the frames from the bottom. The crash is not in your code at all; it is inside heappush, comparing the two items you handed it. The first move went in without complaint because an empty heap compares nothing. The second one had to be placed relative to the first, the two -2 keys tied, and the comparison fell through to the actions themselves. Now the tempting repair: use time.time() as the tiebreaker, since it also increases and carries useful information. It works on the bench and then fails in the field, because the system clock has a resolution, two actions enqueued inside the same tick get identical timestamps, and the same crash returns at whatever rate your platform's clock granularity dictates. A counter cannot tie. Keep the timestamp for the log, where being a real time is the point, and let an integer that only counts do the ordering.

Checkpoint, and a record of what her body did

✓ Checkpoint — what you can now do

I can say why the stored key is -priority and what would come out first if the minus sign were dropped.
Given (-2, 7, a) and (-2, 4, b), I can name which dispatches first and which field decided it.
I can explain why a monotonic counter is a safer tiebreaker than a timestamp, and describe the failure a timestamp produces on a coarse clock.
I can state what a priority queue does not do: reorder work already inside a handler.
I can explain why cancel has to empty and refill the queue, and when that stops being the right technique.
Handed the action log, I can reconstruct the order her hardware actually moved in, and say why the behavior engine's own order would not tell me.

⚡ Exercises — try first, then reveal

Exercise 1 — refuse to grow without limit. Give the controller a max_depth of 20. When an enqueue would exceed it, drop the lowest-priority pending action, print which one, and queue the newcomer. Prove it with a depth of 2 and three actions.

Drain the queue into a list and sort it: because the key is negated, the largest item is the least urgent, so items.pop() after a sort discards exactly the right one. Put the survivors back, then add the new action. With a depth of 2, a p1 lamp queued first and two moves behind it, the lamp is evicted and the print says so. The design question the exercise is really asking: a full queue means the dispatcher is not keeping up, and dropping the least urgent thing is a decision that should be visible in the log rather than silent.

Exercise 2 — make a move interruptible. Rewrite the move handler to travel in 10-degree steps, checking between steps whether anything more urgent is pending, and abandoning the rest of the travel if so. Watch the arm stop mid-sweep instead of at the next angle.

Add a peek to the controller: with self._queue.mutex: then return -self._queue.queue[0][0] if the list is non-empty, else None. Holding the mutex is what makes reading the heap's root safe while a sensor thread may be pushing. The handler compares that number against its own action's priority after each step and returns early when something bigger is waiting. This is cooperative preemption: nothing forces the handler to yield, it agrees to look. That agreement is the only kind available without threads per action, and it is why long handlers are a safety problem and not only a latency one.

Exercise 3 — feed the starving. Enqueue one p1 ambient action, then a p2 move every second forever. Confirm the ambient action never runs, then age the queue so that waiting long enough earns priority.

Starvation is the standing cost of strict priority ordering, and on her bench it looks like a status light that never updates while she is busy. Aging fixes it: once a second, drain the queue and re-enqueue each action with an effective priority of priority + int(waited_seconds / 10), storing the original enqueue time on the Action so the bonus is recomputed rather than compounded. Print the queue depth and the ambient action's effective priority each round, and watch it climb until it dispatches. Then decide, on purpose, which actions should be exempt: an action that ages past PRIORITY["halt"] would let an old lamp command outrank an emergency stop.

The controller now knows what ran, in what order, and how many milliseconds each exchange with the board took. Those millisecond figures are sitting in a list that grows for as long as she is awake, which makes them both the most useful numbers in the system and, averaged over a whole session, a lie: one four-second stall disappears into five hundred fast turns. Next comes a window that only remembers the recent past, and the arithmetic that keeps its average honest.