Agent Skills: Concurrency

Concurrency

Overview

Cross-language concurrency reference organized by concept, not by language. Each concept lists the shared principle plus per-language gotcha callouts for Rust (tokio), Python (asyncio), TypeScript (event loop), and Go. The hard part is never the happy-path API; it's the failure modes — data races, deadlocks, latent OOM from unbounded queues, futures dropped mid-flight by cancellation, and memory-ordering bugs that only surface under load on weakly-ordered CPUs.

Agent Delegation

• loom-senior-software-engineer (Opus) — DEFAULT. Threading models, shared-state vs message-passing, race/TOCTOU analysis, lock ordering, memory ordering, distributed concurrency/consistency, saga patterns.
• loom-software-engineer (Sonnet) — ONLY boilerplate async handlers or unit tests following an established pattern.

Mental Model: Pick the Right Executor

| Workload | Rust | Python | TS/JS | Go | | --- | --- | --- | --- | --- | | I/O-bound | tokio tasks | asyncio | async/Promises | goroutines | | CPU-bound | rayon / spawn_blocking | ProcessPoolExecutor | Worker threads | goroutines (real parallelism) | | Blocking call in async | spawn_blocking | run_in_executor/to_thread | Worker | fine (goroutines block cheaply) |

Runtime model determines everything:

• Rust/tokio, Go — genuinely multi-threaded; tasks run on a thread pool → shared mutable state needs real synchronization.
• Python asyncio — single OS thread; the GIL means asyncio gives you concurrency, not parallelism. Great for I/O, useless for CPU (one core). CPU work → ProcessPoolExecutor. A blocking call (sync DB driver, time.sleep, heavy compute) freezes the entire event loop.
• JS/TS — single-threaded event loop. Same rule: a long synchronous loop blocks all timers, I/O callbacks, and rendering. CPU work → Web/Worker threads. There are no data races on shared JS objects (no preemption between awaits within a microtask), but state can change across an await.

⚠ The blocking-in-async footgun (asyncio + JS): await only yields at await points. Synchronous CPU work or a blocking syscall between awaits stalls every other task. Never call a sync HTTP/DB client, time.sleep, or a tight compute loop directly in an async function — offload it.

Data Races & Shared State

A data race = two threads touch the same memory, ≥1 writes, no synchronization. Rust makes most compile-errors via Send/Sync; Go/Python/JS do not.

Principle: minimize shared mutable state. Prefer message passing (channels) or immutable snapshots. When you must share:

// Rust: Arc<Mutex<T>> for writes, Arc<RwLock<T>> for read-heavy.
use std::sync::Arc;
use tokio::sync::Mutex;               // async Mutex; std::sync::Mutex for sync-only sections
let cache = Arc::new(Mutex::new(HashMap::<String, String>::new()));
let mut g = cache.lock().await;
g.insert(k, v);                        // guard drops at end of scope → keep scope tiny

Per-language gotchas:

• Rust — RwLock can starve writers under continuous readers; prefer Mutex unless reads vastly dominate. std::sync::Mutex is not poisoned-safe across await — use tokio::sync::Mutex only when the guard must cross .await (see below), otherwise the std mutex is faster.
• Go — the race detector (go test -race, go run -race) is your primary tool; run it in CI. Copying a sync.Mutex by value silently breaks it (go vet catches most). Map access is not goroutine-safe — a concurrent read+write panics ("concurrent map writes"); use sync.RWMutex or sync.Map.
• Python — even with the GIL, x += 1 is not atomic (LOAD/ADD/STORE bytecodes can interleave on thread switch). Threaded code still needs threading.Lock. asyncio code needs asyncio.Lock only around state that spans an await.
• JS — no locks needed for sync sections, but "read state → await → write state" is a check-then-act race: the awaited gap lets other tasks mutate. Re-read after the await or guard with an async mutex.

Locks, Lock Ordering & Deadlock

Deadlock rule #1: acquire locks in a globally consistent order. Two code paths taking A→B and B→A will eventually deadlock. Assign a total order (by address, id, or name) and always lock ascending.

Thread 1: lock(A); lock(B)   ┐  both block forever
Thread 2: lock(B); lock(A)   ┘  → enforce A-before-B everywhere

Defenses, in order of preference:

