Algorithm Search⌘K

Strict Fibonacci heap in Python

Fibonacci-like heap with worst-case O(1) decrease-key and O(log n) delete.

File
strict_fibonacci_heap.py, 426 lines
Level
I0 not checked yet
Toolchain
Python 3.14, pytest 9
Test framework
pytest
Run the tests
python3 -m pytest
strict_fibonacci_heap.pyRawGitHub
# SPDX-License-Identifier: MIT
# Part of https://algorithm.tamnd.com/strict-fibonacci-heap/. The authors are listed in prov/V45MQ4.jsonl.

"""Strict Fibonacci heap after Brodal, Lagogiannis and Tarjan (STOC 2012), simplified.

This follows the abstract structure of Sections 2 to 4 of the paper: one
heap-ordered tree, active and passive nodes with a shared flag, rank and loss,
the queue Q, and the five transformations. It does not build the pointer-level
fix-list and rank-list of Section 6. It finds the transformations with sets
and short scans of the children of the root, so an operation takes O(log n)
time in the worst case here, not O(1).
"""

import math
from typing import Any, Dict, Generic, List, Optional, Set, Tuple, TypeVar

K = TypeVar("K")


class Record:
    """The shared active record: a node is active if its record has flag True."""

    __slots__ = ("flag",)

    def __init__(self) -> None:
        self.flag = True


class Item(Generic[K]):
    """A handle: the item with its key, and the node that holds it now."""

    __slots__ = ("key", "uid", "node", "minus_inf")
    _next = 0

    def __init__(self, key: K) -> None:
        self.key = key
        Item._next += 1
        self.uid = Item._next
        self.node: Optional["Node[K]"] = None
        self.minus_inf = False

    def order(self) -> Tuple[Any, ...]:
        """Distinct keys: ties are broken by the item identifier."""
        return (0, self.uid) if self.minus_inf else (1, self.key, self.uid)


class Node(Generic[K]):
    __slots__ = ("item", "parent", "children", "rec", "rank", "loss", "qprev", "qnext")

    def __init__(self, item: Item[K]) -> None:
        self.item = item
        item.node = self
        self.parent: Optional[Node[K]] = None
        self.children: List[Node[K]] = []  # left to right
        self.rec: Optional[Record] = None
        self.rank = 0
        self.loss = 0
        self.qprev: Node[K] = self
        self.qnext: Node[K] = self

    @property
    def active(self) -> bool:
        return self.rec is not None and self.rec.flag

    def key(self) -> Tuple[Any, ...]:
        return self.item.order()


