Algorithm Search⌘K

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
violation_heap.pyRawGitHub
# 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

Read the full page about Violation heap