Sifr’s concurrency model is built on structured concurrency: every task lives inside an explicit scope, and the scope does not exit until all child tasks have finished or been cancelled. There are no fire-and-forget tasks and no global event loops. The compiler enforces that values crossing task, thread, or process boundaries are owned and sendable — the same ownership rules that apply to the rest of the language.
Async Functions
Declare an async function with async def. Call it with await inside another async function or an async scope.
async def one() -> int:
await task.sleep(0.0)
return 1
async def two() -> int:
await task.sleep(0.0)
return 2
Result[T, E]:
async def fail_fast() -> Result[int, ValueError]:
await task.sleep(0.0)
raise ValueError("group child failed")
task.scope and Scoped Spawn
Use task.scope() as an async context manager to create a structured scope. Spawn child tasks with scope.spawn(...). The scope awaits all children before the async with block exits.
async with task.scope() as scope:
gathered = await task.gather([scope.spawn(one()), scope.spawn(two())])
task.gather collects results in the order the handles were passed, preserving deterministic output ordering.
Scoped spawn requires an active owner. Values that a spawned task captures must be owned and sendable across the task boundary. Lock guards, borrowed references, and other non-send resources are rejected at compile time.
task.select — Racing Tasks
Use task.select to race two tasks against each other. The first task to finish wins; the compiler-generated runtime cancels the losing child automatically.
async def fast() -> int:
await task.sleep(0.0)
return 10
async def slow_writes_marker() -> int:
await task.sleep(0.20)
try:
_written: None = write_text(marker_path(), "not cancelled")
except IOError as e:
_message: str = str(e.message)
return 20
async with task.scope() as scope:
selected = await task.select(
first=scope.spawn(fast()),
second=scope.spawn(slow_writes_marker()),
)
task.select consumes both handles. Whichever child loses is cancelled before the scope exits, so you never observe a dangling task writing to shared state after the race.
TaskGroup — Structured Error Propagation
task.TaskGroup is the right tool when you want a group of tasks to run concurrently and you need to observe individual results. If one child fails, the TaskGroup cancels remaining siblings at scope exit.
async with task.TaskGroup() as group:
slow = group.spawn(slow_writes_marker())
failing = group.spawn(fail_fast())
failure = await failing
TaskHandle[T, E] values are linear: awaiting or joining a handle consumes it. Attempting to await the same handle twice is a compile-time ownership error.
TaskGroup cancels unfinished siblings when any child fails. Do not rely on a sibling completing work after another child in the same group has raised an error.
A Complete Structured Concurrency Example
from sifr.io import exists, write_text
from sifr.os import getpid, remove_file
def marker_path() -> str:
return "/tmp/sifr_structured_concurrency_demo_" + str(getpid()) + ".txt"
async def one() -> int:
await task.sleep(0.0)
return 1
async def two() -> int:
await task.sleep(0.0)
return 2
async def fast() -> int:
await task.sleep(0.0)
return 10
async def slow_writes_marker() -> int:
await task.sleep(0.20)
try:
_written: None = write_text(marker_path(), "not cancelled")
except IOError as e:
_message: str = str(e.message)
return 20
async def fail_fast() -> Result[int, ValueError]:
await task.sleep(0.0)
raise ValueError("group child failed")
async def main() -> Result[None, Error]:
path: str = marker_path()
if exists(path):
try:
_removed_existing: None = remove_file(path)
except IOError as e:
_cleanup_message: str = str(e.message)
# Scoped gather: deterministic input ownership and ordering
async with task.scope() as scope:
gathered = await task.gather([scope.spawn(one()), scope.spawn(two())])
# Select: cancels the losing child
async with task.scope() as scope:
selected = await task.select(
first=scope.spawn(fast()),
second=scope.spawn(slow_writes_marker()),
)
# TaskGroup: cancels siblings when a child fails
async with task.TaskGroup() as group:
slow = group.spawn(slow_writes_marker())
failing = group.spawn(fail_fast())
failure = await failing
assert not exists(path)
return None
sifr.sync — Shared State and Channels
Use sifr.sync for same-process communication between tasks.
Channels
Shared[T]
Lock[T]
Channels transfer ownership of values between tasks. Sending moves the value into the channel; receiving moves it back out.from sifr.sync import channel
sender, receiver = channel()
async with task.scope() as scope:
scope.spawn(sender.send(42))
value = await receiver.receive()
bounded_channel(capacity) for backpressure — sends block when the buffer is full. Shared[T] wraps a value for concurrent read access by multiple tasks. Multiple tasks can hold a Shared[T] handle simultaneously without violating ownership rules.from sifr.sync import Shared
counter: Shared[int] = Shared(0)
Lock[T] provides exclusive mutable access. A LockGuard is returned when the lock is acquired, and it is released when the guard goes out of scope.from sifr.sync import Lock
state: Lock[int] = Lock(0)
await point is rejected by the compiler’s ownership diagnostics.sifr.parallel — CPU-Parallel Work
Use sifr.parallel to distribute CPU-heavy work across native worker threads. parallel.map preserves output order; parallel.try_map returns a typed error when any item fails.
from sifr.parallel import map as par_map
results = par_map(items, heavy_compute)
Captured values and return values passed to parallel.map must satisfy worker-boundary ownership and sendability requirements. The same rules that apply to task.scope apply here.
Pool with PoolConfig:
from sifr.parallel import Pool, PoolConfig
config: PoolConfig = PoolConfig(workers=4)
pool: Pool = Pool(config)
results = pool.map(items, heavy_compute)
Concurrency Modules at a Glance
| Module | Purpose |
|---|
sifr.task | Structured async tasks, scopes, gather, select, TaskGroup |
sifr.sync | Channels, Shared[T], Lock[T], RwLock[T], Semaphore, Notify |
sifr.parallel | CPU-parallel map, try_map, and Pool |
sifr.process | Native subprocess spawning and I/O |
sifr.signal | Structured shutdown and signal handling |
sifr.runtime | Structured diagnostic events and observability |
sifr.resource | Deterministic cleanup helpers (nullcontext) |
sifr.ipc | Typed IPC substrate for future process workers |