class StrictFibonacciHeap(Generic[K]):
    def __init__(self) -> None:
        self.root: Optional[Node[K]] = None
        self.n = 0
        self.rec = Record()
        self.qhead: Optional[Node[K]] = None
        self.aroots: Set[Node[K]] = set()  # active roots
        self.lossy: Set[Node[K]] = set()  # active nodes with positive loss
        self.counts: Dict[str, int] = {}

    def __len__(self) -> int:
        return self.n

    # -- helpers -------------------------------------------------------------

    def _count(self, name: str) -> None:
        self.counts[name] = self.counts.get(name, 0) + 1

    def _is_aroot(self, x: Node[K]) -> bool:
        return x.active and x.parent is not None and not x.parent.active

    def _cut(self, x: Node[K]) -> Node[K]:
        """Remove x from the child list of its parent and return the parent."""
        p = x.parent
        assert p is not None
        p.children.remove(x)
        x.parent = None
        return p

    def _link(self, x: Node[K], y: Node[K]) -> None:
        """Make x a child of y: leftmost if x is active, rightmost if passive."""
        if x.parent is not None:
            self._cut(x)
        x.parent = y
        if x.active:
            y.children.insert(0, x)
        else:
            y.children.append(x)

    def _set_loss(self, x: Node[K], loss: int) -> None:
        x.loss = loss
        if loss > 0 and x.active:
            self.lossy.add(x)
        else:
            self.lossy.discard(x)

    def _refresh_aroot(self, x: Node[K]) -> None:
        if self._is_aroot(x):
            self.aroots.add(x)
        else:
            self.aroots.discard(x)

    @staticmethod
    def _linkable(x: Node[K]) -> bool:
        return not x.active and (not x.children or not x.children[0].active)

    # -- the queue Q ---------------------------------------------------------

    @staticmethod
    def _concat(a: Optional[Node[K]], b: Optional[Node[K]]) -> Optional[Node[K]]:
        if a is None:
            return b
        if b is None:
            return a
        at, bt = a.qprev, b.qprev
        at.qnext, b.qprev = b, at
        bt.qnext, a.qprev = a, bt
        return a

    def _q_remove(self, x: Node[K]) -> None:
        if x.qnext is x:
            self.qhead = None
        else:
            x.qprev.qnext, x.qnext.qprev = x.qnext, x.qprev
            if self.qhead is x:
                self.qhead = x.qnext
        x.qprev = x.qnext = x

    # -- transformations -----------------------------------------------------

    def _active_root_reduction(self) -> bool:
        """Link two active roots of equal rank."""
        by_rank: Dict[int, Node[K]] = {}
        for a in self.aroots:
            b = by_rank.get(a.rank)
            if b is None:
                by_rank[a.rank] = a
                continue
            self._count("active-root")
            x, y = (a, b) if a.key() < b.key() else (b, a)
            self._link(y, x)
            x.rank += 1
            self.aroots.discard(y)
            z = x.children[-1]
            if not z.active:
                assert self.root is not None
                self._link(z, self.root)
            return True
        return False

    def _root_degree_reduction(self) -> bool:
        """Make three passive linkable children of the root into two active nodes."""
        root = self.root
        assert root is not None
        cand = [c for c in reversed(root.children) if self._linkable(c)][:3]
        if len(cand) < 3:
            return False
        self._count("root-degree")
        x, y, z = sorted(cand, key=lambda v: v.key())
        x.rec = y.rec = self.rec
        x.rank, y.rank = 0, 0
        self._link(z, y)
        self._link(y, x)
        x.rank = 1
        self._link(x, root)  # x is active, so it becomes the leftmost child
        self._set_loss(x, 0)
        self._set_loss(y, 0)
        self.aroots.add(x)
        return True

    def _loss_reduction(self) -> bool:
        """One-node or two-node loss reduction, if possible."""
        by_rank: Dict[int, Node[K]] = {}
        for x in list(self.lossy):
            if x.loss >= 2:
                self._count("one-node-loss")
                y = self._cut(x)
                assert self.root is not None
                self._link(x, self.root)
                self._set_loss(x, 0)
                self.aroots.add(x)
                y.rank -= 1
                if not self._is_aroot(y):
                    self._set_loss(y, y.loss + 1)
                return True
        for x in list(self.lossy):
            other = by_rank.get(x.rank)
            if other is None:
                by_rank[x.rank] = x
                continue
            self._count("two-node-loss")
            a, b = (x, other) if x.key() < other.key() else (other, x)
            z = self._cut(b)
            self._link(b, a)
            a.rank += 1
            self._set_loss(a, 0)
            self._set_loss(b, 0)
            # The old parent z is updated last: z may be a itself.
            z.rank -= 1
            if not self._is_aroot(z):
                self._set_loss(z, z.loss + 1)
            return True
        return False

    def _reductions(self, arr: int, rdr: int) -> None:
        """Do up to arr active root reductions and rdr root degree reductions."""
        while arr > 0 or rdr > 0:
            done = False
            if arr > 0 and self._active_root_reduction():
                arr -= 1
                done = True
            if rdr > 0 and self._root_degree_reduction():
                rdr -= 1
                done = True
            if not done:
                return

    # -- operations ----------------------------------------------------------

    def insert(self, key: K) -> Item[K]:
        """Insert key and return its handle."""
        it: Item[K] = Item(key)
        other: StrictFibonacciHeap[K] = StrictFibonacciHeap()
        other.root = Node(it)  # @line L1
        other.n = 1
        self.meld(other)
        return it

    def meld(self, other: "StrictFibonacciHeap[K]") -> None:
        """Move all items of other into this heap. other becomes empty."""
        if other.root is None:
            return
        if self.root is None:
            self.root, self.n, self.rec, self.qhead = other.root, other.n, other.rec, other.qhead
            self.aroots, self.lossy = other.aroots, other.lossy
        else:
            big, small = (self, other) if other.n <= self.n else (other, self)
            small.rec.flag = False  # @line L2
            rx, ry = small.root, big.root
            assert rx is not None and ry is not None
            u, v = (rx, ry) if rx.key() < ry.key() else (ry, rx)
            v.parent = None
            self._link_root(v, u)  # @line L3
            vq: Node[K] = v
            v.qprev = v.qnext = v
            q = self._concat(self._concat(small.qhead, vq), big.qhead)
            self.root, self.n, self.rec, self.qhead = u, self.n + other.n, big.rec, q
            self.aroots, self.lossy = big.aroots, big.lossy
            for a in [a for a in self.aroots if a.parent is v or a.parent is u]:
                self._refresh_aroot(a)
            self._reductions(1, 1)  # @line L4
        other.root, other.n, other.qhead = None, 0, None
        other.rec, other.aroots, other.lossy = Record(), set(), set()

    def _link_root(self, v: Node[K], u: Node[K]) -> None:
        v.parent = u
        if v.active:
            u.children.insert(0, v)
        else:
            u.children.append(v)

    def find_min(self) -> K:
        """Return the smallest key, stored at the root."""
        if self.root is None:
            raise IndexError("the heap is empty")
        return self.root.item.key  # @line L5

    def decrease_key(self, it: Item[K], key: K) -> None:
        """Lower the key of the item it to key."""
        if key > it.key:  # type: ignore[operator]
            raise ValueError("new key is larger than the current key")
        it.key = key  # @line L6
        self._after_decrease(it)

    def _after_decrease(self, it: Item[K]) -> None:
        x, z = it.node, self.root
        assert x is not None and z is not None
        if x is z:
            return
        if x.key() < z.key():
            x.item, z.item = z.item, x.item
            x.item.node, z.item.node = x, z
        y = x.parent
        assert y is not None
        was_inner_active = x.active and y.active
        if y is not z:
            self._link(x, z)  # @line L7
        if was_inner_active:  # @line L8
            self._set_loss(x, 0)
            self.aroots.add(x)
            y.rank -= 1
            if not self._is_aroot(y):
                self._set_loss(y, y.loss + 1)
        self._loss_reduction()  # @line L9
        self._reductions(6, 4)

    def delete(self, it: Item[K]) -> None:
        """Delete the item it: decrease its key to minus infinity, then delete-min."""
        it.minus_inf = True
        self._after_decrease(it)
        self.delete_min()

    def delete_min(self) -> K:
        """Remove and return the smallest key."""
        z = self.root
        if z is None:
            raise IndexError("the heap is empty")
        key = z.item.key
        z.item.node = None
        self.n -= 1
        if not z.children:
            self.root = None
            self.qhead = None
            self.aroots, self.lossy = set(), set()
            return key
        x = min(z.children, key=lambda c: c.key())  # @line L10
        if x.active:
            self.aroots.discard(x)
            self.lossy.discard(x)
            x.rec = None
        z.children.remove(x)
        x.parent = None
        for c in x.children:
            if c.active:
                self._set_loss(c, 0)
                self.aroots.add(c)
        for c in list(z.children):
            c.parent = None
            self._link_root(c, x)  # @line L11
            self._refresh_aroot(c)
        z.children = []
        self._q_remove(x)
        self.root = x
        for _ in range(2):  # @line L12
            if self.qhead is None:
                break
            y = self.qhead
            self.qhead = y.qnext
            for c in [c for c in y.children[-2:] if not c.active]:
                self._link(c, x)
                self._refresh_aroot(c)
        self._loss_reduction()  # @line L13
        while self._active_root_reduction() or self._root_degree_reduction():  # @line L14
            pass
        return key

    # -- checking ------------------------------------------------------------

    def nodes(self) -> List[Node[K]]:
        out: List[Node[K]] = []
        stack = [self.root] if self.root is not None else []
        while stack:
            v = stack.pop()
            out.append(v)
            stack.extend(v.children)
        return out

    def check_invariant(self) -> None:
        """Check heap order, the queue, the sets, and the invariants I1 to I4 of the paper."""
        if self.root is None:
            assert self.n == 0 and self.qhead is None
            return
        root = self.root
        nodes = self.nodes()
        assert len(nodes) == self.n and root.parent is None and not root.active
        pos: Dict[Node[K], int] = {}
        if self.qhead is not None:
            v, p = self.qhead, 1
            while True:
                pos[v] = p
                assert v.qnext.qprev is v
                v, p = v.qnext, p + 1
                if v is self.qhead:
                    break
        assert set(pos) == set(nodes) - {root}
        total_loss = 0
        R = 2 * math.log2(self.n) + 6
        for v in nodes:
            assert v.item.node is v
            seen_passive = False
            act = 0
            for c in v.children:
                assert c.parent is v and v.key() < c.key()
                if c.active:
                    assert not seen_passive, "I1: active children must be left of passive ones"
                    act += 1
                else:
                    seen_passive = True
            if v.active:
                assert v.rank == act
                total_loss += v.loss
                assert (v in self.lossy) == (v.loss > 0)
                actives = [c for c in v.children if c.active]
                for i, c in enumerate(reversed(actives), start=1):
                    assert c.rank + c.loss >= i - 1, "I1: rank + loss of the i-th active child"
                if self._is_aroot(v):
                    assert v.loss == 0 and v in self.aroots
            else:
                assert v not in self.aroots and v not in self.lossy
            if v is not root:
                bound = 2 * math.log2(2 * self.n - pos[v]) + 9
                if v.active and v.loss == 0:
                    bound += 1
                assert len(v.children) <= bound, "I4: degree"
        assert all(a.active for a in self.aroots) and all(a.active for a in self.lossy)
        assert len(self.aroots) <= R + 1, "I2"
        assert total_loss <= R + 1, "I3"
        assert len(root.children) <= R + 3, "I4: root degree"
