Worker Pools

Worker Pools

Why Worker Pools

Worker pools prevent goroutine leaks and OOM by bounding concurrency. Without a pool, every incoming request that spawns a goroutine can create unbounded parallelism:

• 10,000 requests/second = 10,000 goroutines = ~20 MB minimum (stacks start at ~2 KB, grow as needed)
• Each goroutine may hold open database connections, file descriptors, or network sockets
• The Go scheduler slows down with millions of goroutines

A worker pool with N workers guarantees at most N concurrent background tasks, regardless of request volume.

Sizing Workers

CPU-Bound Tasks

For tasks that primarily consume CPU (compression, hashing, image processing):

• Workers = runtime.NumCPU() or slightly more
• More workers than CPUs adds context-switching overhead with no throughput gain

I/O-Bound Tasks

For tasks that wait on external services (HTTP calls, database queries, email sending):

• Workers = 10x to 100x the number of CPUs is common
• The bottleneck is the external service, not CPU
• Tune based on the external service's capacity and latency

General Guidance

// CPU-bound: match CPU count
pool := NewWorkerPool(runtime.NumCPU(), 1000, logger)

// I/O-bound: more workers, they spend most time waiting
pool := NewWorkerPool(50, 5000, logger)

Queue Sizing and Backpressure

The buffered channel acts as a queue. Its size determines backpressure behavior:

// Small queue = fast backpressure signal
jobs: make(chan Job, 10)

// Large queue = absorbs bursts but uses more memory
jobs: make(chan Job, 10000)

Handling a Full Queue

When the queue is full, Submit blocks. For HTTP handlers, blocking is usually unacceptable. Use a non-blocking submit:

func (wp *WorkerPool) TrySubmit(job Job) bool {
    select {
    case wp.jobs <- job:
        return true
    default:
        wp.logger.Warn("worker pool full, dropping job", "job_id", job.ID)
        return false
    }
}

// In handler
func (s *Server) handleWebhook(w http.ResponseWriter, r *http.Request) {
    webhook := decodeWebhook(r)
    if !s.workers.TrySubmit(Job{
        ID:      "webhook-" + webhook.ID,
        Execute: func(ctx context.Context) error {
            return s.processWebhook(ctx, webhook)
        },
    }) {
        http.Error(w, "server busy, try again later", http.StatusServiceUnavailable)
        return
    }
    w.WriteHeader(http.StatusAccepted)
}

Graceful Shutdown Integration

Worker pools must integrate with your server's shutdown sequence. Process in-flight jobs before exiting:

func main() {
    logger := slog.Default()
    pool := NewWorkerPool(10, 1000, logger)

    srv := &http.Server{
        Addr:    ":8080",
        Handler: newRouter(pool),
    }

    // Start server
    go func() {
        if err := srv.ListenAndServe(); err != http.ErrServerClosed {
            logger.Error("server error", "err", err)
        }
    }()

    // Wait for interrupt
    quit := make(chan os.Signal, 1)
    signal.Notify(quit, syscall.SIGINT, syscall.SIGTERM)
    <-quit

    logger.Info("shutting down server...")

    // 1. Stop accepting new requests
    ctx, cancel := context.WithTimeout(context.Background(), 30*time.Second)
    defer cancel()
    srv.Shutdown(ctx)

    // 2. Drain the worker pool (finish in-flight jobs)
    logger.Info("draining worker pool...")
    pool.Shutdown()

    logger.Info("shutdown complete")
}

Error Handling and Retry Patterns

Logging Errors

At minimum, log all job failures:

func (wp *WorkerPool) worker(id int) {
    defer wp.wg.Done()
    for job := range wp.jobs {
        start := time.Now()
        if err := job.Execute(context.Background()); err != nil {
            wp.logger.Error("job failed",
                "worker", id,
                "job_id", job.ID,
                "duration", time.Since(start),
                "err", err,
            )
        } else {
            wp.logger.Info("job completed",
                "worker", id,
                "job_id", job.ID,
                "duration", time.Since(start),
            )
        }
    }
}

Retry with Backoff

For transient failures, wrap jobs with retry logic:

type RetryJob struct {
    Job
    MaxRetries int
    Backoff    time.Duration
}

func (wp *WorkerPool) workerWithRetry(id int) {
    defer wp.wg.Done()
    for job := range wp.jobs {
        retryJob, hasRetry := job.(RetryJob) // type assertion if using interface
        maxRetries := 1
        backoff := time.Second

        if hasRetry {
            maxRetries = retryJob.MaxRetries
            backoff = retryJob.Backoff
        }

        var lastErr error
        for attempt := 0; attempt < maxRetries; attempt++ {
            if attempt > 0 {
                time.Sleep(backoff * time.Duration(attempt))
            }
            if err := job.Execute(context.Background()); err != nil {
                lastErr = err
                wp.logger.Warn("job attempt failed",
                    "worker", id,
                    "job_id", job.ID,
                    "attempt", attempt+1,
                    "err", err,
                )
                continue
            }
            lastErr = nil
            break
        }

        if lastErr != nil {
            wp.logger.Error("job exhausted retries",
                "worker", id,
                "job_id", job.ID,
                "err", lastErr,
            )
        }
    }
}

Context Propagation

Workers should use their own context, not the request context. The request context is cancelled when the HTTP response is sent, which happens before the background job runs:

// BAD — request context cancels when response is written
s.workers.Submit(Job{
    ID: "send-email",
    Execute: func(ctx context.Context) error {
        return s.email.Send(r.Context(), user) // r.Context() is already cancelled!
    },
})

// GOOD — use background context with timeout
s.workers.Submit(Job{
    ID: "send-email",
    Execute: func(ctx context.Context) error {
        ctx, cancel := context.WithTimeout(ctx, 30*time.Second)
        defer cancel()
        return s.email.Send(ctx, user)
    },
})

If you need to pass values from the request context (like trace IDs), extract them before submitting:

traceID := traceIDFromContext(r.Context())
userID := userIDFromContext(r.Context())

s.workers.Submit(Job{
    ID: "audit-log-" + userID,
    Execute: func(ctx context.Context) error {
        ctx = withTraceID(ctx, traceID)
        return s.audit.Log(ctx, userID, action)
    },
})

errgroup as a Simpler Alternative

For fan-out/fan-in within a single request (parallel API calls, batch processing), golang.org/x/sync/errgroup is simpler than a worker pool:

// errgroup for bounded parallel work
func (s *Server) handleBatchProcess(w http.ResponseWriter, r *http.Request) {
    items := decodeItems(r)

    g, ctx := errgroup.WithContext(r.Context())
    g.SetLimit(10) // max 10 concurrent goroutines

    results := make([]*Result, len(items))
    for i, item := range items {
        i, item := i, item
        g.Go(func() error {
            result, err := s.processItem(ctx, item)
            if err != nil {
                return fmt.Errorf("processing item %d: %w", i, err)
            }
            results[i] = result
            return nil
        })
    }

    if err := g.Wait(); err != nil {
        handleError(w, r, err)
        return
    }

    writeJSON(w, http.StatusOK, results)
}

When to Use errgroup vs Worker Pool

errgroup Key Points

• g.SetLimit(n) bounds concurrency (available since Go 1.20)
• Context cancellation propagates automatically — if one goroutine returns an error, the context is cancelled for all others
• g.Wait() blocks until all goroutines complete and returns the first error
• Safe to write to results[i] from goroutine i without a mutex because each goroutine writes to a distinct index