Problem. This comparison models a single-threaded bounded stream that receives items one at a time, returns the smallest item when its window is full, and cannot wait until the end to sort everything. The baseline is a sorted list maintained with bisect.insort_right, but each insertion still has to place an item inside the list.[1]
Baseline. In the list implementation below, with a positive queue window w, inserting into the ordered list can move O(w) references and pop(0) moves the remaining references towards the front. Over E events, the shown code therefore has O(Ew) worst-case data movement, plus output construction shared by both functions. This is an analysis of this CPython list path, not a universal statement about every sorted collection.
Solution. The measured alternative is a min-heap using separate heappush() and heappop() calls. heapq keeps the smallest item at index zero, and the benchmark measures that exact operation path; fused functions such as heappushpop() and heapreplace() are deliberately outside this comparison.[2]
Measured result. On CPython 3.13.5 running on Linux aarch64, 100,000 events took a median of 0.094961 seconds with the sorted list and 0.153319 seconds with the separate-call heap at window 64. At window 1,024, the medians were 0.338253 and 0.254629 seconds, so the heap was 1.33 times faster in that row. The output retains all five samples and each implementation’s minimum.
Reproduce the result
Complete code
from __future__ import annotationsimport bisectimport heapqimport platformimport randomimport timeitfrom statistics import medianSEED = 20261005PRIORITY_RANGE = 1_000_000EVENT_COUNTS = (20_000, 50_000, 100_000)WINDOWS = (64, 1_024)REPEATS = 5NUMBER = 1def make_events(count: int) -> list[tuple[int, int]]: generator = random.Random(SEED) return [(generator.randrange(PRIORITY_RANGE), sequence) for sequence in range(count)]def sorted_list_queue(events: list[tuple[int, int]], window: int) -> list[tuple[int, int]]: queue: list[tuple[int, int]] = [] emitted: list[tuple[int, int]] = [] for item in events: bisect.insort_right(queue, item) if len(queue) > window: emitted.append(queue.pop(0)) emitted.extend(queue) return emitteddef heap_queue(events: list[tuple[int, int]], window: int) -> list[tuple[int, int]]: queue: list[tuple[int, int]] = [] emitted: list[tuple[int, int]] = [] for item in events: heapq.heappush(queue, item) if len(queue) > window: emitted.append(heapq.heappop(queue)) while queue: emitted.append(heapq.heappop(queue)) return emitteddef timed(function, events: list[tuple[int, int]], window: int) -> list[float]: timer = timeit.Timer(lambda: function(events, window)) return timer.repeat(repeat=REPEATS, number=NUMBER)def format_samples(samples: list[float]) -> str: return "|".join(f"{sample / NUMBER:.6f}" for sample in samples)def checksum(items: list[tuple[int, int]]) -> int: return sum((priority * 1_000_003 + sequence) for priority, sequence in items)def main() -> None: print(f"Python {platform.python_version()} on {platform.platform()}") print(f"seed={SEED}, priority_range={PRIORITY_RANGE}, repeats={REPEATS}, number={NUMBER}") print("timer=timeit.default_timer (perf_counter); setup=pre-generated events; gc=timeit_default_disabled") print("correctness_check=full_emission_sequence_equality_before_timing") print("events,window,baseline_samples_s,heap_samples_s,baseline_min_s,heap_min_s,baseline_median_s,heap_median_s,speedup,baseline_checksum,heap_checksum") for count in EVENT_COUNTS: events = make_events(count) for window in WINDOWS: baseline_result = sorted_list_queue(events, window) heap_result = heap_queue(events, window) if baseline_result != heap_result: raise AssertionError(f"emission mismatch for events={count}, window={window}") # A single untimed run warms both code paths before timing. sorted_list_queue(events, window) heap_queue(events, window) baseline_samples = timed(sorted_list_queue, events, window) heap_samples = timed(heap_queue, events, window) baseline_median = median(baseline_samples) / NUMBER heap_median = median(heap_samples) / NUMBER print( f"{count},{window},{format_samples(baseline_samples)}," f"{format_samples(heap_samples)},{min(baseline_samples):.6f}," f"{min(heap_samples):.6f},{baseline_median:.6f}," f"{heap_median:.6f},{baseline_median / heap_median:.2f}," f"{checksum(baseline_result)},{checksum(heap_result)}" )if __name__ == "__main__": main()
Command
python3 priority_queue_benchmark.py
Output
$ python3 priority_queue_benchmark.pyPython 3.13.5 on Linux-6.12.47+rpt-rpi-v8-aarch64-with-glibc2.41seed=20261005, priority_range=1000000, repeats=5, number=1timer=timeit.default_timer (perf_counter); setup=pre-generated events; gc=timeit_default_disabledcorrectness_check=full_emission_sequence_equality_before_timingevents,window,baseline_samples_s,heap_samples_s,baseline_min_s,heap_min_s,baseline_median_s,heap_median_s,speedup,baseline_checksum,heap_checksum20000,64,0.018916|0.018804|0.018853|0.019146|0.018616,0.029230|0.029281|0.029514|0.029879|0.029900,0.018616,0.029230,0.018853,0.029514,0.64,10016636048807397,1001663604880739720000,1024,0.069578|0.063810|0.066971|0.066302|0.064753,0.045922|0.046924|0.046665|0.049105|0.047687,0.063810,0.045922,0.066302,0.046924,1.41,10016636048807397,1001663604880739750000,64,0.048513|0.048371|0.048484|0.048726|0.048109,0.079739|0.077374|0.079360|0.077248|0.080088,0.048109,0.077248,0.048484,0.079360,0.61,25087836968256365,2508783696825636550000,1024,0.158761|0.163999|0.169078|0.161562|0.159615,0.116352|0.115092|0.115157|0.115553|0.116586,0.158761,0.115092,0.161562,0.115553,1.40,25087836968256365,25087836968256365100000,64,0.096114|0.094961|0.095764|0.094712|0.094128,0.153319|0.147629|0.154015|0.148722|0.159379,0.094128,0.147629,0.094961,0.153319,0.62,50101902161190568,50101902161190568100000,1024,0.325749|0.326223|0.347900|0.338253|0.341066,0.254629|0.255743|0.257060|0.253898|0.244638,0.325749,0.244638,0.338253,0.254629,1.33,50101902161190568,50101902161190568exit=0
Environment
The run used CPython 3.13.5 on Linux 6.12.47, aarch64, with only Python’s standard library. The source prints the runtime and platform as its first line. The benchmark does not install packages or read external data.
Methodology
make_events() creates (priority, sequence) tuples with random.Random(20261005). Priorities are drawn from 0 through 999999. The run tests 20,000, 50,000 and 100,000 events with queue windows of 64 and 1,024.
For each event, both implementations add one tuple. When the queue exceeds its window, they emit the smallest tuple. Each implementation then drains the remaining queue. The full emitted sequences are compared before timing, rather than checking only their final counts or checksums.
The event lists are generated before timing. Each case gets one untimed warm-up call per implementation, followed by five timeit.Timer.repeat() samples with one call per sample. The code reports every raw sample, its minimum and its median. The median gives a compact centre for the short comparison table; it is not a lower bound or a portable performance factor. The minimum is retained as the lower-bound reference recommended by the timeit documentation. timeit uses perf_counter() as its default timer and disables garbage collection during a timed call by default.[3]
The timed function includes queue allocation, tuple comparisons, output-list appends and the final drain. It excludes event generation. No RSS, allocation count or cache-miss measurement is included.
Limits
The input priorities are pseudo-random integers from a fixed seed, not a production trace. A different distribution, event order or queue-width pattern can change the constant factors and the point at which one representation becomes faster.
The timings describe CPython 3.13.5 on one aarch64 host. They do not establish behaviour for another interpreter, CPU, operating-system load or object type. The code uses two-element tuples; expensive comparison methods would change the balance.
The benchmark is single-threaded. The Python documentation warns that the bisect functions are not thread-safe, so concurrent producers or consumers need an external synchronisation policy.[1] The integer sequence field is also a deliberate tie-breaker. For arbitrary task objects with equal priorities, use a comparable entry counter or wrapper rather than assuming the payloads can be compared.[2]
The benchmark also measures wall-clock time rather than memory use. A separate memory experiment would need its own start and stop boundaries, allocator policy and result checks. No such claim is made here.
The comparison uses separate heappush() and heappop() calls. It does not measure heappushpop() or heapreplace(), whose semantics and constants differ. The timed function also accumulates all emitted values in a list, so a streaming consumer could have a different memory and timing profile.
To test a real workload, replace make_events() with a retained input trace, keep the full-sequence correctness check, set WINDOWS to the queue widths that matter, and rerun the same command.
The problem
A priority queue has a different shape from a batch sort. The next smallest item must be available while later items are still arriving. That rules out collecting the whole stream and calling sorted() once at the end.
A sorted list is an understandable first implementation. Keep the list ordered, insert the new tuple at its position, then remove the first tuple when the window is full. The list is always easy to inspect, and an insertion point can be found with binary search.
The catch is the physical move that follows the search. insort_right() finds the position and then calls list insertion; the Python documentation says that the logarithmic search is dominated by the linear insertion step.[1] Removing index zero also moves the remaining list entries towards the front in this exact implementation. The search is cheap. Keeping an array ordered is the expensive part.
Baseline complexity
Let E be the number of arriving events and w the queue window. The list version performs one ordered insertion and, after the warm-up phase, one front removal per event. The insertion is O(w) in the worst case, so the shown stream has O(Ew) worst-case data movement. The code also builds an output list, which adds work that is common to both implementations.
The tuple is (priority, sequence). The priority decides which item is smallest; the sequence number makes equal priorities deterministic. That tie-breaker matters because the correctness check compares the complete emission sequence.
Solution: keep the boundary, not the whole order
A min-heap stores only the ordering boundary needed for the next removal. Its root is the smallest item, while the rest of the array is arranged to satisfy the heap invariant rather than fully sorted order.[2]
heappush() adds the new tuple and repairs the path towards the root. heappop() removes the root, moves a remaining item into its place and repairs a path down the tree. The heap documentation describes that repair as logarithmic in the number of items.[2]
That is the change being measured. Both functions see the same tuples in the same order, emit the same sequence, and maintain a queue of the same maximum size. Only the representation and the repair work differ.
The comparison deliberately does not use heappushpop() or heapreplace(). Those fused operations are relevant alternatives for a bounded queue, but measuring them would be a third implementation and a different comparison. The results below apply only to separate push-then-pop calls.[2]
What the measurement shows
At 100,000 events and window 64, the sorted list’s median was 0.094961 seconds. The separate-call heap’s median was 0.153319 seconds, and the output reports a baseline-to-heap ratio of 0.62. The list was 1.61 times faster in this small-window case.
At the same event count and window 1,024, the sorted list took 0.338253 seconds at the median and the separate-call heap took 0.254629 seconds. The heap was 1.33 times faster. The direction is consistent across the 20,000 and 50,000 rows in this run: the list wins at window 64, while the separate-call heap wins at window 1,024.
The benchmark does not measure why the crossover occurs. It measures complete function time, including Python-level loop and output work, on one host. The result is evidence for this workload and configuration, not a portable ranking of all priority-queue implementations.
Choosing the structure
For the separate push-then-pop operation contract shown here, a heap avoids the list’s repeated full-width shifts. Its repair path grows logarithmically with the queue width, but this run does not locate a universal crossover.
If the queue stays small, the sorted list can be faster in practice and gives direct access to every item in order. It can also be the clearer choice when the workload needs insertion-point queries rather than repeated removal of the smallest item. The 64-item rows are a measured example of that trade-off, not a fixed threshold.
The important boundary is the operation contract. This programme only inserts, evicts the smallest item and drains the queue. It does not support arbitrary deletion or in-place priority changes. The heapq documentation treats those as separate priority-queue problems, with a lazy-removal pattern for one common approach.[2]
Sources
[1] Python 3.13 documentation: bisect — Array bisection algorithm
[2] Python 3.13 documentation: heapq — Heap queue algorithm
[3] Python 3.13 documentation: timeit — Measure execution time of small code snippets
