Priority Queues in Data Structures - CSU083 | Shoolini University

Priority Queues in Data Structures

2024, August 5

1. Prerequisites

Before understanding priority queues, you should be familiar with:

2. What is a Priority Queue?

A priority queue is a special type of queue where elements are dequeued based on their priority rather than the order they were enqueued.

3. Why Does This Algorithm Exist?

Priority queues exist because real-world systems often require prioritization rather than simple FIFO order. Key applications include:

4. When Should You Use It?

Use a priority queue when:

5. How Does It Compare to Alternatives?

Strengths:
Weaknesses:

6. Python Code Implementation


import heapq

class PriorityQueue:
    def __init__(self):
        self.queue = []

    def insert(self, element, priority):
        heapq.heappush(self.queue, (priority, element))

    def extract_min(self):
        if self.is_empty():
            return None
        return heapq.heappop(self.queue)[1]

    def peek(self):
        if self.is_empty():
            return None
        return self.queue[0][1]

    def is_empty(self):
        return len(self.queue) == 0

    def display(self):
        print("Priority Queue:", self.queue)

# Example usage
pq = PriorityQueue()
pq.insert("Task A", 3)
pq.insert("Task B", 1)
pq.insert("Task C", 2)

pq.display()  # Shows the internal heap structure

print("Extract Min:", pq.extract_min())  # Should return "Task B" (priority 1)
pq.display()

print("Extract Min:", pq.extract_min())  # Should return "Task C" (priority 2)
pq.display()

print("Extract Min:", pq.extract_min())  # Should return "Task A" (priority 3)
pq.display()

7. Dry Run - Step-by-Step Execution

Step 1: Insert "Task A" (Priority = 3)
Step 2: Insert "Task B" (Priority = 1)
Step 3: Insert "Task C" (Priority = 2)

8. Extracting Elements (Min Priority First)

Step 4: Extract Min
Step 5: Extract Min
Step 6: Extract Min

9. Final Output (Stepwise Execution)


Priority Queue: [(1, 'Task B'), (3, 'Task A'), (2, 'Task C')]
Extract Min: Task B
Priority Queue: [(2, 'Task C'), (3, 'Task A')]
Extract Min: Task C
Priority Queue: [(3, 'Task A')]
Extract Min: Task A
Priority Queue: []

10. Key Observations

11. Time Complexity Analysis

The time complexity of a priority queue depends on its underlying data structure. Typically, heaps (binary, Fibonacci) are used for efficient operations.

11.1 Worst-Case, Best-Case, and Average-Case Complexity
Operation Binary Heap Complexity Sorted Array Complexity Unsorted Array Complexity
Insertion O(log n) (heapify up) O(n) (maintain sorted order) O(1) (append at end)
Extract Min/Max O(log n) (heapify down) O(1) (direct access) O(n) (linear search for min/max)
Peek (Get Min/Max) O(1) (root access) O(1) (first element) O(n) (search for min/max)
Search for an element O(n) (linear search) O(log n) (binary search) O(n) (linear search)

12. Space Complexity Analysis

The space complexity of a priority queue is determined by the data structure used.

12.1 Space Consumption vs Input Size

For a priority queue storing n elements:

13. Trade-offs and Considerations

13.1 When to Use Heaps vs Other Implementations
13.2 Trade-offs Between Time and Space

14. Common Optimizations

Optimizing a priority queue involves improving insertion, deletion, and retrieval efficiency. Key optimizations include:

15. Different Versions of the Algorithm

15.1 Min Heap vs Max Heap
15.2 Binary Heap vs Fibonacci Heap
Feature Binary Heap Fibonacci Heap
Insertion O(log n) O(1)
Extract Min O(log n) O(log n)
Decrease Key O(log n) O(1)
Merge O(n) O(1)
15.3 Iterative vs Recursive Implementations
Iterative Insert (Binary Heap)