test_strict_fibonacci_heap.py 146 lines
# SPDX-License-Identifier: MIT
# Part of https://algorithm.tamnd.com/strict-fibonacci-heap/. The authors are listed in prov/V45MQ4.jsonl.

import random
from typing import Dict, List

import pytest

from strict_fibonacci_heap import Item, Node, StrictFibonacciHeap


def show(v: Node[int]) -> str:
    """Write a tree as key(children); an active node gets [rank,loss] after its key."""
    s = str(v.item.key) + (f"[{v.rank},{v.loss}]" if v.active else "")
    return s + ("(" + " ".join(show(c) for c in v.children) + ")" if v.children else "")


def queue(h: StrictFibonacciHeap[int]) -> List[int]:
    out: List[int] = []
    v = h.qhead
    while v is not None:
        out.append(v.item.key)
        v = v.qnext
        if v is h.qhead:
            break
    return out


def test_page_example() -> None:
    h: StrictFibonacciHeap[int] = StrictFibonacciHeap()
    items = {}
    for k in [50, 20, 70, 10, 60]:
        items[k] = h.insert(k)
    assert show(h.root) == "10(20(50 70) 60)"  # type: ignore[arg-type]
    items[30] = h.insert(30)
    assert show(h.root) == "10(20[1,0](30[0,0](60) 50 70))"  # type: ignore[arg-type]
    assert queue(h) == [30, 60, 20, 70, 50]
    for k in [80, 40]:
        items[k] = h.insert(k)
    assert show(h.root) == "10(20[1,0](30[0,0](60) 50 70) 80 40)"  # type: ignore[arg-type]
    items[90] = h.insert(90)
    assert show(h.root) == "10(20[2,0](40[1,0](80[0,0](90)) 30[0,0](60) 50) 70)"  # type: ignore[arg-type]
    assert h.counts == {"root-degree": 2, "active-root": 1}
    assert queue(h) == [90, 40, 80, 30, 60, 20, 70, 50]
    h.check_invariant()
    assert h.delete_min() == 10
    assert show(h.root) == "20(40[1,0](80[0,0](90)) 30[0,0](60) 50 70)"  # type: ignore[arg-type]
    assert queue(h) == [80, 30, 60, 70, 50, 90, 40]
    h.check_invariant()
    h.decrease_key(items[80], 15)
    assert h.find_min() == 15
    h.check_invariant()
    assert [h.delete_min() for _ in range(len(h))] == [15, 20, 30, 40, 50, 60, 70, 90]


