Backpressure — Middle Level¶

Table of Contents¶

1. Introduction
2. Prerequisites
3. Glossary
4. From Channels to Semaphores
5. Weighted Semaphores
6. Worker Pools That Mean It
7. Pipeline Backpressure
8. Propagating Context
9. Load Shedding Strategies
10. Watermarks and Hysteresis
11. Per-Tenant and Per-Class Queues
12. Observability Hooks
13. Code Examples
14. Coding Patterns
15. Clean Code
16. Product Use / Feature
17. Error Handling
18. Security Considerations
19. Performance Tips
20. Best Practices
21. Edge Cases & Pitfalls
22. Common Mistakes
23. Common Misconceptions
24. Tricky Points
25. Test
26. Tricky Questions
27. Cheat Sheet
28. Self-Assessment Checklist
29. Summary
30. What You Can Build
31. Further Reading
32. Related Topics
33. Diagrams & Visual Aids

Introduction¶

Focus: "Semaphores, pipeline backpressure, per-class queues, watermarks, and observability."

At the junior level, backpressure was a property of one channel. At the middle level, backpressure becomes a property of the whole pipeline: producer, queue, workers, downstream service, and shutdown — every link must participate. A single missing link defeats the rest.

This page covers:

• Semaphores as a generalisation of bounded channels — sometimes more natural than a queue.
• Weighted semaphores (golang.org/x/sync/semaphore) when items have different "cost."
• Worker pools with Submit, TrySubmit, SubmitCtx, drop counters, and graceful shutdown.
• Pipeline backpressure — slowing fan-in/fan-out stages without unbounded buffering.
• Context propagation so deadlines flow end-to-end.
• Load shedding strategies — drop newest, drop oldest, priority drop, sample drop.
• Watermarks and hysteresis — start and stop shedding at different thresholds.
• Per-tenant queues — preventing one noisy neighbour from starving the rest.
• Observability — metrics, traces, and logs that make backpressure visible.

You should already be comfortable with everything in junior.md — bounded channels, select with default, context with deadline, the bounded worker pool API. If those are still rough, go back.

Prerequisites¶

• Comfortable with Go channels, select, and context.Context.
• Have written or read a worker pool in production.
• Familiar with sync.WaitGroup, sync.Mutex, and atomic counters.
• Aware of runtime.NumCPU, runtime.NumGoroutine, and have read a goroutine dump.
• Have used a metrics library (Prometheus, expvar, OpenTelemetry) in a real service.

Glossary¶

From Channels to Semaphores¶

A bounded channel is a semaphore in disguise. Both limit concurrent work; the channel happens to carry data, the semaphore does not.

// Channel-as-queue: data flows through it.
jobs := make(chan Job, 100)
// Channel-as-semaphore: only the count matters.
sem := make(chan struct{}, 100)

When you use a chan struct{} purely for counting, you have a semaphore. Acquire by sending; release by receiving:

sem <- struct{}{} // acquire (blocks if full)
defer func() { <-sem }() // release

When to prefer a semaphore over a queue:

• The work is already in your goroutine — you do not need to hand it off, just to gate execution.
• Many goroutines need to share a single budget without serialising through one consumer.
• The work order is not strict FIFO; you want any free worker to start any item immediately.

When to prefer a queue:

• You want strict FIFO order.
• You want a fixed number of worker goroutines (rather than one goroutine per item).
• You need backpressure on the enqueue side, not on the work itself.

In practice, larger Go services use both: a queue for queued work, a semaphore for resource limits.

// Pattern: queue + semaphore for resource limit.
jobs := make(chan Job, 1000)         // shock absorber
sem  := make(chan struct{}, 8)        // CPU budget

for j := range jobs {
    sem <- struct{}{}
    go func(j Job) {
        defer func() { <-sem }()
        j.Run()
    }(j)
}

The queue provides bounded buffering between producer and dispatcher; the semaphore caps concurrent CPU work. Both are bounded; the system has both temporal smoothing and instantaneous limits.

Weighted Semaphores¶

When all work items are the same size, an unweighted semaphore suffices. When items vary — a 10 KB request and a 10 MB request both compete for memory — an unweighted semaphore is wrong: it would allow 8 large requests when you meant 1 large or 8 small.

The standard library does not have a weighted semaphore, but golang.org/x/sync/semaphore does:

import "golang.org/x/sync/semaphore"

s := semaphore.NewWeighted(64) // 64 units of "weight"

func (h *Handler) HandleSmall(ctx context.Context) error {
    if err := s.Acquire(ctx, 1); err != nil { return err }
    defer s.Release(1)
    // ...
}

func (h *Handler) HandleLarge(ctx context.Context) error {
    if err := s.Acquire(ctx, 16); err != nil { return err }
    defer s.Release(16)
    // ...
}

64 small jobs can run concurrently. 4 large jobs can run concurrently. 1 large + 48 small can run concurrently. The single budget mixes work of different sizes naturally.

Weights model real resources: memory, CPU cores, expected duration, downstream API quota. Pick a unit and stick with it across the codebase.

