# Async Programming > Implements asynchronous programming patterns (asyncio task groups, goroutine pools, cancellation scopes, structured concurrency) to build high-throughput, non-blocking systems across Python and Go runtimes. - Skill: `paulpas/async-programming` (Agent Skill) - Install (CLI): `npx skillmds@latest add paulpas/async-programming` - Raw SKILL.md: https://api.skillmd.com/api/skills/paulpas/async-programming/raw - Safety review: pending - Works with: Claude Code, Claude.ai, OpenAI Codex - Category: Productivity - License: MIT - Author: paulpas (https://skillmd.com/u/paulpas) - Updated: 2026-09-17 - Page: https://skillmd.com/skills/paulpas/async-programming --- # Async Programming Engineer I design and implement high-throughput asynchronous systems using structured concurrency, bounded worker pools, and proper cancellation semantics across Python's asyncio and Go's goroutine model. When I am active, I ensure concurrent code is safe from race conditions, resource leaks, and unhandled errors while delivering measurable latency and throughput improvements over synchronous equivalents. ## TL;DR Checklist - [ ] Match the concurrency model to the workload: asyncio/goroutines for I/O-bound workloads, multiprocessing/cargo for CPU-bound workloads - [ ] Never launch fire-and-forget tasks or goroutines — always track them in a bounded group, semaphore pool, or WaitGroup - [ ] Propagate cancellation via context upstream; set explicit deadlines on all external calls (HTTP, DB, cache) - [ ] Aggregate errors across await boundaries using `asyncio.TaskGroup`, Go `errgroup.Group`, or collector patterns — never swallow exceptions silently - [ ] Protect shared mutable state with mutexes, channels, or actor-pattern message passing — race detector must run clean - [ ] Benchmark async code against its synchronous baseline and verify ≥2x improvement for I/O-bound workloads --- ## When to Use Use this skill when: - Building a high-throughput HTTP server that needs to handle thousands of concurrent connections efficiently - Implementing data pipelines that fetch from multiple external sources in parallel (APIs, databases, queues) - Designing background worker systems with bounded concurrency (email dispatch, report generation, media processing) - Writing event-driven microservices where requests fan out to several downstream services - Refactoring a synchronous, blocking codebase to use non-blocking I/O for improved latency - Implementing retry logic with exponential backoff across concurrent operations ## When NOT to Use Avoid this skill for: - CPU-bound computational workloads (matrix multiplication, image encoding, cryptographic hashing) — use multiprocessing or native extensions instead - Simple scripts that perform one sequential task — async adds complexity without benefit for linear workflows - Code where all dependencies are synchronous and non-overridable with async counterparts — wrapping blocking calls in `run_in_executor` adds overhead --- ## Core Workflow 1. **Choose Concurrency Model** — Pick asyncio for Python (I/O-bound, event-loop driven) or goroutines + channels for Go (lightweight, M:N scheduler). For CPU-bound tasks, use Python's `multiprocessing` or Go's `worker pool + sync.WaitGroup`. **Checkpoint:** Confirm all blocking calls are removed from the hot path; every blocking operation is wrapped in an executor or replaced with its async counterpart. 2. **Structure with Task Groups / WaitGroups** — Never launch fire-and-forget goroutines or tasks. Use bounded task groups (`asyncio.TaskGroup`), semaphore pools, or `sync.WaitGroup` to track and await all concurrent units of work. **Checkpoint:** Every launched unit has a defined lifecycle: creation → execution → completion → resource cleanup. No orphaned goroutines or zombie tasks. 3. **Implement Cancellation & Timeouts** — Propagate cancellation context upstream from the top-level entry point. Set explicit deadlines on all external calls (HTTP, database queries, cache lookups). Use `context.WithTimeout` and `asyncio.wait_for`. **Checkpoint:** When the parent context is cancelled or times out, all child tasks receive the signal within a bounded delay (<100ms) and release their resources. 4. **Handle Errors Across Await Boundaries** — Collect errors without crashing the parent coroutine or goroutine. Use `asyncio.TaskGroup` error aggregation (Python 3.11+), Go `errgroup.Group`, or fan-out/fan-in with an error channel. **Checkpoint:** No unhandled exceptions leak to the event loop or main goroutine; every failure path is either recovered from or collected for downstream handling. 5. **Protect Shared State** — Use mutexes (`sync.Mutex`), channels, or actor-pattern message passing. Never share mutable state between concurrent units without explicit synchronization. Run the race detector on all tests. **Checkpoint:** `go test -race ./...` passes clean; Python code avoids shared mutable globals in favor of channel-based communication or async-safe data structures. 6. **Benchmark & Profile Async Code** — Measure throughput (requests/sec), latency (p50, p95, p99), and memory usage under load with tools like `wrk`, `hey`, or Go's `testing.B`. Compare against the synchronous baseline. **Checkpoint:** The async version shows ≥2x improvement over sync for I/O-bound workloads; memory usage does not grow linearly with concurrency level. --- ## Implementation Patterns ### Pattern 1: Python asyncio TaskGroup with Timeout & Error Aggregation Python 3.11+ `asyncio.TaskGroup` provides structured concurrency — all tasks are awaited and errors are aggregated when the group exits. ```python # ❌ BAD — fire-and-forget tasks, swallowed exceptions, no cancellation import aiohttp import asyncio async def fetch_all_users_bad(): """Launches unbounded requests with no error handling or cancellation.""" session = aiohttp.ClientSession() tasks = [] for org_id in range(100): async def fetch(org=org_id): # Closure trap — all use same org_id! resp = await session.get(f"https://api.example.com/orgs/{org}/users") return await resp.json() tasks.append(asyncio.create_task(fetch())) # Fire-and-forget, no tracking # No timeout, no cancellation propagation results = await asyncio.gather(*tasks, return_exceptions=True) # Exceptions are swallowed in return_exceptions — caller never knows what failed for i, result in enumerate(results): if isinstance(result, Exception): print(f"Task {i} failed: {result}") # Logged but not collected else: print(f"Got {len(result)} users from org {i}") await session.close() # May never run if an exception above re-raises # ✅ GOOD — TaskGroup with timeout, error aggregation, and structured cleanup import asyncio from dataclasses import dataclass, field from typing import Any @dataclass class FetchResult: """Result from a single org fetch — success or failure.""" org_id: int success: bool user_count: int | None = None error: str | None = None latency_ms: float = 0.0 async def fetch_org_users( session: aiohttp.ClientSession, org_id: int, timeout: float = 5.0, ) -> FetchResult: """Fetch users for a single org with timeout and structured error handling.""" start = asyncio.get_event_loop().time() try: async with asyncio.timeout(timeout): async with session.get( f"https://api.example.com/orgs/{org_id}/users", headers={"Accept": "application/json"}, ) as resp: resp.raise_for_status() data = await resp.json() latency_ms = (asyncio.get_event_loop().time() - start) * 1000 return FetchResult( org_id=org_id, success=True, user_count=len(data), latency_ms=latency_ms, ) except asyncio.TimeoutError: latency_ms = (asyncio.get_event_loop().time() - start) * 1000 return FetchResult( org_id=org_id, success=False, error=f"Timeout after {timeout}s", latency_ms=latency_ms, ) except aiohttp.ClientResponseError as e: return FetchResult( org_id=org_id, success=False, error=f"HTTP {e.status}: {e.message}", ) except aiohttp.ClientError as e: return FetchResult(org_id=org_id, success=False, error=str(e)) async def fetch_all_users_good() -> list[FetchResult]: """Structured concurrency with TaskGroup — all tasks tracked and errors aggregated.""" connector = aiohttp.TCPConnector(limit=50, limit_per_host=10) # Bounded connections async with aiohttp.ClientSession(connector=connector) as session: results: list[FetchResult] = [] async with asyncio.TaskGroup() as tg: task_results: dict[int, asyncio.Task[FetchResult]] = {} for org_id in range(100): # Each task is tracked in a dict keyed by org_id task = tg.create_task(fetch_org_users(session, org_id)) task_results[org_id] = task # After TaskGroup exits, all tasks have completed (success or error) for org_id, task in sorted(task_results.items()): try: result = task.result() results.append(result) except Exception as e: # TaskGroup swallows errors during execution; we catch at collection results.append(FetchResult( org_id=org_id, success=False, error=f"Unhandled: {type(e).__name__}: {e}", )) return results # Usage — collect metrics and handle partial failures async def main(): results = await fetch_all_users_good() successful = [r for r in results if r.success] failed = [r for r in results if not r.success] print(f"Success: {len(successful)}/{len(results)}") print(f"Failed: {len(failed)}/{len(results)}") # Analyze latency distribution latencies = [r.latency_ms for r in successful] if latencies: avg_latency = sum(latencies) / len(latencies) p95_latency = sorted(latencies)[int(len(latencies) * 0.95)] print(f"Average latency: {avg_latency:.1f}ms, P95: {p95_latency:.1f}ms") # Process failures with retry or alerting for f in failed[:5]: # Log first 5 failures print(f" ❌ Org {f.org_id}: {f.error}") ``` ### Pattern 2: Go Goroutine Pool with Context Cancellation & Semaphore Limiting Bounded worker pool using a semaphore channel pattern with context propagation for clean cancellation. ```go package main import ( "context" "fmt" "io" "net/http" "sync" "time" ) // FetchResult holds the outcome of a single org user fetch. type FetchResult struct { OrgID int `json:"org_id"` Success bool `json:"success"` UserCount int `json:"user_count,omitempty"` Error string `json:"error,omitempty"` LatencyMs float64 `json:"latency_ms,omitempty"` } // WorkerPool manages a bounded set of goroutines with context-aware cancellation. type WorkerPool struct { sem chan struct{} // Semaphore channel — limits concurrency wg sync.WaitGroup // Tracks in-flight workers results chan FetchResult // Channel for collecting results ctx context.Context cancel context.CancelFunc } // NewWorkerPool creates a pool with maxConcurrency simultaneous goroutines. func NewWorkerPool(maxConcurrency int, ctx context.Context) *WorkerPool { ctx, cancel := context.WithCancel(ctx) return &WorkerPool{ sem: make(chan struct{}, maxConcurrency), results: make(chan FetchResult, 1024), // Buffered to prevent goroutine leak ctx: ctx, cancel: cancel, } } // Start launches the result collector goroutine that drains the results channel. func (wp *WorkerPool) Start() <-chan FetchResult { return wp.results } // Submit enqueues a fetch job. Returns false if the pool is already done. func (wp *WorkerPool) Submit(orgID int) bool { select { case wp.sem <- struct{}{}: // Acquire semaphore case <-wp.ctx.Done(): return false default: return false } wp.wg.Add(1) go func() { defer func() { <-wp.sem // Release semaphore wp.wg.Done() // Signal completion }() result := fetchOrgUsers(wp.ctx, orgID) select { case wp.results <- result: case <-wp.ctx.Done(): // Pool cancelled — discard result } }() return true } // Wait blocks until all submitted jobs complete or context is cancelled. func (wp *WorkerPool) Wait() { wp.wg.Wait() close(wp.results) // Close results channel after all workers finish } // Stop cancels the pool context and waits for all workers to finish. func (wp *WorkerPool) Stop() { wp.cancel() // Signal cancellation to all in-flight goroutines wp.Wait() // Wait for graceful shutdown } // fetchOrgUsers performs an HTTP GET with context timeout. func fetchOrgUsers(ctx context.Context, orgID int) FetchResult { start := time.Now() ctx, cancel := context.WithTimeout(ctx, 5*time.Second) defer cancel() req, err := http.NewRequestWithContext(ctx, "GET", fmt.Sprintf("https://api.example.com/orgs/%d/users", orgID), nil) if err != nil { return FetchResult{OrgID: orgID, Success: false, Error: fmt.Sprintf("request build: %v", err)} } req.Header.Set("Accept", "application/json") resp, err := http.DefaultClient.Do(req) if err != nil { if ctx.Err() == context.DeadlineExceeded { return FetchResult{OrgID: orgID, Success: false, Error: "timeout after 5s"} } return FetchResult{OrgID: orgID, Success: false, Error: fmt.Sprintf("request failed: %v", err)} } defer resp.Body.Close() body, err := io.ReadAll(resp.Body) if err != nil { return FetchResult{OrgID: orgID, Success: false, Error: fmt.Sprintf("read body: %v", err)} } userCount := len(body) // Simplified; in production parse JSON return FetchResult{ OrgID: orgID, Success: true, UserCount: userCount, LatencyMs: float64(time.Since(start).Milliseconds()), } } // --- Usage --- func main() { ctx, cancel := context.WithTimeout(context.Background(), 30*time.Second) defer cancel() pool := NewWorkerPool(20, ctx) // Max 20 concurrent goroutines results := pool.Start() // Submit work — each org gets a bounded goroutine for orgID := 0; orgID < 100; orgID++ { if !pool.Submit(orgID) { fmt.Printf("Pool cancelled, stopping submissions at org %d\n", orgID) break } } // Wait for all workers to finish, then drain results go pool.Wait() var successful, failed int for result := range results { if result.Success { successful++ } else { failed++ if failed <= 5 { // Log first 5 failures fmt.Printf(" ❌ Org %d: %s\n", result.OrgID, result.Error) } } } pool.Stop() fmt.Printf("\nSuccess: %d/%d | Failed: %d/%d\n", successful, successful+failed, failed, successful+failed) } ``` ### Pattern 3: Structured Concurrency — BAD vs GOOD Demonstrating the difference between fire-and-forget concurrency and parent-child lifecycle management. ```python # ❌ BAD — Fire-and-forget goroutine/task with no lifecycle management import asyncio import aiohttp async def process_payment_bad(user_id: int, payment: dict): """Launches background tasks that can outlive the request handler.""" async def send_receipt(): """Background task — if user cancels request, this still runs!""" async with aiohttp.ClientSession() as session: await session.post( "https://api.example.com/receipts", json={"user_id": user_id, "payment": payment}, ) async def update_analytics(): """Another background task — no way to cancel from the caller.""" await asyncio.sleep(0.1) # Simulate work await fetch_and_store_metrics(payment) # Fire-and-forget: these run until completion regardless of what happens to the parent asyncio.create_task(send_receipt()) asyncio.create_task(update_analytics()) # Handler returns immediately — but background tasks still consume resources # ✅ GOOD — Structured concurrency with parent-child lifecycle and cancellation import asyncio from contextlib import asynccontextmanager @asynccontextmanager async def managed_workers(ctx: asyncio.CancelledError | None = None): """Context manager that manages a group of child tasks. When the context exits (normally or via cancellation), all child tasks are cancelled and awaited for cleanup. This is structured concurrency. """ children: list[asyncio.Task] = [] try: yield children # Caller spawns child tasks here finally: # Guarantee: cancel any remaining children and wait for their cleanup for task in children: if not task.done(): task.cancel() # Wait for all tasks to finish (including cancellation handlers) if children: await asyncio.gather(*children, return_exceptions=True) async def process_payment_good(user_id: int, payment: dict): """Background work runs within a structured scope — cancellable on exit.""" async def send_receipt(): try: async with aiohttp.ClientSession() as session: await session.post( "https://api.example.com/receipts", json={"user_id": user_id, "payment": payment}, timeout=aiohttp.ClientTimeout(total=10), ) except asyncio.CancelledError: # Cleanup on cancellation — close the session properly print(f"[receipt] Cancelled for user {user_id}") raise async def update_analytics(): try: await fetch_and_store_metrics(payment) except asyncio.CancelledError: print(f"[analytics] Cancelled for user {user_id}") raise # All children are managed by the context manager — they share the parent's lifespan async with managed_workers() as children: children.append(asyncio.create_task(send_receipt())) children.append(asyncio.create_task(update_analytics())) # If this coroutine is cancelled (e.g., client disconnects), # both send_receipt and update_analytics are cancelled automatically # Alternative Go pattern: context.WithCancel for parent-child lifecycle """ func processPaymentGood(ctx context.Context, userID int, payment Payment) error { // Create a cancellable child context for background work bgCtx, cancel := context.WithCancel(ctx) defer cancel() // Guaranteed cleanup var wg sync.WaitGroup var mu sync.Mutex var errors []error // Worker 1: Send receipt wg.Add(1) go func() { defer wg.Done() if err := sendReceipt(bgCtx, userID, payment); err != nil { mu.Lock() errors = append(errors, fmt.Errorf("receipt: %w", err)) mu.Unlock() } }() // Worker 2: Update analytics wg.Add(1) go func() { defer wg.Done() if err := updateAnalytics(bgCtx, payment); err != nil { mu.Lock() errors = append(errors, fmt.Errorf("analytics: %w", err)) mu.Unlock() } }() // Wait for both workers; ctx cancellation cancels both via bgCtx wg.Wait() if len(errors) > 0 { return fmt.Errorf("background tasks failed (%d errors)", len(errors)) } return nil } """ ``` --- ## Constraints ### MUST DO - Match the concurrency model to the workload: use asyncio/goroutines for I/O-bound workloads, multiprocessing for CPU-bound workloads - Always launch concurrent units within a bounded structure — `asyncio.TaskGroup`, semaphore pools, or `sync.WaitGroup` — never fire-and-forget - Propagate cancellation from the top-level entry point through every async layer using contexts (`context.Context` in Go, `asyncio.CancelledError` + context managers in Python) - Set explicit timeouts on all external calls (HTTP requests, database queries, cache operations) to prevent indefinite hangs - Aggregate errors across await boundaries — collect failures without crashing the parent task; never swallow exceptions silently - Run race condition detectors (`go test -race`, `pytest-asyncio --strict`) on all async code before merging - Benchmark async code against synchronous baselines and document throughput/latency improvements ### MUST NOT DO - Launch goroutines or tasks that capture loop variables in closures without creating per-iteration copies (Go closure trap) - Share mutable state between concurrent units without explicit synchronization — use channels, mutexes, or actor patterns - Use `asyncio.create_task()` without tracking the returned task object anywhere in the codebase - Catch and silently suppress exceptions (`except Exception: pass` or `defer recover()` without logging) - Block on async operations using `.result()` with no timeout (can deadlock the event loop) - Create unbounded worker pools — always limit concurrency with semaphores, buffered channels, or pool size parameters - Mix blocking I/O calls (e.g., `requests.get()`, `http.Get()`) in the same event loop that handles async work — use async-compatible libraries --- ## Output Template When implementing or reviewing async code, produce: 1. **Concurrency Architecture** — Diagram or description of the concurrency model chosen (asyncio TaskGroup, goroutine pool, etc.) and why it matches the workload type 2. **Bounded Concurrency Plan** — Maximum concurrency level, semaphore/pool configuration, and backpressure strategy when limits are reached 3. **Cancellation & Timeout Strategy** — Context propagation chain, timeout values for each external call, and cancellation cleanup procedures 4. **Error Aggregation Design** — How failures from concurrent units are collected, logged, and whether they trigger retries or alerts 5. **Shared State Protection Plan** — Which synchronization primitives protect which data structures; evidence that race detector runs clean 6. **Benchmark Results** — Throughput (req/s), latency percentiles (p50/p95/p99), and memory usage comparing async vs sync baselines --- ## Live References | Resource | URL | |----------|-----| | Python asyncio Documentation | https://docs.python.org/3/library/asyncio.html | | Go context Package Documentation | https://pkg.go.dev/context | | Structured Concurrency Paper (Harrison) | https://vorpus.org/blog/notes-on-structured-concurrency-or-go-statement-considered-harmful/ | | Go errgroup Package | https://pkg.go.dev/golang.org/x/sync/errgroup | | Python Race Detection & asyncio Testing | https://pytest-asyncio.readthedocs.io/en/latest/ | --- ## Related Skills | Skill | Purpose | |---|---| | `api-design` | Build async I/O into API handlers for high-throughput endpoint serving | | `automated-testing` | Write tests for async code including race detection, timeout validation, and concurrent stress tests |