Algorithm Search⌘K

Cuckoo hashing in Python

Two tables with two hash functions and worst-case O(1) lookup, using eviction on insert.

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

import math
import random
from fractions import Fraction
from collections.abc import Callable, Hashable, Iterator
from typing import Any

HashFn = Callable[[Any], int]
Factory = Callable[[int, random.Random], tuple[HashFn, HashFn]]

P = (1 << 61) - 1  # a Mersenne prime


def random_pair(r: int, rng: random.Random) -> tuple[HashFn, HashFn]:
    """Two independent functions k -> ((a * hash(k) + b) mod P) mod r."""

    def make() -> HashFn:
        a, b = rng.randrange(1, P), rng.randrange(P)
        return lambda k: ((a * (hash(k) % P) + b) % P) % r

    return make(), make()


class CuckooHashing:
    """A dictionary stored by cuckoo hashing in two tables of r cells.

    Each key k is in T1[h1(k)] or in T2[h2(k)], so search and delete look at
    two cells: O(1) time in the worst case. An insert moves keys between their
    two cells for at most MaxLoop rounds, then rehashes with new functions.
    The tables keep r >= (1 + eps) * n. With random enough hash functions an
    insert takes O(1) expected amortized time.
    """

    def __init__(
        self,
        r: int = 8,
        eps: float | Fraction = Fraction(1, 10),
        factory: Factory = random_pair,
        rng: random.Random | None = None,
    ) -> None:
        self.eps = Fraction(str(eps)) if isinstance(eps, float) else Fraction(eps)
        if r < 1 or not 0 < self.eps <= 1:
            raise ValueError("need r >= 1 and 0 < eps <= 1")
        self.factory = factory
        self.rng = rng if rng is not None else random.Random()
        self.rehashes = 0
        self._reset(r)

    def _reset(self, r: int) -> None:
        self.r = r
        self.h1, self.h2 = self.factory(r, self.rng)
        self.max_loop = max(1, math.ceil(3 * math.log(r) / math.log(1 + float(self.eps))))
        self.t1: list[tuple[Any, Any] | None] = [None] * r
        self.t2: list[tuple[Any, Any] | None] = [None] * r
        self.n = 0

    def search(self, k: Hashable) -> Any:
        """Return the value of k, or None if k is not a key. Two cells."""
        c = self.t1[self.h1(k)]
        if c is not None and c[0] == k:  # @line L1
            return c[1]
        c = self.t2[self.h2(k)]
        if c is not None and c[0] == k:  # @line L2
            return c[1]
        return None  # @line L3

    def contains(self, k: Hashable) -> bool:
        c1, c2 = self.t1[self.h1(k)], self.t2[self.h2(k)]
        return (c1 is not None and c1[0] == k) or (c2 is not None and c2[0] == k)

    def _place(self, x: tuple[Any, Any]) -> tuple[Any, Any] | None:
        """The eviction walk. Return None if x found a cell, else the pair left over."""
        y: tuple[Any, Any] | None = x  # @line L6
        for _ in range(self.max_loop):  # @line L7
            i = self.h1(y[0])
            y, self.t1[i] = self.t1[i], y  # @line L8
            if y is None:  # @line L9
                self.n += 1
                return None
            j = self.h2(y[0])
            y, self.t2[j] = self.t2[j], y  # @line L10
            if y is None:  # @line L11
                self.n += 1
                return None
        return y

    def insert(self, k: Hashable, v: Any) -> None:
        """Set the value of k to v."""
        i, j = self.h1(k), self.h2(k)
        for t, p in ((self.t1, i), (self.t2, j)):
            c = t[p]
            if c is not None and c[0] == k:  # @line L4
                t[p] = (k, v)
                return
        if self.r < (1 + self.eps) * (self.n + 1):  # @line L5
            self._rehash(2 * self.r, [])
        left = self._place((k, v))
        if left is not None:
            self._rehash(self.r, [left])  # @line L12

    def delete(self, k: Hashable) -> bool:
        """Remove k. Return True if k was a key. Two cells, no tombstone."""
        i = self.h1(k)
        c = self.t1[i]
        if c is not None and c[0] == k:  # @line L13
            self.t1[i] = None
            self.n -= 1
            return True
        j = self.h2(k)
        c = self.t2[j]
        if c is not None and c[0] == k:  # @line L14
            self.t2[j] = None
            self.n -= 1
            return True
        return False

    def _rehash(self, r2: int, extra: list[tuple[Any, Any]]) -> None:
        """Put all pairs and the pairs of extra into new tables of r2 cells."""
        pairs = [c for c in self.t1 + self.t2 if c is not None] + extra  # @line L15
        while True:
            self.rehashes += 1
            self._reset(r2)  # @line L16
            if all(self._place(x) is None for x in pairs):  # @line L17
                return

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

    def items(self) -> Iterator[tuple[Any, Any]]:
        for c in self.t1 + self.t2:
            if c is not None:
                yield c

    def check_invariant(self) -> None:
        """Assert (C1) to (C4) of the page."""
        keys = []
        for i, c in enumerate(self.t1):
            if c is not None:
                assert self.h1(c[0]) == i  # (C1)
                keys.append(c[0])
        for j, c in enumerate(self.t2):
            if c is not None:
                assert self.h2(c[0]) == j  # (C1)
                keys.append(c[0])
        assert len(keys) == len(set(keys))  # (C2)
        assert self.n == len(keys)  # (C3)
        assert self.r >= (1 + self.eps) * self.n  # (C4)