semaphore.Acquire blocks until weight is available or the context expires. There is no TryAcquire(ctx); for non-blocking, call s.TryAcquire(n) which returns immediately with true/false.

Worker Pools That Mean It¶

A production worker pool needs more than three methods. Here is a fuller API.

type Pool struct {
    jobs    chan Job
    sem     *semaphore.Weighted
    wg      sync.WaitGroup
    closed  atomic.Bool

    Stats Stats
}

type Stats struct {
    Submitted atomic.Uint64
    Completed atomic.Uint64
    Dropped   atomic.Uint64
    Rejected  atomic.Uint64
    Errored   atomic.Uint64
    InFlight  atomic.Int64
}

func NewPool(workers, buffer int, totalWeight int64) *Pool {
    p := &Pool{
        jobs: make(chan Job, buffer),
        sem:  semaphore.NewWeighted(totalWeight),
    }
    for i := 0; i < workers; i++ {
        p.wg.Add(1)
        go p.worker()
    }
    return p
}

func (p *Pool) worker() {
    defer p.wg.Done()
    for j := range p.jobs {
        p.Stats.InFlight.Add(1)
        if err := j.Run(); err != nil {
            p.Stats.Errored.Add(1)
        }
        p.Stats.Completed.Add(1)
        p.Stats.InFlight.Add(-1)
    }
}

func (p *Pool) Submit(j Job) {
    if p.closed.Load() {
        p.Stats.Rejected.Add(1)
        return
    }
    p.Stats.Submitted.Add(1)
    p.jobs <- j
}

func (p *Pool) TrySubmit(j Job) bool {
    if p.closed.Load() {
        return false
    }
    select {
    case p.jobs <- j:
        p.Stats.Submitted.Add(1)
        return true
    default:
        p.Stats.Dropped.Add(1)
        return false
    }
}

func (p *Pool) SubmitCtx(ctx context.Context, j Job) error {
    if p.closed.Load() {
        return errors.New("pool closed")
    }
    select {
    case p.jobs <- j:
        p.Stats.Submitted.Add(1)
        return nil
    case <-ctx.Done():
        p.Stats.Rejected.Add(1)
        return ctx.Err()
    }
}

func (p *Pool) Close(ctx context.Context) error {
    if !p.closed.CompareAndSwap(false, true) {
        return nil
    }
    close(p.jobs)
    done := make(chan struct{})
    go func() {
        p.wg.Wait()
        close(done)
    }()
    select {
    case <-done:
        return nil
    case <-ctx.Done():
        return ctx.Err()
    }
}

Notice the additions:

• closed flag prevents submits after Close.
• Stats exposes counters that map directly to Prometheus metrics.
• Close(ctx) has a deadline — operators can choose how long to wait for drain.

A pool with this API is a building block you can drop into any service.

Pipeline Backpressure¶

A pipeline is a chain of goroutines joined by channels. Backpressure must flow from the last stage to the first.

stage1Out := stage1(input)
stage2Out := stage2(stage1Out)
stage3Out := stage3(stage2Out)
// consumer reads stage3Out

If every stage uses a bounded channel for its output and a blocking write, backpressure is automatic: when stage 3 is slow, stage 2's output fills, stage 2 blocks, stage 1's output fills, stage 1 blocks, the source slows.

The trap is that all stages must be bounded. One unbounded slice or one make(chan T, 1<<20) in the middle defeats the chain.

func stage1(in <-chan A) <-chan B {
    out := make(chan B, 4) // small buffer to absorb jitter
    go func() {
        defer close(out)
        for a := range in {
            out <- transform(a) // blocks when stage2's input is full
        }
    }()
    return out
}

A common idiom is to thread a context.Context through every stage so the whole pipeline can be cancelled:

func stage1(ctx context.Context, in <-chan A) <-chan B {
    out := make(chan B, 4)
    go func() {
        defer close(out)
        for a := range in {
            select {
            case out <- transform(a):
            case <-ctx.Done():
                return
            }
        }
    }()
    return out
}

Without the ctx.Done() case, a stuck downstream means a stuck pipeline means a leaked goroutine.

Fan-out with bounded fan-in¶

A common pattern is fan-out (one input → N workers) then fan-in (N outputs → one consumer). The fan-in channel is the bottleneck if not bounded.

func fanOutIn(ctx context.Context, in <-chan A, workers int) <-chan B {
    out := make(chan B, workers) // bounded by worker count
    var wg sync.WaitGroup
    for i := 0; i < workers; i++ {
        wg.Add(1)
        go func() {
            defer wg.Done()
            for a := range in {
                b := transform(a)
                select {
                case out <- b:
                case <-ctx.Done():
                    return
                }
            }
        }()
    }
    go func() { wg.Wait(); close(out) }()
    return out
}

The output channel is bounded, the input channel is presumably bounded by the upstream, and every send respects context. This pipeline cannot leak goroutines and cannot grow memory.

Propagating Context¶