def insert(heap, element):
    heap.append(element)
    i = len(heap) - 1
    while i > 0 and heap[i] < heap[(i-1)//2]:  # Heapify up
        heap[i], heap[(i-1)//2] = heap[(i-1)//2], heap[i]
        i = (i-1)//2
Recursive Insert (Binary Heap)

def heapify_up(heap, i):
    if i == 0:
        return
    parent = (i - 1) // 2
    if heap[i] < heap[parent]:
        heap[i], heap[parent] = heap[parent], heap[i]
        heapify_up(heap, parent)
15.4 Comparison: Iterative vs Recursive

16. Selecting the Best Priority Queue Variant

17. Common Pitfalls and Edge Cases

Priority queues can fail or behave unexpectedly in specific scenarios. Understanding these cases is crucial for robust implementations.

17.1 Edge Cases
17.2 Failure Scenarios

18. Test Cases to Verify Correctness

Test cases ensure the priority queue functions correctly under all conditions.

18.1 Test Case: Basic Insert & Extract

def test_basic_operations():
    pq = PriorityQueue()
    pq.insert("Task A", 3)
    pq.insert("Task B", 1)
    pq.insert("Task C", 2)
    
    assert pq.extract_min() == "Task B"
    assert pq.extract_min() == "Task C"
    assert pq.extract_min() == "Task A"
18.2 Test Case: Handling an Empty Queue

def test_empty_queue():
    pq = PriorityQueue()
    assert pq.extract_min() is None
    assert pq.peek() is None
18.3 Test Case: Duplicate Priorities

def test_duplicate_priorities():
    pq = PriorityQueue()
    pq.insert("Task A", 2)
    pq.insert("Task B", 2)
    pq.insert("Task C", 2)
    
    assert pq.extract_min() in ["Task A", "Task B", "Task C"]
18.4 Test Case: Large Input Handling

def test_large_input():
    pq = PriorityQueue()
    for i in range(1000000):
        pq.insert(f"Task {i}", i)
    
    assert pq.extract_min() == "Task 0"  # Smallest priority should be first

19. Real-World Failure Scenarios

19.1 Handling Non-Comparable Elements

Some priority queue implementations allow objects as elements, but without a comparator, they fail.


class CustomTask:
    def __init__(self, name, priority):
        self.name = name
        self.priority = priority  # Missing comparison operator

pq = PriorityQueue()
pq.insert(CustomTask("Task X", 1), 1)  # May raise a TypeError

Fix: Implement a comparison method in the class.

19.2 Memory Management in Continuous Insertions

Repeated insertions without deletions can cause memory exhaustion.


def memory_leak_scenario():
    pq = PriorityQueue()
    for i in range(10**9):  # Excessive insertions
        pq.insert(f"Task {i}", i)

Fix: Limit queue size or implement periodic cleanup.

20. Best Practices for Failure Handling

21. Real-World Applications

Priority Queues are widely used in real-world systems where prioritization is crucial.

21.1 Operating Systems
21.2 Networking & Communication
21.3 AI & Machine Learning
21.4 Finance & Stock Markets

22. Open-Source Implementations

Many open-source libraries provide optimized implementations of Priority Queues:

23. Practical Project: Task Scheduler Using Priority Queue

A real-world application of a priority queue is a Task Scheduler, where higher-priority tasks are executed first.

23.1 Project Description

We will implement a task scheduler where tasks with higher priority are executed first.

23.2 Python Implementation

import heapq
import time

class TaskScheduler:
    def __init__(self):
        self.queue = []
    
    def add_task(self, task_name, priority, execution_time):
        heapq.heappush(self.queue, (priority, execution_time, task_name))
    
    def run_tasks(self):
        while self.queue:
            priority, execution_time, task_name = heapq.heappop(self.queue)
            print(f"Executing Task: {task_name} (Priority: {priority})")
            time.sleep(execution_time)  # Simulating task execution

# Example Usage
scheduler = TaskScheduler()
scheduler.add_task("Low Priority Task", 3, 2)
scheduler.add_task("High Priority Task", 1, 1)
scheduler.add_task("Medium Priority Task", 2, 1.5)

print("Starting Task Execution...")
scheduler.run_tasks()
23.3 Expected Output

Starting Task Execution...
Executing Task: High Priority Task (Priority: 1)
Executing Task: Medium Priority Task (Priority: 2)
Executing Task: Low Priority Task (Priority: 3)

24. Quick Recap

25. Assignments: Solve 10 Competitive Programming Problems

Practice priority queue implementation by solving the following problems on platforms like LeetCode, Codeforces, and HackerRank.

  1. Kth Largest Element in an Array (LeetCode 215) - Use a min-heap of size K.
  2. Top K Frequent Elements (LeetCode 347) - Use a max-heap or bucket sort.
  3. Find Median from Data Stream (LeetCode 295) - Use two heaps to maintain median dynamically.
  4. Merge K Sorted Lists (LeetCode 23) - Use a min-heap for efficient merging.
  5. Task Scheduler (LeetCode 621) - Use a max-heap to schedule tasks efficiently.
  6. Dijkstra’s Algorithm (Shortest Path) - Use a priority queue for efficient graph traversal.
  7. Prim’s Algorithm (Minimum Spanning Tree) - Use a min-heap to find the minimum-cost edges.
  8. Rearrange String K Distance Apart (LeetCode 358) - Use a max-heap to maintain order constraints.
  9. Minimum Cost to Connect Ropes (LeetCode 1167) - Use a min-heap for greedy merging.
  10. IPO (Investment in Projects) (LeetCode 502) - Use a priority queue to maximize capital.

26. System Design Problem: Job Scheduling System

Design a job scheduling system where jobs are executed based on priority.

26.1 Requirements
26.2 System Architecture
26.3 Python Implementation

import heapq
import time
from threading import Thread

class JobScheduler:
    def __init__(self):
        self.job_queue = []
    
    def add_job(self, job_id, priority, execution_time):
        heapq.heappush(self.job_queue, (priority, job_id, execution_time))
    
    def execute_jobs(self):
        while self.job_queue:
            priority, job_id, execution_time = heapq.heappop(self.job_queue)
            print(f"Executing Job {job_id} with Priority {priority}")
            time.sleep(execution_time)  # Simulating execution time

# Example usage
scheduler = JobScheduler()
scheduler.add_job("Job1", 2, 2)
scheduler.add_job("Job2", 1, 1)
scheduler.add_job("Job3", 3, 1.5)

print("Starting Job Execution...")
Thread(target=scheduler.execute_jobs).start()
26.4 Expected Output

Starting Job Execution...
Executing Job Job2 with Priority 1
Executing Job Job1 with Priority 2
Executing Job Job3 with Priority 3

27. Time-Constrained Implementation Practice

To simulate real-world coding interviews, implement a priority queue from scratch in 15 minutes without using built-in heap functions.

27.1 Challenge: Implement a Min-Heap Based Priority Queue

Implement a custom priority queue using an array-based binary heap.


class MinHeap:
    def __init__(self):
        self.heap = []

    def insert(self, element):
        self.heap.append(element)
        self._heapify_up(len(self.heap) - 1)

    def extract_min(self):
        if len(self.heap) == 0:
            return None
        if len(self.heap) == 1:
            return self.heap.pop()
        
        min_val = self.heap[0]
        self.heap[0] = self.heap.pop()
        self._heapify_down(0)
        return min_val

    def _heapify_up(self, index):
        parent = (index - 1) // 2
        while index > 0 and self.heap[index] < self.heap[parent]:
            self.heap[index], self.heap[parent] = self.heap[parent], self.heap[index]
            index = parent
            parent = (index - 1) // 2

    def _heapify_down(self, index):
        left = 2 * index + 1
        right = 2 * index + 2
        smallest = index
        
        if left < len(self.heap) and self.heap[left] < self.heap[smallest]:
            smallest = left
        if right < len(self.heap) and self.heap[right] < self.heap[smallest]:
            smallest = right
        
        if smallest != index:
            self.heap[index], self.heap[smallest] = self.heap[smallest], self.heap[index]
            self._heapify_down(smallest)

# Testing
pq = MinHeap()
pq.insert(3)
pq.insert(1)
pq.insert(2)
print(pq.extract_min())  # Output: 1
print(pq.extract_min())  # Output: 2
print(pq.extract_min())  # Output: 3

28. Quick Recap