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.

# ❌ 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.

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.

# ❌ 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

Related Skills