Every queue operation that can wait should accept a context. Every worker that does meaningful work should check ctx.Err() on long operations. This is the discipline that makes shutdowns clean and deadlines real.

func (w *Worker) Do(ctx context.Context, j Job) error {
    for _, step := range j.Steps {
        if err := ctx.Err(); err != nil {
            return err
        }
        if err := step.Run(ctx); err != nil {
            return err
        }
    }
    return nil
}

Inside the step:

func (s *Step) Run(ctx context.Context) error {
    ctx, cancel := context.WithTimeout(ctx, s.Budget)
    defer cancel()
    return s.client.Do(ctx, s.Request)
}

A request's context is the leash on every operation it triggers. When the leash expires, work in flight cancels itself; no orphaned goroutines, no zombie queue entries.

For long-lived workers (those that loop forever), the context should be the worker's context, not any particular request's context. Each request's ctx is derived from the worker's ctx for the duration of that request.

func (w *Worker) Run(workerCtx context.Context) {
    for {
        select {
        case j, ok := <-w.jobs:
            if !ok { return }
            reqCtx, cancel := context.WithTimeout(workerCtx, j.Budget)
            w.handle(reqCtx, j)
            cancel()
        case <-workerCtx.Done():
            return
        }
    }
}

Load Shedding Strategies¶

When you must drop, you can drop:

• Newest. Reject the next item. Simplest. Default.
• Oldest. Drop the oldest queued item, accept the new one. Useful when freshness matters (live data, telemetry).
• By priority. Drop low-priority items first.
• By tenant. Drop the noisiest tenant's items first.
• By size. Drop items whose weight (CPU, memory) is largest.
• Sampled. Drop a configurable percentage uniformly at random.

Each policy has a code shape.

Drop newest¶

select {
case ch <- j:
default:
    drops.Inc()
}

Drop oldest¶

for {
    select {
    case ch <- j:
        return
    default:
        select {
        case <-ch:
        default:
        }
    }
}

This pops one old item and tries again. Be careful — under heavy contention, this can spin. A safer version uses a buffer of size 1 and overwrites:

select {
case ch <- j:
default:
    <-ch
    ch <- j
}

By priority¶

type Item struct {
    Priority int
    Payload  any
}

// Use a heap, not a channel, when priority matters strictly.

Channels do not have priority. For priority queues, use container/heap behind a mutex; or two channels (high, low) read in a select that always tries high first via nested non-blocking selects.

Sampled drop¶

if rand.Float64() < 0.1 {
    drops.Inc()
    return // drop 10%
}
ch <- j

Useful for telemetry: keep approximately 90% of the data, never block, never grow.

Watermarks and Hysteresis¶

A single threshold ("shed when len(ch) > 80") can cause thrashing if the queue oscillates near the threshold. Two thresholds — high (start shedding) and low (stop shedding) — produce stable behaviour.

type Shedder struct {
    jobs       chan Job
    high, low  int
    shedding   atomic.Bool
}

func (s *Shedder) Submit(j Job) bool {
    n := len(s.jobs)
    if s.shedding.Load() {
        if n <= s.low {
            s.shedding.Store(false)
        }
    } else {
        if n >= s.high {
            s.shedding.Store(true)
        }
    }
    if s.shedding.Load() {
        return false
    }
    select {
    case s.jobs <- j:
        return true
    default:
        return false
    }
}

Pick high somewhere like 80% and low like 40% of capacity. The gap absorbs noise. Without hysteresis, every fluctuation around the threshold flips the state, producing alarm noise and uneven throughput.

Per-Tenant and Per-Class Queues¶

A single queue is vulnerable to one noisy neighbour. If Tenant A submits 10× the load of others, they fill the queue and starve everyone. Per-tenant queues fix this.

type Multiplexer struct {
    queues map[string]chan Job
    mu     sync.Mutex
}

func (m *Multiplexer) Submit(tenant string, j Job) bool {
    m.mu.Lock()
    q, ok := m.queues[tenant]
    if !ok {
        q = make(chan Job, 32)
        m.queues[tenant] = q
        go consumer(q)
    }
    m.mu.Unlock()
    select {
    case q <- j:
        return true
    default:
        return false // tenant-local shedding
    }
}

Now each tenant has their own bounded queue. A spike from one tenant fills only that tenant's queue. The drop is local; other tenants continue normally.

The cost: more goroutines and more state. For 10,000 tenants you would not give each one a goroutine — you would use a single worker pool that picks from per-tenant queues (a scheduler). Even then, per-tenant queues are the right model; the workers are shared.

Round-robin fairness¶

When workers must serve many tenant queues, round-robin scheduling ensures no tenant is starved:

func dispatcher(tenants []chan Job, work chan<- Job) {
    i := 0
    for {
        q := tenants[i%len(tenants)]
        i++
        select {
        case j := <-q:
            work <- j
        default:
            // skip this tenant; they have no work right now
        }
    }
}

A more sophisticated version uses weighted round-robin where each tenant has a priority or quota.

Observability Hooks¶

A backpressure-aware service exposes (at minimum) these signals:

