Heaps and priority queues in object design

Scheduling, ranking and timers with a heap behind a clean interface.

Low-level design conceptState, data structures and algorithms inside objects2 Nov 2026

01The intuition

The problem it solves

Imagine a small clinic with one doctor. Patients arrive and each one is given an urgency number from 1 (emergency) to 5 (routine). The receptionist must always send in the most urgent patient next. The obvious program keeps a list and, when the doctor is free, scans the whole list for the smallest number.

class Clinic:
    def __init__(self) -> None:
        self.waiting: list[tuple[int, str]] = []   # (urgency, name)

    def arrive(self, urgency: int, name: str) -> None:
        self.waiting.append((urgency, name))

    def next_patient(self) -> str:
        best = 0
        for i in range(1, len(self.waiting)):
            if self.waiting[i][0] < self.waiting[best][0]:
                best = i
        return self.waiting.pop(best)[1]

With 10 patients this is fine. With 100,000 queued tasks in a scheduler, every next_patient scans 100,000 entries, and the pop shifts the rest of the list down. Count it: serving everyone costs about 100,000 × 100,000 / 2 = 5 billion comparisons. The structure is wrong, not the idea.

Build it one piece at a time

Step one: notice that the only question ever asked is "who is smallest right now". A heap answers exactly that. It is a list kept in a loose order where position 0 is always the smallest, and fixing the order after an insert or a removal costs log₂(n) swaps, about 17 for 100,000 items. Python ships one in heapq.

import heapq

class Clinic:
    def __init__(self) -> None:
        self.waiting: list[tuple[int, str]] = []

    def arrive(self, urgency: int, name: str) -> None:
        heapq.heappush(self.waiting, (urgency, name))

    def next_patient(self) -> str:
        return heapq.heappop(self.waiting)[1]

Step two: two patients with urgency 2 now compare by name, which is arbitrary and, for objects without ordering, a crash. Add an arrival counter so ties break by who came first. The tuple becomes (urgency, arrival_number, patient), and the patient itself is never compared.

from dataclasses import dataclass, field
from itertools import count

@dataclass(order=True)
class Entry:
    urgency: int
    seq: int
    patient: str = field(compare=False)

class Clinic:
    def __init__(self) -> None:
        self.waiting: list[Entry] = []
        self.seq = count()

    def arrive(self, urgency: int, name: str) -> None:
        heapq.heappush(self.waiting, Entry(urgency, next(self.seq), name))

    def next_patient(self) -> str:
        return heapq.heappop(self.waiting).patient

Step three: a patient leaves, or an urgency changes. Heaps cannot remove from the middle cheaply, so mark the entry dead and skip dead entries when they reach the top. This is called lazy deletion, and it is the move interviewers wait for.

class Clinic:
    def __init__(self) -> None:
        self.waiting: list[Entry] = []
        self.seq = count()
        self.alive: dict[str, Entry] = {}

    def arrive(self, urgency: int, name: str) -> None:
        e = Entry(urgency, next(self.seq), name)
        self.alive[name] = e
        heapq.heappush(self.waiting, e)

    def leave(self, name: str) -> None:
        self.alive.pop(name, None)          # the heap entry stays, now stale

    def next_patient(self) -> str:
        while self.waiting:
            e = heapq.heappop(self.waiting)
            if self.alive.get(e.patient) is e:
                del self.alive[e.patient]
                return e.patient
        raise LookupError("nobody waiting")

The table below shows the heap list after three arrivals and one departure. Notice that the list is not sorted; only the first slot is guaranteed smallest.

Operation	Heap list (urgency, seq)	alive
arrive(3, Ana)	[(3,0)]	Ana
arrive(1, Bo)	[(1,1), (3,0)]	Ana, Bo
arrive(2, Cy)	[(1,1), (3,0), (2,2)]	Ana, Bo, Cy
leave(Bo)	[(1,1), (3,0), (2,2)]	Ana, Cy
next_patient()	pops (1,1) stale, pops (2,2) → Cy	Ana

Why it works

The invariant is simple: the element at index 0 is less than or equal to everything else in the list. heappush and heappop each repair that invariant along one root-to-leaf path, which is why they cost log n rather than n. The interface of the clinic never mentions the heap, so callers only see arrive, leave and next_patient; the data structure is a private choice.

Where it breaks: if the key of an entry changes after it is pushed, the heap order is silently wrong. That is why urgency is frozen inside the entry and a change is modelled as leave plus arrive. The second trap is memory: with lazy deletion, stale entries pile up if many items leave and nobody calls next. A periodic rebuild when stale entries exceed half the heap keeps the size bounded.

serve n items: n × (push + pop) = n × 2 log₂ n, about 3.4 million steps for n = 100,000, against 5 billion with the scan

The trade-off you must say out loud

A heap gives you the minimum, and nothing else. If the system also needs "show me everyone with urgency 2" or "remove the third person", a sorted container or a balanced tree may be better, at the price of a larger constant factor. Say that a heap is the right choice when the dominant operation is take-the-best, and name the lazy-deletion cost. The follow-up that usually comes next is "what if the priority can change", and the answer is the leave-and-re-add trick, or an indexed heap that stores each item's position so a decrease-key can bubble it up in place.

Where it shows up in the 100 Google problems

Concurrent priority task scheduler: the heap holds tasks by priority and submission order; cancellation is lazy deletion.
Timer with many concurrent timers: a heap keyed by fire time tells the single primitive timer when to wake next.
Running-median container: two heaps, a max-heap for the lower half and a min-heap for the upper half.
Game matchmaking: waiting players ordered by rating, with timeouts handled as stale entries.
Leaderboard class: top(k) from a heap of scores, or a heap of size k over a stream.