04 · Unfinished work: stacks, queues and heaps

A collection of unfinished work needs a rule for choosing what happens next. A stack removes the most recently added item, a queue removes the earliest, and a priority queue removes according to a key. The storage mechanism and the scheduling policy are related but not interchangeable. A heap cannot make a dependency-ready job ready; it can only choose among jobs already admitted.

Start with the removal contract

Use a Python list's append and end pop for a stack. Use collections.deque for queue operations at both ends, not repeated list.pop(0): the latter shifts later elements. The Python data-structures tutorial explains that distinction. The deque reference documents approximately O(1) append/pop operations at either end; random access near the middle is not its strength.

from collections import deque
 
stack = []
queue = deque()
for job in ("a", "b", "c"):
    stack.append(job)
    queue.append(job)
assert [stack.pop() for _ in range(3)] == ["c", "b", "a"]
assert [queue.popleft() for _ in range(3)] == ["a", "b", "c"]

This is not a benchmark. It checks order. It also does not establish a concurrent worker protocol: individually supported operations do not make a multi-operation “check then remove” sequence atomic.

A monotonic stack stores unanswered questions

For each duration, find the index of the first later duration that is strictly greater. A direct solution scans forward separately from every index, potentially O(n²). Instead, keep the indices whose questions remain unanswered. Their values are nonincreasing from stack bottom to top.

When a new value arrives, it answers every smaller pending value at the top. Those indices have not seen a greater value earlier—otherwise they would already have been removed—so this is the first greater position. Stop at an equal or larger value because the new value cannot answer it. Store indices rather than values so duplicate durations retain separate identities.

from itertools import product
 
 
def next_greater_index(values):
    answer = [-1] * len(values)
    pending = []
    for i, value in enumerate(values):
        while pending and values[pending[-1]] < value:
            answer[pending.pop()] = i
        pending.append(i)
    return answer
 
 
def brute_next(values):
    return [
        next((j for j in range(i + 1, len(values)) if values[j] > value), -1)
        for i, value in enumerate(values)
    ]
 
assert next_greater_index([3, 1, 1, 4, 2]) == [3, 3, 3, -1, -1]
assert next_greater_index([2, 2]) == [-1, -1]
for n in range(7):
    for values in product(range(3), repeat=n):
        assert next_greater_index(values) == brute_next(values)

Before the four arrives, indices 0, 1, 2 are pending with values 3, 1, 1. Four removes them in reverse insertion order and answers each with index three. The value two remains unanswered at the end. Every index is pushed once and popped at most once, so the nested loops perform O(n) pushes/pops overall; output and pending storage are O(n). This is an amortized accounting argument, not a claim that every single arrival costs constant time.

Changing < to <= changes the question to “next greater or equal.” A monotonic stack is not automatically appropriate for any problem containing the words “next” or “greater”; prove what removed entries can no longer contribute.

A heap is partially ordered, not a sorted list

The official heapq reference defines the min-heap invariant: a parent is no greater than its children, and heap[0] is the smallest item. It specifies linear-time heapify and distinguishes heapreplace from heappushpop. Iterating the underlying list does not generally yield sorted order.

To retain the largest k durations seen so far, keep a min-heap of at most k items. Its smallest item is the weakest candidate among the retained large values. A larger arrival replaces it; a smaller or equal arrival cannot improve the retained multiset. Until the heap fills, admit every value.

import heapq
from itertools import product
 
 
def largest_k(values, k):
    if type(k) is not int or k < 0:
        raise ValueError("k must be a nonnegative integer")
    if k == 0:
        return []
    heap = []
    for x in values:
        if len(heap) < k:
            heapq.heappush(heap, x)
        elif x > heap[0]:
            heapq.heapreplace(heap, x)
    return sorted(heap, reverse=True)
 
assert largest_k([5, 1, 5, 2], 2) == [5, 5]
assert largest_k([5, 1], 8) == [5, 1]
for n in range(6):
    for values in product(range(3), repeat=n):
        for k in range(5):
            assert largest_k(values, k) == sorted(values, reverse=True)[:k]

If m = min(n, k) and k is positive, ingestion costs O(n log(m+1)) as an upper bound, the final sort costs O(m log(m+1)), and storage is O(m). For k = 0 this API immediately returns without consuming the iterable. If k is close to n and all records are already in memory, sorting once may be simpler and faster in practice. “Top k always means heap” is a cue, not a theorem.

For task objects with equal priorities, pairs (priority, task) may fail if task objects are not orderable. The documented pattern is (priority, unique_sequence, task), where the sequence breaks ties before Python compares the payload. Updating a priority in place can break heap order; use a deliberate replacement or lazy-invalidating design rather than mutating a nested field and hoping the heap notices.

Exercises

1. Give an input where a single next-greater arrival pops every existing pending index. Why is total work still linear?
2. Would a max-heap of the largest k values give direct access to the candidate you want to evict?
3. Define how two jobs with the same priority should be scheduled and choose an appropriate tuple key.

Exit artifact: state the removal rule for each structure and test duplicate values. No custom linked-list implementation or concurrent queue is certified by these examples.

Next: Graphs and dependency order.