crates/admit/src/table.rs
//! **The consumer table**: rolling-window bytes per consumer, multi-core, with
//! no global lock anywhere on the path.
//!
//! # Shape
//!
//! * **Sharded by consumer hash**, one shard per core (rounded up to a power
//! of two), each cache-line aligned so two cores writing two shards never
//! share a line.
//! * **Preallocated.** Every shard is a fixed open-addressing array of
//! `capacity` slots allocated once at construction; nothing on the path
//! allocates, and the steady state never grows or rehashes.
//! * **A slot is one hash word and `window_days` bucket words**, all
//! `AtomicU64`. A bucket packs its UTC day (low [`DAY_BITS`]) and its bytes
//! (high [`BYTE_BITS`], saturating), so a charge is ONE compare-and-swap and a
//! 30-day sum is 30 loads — no lock, no second word to keep in step.
//!
//! # Who locks what
//!
//! * **Reading an existing consumer** — the plan's usage — is a probe of hash
//! words and a pass over its buckets: loads only.
//! * **Charging an existing consumer** — `record` — is the same probe and one
//! CAS on today's bucket.
//! * **A new consumer** takes its shard's insert lock (a per-SHARD mutex, so
//! it serialises only against other first-sights hashing to the same shard,
//! never against readers). Its buckets are filled BEFORE its hash word is
//! published with `Release`, so a reader that sees the hash sees the buckets.
//! * **Eviction** runs only inside that insert lock, when the shard is past
//! three quarters full: a consumer none of whose buckets is inside the window
//! any more is tombstoned. Off the hot path by construction — it costs the
//! insert that needed the room, never a read or a charge.
//!
//! # The identity is a 64-bit keyed hash
//!
//! A consumer is stored as `SipHash-1-3(random per-process key, consumer)`
//! (std's `RandomState`), not as its name: the table holds no string and
//! allocates none. Two consumers colliding on 64 bits among the table's
//! (bounded) population is ~2^-35 for 16 384 of them, and the only
//! consequence would be two tenants sharing one usage figure. The per-process
//! key means nobody outside can aim a collision.
//!
//! # Honest limits
//!
//! * **Full is fail-OPEN.** A shard with no free slot after eviction does not
//! track the newcomer: its usage reads zero and its charges are dropped. A
//! flood of names cannot grow the table, and nobody already tracked loses
//! their history.
//! * **A charge racing its own consumer's eviction** can land in a slot that
//! was just tombstoned (lost) or, if the slot is re-used within that window
//! by another consumer, in the wrong one. Eviction only takes a consumer idle
//! for the whole window, so this needs a transfer that recorded nothing for
//! 30 days to finish in the microseconds of a sweep.
use std::collections::hash_map::RandomState;
use std::hash::{BuildHasher, Hash};
use std::sync::atomic::{AtomicU64, AtomicUsize, Ordering};
use std::sync::Mutex;
/// Bits of a bucket word holding the day (≈ 2 870 years of distinct days).
pub const DAY_BITS: u32 = 20;
/// Bits holding the bytes: 16 TiB per consumer per day, saturating.
pub const BYTE_BITS: u32 = 64 - DAY_BITS;
const DAY_MASK: u64 = (1 << DAY_BITS) - 1;
const BYTE_MAX: u64 = (1 << BYTE_BITS) - 1;
const EMPTY: u64 = 0;
const TOMB: u64 = u64::MAX;
fn pack(day: u64, bytes: u64) -> u64 {
(bytes.min(BYTE_MAX) << DAY_BITS) | (day & DAY_MASK)
}
/// The full day a bucket holds, read relative to `reference` (within ±2^19 days).
fn day_of(word: u64, reference: u64) -> u64 {
let half = 1u64 << (DAY_BITS - 1);
let diff = ((word & DAY_MASK).wrapping_sub(reference & DAY_MASK).wrapping_add(half)) & DAY_MASK;
(reference + diff).wrapping_sub(half)
}
fn bytes_of(word: u64) -> u64 {
word >> DAY_BITS
}
#[repr(align(64))]
struct Shard {
insert: Mutex<()>,
live: AtomicUsize,
hashes: Box<[AtomicU64]>,
buckets: Box<[AtomicU64]>,
}
/// Rolling-window bytes per consumer hash. See the module docs.
pub struct Table {
shards: Box<[Shard]>,
shard_bits: u32,
capacity: usize,
window: usize,
keys: RandomState,
}
impl std::fmt::Debug for Table {
fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
f.debug_struct("Table")
.field("shards", &self.shards.len())
.field("capacity", &self.capacity)
.field("window", &self.window)
.finish()
}
}
/// One shard per core, rounded up to a power of two.
pub fn default_shards() -> usize {
std::thread::available_parallelism()
.map(|n| n.get())
.unwrap_or(1)
.next_power_of_two()
}
impl Table {
/// A table of about `consumers` slots over `shards` shards, each consumer
/// holding `window_days` day buckets. Everything is allocated here.
pub fn new(consumers: usize, window_days: u32, shards: usize) -> Table {
let shards = shards.max(1).next_power_of_two();
let capacity = consumers.div_ceil(shards).max(8);
let window = window_days.max(1) as usize;
let shard = || Shard {
insert: Mutex::new(()),
live: AtomicUsize::new(0),
hashes: (0..capacity).map(|_| AtomicU64::new(EMPTY)).collect(),
buckets: (0..capacity * window).map(|_| AtomicU64::new(0)).collect(),
};
Table {
shards: (0..shards).map(|_| shard()).collect(),
shard_bits: shards.trailing_zeros(),
capacity,
window,
keys: RandomState::new(),
}
}
pub fn window_days(&self) -> u32 {
self.window as u32
}
/// Consumers this table can hold.
pub fn capacity(&self) -> usize {
self.capacity * self.shards.len()
}
/// Consumers held now.
pub fn len(&self) -> usize {
self.shards.iter().map(|s| s.live.load(Ordering::Relaxed)).sum()
}
pub fn is_empty(&self) -> bool {
self.len() == 0
}
/// The keyed 64-bit identity of anything hashable, never EMPTY or TOMB.
pub fn key<T: Hash + ?Sized>(&self, what: &T) -> u64 {
Self::fix(self.keys.hash_one(what))
}
/// A hash taken elsewhere (a day token's own keyed bytes), made valid.
pub fn fix(h: u64) -> u64 {
match h {
EMPTY => 1,
TOMB => TOMB - 1,
h => h,
}
}
fn shard(&self, h: u64) -> &Shard {
let i = if self.shard_bits == 0 { 0 } else { (h >> (64 - self.shard_bits)) as usize };
&self.shards[i]
}
/// Lock-free probe: the slot index holding `h` in its shard.
fn find(&self, shard: &Shard, h: u64) -> Option<usize> {
let start = (h as usize) % self.capacity;
for step in 0..self.capacity {
let i = (start + step) % self.capacity;
match shard.hashes[i].load(Ordering::Acquire) {
EMPTY => return None,
x if x == h => return Some(i),
_ => {}
}
}
None
}
fn bucket<'a>(&self, shard: &'a Shard, slot: usize, day: u64) -> &'a AtomicU64 {
&shard.buckets[slot * self.window + (day % self.window as u64) as usize]
}
/// Whether `h` is held.
pub fn contains(&self, h: u64) -> bool {
let shard = self.shard(h);
self.find(shard, h).is_some()
}
/// **Bytes charged to `h` over the window ending on (and including)
/// `today`**, or `None` when `h` is not held. Loads only.
pub fn sum(&self, h: u64, today: u64) -> Option<u64> {
let shard = self.shard(h);
let slot = self.find(shard, h)?;
let oldest = today.saturating_sub(self.window as u64 - 1);
let base = slot * self.window;
let total = shard.buckets[base..base + self.window]
.iter()
.map(|b| b.load(Ordering::Acquire))
.filter(|w| *w != 0)
.filter(|w| {
let d = day_of(*w, today);
d >= oldest && d <= today
})
.fold(0u64, |acc, w| acc.saturating_add(bytes_of(w)));
// The slot may have been evicted and re-used while it was read.
(shard.hashes[slot].load(Ordering::Acquire) == h).then_some(total)
}
/// Bytes charged to `h` on exactly `day`, or `None` when not held.
pub fn on(&self, h: u64, day: u64) -> Option<u64> {
let shard = self.shard(h);
let slot = self.find(shard, h)?;
let w = self.bucket(shard, slot, day).load(Ordering::Acquire);
Some(if w != 0 && day_of(w, day) == day { bytes_of(w) } else { 0 })
}
/// **Charge `bytes` to `h` on `day`** — one CAS on an existing consumer.
/// A day older than the one its bucket already holds (a whole window
/// later) is outside that day's window and is dropped. Returns `false`
/// when `h` is not held (see [`Table::insert`]).
pub fn add(&self, h: u64, day: u64, bytes: u64) -> bool {
let shard = self.shard(h);
let Some(slot) = self.find(shard, h) else { return false };
let bucket = self.bucket(shard, slot, day);
let mut cur = bucket.load(Ordering::Acquire);
loop {
let held = if cur == 0 { None } else { Some(day_of(cur, day)) };
let next = match held {
Some(d) if d == day => pack(day, bytes_of(cur).saturating_add(bytes)),
Some(d) if d > day => return true,
_ => pack(day, bytes),
};
match bucket.compare_exchange_weak(cur, next, Ordering::AcqRel, Ordering::Acquire) {
Ok(_) => return true,
Err(now) => cur = now,
}
}
}
/// **First sight of `h`**: insert it, its buckets seeded from `seed`
/// (`(day, bytes)` pairs — a durable ledger's closed days, for a product
/// that has one) BEFORE its hash is published. Takes the shard's insert
/// lock; evicts idle consumers first when the shard is ¾ full. Returns
/// `false` when the shard is full (fail-open, see the module docs). An
/// `h` that another thread inserted meanwhile is left as it is.
pub fn insert(&self, h: u64, today: u64, seed: &[(u64, u64)]) -> bool {
let shard = self.shard(h);
let _held = shard.insert.lock().unwrap_or_else(|e| e.into_inner());
if self.find(shard, h).is_some() {
return true;
}
if shard.live.load(Ordering::Relaxed) * 4 >= self.capacity * 3 {
self.sweep(shard, today);
}
let start = (h as usize) % self.capacity;
let free = (0..self.capacity)
.map(|step| (start + step) % self.capacity)
.find(|i| matches!(shard.hashes[*i].load(Ordering::Acquire), EMPTY | TOMB));
let Some(slot) = free else { return false };
let base = slot * self.window;
for b in &shard.buckets[base..base + self.window] {
b.store(0, Ordering::Relaxed);
}
let oldest = today.saturating_sub(self.window as u64 - 1);
for &(day, bytes) in seed {
if day >= oldest && day <= today && bytes > 0 {
let b = self.bucket(shard, slot, day);
let cur = b.load(Ordering::Relaxed);
let have = if cur != 0 && day_of(cur, day) == day { bytes_of(cur) } else { 0 };
b.store(pack(day, have.saturating_add(bytes)), Ordering::Relaxed);
}
}
shard.hashes[slot].store(h, Ordering::Release);
shard.live.fetch_add(1, Ordering::Relaxed);
true
}
/// Tombstone every consumer in `shard` with no bucket inside the window.
/// Called only under the shard's insert lock.
fn sweep(&self, shard: &Shard, today: u64) {
let oldest = today.saturating_sub(self.window as u64 - 1);
for slot in 0..self.capacity {
let h = shard.hashes[slot].load(Ordering::Acquire);
if matches!(h, EMPTY | TOMB) {
continue;
}
let base = slot * self.window;
let active = shard.buckets[base..base + self.window].iter().any(|b| {
let w = b.load(Ordering::Acquire);
w != 0 && day_of(w, today) >= oldest
});
if !active {
shard.hashes[slot].store(TOMB, Ordering::Release);
shard.live.fetch_sub(1, Ordering::Relaxed);
}
}
}
}
#[cfg(test)]
mod tests {
use super::*;
const D: u64 = 20_000;
fn t(window: u32) -> Table {
Table::new(64, window, 4)
}
#[test]
fn a_bucket_word_round_trips_day_and_bytes() {
let w = pack(D, 12_345);
assert_eq!((day_of(w, D), bytes_of(w)), (D, 12_345));
assert_eq!(day_of(w, D + 29), D, "read from later in the window");
assert_eq!(day_of(pack(D + 5, 1), D), D + 5, "a NEWER bucket reads as newer");
assert_eq!(bytes_of(pack(D, u64::MAX)), BYTE_MAX, "saturates");
assert_eq!(day_of(pack(D, 1), D + (1 << DAY_BITS) - 3), D + (1 << DAY_BITS), "wrap is relative");
}
#[test]
fn a_window_is_today_and_the_days_before_it() {
let t = t(30);
let h = t.key("a");
assert!(t.insert(h, D, &[]));
t.add(h, D, 10);
t.add(h, D - 29, 5); // the oldest day still inside
t.add(h, D - 30, 1_000); // too old — and shares D's slot, which it must not evict
assert_eq!(t.sum(h, D), Some(15));
assert_eq!(t.sum(t.key("b"), D), None);
}
#[test]
fn days_fall_out_as_the_window_moves() {
let t = t(30);
let h = t.key("a");
t.insert(h, D, &[]);
for d in 0..30 {
t.add(h, D + d, 1);
}
assert_eq!(t.sum(h, D + 29), Some(30));
assert_eq!(t.sum(h, D + 30), Some(29), "day D left the window");
assert_eq!(t.sum(h, D + 58), Some(1));
assert_eq!(t.sum(h, D + 59), Some(0));
}
/// ★ A slot reused by a later day starts from zero; the stale day is not summed.
#[test]
fn a_reused_bucket_does_not_carry_the_old_day() {
let t = t(30);
let h = t.key("a");
t.insert(h, D, &[]);
t.add(h, D, 100);
t.add(h, D + 30, 7);
assert_eq!(t.sum(h, D + 30), Some(7));
t.add(h, D + 400, 1);
assert_eq!(t.sum(h, D + 400), Some(1));
assert_eq!(t.on(h, D + 400), Some(1));
assert_eq!(t.on(h, D + 399), Some(0));
}
#[test]
fn a_seed_fills_the_window_before_the_consumer_is_visible() {
let t = t(30);
let h = t.key("ns");
assert!(!t.add(h, D, 1), "not held yet");
t.insert(h, D, &[(D - 40, 999), (D - 29, 4), (D - 1, 5), (D, 6), (D + 1, 777)]);
assert_eq!(t.sum(h, D), Some(15), "only days inside the window are seeded");
t.insert(h, D, &[(D, 1_000_000)]);
assert_eq!(t.sum(h, D), Some(15), "a second insert leaves it as it is");
}
#[test]
fn a_full_shard_evicts_the_idle_and_fails_open_when_nobody_is() {
let t = Table::new(8, 30, 1); // one shard, 8 slots
let names: Vec<u64> = (0..8).map(|i| t.key(&i)).collect();
for h in &names {
assert!(t.insert(*h, D, &[]));
t.add(*h, D, 1);
}
assert!(!t.insert(t.key(&100), D, &[]), "full, all active: fail open");
// A window later, all eight are idle; the next newcomer sweeps them.
let late = D + 30;
assert!(t.insert(t.key(&101), late, &[]));
assert_eq!(t.len(), 1);
assert_eq!(t.sum(names[0], late), None, "evicted");
}
/// ★ Many threads charging one consumer lose nothing.
#[test]
fn concurrent_charges_are_all_counted() {
let t = std::sync::Arc::new(t(30));
let h = t.key("hot");
t.insert(h, D, &[]);
let threads: Vec<_> = (0..8)
.map(|_| {
let t = std::sync::Arc::clone(&t);
std::thread::spawn(move || {
for _ in 0..10_000 {
t.add(h, D, 3);
}
})
})
.collect();
for th in threads {
th.join().unwrap();
}
assert_eq!(t.sum(h, D), Some(8 * 10_000 * 3));
}
}