50 lines
1.2 KiB
Python
50 lines
1.2 KiB
Python
"""Rendezvous (highest random weight) hashing."""
|
|
from __future__ import annotations
|
|
|
|
import hashlib
|
|
from typing import Any, Iterable
|
|
|
|
|
|
def _weight(key: Any, node: Any) -> int:
|
|
h = hashlib.sha256(repr(key).encode() + b"\x00" + repr(node).encode()).digest()
|
|
return int.from_bytes(h[:8], "big")
|
|
|
|
|
|
def _nodes(nodes: Iterable[Any]) -> list[Any]:
|
|
seen = []
|
|
seen_set = set()
|
|
for n in nodes:
|
|
if n not in seen_set:
|
|
seen_set.add(n)
|
|
seen.append(n)
|
|
return seen
|
|
|
|
|
|
def rendezvous(key: Any, nodes: Iterable[Any]) -> Any | None:
|
|
ns = _nodes(nodes)
|
|
if not ns:
|
|
return None
|
|
best = None
|
|
best_w = -1
|
|
for n in ns:
|
|
w = _weight(key, n)
|
|
if w > best_w:
|
|
best_w = w
|
|
best = n
|
|
return best
|
|
|
|
|
|
def rendezvous_topk(key: Any, nodes: Iterable[Any], k: int) -> list[Any]:
|
|
ns = _nodes(nodes)
|
|
if k <= 0 or not ns:
|
|
return []
|
|
indexed = sorted(range(len(ns)), key=lambda i: (-_weight(key, ns[i]), i))
|
|
return [ns[i] for i in indexed[:k]]
|
|
|
|
|
|
def weights(key: Any, nodes: Iterable[Any]) -> list[tuple[Any, int]]:
|
|
ns = _nodes(nodes)
|
|
out = [(n, _weight(key, n)) for n in ns]
|
|
out.sort(key=lambda t: (-t[1], ns.index(t[0])))
|
|
return out
|