class CircularQueue:
def __init__(self, capacity=10):
self.capacity = capacity
self.queue = [None] * capacity
self.front = 0
self.rear = -1
self.size = 0
def enqueue(self, item):
if self.is_full():
self._resize(2 * self.capacity)
self.rear = (self.rear + 1) % self.capacity
self.queue[self.rear] = item
self.size += 1
def dequeue(self):
if self.is_empty():
raise IndexError("Queue is empty")
item = self.queue[self.front]
self.queue[self.front] = None
self.front = (self.front + 1) % self.capacity
self.size -= 1
return item
def peek(self):
if self.is_empty():
raise IndexError("Queue is empty")
return self.queue[self.front]
def is_empty(self):
return self.size == 0
def is_full(self):
return self.size == self.capacity
def _resize(self, new_capacity):
new_queue = [None] * new_capacity
for i in range(self.size):
new_queue[i] = self.queue[(self.front + i) % self.capacity]
self.queue = new_queue
self.front = 0
self.rear = self.size - 1
self.capacity = new_capacity
class Node:
def __init__(self, data):
self.data = data
self.next = None
class LinkedQueue:
def __init__(self):
self.front = None
self.rear = None
self.size = 0
def enqueue(self, item):
new_node = Node(item)
if self.is_empty():
self.front = self.rear = new_node
else:
self.rear.next = new_node
self.rear = new_node
self.size += 1
def dequeue(self):
if self.is_empty():
raise IndexError("Queue is empty")
item = self.front.data
self.front = self.front.next
if self.front is None:
self.rear = None
self.size -= 1
return item
def peek(self):
if self.is_empty():
raise IndexError("Queue is empty")
return self.front.data
def is_empty(self):
return self.front is None
import heapq
class PriorityQueue:
def __init__(self):
self.queue = []
self.index = 0
def enqueue(self, item, priority):
heapq.heappush(self.queue, (priority, self.index, item))
self.index += 1
def dequeue(self):
if self.is_empty():
raise IndexError("Queue is empty")
return heapq.heappop(self.queue)[2]
def is_empty(self):
return len(self.queue) == 0
class Deque:
def __init__(self):
self.items = []
def add_front(self, item):
self.items.insert(0, item)
def add_rear(self, item):
self.items.append(item)
def remove_front(self):
if not self.is_empty():
return self.items.pop(0)
raise IndexError("Deque is empty")
def remove_rear(self):
if not self.is_empty():
return self.items.pop()
raise IndexError("Deque is empty")
def is_empty(self):
return len(self.items) == 0
def dequeue(self):
if self.is_empty():
raise IndexError("Queue underflow")
def dequeue(self):
if self.front == self.rear:
item = self.queue[self.front]
self.front = self.rear = -1
return item
def enqueue(self, item):
if (self.rear + 1) % self.capacity == self.front:
self._resize(2 * self.capacity)
class BatchQueue:
def __init__(self, batch_size=1000):
self.queue = []
self.batch_size = batch_size
def enqueue_batch(self, items):
if len(items) > self.batch_size:
for i in range(0, len(items), self.batch_size):
chunk = items[i:i + self.batch_size]
self.queue.extend(chunk)
else:
self.queue.extend(items)
class QueueNodePool:
def __init__(self, pool_size=1000):
self.pool = [Node(None) for _ in range(pool_size)]
self.available = set(range(pool_size))
def get_node(self, data):
if self.available:
idx = self.available.pop()
node = self.pool[idx]
node.data = data
return node
return Node(data)
class SmartQueue:
def _resize(self, new_capacity):
if new_capacity < self.min_capacity:
return False
if self.size < new_capacity // 4:
new_capacity = max(new_capacity // 2, self.min_capacity)
new_queue = [None] * new_capacity
for i in range(self.size):
new_queue[i] = self.queue[(self.front + i) % self.capacity]
self.queue = new_queue
self.capacity = new_capacity
self.front = 0
self.rear = self.size - 1
return True
def clear(self):
while not self.is_empty():
self.dequeue()
self.front = self.rear = None
class CacheOptimizedQueue:
def __init__(self):
self.block_size = 64
self.queue = bytearray(self.block_size * 16)
self.front = 0
self.rear = 0
from threading import Lock
class ThreadSafeQueue:
def __init__(self):
self.queue = []
self.lock = Lock()
def enqueue(self, item):
with self.lock:
self.queue.append(item)
def dequeue(self):
with self.lock:
if not self.queue:
raise IndexError("Queue is empty")
return self.queue.pop(0)
def is_full(self):
return self.rear == self.capacity - 1
def is_full(self):
return (self.rear + 1) % self.capacity == self.front
def dequeue(self):
item = self.queue[self.front]
self.front += 1
return item
def dequeue(self):
item = self.queue[self.front]
self.queue[self.front] = None
self.front = (self.front + 1) % self.capacity
return item
def dequeue(self):
item = self.front.data
self.front = self.front.next
return item
def dequeue(self):
item = self.front.data
old_front = self.front
self.front = self.front.next
old_front.next = None
if self.front is None:
self.rear = None
return item
def enqueue(self, item):
self.rear = (self.rear + 1) % self.capacity
self.queue[self.rear] = item
def enqueue(self, item):
self.rear = (self.rear + 1) % self.capacity
self.queue[self.rear] = item
self.size += 1