func (p *Pool) instrumentSubmit(ctx context.Context, j Job) error {
    start := time.Now()
    err := p.SubmitCtx(ctx, j)
    submitWait.Observe(time.Since(start).Seconds())
    queueDepth.Set(float64(len(p.jobs)))
    inFlight.Set(float64(p.Stats.InFlight.Load()))
    switch {
    case err == nil:
        submitTotal.WithLabelValues("accepted").Inc()
    case errors.Is(err, context.DeadlineExceeded):
        submitTotal.WithLabelValues("rejected").Inc()
    default:
        submitTotal.WithLabelValues("errored").Inc()
    }
    return err
}

A dashboard combining these tells the whole backpressure story: queue depth climbs, wait time grows, rejections begin, in-flight stays flat (workers are saturated), shedding goes 0→1. At a glance, the operator knows the system is shedding load and can act.

Tracing¶

For distributed tracing, every SubmitCtx call should produce a span. Tag the span with accepted / rejected outcome and the wait duration. When a request is rejected, the trace shows exactly where.

ctx, span := tracer.Start(ctx, "pool.SubmitCtx")
defer span.End()
err := p.SubmitCtx(ctx, j)
if err != nil { span.SetStatus(codes.Error, err.Error()) }

Code Examples¶

Example 1 — A semaphore-based admission controller¶

package admit

import (
    "context"
    "sync/atomic"

    "golang.org/x/sync/semaphore"
)

type Controller struct {
    sem      *semaphore.Weighted
    accepted atomic.Uint64
    rejected atomic.Uint64
    inFlight atomic.Int64
}

func New(total int64) *Controller {
    return &Controller{sem: semaphore.NewWeighted(total)}
}

func (c *Controller) Acquire(ctx context.Context, n int64) (release func(), err error) {
    if err := c.sem.Acquire(ctx, n); err != nil {
        c.rejected.Add(1)
        return func() {}, err
    }
    c.accepted.Add(1)
    c.inFlight.Add(n)
    return func() {
        c.sem.Release(n)
        c.inFlight.Add(-n)
    }, nil
}

func (c *Controller) Stats() (accepted, rejected uint64, inFlight int64) {
    return c.accepted.Load(), c.rejected.Load(), c.inFlight.Load()
}

Drop this into an HTTP handler:

func handler(w http.ResponseWriter, r *http.Request) {
    ctx, cancel := context.WithTimeout(r.Context(), 100*time.Millisecond)
    defer cancel()
    release, err := admit.Acquire(ctx, 1)
    if err != nil {
        http.Error(w, "busy", 503)
        return
    }
    defer release()
    // ...
}

Example 2 — Drop-oldest channel¶

type LatestN struct {
    ch chan Event
}

func NewLatestN(n int) *LatestN { return &LatestN{ch: make(chan Event, n)} }

func (l *LatestN) Push(e Event) {
    for {
        select {
        case l.ch <- e:
            return
        default:
            select {
            case <-l.ch: // drop oldest
            default:
            }
        }
    }
}

func (l *LatestN) Drain() <-chan Event { return l.ch }

Useful for "most recent N" telemetry buffers.

Example 3 — Pipeline with end-to-end backpressure¶

func source(ctx context.Context) <-chan int {
    out := make(chan int, 4)
    go func() {
        defer close(out)
        for i := 0; ; i++ {
            select {
            case out <- i:
            case <-ctx.Done():
                return
            }
        }
    }()
    return out
}

func square(ctx context.Context, in <-chan int) <-chan int {
    out := make(chan int, 4)
    go func() {
        defer close(out)
        for x := range in {
            select {
            case out <- x * x:
            case <-ctx.Done():
                return
            }
        }
    }()
    return out
}

func sink(ctx context.Context, in <-chan int) {
    for x := range in {
        time.Sleep(10 * time.Millisecond) // slow consumer
        _ = x
    }
}

func main() {
    ctx, cancel := context.WithTimeout(context.Background(), time.Second)
    defer cancel()
    sink(ctx, square(ctx, source(ctx)))
}

The sink is slow. The square stage's output fills. Square blocks on its output channel. Source's output fills. Source blocks. Memory stays bounded; producers slow to match consumer; everything cancels cleanly when context fires.

Example 4 — Per-tenant queues with shared workers¶

type TenantPool struct {
    mu      sync.Mutex
    queues  map[string]chan Job
    workCh  chan tenantJob
    workers int
}

type tenantJob struct {
    tenant string
    j      Job
}

func NewTenantPool(workers int) *TenantPool {
    p := &TenantPool{
        queues:  make(map[string]chan Job),
        workCh:  make(chan tenantJob, workers*2),
        workers: workers,
    }
    for i := 0; i < workers; i++ {
        go func() {
            for tj := range p.workCh {
                tj.j.Run()
            }
        }()
    }
    return p
}

func (p *TenantPool) Submit(tenant string, j Job) bool {
    p.mu.Lock()
    q, ok := p.queues[tenant]
    if !ok {
        q = make(chan Job, 16)
        p.queues[tenant] = q
        go p.drain(tenant, q)
    }
    p.mu.Unlock()
    select {
    case q <- j:
        return true
    default:
        return false
    }
}