1. Don't hold two locks. Restructure so critical sections don't nest.
2. Consistent ordering when you must nest.
3. try_lock with timeout + backoff as a last resort (detects, doesn't prevent).
4. Keep critical sections tiny — never do I/O, call user callbacks, or .await unrelated work while holding a lock.

⚠ Holding a lock across .await (the async deadlock/Send trap):

• Rust — holding std::sync::MutexGuard across .await makes the future !Send, so tokio::spawn won't compile. Fix: drop the guard before .await, or use tokio::sync::Mutex (designed to be held across await — but then two tasks awaiting the same lock on one thread can deadlock the logical flow). Best: copy the data out, drop the guard, then await.
• Python/JS — holding an asyncio.Lock / async mutex across an await that (transitively) tries to re-acquire the same lock is a self-deadlock. Async locks are not reentrant.
• Go — sync.Mutex is not reentrant either; a method holding the lock calling another method that locks the same mutex deadlocks.

Re-entrancy: none of these mutexes are reentrant. If a locked method calls another locked method on the same object → deadlock. Split into a public (locks) + private (assumes locked) method pair.

Channels & Backpressure

Channels decouple producers from consumers. The single most important decision: bounded vs unbounded.

⚠ Unbounded channels are latent OOM. If producers outpace consumers, an unbounded queue grows without limit until the process is killed. Default to bounded; a full bounded channel applies backpressure — send blocks/awaits, throttling the producer to the consumer's rate. Only use unbounded when you can prove the producer is rate-limited by something else.

| | Bounded | Unbounded | | --- | --- | --- | | Rust tokio | mpsc::channel(N) — send().await | mpsc::unbounded_channel() — ⚠ OOM risk | | Python | asyncio.Queue(maxsize=N) | asyncio.Queue() (maxsize=0) — ⚠ | | Go | make(chan T, N) | make(chan T) (unbuffered = rendezvous, or N too small = stall) |

Per-language gotchas:

• Rust — tokio::sync::mpsc::Receiver is not clonable (single consumer). For fan-out to many workers, share one Arc<Mutex<Receiver>> or use async_channel/flume (MPMC). The old skill's "clone the receiver" pattern does not compile. Dropping all senders closes the channel; recv() then returns None — the clean shutdown signal.
• Go — only the sender closes a channel, and only once; sending on a closed channel panics, closing twice panics. Ranging over a channel (for v := range ch) exits when it's closed. Reading from a nil channel blocks forever (useful in select to disable a case). Leaked goroutines blocked on a channel nobody closes are a top Go bug.
• Python/JS — bound the queue and handle QueueFull/awaiting put as backpressure; an unbounded producer feeding a slow consumer is the classic memory leak.

Structured Concurrency

Principle: spawned tasks have a bounded lifetime tied to a scope; the scope waits for all children and propagates the first error, cancelling siblings. This kills orphaned/leaked tasks and lost exceptions.

# Python 3.11+: TaskGroup — cancels siblings on first exception, raises ExceptionGroup
async with asyncio.TaskGroup() as tg:
    tg.create_task(fetch(a))
    tg.create_task(fetch(b))
# exits only when all done; any failure cancels the rest

// Rust tokio: JoinSet — owns tasks, join_next() yields results as they finish;
// dropping the JoinSet aborts all remaining tasks.
let mut set = tokio::task::JoinSet::new();
for url in urls { set.spawn(fetch(url)); }
while let Some(res) = set.join_next().await {
    let _ = res?;   // JoinError on panic/abort
}

// Go: errgroup — first non-nil error cancels the derived ctx; Wait returns it.
g, ctx := errgroup.WithContext(ctx)
for _, u := range urls {
    u := u                       // pre-1.22 loop-var capture!
    g.Go(func() error { return fetch(ctx, u) })
}
if err := g.Wait(); err != nil { return err }

Gotchas:

• Python — bare asyncio.gather(*tasks) does NOT cancel siblings on failure and, without return_exceptions=True, leaves other tasks running as it raises. TaskGroup (3.11+) is the correct default. A task created with create_task and never awaited can have its exception silently swallowed ("Task exception was never retrieved").
• Rust — tokio::spawn detaches: the task keeps running even if the handle is dropped, and its panic is isolated (only surfaced via the JoinHandle). Prefer JoinSet when you need lifetime + error propagation. join!/try_join! run futures on the current task (no parallelism across threads, but concurrent) and try_join! short-circuits on first error.
• Go — sync.WaitGroup gives you waiting but not error propagation or cancellation; errgroup adds both. Classic footgun: capturing the loop variable by reference (fixed in Go 1.22, still bites older code) — every goroutine sees the last value.
• JS — Promise.all rejects on first failure but does NOT cancel the other in-flight promises (JS promises aren't cancellable); they run to completion, their results discarded. Use Promise.allSettled when you need every outcome. AbortController is the cancellation mechanism for fetch/streams.

Cancellation & Timeouts

Cancellation must propagate. A timeout at the top is useless if inner operations ignore it.

// Rust: tokio::time::timeout returns Err(Elapsed) and DROPS the future at the
// current await point → cancellation. The dropped future must be cancel-safe.
match tokio::time::timeout(Duration::from_secs(5), op()).await {
    Ok(v) => v?, Err(_) => return Err("timed out".into()),
}

// Go: context is the cancellation currency. Thread ctx through every call;
// select on ctx.Done(). ALWAYS defer cancel() to avoid ctx leaks.
ctx, cancel := context.WithTimeout(parent, 5*time.Second)
defer cancel()
select {
case r := <-ch:   return r, nil
case <-ctx.Done(): return zero, ctx.Err()
}

⚠ Cancellation safety (Rust select!, Go select, any timeout): when a future is dropped mid-flight, work already in progress is lost. If that future had taken an item from a channel or half-written to a buffer, dropping it loses the item / corrupts state. In tokio::select!, only use futures whose partial execution is safe to discard; otherwise use Semaphore::acquire_owned / hold the value outside the select!, or use cancellation tokens. mpsc::Receiver::recv is cancel-safe; a hand-rolled read-modify-write across await usually is not.

• Python — cancellation raises asyncio.CancelledError inside the task at its next await. Never swallow it with a bare except Exception (in 3.8+ it derives from BaseException, so except Exception won't catch it — but except: will; don't). Cleanup goes in finally; if you must shield critical cleanup from cancellation, use asyncio.shield.
• JS — pass an AbortSignal to fetch/streams; a rejected/aborted promise still lets already-started microtasks finish. There is no forced cancellation of arbitrary async work.
• Go — a goroutine only stops if it checks ctx.Done(); there is no preemptive kill. Blocking on a channel without a ctx case = uncancellable = leak.

Atomics & Memory Ordering

For a single counter/flag, an atomic beats a mutex. But memory ordering is the trap.

Ordering, weakest→strongest: Relaxed < Acquire/Release < AcqRel < SeqCst.

• Relaxed — atomicity only, NO ordering guarantee vs other memory. Safe for a standalone counter you only read at the end (e.g. metrics). NOT safe to publish data ("store flag, then reader sees the data") — the flag store can be reordered before the data writes.
• Acquire (loads) / Release (stores) — the workhorse. A Release store publishes all prior writes; an Acquire load that sees it observes those writes. This is what you want for lock-free publish/handoff (the "message passing" pattern). Most code that isn't a plain counter wants Acquire/Release.
• SeqCst — single global total order; simplest to reason about, slowest. Use only when multiple atomics must agree on one global order (rare). Reaching for SeqCst "to be safe" is a performance smell, but it's the correct default if you're unsure and can't prove Acquire/Release suffices.

use std::sync::atomic::{AtomicBool, AtomicU64, Ordering};
COUNT.fetch_add(1, Ordering::Relaxed);          // pure counter: Relaxed is fine
DATA.store(payload, Ordering::Relaxed);
READY.store(true, Ordering::Release);           // publish: Release
// reader:
if READY.load(Ordering::Acquire) { use(DATA.load(Ordering::Relaxed)); }  // Acquire pairs with Release

Per-language:

• Go — sync/atomic operations are effectively sequentially consistent; you don't pick an ordering, but you still must use atomics (or a mutex) for any concurrent access — a plain int read/write across goroutines is a race even if "it looks atomic." Prefer atomic.Int64 (Go 1.19+) typed wrappers over the free functions.
• Python/JS — no user-facing memory-ordering knobs (GIL / single event loop). JS SharedArrayBuffer + Atomics is the exception (shared memory across Workers) and there ordering matters again.

⚠ ABA problem (lock-free CAS): a value reads A, changes to B, back to A; a naive compare-and-swap succeeds though the world moved underneath it. Mitigate with tagged pointers / version counters (AtomicU64 epoch) or hazard pointers. When NOT to go lock-free: unless you've measured lock contention as a real bottleneck, a Mutex is simpler, correct, and usually fast enough. Lock-free code is a maintenance and correctness liability — reserve it for proven hot paths.

⚠ False sharing: two unrelated atomics/fields on the same 64-byte cache line cause cores to ping-pong ownership, silently tanking throughput. Pad hot per-thread counters to a cache line (#[repr(align(64))] in Rust, //go:align/padding in Go) when profiling shows it.

Worker Pools & Rate Limiting

Bounded concurrency = fixed workers draining a bounded channel, OR a semaphore capping in-flight tasks.

// Semaphore caps concurrency without spawning a fixed pool.
let sem = Arc::new(tokio::sync::Semaphore::new(max_concurrent));
let tasks = urls.into_iter().map(|u| {
    let permit = sem.clone();
    tokio::spawn(async move { let _p = permit.acquire_owned().await.unwrap(); fetch(&u).await })
});

sem = asyncio.Semaphore(max_concurrent)
async def limited(u):
    async with sem:            # acquire/release even on exception
        return await fetch(u)
await asyncio.gather(*(limited(u) for u in urls))

// Worker pool: N goroutines range over a bounded jobs channel; close(jobs) to drain.
jobs := make(chan Job, 100)
var wg sync.WaitGroup
for i := 0; i < numWorkers; i++ {
    wg.Add(1)
    go func() { defer wg.Done(); for j := range jobs { process(j) } }()
}
// producer: for _, j := range work { jobs <- j }; close(jobs); wg.Wait()

• Fixed pool (channel + N workers) gives strict backpressure and bounded memory. Semaphore-per-task is simpler but each task still allocates until it acquires — bound the spawn rate too for huge inputs (spawn inside the semaphore, or chunk the input).
• Rate limiting (requests/sec) is orthogonal to concurrency (max in-flight): a token bucket / time.Ticker throttles rate; a semaphore caps simultaneity. Real clients usually need both.

Gotchas Quick Reference

• ⚠ Unbounded channel/queue = latent OOM. Default bounded.
• ⚠ Blocking call (sync I/O, sleep, tight loop) in asyncio/JS freezes the whole loop.
• ⚠ Lock held across .await: Rust !Send (won't spawn) or async self-deadlock; copy-out then await.
• ⚠ Inconsistent lock order between two paths → deadlock. Total-order your locks.
• ⚠ Mutexes are non-reentrant (Rust/Go/asyncio) — nested same-lock acquire deadlocks.
• ⚠ gather/Promise.all don't cancel siblings; use TaskGroup/JoinSet/errgroup.
• ⚠ Cancelled/timed-out future dropped mid-flight loses in-progress work — verify cancel-safety.
• ⚠ Go: only sender closes a channel, once; send-on-closed and double-close panic; loop-var capture (<1.22).
• ⚠ Rust: tokio::mpsc::Receiver is single-consumer (not clonable); spawn detaches & isolates panics.
• ⚠ Python: x += 1 isn't atomic even under the GIL; asyncio never gives CPU parallelism.
• ⚠ Atomics: Relaxed for pure counters; Acquire/Release to publish data; SeqCst only for cross-atomic global order.
• ⚠ Lock-free before profiling = premature; ABA and false sharing bite silently.

Verification Checklist

Before declaring concurrent code done:

• [ ] Every channel/queue is bounded, or unboundedness is justified by an upstream rate limit
• [ ] No lock (sync mutex) is held across an .await / blocking call; critical sections are minimal
• [ ] Locks acquired in one consistent global order everywhere; no reentrant same-lock nesting
• [ ] Cancellation/timeout propagates to inner ops (ctx threaded / AbortSignal passed) and dropped futures are cancel-safe
• [ ] Spawned tasks are owned by a scope (JoinSet/TaskGroup/errgroup) — no detached, unawaited, or leaked tasks
• [ ] Errors from tasks propagate and cancel siblings where required (not swallowed by gather/spawn)
• [ ] Shared mutable state is guarded; atomic orderings justified (default Acquire/Release, not reflexive SeqCst/Relaxed)
• [ ] Go: go test -race green in CI; only the sender closes channels; every WithCancel/Timeout has a defer cancel()
• [ ] Python/JS: no blocking/CPU work on the event loop (offloaded to executor/Worker); CancelledError not swallowed
• [ ] Shutdown path drains queues, joins workers, and closes channels cleanly (no leaked goroutines/tasks)
• [ ] Stress-tested under load (many iterations / high concurrency), not just a single happy-path run