Software Systems Atlas

English

中文

English

中文

Theme

VOL 02 / CH 07 / LESSON 02

7.2 Priority Queues ​

Prerequisites: Heap | Next Chapter: Graph. This section encapsulates the heap as a software interface and clearly defines the contract for behavior under various scenarios: tie-breaking rules, updates, cancellation, Top-K retrieval, and K-way merging.

In the previous section, we manually implemented a heap. In real-world engineering code, developers are more concerned with interfaces: how to add a task, how to retrieve the next item, who gets priority when tasks have equal priority, and what happens when a task is canceled.

A priority queue is an abstract data type; a heap is one of its implementations. An unsorted array can implement a priority queue, appending takes amortized O(1), but each removal requires scanning O(n) elements. A sorted array makes removal efficient, but insertion degrades to O(n). A binary heap bounds both sift-up and sift-down by O(log n), and retrieving the root element is simply O(1).

The array-based relationship of the heap remains unchanged: parent node at (i - 1) // 2, left child at 2 * i + 1, right child at 2 * i + 2. On top of this structural foundation, the priority queue defines the business logic for ordering elements.

1. Equal Priority Requires a Second Criterion ​

Suppose a scheduler receives two dictionary tasks with priority 2. If Python stores only (priority, task), equal priorities force it to compare the task objects. Dictionaries are not orderable, so the program raises TypeError. Even if the task objects were orderable, their natural order would not mean “first come, first served.”

A solution is to insert a monotonic sequence number in between:

text
(priority, sequence, task)

Priority determines the primary order; when priorities are equal, the smaller sequence number is processed first. The sequence number is unique, so comparisons never need to reach the task object itself. A standard heap does not guarantee stability; the sequence number establishes a stable ordering contract.

2. Complete Implementation with Support for Updates and Cancellations ​

Binary heaps excel at root reassignment but are poor at locating a named task within the middle of an array. To address this, we use a map to locate the current entry. When canceling, we first mark the entry as deleted. The actual removal occurs only when the item reaches the root, then it is skipped. This approach is called lazy deletion. Smaller priorities leave first; ties follow the order of the most recent add. Updating a task gives it a new sequence number, so an equal-priority update rejoins at the back.

Names must be strings and priorities integers, checked before an old entry is changed. remove raises KeyError for an unknown name; pop raises KeyError when no active tasks remain. This teaching class provides no thread synchronization.

python
from __future__ import annotations

import heapq
import itertools
from operator import index as integer_index
from collections.abc import Iterable
from typing import Any, TypeVar

T = TypeVar("T")

class StablePriorityQueue:
    def __init__(self) -> None:
        self._heap: list[list[Any]] = []
        self._entries: dict[str, list[Any]] = {}
        self._counter = itertools.count()
        self._removed = object()

    def __len__(self) -> int:
        return len(self._entries)

    def add(self, name: str, payload: Any, priority: int = 0) -> None:
        if not isinstance(name, str):
            raise TypeError("name must be a string")
        if not isinstance(priority, int):
            raise TypeError("priority must be an integer")
        if name in self._entries:
            self.remove(name)
        entry = [priority, next(self._counter), name, payload]
        self._entries[name] = entry
        heapq.heappush(self._heap, entry)

    def remove(self, name: str) -> None:
        if not isinstance(name, str):
            raise TypeError("name must be a string")
        entry = self._entries.pop(name)
        entry[2] = self._removed
        entry[3] = None

    def compact(self) -> None:
        self._heap = list(self._entries.values())
        heapq.heapify(self._heap)

    def pop(self) -> tuple[str, Any, int]:
        while self._heap:
            priority, _sequence, name, payload = heapq.heappop(self._heap)
            if name is not self._removed:
                del self._entries[name]
                return name, payload, priority
        raise KeyError("pop from an empty priority queue")

def top_k_largest(values: Iterable[T], k: int) -> list[T]:
    k = integer_index(k)
    if k <= 0:
        return []
    heap: list[T] = []
    for value in values:
        if len(heap) < k:
            heapq.heappush(heap, value)
        elif heap[0] < value:
            heapq.heapreplace(heap, value)
    return sorted(heap, reverse=True)

def merge_k_sorted(groups: list[list[T]]) -> list[T]:
    heap: list[tuple[T, int, int]] = []
    for group_index, group in enumerate(groups):
        if group:
            heapq.heappush(heap, (group[0], group_index, 0))
    merged: list[T] = []
    while heap:
        value, group_index, item_index = heapq.heappop(heap)
        merged.append(value)
        next_index = item_index + 1
        if next_index < len(groups[group_index]):
            next_value = groups[group_index][next_index]
            heapq.heappush(heap, (next_value, group_index, next_index))
    return merged

if __name__ == "__main__":
    queue = StablePriorityQueue()
    queue.add("backup", {"node": 3}, priority=2)
    queue.add("report", {"node": 1}, priority=2)
    queue.add("alert", {"node": 7}, priority=1)
    print(queue.pop())
    print(queue.pop())
    print(top_k_largest([3, 2, 1, 5, 6, 4], 2))
    print(merge_k_sorted([[1, 4, 5], [1, 3, 4], [2, 6]]))

The complete output is below. Dictionary payloads are never compared with one another:

text
('alert', {'node': 7}, 1)
('backup', {'node': 3}, 2)
[6, 5]
[1, 1, 2, 3, 4, 4, 5, 6]

An update leaves the old entry in the heap array; cancellation clears its payload reference. len(queue) counts active tasks only. compact() rebuilds the heap from active entries and preserves their sequence numbers, so tie order stays the same. When to call it depends on memory and latency targets; this example has no automatic threshold.