func (p *TenantPool) drain(tenant string, q chan Job) {
    for j := range q {
        p.workCh <- tenantJob{tenant, j}
    }
}

Each tenant has a private queue (sized 16); a small per-tenant goroutine drains into a shared work channel. Shedding happens at the per-tenant queue, not at the shared pool. Noisy tenants cannot starve quiet ones.

Coding Patterns¶

Pattern 1 — The dual-budget pool¶

Separate budgets for I/O wait and CPU work:

type DualPool struct {
    io  *semaphore.Weighted
    cpu *semaphore.Weighted
}

func (p *DualPool) Handle(ctx context.Context, r *http.Request) error {
    if err := p.io.Acquire(ctx, 1); err != nil { return err }
    defer p.io.Release(1)
    data, err := io.ReadAll(r.Body)
    if err != nil { return err }

    if err := p.cpu.Acquire(ctx, 1); err != nil { return err }
    defer p.cpu.Release(1)
    return process(data)
}

Pattern 2 — Concurrency limit via a "permit token" type¶

type Permit struct{ release func() }

func (p Permit) Release() { p.release() }

func (a *Admit) Acquire(ctx context.Context) (Permit, error) {
    if err := a.sem.Acquire(ctx, 1); err != nil { return Permit{}, err }
    return Permit{release: func() { a.sem.Release(1) }}, nil
}

A typed permit prevents the common bug of forgetting to release. The compiler at least reminds you that Permit is a value to be held.

Pattern 3 — Submit with retry¶

func SubmitWithRetry(ctx context.Context, p *Pool, j Job, retries int) error {
    for i := 0; i < retries; i++ {
        if err := p.SubmitCtx(ctx, j); err == nil {
            return nil
        }
        select {
        case <-time.After(backoff(i)):
        case <-ctx.Done():
            return ctx.Err()
        }
    }
    return errors.New("submit retries exhausted")
}

When the caller can wait, retrying with backoff smooths over brief overloads. When the caller cannot wait, do not retry — return immediately.

Pattern 4 — Submit with a fallback¶

func SubmitOrFallback(p *Pool, primary, fallback Job) {
    if !p.TrySubmit(primary) {
        // skip the expensive primary; run the cheap fallback inline
        fallback.Run()
    }
}

When primary work is too expensive to queue, fall back to cheap work. Common for caches: "if the heavy refresh queue is full, just serve the stale cache."

Clean Code¶

• A function that submits work should be named to reflect the policy: Submit, TrySubmit, SubmitCtx, SubmitOrDrop, SubmitOrFallback. Not just one Submit that hides the behaviour.
• Each backpressure-related constant (workers, bufferSize, submitTimeout) should be configurable. Hard-coded constants make tuning a code change.
• The pool struct should not hold a context.Context field. Contexts are passed in, not stored.
• The pool's Close method should accept a context for shutdown timeout, not block forever.
• Document the policy at the type doc, not just inline: "Pool drops jobs when the buffer is full; use SubmitCtx for caller-driven timeouts."

Product Use / Feature¶

A backpressure-aware service is a better product. Some examples:

• A search endpoint that returns "partial results in 200 ms" instead of "full results in 5 seconds" wins customer satisfaction.
• A bulk-import API that returns 202 Accepted with a job ID, then processes the work async with a bounded queue, scales gracefully.
• A real-time chat that drops the slowest 1% of recipients keeps the median experience excellent. One slow client cannot pause everyone.
• An admin batch job that pauses when the production DB is busy is a better batch job — it does not need scheduling tricks; it self-throttles.

Customers cannot articulate "your service has backpressure," but they notice when it does not: timeouts, errors, slowness, lost data. Backpressure design is therefore a UX investment.

Error Handling¶

• Distinguish error types: ErrBusy (overload), ErrTimeout (deadline), ErrClosed (shutdown). Callers may handle each differently.
• For HTTP, map errors to status codes deliberately: ErrBusy → 503, ErrTimeout → 504, ErrClosed → 503 with a friendly message.
• Include a Retry-After header on 503 when the system has a known recovery time. Otherwise, omit it and let clients use their own backoff.
• Log overload events at info level, not error. They are not bugs; they are the system doing its job. Errors should be reserved for unexpected failures.
• A drop-counter increment is not an error to be returned. It is telemetry.

const (
    StatusErrBusy = "busy"
    StatusErrShutdown = "shutdown"
)

type SubmitError struct {
    Status string
    Wait   time.Duration
}

func (e *SubmitError) Error() string { return e.Status }

Security Considerations¶

• Per-IP or per-API-key rate limits should be applied before the worker pool's admission control. Otherwise, a single attacker can occupy all slots.
• Be aware that admission control with a long submit timeout (SubmitCtx with 5-second context) is itself a DoS vector — attackers can pin slots by being slow.
• For sensitive workloads (audit logging, fraud detection), prefer "block then alert" over "drop." Dropped audit records are an incident.
• Use context.WithDeadline rather than context.WithTimeout at outer layers, so the absolute time budget is preserved across hops.

