Python - deque (Double-Ended Queue)
Python's `list` type performs poorly when you need to add or remove elements from the left side. Every insertion at index 0 requires shifting all existing elements, resulting in O(n) complexity. The...
Key Insights
- The
deque(double-ended queue) from Python’scollectionsmodule provides O(1) time complexity for append and pop operations from both ends, unlike lists which have O(n) for left-side operations - Deques excel at implementing sliding windows, task scheduling, breadth-first search algorithms, and maintaining bounded histories where you need efficient operations at both ends
- Thread-safe atomic operations and optional maximum length with automatic eviction make deques ideal for producer-consumer patterns and fixed-size buffers
Understanding deque Performance Characteristics
Python’s list type performs poorly when you need to add or remove elements from the left side. Every insertion at index 0 requires shifting all existing elements, resulting in O(n) complexity. The deque class solves this with a doubly-linked list implementation that maintains O(1) performance for operations at both ends.
from collections import deque
import time
# Performance comparison: list vs deque
def benchmark_left_operations(n=100000):
# List performance
start = time.perf_counter()
lst = []
for i in range(n):
lst.insert(0, i)
list_time = time.perf_counter() - start
# Deque performance
start = time.perf_counter()
dq = deque()
for i in range(n):
dq.appendleft(i)
deque_time = time.perf_counter() - start
print(f"List insert(0): {list_time:.4f}s")
print(f"Deque appendleft: {deque_time:.4f}s")
print(f"Speedup: {list_time/deque_time:.1f}x")
benchmark_left_operations()
# Output example:
# List insert(0): 2.3456s
# Deque appendleft: 0.0089s
# Speedup: 263.6x
Basic Operations and API
The deque API mirrors Python’s list with additional methods for both-ended operations. All append and pop operations from either end execute in constant time.
from collections import deque
# Creating deques
dq = deque([1, 2, 3, 4, 5])
empty_dq = deque()
bounded_dq = deque(maxlen=3) # Fixed-size deque
# Append operations
dq.append(6) # Add to right: [1, 2, 3, 4, 5, 6]
dq.appendleft(0) # Add to left: [0, 1, 2, 3, 4, 5, 6]
# Pop operations
right = dq.pop() # Remove from right: returns 6
left = dq.popleft() # Remove from left: returns 0
# Extend operations
dq.extend([7, 8, 9]) # Extend right
dq.extendleft([10, 11, 12]) # Extend left (reverses order!)
# Rotation
dq = deque([1, 2, 3, 4, 5])
dq.rotate(2) # Rotate right: [4, 5, 1, 2, 3]
dq.rotate(-1) # Rotate left: [5, 1, 2, 3, 4]
# Access by index (O(n) operation)
print(dq[0]) # First element
print(dq[-1]) # Last element
Implementing a Sliding Window
Deques excel at sliding window algorithms where you need to maintain a window of elements while efficiently adding new items and removing old ones.
from collections import deque
def max_sliding_window(nums, k):
"""
Find maximum value in each sliding window of size k.
Time complexity: O(n)
"""
if not nums or k == 0:
return []
result = []
# Store indices of elements in decreasing order of their values
window = deque()
for i, num in enumerate(nums):
# Remove indices outside current window
while window and window[0] <= i - k:
window.popleft()
# Remove indices of smaller elements (they won't be max)
while window and nums[window[-1]] < num:
window.pop()
window.append(i)
# Add to result once we have a full window
if i >= k - 1:
result.append(nums[window[0]])
return result
# Example usage
nums = [1, 3, -1, -3, 5, 3, 6, 7]
k = 3
print(max_sliding_window(nums, k))
# Output: [3, 3, 5, 5, 6, 7]
Bounded Deques for Fixed-Size Buffers
Setting maxlen creates a bounded deque that automatically evicts the oldest element when full. This is perfect for maintaining recent history or implementing LRU-style caches.
from collections import deque
import time
class RecentCounter:
"""Count requests in the last 3000 milliseconds."""
def __init__(self):
self.requests = deque()
def ping(self, t: int) -> int:
self.requests.append(t)
# Remove requests older than 3000ms
while self.requests and self.requests[0] < t - 3000:
self.requests.popleft()
return len(self.requests)
# Usage
counter = RecentCounter()
print(counter.ping(1)) # 1
print(counter.ping(100)) # 2
print(counter.ping(3001)) # 3
print(counter.ping(3002)) # 3 (request at t=1 is now too old)
class CircularBuffer:
"""Fixed-size buffer that overwrites oldest data."""
def __init__(self, capacity):
self.buffer = deque(maxlen=capacity)
def add(self, item):
self.buffer.append(item) # Automatically removes oldest if full
def get_all(self):
return list(self.buffer)
# Usage
buffer = CircularBuffer(3)
for i in range(5):
buffer.add(i)
print(f"After adding {i}: {buffer.get_all()}")
# Output:
# After adding 0: [0]
# After adding 1: [0, 1]
# After adding 2: [0, 1, 2]
# After adding 3: [1, 2, 3] # 0 was evicted
# After adding 4: [2, 3, 4] # 1 was evicted
Task Queue and BFS Implementation
Deques are the standard choice for implementing breadth-first search and task queues where you process items in FIFO order.
from collections import deque
def bfs_shortest_path(graph, start, end):
"""
Find shortest path in unweighted graph using BFS.
"""
if start == end:
return [start]
queue = deque([(start, [start])])
visited = {start}
while queue:
node, path = queue.popleft()
for neighbor in graph.get(node, []):
if neighbor in visited:
continue
if neighbor == end:
return path + [neighbor]
visited.add(neighbor)
queue.append((neighbor, path + [neighbor]))
return None
# Example graph
graph = {
'A': ['B', 'C'],
'B': ['A', 'D', 'E'],
'C': ['A', 'F'],
'D': ['B'],
'E': ['B', 'F'],
'F': ['C', 'E']
}
print(bfs_shortest_path(graph, 'A', 'F'))
# Output: ['A', 'C', 'F']
class TaskScheduler:
"""Simple task queue with priority levels."""
def __init__(self):
self.high_priority = deque()
self.normal_priority = deque()
def add_task(self, task, priority='normal'):
if priority == 'high':
self.high_priority.append(task)
else:
self.normal_priority.append(task)
def process_next(self):
if self.high_priority:
return self.high_priority.popleft()
elif self.normal_priority:
return self.normal_priority.popleft()
return None
# Usage
scheduler = TaskScheduler()
scheduler.add_task("Regular task 1")
scheduler.add_task("Urgent task", priority='high')
scheduler.add_task("Regular task 2")
while task := scheduler.process_next():
print(f"Processing: {task}")
# Output:
# Processing: Urgent task
# Processing: Regular task 1
# Processing: Regular task 2
Thread Safety and Concurrent Operations
Deque operations are atomic and thread-safe for append and pop operations, making them suitable for simple producer-consumer patterns without additional locking.
from collections import deque
import threading
import time
import random
class ThreadSafeQueue:
"""Producer-consumer queue using deque."""
def __init__(self, maxlen=None):
self.queue = deque(maxlen=maxlen)
self.running = True
def produce(self, producer_id, count):
for i in range(count):
item = f"P{producer_id}-Item{i}"
self.queue.append(item)
print(f"Produced: {item}")
time.sleep(random.uniform(0.01, 0.05))
def consume(self, consumer_id):
while self.running or self.queue:
try:
item = self.queue.popleft()
print(f"Consumer {consumer_id} consumed: {item}")
time.sleep(random.uniform(0.01, 0.05))
except IndexError:
time.sleep(0.01)
def stop(self):
self.running = False
# Usage
queue = ThreadSafeQueue()
producers = [
threading.Thread(target=queue.produce, args=(i, 5))
for i in range(2)
]
consumers = [
threading.Thread(target=queue.consume, args=(i,))
for i in range(3)
]
for t in producers + consumers:
t.start()
for t in producers:
t.join()
queue.stop()
for t in consumers:
t.join()
When to Use Deque vs List
Choose deque when you need frequent operations at both ends of a sequence. Use list when you primarily append to the end and access by index. Deques have slightly higher memory overhead per element and slower index access (O(n) vs O(1) for lists), but this is negligible for most applications compared to the performance gains for end operations.
# Use deque for:
# - Queues and stacks
# - Sliding windows
# - BFS algorithms
# - Recent history tracking
# - Palindrome checking
# Use list for:
# - Random access by index
# - Sorting operations
# - Primarily right-side operations
# - When you need list comprehensions extensively