Go — Concurrency
Goroutines, channels, select, context
Goroutines
A goroutine is a lightweight, cooperatively-scheduled function execution managed by the Go runtime. Starting one costs roughly 2–8 KB of stack (vs. ~1 MB for an OS thread) and the runtime multiplexes thousands of goroutines onto a small pool of OS threads. `go f()` starts `f` concurrently — no callbacks, no futures, just a keyword.
package main
import (
"fmt"
"time"
)
func worker(id int) {
fmt.Printf("Worker %d starting\n", id)
time.Sleep(100 * time.Millisecond) // simulate work
fmt.Printf("Worker %d done\n", id)
}
func main() {
for i := 1; i <= 5; i++ {
go worker(i) // launch concurrently
}
time.Sleep(500 * time.Millisecond) // wait for workers (use sync.WaitGroup in real code)
}Channels
Channels are typed, goroutine-safe communication pipes. `make(chan T)` creates an unbuffered channel — a send blocks until a receiver is ready, and vice versa. `make(chan T, n)` creates a buffered channel that holds up to `n` values without blocking. The rule: *Do not communicate by sharing memory; share memory by communicating.*
// Unbuffered channel — synchronizes sender and receiver
ch := make(chan int)
go func() { ch <- 42 }() // sender blocks until receiver is ready
val := <-ch // receiver blocks until sender sends
fmt.Println(val) // 42
// Buffered channel — sender only blocks when buffer is full
buf := make(chan string, 3)
buf <- "a"
buf <- "b"
buf <- "c"
// buf <- "d" would block here
// Close signals no more values; range drains the channel
close(buf)
for s := range buf {
fmt.Println(s) // a, b, c
}Select
`select` waits on multiple channel operations simultaneously, executing the first one that is ready. If multiple are ready, it picks one at random. A `default` case makes `select` non-blocking. `select` is the primary tool for timeouts, cancellation, and fan-in patterns.
func fetchWithTimeout(url string, timeout time.Duration) (string, error) {
resultCh := make(chan string, 1)
go func() {
resultCh <- fetch(url) // assume fetch returns string
}()
select {
case result := <-resultCh:
return result, nil
case <-time.After(timeout):
return "", fmt.Errorf("timeout after %v", timeout)
}
}sync.WaitGroup
`sync.WaitGroup` coordinates a set of goroutines: `Add(n)` registers n tasks, each goroutine calls `Done()` when finished, and `Wait()` blocks until the count reaches zero. It is the standard way to launch a batch of goroutines and wait for all of them to complete before proceeding.
import "sync"
func processAll(items []string) {
var wg sync.WaitGroup
for _, item := range items {
wg.Add(1)
go func(item string) {
defer wg.Done()
process(item)
}(item) // pass item to avoid closure capture (pre-Go1.22)
}
wg.Wait() // blocks until all goroutines call Done
fmt.Println("all done")
}sync.Mutex and sync.RWMutex
When goroutines share mutable state (instead of communicating via channels), a mutex serializes access. `sync.Mutex` provides exclusive locking (`Lock`/`Unlock`). `sync.RWMutex` allows multiple concurrent readers OR one exclusive writer — use it when reads vastly outnumber writes.
type SafeCounter struct {
mu sync.Mutex
count int
}
func (c *SafeCounter) Inc() {
c.mu.Lock()
defer c.mu.Unlock()
c.count++
}
func (c *SafeCounter) Value() int {
c.mu.Lock()
defer c.mu.Unlock()
return c.count
}
// RWMutex for read-heavy cache
type Cache struct {
mu sync.RWMutex
data map[string]string
}
func (c *Cache) Get(key string) (string, bool) {
c.mu.RLock() // multiple readers allowed simultaneously
defer c.mu.RUnlock()
v, ok := c.data[key]
return v, ok
}context.Context
The `context` package propagates deadlines, cancellation signals, and request-scoped values across goroutines and API boundaries. `context.WithTimeout` and `context.WithCancel` return a derived context and a cancel function. When the context is cancelled (or times out), all operations using it receive a `ctx.Done()` signal and should clean up and return.
func fetchData(ctx context.Context, url string) ([]byte, error) {
req, err := http.NewRequestWithContext(ctx, "GET", url, nil)
if err != nil {
return nil, err
}
resp, err := http.DefaultClient.Do(req)
if err != nil {
return nil, err // includes context.DeadlineExceeded or context.Canceled
}
defer resp.Body.Close()
return io.ReadAll(resp.Body)
}
// Caller
ctx, cancel := context.WithTimeout(context.Background(), 3*time.Second)
defer cancel() // always cancel to release resources
data, err := fetchData(ctx, "https://api.example.com/data")Worker Pools
A worker pool limits concurrency: instead of launching one goroutine per task (which can overwhelm resources), a fixed number of worker goroutines read from a shared task channel. This pattern controls memory usage and prevents thundering herds when processing large batches.
func workerPool(tasks []Task, numWorkers int) []Result {
taskCh := make(chan Task, len(tasks))
resultCh := make(chan Result, len(tasks))
// Start fixed pool of workers
var wg sync.WaitGroup
for i := 0; i < numWorkers; i++ {
wg.Add(1)
go func() {
defer wg.Done()
for task := range taskCh {
resultCh <- process(task)
}
}()
}
// Send all tasks
for _, t := range tasks {
taskCh <- t
}
close(taskCh) // signals workers: no more tasks
// Wait for workers, then close results
go func() { wg.Wait(); close(resultCh) }()
var results []Result
for r := range resultCh {
results = append(results, r)
}
return results
}sync/atomic
The `sync/atomic` package provides lock-free, hardware-atomic operations on primitive types (`int32`, `int64`, `uint64`, `Pointer`, etc.). Atomic operations are faster than mutexes for simple counters or flags because they use CPU instructions (CAS, fetch-and-add) rather than OS-level locking.
import "sync/atomic"
var requests atomic.Int64 // zero value is valid
func handleRequest() {
requests.Add(1)
// ... handle request
}
func stats() {
fmt.Printf("Total requests: %d\n", requests.Load())
}
// Atomic compare-and-swap (CAS) — update only if current value matches expected
var flag atomic.Bool
if flag.CompareAndSwap(false, true) {
fmt.Println("acquired flag")
}Pipelines
A pipeline is a series of stages connected by channels, where each stage receives values from upstream, transforms them, and sends to downstream. Pipelines make concurrent data processing composable and readable. `done` channels (or `context.Context`) propagate cancellation through the pipeline.
// Stage 1: generate integers
func generate(nums ...int) <-chan int {
out := make(chan int)
go func() {
for _, n := range nums {
out <- n
}
close(out)
}()
return out
}
// Stage 2: square each integer
func square(in <-chan int) <-chan int {
out := make(chan int)
go func() {
for n := range in {
out <- n * n
}
close(out)
}()
return out
}
// Wire stages together
func main() {
for n := range square(generate(2, 3, 4)) {
fmt.Println(n) // 4, 9, 16
}
}sync.Once
`sync.Once` guarantees a function runs exactly once, no matter how many goroutines call it or how often. The first call to `once.Do(f)` runs `f`; every later call blocks until that first run completes, then returns immediately without re-running. It is the idiomatic, race-free way to do lazy one-time initialization.
import "sync"
var (
once sync.Once
instance *Database
)
func GetDatabase() *Database {
once.Do(func() {
// Runs exactly once, even under concurrent calls
instance = connectToDatabase()
})
return instance
}sync.Pool
`sync.Pool` is a cache of temporary objects that can be reused to relieve pressure on the garbage collector. `Get` returns an object (creating one via the `New` func if the pool is empty); `Put` returns it for reuse. The pool may drop its contents at any time — never store anything that must persist.
import (
"bytes"
"sync"
)
var bufPool = sync.Pool{
New: func() any { return new(bytes.Buffer) },
}
func handle() {
buf := bufPool.Get().(*bytes.Buffer)
buf.Reset() // always reset reused objects
defer bufPool.Put(buf) // return it for the next caller
buf.WriteString("response body")
// ... use buf ...
}Channel Direction
A channel type can be restricted to send-only (`chan<- T`) or receive-only (`<-chan T`). A bidirectional `chan T` converts implicitly to either restricted form, but not the reverse. Declaring direction in function signatures documents intent and lets the compiler catch misuse — a producer can't accidentally read, a consumer can't accidentally write.
// Send-only parameter — function may only write
func produce(out chan<- int) {
for i := 0; i < 3; i++ {
out <- i
}
close(out)
}
// Receive-only parameter — function may only read
func consume(in <-chan int) {
for v := range in {
fmt.Println(v)
}
}
ch := make(chan int) // bidirectional
go produce(ch) // passed as chan<- int
consume(ch) // passed as <-chan intThe Race Detector
A data race occurs when two goroutines access the same memory concurrently and at least one writes, with no synchronization. Go ships a built-in race detector enabled with the `-race` flag (`go run -race`, `go test -race`, `go build -race`). It instruments memory accesses and reports the conflicting goroutines, stacks, and addresses at runtime.
// This code has a data race — run with: go run -race race_detector.go
var counter int
func main() {
var wg sync.WaitGroup
for i := 0; i < 100; i++ {
wg.Add(1)
go func() {
defer wg.Done()
counter++ // RACE: unsynchronized write
}()
}
wg.Wait()
}
// The fix: protect counter with sync.Mutex or use atomic.Int64.
// Always run tests with -race in CI to catch races early.The Global Interpreter Lock (GIL)
The GIL is a mutex in CPython that allows only one thread to execute Python bytecode at a time, even on multi-core CPUs. It exists because CPython's memory management (reference counting) is not thread-safe. The GIL makes single-threaded programs faster (no locking overhead) and C extension writing simpler. The consequence: CPU-bound multi-threaded Python programs don't scale across cores. I/O-bound workloads are largely unaffected — the GIL is released while waiting for I/O. Python 3.13 introduced an optional free-threaded build (`python3.13t`) that removes the GIL.
import threading
import time
def cpu_bound(n: int) -> int:
"""Count-down loop — pure Python, GIL-bound."""
while n > 0:
n -= 1
return n
N = 10_000_000
# Single-threaded baseline
start = time.perf_counter()
cpu_bound(N)
cpu_bound(N)
single = time.perf_counter() - start
# Multi-threaded — GIL prevents true parallelism
t1 = threading.Thread(target=cpu_bound, args=(N,))
t2 = threading.Thread(target=cpu_bound, args=(N,))
start = time.perf_counter()
t1.start(); t2.start()
t1.join(); t2.join()
multi = time.perf_counter() - start
print(f"Single: {single:.2f}s")
print(f"Multi: {multi:.2f}s (should be ~same or slower due to GIL)")`threading` — OS-Level Threads
`threading.Thread` runs a callable in an OS thread. Threads share the process's memory — communication is via shared objects, protected by `threading.Lock`, `RLock`, `Semaphore`, `Event`, `Condition`, and `Barrier`. `threading.local()` provides thread-local storage. Threads are well-suited for I/O-bound concurrency: database queries, HTTP requests, file I/O. A `Daemon` thread dies automatically when the main thread exits.
import threading
import time
import random
from queue import Queue
results: Queue[tuple[str, int]] = Queue()
def fetch(url: str) -> None:
"""Simulates an HTTP request with random latency."""
time.sleep(random.uniform(0.05, 0.3))
status = 200 if random.random() > 0.1 else 500
results.put((url, status))
urls = [f"https://api.example.com/item/{i}" for i in range(10)]
# Thread pool via list
start = time.perf_counter()
threads = [threading.Thread(target=fetch, args=(u,)) for u in urls]
for t in threads: t.start()
for t in threads: t.join()
elapsed = time.perf_counter() - start
print(f"Fetched {results.qsize()} URLs in {elapsed:.2f}s")
successes = sum(1 for _ in range(results.qsize()) if results.get()[1] == 200)`multiprocessing` — True Parallelism
`multiprocessing.Process` launches a separate Python interpreter process — no GIL, true CPU parallelism. `Pool.map()` distributes work across N worker processes. `multiprocessing.Queue` and `Pipe` enable inter-process communication (IPC). `Value` and `Array` provide shared memory. The `spawn` start method (default on Windows/macOS) re-imports the module in each worker — guard the entry point with `if __name__ == "__main__":`. `ProcessPoolExecutor` (from `concurrent.futures`) is the higher-level API.
import multiprocessing
import os
import time
def is_prime(n: int) -> bool:
if n < 2: return False
for i in range(2, int(n**0.5) + 1):
if n % i == 0: return False
return True
def count_primes_in_range(args: tuple[int, int]) -> int:
lo, hi = args
return sum(1 for n in range(lo, hi) if is_prime(n))
if __name__ == "__main__":
N = 200_000
num_workers = os.cpu_count() or 4
chunk_size = N // num_workers
ranges = [(i * chunk_size, (i + 1) * chunk_size) for i in range(num_workers)]
start = time.perf_counter()
with multiprocessing.Pool(num_workers) as pool:
counts = pool.map(count_primes_in_range, ranges)
elapsed = time.perf_counter() - start
print(f"Primes below {N}: {sum(counts)}")
print(f"Time ({num_workers} workers): {elapsed:.2f}s")`concurrent.futures` — High-Level Executor Interface
`concurrent.futures` provides `ThreadPoolExecutor` (threads, good for I/O) and `ProcessPoolExecutor` (processes, good for CPU). Both expose `submit()` → `Future` and `map()`. `Future.result()` blocks until complete; `as_completed()` yields futures as they finish (fastest first). The executor is a context manager — `__exit__` calls `shutdown(wait=True)`. Switching between thread and process pools requires changing only one import.
from concurrent.futures import ThreadPoolExecutor, as_completed
import time
import random
def download(url: str) -> tuple[str, int]:
"""Simulates downloading a file."""
time.sleep(random.uniform(0.1, 0.5))
size = random.randint(1024, 1_048_576)
return url, size
urls = [f"https://cdn.example.com/file{i}.dat" for i in range(12)]
with ThreadPoolExecutor(max_workers=4) as executor:
futures = {executor.submit(download, url): url for url in urls}
total_bytes = 0
for future in as_completed(futures):
url = futures[future]
_, size = future.result()
total_bytes += size
print(f" Downloaded {url.split('/')[-1]} ({size:,} bytes)")
print(f"\nTotal: {total_bytes:,} bytes across {len(urls)} files")`asyncio` — Cooperative Concurrency
`asyncio` runs many coroutines on a single thread using cooperative multitasking. A coroutine (defined with `async def`) yields control at `await` points. The event loop drives execution, switching between coroutines whenever one awaits I/O. `asyncio.run()` is the entry point. `asyncio.gather()` runs multiple coroutines concurrently. `asyncio.sleep(0)` yields control without actually sleeping. All I/O must use `async`-aware libraries (`aiohttp`, `asyncpg`, `aiofiles`) — calling blocking I/O inside a coroutine blocks the entire event loop.
import asyncio
import random
async def fetch(session_id: int, url: str) -> dict:
"""Simulates an async HTTP request."""
await asyncio.sleep(random.uniform(0.05, 0.3))
return {"session": session_id, "url": url, "status": 200}
async def main() -> None:
urls = [f"https://api.example.com/v1/item/{i}" for i in range(8)]
# Run all concurrently, collect results
tasks = [fetch(i, url) for i, url in enumerate(urls)]
results = await asyncio.gather(*tasks)
for r in results:
print(f"[{r['session']}] {r['url']} → {r['status']}")
print(f"\nCompleted {len(results)} requests")
asyncio.run(main())Coroutines and `async`/`await`
A coroutine is a function defined with `async def` that returns a coroutine object when called (it does not run immediately). `await expr` suspends the coroutine until the awaitable (coroutine, `Task`, or `Future`) completes. `async for` iterates an async iterable; `async with` uses an async context manager. `asyncio.create_task()` schedules a coroutine as a background task — unlike `await`, the caller continues without waiting.
import asyncio
async def slow_operation(name: str, delay: float) -> str:
print(f" Starting {name}")
await asyncio.sleep(delay)
print(f" Finished {name}")
return f"{name}: done"
async def sequential() -> None:
# Each await blocks until completion — total ≈ 0.3 + 0.2 + 0.1s
r1 = await slow_operation("A", 0.3)
r2 = await slow_operation("B", 0.2)
r3 = await slow_operation("C", 0.1)
print([r1, r2, r3])
async def concurrent() -> None:
# All three run concurrently — total ≈ max(0.3, 0.2, 0.1)s
t1 = asyncio.create_task(slow_operation("X", 0.3))
t2 = asyncio.create_task(slow_operation("Y", 0.2))
t3 = asyncio.create_task(slow_operation("Z", 0.1))
results = await asyncio.gather(t1, t2, t3)
print(results)
print("=== Sequential ===")
asyncio.run(sequential())
print("\n=== Concurrent ===")
asyncio.run(concurrent())`asyncio.Queue` — Async Producer-Consumer
`asyncio.Queue` is a thread-safe, async-friendly queue for producer-consumer patterns within an asyncio event loop. `await queue.put(item)` blocks if the queue is full (maxsize > 0); `await queue.get()` blocks until an item is available. `queue.task_done()` signals completion; `await queue.join()` waits until all items are processed — the canonical pattern for bounded work queues.
import asyncio
import random
async def producer(queue: asyncio.Queue, n: int) -> None:
for i in range(n):
item = f"item-{i}"
await queue.put(item)
print(f" Produced {item}")
await asyncio.sleep(random.uniform(0.01, 0.05))
# Signal consumers to stop
for _ in range(3): # 3 consumers
await queue.put(None)
async def consumer(name: str, queue: asyncio.Queue) -> int:
processed = 0
while True:
item = await queue.get()
if item is None:
queue.task_done()
break
await asyncio.sleep(random.uniform(0.02, 0.08)) # simulate work
queue.task_done()
processed += 1
return processed
async def main() -> None:
queue: asyncio.Queue[str | None] = asyncio.Queue(maxsize=5)
prod = asyncio.create_task(producer(queue, 10))
consumers = [asyncio.create_task(consumer(f"C{i}", queue)) for i in range(3)]
await prod
counts = await asyncio.gather(*consumers)
print(f"Processed: {counts}, total={sum(counts)}")
asyncio.run(main())Locks, Events, and Semaphores
`threading.Lock` is a mutual exclusion lock — only one thread can hold it at a time. `RLock` (reentrant lock) can be acquired multiple times by the same thread. `threading.Event` coordinates threads: `event.wait()` blocks until `event.set()` is called. `threading.Semaphore(n)` limits concurrent access to `n` threads. `asyncio` mirrors all these: `asyncio.Lock`, `asyncio.Event`, `asyncio.Semaphore` — same semantics but `await`-based.
import asyncio
async def rate_limited_fetch(sem: asyncio.Semaphore, url: str) -> str:
async with sem: # At most 3 concurrent requests
await asyncio.sleep(0.1) # simulate network I/O
return f"OK: {url}"
async def main() -> None:
sem = asyncio.Semaphore(3) # max 3 concurrent
urls = [f"https://api.example.com/{i}" for i in range(10)]
results = await asyncio.gather(*[rate_limited_fetch(sem, u) for u in urls])
print(f"Completed {len(results)} requests with concurrency limit 3")
asyncio.run(main())
# threading.Event example
import threading
ready = threading.Event()
def worker():
ready.wait() # Block until main signals
print("Worker: starting work")
t = threading.Thread(target=worker)
t.start()
print("Main: initializing...")
import time; time.sleep(0.1)
ready.set() # Unblock the worker
t.join()Shared State and Thread Safety
Sharing mutable state between threads without synchronization causes race conditions. Use `threading.Lock` to protect shared state. `queue.Queue` is inherently thread-safe — prefer it over passing data via shared variables. Immutable objects (tuples, frozensets, numbers) need no locking. `collections.deque` is thread-safe for `appendleft`/`popleft` from different ends. Avoid `global` mutable state in threaded code — encapsulate in locked objects or thread-local storage.
import threading
class ThreadSafeCounter:
def __init__(self) -> None:
self._value = 0
self._lock = threading.Lock()
def increment(self) -> None:
with self._lock:
self._value += 1
@property
def value(self) -> int:
with self._lock:
return self._value
counter = ThreadSafeCounter()
def worker(n: int) -> None:
for _ in range(n):
counter.increment()
threads = [threading.Thread(target=worker, args=(1000,)) for _ in range(10)]
for t in threads: t.start()
for t in threads: t.join()
print(f"Counter: {counter.value}") # Always 10000 with the lockPerformance Patterns — Choosing the Right Concurrency Model
Python offers three concurrency models: `threading` (I/O-bound, shared memory, GIL limited), `multiprocessing` (CPU-bound, process isolation, higher overhead), and `asyncio` (I/O-bound, single-thread, highest concurrency per memory). The rules: I/O-bound + simple → `asyncio`; I/O-bound + sync libraries → `threading`; CPU-bound → `multiprocessing` or C extensions (NumPy, Cython). Mix: use `asyncio.to_thread()` to run blocking code in a thread pool from an async context.
import asyncio
import time
import concurrent.futures
def cpu_heavy(n: int) -> int:
"""Blocks the thread — not suitable to call directly in async."""
return sum(i * i for i in range(n))
async def mixed_workload() -> None:
# I/O-bound: run concurrently in the event loop
io_tasks = [asyncio.sleep(0.1) for _ in range(5)]
await asyncio.gather(*io_tasks)
# CPU-bound: offload to a thread pool
loop = asyncio.get_event_loop()
with concurrent.futures.ThreadPoolExecutor() as pool:
result = await loop.run_in_executor(pool, cpu_heavy, 1_000_000)
print(f"CPU result: {result}")
# Python 3.9+ shorthand
result2 = await asyncio.to_thread(cpu_heavy, 500_000)
print(f"to_thread result: {result2}")
asyncio.run(mixed_workload())`asyncio.TaskGroup` and `ExceptionGroup` (Python 3.11+)
`asyncio.TaskGroup` (PEP 654) is the structured concurrency primitive for asyncio: all tasks in a group are cancelled if one raises, and multiple exceptions are bundled into an `ExceptionGroup`. `except*` syntax handles specific exception types from a group, ignoring others. This replaces the fragile pattern of manually cancelling tasks on failure and re-raising the first exception — common bugs with bare `gather()`.
import asyncio
import random
async def fetch_service(name: str, fail: bool = False) -> str:
await asyncio.sleep(random.uniform(0.05, 0.2))
if fail:
raise ValueError(f"{name} returned an error")
return f"{name}: OK"
async def main() -> None:
# TaskGroup: structured concurrency (Python 3.11+)
try:
async with asyncio.TaskGroup() as tg:
t1 = tg.create_task(fetch_service("auth"))
t2 = tg.create_task(fetch_service("inventory"))
t3 = tg.create_task(fetch_service("pricing", fail=True))
except* ValueError as eg:
print(f"Some services failed: {[str(e) for e in eg.exceptions]}")
return
print([t1.result(), t2.result(), t3.result()])
asyncio.run(main())