latchkit

One-shot synchronization primitives for threads and asyncio: a count-down latch, a gate and a fire-once event, each with abort-with-reason, typed errors and timeouts that behave. Zero dependencies, fully typed.

Install

pip install latchkit

Usage

Wait for a fixed number of workers to finish before moving on:

import threading

from latchkit import CountDownLatch

latch = CountDownLatch(3)


def worker(i: int) -> None:
    with latch.arrival():  # counts down when the block exits, even on error
        ...  # do the work


for i in range(3):
    threading.Thread(target=worker, args=(i,)).start()

latch.wait()  # blocks until all three have counted down

wait takes a timeout and returns whether the latch opened in time; wait_or_raise raises LatchTimeoutError (a TimeoutError) instead:

if not latch.wait(timeout=5.0):
    print("still waiting")
latch.wait_or_raise(timeout=5.0)

When something goes wrong, abort instead of leaving everyone blocked. Every current and future waiter raises BrokenLatchError with your reason:

from latchkit import BrokenLatchError

latch.abort(RuntimeError("database unreachable"))
try:
    latch.wait()
except BrokenLatchError as error:
    print(error.reason)  # the RuntimeError, also chained as __cause__

A Gate is a manual switch that, once opened, stays open for every waiter. opening() guards a run-once initialisation: it opens on success and aborts on any failure (cancellation included), so waiters never hang on a crashed initialiser:

from latchkit import Gate

ready = Gate()

with ready.opening():
    ...  # load config, warm caches

ready.wait()  # returns at once, now and forever

A OneShotEvent fires exactly once and tells the caller who won the race:

from latchkit import OneShotEvent

claimed = OneShotEvent()
if claimed.set():  # True for the first caller only
    print("this thread won")

Every primitive has an asyncio twin with the same API. Signalling is thread-safe, so worker threads can release coroutines:

import asyncio

from latchkit import AsyncCountDownLatch


async def main() -> None:
    latch = AsyncCountDownLatch(4)
    loop = asyncio.get_running_loop()
    for _ in range(4):
        loop.run_in_executor(None, latch.count_down)
    await latch.wait(timeout=10)


asyncio.run(main())

The guide covers each primitive, abort and errors, asyncio and thread bridging, timeouts, and the exact memory-ordering guarantees.

API at a glance

All six share wait(timeout), wait_or_raise(timeout), abort(reason), is_broken and abort_reason. Errors: LatchError, BrokenLatchError, LatchTimeoutError.

Why

The standard library has threading.Event, threading.Barrier, threading.Condition and their asyncio cousins, and none of them is a count-down latch:

• threading.Barrier is cyclic and symmetric: every participant must both arrive and wait, so one thread cannot wait while N unrelated threads each signal once. If you need a rendezvous, use it; latchkit does not duplicate it.
• threading.Event and asyncio.Event are a single resettable boolean with no count, no notion of failure and no winner. A clear() can make a late waiter miss the signal, and a crashed producer leaves waiters blocked forever. asyncio.Event is also not thread-safe, so a worker thread cannot set it.
• threading.Condition is the right building block, which is why people hand-roll a counter plus a condition in every codebase and get the edge cases (counting past zero, waking everyone, failure, timeouts) subtly wrong.

The "wait for N things, then proceed, once" primitive is CountDownLatch in Java and CountdownEvent in .NET. latchkit is that primitive done once, correctly, for both threads and asyncio, plus the two one-shot shapes that come up next to it, and an abort path modelled on threading.Barrier.abort() so failures reach waiters instead of hanging them.

There is no CLI: these are in-process primitives, and a command line could not share one with anything.

License

MIT. See LICENSE.

Links

• Website: https://lianes8studio.com
• GitHub: https://github.com/JaviLianes8
• LinkedIn: https://www.linkedin.com/in/jlianes/