Swiss table in Python
Cache-friendly open-addressing table that uses SIMD group probing and control bytes.
- File
swiss_table.py, 169 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/swiss-table/. The authors are listed in prov/5PH332.jsonl.
from collections.abc import Callable, Hashable
from typing import Generic, TypeVar
K = TypeVar("K", bound=Hashable)
V = TypeVar("V")
EMPTY = -128 # 0b10000000, as in Abseil
DELETED = -2 # 0b11111110
MASK64 = (1 << 64) - 1
def max_fill(capacity: int) -> int:
"""The most slots that may be full or deleted: 7/8 of the slots, and at least one stays EMPTY."""
return min(capacity * 7 // 8, capacity - 1)
class SwissTable(Generic[K, V]):
"""A map in the style of the Swiss table: groups of slots, one control byte per slot.
A control byte is EMPTY, DELETED, or the 7-bit H2 part of the hash of
the key in the slot. The groups are aligned and probed in the triangular
order g, g+1, g+3, g+6, ... modulo the number of groups, a power of two.
The scan of one group stands for one SIMD compare. Each operation costs
O(capacity) in the worst case and O(1) expected probes in practice.
"""
def __init__(
self,
group_width: int = 16,
groups: int = 1,
hash_fn: Callable[[K], int] = hash,
) -> None:
if group_width < 1:
raise ValueError("group_width must be positive")
if groups < 1 or groups & (groups - 1):
raise ValueError("groups must be a power of two")
self._w = group_width
self._hash = hash_fn
self._alloc(groups)
def _alloc(self, groups: int) -> None:
self._groups = groups
cap = groups * self._w
self._ctrl: list[int] = [EMPTY] * cap
self._keys: list[K | None] = [None] * cap
self._vals: list[V | None] = [None] * cap
self._n = 0
self._dead = 0
def __len__(self) -> int:
return self._n
@property
def capacity(self) -> int:
return len(self._ctrl)
def _split(self, k: K) -> tuple[int, int]:
h = self._hash(k) & MASK64
return h >> 7, h & 0x7F # H1, H2
def _slots(self, g: int) -> range:
return range(g * self._w, (g + 1) * self._w)
def find(self, k: K) -> int | None:
"""Return the slot of k, or None."""
h1, h2 = self._split(k) # @line L1
g, i = h1 % self._groups, 0
while True: # @line L2
for s in self._slots(g): # @line L3: one SIMD compare in a real table
if self._ctrl[s] == h2 and self._keys[s] == k: # @line L4
return s
if any(self._ctrl[s] == EMPTY for s in self._slots(g)): # @line L5
return None
i += 1 # @line L6
g = (g + i) % self._groups
def get(self, k: K, default: V | None = None) -> V | None:
s = self.find(k)
return default if s is None else self._vals[s]
def __contains__(self, k: K) -> bool:
return self.find(k) is not None
def insert(self, k: K, v: V) -> None:
"""Insert k with value v, or replace the value of k."""
s = self.find(k) # @line L7
if s is not None: # @line L8
self._vals[s] = v
return
if self._n + self._dead + 1 > max_fill(self.capacity): # @line L9
self._rebuild()
self._place(k, v)
def _place(self, k: K, v: V) -> None:
"""Lines L10 to L15: put a key that is not present into the first free slot."""
h1, h2 = self._split(k) # @line L10
g, i = h1 % self._groups, 0
while all(self._ctrl[s] >= 0 for s in self._slots(g)): # @line L11
i += 1 # @line L12
g = (g + i) % self._groups
s = next(s for s in self._slots(g) if self._ctrl[s] < 0) # @line L13
if self._ctrl[s] == DELETED: # @line L14
self._dead -= 1
self._ctrl[s], self._keys[s], self._vals[s] = h2, k, v # @line L15
self._n += 1
def erase(self, k: K) -> bool:
"""Remove k. Return True if it was present."""
s = self.find(k) # @line L16
if s is None: # @line L17
return False
g = s // self._w
if any(self._ctrl[t] == EMPTY for t in self._slots(g)): # @line L18
self._ctrl[s] = EMPTY
else:
self._ctrl[s] = DELETED # @line L19
self._dead += 1
self._keys[s] = self._vals[s] = None
self._n -= 1 # @line L20
return True
def _rebuild(self) -> None:
"""Rehash into a table with no tombstones, doubled until n + 1 fits in half of max_fill."""
old = self.items()
groups = self._groups
while 2 * (self._n + 1) > max_fill(groups * self._w): # @line L21
groups *= 2
self._alloc(groups) # @line L22
for k, v in old:
self._place(k, v)
def items(self) -> list[tuple[K, V]]:
return [(self._keys[s], self._vals[s]) for s in range(self.capacity) if self._ctrl[s] >= 0] # type: ignore[misc]
def ctrl(self) -> list[int]:
return list(self._ctrl)
def check_invariant(self) -> None:
"""Check (S1) control bytes and counts, (S2) no EMPTY group before a key's group, (S3) an EMPTY slot."""
n = dead = 0
seen = set()
for s in range(self.capacity):
c = self._ctrl[s]
if c == DELETED:
dead += 1
continue
if c == EMPTY:
continue
assert 0 <= c < 128
k = self._keys[s]
assert k not in seen
seen.add(k)
n += 1
h1, h2 = self._split(k) # type: ignore[arg-type]
assert c == h2
g, i = h1 % self._groups, 0
for _ in range(self._groups):
if g == s // self._w:
break
assert all(self._ctrl[t] != EMPTY for t in self._slots(g))
i += 1
g = (g + i) % self._groups
else:
raise AssertionError("slot not on the probe sequence")
assert n == self._n and dead == self._dead
assert n + dead <= max_fill(self.capacity) and EMPTY in self._ctrl
test_swiss_table.py 109 lines
# SPDX-License-Identifier: MIT
# Part of https://algorithm.tamnd.com/swiss-table/. The authors are listed in prov/5PH332.jsonl.
import random
from swiss_table import DELETED, EMPTY, SwissTable, max_fill
def ident(k: int) -> int:
return k
def layout(t: SwissTable) -> list:
c = t.ctrl()
return ["." if x == EMPTY else "D" if x == DELETED else t._keys[s] for s, x in enumerate(c)]
def test_page_example() -> None:
t = SwissTable(group_width=4, groups=4, hash_fn=ident)
for k in [3, 7, 131, 11, 15, 19, 23, 27]:
t.insert(k, -k)
t.check_invariant()
assert layout(t) == [3, 7, 11, 15, 131, 19, 23, 27] + ["."] * 8
assert t.ctrl()[:8] == [3, 7, 11, 15, 3, 19, 23, 27]
assert t.find(27) == 7 and t.get(27) == -27
assert t.find(259) is None # start group 2 holds an EMPTY slot
assert t.find(1027) is None # H2 = 3 matches 3 and 131, then group 3 stops it
assert t.erase(7)
assert layout(t)[:4] == [3, "D", 11, 15]
assert t.find(27) == 7 # the tombstone keeps group 0 "full"
assert t.erase(27)
assert layout(t)[4:8] == [131, 19, 23, "D"]
t.insert(35, -35)
assert layout(t)[:4] == [3, 35, 11, 15]
assert len(t) == 7 and t._dead == 1
t.check_invariant()
def test_erase_to_empty_when_group_has_empty() -> None:
t = SwissTable(group_width=4, groups=4, hash_fn=ident)
for k in [3, 7]:
t.insert(k, k)
assert t.erase(3)
assert t.ctrl()[0] == EMPTY and t._dead == 0
assert 7 in t and 3 not in t
t.check_invariant()
def test_edges() -> None:
t = SwissTable(group_width=1, groups=1, hash_fn=ident)
assert len(t) == 0 and t.find(5) is None and not t.erase(5)
assert t.get(5, "x") == "x"
t.insert(5, "a")
t.insert(5, "b")
assert len(t) == 1 and t.get(5) == "b"
t.check_invariant()
for bad in [(0, 1), (4, 3), (4, 0)]:
try:
SwissTable(group_width=bad[0], groups=bad[1])
except ValueError:
pass
else:
raise AssertionError(bad)
assert max_fill(16) == 14 and max_fill(4) == 3 and max_fill(1) == 0
def test_all_keys_one_group_start() -> None:
# Every key has the same H1, so they spread over all groups in probe order.
t = SwissTable(group_width=2, groups=8, hash_fn=lambda k: k & 0x7F)
for k in range(13):
t.insert(k, k)
t.check_invariant()
for k in range(13):
assert t.get(k) == k
assert t.capacity == 16
def test_tombstones_force_rebuild_not_growth() -> None:
t = SwissTable(group_width=4, groups=4, hash_fn=ident)
for k in [128 * i for i in range(4)]: # H1 = i, H2 = 0
t.insert(k, k)
cap = t.capacity
for r in range(200):
k = 10_000 + r
t.insert(k, k)
t.erase(k)
t.check_invariant()
assert t.capacity == cap and len(t) == 4
def test_random_against_dict() -> None:
rng = random.Random(2017)
for case in range(600):
w = rng.choice([1, 2, 4, 8, 16])
t: SwissTable = SwissTable(group_width=w, groups=1, hash_fn=(ident if case % 2 else hash))
ref: dict = {}
for _ in range(rng.randint(0, 80)):
k = rng.randrange(60) * rng.choice([1, 128])
op = rng.random()
if op < 0.5:
t.insert(k, case)
ref[k] = case
elif op < 0.8:
assert t.erase(k) == (k in ref)
ref.pop(k, None)
else:
assert t.get(k) == ref.get(k)
t.check_invariant()
assert sorted(t.items()) == sorted(ref.items())
What the next level needs
- the tests did not run in this build