Priority Queue via Binary Heap, in depth: an indexed heap with update and remove, stable ties, a scheduler and the tests that keep it honest

The contract before the structure

Start from the interface. A minimal priority queue inserts, peeks at the minimum and removes it; for that, the standard library heap is enough. Once a caller holds an identity, such as a job id or a vertex, and wants to change or withdraw that item, you need two more operations, and they change the design.

The indexed version costs a hash map, one extra write per swap, and a rule that each item appears once, which turns duplicate-entry bugs into immediate errors.

The layout, and the one extra structure

A binary heap stores a complete binary tree level by level in an array. The parent of slot i is (i-1)//2 and its children are 2i+1 and 2i+2, so there are no pointers to maintain. The heap property says every parent is no larger than its children, which puts the minimum at slot 0 and says nothing about the order of anything else.

The indexed heap adds a position map from each item to its current slot. All movement goes through a single swap function that updates the map for both entries. That is the whole trick: if any path writes the array directly, the map goes stale and the next update moves the wrong item.

A complete indexed heap

Here is the full implementation in Python, with no dependencies. Each entry is a mutable list of priority, sequence number and item; the sequence number breaks ties, making the queue first-in, first-out among equal priorities.

import itertools

class IndexedMinPQ:
    """Min-priority queue keyed by hashable item ids. FIFO among equal priorities."""

    def __init__(self):
        self._heap = []                 # entries: [priority, seq, item]
        self._pos = {}                  # item -> index in _heap
        self._seq = itertools.count()   # tie-breaker, strictly increasing

    def __len__(self):
        return len(self._heap)

    def __contains__(self, item):
        return item in self._pos

    def push(self, item, priority):
        if item in self._pos:
            raise KeyError(f"{item!r} already queued; use update()")
        self._heap.append([priority, next(self._seq), item])
        self._pos[item] = len(self._heap) - 1
        self._sift_up(len(self._heap) - 1)

    def peek(self):
        if not self._heap:
            raise IndexError("peek from empty queue")
        priority, _, item = self._heap[0]
        return item, priority

    def pop(self):
        item, priority = self.peek()
        self._delete_at(0)
        return item, priority

    def update(self, item, priority):
        i = self._pos[item]                      # KeyError if absent: a caller bug
        old = self._heap[i][0]
        self._heap[i][0] = priority              # seq kept: item keeps its FIFO place
        if priority < old:
            self._sift_up(i)
        elif priority > old:
            self._sift_down(i)

    def remove(self, item):
        self._delete_at(self._pos[item])

    # -- internals -------------------------------------------------------
    def _less(self, i, j):
        a, b = self._heap[i], self._heap[j]
        return (a[0], a[1]) < (b[0], b[1])       # never compares items

    def _swap(self, i, j):
        h = self._heap
        h[i], h[j] = h[j], h[i]
        self._pos[h[i][2]] = i
        self._pos[h[j][2]] = j

    def _delete_at(self, i):
        last = len(self._heap) - 1
        if i != last:
            self._swap(i, last)
        _, _, item = self._heap.pop()
        del self._pos[item]
        if i < len(self._heap):                  # moved entry may need either direction
            self._sift_up(i)
            self._sift_down(i)

    def _sift_up(self, i):
        while i > 0:
            parent = (i - 1) // 2
            if not self._less(i, parent):
                break
            self._swap(i, parent)
            i = parent

    def _sift_down(self, i):
        n = len(self._heap)
        while True:
            smallest, left, right = i, 2 * i + 1, 2 * i + 2
            if left < n and self._less(left, smallest):
                smallest = left
            if right < n and self._less(right, smallest):
                smallest = right
            if smallest == i:
                return
            self._swap(i, smallest)
            i = smallest

Three details deserve attention. First, _less compares only priority and sequence number, so items can be any hashable value, even ones that cannot be ordered; a tuple of priority and payload would instead fall through to comparing payloads on a tie and raise TypeError. Use a counter, not a timestamp, because two pushes in one clock tick share a timestamp. Second, update keeps the original sequence number, so a rescheduled item keeps its place among equals; assign a fresh one if you want it sent to the back, but choose deliberately. Third, _delete_at handles removing the last slot, where there is nothing to swap or sift.

Why remove must try both directions

Popping the root moves the last entry to slot 0, and since nothing can be above slot 0, sifting down is enough. Removing from the middle is different. The last entry, which replaces the removed one, came from a different branch of the tree. It is guaranteed to be no smaller than its own old parent, but it has no relationship at all to its new parent. It may need to go down, or it may need to go up.

A worked example makes this concrete. Take the heap [1, 10, 2, 11, 12, 3, 4]. Slot 1 holds 10 with children 11 and 12; slot 2 holds 2 with children 3 and 4. Remove the 11 at slot 3. The last entry, 4 from slot 6, moves into slot 3. Its new parent is 10 at slot 1, and 4 is smaller than 10, so the heap property is now broken upwards. Sifting down from slot 3 does nothing, because slot 3 has no children. Only a sift up repairs it, swapping 4 with 10 to give [1, 4, 2, 10, 12, 3].

Sift-down-only implementations pass most tests, because the bad case is rare. Calling both sifts costs one extra comparison, since at most one moves anything, and removes the bug.

Worked example: a delayed-job scheduler

