External stack in Python
Stack that keeps one block of its top in memory so that push and pop cost O(1/B) I/Os amortized.
- File
external_stack.py, 61 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-stack/. The authors are listed in prov/A69Z23.jsonl.
from typing import Generic, TypeVar
T = TypeVar("T")
class ExternalStack(Generic[T]):
"""A stack whose bottom part lives on a simulated disk in blocks of B items.
Memory holds a buffer of at most 2B items: the top of the stack. A push
into a full buffer first writes its B oldest items as one block; a pop
from an empty buffer first reads the last block back. After each block
transfer the buffer holds exactly B items, so any n operations make at
most n / 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.buf: list[T] = []
self.disk: list[list[T]] = []
self.reads = 0
self.writes = 0
def push(self, x: T) -> None:
"""Put x on top of the stack."""
if len(self.buf) == 2 * self.B: # @line L1
self.disk.append(self.buf[: self.B]) # @line L2
self.writes += 1
del self.buf[: self.B] # @line L3
self.buf.append(x) # @line L4
def pop(self) -> T:
"""Remove and return the top item. Raise IndexError if empty."""
if not self.buf: # @line L5
if not self.disk: # @line L6
raise IndexError("pop from an empty stack")
self.buf = self.disk.pop() # @line L7
self.reads += 1
return self.buf.pop() # @line L8
def top(self) -> T:
"""Return the top item without removing it."""
x = self.pop()
self.buf.append(x)
return x
def __len__(self) -> int:
return len(self.buf) + self.B * len(self.disk)
def items(self) -> list[T]:
"""Return the items from the bottom to the top, without I/O counts."""
return [x for blk in self.disk for x in blk] + self.buf
def check_invariant(self) -> None:
"""Raise AssertionError if a representation rule fails."""
assert len(self.buf) <= 2 * self.B
assert all(len(blk) == self.B for blk in self.disk)
test_external_stack.py 72 lines
# SPDX-License-Identifier: MIT
# Part of https://algorithm.tamnd.com/external-stack/. The authors are listed in prov/A69Z23.jsonl.
import random
import pytest
from external_stack import ExternalStack
def test_example() -> None:
s: ExternalStack[int] = ExternalStack(2)
for x in range(1, 8):
s.push(x)
s.check_invariant()
assert s.disk == [[1, 2], [3, 4]] and s.buf == [5, 6, 7]
assert (s.reads, s.writes) == (0, 2)
assert [s.pop() for _ in range(3)] == [7, 6, 5]
assert s.reads == 0
assert s.pop() == 4
assert s.reads == 1 and s.buf == [3] and s.disk == [[1, 2]]
s.push(8)
assert s.top() == 8 and len(s) == 4
assert [s.pop() for _ in range(4)] == [8, 3, 2, 1]
assert (s.reads, s.writes) == (2, 2)
s.check_invariant()
def test_edges() -> None:
with pytest.raises(ValueError):
ExternalStack(0)
s: ExternalStack[str] = ExternalStack(1)
with pytest.raises(IndexError):
s.pop()
s.push("a")
assert s.pop() == "a" and len(s) == 0
with pytest.raises(IndexError):
s.top()
def test_no_thrashing() -> None:
B = 8
s: ExternalStack[int] = ExternalStack(B)
for x in range(2 * B):
s.push(x)
before = s.reads + s.writes
for _ in range(1000):
s.push(0)
s.pop()
s.pop()
s.push(0)
assert s.reads + s.writes - before <= 1
def test_against_list() -> None:
rng = random.Random(5)
for _ in range(500):
B = rng.randrange(1, 6)
s: ExternalStack[int] = ExternalStack(B)
model: list[int] = []
n = rng.randrange(1, 200)
p = rng.random()
for _ in range(n):
if model and rng.random() < p:
assert s.pop() == model.pop()
else:
x = rng.randrange(1000)
s.push(x)
model.append(x)
s.check_invariant()
assert s.items() == model and len(s) == len(model)
assert s.reads + s.writes <= n // B
What the next level needs
- the tests did not run in this build