A hazard-pointer-protected lock-free read path will show higher…

import json
import sys
import time
import threading
import statistics
import random
from collections import deque

# ---------------------------------------------------------------------------
# Shared read-mostly data structure with two reclamation strategies.
#
# We simulate a lock-free read path over a shared dict-like structure.
# Readers perform lookups on a snapshot of the current "table".
# A writer periodically replaces the table (reclamation event).
#
# Strategy A: Hazard pointers
# - Each reader registers the table it is currently using in a per-thread
# hazard slot. The writer, before freeing an old table, scans all hazard
# slots and only frees tables that are not currently protected.
# - This allows immediate reclamation once no reader holds the old table.
#
# Strategy B: RCU-style epoch / grace period
# - Readers increment an epoch counter on entry, decrement on exit.
# - Writer waits for a grace period (all readers that were active at the
# time of the swap to finish) before freeing the old table.
# - This introduces a grace-period delay before memory is reclaimed.
#
# The workload is CPU-only: readers do many lookups, writer swaps tables.
# We measure read throughput (ops/sec) for each strategy.
# ---------------------------------------------------------------------------

class HazardPointerTable:
 """Shared table protected by hazard pointers."""

 def __init__(self, size=1024):
 self.size = size
 self._lock = threading.Lock()
 self._table = self._make_table()
 # hazard slots: one per reader thread (indexed by thread id)
 self._hazard = {}
 self._hazard_lock = threading.Lock()
 # retired tables waiting for reclamation
 self._retired = deque()
 self._retired_lock = threading.Lock()
 self.reclaimed_count = 0

 def _make_table(self):
 return {i: i * 2 for i in range(self.size)}

 def register_reader(self, tid):
 with self._hazard_lock:
 self._hazard[tid] = None

 def unregister_reader(self, tid):
 with self._hazard_lock:
 self._hazard.pop(tid, None)

 def acquire(self, tid):
 """Reader acquires a protected reference to the current table."""
 while True:
 with self._lock:
 tbl = self._table
 with self._hazard_lock:
 self._hazard[tid] = tbl
 # re-check: table may have changed between read and hazard store
 with self._lock:
 if self._table is tbl:
 return tbl
 # retry

 def release(self, tid):
 with self._hazard_lock:
 self._hazard[tid] = None

 def lookup(self, tbl, key):
 return tbl.get(key)

 def swap(self):
 """Writer swaps in a new table and retires the old one."""
 new_tbl = self._make_table()
 with self._lock:
 old = self._table
 self._table = new_tbl
 with self._retired_lock:
 self._retired.append(old)
 self._try_reclaim()

 def _try_reclaim(self):
 """Free retired tables that are not protected by any hazard pointer."""
 with self._hazard_lock:
 protected = set(id(t) for t in self._hazard.values() if t is not None)
 with self._retired_lock:
 remaining = deque()
 while self._retired:
 t = self._retired.popleft()
 if id(t) in protected:
 remaining.append(t)
 else:
 self.reclaimed_count += 1
 self._retired = remaining

class RCUTable:
 """Shared table protected by RCU-style epoch / grace period."""

 def __init__(self, size=1024):
 self.size = size
 self._lock = threading.Lock()
 self._table = self._make_table()
 # epoch counters per reader
 self._epochs = {}
 self._epoch_lock = threading.Lock()
 self._retired = deque()
 self._retired_lock = threading.Lock()
 self.reclaimed_count = 0
 self._writer_lock = threading.Lock()

 def _make_table(self):
 return {i: i * 2 for i in range(self.size)}

 def register_reader(self, tid):
 with self._epoch_lock:
 self._epochs[tid] = 0

 def unregister_reader(self, tid):
 with self._epoch_lock:
 self._epochs.pop(tid, None)

 def acquire(self, tid):
 with self._epoch_lock:
 self._epochs[tid] = self._epochs.get(tid, 0) + 1
 with self._lock:
 return self._table

 def release(self, tid):
 with self._epoch_lock:
 self._epochs[tid] = max(0, self._epochs.get(tid, 0) - 1)

 def lookup(self, tbl, key):
 return tbl.get(key)

 def swap(self):
 """Writer swaps in a new table, then waits for a grace period."""
 with self._writer_lock:
 new_tbl = self._make_table()
 with self._lock:
 old = self._table
 self._table = new_tbl
 with self._retired_lock:
 self._retired.append(old)
 # Grace period: wait until all readers that were active
 # at swap time have finished. We approximate by waiting
 # for all epoch counters to reach 0.
 deadline = time.perf_counter() + 0.005 # cap wait
 while time.perf_counter() < deadline:
 with self._epoch_lock:
 if all(v == 0 for v in self._epochs.values()):
 break
 time.sleep(0.0001)
 # Now reclaim all retired tables
 with self._retired_lock:
 n = len(self._retired)
 self._retired.clear()
 self.reclaimed_count += n

