Circular Queue Implementation

A regular array-based queue has a problem: after many enqueue/dequeue operations, the front pointer moves forward, leaving wasted space at the beginning. A circular queue (ring buffer) solves this by wrapping around to reuse that space.

The Problem with Linear Queues

In a linear array-based queue:

# After enqueue(A), enqueue(B), enqueue(C):
# [A, B, C, _, _]  front=0, rear=2

# After dequeue(), dequeue():
# [_, _, C, _, _]  front=2, rear=2

# Slots 0 and 1 are wasted!
# Even though the queue has room, we can't use those slots
# without shifting everything left — which is O(n)

A circular queue uses modulo arithmetic to wrap the pointers around:

After the rear reaches the end, it wraps: rear = (rear + 1) % size
This reuses the empty slots at the beginning!

Array-Based Circular Queue

The core idea

• front: index of the first element
• rear: index where the next element will be inserted
• count: number of elements currently in the queue
• All index math uses % capacity to wrap around

class CircularQueue:
    """
    Fixed-size circular queue (ring buffer) using an array.
    All operations are O(1).
    """

    def __init__(self, capacity):
        self.capacity = capacity
        self.queue = [None] * capacity
        self.front = 0
        self.rear = 0
        self.count = 0

    def is_empty(self):
        """Check if queue is empty. O(1)."""
        return self.count == 0

    def is_full(self):
        """Check if queue is full. O(1)."""
        return self.count == self.capacity

    def size(self):
        """Return number of elements. O(1)."""
        return self.count

    def enqueue(self, item):
        """
        Add item to the rear of the queue. O(1).
        Raises OverflowError if full.
        """
        if self.is_full():
            raise OverflowError("Queue is full")

        self.queue[self.rear] = item
        self.rear = (self.rear + 1) % self.capacity  # Wrap around!
        self.count += 1

    def dequeue(self):
        """
        Remove and return item from the front. O(1).
        Raises IndexError if empty.
        """
        if self.is_empty():
            raise IndexError("Queue is empty")

        item = self.queue[self.front]
        self.queue[self.front] = None  # Help garbage collection
        self.front = (self.front + 1) % self.capacity  # Wrap around!
        self.count -= 1
        return item

    def peek(self):
        """Return front item without removing. O(1)."""
        if self.is_empty():
            raise IndexError("Queue is empty")
        return self.queue[self.front]

    def __repr__(self):
        if self.is_empty():
            return "CircularQueue([])"
        items = []
        idx = self.front
        for _ in range(self.count):
            items.append(str(self.queue[idx]))
            idx = (idx + 1) % self.capacity
        return f"CircularQueue([{', '.join(items)}])"

Testing the implementation

cq = CircularQueue(5)

# Fill the queue
cq.enqueue('A')
cq.enqueue('B')
cq.enqueue('C')
cq.enqueue('D')
cq.enqueue('E')
print(cq)  # CircularQueue([A, B, C, D, E])
print(f"Full: {cq.is_full()}")  # True

# Dequeue some items
print(cq.dequeue())  # A
print(cq.dequeue())  # B
print(cq)  # CircularQueue([C, D, E])

# Now enqueue wraps around to reuse slots 0 and 1
cq.enqueue('F')
cq.enqueue('G')
print(cq)  # CircularQueue([C, D, E, F, G])

# Internal array looks like: [F, G, C, D, E]
# front=2, rear=2 (wrapped!), count=5
print(f"Internal: {cq.queue}")
print(f"front={cq.front}, rear={cq.rear}, count={cq.count}")

Trace: How wrapping works

Step           | Array          | front | rear | count
---------------|----------------|-------|------|------
Initial        | [_, _, _, _, _] | 0     | 0    | 0
enqueue(A)     | [A, _, _, _, _] | 0     | 1    | 1
enqueue(B)     | [A, B, _, _, _] | 0     | 2    | 2
enqueue(C)     | [A, B, C, _, _] | 0     | 3    | 3
dequeue() → A  | [_, B, C, _, _] | 1     | 3    | 2
dequeue() → B  | [_, _, C, _, _] | 2     | 3    | 1
enqueue(D)     | [_, _, C, D, _] | 2     | 4    | 2
enqueue(E)     | [_, _, C, D, E] | 2     | 0    | 3  ← rear wraps!
enqueue(F)     | [F, _, C, D, E] | 2     | 1    | 4  ← uses slot 0

Full vs Empty Detection

The classic challenge with circular queues: when front == rear, is the queue full or empty? There are two approaches:

Approach 1: Count variable (recommended)

Keep a count of elements. This is what our implementation uses.

def is_empty(self):
    return self.count == 0

