11. Queues — FIFO Data Structure

What problem does this solve? You need to process items in the order they arrive — first come, first served. Think print queue, task scheduling, or BFS traversal.

1. The Queue ADT — First In, First Out (FIFO)

Mental Model

Interface

2. Array-Based Queue (Circular Buffer)

Mental Model

python
Run Inline
# runnable
class CircularQueue:
    """Queue using circular buffer (fixed capacity array)."""
    def __init__(self, capacity=10):
        self._data = [None] * capacity
        self._capacity = capacity
        self._front = 0      # Index of front element
        self._back = 0       # Index of next insertion point
        self._size = 0
    def __len__(self):
        return self._size
    def __repr__(self):
        if self.is_empty():
            return "Queue: []"
        items = []
        idx = self._front
        for _ in range(self._size):
            items.append(str(self._data[idx]))
            idx = (idx + 1) % self._capacity
        return "Queue: [" + ", ".join(items) + "] (front → " + items[0] + ")"
    def is_empty(self):
        return self._size == 0
    def is_full(self):
        return self._size == self._capacity
    def enqueue(self, item):
        """Add item to back. O(1)."""
        if self.is_full():
            self._resize(2 * self._capacity)
        self._data[self._back] = item
        self._back = (self._back + 1) % self._capacity
        self._size += 1
    def dequeue(self):
        """Remove and return front item. O(1)."""
        if self.is_empty():
            raise IndexError("Dequeue from empty queue")
        item = self._data[self._front]
        self._data[self._front] = None  # Help garbage collection
        self._front = (self._front + 1) % self._capacity
        self._size -= 1
        return item
    def peek(self):
        """Return front item without removing. O(1)."""
        if self.is_empty():
            raise IndexError("Peek from empty queue")
        return self._data[self._front]
    def _resize(self, new_capacity):
        """Resize the circular buffer."""
        old_data = self._data
        self._data = [None] * new_capacity
        idx = self._front
        for i in range(self._size):
            self._data[i] = old_data[idx]
            idx = (idx + 1) % self._capacity
        self._front = 0
        self._back = self._size
        self._capacity = new_capacity
# Test
q = CircularQueue(4)
q.enqueue(10)
q.enqueue(20)
q.enqueue(30)
print(q)                    # Queue: [10, 20, 30] (front → 10)
print(f"Dequeue: {q.dequeue()}")  # 10
print(f"Peek: {q.peek()}")       # 20
q.enqueue(40)
q.enqueue(50)
print(q)                    # Queue: [20, 30, 40, 50]

Trace

pseudo
enqueue(10):  front=0, back=1, size=1   data=[10, _, _, _]
enqueue(20):  front=0, back=2, size=2   data=[10, 20, _, _]
enqueue(30):  front=0, back=3, size=3   data=[10, 20, 30, _]
dequeue():    front=1, back=3, size=2   returns 10
enqueue(40):  front=1, back=0, size=3   data=[40, 20, 30, _]  ← wraps!

3. Linked List-Based Queue

How It Works

python
Run Inline
# runnable
class Node:
    def __init__(self, data, next_node=None):
        self.data = data
        self.next = next_node
class LinkedQueue:
    """Queue implementation using singly linked list."""
    def __init__(self):
        self._head = None
        self._tail = None
        self._size = 0
    def __len__(self):
        return self._size
    def __repr__(self):
        items = []
        curr = self._head
        while curr:
            items.append(str(curr.data))
            curr = curr.next
        return "Queue: [" + ", ".join(items) + "]"
    def is_empty(self):
        return self._size == 0
    def enqueue(self, item):
        """Add item to back. O(1)."""
        new_node = Node(item)
        if self.is_empty():
            self._head = new_node
        else:
            self._tail.next = new_node
        self._tail = new_node
        self._size += 1
    def dequeue(self):
        """Remove and return front item. O(1)."""
        if self.is_empty():
            raise IndexError("Dequeue from empty queue")
        data = self._head.data
        self._head = self._head.next
        self._size -= 1
        if self.is_empty():
            self._tail = None
        return data
    def peek(self):
        """Return front item without removing. O(1)."""
        if self.is_empty():
            raise IndexError("Peek from empty queue")
        return self._head.data
# Test
q = LinkedQueue()
q.enqueue(10)
q.enqueue(20)
q.enqueue(30)
print(q)                    # Queue: [10, 20, 30]
print(f"Dequeue: {q.dequeue()}")  # 10
print(q)                    # Queue: [20, 30]

4. Classic Queue Applications

Application 1: BFS (Breadth-First Search) — Preview

python
Run Inline
# runnable
from collections import deque
def bfs(graph, start):
    """BFS traversal using a queue."""
    visited = {start}
    queue = deque([start])
    order = []
    while queue:
        vertex = queue.popleft()
        order.append(vertex)
        for neighbor in graph[vertex]:
            if neighbor not in visited:
                visited.add(neighbor)
                queue.append(neighbor)
    return order
graph = {
    'A': ['B', 'C'],
    'B': ['D', 'E'],
    'C': ['F'],
    'D': [],
    'E': ['F'],
    'F': []
}
print(f"BFS order: {bfs(graph, 'A')}")  # ['A', 'B', 'C', 'D', 'E', 'F']

Application 2: Print Queue Simulation

python
Run Inline
# runnable
from collections import deque
import time
class PrintQueue:
    def __init__(self):
        self.queue = deque()
    def add_job(self, doc_name):
        self.queue.append(doc_name)
        print(f"Added: {doc_name}")
    def process_jobs(self):
        while self.queue:
            job = self.queue.popleft()
            print(f"Printing: {job}")
            time.sleep(0.5)  # Simulate printing time
        print("All jobs done!")
