Concurrency Module

Concurrency backends and primitives for Katharos.

A BaseThreadingBackend abstracts thread spawning and synchronization primitives so concurrency types (such as Lazy and Channel) are not tied to a specific threading library. ThreadingBackend is the standard-library default.

Threading Backends

BaseThreadingBackend

class katharos.concurrency.BaseThreadingBackend

Bases: ABC

Abstraction over a concurrency implementation.

A backend decouples Katharos’ concurrency types (such as Lazy and Channel) from a specific threading library. The default backend is built on the standard library threading module, but an alternative (for example one backed by greenlet or gevent) can be supplied instead, as long as it provides compatible thread spawning and synchronization primitives.

A backend exposes three capabilities:

• Spawning lightweight units of work via spawn().

• Synchronization primitives via create_lock() and create_condition(), so consumers coordinate using primitives that cooperate with the backend’s scheduler.

• Context-local storage via create_local(), isolated per unit of concurrency the backend schedules.

abstractmethod spawn(fn)

Run a zero-argument callable concurrently.

Parameters:

fn (Callable[[], Any]) – The work to run. Its return value is ignored; callers that need a result should arrange for fn to store it.

Return type:

BaseThreadHandle

Returns:

A BaseThreadHandle for the started work.

abstractmethod create_lock()

Create a new mutual-exclusion lock.

Return type:

AbstractLock

Returns:

A lock that cooperates with this backend’s scheduler.

abstractmethod create_condition()

Create a new condition variable.

Return type:

AbstractCondition

Returns:

A condition variable that cooperates with this backend’s scheduler.

abstractmethod create_local()

Create a context-local storage object.

The returned object stores attributes independently for each unit of concurrency the backend schedules: per OS thread for a thread-based backend, per green thread for a green-thread backend. It is used to hold execution-context-specific state, such as the active scopes of a Go launcher.

Using the backend’s own context-local (rather than hardcoding threading.local) is what keeps that per-context state correct when the backend multiplexes many tasks onto a single OS thread.

Return type:

Any

Returns:

An object supporting arbitrary attribute get/set/delete that is isolated per scheduled context (for example threading.local).

ThreadingBackend

class katharos.concurrency.ThreadingBackend

Bases: BaseThreadingBackend

A threading backend built on the standard library threading.

This is the default backend used by Katharos’ concurrency types. Spawned work runs on daemon threading.Thread instances, and the synchronization primitives are plain threading.Lock and threading.Condition objects.

spawn(fn)

Run fn on a new daemon thread.

Parameters:

fn (Callable[[], Any]) – The zero-argument callable to run. Its return value is ignored.

Return type:

ThreadingHandle

Returns:

A ThreadingHandle for the started thread.

create_lock()

Create a new threading.Lock.

Return type:

AbstractLock

Returns:

A standard-library lock.

create_condition()

Create a new threading.Condition.

Return type:

AbstractCondition

Returns:

A standard-library condition variable.

create_local()

Create a new threading.local.

Return type:

local

Returns:

Standard-library thread-local storage, isolated per OS thread.

default_backend

katharos.concurrency.default_backend()

Return the shared default threading backend.

Consumers that are not given an explicit backend use this one. The instance is stateless and safe to share across threads.

Return type:

ThreadingBackend

Returns:

The process-wide default ThreadingBackend.

Thread Handles

BaseThreadHandle

class katharos.concurrency.BaseThreadHandle

Bases: ABC

A handle to a unit of work started by a BaseThreadingBackend.

Returned by BaseThreadingBackend.spawn(); lets callers wait for the spawned work to finish or check whether it is still running, without exposing the underlying thread implementation.

abstractmethod join(timeout=None)

Block until the spawned work finishes (or timeout elapses).

Parameters:

timeout (float | None) – Maximum seconds to wait, or None to wait forever.

Return type:

None

abstractmethod is_alive()

Return whether the spawned work is still running.

Return type:

bool

Returns:

True while the work is in progress, False once it has finished.

ThreadingHandle