Performance Tips¶

• A chan struct{} semaphore is the fastest in pure throughput. golang.org/x/sync/semaphore is slightly slower but supports weights.
• Avoid select with more than 4–5 cases in hot paths; the runtime cost grows.
• For very high-frequency submit paths (millions per second), consider sharding: N independent pools, hashed by some key.
• Profile producer wait time with a histogram. The shape of the distribution tells you whether the system is saturated (heavy tail) or just jittery (narrow distribution).
• len(ch) and cap(ch) are O(1); read them freely in metrics paths.

Best Practices¶

• Every public submit method returns a clear signal: bool, error, or both. Never silently swallow.
• Every backpressure mechanism is observable: counters, gauges, histograms.
• Test the overload path: a load test that drives the pool past capacity and verifies the right error/drop behaviour.
• Test the shutdown path: an integration test that submits work, calls Close, and verifies all work completes.
• Tune parameters one at a time: change buffer size, run a load test, change worker count, run a load test.
• Keep backpressure-related parameters in a config struct with sensible defaults, not as scattered constants.

Edge Cases & Pitfalls¶

• A closed pool that still has TrySubmit callers. Decide explicitly: do they get false (drop) or an error?
• A worker that holds a lock across <-jobs. Deadlocks if other code wants the lock to push.
• A select with both submit and ctx.Done() where both are ready. Random choice; the submit may succeed even though the context is done. Always re-check.
• A queue whose items have side-effects on submit. If j := newJob(...) allocates resources, dropping j leaks those resources unless cleanup is explicit.

Common Mistakes¶

1. Forgetting to release the semaphore on every code path. Use defer s.Release(n) immediately after Acquire.
2. Using one semaphore for both I/O and CPU concurrency. Either limit dominates; use two.
3. Reading len(ch) and acting on it without realising the value is stale immediately.
4. Sizing the queue too large because "small queues fill up too often." That is the point — they tell you about overload.
5. Returning the same error string for drop and reject. Operators cannot distinguish.

Common Misconceptions¶

• "Backpressure is just rate limiting." False. Rate limiting caps arrivals; backpressure caps work in flight. Both can be in the same system.
• "Per-tenant queues are too expensive." For small tenant counts (< 1000), they are cheap. For larger counts, share workers across tenant queues.
• "We don't need a Close because we restart on deploy." Then your in-flight work is lost. Drain costs you nothing if there is no work, and saves you a lot if there is.

Tricky Points¶

• semaphore.Acquire(ctx, n) blocks on the full weight; you cannot partial-acquire. If you need partial, build it yourself.
• An infinite goroutine in a pool worker means wg.Wait() will never return. Always have a way out.
• Pool stats counters can be read inconsistently across multiple counters (no atomic snapshot). For dashboards, that is fine. For correctness, lock or accept the inconsistency.

Test¶

func TestPoolDrainsOnClose(t *testing.T) {
    p := NewPool(2, 4, 8)
    var done int32
    for i := 0; i < 10; i++ {
        p.SubmitCtx(context.Background(), JobFunc(func() error {
            time.Sleep(20 * time.Millisecond)
            atomic.AddInt32(&done, 1)
            return nil
        }))
    }
    ctx, cancel := context.WithTimeout(context.Background(), time.Second)
    defer cancel()
    if err := p.Close(ctx); err != nil {
        t.Fatal(err)
    }
    if atomic.LoadInt32(&done) != 10 {
        t.Fatalf("expected 10 jobs, got %d", done)
    }
}

Tricky Questions¶

1. What is the difference between a semaphore and a buffered channel? Channels carry data; semaphores carry permission.
2. When is a weighted semaphore preferable to an unweighted one? When items differ in cost.
3. How does drop-oldest avoid spinning? Use overwrite-on-full; do not loop indefinitely.
4. Why do you want hysteresis on a shedding threshold? To prevent thrashing when the queue oscillates.
5. Why should every worker check ctx.Done()? Otherwise the pool cannot be shut down promptly.
6. What is the relationship between len(ch) and cap(ch) in metrics? The ratio reveals fill; the trend reveals drift.
7. Why prefer per-tenant queues? Bulkheading — one tenant cannot starve the rest.
8. What happens if you call sem.Release(n) more times than Acquire? Panic — release of unowned weight.

Cheat Sheet¶

Self-Assessment Checklist¶

• I can choose between a queue and a semaphore for a given problem.
• I use weighted semaphores when item costs vary.
• I size buffers per channel and document why.
• I propagate context.Context through every blocking operation.
• I have written a worker pool with full submit/close/stats API.
• I expose at least three backpressure metrics in any service I write.
• I test the overload path, not just the happy path.
• I implement hysteresis when I have a shedding threshold.

Summary¶

