External queue in Python
FIFO queue that buffers its head and tail blocks in memory, giving O(1/B) amortized I/Os per operation.
- File
external_queue.py, 68 lines- Level
- I0 not checked yet
- Toolchain
- Python 3.14, pytest 9
- Test framework
- pytest
- Run the tests
python3 -m pytest
# SPDX-License-Identifier: MIT
# Part of https://algorithm.tamnd.com/external-queue/. The authors are listed in prov/HQ3Q31.jsonl.
from collections import deque
from typing import Generic, TypeVar
T = TypeVar("T")
class ExternalQueue(Generic[T]):
"""A FIFO queue whose middle part lives on a simulated disk in blocks of B.
Memory holds two buffers of at most B items: `head`, the front of the
queue, and `tail`, the back. A full tail is written to disk as one block;
an empty head is refilled with the oldest block, or with the tail when
the disk holds no block. Each item is written at most once and read at
most once, so n operations make at most 2n / B transfers. `reads` and
`writes` count the block transfers.
"""
def __init__(self, block_size: int) -> None:
if block_size < 1:
raise ValueError("the block size must be at least 1")
self.B = block_size
self.head: deque[T] = deque()
self.tail: list[T] = []
self.disk: deque[list[T]] = deque()
self.reads = 0
self.writes = 0
def enqueue(self, x: T) -> None:
"""Put x at the back of the queue."""
self.tail.append(x) # @line L1
if len(self.tail) == self.B: # @line L2
self.disk.append(self.tail) # @line L3
self.writes += 1
self.tail = [] # @line L4
def dequeue(self) -> T:
"""Remove and return the front item. Raise IndexError if empty."""
if not self.head: # @line L5
if self.disk: # @line L6
self.head = deque(self.disk.popleft()) # @line L7
self.reads += 1
else:
if not self.tail: # @line L8
raise IndexError("dequeue from an empty queue")
self.head, self.tail = deque(self.tail), [] # @line L9
return self.head.popleft() # @line L10
def front(self) -> T:
"""Return the front item without removing it."""
x = self.dequeue()
self.head.appendleft(x)
return x
def __len__(self) -> int:
return len(self.head) + self.B * len(self.disk) + len(self.tail)
def items(self) -> list[T]:
"""Return the items from the front to the back, without I/O counts."""
return list(self.head) + [x for blk in self.disk for x in blk] + self.tail
def check_invariant(self) -> None:
"""Raise AssertionError if a representation rule fails."""
assert len(self.head) <= self.B
assert len(self.tail) < self.B
assert all(len(blk) == self.B for blk in self.disk)
test_external_queue.py 73 lines
# SPDX-License-Identifier: MIT
# Part of https://algorithm.tamnd.com/external-queue/. The authors are listed in prov/HQ3Q31.jsonl.
import random
from collections import deque
import pytest
from external_queue import ExternalQueue
def test_example() -> None:
q: ExternalQueue[int] = ExternalQueue(3)
for x in range(1, 8):
q.enqueue(x)
q.check_invariant()
assert list(q.disk) == [[1, 2, 3], [4, 5, 6]] and q.tail == [7]
assert q.writes == 2 and q.reads == 0
assert q.dequeue() == 1
assert q.reads == 1 and list(q.head) == [2, 3]
q.enqueue(8)
assert [q.dequeue() for _ in range(5)] == [2, 3, 4, 5, 6]
assert q.reads == 2
assert q.dequeue() == 7
assert q.reads == 2 and list(q.head) == [8] and q.tail == []
assert q.front() == 8 and len(q) == 1
assert q.dequeue() == 8 and len(q) == 0
assert (q.reads, q.writes) == (2, 2)
def test_edges() -> None:
with pytest.raises(ValueError):
ExternalQueue(0)
q: ExternalQueue[str] = ExternalQueue(1)
with pytest.raises(IndexError):
q.dequeue()
q.enqueue("a")
assert q.writes == 1
assert q.front() == "a" and q.dequeue() == "a"
with pytest.raises(IndexError):
q.front()
def test_alternating_is_free() -> None:
q: ExternalQueue[int] = ExternalQueue(16)
for x in range(1000):
q.enqueue(x)
assert q.dequeue() == x
# each block of 16 items is written once and read once at most
assert q.reads + q.writes <= 2 * 1000 // 16
def test_against_deque() -> None:
rng = random.Random(11)
for _ in range(500):
B = rng.randrange(1, 6)
q: ExternalQueue[int] = ExternalQueue(B)
model: deque[int] = deque()
n = rng.randrange(1, 200)
p = rng.random()
enq = 0
for _ in range(n):
if model and rng.random() < p:
assert q.dequeue() == model.popleft()
else:
x = rng.randrange(1000)
q.enqueue(x)
model.append(x)
enq += 1
q.check_invariant()
assert q.items() == list(model) and len(q) == len(model)
assert q.writes <= enq // B
assert q.reads <= q.writes
What the next level needs
- the tests did not run in this build