class katharos.concurrency.ThreadingHandle(thread)

Bases: BaseThreadHandle

A BaseThreadHandle backed by a threading.Thread.

__init__(thread)

Wrap an already-started thread.

Parameters:

thread (Thread) – The running thread this handle refers to.

join(timeout=None)

Block until the underlying thread finishes (or timeout elapses).

Parameters:

timeout (float | None) – Maximum seconds to wait, or None to wait forever.

Return type:

None

is_alive()

Return whether the underlying thread is still running.

Return type:

bool

Returns:

True while the thread is running, False once it finishes.

Synchronization Protocols

AbstractLock

class katharos.concurrency.AbstractLock(*args, **kwargs)

Bases: Protocol

A mutual-exclusion lock usable as a context manager.

Only the context-manager surface is required, which is all the concurrency primitives in Katharos rely on. threading.Lock satisfies this protocol structurally, as does any green-thread equivalent.

__enter__()

Acquire the lock on entering a with block.

Return type:

bool

__exit__(exc_type, exc_value, traceback, /)

Release the lock on leaving a with block.

Return type:

bool | None

__init__(*args, **kwargs)

AbstractCondition

class katharos.concurrency.AbstractCondition(*args, **kwargs)

Bases: Protocol

A condition variable usable as a context manager.

Captures the subset of threading.Condition that Katharos uses: acquiring the underlying lock via with, waiting for a notification, and notifying waiters. threading.Condition satisfies this protocol structurally.

__enter__()

Acquire the underlying lock on entering a with block.

Return type:

bool

__exit__(exc_type, exc_value, traceback, /)

Release the underlying lock on leaving a with block.

Return type:

bool | None

wait(timeout=None)

Release the lock and block until notified (or timeout elapses).

Parameters:

timeout (float | None) – Maximum seconds to wait, or None to wait forever.

Return type:

bool

Returns:

False if the call returned because timeout elapsed, else True.

notify(n=1)

Wake at most n threads waiting on this condition.

Parameters:

n (int) – The maximum number of waiters to wake.

Return type:

None

notify_all()

Wake all threads waiting on this condition.

Return type:

None

__init__(*args, **kwargs)

CSP Primitives

CSP (Communicating Sequential Processes) style concurrency for Katharos.

This module provides Go-style concurrency primitives for coordinating concurrent computations in the CSP tradition:

• CSPRuntime: a runtime that binds channels and goroutines to a single threading backend; the csp instance is the default runtime.

• Go: launch a callable concurrently, mirroring Go’s go func().

• Channel: a Go-style, thread-safe channel for communicating values between threads (paired with Go).

• ChannelClosedError: raised on sending to or closing a closed channel.

• select() / recv(): wait on whichever of several channels becomes ready first, mirroring Go’s select; SelectResult describes the outcome.

CSPRuntime

class katharos.concurrency.csp.CSPRuntime(backend=None)

Bases: object

A CSP runtime that binds channels and goroutines to one backend.

CSPRuntime bundles Katharos’ Communicating Sequential Processes primitives – Channel and Go – around a single BaseThreadingBackend. Channels created via Channel and work launched via go all run on (and synchronize through) that one backend, so the whole concurrency model is swappable in a single place.

The two members are:

• Channel: a backend-bound channel class. Subscript it with the element type and call it to construct a channel: ch = csp.Channel[int](capacity=1). The backend is supplied automatically, so callers never pass it.

• go: a backend-bound Go launcher. Call it to run work concurrently (csp.go(fn, *args)) or use it as a with block for structured concurrency.

Examples

>>> from katharos.concurrency.threading_backend import ThreadingBackend
>>> csp = CSPRuntime(ThreadingBackend())
>>> ch = csp.Channel[int](capacity=1)
>>> _ = csp.go(ch.send, 42)
>>> ch.recv()
Success(42)

Used as a structured-concurrency scope, the block waits for everything launched inside it:

>>> ch = csp.Channel[int]()
>>> with csp.go:
...     _ = csp.go(ch.send, 1)
...     received = ch.recv()
>>> received
Success(1)

