/ Data structures - Circular Queue
Data structures - Circular Queue¶
A circular queue (also called a ring buffer) is a fixed-capacity queue where the tail wraps around to the front of the underlying array once it reaches the end. This avoids wasted space and eliminates the need to shift elements on dequeue.
This notebook covers:
- What is a Circular Queue?
- The
CircularQueueclass - Core operations:
enqueue,dequeue,peek,is_empty,is_full,size - Example: fixed-size data logger (sensor readings)
1. What is a Circular Queue?¶
A regular queue backed by an array wastes space once elements are dequeued from the front. A circular queue solves this by treating the array as a ring:
Circular Queue (capacity = 5)
+-----+-----+-----+-----+-----+
index → | 0 | 1 | 2 | 3 | 4 |
+-----+-----+-----+-----+-----+
data → | C | D | _ | A | B |
+-----+-----+-----+-----+-----+
▲ ▲
tail head
Dequeue removes from head (index 3).
Enqueue inserts at tail (index 2).
Both pointers advance with (ptr + 1) % capacity.
| Operation | Description | Time complexity |
|---|---|---|
enqueue(item) |
Add an item at the tail | O(1) |
dequeue() |
Remove and return the item at the head | O(1) |
peek() |
Return the head item without removing it | O(1) |
is_empty() |
Return True if no items are stored |
O(1) |
is_full() |
Return True if capacity is reached |
O(1) |
size() |
Return the number of stored items | O(1) |
2. The CircularQueue class¶
class CircularQueue:
"""
Fixed-capacity FIFO queue backed by a ring buffer.
Parameters
----------
capacity : int
Maximum number of elements the queue can hold.
"""
def __init__(self, capacity: int):
if capacity < 1:
raise ValueError("capacity must be at least 1")
self._capacity = capacity
self._buffer = [None] * capacity
self._head = 0 # index of the front element
self._tail = 0 # index where the next element will be written
self._count = 0
# ── Core operations ────────────────────────────────────────────────
def enqueue(self, item):
"""Add *item* to the tail. Raises OverflowError if the queue is full."""
if self.is_full():
raise OverflowError("enqueue into a full circular queue")
self._buffer[self._tail] = item
self._tail = (self._tail + 1) % self._capacity
self._count += 1
def dequeue(self):
"""Remove and return the front item. Raises IndexError if the queue is empty."""
if self.is_empty():
raise IndexError("dequeue from an empty circular queue")
item = self._buffer[self._head]
self._buffer[self._head] = None # release reference
self._head = (self._head + 1) % self._capacity
self._count -= 1
return item
def peek(self):
"""Return the front item without removing it. Raises IndexError if empty."""
if self.is_empty():
raise IndexError("peek at an empty circular queue")
return self._buffer[self._head]
# ── State queries ──────────────────────────────────────────────────
def is_empty(self):
"""Return True if the queue holds no items."""
return self._count == 0
def is_full(self):
"""Return True if the queue has reached its capacity."""
return self._count == self._capacity
def size(self):
"""Return the number of items currently stored."""
return self._count
@property
def capacity(self):
"""Maximum number of items the queue can hold."""
return self._capacity
# ── Display ────────────────────────────────────────────────────────
def __repr__(self):
items = []
for i in range(self._count):
items.append(self._buffer[(self._head + i) % self._capacity])
return f"CircularQueue({items}, capacity={self._capacity})"
def __str__(self):
items = []
for i in range(self._count):
items.append(str(self._buffer[(self._head + i) % self._capacity]))
return " <- ".join(items) or "(empty)"
3. Core operations¶
cq = CircularQueue(capacity=4)
print(f"Empty? {cq.is_empty()} Full? {cq.is_full()} Size: {cq.size()}/{cq.capacity}")
print()
for v in ["A", "B", "C", "D"]:
cq.enqueue(v)
print(f"enqueue({v!r}) → {cq} full={cq.is_full()}")
Empty? True Full? False Size: 0/4
enqueue('A') → A full=False
enqueue('B') → A <- B full=False
enqueue('C') → A <- B <- C full=False
enqueue('D') → A <- B <- C <- D full=True
# enqueue into a full queue raises OverflowError
try:
cq.enqueue("E")
except OverflowError as e:
print("Caught:", e)
Caught: enqueue into a full circular queue
# dequeue two items, then enqueue two more — observe the wrap-around
for _ in range(2):
item = cq.dequeue()
print(f"dequeue() → {item!r} queue: {cq}")
print()
for v in ["E", "F"]:
cq.enqueue(v)
print(f"enqueue({v!r}) → {cq} (tail wrapped around)")
dequeue() → 'A' queue: B <- C <- D
dequeue() → 'B' queue: C <- D
enqueue('E') → C <- D <- E (tail wrapped around)
enqueue('F') → C <- D <- E <- F (tail wrapped around)
print(f"peek() → {cq.peek()!r}")
print()
while not cq.is_empty():
print(f"dequeue() → {cq.dequeue()!r} remaining: {cq}")
peek() → 'C' dequeue() → 'C' remaining: D <- E <- F dequeue() → 'D' remaining: E <- F dequeue() → 'E' remaining: F dequeue() → 'F' remaining: (empty)
try:
cq.dequeue()
except IndexError as e:
print("Caught:", e)
Caught: dequeue from an empty circular queue
4. Example: fixed-size data logger¶
A circular queue is the natural choice for a rolling sensor log: when the buffer is full the oldest reading is overwritten by the newest, keeping only the last N measurements in memory at constant cost.
import random, time
random.seed(0)
class SensorLogger:
"""Rolling logger that keeps only the last *window* readings."""
def __init__(self, window: int):
self._queue = CircularQueue(window)
def record(self, value: float):
"""Add a new reading, evicting the oldest if the buffer is full."""
if self._queue.is_full():
self._queue.dequeue() # drop oldest
self._queue.enqueue(value)
def readings(self):
"""Return the current window of readings in order."""
buf = self._queue
return [buf._buffer[(buf._head + i) % buf._capacity] for i in range(buf._count)]
def average(self):
vals = self.readings()
return sum(vals) / len(vals) if vals else 0.0
# Simulate a temperature sensor that emits a reading every second
# Keep only the last 5 readings
logger = SensorLogger(window=5)
print(f"{'Reading #':<12} {'Temp (°C)':<12} {'Last 5 readings':<35} {'Avg (°C)'}")
print("-" * 72)
for i in range(1, 13):
temp = round(36.5 + random.uniform(-1.5, 1.5), 2)
logger.record(temp)
window = logger.readings()
print(f"{i:<12} {temp:<12} {str(window):<35} {logger.average():.2f}")
time.sleep(0.05) # simulate real-time sampling
Reading # Temp (°C) Last 5 readings Avg (°C) ------------------------------------------------------------------------ 1 37.53 [37.53] 37.53 2 37.27 [37.53, 37.27] 37.40 3 36.26 [37.53, 37.27, 36.26] 37.02 4 35.78 [37.53, 37.27, 36.26, 35.78] 36.71 5 36.53 [37.53, 37.27, 36.26, 35.78, 36.53] 36.67 6 36.21 [37.27, 36.26, 35.78, 36.53, 36.21] 36.41 7 37.35 [36.26, 35.78, 36.53, 36.21, 37.35] 36.43 8 35.91 [35.78, 36.53, 36.21, 37.35, 35.91] 36.36 9 36.43 [36.53, 36.21, 37.35, 35.91, 36.43] 36.49 10 36.75 [36.21, 37.35, 35.91, 36.43, 36.75] 36.53 11 37.72 [37.35, 35.91, 36.43, 36.75, 37.72] 36.83 12 36.51 [35.91, 36.43, 36.75, 37.72, 36.51] 36.66
© 2026 Ivan Cao-Berg Pittsburgh Supercomputing Center, Carnegie Mellon University
Licensed under the GNU General Public License v2.0 (GPL-2).
You may redistribute and/or modify this work under the terms of GPL-2.
This work is distributed WITHOUT ANY WARRANTY.
Happy computing.