def test_empty() -> None:
    h: StrictFibonacciHeap[int] = StrictFibonacciHeap()
    h.check_invariant()
    with pytest.raises(IndexError):
        h.find_min()
    with pytest.raises(IndexError):
        h.delete_min()
    h.meld(StrictFibonacciHeap())
    assert len(h) == 0


def test_single_and_larger_key() -> None:
    h: StrictFibonacciHeap[int] = StrictFibonacciHeap()
    it = h.insert(7)
    with pytest.raises(ValueError):
        h.decrease_key(it, 8)
    h.decrease_key(it, 3)
    assert h.find_min() == 3
    assert h.delete_min() == 3
    assert len(h) == 0
    h.check_invariant()


def test_equal_keys_and_delete() -> None:
    h: StrictFibonacciHeap[int] = StrictFibonacciHeap()
    its = [h.insert(5) for _ in range(20)]
    h.delete(its[3])
    h.check_invariant()
    assert [h.delete_min() for _ in range(19)] == [5] * 19


def test_meld_both_ways() -> None:
    a: StrictFibonacciHeap[int] = StrictFibonacciHeap()
    b: StrictFibonacciHeap[int] = StrictFibonacciHeap()
    for k in range(0, 40, 2):
        a.insert(k)
    for k in range(1, 9, 2):
        b.insert(k)
    b.meld(a)  # the larger heap is the argument
    assert len(a) == 0 and len(b) == 24
    b.check_invariant()
    a.check_invariant()
    assert [b.delete_min() for _ in range(24)] == sorted(list(range(0, 40, 2)) + [1, 3, 5, 7])