__init__(backend=None)

Initialize a CSP runtime bound to a backend.

Parameters:

backend (BaseThreadingBackend | None) – The threading backend that channels and goroutines from this runtime use. Defaults to the shared default_backend().

property backend: BaseThreadingBackend

The threading backend shared by this runtime’s channels and goroutines.

katharos.concurrency.csp.csp = <katharos.concurrency.csp.csp.CSPRuntime object>

A CSP runtime that binds channels and goroutines to one backend.

CSPRuntime bundles Katharos’ Communicating Sequential Processes primitives – Channel and Go – around a single BaseThreadingBackend. Channels created via Channel and work launched via go all run on (and synchronize through) that one backend, so the whole concurrency model is swappable in a single place.

The two members are:

• Channel: a backend-bound channel class. Subscript it with the element type and call it to construct a channel: ch = csp.Channel[int](capacity=1). The backend is supplied automatically, so callers never pass it.

• go: a backend-bound Go launcher. Call it to run work concurrently (csp.go(fn, *args)) or use it as a with block for structured concurrency.

Examples

>>> from katharos.concurrency.threading_backend import ThreadingBackend
>>> csp = CSPRuntime(ThreadingBackend())
>>> ch = csp.Channel[int](capacity=1)
>>> _ = csp.go(ch.send, 42)
>>> ch.recv()
Success(42)

Used as a structured-concurrency scope, the block waits for everything launched inside it:

>>> ch = csp.Channel[int]()
>>> with csp.go:
...     _ = csp.go(ch.send, 1)
...     received = ch.recv()
>>> received
Success(1)

Channel

class katharos.concurrency.csp.Channel(capacity=0, *, backend)

Bases: Generic

A Go-style channel for communicating values between threads.

A Channel is a typed, thread-safe conduit. Producers send() values and consumers recv() them; both operations block to coordinate the two sides, mirroring Go’s channels.

The capacity controls buffering:

• Unbuffered (capacity == 0, the default): every send() blocks until another thread is ready to recv() the value, giving a synchronous hand-off (rendezvous).

• Buffered (capacity > 0): send() only blocks once the buffer is full, and recv() only blocks once it is empty.

Unlike Go, recv() returns a Result: Success(value) while values are available, Failure(ChannelClosedError) once the channel is closed and drained, or Failure(ChannelTimeoutError) if a timeout is specified and expires. This makes “the channel is closed” a type-safe outcome rather than a second return value. Iterating over a channel yields its values until it is closed.

Note

Every send(), recv(), and close() wakes all waiters (notify_all) to keep coordination correct under contention. This is O(number of waiters) per operation, so a single channel shared by a very large number of blocked senders/receivers trades throughput for simplicity.

Examples

>>> from katharos.concurrency.threading_backend import ThreadingBackend
>>> ch = Channel[int](capacity=1, backend=ThreadingBackend())
>>> ch.send(42)
>>> ch.recv()
Success(42)
>>> ch.close()
>>> ch.recv()
Failure(ChannelClosedError('recv on closed channel'))

__init__(capacity=0, *, backend)

Initialize a channel.

Parameters:

• capacity (int) – Buffer size. 0 (the default) makes the channel unbuffered, so each send blocks for a matching receive. A positive value buffers that many pending values.

• backend (BaseThreadingBackend) – The threading backend whose condition variable coordinates senders and receivers.

Raises:

ValueError – If capacity is negative.

property capacity: int

The channel’s buffer capacity (0 for an unbuffered channel).

property backend: BaseThreadingBackend

The threading backend coordinating this channel’s senders and receivers.

send(value)

Send a value into the channel, blocking as needed.

For a buffered channel this blocks while the buffer is full. For an unbuffered channel it blocks until another thread receives the value.

Parameters:

value (TypeVar(A)) – The value to send.

Raises:

ChannelClosedError – If the channel is (or becomes) closed before the value can be delivered.

Return type:

None

recv(timeout=None)

Receive a value, blocking until one is available or the channel closes.