test_cuckoo_hashing.py 76 lines
# SPDX-License-Identifier: MIT
# Part of https://algorithm.tamnd.com/cuckoo-hashing/. The authors are listed in prov/FHYWMC.jsonl.

import random

from cuckoo_hashing import CuckooHashing, random_pair


def example_then_random():
    calls = [0]

    def factory(r, rng):
        calls[0] += 1
        if calls[0] == 1:
            return (lambda k: k % r), (lambda k: (k // r) % r)
        return random_pair(r, rng)

    return factory


def test_page_example() -> None:
    t = CuckooHashing(11, factory=example_then_random(), rng=random.Random(1))
    assert t.max_loop == 76
    for k in [20, 50, 53, 75, 100, 67, 105, 3, 36, 39]:
        t.insert(k, -k)
        t.check_invariant()
    assert t.rehashes == 0 and t.r == 11 and len(t) == 10
    assert [c and c[0] for c in t.t1] == [None, 100, None, 36, None, None, 50, None, None, 75, None]
    assert [c and c[0] for c in t.t2] == [3, 20, None, 39, 53, None, 67, None, None, 105, None]
    assert t.search(67) == -67 and t.search(4) is None
    assert t.delete(67) and t.search(67) is None and not t.delete(67)
    t.check_invariant()


def test_cycle_forces_rehash() -> None:
    t = CuckooHashing(11, factory=example_then_random(), rng=random.Random(2))
    for k in [1, 122, 243]:
        t.insert(k, k)
        t.check_invariant()
    assert t.rehashes >= 1
    assert all(t.search(k) == k for k in [1, 122, 243])


def test_edges() -> None:
    t = CuckooHashing(1, rng=random.Random(3))
    assert t.search("x") is None and not t.delete("x") and len(t) == 0
    t.insert("x", 1)
    t.insert("x", 2)
    assert len(t) == 1 and t.search("x") == 2
    t.check_invariant()
    for k in [2**63 - 1, -(2**63), -1, 0, 2**64]:
        t.insert(k, k)
        t.check_invariant()
    assert all(t.search(k) == k for k in [2**63 - 1, -(2**63), -1, 0, 2**64])


def test_random_against_dict() -> None:
    rng = random.Random(777)
    for _ in range(600):
        t = CuckooHashing(rng.randint(1, 6), rng=random.Random(rng.random()))
        model: dict[int, int] = {}
        for _ in range(rng.randint(0, 60)):
            k = rng.randint(-30, 30)
            op = rng.random()
            if op < 0.5:
                v = rng.randint(0, 99)
                t.insert(k, v)
                model[k] = v
            elif op < 0.75:
                assert t.delete(k) == (k in model)
                model.pop(k, None)
            else:
                assert t.search(k) == model.get(k)
            t.check_invariant()
            assert len(t) == len(model)
        assert dict(t.items()) == model

What the next level needs

  • the tests did not run in this build

Read the full page about Cuckoo hashing