Priority Queue — Poisoned by Equal Priorities

How a Priority Queue Works — Plain English

A priority queue always dequeues the element with the highest (or lowest) priority, not the oldest one. It is implemented with a binary heap — a complete binary tree where every parent is smaller than its children (min-heap) or larger (max-heap).

Core operations: 1. insert(x, priority): add element. Heap 'bubbles up' to maintain heap property. O(log n). 2. extract_min() / extract_max(): remove and return the minimum/maximum. Heap 'sinks down'. O(log n). 3. peek(): return min/max without removing. O(1).

Python's heapq is a min-heap. For max-heap, negate values.

Worked example — min-heap insert then extract: Insert 5: heap=[5]. Insert 3: heap=[5,3]. Bubble up: 3 < parent 5. Swap. heap=[3,5]. Insert 7: heap=[3,5,7]. 7 >= parent 3 (index 0 parent of index 2). No swap. Insert 1: heap=[3,5,7,1]. Bubble up: 1 < parent 5. Swap: [3,1,7,5]. 1 < parent 3. Swap: [1,3,7,5]. extract_min(): remove 1. Move 5 (last) to root: [5,3,7]. Sink down: 5>min(3,7)=3. Swap: [3,5,7]. Extracted: 1.

The Binary Heap: Array-Based Tree Logic

Picture a heap as a very disciplined family tree where every parent is stricter (smaller or larger) than its kids, and the entire tree is always packed to the left. Because it's complete, we can throw away the pointers and just use a plain list — that's why heaps are so memory efficient and cache-friendly.

For any index i in the array: • Left child → 2i + 1 • Right child → 2i + 2 • Parent → (i - 1) // 2

The magic rule (Heap Invariant) is simple: no parent is ever “worse” than its children. In a min-heap the smallest guy always bubbles to index 0. Break this rule even once and the whole thing collapses — that's why we have heapify and sifting operations. I still draw this array-as-tree on a whiteboard every time I explain it to someone; it clicks instantly.

# Package: io.thecodeforge.algorithms.ds

# The heap [1, 3, 5, 7, 9, 8] represented as a tree:
#         1          (index 0)
#        / \
#       3   5        (index 1, 2)
#      / \ / \
#     7  9 8        (index 3, 4, 5)

def get_heap_relationships(heap):
    relationships = []
    for i in range(len(heap)):
        left_idx  = 2 * i + 1
        right_idx = 2 * i + 2
        parent_idx = (i - 1) // 2 if i > 0 else None
        
        relationships.append({
            "val": heap[i],
            "parent": heap[parent_idx] if parent_idx is not None else None,
            "left": heap[left_idx] if left_idx < len(heap) else None,
            "right": heap[right_idx] if right_idx < len(heap) else None
        })
    return relationships

# Standard Heap Layout
my_heap = [1, 3, 5, 7, 9, 8]
for node in get_heap_relationships(my_heap):
    print(f"Node: {node['val']} | Parent: {node['parent']} | L: {node['left']} | R: {node['right']}")

Production Usage: Python's heapq Module

Real talk — in 99% of production code and interviews you will never implement a heap from scratch. Python's heapq is battle-tested, written in C under the hood, and ridiculously fast. The golden pattern every senior engineer uses is storing tuples: (priority, counter, data). The counter prevents comparison crashes when priorities are equal (a bug I've debugged in live systems more than once).

You'll see this everywhere: background job queues, rate-limiters, merge k sorted streams, and even in custom schedulers.

import heapq

# 1. Initialize a Priority Queue for Task Management
tasks = []
# Format: (priority_level, unique_id, task_description)
# unique_id prevents comparison errors if priorities are equal
heapq.heappush(tasks, (3, 101, 'Cleanup logs'))
heapq.heappush(tasks, (1, 102, 'Fix production crash'))
heapq.heappush(tasks, (2, 103, 'Update documentation'))

# 2. Efficient Extraction
while tasks:
    priority, _id, task = heapq.heappop(tasks)
    print(f'Processing: {task} (Priority {priority})')

# 3. Heapify: O(n) construction from unsorted list
# Much faster than n-calls to heappush [O(n log n)]
raw_metrics = [45, 12, 98, 3, 7, 22]
heapq.heapify(raw_metrics)
print(f"Smallest metric: {raw_metrics[0]}") # 3

The Max-Heap Pattern & Top-K Problems

Python only gives you a min-heap, so we just flip the sign — a trick every LeetCode grinder and system designer knows by heart. This pattern shows up in "k closest points", "k largest elements in stream", "median in data stream", and even in designing Twitter's trending topics.

For huge streams where you only care about the top 10, a heap of size k eats way less memory than sorting millions of items. I still remember the first time I replaced a full sort with a heap in a production analytics pipeline — CPU dropped 70% overnight.

import heapq

class MaxForgeHeap:
    """Wrapper to facilitate Max-Heap behavior in Python"""
    def __init__(self, data=None):
        self.data = [-x for x in data] if data else []
        heapq.heapify(self.data)

    def push(self, val):
        heapq.heappush(self.data, -val)

    def pop(self):
        return -heapq.heappop(self.data)

# Practical: Top 3 high scores
scores = [1500, 3200, 1200, 5000, 2800]
top_k = heapq.nlargest(3, scores) # Optimized internal implementation
print(f"Top 3 Scores: {top_k}")

Priority Queue in System Design: Real-World Use Cases

Priority queues aren't just for LeetCode — they're the backbone of scheduling, network routing, and event-driven systems. In distributed systems, they appear in task queues (Celery, Redis sorted sets), in rate limiters (token bucket with priority), and in event processing (Kafka consumers processing based on event priority).

One key pattern: bounded priority queues. If you're processing a stream of events with priorities, you cannot let the queue grow unbounded — memory will kill you. Set a max size, and when full, either drop the lowest priority item or reject new low-priority items.

Another pattern: priority inversion. In real-time systems, a low-priority task holding a lock can block a high-priority task. The heap itself doesn't cause this, but the scheduling policy built on top must handle it (priority inheritance).

# Package: io.thecodeforge.production.scheduling

import heapq

class BoundedPriorityQueue:
    """Fixed-size priority queue with drop of lowest priority when full"""
    def __init__(self, max_size):
        self._queue = []
        self._max_size = max_size
        self._counter = 0

    def push(self, priority, item):
        if len(self._queue) >= self._max_size:
            # Drop lowest priority (highest numerical value since min-heap)
            heapq.heappushpop(self._queue, (priority, self._counter, item))
        else:
            heapq.heappush(self._queue, (priority, self._counter, item))
        self._counter += 1

    def pop(self):
        if not self._queue:
            return None
        _, _, item = heapq.heappop(self._queue)
        return item

    @property
    def size(self):
        return len(self._queue)

# Example: log processor with max 1000 pending
log_q = BoundedPriorityQueue(1000)
log_q.push(1, "CRITICAL: disk full")
log_q.push(5, "INFO: health check passed")
log_q.push(2, "ERROR: connection timeout")
print(log_q.pop())  # CRITICAL: disk full