Let the heap contain m entries, of which a are active. Assuming average dictionary-access costs, marking an entry removed takes O(1). Adding one requires O(log(m+1)) heap adjustment. A pop that skips s stale entries can require O((s+1) log(m+1)) heap adjustment. Underlying list and dictionary resizes must also be amortized over the operation sequence. A single pop can clear many stale entries even with just one active task remaining. Building the active heap costs O(a), while releasing the old list processes m references: total compaction cost is O(m+a). Storage falls from O(m+a) to O(a), excluding payload objects.

3. Top K: Keep a Cutoff for the Candidates ​

As scores arrive, retain only the largest k. A min-heap of at most k elements keeps the current cutoff at its root:

  1. If the heap is not full, simply insert the new value;
  2. If the new value is less than or equal to the root, it cannot enter the top k;
  3. If the new value exceeds the root, replace the current smallest candidate with the new value.

k must support integer indexing; a floating-point value is rejected before input is read. If k <= 0, the function returns [] without consuming the input. If k >= n, it returns all values in descending order, preserving duplicate counts. For k > 0, let q=min(k,n). Scanning and final sorting together take O(n log(q+1)); the auxiliary heap and result each use O(q) space. With k=1, scanning still takes linear time.

Elements must have a consistent ordering, excluding NaN, and comparison costs are treated as constant. Top K preserves values and duplicate counts, but does not promise the original order among equal objects.

We use heapreplace because the code has already proven that the new value is greater than the heap root. It first removes the root from the heap and then inserts the new value. heappushpop(heap, value) instead compares the new value with the current root and returns the smaller of the two, keeping the larger one in the heap. In fixed-size filtering, both approaches may be useful, but their semantics cannot be inferred from their names alone.

4. K-Way Merge: One Current Item per Input ​

Start with input lists already sorted in ascending order. Insert the first item of each nonempty list into the heap. After removing an item, add only the next item from the same list. The function leaves inputs unchanged and assumes they are sorted; it does not validate that precondition.

Let there be K lists, N elements, and r nonempty lists. Scanning list heads takes O(K), repeated insertion into the initial heap takes O(r log(r+1)), and processing the elements takes O(N log(r+1)). Since r <= N, the total is O(K + N log(r+1)), also covering one list or many empty lists. The auxiliary heap uses O(r) space; the returned list adds Θ(N) output space.

Elements need consistent ordering and equality. Equal values from a lower group_index leave first, and each list retains its own order. These are the tuple's tie-breaking rules. To produce results lazily, use a generator or heapq.merge.

5. Three different directions of the standard library ​

EnvironmentDefault Root / HeaderImportant Contracts
Python heapqminimummaintains a heap on a regular list; heapify is linear time
Java PriorityQueueminimum value under comparatorany arbitrary head with equal values; iterator does not guarantee order; prohibits null; unsynchronized
C++ std::priority_queue<T>Default maximumContainer adapter; using std::greater<T> to get a min priority queue

Python 3.14 introduces heapify_max, heappush_max, heappop_max, and other public max-heap APIs. When maintaining compatibility with earlier versions, negate numeric values or wrap comparison keys. Python integer negation does not overflow a fixed width; in other languages, check the boundary when negating the minimum integer. heapq is merely an algorithms module for lists; for thread-blocking task queues, refer to queue.PriorityQueue.

Java's offer and poll are O(log n), and peek is O(1). Heads with the same minimum key are arbitrary and PriorityQueue cannot be treated as a stable queue. The internal ordering shown by iterator() and toString() is not the sorted result; repeated poll is needed when ordered extraction is required, but this will empty the queue. Thread-safe modifications require appropriate synchronization or PriorityBlockingQueue.

The C++ default comparator std::less<T> causes the maximum to appear at top(). The Top K maximum problem requires maintaining a minimum heap:

cpp
#include <functional>
#include <queue>
#include <vector>

std::priority_queue<int, std::vector<int>, std::greater<int>> candidates;

The standard adapter provides push, pop, and top, without support for ordered iteration or arbitrary element removal. When priority updates are needed, a common approach is to add new entries and skip old versions during popping, or alternatively, to choose a data structure that supports handles.

6. Extended Mode: Maintaining the Median in Two Heaps ​

The median in a data stream is maintained by placing the smaller half in a max heap and the larger half in a min heap, ensuring the sizes of the two heaps differ by at most one. For an odd count, the median is the root of the larger heap. For an even count of numeric values, it is usually the mean of the two roots. Empty input has no median yet. This is another example of "maintaining only boundaries, not full order." Begin by mastering the single-heap implementation before extending it to this more complex scenario.

7. Troubleshooting ​

  • Occasional same-priority events TypeError: The tuple lacks a unique sequence number, causing comparisons to fall on the payload.
  • Items still dequeued after cancellation: pop does not skip over deletion markers during cycling.
  • Number of active tasks increases after update: The old entry with the same name is not removed prior to replacement.
  • Java print output appears out of order: The internal array traversal lacks a guaranteed sort order.
  • C++ Top K algorithm eliminates the maximum value: A default max-heap priority queue was incorrectly used.
  • Memory continues to grow: Lazy deletion leaves too many stale entries; a rebuild strategy is required.

Official Contracts ​

From Task Order to Graph Traversal ​

The engineering challenges of priority queues often don’t lie in sift-up or sift-down, but in business logic: does a smaller number mean higher urgency? Should equal-priority items follow first-come, first-served? Are updates and cancellations permitted? And how should an empty queue report its state? First, define the interface contract clearly, then let the heap execute it. If lazy deletion is used, monitor stale entries and schedule rebuilds accordingly.

The next chapter moves into Graphs, where paths are no longer limited to parent-child relationships, and pending nodes expand from a single queue into a full network.

Built with VitePress | Software Systems Atlas

Copyright 2026 Steven Yang