Wrap any shared object with protect(). Unsynchronized concurrent
access instantly throws a RaceConditionError
— with thread names, timing, and fix instructions. Zero overhead in production.
list, dict, set,
and any custom object.
data = protect([]) — that's it. The proxy behaves transparently, automatically protecting
nested child objects without manual wrapping.
RACEGUARD_ENABLED=0 (or call configure(enabled=False) before ANY objects are
wrapped) in production for a true zero-cost passthrough of your original
objects. In dev-mode, proxy overhead
is mere microseconds per access.
AtomicGroup to enforce strict joint transactional safety, preventing logical blindspots and TOCTOU (Time-Of-Check to Time-Of-Use) races.
configure(strict=True) in CI to catch lockless accesses regardless of time delays.
Overcomes the heuristic limits of highly-loaded systems.
asyncio task identities to catch thread interleaving between pure
async tasks on the event loop.
import threading results = [] # shared, unprotected def worker(x): results.append(x) # silent data corruption in prod threads = [ threading.Thread(target=worker, args=(i,)) for i in range(10) ] for t in threads: t.start() for t in threads: t.join() # bug impossible to reproduce reliably
import threading from raceguard import protect, with_lock results = protect([]) # ← one change @with_lock(results) def worker(x): results.append(x) # lock held → safe threads = [ threading.Thread(target=worker, args=(i,)) for i in range(10) ] for t in threads: t.start() for t in threads: t.join()
from raceguard import protect, locked, AtomicGroup a = protect([]) b = protect({}) # bind objects transactionally group = AtomicGroup(a, b) def transfer(): with locked(group): a.append(1) b['x'] = 1 # locks automatically resolve
from raceguard import protect, with_lock, configure class Counter: def __init__(self): self.value = 0 def increment(self): self.value += 1 c = protect(Counter()) @with_lock(c) def safe_inc(): c.increment() # attribute read proxied # log, warn, or raise (default) configure(mode="log") # true zero-overhead passthrough configure(enabled=False)
_rg_check() decides.RaceConditionError. Two unguarded threads touching the
same object in rapid succession.race_window (default 10 ms), treated as
sequential. (Note: ABA-style races may evade this heuristic unless Strict Mode is used).One import. One function call. Ship threading bugs to the grave.