At the middle level, backpressure is no longer "put a buffer on a channel." It is a system property: every stage bounded, every blocking operation context-aware, every drop counted, every threshold hysteretic, every queue per-class where neighbours can starve each other. The worker pool with Submit, TrySubmit, SubmitCtx, stats, and graceful close is the building block; semaphores (plain and weighted) are the generalisation; pipelines are the chain. Observability — metrics, traces, logs — is what makes the system honest under load.

What You Can Build¶

• A production-grade worker pool library with full submit/close/stats API.
• An admission controller for HTTP handlers with weighted semaphores.
• A multi-tenant job system with per-tenant queues and shared workers.
• A pipeline of decoded → transformed → encoded data with bounded buffers and context-aware sends.
• A telemetry shipper with drop-oldest semantics and a drop-rate metric.

Further Reading¶

• golang.org/x/sync/semaphore source code — short and instructive.
• golang.org/x/sync/errgroup documentation.
• Concurrency in Go (Cox) — book.
• Cloudflare engineering blog on bounded concurrency and admission control.
• Google SRE book — chapter on overload.

Related Topics¶

• Dynamic worker scaling
• Batching
• Graceful shutdown / drain
• Rate limiting (token bucket, leaky bucket)
• Circuit breakers
• Bulkhead pattern

Diagrams & Visual Aids¶

Per-tenant queues feeding a shared worker pool:

  Tenant A ──► [Q_A: ████░░] ──┐
  Tenant B ──► [Q_B: ██░░░░] ──┼──► [shared work channel] ──► workers
  Tenant C ──► [Q_C: ░░░░░░] ──┘

Noisy A fills Q_A; B and C unaffected.

Hysteresis: shedding state vs queue depth

  shedding ─────┐                  ┌─────
              ON│                  │ON
                │                  │
  shedding ─────┴──────────────────┴───── time
              OFF                  OFF

         queue:  0───low───high────cap
         depth:           ▲      ▲
                          │      │
                       stop      start
                       shed      shed

Pipeline backpressure:

   source ──► [4] ──► square ──► [4] ──► encode ──► [4] ──► sink (slow)

   slow sink → encode blocks → its buffer fills → square blocks → ...
   ... → source blocks. Memory bounded; rate limited by sink.

Deep Dive: Implementing a Production Worker Pool¶

Below is the full reference implementation a middle-level engineer should be able to produce. It supports:

• Three submit modes.
• Weighted concurrency.
• Per-job timeout.
• Graceful shutdown.
• Stats with atomic counters.
• A "drain wait" semantic when closing.
• A WithObserver hook for metrics.

package pool

import (
    "context"
    "errors"
    "sync"
    "sync/atomic"
    "time"

    "golang.org/x/sync/semaphore"
)

var (
    ErrPoolClosed = errors.New("pool: closed")
    ErrBusy       = errors.New("pool: busy")
)

type Job interface {
    Weight() int64
    Run(ctx context.Context) error
}

type Stats struct {
    Submitted atomic.Uint64
    Completed atomic.Uint64
    Failed    atomic.Uint64
    Dropped   atomic.Uint64
    Rejected  atomic.Uint64
    Latency   atomic.Uint64 // nanoseconds, last finished
    InFlight  atomic.Int64
}

type Observer interface {
    OnSubmit(j Job)
    OnDrop(j Job)
    OnReject(j Job)
    OnStart(j Job)
    OnFinish(j Job, err error, took time.Duration)
}

type noopObserver struct{}

func (noopObserver) OnSubmit(Job)                           {}
func (noopObserver) OnDrop(Job)                             {}
func (noopObserver) OnReject(Job)                           {}
func (noopObserver) OnStart(Job)                            {}
func (noopObserver) OnFinish(Job, error, time.Duration)     {}

type Pool struct {
    queue   chan Job
    sem     *semaphore.Weighted
    closed  atomic.Bool
    wg      sync.WaitGroup
    cancel  context.CancelFunc
    obs     Observer
    Stats   Stats
}

type Config struct {
    QueueSize   int
    Workers     int
    TotalWeight int64
    JobTimeout  time.Duration
    Observer    Observer
}

func New(cfg Config) *Pool {
    ctx, cancel := context.WithCancel(context.Background())
    obs := cfg.Observer
    if obs == nil {
        obs = noopObserver{}
    }
    p := &Pool{
        queue:  make(chan Job, cfg.QueueSize),
        sem:    semaphore.NewWeighted(cfg.TotalWeight),
        cancel: cancel,
        obs:    obs,
    }
    for i := 0; i < cfg.Workers; i++ {
        p.wg.Add(1)
        go p.worker(ctx, cfg.JobTimeout)
    }
    return p
}

func (p *Pool) worker(workerCtx context.Context, timeout time.Duration) {
    defer p.wg.Done()
    for {
        select {
        case j, ok := <-p.queue:
            if !ok {
                return
            }
            if err := p.sem.Acquire(workerCtx, j.Weight()); err != nil {
                continue
            }
            p.run(workerCtx, j, timeout)
            p.sem.Release(j.Weight())
        case <-workerCtx.Done():
            return
        }
    }
}

