Lab: Priority Queue
Duc-Minh VU
FDA - SLSCM Lab
National Economics University
[email protected]
February 12, 2025
Objective
The purpose of this lab is to understand the priority queue data structure,
implement it using Python, explore its common applications, and solve re-
lated exercises.
Definition
A priority queue is an abstract data structure where each element is asso-
ciated with a priority. The element with the highest priority is served before
other elements. It differs from a regular queue in that the order of service is
determined by priority rather than the order of arrival.
Types of Priority Queue
• Min-Priority Queue: The element with the smallest priority is served
first.
• Max-Priority Queue: The element with the largest priority is served
first.
Supported Operations in Priority Queue
A priority queue provides several key operations:
• is empty(): Check if the priority queue is empty.
1
• insert(data): Insert a new element into the priority queue.
• delete(): Remove and return the element with the highest priority.
In a min-priority queue, this would be the element with the smallest
priority value.
• peek(): Retrieve the element with the highest priority without remov-
ing it.
• size(): Return the number of elements in the priority queue.
• change priority(data, new priority): Change the priority of a spe-
cific element (optional but useful in advanced applications).
Implementation
Using List (Basic Implementation)
Listing 1: Priority Queue Implementation Using List
class PriorityQueue:
def __init__(self):
self.queue = []
def is_empty(self):
return len(self.queue) == 0
def insert(self, data):
"""Insert an element into the queue."""
self.queue.append(data)
def delete(self):
"""Delete and return the element with the highest priority.
,→ """
if not self.is_empty():
max_val = 0
for i in range(len(self.queue)):
if self.queue[i] > self.queue[max_val]:
max_val = i
return self.queue.pop(max_val)
return "Queue is empty!"
def peek(self):
2
"""Return the element with the highest priority without
,→ removing it."""
if not self.is_empty():
max_val = 0
for i in range(len(self.queue)):
if self.queue[i] > self.queue[max_val]:
max_val = i
return self.queue[max_val]
return "Queue is empty!"
# Example usage:
pq = PriorityQueue()
pq.insert(3)
pq.insert(5)
pq.insert(2)
pq.insert(4)
print(pq.peek()) # Output: 5 (peek without removing it)
print(pq.delete()) # Output: 5 (remove the highest-priority element
,→ )
print(pq.delete()) # Output: 4
Time Complexity Analysis
The time complexity of the operations in the basic list-based implementation
is as follows:
• Insertion (insert()): O(1) — Adding an element to the end of the
list takes constant time.
• Deletion (delete()): O(n) — The method iterates through the list
to find the element with the highest priority, which takes linear time.
• Checking if the queue is empty (is empty()): O(1) — Checking
the length of the list is a constant time operation.
Using List with Sorted Elements (Ascending Order)
This implementation maintains the list in sorted order (ascending). Insertion
takes O(n) time, while deletion (removing the element with the smallest
priority) takes O(1).
Listing 2: Priority Queue with Sorted List (Ascending Order)
3
class PriorityQueueSorted:
def __init__(self):
self.queue = []
def is_empty(self):
return len(self.queue) == 0
def insert(self, data):
self.queue.append(data)
self.queue.sort() # Maintain the list in ascending order
def delete(self):
if not self.is_empty():
return self.queue.pop(0) # Remove the element with the
,→ smallest priority
return "Queue is empty!"
def peek(self):
if not self.is_empty():
return self.queue[0]
return "Queue is empty!"
# Example usage:
pq = PriorityQueueSorted()
pq.insert(5)
pq.insert(3)
pq.insert(4)
pq.insert(1)
print(pq.delete()) # Output: 1
print(pq.delete()) # Output: 3
Time Complexity Analysis
• Insertion: O(n) — Requires sorting the list after each insertion.
• Deletion (removing the smallest element): O(1).
• Peek (getting the smallest element without removal): O(1).
Using heapq Library
4
Listing 3: Priority Queue Implementation Using heapq
import heapq
queue = []
heapq.heappush(queue, (1, "Task 1"))
heapq.heappush(queue, (3, "Task 3"))
heapq.heappush(queue, (2, "Task 2"))
while queue:
priority, task = heapq.heappop(queue)
print(f"Processing {task} with priority {priority}")
Applications
Task Scheduling
Problem: Schedule tasks based on their priority.
Solution: Use a priority queue to process tasks in the order of their priority.
Listing 4: Task Scheduling Using heapq
import heapq
def task_scheduler(tasks):
queue = []
for priority, task in tasks:
heapq.heappush(queue, (priority, task))
while queue:
priority, task = heapq.heappop(queue)
print(f"Processing {task} with priority {priority}")
# Example usage:
tasks = [(2, "Task A"), (1, "Task B"), (3, "Task C")]
task_scheduler(tasks)
Job Scheduling with Deadline and Profit
Problem: Schedule jobs to maximize profit.
Objective: Write a program to simulate a job scheduling system using a
priority queue. The program should:
5
• Accept a list of jobs with deadlines and profits.
• Schedule jobs to maximize profit without exceeding deadlines.
Exercises
1. Task Scheduling: Implement a task scheduler using a priority queue.
2. Dijkstra’s Algorithm: Implement Dijkstra’s algorithm for shortest
path.
3. Job Scheduling: Solve the job scheduling problem using a priority
queue.
4. Huffman Coding: Implement Huffman Coding to compress data.
5. Patient Triage Simulation: Simulate a hospital triage system where
patients are treated based on priority.
Conclusion
In this lab, you learned the fundamentals of priority queues and implemented
them in Python using both basic and optimized approaches. Understand-
ing priority queues is crucial for solving real-world problems like scheduling,
pathfinding, and task management.
References
• Introduction to Priority Queue
• Priority Queue in Python - GeeksforGeeks
• Applications of Priority Queue
• Coding Problems