def test_decrease_root_and_handles_follow_items() -> None:
    h: StrictFibonacciHeap[int] = StrictFibonacciHeap()
    its: Dict[int, Item[int]] = {k: h.insert(k) for k in range(1, 30)}
    h.delete_min()
    h.decrease_key(its[2], 0)
    for k in range(29, 10, -1):
        h.decrease_key(its[k], -k)
        h.check_invariant()
    for k in range(29, 10, -1):
        assert h.delete_min() == -k
    assert h.delete_min() == 0


def test_random() -> None:
    rng = random.Random(2012)
    for _ in range(600):
        h: StrictFibonacciHeap[int] = StrictFibonacciHeap()
        ref: Dict[Item[int], int] = {}
        for _ in range(rng.randint(1, 120)):
            r = rng.random()
            if r < 0.4 or not ref:
                k = rng.randint(0, 1000)
                ref[h.insert(k)] = k
            elif r < 0.7:
                it = rng.choice(list(ref))
                k = ref[it] - rng.randint(0, 500)
                h.decrease_key(it, k)
                ref[it] = k
            elif r < 0.75:
                it = rng.choice(list(ref))
                h.delete(it)
                del ref[it]
            elif r < 0.8:
                o: StrictFibonacciHeap[int] = StrictFibonacciHeap()
                for _ in range(rng.randint(0, 25)):
                    k = rng.randint(0, 1000)
                    ref[o.insert(k)] = k
                h.meld(o)
            else:
                assert h.find_min() == min(ref.values())
                k = h.delete_min()
                it = min(ref, key=lambda i: (ref[i], i.uid))
                assert k == ref.pop(it)
            assert len(h) == len(ref)
            h.check_invariant()
        assert [h.delete_min() for _ in range(len(h))] == sorted(ref.values())

What the next level needs

  • the tests did not run in this build

Read the full page about Strict Fibonacci heap