pq = PrintQueue()
pq.add_job("resume.pdf")
pq.add_job("letter.docx")
pq.add_job("photo.jpg")
pq.process_jobs()

Application 3: Task Scheduling (Round Robin)

python
Run Inline
# runnable
from collections import deque
def round_robin(tasks, time_quantum):
    """Simple round-robin scheduler simulation."""
    queue = deque([(name, burst) for name, burst in tasks])
    time = 0
    while queue:
        name, burst = queue.popleft()
        if burst <= time_quantum:
            time += burst
            print(f"[{time:3d}] {name} completed")
        else:
            time += time_quantum
            queue.append((name, burst - time_quantum))
            print(f"[{time:3d}] {name} paused (remaining: {burst - time_quantum})")
tasks = [("P1", 10), ("P2", 5), ("P3", 8)]
round_robin(tasks, 4)

Application 4: Sliding Window Maximum

python
Run Inline
# runnable
from collections import deque
def max_sliding_window(arr, k):
    """Find maximum in every sliding window of size k.
    Uses deque to maintain candidates in O(n) time.
    """
    dq = deque()  # Stores indices
    result = []
    for i in range(len(arr)):
        # Remove elements outside window
        if dq and dq[0] <= i - k:
            dq.popleft()
        # Remove elements smaller than current (they'll never be max)
        while dq and arr[dq[-1]] <= arr[i]:
            dq.pop()
        dq.append(i)
        # Record maximum when window is full
        if i >= k - 1:
            result.append(arr[dq[0]])
    return result
arr = [1, 3, -1, -3, 5, 3, 6, 7]
print(f"Sliding max (k=3): {max_sliding_window(arr, 3)}")
# [3, 3, 5, 5, 6, 7]

5. Deque — Double-Ended Queue

python
Run Inline
# runnable
from collections import deque
dq = deque()
dq.append('right')        # Add to right
dq.appendleft('left')     # Add to left
print(dq)                 # deque(['left', 'right'])
print(dq.pop())           # 'right' — remove from right
print(dq.popleft())       # 'left' — remove from left

6. Comparison: Queue Implementations

7. Common Bugs

python
Run Inline
# BUG 1: Using list as queue (pop(0) is O(n))
bad_queue = []
for i in range(10000):
    bad_queue.append(i)
for _ in range(10000):
    bad_queue.pop(0)  # O(n) each! Total: O(n²)
# FIX: Use collections.deque
# BUG 2: Not checking whether queue is full in circular buffer
def buggy_enqueue(self, item):
    self._data[self._back] = item
    self._back = (self._back + 1) % self._capacity
    self._size += 1
# Overwrites existing data if queue is full!
# BUG 3: Forgetting to wrap indices in circular buffer
def buggy_dequeue(self):
    item = self._data[self._front]
    self._front += 1  # Should be self._front = (self._front + 1) % self._capacity
    self._size -= 1
    return item
# Eventually goes out of bounds after front reaches capacity

Practice Questions

Answers

A1.

pseudo
Enqueue A: [A]
Dequeue A: visit A → enqueue B, C: [B, C]
Dequeue B: visit B → enqueue D, E: [C, D, E]
Dequeue C: visit C → enqueue F: [D, E, F]
Dequeue D: visit D: [E, F]
Dequeue E: visit E: [F]
Dequeue F: visit F: []

A2. See stack file — push: enqueue to q2, move q1 to q2, swap. Pop: dequeue from q1.

A3. Queue allows operations at one end (FIFO). Deque allows O(1) operations at BOTH ends (add/remove front and back).

A4. Use a deque with maxlen. On every access, remove and re-add the item to maintain recency order.

A5. Queues model real-world FIFO scenarios where order of arrival matters. The printer processes jobs in the order they were submitted.

A6.

python
Run Inline
class QueueWithStacks:
    def __init__(self):
        self.in_stack, self.out_stack = [], []
    def enqueue(self, x):
        self.in_stack.append(x)
    def dequeue(self):
        if not self.out_stack:
            while self.in_stack:
                self.out_stack.append(self.in_stack.pop())
        return self.out_stack.pop()

A7. Use a heap (priority queue) — see the heaps module for implementation.

A8.

pseudo
Start:        front=0, back=0, size=0 [_, _, _, _]
enqueue(1):   front=0, back=1, size=1 [1, _, _, _]
enqueue(2):   front=0, back=2, size=2 [1, 2, _, _]
dequeue():    front=1, back=2, size=1 [_, 2, _, _] → returns 1
enqueue(3):   front=1, back=3, size=2 [_, 2, 3, _]
enqueue(4):   front=1, back=0, size=3 [_, 2, 3, 4]
dequeue():    front=2, back=0, size=2 [_, _, 3, 4] → returns 2
enqueue(5):   front=2, back=1, size=3 [5, _, 3, 4]

Final: front=2 (value 3), back=1 (next insertion), data=[5, _, 3, 4]

A9. deque uses a doubly linked list of fixed-size blocks. pop(0) on a list requires shifting all elements — O(n). popleft() on deque is O(1).

A10.

python
Run Inline
def generate_binary(n):
    q = deque(["1"])
    result = []
    for _ in range(n):
        x = q.popleft()
        result.append(x)
        q.append(x + "0")
        q.append(x + "1")
    return result
print(generate_binary(5))  # ['1', '10', '11', '100', '101']