Parameters:

timeout (float | None) – Maximum seconds to wait for a value. None (default) blocks indefinitely. A positive float will return a ChannelTimeoutError failure if no value arrives in time.

Return type:

Result[ChannelClosedError | ChannelTimeoutError, TypeVar(A)]

Returns:

Success(value) when a value is received. Failure(ChannelClosedError) when the channel is closed and empty. Failure(ChannelTimeoutError) when the timeout expires.

Examples

>>> from katharos.concurrency.threading_backend import ThreadingBackend
>>> ch = Channel[int](capacity=1, backend=ThreadingBackend())
>>> ch.send(42)
>>> ch.recv()
Success(42)

>>> ch = Channel[int](backend=ThreadingBackend())
>>> ch.close()
>>> ch.recv()
Failure(ChannelClosedError('recv on closed channel'))

Timeout on an empty channel:

>>> ch = Channel[int](backend=ThreadingBackend())
>>> ch.recv(timeout=0.01)
Failure(ChannelTimeoutError('recv timed out after 0.01 seconds'))

register_observer(callback)

Register a callback notified when this channel changes state.

Used by select() to learn when a value becomes available (or the channel closes) without polling. The callback is invoked while this channel’s internal lock is held, so it must do no more than set a flag and signal its own condition; it must never call back into this channel (doing so would deadlock the non-reentrant lock).

Parameters:

callback (Callable[[], None]) – A zero-argument callable invoked on each state change.

Return type:

None

unregister_observer(callback)

Remove a callback previously registered with register_observer().

Removing a callback that is not registered is a no-op.

Parameters:

callback (Callable[[], None]) – The callback to remove.

Return type:

None

close()

Close the channel.

After closing, send() raises ChannelClosedError and recv() returns the remaining buffered values followed by a Failure(ChannelClosedError). Any blocked senders or receivers are woken.

Raises:

ChannelClosedError – If the channel is already closed.

Return type:

None

__iter__()

Iterate over received values until the channel is closed and drained.

Yields:

Each received value, in order.

__repr__()

Return a string representation of the channel.

Return type:

str

Go

class katharos.concurrency.csp.Go(backend)

Bases: object

Launch callables concurrently, mimicking Go’s go func() statement.

A Go instance is callable: go(fn, *args, **kwargs) runs fn(*args, **kwargs) concurrently on a threading backend and returns immediately, just like go f(x, y) starts a goroutine in Go. The work runs on the instance’s BaseThreadingBackend, so the concurrency model is swappable (standard threads by default, or a green-thread backend if supplied).

Used as a context manager, Go becomes a structured-concurrency scope: every call made inside the with block is tracked, and leaving the block blocks until all of that work has finished (similar to a sync.WaitGroup). Scopes are tracked per execution context (via the backend’s context-local storage) and nest correctly, so a single Go instance is safe to share across contexts. Calls made outside any with block are pure fire-and-forget and are not tracked.

You normally reach a launcher through a CSPRuntime runtime (the csp default exposes one as csp.go); construct your own Go only to pin a specific backend.

Note

Like a goroutine, a callable launched here is fire-and-forget: its return value is discarded and an exception it raises does not propagate out of the scope. Use a Lazy or a Channel to communicate results or failures back to the caller.

Examples

>>> import threading
>>> from katharos.concurrency.threading_backend import ThreadingBackend
>>> go = Go(ThreadingBackend())
>>> done = threading.Event()
>>> handle = go(done.set)
>>> handle.join()
>>> done.is_set()
True

>>> results = []
>>> handle = go(results.append, 42)
>>> handle.join()
>>> results
[42]

Used as a scope, the block waits for everything spawned inside it:

>>> lock = threading.Lock()
>>> collected = []
>>> def collect(x):
...     with lock:
...         collected.append(x)
>>> with go:
...     _ = go(collect, 1)
...     _ = go(collect, 2)
>>> sorted(collected)
[1, 2]

__init__(backend)

Initialize a launcher bound to a backend.

Parameters:

