Violation heap in Python
Heap that tracks heap-order violations to achieve Fibonacci-heap bounds with simpler structure.
- File
violation_heap.py, 256 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/violation-heap/. The authors are listed in prov/HWWSW6.jsonl.
from typing import Any, Dict, Generic, Iterator, List, Optional, Protocol, TypeVar
class Ordered(Protocol):
def __lt__(self, other: Any, /) -> bool: ...
K = TypeVar("K", bound=Ordered)
class Node(Generic[K]):
"""A node of a violation heap.
The children of a node are in a doubly linked list, oldest first. last
points to the last (newest) child. left points to the previous sibling,
or is None for the first child and for a root. right points to the next
sibling; for the last child it points to the parent, and for a root it is
None. There is no parent pointer.
"""
__slots__ = ("key", "rank", "last", "left", "right")
def __init__(self, key: K) -> None:
self.key = key
self.rank = 0
self.last: Optional[Node[K]] = None
self.left: Optional[Node[K]] = None
self.right: Optional[Node[K]] = None
def _rank(x: "Optional[Node[Any]]") -> int:
return -1 if x is None else x.rank
def is_root(x: "Node[Any]") -> bool:
return x.right is None
def parent_if_active(x: "Node[Any]") -> "Optional[Node[Any]]":
"""The parent of x if x is one of the last two children, else None."""
y = x.right
if y is None:
return None
if y.last is x:
return y
if y.right is not None and y.right.last is y:
return y.right
return None
def active_children(z: "Node[Any]") -> "tuple[Optional[Node[Any]], Optional[Node[Any]]]":
a = z.last
return a, (None if a is None else a.left)
def formula_rank(z: "Node[Any]") -> int:
"""Formula (1) of Elmasry: ceil((r1 + r2) / 2) + 1 over the active children."""
a, b = active_children(z)
s = _rank(a) + _rank(b)
return -((-s) // 2) + 1
class ViolationHeap(Generic[K]):
"""Violation heap of Elmasry (2010): a min-heap with Fibonacci-heap bounds."""
def __init__(self) -> None:
self.roots: List[Node[K]] = []
self.min: Optional[Node[K]] = None
self.n = 0
self.joins = 0
self.rank_steps = 0
def __len__(self) -> int:
return self.n
def _add_root(self, x: Node[K]) -> None:
x.left = x.right = None
self.roots.append(x)
if self.min is None or x.key < self.min.key:
self.min = x
def _append_child(self, z: Node[K], c: Node[K]) -> None:
c.left, c.right = z.last, z
if z.last is not None:
z.last.right = c
z.last = c
def join3(self, z: Node[K], z1: Node[K], z2: Node[K]) -> Node[K]:
"""3-way join of three roots of equal rank."""
self.joins += 1
if z1.key < z.key and not z2.key < z1.key: # @line L1
z, z1 = z1, z
elif z2.key < z.key:
z, z2 = z2, z
a, b = active_children(z) # @line L2
if a is not None and b is not None and a.rank < b.rank:
self._swap_last_two(z)
self._append_child(z, z1) # @line L3
self._append_child(z, z2)
z.rank += 1
return z
def _swap_last_two(self, z: Node[K]) -> None:
a = z.last
assert a is not None and a.left is not None
b = a.left
c = b.left
a.left, a.right = c, b
b.left, b.right = a, z
if c is not None:
c.right = a
z.last = b
def insert(self, key: K) -> Node[K]:
"""Insert key and return its node, the handle for decrease_key."""
x = Node(key) # @line L4
self._add_root(x)
self.n += 1
return x
def meld(self, other: "ViolationHeap[K]") -> None:
"""Move all items of other into this heap."""
self.roots.extend(other.roots) # @line L5
if other.min is not None and (self.min is None or other.min.key < self.min.key):
self.min = other.min
self.n += other.n
self.joins += other.joins
other.roots, other.min, other.n = [], None, 0
def find_min(self) -> K:
if self.min is None:
raise IndexError("find_min from an empty heap")
return self.min.key
def delete_min(self) -> K:
"""Remove and return a smallest key."""
if self.min is None:
raise IndexError("delete_min from an empty heap")
z = self.min # @line L6
trees = [r for r in self.roots if r is not z]
c = z.last
while c is not None:
nxt = c.left
c.left = c.right = None
trees.append(c)
c = nxt
bucket: Dict[int, List[Node[K]]] = {}
for x in trees: # @line L7
while len(bucket.get(x.rank, [])) == 2:
y1, y2 = bucket.pop(x.rank)
x = self.join3(x, y1, y2)
bucket.setdefault(x.rank, []).append(x) # @line L8
self.roots, self.min = [], None # @line L9
for group in bucket.values():
for x in group:
self._add_root(x)
self.n -= 1
return z.key
def _replace(self, x: Node[K], y: Optional[Node[K]]) -> None:
"""Put y in the place of x in its sibling list, or remove x if y is None."""
l, r = x.left, x.right
assert r is not None
if y is None:
if r.last is x:
r.last = l
if l is not None:
l.right = r
else:
r.left = l
if l is not None:
l.right = r
else:
y.left, y.right = l, r
if l is not None:
l.right = y
if r.last is x:
r.last = y
else:
r.left = y
x.left = x.right = None
def _detach_child(self, x: Node[K], c: Node[K]) -> None:
"""Remove the child c from the list of x."""
self._replace(c, None)
def decrease_key(self, x: Node[K], key: K) -> None:
"""Lower the key of node x to key, which must not be larger."""
if x.key < key:
raise ValueError("new key is larger than the current key")
x.key = key # @line L10
if is_root(x):
if self.min is None or key < self.min.key:
self.min = x
return
p = parent_if_active(x) # @line L11
if p is not None and not key < p.key:
return
a, b = active_children(x) # @line L12
z1 = a if b is None or not a.rank < b.rank else b # type: ignore[union-attr]
if z1 is not None: # @line L13
self._detach_child(x, z1)
self._replace(x, z1)
x.rank = formula_rank(x) # @line L14
self._add_root(x)
while p is not None: # @line L15
k = formula_rank(p)
if k >= p.rank:
return
self.rank_steps += 1 # @line L16
p.rank = k
p = parent_if_active(p)
def tree_nodes(self, r: Node[K]) -> Iterator[Node[K]]:
"""All nodes of the tree rooted at r, without recursion."""
stack = [r]
while stack:
v = stack.pop()
yield v
c = v.last
while c is not None:
stack.append(c)
c = c.left
def check_invariant(self) -> None:
"""Check the links, heap order, the rank rule and the size bound."""
fib = [0, 1]
while len(fib) < 200:
fib.append(fib[-1] + fib[-2])
count = 0
for r in self.roots:
assert r.right is None and r.left is None
assert self.min is not None and not r.key < self.min.key
order: List[Node[K]] = []
for v in self.tree_nodes(r):
order.append(v)
c, prev = v.last, None
while c is not None:
assert c.right is (v if prev is None else prev)
assert not c.key < v.key
prev, c = c, c.left
assert 0 <= v.rank <= formula_rank(v)
size: Dict[int, int] = {}
for v in reversed(order):
s, c = 1, v.last
while c is not None:
s += size[id(c)]
c = c.left
size[id(v)] = s
assert s >= fib[v.rank + 2]
count += len(order)
assert count == self.n
assert (self.min is None) == (self.n == 0)
test_violation_heap.py 146 lines
# SPDX-License-Identifier: MIT
# Part of https://algorithm.tamnd.com/violation-heap/. The authors are listed in prov/HWWSW6.jsonl.
import heapq
import math
import random
from typing import Dict, List
import pytest
from violation_heap import Node, ViolationHeap, formula_rank
def show(x: Node[int]) -> str:
"""Write a tree as key:rank(children, oldest first)."""
kids: List[Node[int]] = []
c = x.last
while c is not None:
kids.append(c)
c = c.left
kids.reverse()
s = f"{x.key}:{x.rank}"
return s if not kids else s + "(" + ", ".join(show(k) for k in kids) + ")"
def dump(h: ViolationHeap[int]) -> str:
return " ".join(show(r) for r in h.roots)
def test_page_example() -> None:
h: ViolationHeap[int] = ViolationHeap()
nd = {k: h.insert(k) for k in range(10, 110, 10)}
assert h.delete_min() == 10 and h.joins == 4
assert dump(h) == "20:2(40:0, 30:0, 80:1(100:0, 90:0), 50:1(70:0, 60:0))"
steps = []
for old, new in [(90, 5), (50, 45), (40, 35), (50, 15), (100, 8)]:
h.rank_steps = 0
h.decrease_key(nd[old], new)
h.check_invariant()
steps.append(h.rank_steps)
assert steps == [0, 0, 0, 0, 2]
assert dump(h) == "20:1(30:0, 80:0, 60:0) 5:0 35:0 15:1(70:0) 8:0"
h.joins = 0
assert h.delete_min() == 5 and h.joins == 0
assert dump(h) == "20:1(30:0, 80:0, 60:0) 15:1(70:0) 35:0 8:0"
h.check_invariant()
def test_formula() -> None:
z: Node[int] = Node(0)
assert formula_rank(z) == 0
h: ViolationHeap[int] = ViolationHeap()
for k in range(4):
h.insert(k)
h.delete_min()
(r,) = h.roots
assert r.key == 1 and r.rank == 1 and formula_rank(r) == 1
def test_empty_and_single() -> None:
h: ViolationHeap[int] = ViolationHeap()
with pytest.raises(IndexError):
h.delete_min()
with pytest.raises(IndexError):
h.find_min()
x = h.insert(7)
with pytest.raises(ValueError):
h.decrease_key(x, 8)
h.decrease_key(x, 3)
assert h.find_min() == 3 and h.delete_min() == 3 and len(h) == 0
h.check_invariant()
def test_duplicates_and_meld() -> None:
a: ViolationHeap[int] = ViolationHeap()
b: ViolationHeap[int] = ViolationHeap()
for k in [5, 5, 1, 5]:
a.insert(k)
for k in [1, 2, 5]:
b.insert(k)
a.meld(b)
assert len(b) == 0 and len(a) == 7
assert [a.delete_min() for _ in range(7)] == [1, 1, 2, 5, 5, 5, 5]
def test_rank_bound_after_many_joins() -> None:
h: ViolationHeap[int] = ViolationHeap()
for k in range(3**6 + 1):
h.insert(k)
h.delete_min()
(r,) = h.roots
assert r.rank == 6
h.check_invariant()
def test_random_against_heapq() -> None:
rng = random.Random(20100627)
for case in range(600):
h: ViolationHeap[int] = ViolationHeap()
ref: List[tuple[int, int]] = []
handles: Dict[int, Node[int]] = {}
keys: Dict[int, int] = {}
nxt = 0
cost = 0
budget = 0.0
for _ in range(rng.randint(1, 120)):
op = rng.random()
if op < 0.4 or not keys:
k = rng.randint(0, 60)
handles[nxt] = h.insert(k)
keys[nxt] = k
heapq.heappush(ref, (k, nxt))
nxt += 1
budget += 1
elif op < 0.75:
i = rng.choice(list(keys))
k = keys[i] - rng.randint(0, 20)
h.rank_steps = 0
h.decrease_key(handles[i], k)
cost += h.rank_steps
keys[i] = k
heapq.heappush(ref, (k, i))
budget += 1
else:
while ref[0][0] != keys.get(ref[0][1]):
heapq.heappop(ref)
m = h.find_min()
assert m == ref[0][0]
n_before = len(h)
h.joins = 0
assert h.delete_min() == m
cost += h.joins + len(h.roots)
budget += math.log2(n_before) + 1
for i, v in list(keys.items()):
if v == m and handles[i].right is None and handles[i] not in h.roots:
del keys[i]
break
if case % 20 == 0:
h.check_invariant()
h.check_invariant()
assert len(h) == len(keys)
assert cost <= 8 * budget
out = []
while len(h):
out.append(h.delete_min())
assert out == sorted(keys.values())
What the next level needs
- the tests did not run in this build