def is_full(self):
    return self.count == self.capacity

Pros: Simple, unambiguous. Cons: Extra variable to maintain.

Approach 2: Waste one slot

Use only capacity - 1 slots. The queue is full when (rear + 1) % capacity == front:

class CircularQueueGapBased:
    def __init__(self, capacity):
        # Allocate one extra slot
        self.capacity = capacity + 1
        self.queue = [None] * self.capacity
        self.front = 0
        self.rear = 0

    def is_empty(self):
        return self.front == self.rear

    def is_full(self):
        return (self.rear + 1) % self.capacity == self.front

    def enqueue(self, item):
        if self.is_full():
            raise OverflowError("Queue is full")
        self.queue[self.rear] = item
        self.rear = (self.rear + 1) % self.capacity

    def dequeue(self):
        if self.is_empty():
            raise IndexError("Queue is empty")
        item = self.queue[self.front]
        self.front = (self.front + 1) % self.capacity
        return item

Pros: No count variable needed. Cons: Wastes one slot, capacity+1 array for capacity items.

Dynamic Circular Queue (Auto-Resize)

A fixed-size queue is limiting. Let’s make one that doubles when full:

class DynamicCircularQueue:
    """Circular queue that doubles in size when full."""

    def __init__(self, initial_capacity=8):
        self.capacity = initial_capacity
        self.queue = [None] * self.capacity
        self.front = 0
        self.rear = 0
        self.count = 0

    def _resize(self, new_capacity):
        """Copy elements to a new array in order. O(n)."""
        new_queue = [None] * new_capacity
        idx = self.front
        for i in range(self.count):
            new_queue[i] = self.queue[idx]
            idx = (idx + 1) % self.capacity

        self.queue = new_queue
        self.front = 0
        self.rear = self.count
        self.capacity = new_capacity

    def enqueue(self, item):
        """Add item, resizing if necessary. Amortized O(1)."""
        if self.count == self.capacity:
            self._resize(self.capacity * 2)

        self.queue[self.rear] = item
        self.rear = (self.rear + 1) % self.capacity
        self.count += 1

    def dequeue(self):
        """Remove front item. Optionally shrink. O(1)."""
        if self.count == 0:
            raise IndexError("Queue is empty")

        item = self.queue[self.front]
        self.queue[self.front] = None
        self.front = (self.front + 1) % self.capacity
        self.count -= 1

        # Shrink when 1/4 full to save memory
        if self.count > 0 and self.count <= self.capacity // 4:
            self._resize(self.capacity // 2)

        return item

    def peek(self):
        if self.count == 0:
            raise IndexError("Queue is empty")
        return self.queue[self.front]

    def __len__(self):
        return self.count

    def __repr__(self):
        items = []
        idx = self.front
        for _ in range(self.count):
            items.append(str(self.queue[idx]))
            idx = (idx + 1) % self.capacity
        return f"DynCQ([{', '.join(items)}], cap={self.capacity})"

dq = DynamicCircularQueue(4)
for i in range(10):
    dq.enqueue(i)
    print(f"  After enqueue({i}): {dq}")

# After enqueue(0): DynCQ([0], cap=4)
# After enqueue(3): DynCQ([0, 1, 2, 3], cap=4)
# After enqueue(4): DynCQ([0, 1, 2, 3, 4], cap=8)  ← resized!

LeetCode Design Problem

LeetCode 622 — Design Circular Queue:

class MyCircularQueue:
    """LeetCode-ready implementation."""

    def __init__(self, k: int):
        self.queue = [0] * k
        self.capacity = k
        self.count = 0
        self.front = 0

    def enQueue(self, value: int) -> bool:
        if self.isFull():
            return False
        rear = (self.front + self.count) % self.capacity
        self.queue[rear] = value
        self.count += 1
        return True

    def deQueue(self) -> bool:
        if self.isEmpty():
            return False
        self.front = (self.front + 1) % self.capacity
        self.count -= 1
        return True

    def Front(self) -> int:
        if self.isEmpty():
            return -1
        return self.queue[self.front]

    def Rear(self) -> int:
        if self.isEmpty():
            return -1
        rear = (self.front + self.count - 1) % self.capacity
        return self.queue[rear]

    def isEmpty(self) -> bool:
        return self.count == 0

    def isFull(self) -> bool:
        return self.count == self.capacity

Circular Deque

A circular deque supports insertion and removal at both ends:

class CircularDeque:
    """Double-ended circular queue."""

    def __init__(self, capacity):
        self.capacity = capacity
        self.queue = [None] * capacity
        self.front = 0
        self.rear = 0
        self.count = 0

    def insert_front(self, item):
        if self.count == self.capacity:
            raise OverflowError("Deque is full")
        self.front = (self.front - 1) % self.capacity  # Move front backward
        self.queue[self.front] = item
        self.count += 1

    def insert_rear(self, item):
        if self.count == self.capacity:
            raise OverflowError("Deque is full")
        self.queue[self.rear] = item
        self.rear = (self.rear + 1) % self.capacity
        self.count += 1

    def delete_front(self):
        if self.count == 0:
            raise IndexError("Deque is empty")
        item = self.queue[self.front]
        self.front = (self.front + 1) % self.capacity
        self.count -= 1
        return item

    def delete_rear(self):
        if self.count == 0:
            raise IndexError("Deque is empty")
        self.rear = (self.rear - 1) % self.capacity
        item = self.queue[self.rear]
        self.count -= 1
        return item

    def get_front(self):
        if self.count == 0:
            return -1
        return self.queue[self.front]

    def get_rear(self):
        if self.count == 0:
            return -1
        return self.queue[(self.rear - 1) % self.capacity]

Real-World Use Cases

1. OS Process Scheduling (Round Robin)

class RoundRobinScheduler:
    """Simple round-robin CPU scheduler using circular queue."""

    def __init__(self, time_quantum):
        self.queue = CircularQueue(100)
        self.time_quantum = time_quantum

    def add_process(self, process_id, burst_time):
        self.queue.enqueue({'pid': process_id, 'remaining': burst_time})

    def run(self):
        """Simulate round-robin scheduling."""
        time = 0
        while not self.queue.is_empty():
            process = self.queue.dequeue()
            run_time = min(self.time_quantum, process['remaining'])
            time += run_time
            process['remaining'] -= run_time

            if process['remaining'] > 0:
                # Not finished — put back in queue
                self.queue.enqueue(process)
                print(f"t={time}: P{process['pid']} ran {run_time}ms, "
                      f"{process['remaining']}ms remaining")
            else:
                print(f"t={time}: P{process['pid']} completed")


scheduler = RoundRobinScheduler(time_quantum=4)
scheduler.add_process(1, 10)
scheduler.add_process(2, 5)
scheduler.add_process(3, 8)
scheduler.run()

2. Producer-Consumer Buffer

import threading

class BoundedBuffer:
    """Thread-safe bounded buffer using circular queue."""

    def __init__(self, capacity):
        self.buffer = CircularQueue(capacity)
        self.lock = threading.Lock()
        self.not_full = threading.Condition(self.lock)
        self.not_empty = threading.Condition(self.lock)

    def produce(self, item):
        with self.not_full:
            while self.buffer.is_full():
                self.not_full.wait()
            self.buffer.enqueue(item)
            self.not_empty.notify()

    def consume(self):
        with self.not_empty:
            while self.buffer.is_empty():
                self.not_empty.wait()
            item = self.buffer.dequeue()
            self.not_full.notify()
            return item

Circular Queue vs Regular Queue vs Deque

Feature	Regular Queue	Circular Queue	Python deque
Implementation	Array or LL	Fixed-size array	Doubly linked blocks
Space waste	Front slots wasted	None	None
Enqueue	O(1) amortized	O(1)	O(1)
Dequeue	O(1) or O(n)	O(1)	O(1)
Random access	O(1)	O(1)	O(n)
Size	Dynamic	Fixed	Dynamic
Use case	General	Buffers, embedded	General

Complexity Summary

Operation	Time	Space
enqueue	O(1)	O(1)
dequeue	O(1)	O(1)
peek	O(1)	O(1)
is_empty / is_full	O(1)	O(1)
Space (total)	-	O(n)
Resize (dynamic)	O(n)	O(n)

Practice Problems

1. LeetCode 622 — Design Circular Queue: Core implementation (Medium)
2. LeetCode 641 — Design Circular Deque: Double-ended variant (Medium)
3. LeetCode 346 — Moving Average from Data Stream: Sliding window with circular buffer (Easy)
4. LeetCode 933 — Number of Recent Calls: Queue-based counting (Easy)
5. Implement a circular buffer for a producer-consumer system (System Design)

Key Takeaways

• Circular queues use modulo arithmetic ((index + 1) % capacity) to wrap pointers around, eliminating wasted space.
• The count-based approach for full/empty detection is simpler and more intuitive than wasting a slot.
• Circular queues are ideal for fixed-size buffers in OS scheduling, I/O buffering, and streaming applications.
• For dynamic sizing, use a doubling strategy with O(n) resize — amortized O(1) per operation.
• Python’s collections.deque is a better general-purpose choice, but understanding circular queues is essential for interviews and systems programming.