backend (BaseThreadingBackend) – The threading backend used to spawn work.

__call__(fn, *args, **kwargs)

Run fn(*args, **kwargs) concurrently without blocking.

If called inside a with block on this instance, the spawned work is tracked and joined when the block exits.

Parameters:

• fn (Callable[..., Any]) – The callable to run concurrently. Its return value is discarded; use a Lazy or a Channel to communicate results back.

• *args (Any) – Positional arguments forwarded to fn.

• **kwargs (Any) – Keyword arguments forwarded to fn.

Return type:

BaseThreadHandle

Returns:

A BaseThreadHandle for the spawned work, so callers can join it. It may be ignored for fire-and-forget use.

__enter__()

Open a structured-concurrency scope on the calling context.

Return type:

Go

Returns:

This launcher, so calls made on it inside the block are tracked.

__exit__(exc_type, exc_value, traceback)

Close the scope, blocking until all work spawned in it finishes.

The scope is joined even when the block exits because of an exception, so no spawned work outlives the block. The exception, if any, is allowed to propagate.

Parameters:

• exc_type (type[BaseException] | None) – The exception type if the block raised, else None.

• exc_value (BaseException | None) – The exception instance if the block raised, else None.

• traceback (TracebackType | None) – The traceback if the block raised, else None.

Return type:

None

select

katharos.concurrency.csp.select(*cases, default=False, timeout=None)

Wait for one of several channel operations to become ready.

Mirrors Go’s select: given a number of cases (built with recv()), block until one is ready and return it. When several are ready at once, the first in argument order wins. A closed channel counts as ready (its case yields a Failure(ChannelClosedError)).

Parameters:

• *cases (_RecvCase[TypeVar(A)]) – The receive cases to wait on, in priority order.

• default (bool) – If True, return immediately with a default result when no case is ready, instead of blocking. Mutually useful with, and takes precedence over, timeout.

• timeout (float | None) – Maximum seconds to block waiting for a case. None (the default) blocks indefinitely.

Returns:

the chosen case, the default branch, or a timeout.

Return type:

SelectResult[TypeVar(A)]

Raises:

ValueError – If called with no cases and neither default nor timeout, which would block forever.

Examples

>>> from katharos.concurrency.csp import csp, recv, select
>>> a = csp.Channel[int](capacity=1)
>>> b = csp.Channel[int](capacity=1)
>>> b.send(99)
>>> choice = select(recv(a), recv(b))
>>> choice.index, choice.value.unwrap()
(1, 99)

With default, an empty channel does not block:

>>> select(recv(a), default=True)
SelectResult(default)

katharos.concurrency.csp.recv(channel)

Build a receive case for select().

Parameters:

channel (Channel[TypeVar(A)]) – The channel to receive from when this case is chosen.

Return type:

_RecvCase[TypeVar(A)]

Returns:

A select case that, when ready, yields the result of receiving from channel.

Examples

>>> from katharos.concurrency.csp import csp, recv, select
>>> ch = csp.Channel[int](capacity=1)
>>> ch.send(7)
>>> choice = select(recv(ch))
>>> choice.index, choice.value.unwrap()
(0, 7)

SelectResult

class katharos.concurrency.csp.SelectResult(index=None, channel=None, value=None, is_default=False, is_timeout=False)

Bases: Generic

The outcome of a select() call.

Exactly one of three things happened: a case became ready (index, channel and value are set), the default branch was taken (is_default), or the call timed out (is_timeout). In the latter two cases index, channel and value are all None.

index

Position (in argument order) of the chosen case, or None.

channel

The chosen case’s channel, or None.

value

The chosen case’s Result, or None.

is_default

Whether the default branch was taken.

is_timeout

Whether the call timed out before any case was ready.

__repr__()

Return a string representation of the result.

Return type:

str

Exceptions

exception katharos.concurrency.csp.ChannelClosedError

Bases: Exception

Raised when sending on, or closing, an already-closed channel.

exception katharos.concurrency.csp.ChannelTimeoutError

Bases: Exception

Raised when a channel operation times out.