def run_trial(strategy, duration=3.0, n_readers=4, table_size=1024):
 """Run one trial and return read ops/sec."""
 if strategy == "hazard":
 table = HazardPointerTable(size=table_size)
 else:
 table = RCUTable(size=table_size)

 stop_flag = threading.Event()
 read_counts = [0] * n_readers
 keys = list(range(table_size))

 def reader(tid):
 table.register_reader(tid)
 count = 0
 local_keys = keys
 n = len(local_keys)
 try:
 while not stop_flag.is_set():
 tbl = table.acquire(tid)
 # do a batch of lookups while holding the reference
 for i in range(64):
 k = local_keys[(count + i) % n]
 table.lookup(tbl, k)
 table.release(tid)
 count += 64
 finally:
 table.unregister_reader(tid)
 read_counts[tid] = count

 def writer():
 while not stop_flag.is_set():
 table.swap()
 time.sleep(0.001)

 threads = []
 for tid in range(n_readers):
 t = threading.Thread(target=reader, args=(tid,), daemon=True)
 threads.append(t)
 w = threading.Thread(target=writer, daemon=True)

 start = time.perf_counter()
 for t in threads:
 t.start()
 w.start()

 time.sleep(duration)
 stop_flag.set()

 for t in threads:
 t.join(timeout=2.0)
 w.join(timeout=2.0)

 elapsed = time.perf_counter() - start
 total_ops = sum(read_counts)
 ops_per_sec = total_ops / elapsed if elapsed > 0 else 0.0
 return ops_per_sec, table.reclaimed_count

def paired_t_test(a, b):
 """Simple paired t-test (two-sided) using stdlib only."""
 n = len(a)
 if n < 2:
 return None, None
 diffs = [a[i] - b[i] for i in range(n)]
 mean_d = statistics.mean(diffs)
 sd_d = statistics.stdev(diffs)
 if sd_d == 0:
 return float('inf') if mean_d != 0 else 0.0, 0.0
 se = sd_d / (n ** 0.5)
 t_stat = mean_d / se
 # approximate two-sided p-value using normal approximation for df>=4
 # (df = n-1 = 4). Use a rough t-distribution CDF via math.erf for df=4.
 # For df=4, t-distribution is close to normal; use normal approx.
 import math
 # two-sided p-value
 p = 2 * (1 - 0.5 * (1 + math.erf(abs(t_stat) / (2 ** 0.5))))
 return t_stat, p

def main():
 print("[experiment] hazard-pointers vs RCU lookup throughput", file=sys.stderr)
 random.seed(42)

 n_trials = 5
 duration = 3.0
 hazard_ops = []
 rcu_ops = []
 hazard_reclaimed = []
 rcu_reclaimed = []

 for trial in range(n_trials):
 # alternate order to reduce ordering bias
 if trial % 2 == 0:
 h_ops, h_rec = run_trial("hazard", duration=duration)
 r_ops, r_rec = run_trial("rcu", duration=duration)
 else:
 r_ops, r_rec = run_trial("rcu", duration=duration)
 h_ops, h_rec = run_trial("hazard", duration=duration)
 hazard_ops.append(h_ops)
 rcu_ops.append(r_ops)
 hazard_reclaimed.append(h_rec)
 rcu_reclaimed.append(r_rec)
 print(f"[trial {trial+1}] hazard={h_ops:.0f} ops/s (reclaimed={h_rec}) "
 f"rcu={r_ops:.0f} ops/s (reclaimed={r_rec})", file=sys.stderr)

 med_hazard = statistics.median(hazard_ops)
 med_rcu = statistics.median(rcu_ops)
 mean_hazard = statistics.mean(hazard_ops)
 mean_rcu = statistics.mean(rcu_ops)

 if med_rcu > 0:
 pct_diff = (med_hazard - med_rcu) / med_rcu * 100.0
 else:
 pct_diff = float('inf') if med_hazard > 0 else 0.0

 t_stat, p_value = paired_t_test(hazard_ops, rcu_ops)

 # Hypothesis: hazard-pointer reads sustain >= 10% higher median ops/sec
 supported = bool(med_rcu > 0 and pct_diff >= 10.0)

 evidence = {
 "n_trials": n_trials,
 "duration_seconds_per_trial": duration,
 "n_readers": 4,
 "n_writers": 1,
 "hazard_ops_per_sec": [float(x) for x in hazard_ops],
 "rcu_ops_per_sec": [float(x) for x in rcu_ops],
 "hazard_median_ops_per_sec": float(med_hazard),
 "rcu_median_ops_per_sec": float(med_rcu),
 "hazard_mean_ops_per_sec": float(mean_hazard),
 "rcu_mean_ops_per_sec": float(mean_rcu),
 "median_pct_difference": float(pct_diff),
 "paired_t_statistic": float(t_stat) if t_stat is not None else None,
 "paired_p_value": float(p_value) if p_value is not None else None,
 "hazard_reclaimed_counts": [int(x) for x in hazard_reclaimed],
 "rcu_reclaimed_counts": [int(x) for x in rcu_reclaimed],
 "threshold_pct": 10.0,
 }

 interpretation = (
 f"Hazard-pointer reads achieved a median of {med_hazard:.0f} ops/sec vs "
 f"{med_rcu:.0f} ops/sec for RCU-style reads ({pct_diff:+.1f}% difference). "
 f"The hypothesis that hazard pointers sustain at least 10% higher median "
 f"throughput was {'supported' if supported else 'not supported'}."
 )

 result = {
 "hypothesis": (
 "A hazard-pointer-protected lock-free read path will show higher "
 "throughput than an RCU-style read path on a CPU-only workload with "
 "frequent reclamation, because hazard pointers avoid the grace-period "
 "delay that RCU incurs when reclaiming memory; pass/fail is decided by "
 "whether hazard-pointer reads sustain at least 10% higher median "
 "ops/sec across 5 trials."
 ),
 "hypothesis_supported": supported,
 "evidence": evidence,
 "interpretation": interpretation,
 }

 print(json.dumps(result, indent=2, default=lambda o: o.item() if hasattr(o, "item") else str(o)))
 return 0

if __name__ == "__main__":
 sys.exit(main())