Concurrent hash table in Python
Hash table supporting concurrent insert, lookup and delete, typically with lock striping or lock-free buckets.
- File
concurrent_hash_table.py, 161 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/concurrent-hash-table/. The authors are listed in prov/HQKVBH.jsonl.
import threading
import time
from collections.abc import Generator, Hashable
from typing import Generic, TypeVar
K = TypeVar("K", bound=Hashable)
V = TypeVar("V")
R = TypeVar("R")
MISSING = object() # the result of get or remove when the key is absent
SPIN = "spin" # what a step generator yields while it waits for a lock
Steps = Generator[str | None, None, R]
def acquire(lock: threading.Lock) -> Generator[str | None, None, None]:
"""Take the lock, one try per step; yield SPIN while another holder has it."""
while not lock.acquire(blocking=False):
yield SPIN
def run(steps: "Steps[R]") -> R:
"""Run a step generator to its end in the calling thread and return its result."""
while True:
try:
if next(steps) == SPIN:
time.sleep(0)
except StopIteration as stop:
return stop.value
class ConcurrentHashTable(Generic[K, V]):
"""A chained hash table with lock striping, as in the StripedHashSet of Herlihy and Shavit.
There are L locks. The bucket array has a capacity that is always a
multiple of L, so bucket b is guarded by lock b mod L, and a key k with
hash h is guarded by lock h mod L whatever the capacity is. put, get and
remove hold one lock and take O(1 + chain length) steps; resize takes
all L locks in index order and doubles the capacity. The table is
linearizable and deadlock-free.
The *_steps methods are generators that yield after each step on shared
memory, for tests that choose the interleaving; the plain methods run the
same steps and are safe to call from threads.
"""
def __init__(self, stripes: int = 4, capacity: int = 4, max_load: float = 2.0) -> None:
if stripes < 1 or capacity < stripes or capacity % stripes:
raise ValueError("capacity must be a positive multiple of stripes")
self.locks = [threading.Lock() for _ in range(stripes)]
self.buckets: list[list[tuple[K, V]]] = [[] for _ in range(capacity)]
self.counts = [0] * stripes # counts[i]: pairs in the stripe of lock i
self.max_load = max_load
self.resizes = 0
def _stripe(self, k: K) -> int:
return hash(k) % len(self.locks)
def put_steps(self, k: K, v: V) -> "Steps[object]":
"""Map k to v; return the old value, or MISSING if k was absent."""
i = self._stripe(k)
yield from acquire(self.locks[i]) # @line L1
table = self.buckets # @line L2
b = table[hash(k) % len(table)]
yield None
old: object = MISSING
for j, (key, val) in enumerate(b): # @line L3
if key == k:
old = val
b[j] = (k, v) # @line L4
break
else:
b.append((k, v)) # @line L5
self.counts[i] += 1
yield None
cap = len(table)
self.locks[i].release() # @line L6
if sum(self.counts) > self.max_load * cap: # @line L7
yield from self.resize_steps(cap) # @line L8
return old # @line L9
def get_steps(self, k: K) -> "Steps[object]":
"""Return the value of k, or MISSING."""
i = self._stripe(k)
yield from acquire(self.locks[i]) # @line L10
table = self.buckets
b = table[hash(k) % len(table)]
yield None
res: object = MISSING
for key, val in b: # @line L11
if key == k:
res = val
break
self.locks[i].release() # @line L12
return res
def remove_steps(self, k: K) -> "Steps[object]":
"""Delete k; return its old value, or MISSING if it was absent."""
i = self._stripe(k)
yield from acquire(self.locks[i])
table = self.buckets
b = table[hash(k) % len(table)]
yield None
res: object = MISSING
for j, (key, val) in enumerate(b):
if key == k:
res = val
del b[j]
self.counts[i] -= 1
break
yield None
self.locks[i].release()
return res
def resize_steps(self, old_cap: int) -> "Steps[None]":
"""Double the capacity, unless another thread resized since the caller saw old_cap."""
for lock in self.locks: # @line L13
yield from acquire(lock)
try:
if len(self.buckets) == old_cap: # @line L14
new: list[list[tuple[K, V]]] = [[] for _ in range(2 * old_cap)]
for b in self.buckets: # @line L15
for k, v in b:
new[hash(k) % len(new)].append((k, v))
self.buckets = new # @line L16
self.resizes += 1
yield None
finally:
for lock in self.locks: # @line L17
lock.release()
def put(self, k: K, v: V) -> object:
return run(self.put_steps(k, v))
def get(self, k: K) -> object:
return run(self.get_steps(k))
def remove(self, k: K) -> object:
return run(self.remove_steps(k))
def to_dict(self) -> dict[K, V]:
"""The contents, read at a quiet moment."""
return {k: v for b in self.buckets for k, v in b}
def check_invariant(self) -> None:
"""At a quiet moment: each pair is in its bucket, keys are unique, counts are right."""
cap, n = len(self.buckets), len(self.locks)
assert cap % n == 0
counts = [0] * n
seen = set()
for idx, b in enumerate(self.buckets):
for k, _ in b:
assert hash(k) % cap == idx and k not in seen
assert idx % n == hash(k) % n # the bucket and the key share a stripe
seen.add(k)
counts[hash(k) % n] += 1
assert counts == self.counts
assert not any(lock.locked() for lock in self.locks)
test_concurrent_hash_table.py 211 lines
# SPDX-License-Identifier: MIT
# Part of https://algorithm.tamnd.com/concurrent-hash-table/. The authors are listed in prov/HQKVBH.jsonl.
import random
import threading
import pytest
from concurrent_hash_table import MISSING, ConcurrentHashTable
# ---- deterministic interleaving driver and linearizability check ----
Op = tuple # (thread, name, arg, result, invoke time, response time)
def interleave(threads, schedule, limit=100000):
"""Run the operations of each thread, one atomic step at a time.
threads[t] is a list of (name, arg, make) where make() returns the step
generator of the operation. schedule is a sequence of thread numbers; a
pick of a thread with no work left is skipped. After the schedule, the
remaining steps run round robin. Returns the history as a list of Op.
"""
todo = [list(reversed(ops)) for ops in threads]
cur: list = [None] * len(threads)
history: list = []
clock = 0
def step(t):
nonlocal clock
if cur[t] is None:
if not todo[t]:
return False
name, arg, make = todo[t].pop()
cur[t] = (name, arg, make(), clock)
name, arg, gen, inv = cur[t]
clock += 1
try:
next(gen)
except StopIteration as stop:
history.append((t, name, arg, stop.value, inv, clock))
cur[t] = None
return True
for t in schedule:
step(t)
steps = 0
while any(cur) or any(todo):
for t in range(len(threads)):
step(t)
steps += 1
assert steps < limit, "no progress"
return history
def linearizable(history, init, apply):
"""Wing and Gong search: is there a legal order that respects real time?
apply(state, name, arg) returns (new state, result); states are hashable.
"""
n = len(history)
full = (1 << n) - 1
seen = set()
def dfs(done, state):
if done == full:
return True
if (done, state) in seen:
return False
seen.add((done, state))
for i in range(n):
if done >> i & 1:
continue
inv_i = history[i][4]
if any(not done >> j & 1 and history[j][5] < inv_i for j in range(n)):
continue
s2, r = apply(state, history[i][1], history[i][2])
if r == history[i][3] and dfs(done | 1 << i, s2):
return True
return False
return dfs(0, init)
def map_spec(state, name, arg):
d = dict(state)
if name == "put":
k, v = arg
old = d.get(k, MISSING)
d[k] = v
elif name == "get":
return state, d.get(arg, MISSING)
else:
old = d.pop(arg, MISSING)
return tuple(sorted(d.items())), old
def make_ops(t, plan):
ops = []
for name, arg in plan:
if name == "put":
ops.append((name, arg, lambda a=arg: t.put_steps(*a)))
elif name == "get":
ops.append((name, arg, lambda a=arg: t.get_steps(a)))
else:
ops.append((name, arg, lambda a=arg: t.remove_steps(a)))
return ops
def test_example() -> None:
t: ConcurrentHashTable[int, str] = ConcurrentHashTable(stripes=2, capacity=2, max_load=1.0)
assert t.put(4, "a") is MISSING
assert t.put(7, "b") is MISSING
assert t.buckets == [[(4, "a")], [(7, "b")]]
# keys 6 and 4 are in stripe 0, key 9 in stripe 1
a = make_ops(t, [("put", (6, "c"))])
b = make_ops(t, [("put", (9, "d"))])
c = make_ops(t, [("get", 4)])
h = interleave([a, b, c], [0, 1, 2, 0, 1, 2, 2, 0, 0, 0, 0, 1, 1, 1, 1, 1, 1])
res = {name: r for _, name, _, r, _, _ in h}
assert res["get"] == "a"
assert t.resizes == 1
assert t.buckets == [[(4, "a")], [(9, "d")], [(6, "c")], [(7, "b")]]
assert linearizable(h, ((4, "a"), (7, "b")), map_spec)
t.check_invariant()
def test_edges() -> None:
with pytest.raises(ValueError):
ConcurrentHashTable(stripes=4, capacity=6)
t: ConcurrentHashTable[str, int] = ConcurrentHashTable(stripes=1, capacity=1)
assert t.get("x") is MISSING and t.remove("x") is MISSING
assert t.put("x", 1) is MISSING
assert t.put("x", 2) == 1
assert t.get("x") == 2
assert t.remove("x") == 2
assert t.get("x") is MISSING
t.check_invariant()
def test_same_stripe_waits_other_stripe_runs() -> None:
t: ConcurrentHashTable[int, int] = ConcurrentHashTable(stripes=2, capacity=2, max_load=100)
g = t.put_steps(0, 0)
next(g) # g holds lock 0
assert t.put(1, 1) is MISSING # stripe 1 is free
w = t.get_steps(2) # stripe 0: must wait
assert all(next(w) == "spin" for _ in range(5))
for _ in g:
pass
assert next(w) is None
for _ in w:
pass
t.check_invariant()
def test_random_against_model() -> None:
rng = random.Random(1234)
for _ in range(500):
stripes = rng.choice([1, 2, 4])
t: ConcurrentHashTable[int, int] = ConcurrentHashTable(stripes, stripes * rng.randint(1, 2), rng.choice([0.5, 1.0, 3.0]))
model: dict[int, int] = {}
for _ in range(rng.randint(0, 60)):
k, op = rng.randint(-10, 10), rng.random()
if op < 0.5:
v = rng.randint(0, 99)
assert t.put(k, v) == model.get(k, MISSING)
model[k] = v
elif op < 0.75:
assert t.get(k) == model.get(k, MISSING)
else:
assert t.remove(k) == model.pop(k, MISSING)
t.check_invariant()
assert t.to_dict() == model
def test_random_interleavings_are_linearizable() -> None:
rng = random.Random(777)
for _ in range(500):
t: ConcurrentHashTable[int, int] = ConcurrentHashTable(2, 2, rng.choice([0.5, 1.0, 9.0]))
for k in rng.sample(range(6), rng.randint(0, 3)):
t.put(k, -1)
init = tuple(sorted(t.to_dict().items()))
threads = []
for th in range(3):
plan = []
for i in range(rng.randint(1, 3)):
k = rng.randrange(6)
plan.append(rng.choice([("put", (k, 10 * th + i)), ("get", k), ("remove", k)]))
threads.append(make_ops(t, plan))
h = interleave(threads, [rng.randrange(3) for _ in range(rng.randint(0, 80))])
assert linearizable(h, init, map_spec)
t.check_invariant()
def test_threads() -> None:
t: ConcurrentHashTable[int, int] = ConcurrentHashTable(stripes=4, capacity=4, max_load=1.0)
def work(w: int) -> None:
for i in range(300):
assert t.put(4 * i + w, w) is MISSING
for i in range(0, 300, 2):
assert t.remove(4 * i + w) == w
ts = [threading.Thread(target=work, args=(w,)) for w in range(4)]
for x in ts:
x.start()
for x in ts:
x.join()
t.check_invariant()
assert t.to_dict() == {4 * i + w: w for w in range(4) for i in range(1, 300, 2)}
assert t.resizes >= 1
What the next level needs
- the tests did not run in this build