Brodal queue in Python
Priority queue with worst-case O(1) insert, find-min and decrease-key, and O(log n) extract-min.
- File
brodal_queue.py, 186 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/brodal-queue/. The authors are listed in prov/CZ84MC.jsonl.
"""A simplified, worst-case efficient priority queue in the spirit of Brodal.
This is NOT the full Brodal queue of 1996. It is the skew binomial queue
with a pointer to the minimum root, the structure that Brodal and Okasaki
use as the base of their worst-case queues. It gives, in the worst case:
find_min and insert in O(1); delete_min, delete, decrease_key and meld in
O(log n). The full Brodal queue also has meld and decrease_key in O(1).
"""
from typing import Any, Generic, Optional, Protocol, TypeVar
class Ordered(Protocol):
def __lt__(self, other: Any, /) -> bool: ...
K = TypeVar("K", bound=Ordered)
V = TypeVar("V")
class Item(Generic[K, V]):
__slots__ = ("value", "node")
def __init__(self, value: V) -> None:
self.value = value
self.node: Optional["Node[K, V]"] = None
class Node(Generic[K, V]):
__slots__ = ("item", "key", "rank", "children", "parent")
def __init__(self, item: Item[K, V], key: K) -> None:
self.item = item
self.key = key
self.rank = 0
self.children: list[Node[K, V]] = []
self.parent: Optional[Node[K, V]] = None
item.node = self
def _swap(x: Node[K, V], y: Node[K, V]) -> None:
"""Exchange the items and keys of two nodes and fix the handles."""
x.item, y.item = y.item, x.item
x.key, y.key = y.key, x.key
x.item.node = x
y.item.node = y
def _link(s: Node[K, V], t: Node[K, V]) -> Node[K, V]:
"""Link two roots of the same rank r into one root of rank r + 1."""
if t.key < s.key: # @line L1
s, t = t, s
s.children.append(t) # @line L2
t.parent = s
s.rank += 1
return s
class BrodalQueue(Generic[K, V]):
def __init__(self) -> None:
self.roots: list[Node[K, V]] = [] # by rank; only roots[0], roots[1] may tie
self.min: Optional[Node[K, V]] = None
self.size = 0
def __len__(self) -> int:
return self.size
def find_min(self) -> tuple[V, K]:
if self.min is None:
raise IndexError("find_min from an empty queue")
return self.min.item.value, self.min.key
def insert(self, value: V, key: K) -> Item[K, V]:
e: Item[K, V] = Item(value)
x = Node(e, key) # @line L3
r = self.roots
if len(r) >= 2 and r[0].rank == r[1].rank: # @line L4
was_min = self.min is r[0] or self.min is r[1]
u = _link(r[0], r[1]) # @line L5
if was_min:
self.min = u
u.children.append(x) # @line L6
x.parent = u
if x.key < u.key: # @line L7
_swap(x, u)
r[0:2] = [u]
x = u
else:
r.insert(0, x) # @line L8
if self.min is None or x.key < self.min.key: # @line L9
self.min = x
self.size += 1
return e
def _remove_root(self, m: Node[K, V]) -> None:
self.roots.remove(m) # @line L11
trees = self.roots + m.children # @line L12
for c in m.children:
c.parent = None
m.children = []
buckets: dict[int, Node[K, V]] = {}
for t in trees: # @line L13
while t.rank in buckets:
t = _link(t, buckets.pop(t.rank))
buckets[t.rank] = t
self.roots = [buckets[k] for k in sorted(buckets)] # @line L14
self.min = None
for t in self.roots:
if self.min is None or t.key < self.min.key:
self.min = t
self.size -= 1
def delete_min(self) -> tuple[V, K]:
m = self.min
if m is None: # @line L10
raise IndexError("delete_min from an empty queue")
e, key = m.item, m.key
self._remove_root(m)
e.node = None
return e.value, key
def decrease_key(self, e: Item[K, V], k: K) -> None:
x = e.node
if x is None:
raise KeyError("item not in the queue")
if x.key < k: # @line L15
raise ValueError("new key is larger than the current key")
x.key = k
while x.parent is not None and x.key < x.parent.key: # @line L16
_swap(x, x.parent)
x = x.parent
assert self.min is not None
if x.parent is None and x.key < self.min.key: # @line L17
self.min = x
def delete(self, e: Item[K, V]) -> None:
x = e.node
if x is None:
raise KeyError("item not in the queue")
while x.parent is not None: # @line L18
_swap(x, x.parent)
x = x.parent
self._remove_root(x) # @line L19
e.node = None
def meld(self, other: "BrodalQueue[K, V]") -> None:
trees = self.roots + other.roots # @line L20
buckets: dict[int, Node[K, V]] = {}
for t in trees:
while t.rank in buckets:
t = _link(t, buckets.pop(t.rank))
buckets[t.rank] = t
self.roots = [buckets[k] for k in sorted(buckets)]
if other.min is not None and (self.min is None or other.min.key < self.min.key):
self.min = other.min
self.size += other.size
other.roots, other.min, other.size = [], None, 0
def check_invariant(self) -> None:
r = self.roots
assert all(r[i].rank <= r[i + 1].rank for i in range(len(r) - 1))
assert all(r[i].rank < r[i + 1].rank for i in range(1, len(r) - 1))
count = 0
def walk(x: Node[K, V]) -> int:
assert x.item.node is x
size = 1
for c in x.children:
assert c.parent is x
assert not c.key < x.key
assert c.rank < x.rank
size += walk(c)
assert size >= 1 << x.rank
return size
for t in r:
assert t.parent is None
count += walk(t)
assert count == self.size
if self.size == 0:
assert self.min is None
else:
assert self.min in r
assert all(not t.key < self.min.key for t in r)
test_brodal_queue.py 127 lines
# SPDX-License-Identifier: MIT
# Part of https://algorithm.tamnd.com/brodal-queue/. The authors are listed in prov/CZ84MC.jsonl.
import random
import pytest
from brodal_queue import BrodalQueue, Node
def show(x: Node) -> str:
s = f"{x.key}/r{x.rank}"
return s + ("[" + " ".join(show(c) for c in x.children) + "]" if x.children else "")
def shape(q: BrodalQueue) -> str:
return " ".join(show(t) for t in q.roots)
def test_example() -> None:
q: BrodalQueue[int, str] = BrodalQueue()
it = {}
steps = []
for v, k in [("a", 6), ("b", 4), ("c", 9), ("d", 2), ("e", 7), ("f", 5), ("g", 8)]:
it[v] = q.insert(v, k)
steps.append(shape(q))
assert steps == [
"6/r0",
"4/r0 6/r0",
"4/r1[6/r0 9/r0]",
"2/r0 4/r1[6/r0 9/r0]",
"7/r0 2/r0 4/r1[6/r0 9/r0]",
"2/r1[7/r0 5/r0] 4/r1[6/r0 9/r0]",
"2/r2[7/r0 5/r0 4/r1[6/r0 9/r0] 8/r0]",
]
q.check_invariant()
assert q.find_min() == ("d", 2)
q.decrease_key(it["e"], 1)
assert shape(q) == "1/r2[2/r0 5/r0 4/r1[6/r0 9/r0] 8/r0]"
assert q.find_min() == ("e", 1)
assert q.delete_min() == ("e", 1)
assert shape(q) == "8/r0 2/r2[5/r0 4/r1[6/r0 9/r0]]"
q.check_invariant()
assert q.delete_min() == ("d", 2)
assert shape(q) == "4/r2[6/r0 9/r0 5/r1[8/r0]]"
q.delete(it["c"])
assert shape(q) == "4/r2[6/r0 5/r1[8/r0]]"
q.check_invariant()
assert len(q) == 4
def test_errors() -> None:
q: BrodalQueue[int, int] = BrodalQueue()
q.check_invariant()
with pytest.raises(IndexError):
q.find_min()
with pytest.raises(IndexError):
q.delete_min()
e = q.insert(1, 10)
with pytest.raises(ValueError):
q.decrease_key(e, 11)
q.delete(e)
with pytest.raises(KeyError):
q.delete(e)
with pytest.raises(KeyError):
q.decrease_key(e, 1)
assert len(q) == 0
q.check_invariant()
def test_meld() -> None:
a: BrodalQueue[int, int] = BrodalQueue()
b: BrodalQueue[int, int] = BrodalQueue()
for k in [9, 4, 7, 1]:
a.insert(k, k)
for k in [8, 2, 6]:
b.insert(k, k)
a.meld(b)
a.check_invariant()
assert len(a) == 7 and len(b) == 0
assert [a.delete_min()[1] for _ in range(7)] == [1, 2, 4, 6, 7, 8, 9]
def test_roots_stay_few() -> None:
q: BrodalQueue[int, int] = BrodalQueue()
for i in range(1000):
q.insert(i, (i * 7919) % 1000)
assert len(q.roots) <= (len(q)).bit_length() + 1
q.check_invariant()
out = [q.delete_min()[1] for _ in range(1000)]
assert out == list(range(1000))
def test_random_against_model() -> None:
rng = random.Random(1996)
for _ in range(600):
q: BrodalQueue[int, int] = BrodalQueue()
model: dict[int, int] = {}
items = {}
nxt = 0
for _ in range(rng.randrange(1, 50)):
op = rng.randrange(5)
if op <= 1 or not model:
k = rng.randrange(40)
items[nxt] = q.insert(nxt, k)
model[nxt] = k
nxt += 1
elif op == 2:
v, k = q.delete_min()
assert k == min(model.values())
assert model.pop(v) == k
elif op == 3:
v = rng.choice(sorted(model))
k = rng.randrange(model[v] + 1)
q.decrease_key(items[v], k)
model[v] = k
else:
v = rng.choice(sorted(model))
q.delete(items[v])
del model[v]
q.check_invariant()
assert len(q) == len(model)
if model:
assert q.find_min()[1] == min(model.values())
for v, it in items.items():
if v in model:
assert it.node is not None and it.node.key == model[v]
What the next level needs
- the tests did not run in this build