A scheduler that runs callbacks at due times is the canonical consumer of an indexed priority queue. The priority is the due time on a monotonic clock, the item is the job id, and rescheduling and cancellation are first-class operations.

import time

class DelayedScheduler:
    """Run each job at its due time; jobs can be rescheduled or cancelled by id."""

    def __init__(self, clock=time.monotonic):
        self._pq = IndexedMinPQ()
        self._jobs = {}
        self._clock = clock

    def schedule(self, job_id, due_at, fn):
        self._jobs[job_id] = fn
        if job_id in self._pq:
            self._pq.update(job_id, due_at)      # reschedule: O(log n)
        else:
            self._pq.push(job_id, due_at)

    def cancel(self, job_id):
        if job_id in self._pq:
            self._pq.remove(job_id)              # O(log n), no tombstones
            del self._jobs[job_id]

    def next_wait(self):
        """Seconds until the earliest job, or None when idle."""
        if not self._pq:
            return None
        _, due = self._pq.peek()
        return max(0.0, due - self._clock())

    def run_due(self, budget=100):
        """Run up to `budget` due jobs; bounded so one tick cannot starve the loop."""
        ran, now = 0, self._clock()
        while self._pq and ran < budget:
            job_id, due = self._pq.peek()
            if due > now:
                break
            self._pq.pop()
            self._jobs.pop(job_id)()
            ran += 1
        return ran

Trace it. Schedule A at t=10, B at t=5 and C at t=5. The heap root is B, because it ties with C on time and was pushed first. Now reschedule A to t=3: update lowers its priority and sifts it up to the root. Cancel C: remove takes it out in logarithmic time, with no tombstone left behind. At t=6, run_due pops A then B, sees nothing else is due, and returns 2. An event loop sleeps for next_wait() seconds between ticks.

The budget parameter is operational: after a long pause thousands of jobs can be due at once, and bounding work per tick keeps the loop responsive. The same structure, with deadlines in place of due times, underlies the deadline scheduling greedy algorithm.

Testing: an invariant checker and a model

Heap bugs are quiet: a broken heap still returns items, just not always the right ones. Two tests catch nearly everything: an invariant checker for the heap property and map consistency, and a model-based test that runs random operations against the heap and a trivially correct dictionary, asserting they agree on every pop.

import random

def check_invariant(pq):
    h = pq._heap
    for i in range(1, len(h)):
        assert (h[(i - 1) // 2][0], h[(i - 1) // 2][1]) <= (h[i][0], h[i][1]), i
    assert len(pq._pos) == len(h)
    for item, i in pq._pos.items():
        assert h[i][2] == item, (item, i)

def test_against_model(seed):
    rng, pq, model = random.Random(seed), IndexedMinPQ(), {}
    for step in range(5000):
        op = rng.choice("push push update remove pop".split())
        if op == "push":
            item = rng.randrange(300)
            if item not in model:
                pr = rng.randrange(20)
                pq.push(item, pr); model[item] = (pr, step)
        elif op == "update" and model:
            item = rng.choice(list(model)); pr = rng.randrange(20)
            pq.update(item, pr); model[item] = (pr, model[item][1])
        elif op == "remove" and model:
            item = rng.choice(list(model))
            pq.remove(item); del model[item]
        elif op == "pop" and model:
            want = min(model, key=lambda k: model[k])     # (priority, first-push step)
            got, _ = pq.pop()
            assert got == want, (step, got, want)
            del model[got]
        check_invariant(pq)

for seed in range(50):
    test_against_model(seed)

Run the invariant check in tests only, since it is linear. The model test is what finds the remove-direction bug, missed map updates and tie-breaking regressions. A property-based library with stateful testing, such as Hypothesis or jqwik, adds shrinking: a failing run of thousands of steps is reduced to the few that matter.

What standard libraries give you

Go's design is the most instructive: Fix and Remove work by index and the item-to-index map is yours, exactly the split used here. In Java, a common alternative is to mark cancelled entries and skip them on poll, which leads to the real trade-off.

Indexed or lazy?

Lazy deletion leaves stale entries in the heap and discards them when they reach the root. It needs no position map and keeps every operation a plain push or pop. Its costs are memory and latency variance: a workload that reschedules each job ten times holds ten entries per job, and a pop may have to discard a long run of dead entries before finding a live one. The Dijkstra article shows the lazy pattern where it shines, because each vertex is relaxed only a few times.

Choose the indexed heap when updates and cancellations are frequent, when memory must track live items, or when you need contains. Choose lazy deletion when updates are rare or the queue is short-lived. If you go lazy, rebuild from live entries when stale ones exceed half the array; heapify makes that linear. Either way, neither structure is thread-safe: wrap it in one lock or give it a single owner thread, and export queue length and root age (now minus its due time) as metrics.

Failure modes

What to do next

1. Write down the operations your callers need. If they never change or cancel items, use the standard library heap with a sequence counter and stop there.
2. If they do, implement the indexed heap above, with every array movement going through one swap function.
3. Decide and document whether update keeps an item's place among equals or sends it to the back.
4. Add the invariant checker and the model-based test to CI, and run them on every change to the queue.
5. Export queue length and root age as metrics, and bound work per tick in any scheduler loop.
6. Read the heap operations article for the linear-time build and d-ary heaps, then the k-way merge article for another heap you will meet in practice.