func (p *Pool) run(parent context.Context, j Job, timeout time.Duration) {
    ctx := parent
    var cancel context.CancelFunc
    if timeout > 0 {
        ctx, cancel = context.WithTimeout(parent, timeout)
        defer cancel()
    }
    p.obs.OnStart(j)
    p.Stats.InFlight.Add(j.Weight())
    start := time.Now()
    err := j.Run(ctx)
    took := time.Since(start)
    p.Stats.InFlight.Add(-j.Weight())
    p.Stats.Latency.Store(uint64(took))
    if err != nil {
        p.Stats.Failed.Add(1)
    } else {
        p.Stats.Completed.Add(1)
    }
    p.obs.OnFinish(j, err, took)
}

func (p *Pool) Submit(j Job) error {
    if p.closed.Load() {
        return ErrPoolClosed
    }
    p.queue <- j
    p.Stats.Submitted.Add(1)
    p.obs.OnSubmit(j)
    return nil
}

func (p *Pool) TrySubmit(j Job) error {
    if p.closed.Load() {
        return ErrPoolClosed
    }
    select {
    case p.queue <- j:
        p.Stats.Submitted.Add(1)
        p.obs.OnSubmit(j)
        return nil
    default:
        p.Stats.Dropped.Add(1)
        p.obs.OnDrop(j)
        return ErrBusy
    }
}

func (p *Pool) SubmitCtx(ctx context.Context, j Job) error {
    if p.closed.Load() {
        return ErrPoolClosed
    }
    select {
    case p.queue <- j:
        p.Stats.Submitted.Add(1)
        p.obs.OnSubmit(j)
        return nil
    case <-ctx.Done():
        p.Stats.Rejected.Add(1)
        p.obs.OnReject(j)
        return ctx.Err()
    }
}

func (p *Pool) Close(ctx context.Context) error {
    if !p.closed.CompareAndSwap(false, true) {
        return nil
    }
    close(p.queue)
    done := make(chan struct{})
    go func() {
        p.wg.Wait()
        close(done)
    }()
    select {
    case <-done:
        return nil
    case <-ctx.Done():
        p.cancel() // cancel workers to abort
        <-done
        return ctx.Err()
    }
}

Every detail of this implementation is deliberate:

• The queue is bounded by QueueSize, so producers feel backpressure.
• The sem is the concurrency budget — independent of queue size.
• Close(ctx) drains and then cancels if the context expires.
• The Observer interface lets callers wire in Prometheus, OpenTelemetry, or custom logging without modifying the pool.
• Stats are atomics, readable concurrently from a metrics endpoint.

A team can build many domain-specific pools on top of this single implementation.

Deep Dive: Wiring the Pool into a Real HTTP Server¶

package main

import (
    "context"
    "encoding/json"
    "errors"
    "log"
    "net/http"
    "os"
    "os/signal"
    "syscall"
    "time"

    "example.com/pool"
)

type EmailJob struct {
    To, Body string
}

func (j *EmailJob) Weight() int64 { return 1 }
func (j *EmailJob) Run(ctx context.Context) error {
    select {
    case <-time.After(50 * time.Millisecond):
        return nil
    case <-ctx.Done():
        return ctx.Err()
    }
}

func main() {
    p := pool.New(pool.Config{
        QueueSize:   100,
        Workers:     8,
        TotalWeight: 8,
        JobTimeout:  200 * time.Millisecond,
    })

    http.HandleFunc("/send", func(w http.ResponseWriter, r *http.Request) {
        var j EmailJob
        if err := json.NewDecoder(r.Body).Decode(&j); err != nil {
            http.Error(w, err.Error(), 400)
            return
        }
        ctx, cancel := context.WithTimeout(r.Context(), 100*time.Millisecond)
        defer cancel()
        switch err := p.SubmitCtx(ctx, &j); {
        case err == nil:
            w.WriteHeader(http.StatusAccepted)
        case errors.Is(err, context.DeadlineExceeded):
            http.Error(w, "busy", http.StatusServiceUnavailable)
        default:
            http.Error(w, err.Error(), 500)
        }
    })

    http.HandleFunc("/stats", func(w http.ResponseWriter, r *http.Request) {
        json.NewEncoder(w).Encode(map[string]uint64{
            "submitted": p.Stats.Submitted.Load(),
            "completed": p.Stats.Completed.Load(),
            "failed":    p.Stats.Failed.Load(),
            "dropped":   p.Stats.Dropped.Load(),
            "rejected":  p.Stats.Rejected.Load(),
        })
    })

    srv := &http.Server{Addr: ":8080"}
    go func() { _ = srv.ListenAndServe() }()

    sig := make(chan os.Signal, 1)
    signal.Notify(sig, syscall.SIGINT, syscall.SIGTERM)
    <-sig

    ctx, cancel := context.WithTimeout(context.Background(), 30*time.Second)
    defer cancel()
    _ = srv.Shutdown(ctx)
    if err := p.Close(ctx); err != nil {
        log.Println("pool close:", err)
    }
}