Thread Pool Pattern #

A web server receives thousands of requests per second. Without control, every request would spawn a new goroutine — and during high traffic, there could be hundreds of thousands of active goroutines at once. Even though goroutines are far lighter than OS threads, hundreds of thousands of goroutines still consume significant memory, cause high GC pressure, and can crash the system. The Thread Pool Pattern solves this elegantly: create a fixed number of goroutines (the pool), queue all incoming work, and let the existing goroutines pick up and process that queue — no new goroutines are created, no goroutines are left idle indefinitely.

What Is the Thread Pool Pattern? #

The Thread Pool Pattern is a concurrency pattern that provides a set of pre-spawned goroutines to execute incoming tasks, instead of creating a new goroutine for every task. The pool is bounded in size, so resource consumption stays predictable and controlled.

In Go, “Thread Pool” is more accurately called a “Goroutine Pool” because the concurrency unit is the goroutine, not an OS thread. But the concept and benefits are identical.

Three main benefits of a Thread Pool:

Controlled resources — the number of active goroutines is bounded; no goroutine explosion during high traffic
Reduced overhead — creating and destroying goroutines repeatedly is more expensive than reusing them
Natural backpressure — when the pool and queue are full, new tasks are rejected with a catchable error

flowchart LR
    subgraph "Without Pool — Unlimited"
        T1[Task 1] --> G1[new goroutine]
        T2[Task 2] --> G2[new goroutine]
        T3[Task 3] --> G3[new goroutine]
        TN[Task N] --> GN[N-th goroutine]
        note1["N tasks = N goroutines\\ncan be hundreds of thousands!"]
    end

    subgraph "With Pool — Controlled"
        Q[Task Queue\\nbuffered channel]
        T4[Task 1] --> Q
        T5[Task 2] --> Q
        T6[Task N] --> Q
        Q --> W1[Worker 1\\npersistent goroutine]
        Q --> W2[Worker 2\\npersistent goroutine]
        Q --> W3[Worker 3\\npersistent goroutine]
        note2["N tasks, only 3 goroutines\\nregardless of N"]
    end

Thread Pool Components in Go #

A Thread Pool in Go is built from three primitives working together.

flowchart TD
    subgraph "Thread Pool"
        direction TB
        TQ["Task Queue\\nbuffered channel Task"]
        W1["Worker Goroutine 1\\nfor task := range taskQueue"]
        W2["Worker Goroutine 2\\nfor task := range taskQueue"]
        W3["Worker Goroutine N\\nfor task := range taskQueue"]
        CTX["context.Context\\nfor shutdown signal"]
        WG["sync.WaitGroup\\ntracking workers done"]
    end

    TQ --> W1 & W2 & W3
    CTX -->|cancel| W1 & W2 & W3
    W1 & W2 & W3 -->|Done| WG

Component	Role	Implementation
Task Queue	Buffer of queued tasks waiting	Buffered channel `chan Task`
Worker Goroutines	Goroutines that pick up and execute tasks	`go func()` looping `for range`
Context	Signal for graceful shutdown	`context.WithCancel`
WaitGroup	Tracks that all workers finished	`sync.WaitGroup`

Full Implementation: General-Purpose Thread Pool #

Task Interface and Pool #

package pool

import (
    "context"
    "errors"
    "fmt"
    "log/slog"
    "sync"
    "sync/atomic"
    "time"
)

// Task represents work that can be executed by the pool.
type Task interface {
    Execute(ctx context.Context) error
    TaskID() string
}

// TaskResult stores the result of a task execution.
type TaskResult struct {
    TaskID    string
    Error     error
    Duration  time.Duration
    WorkerID  int
}

// PoolConfig stores the thread pool configuration.
type PoolConfig struct {
    Workers     int           // number of worker goroutines
    QueueSize   int           // task queue capacity
    TaskTimeout time.Duration // per-task timeout (0 = no timeout)
    OnError     func(result TaskResult) // callback when a task fails
    Logger      *slog.Logger
}

// DefaultPoolConfig returns a sensible default configuration.
func DefaultPoolConfig() PoolConfig {
    return PoolConfig{
        Workers:     10,
        QueueSize:   100,
        TaskTimeout: 30 * time.Second,
        Logger:      slog.Default(),
    }
}

// Metrics stores pool statistics that can be monitored.
type Metrics struct {
    TasksSubmitted atomic.Int64
    TasksCompleted atomic.Int64
    TasksFailed    atomic.Int64
    TasksRejected  atomic.Int64
    ActiveWorkers  atomic.Int32
}

// Pool is a thread-safe Thread Pool.
type Pool struct {
    config  PoolConfig
    tasks   chan Task
    results chan TaskResult
    wg      sync.WaitGroup
    ctx     context.Context
    cancel  context.CancelFunc
    metrics Metrics
    started bool
    mu      sync.Mutex
}

// NewPool creates a new pool with the given configuration.
func NewPool(config PoolConfig) *Pool {
    if config.Workers <= 0 {
        config.Workers = 1
    }
    if config.QueueSize <= 0 {
        config.QueueSize = config.Workers * 10
    }

    ctx, cancel := context.WithCancel(context.Background())
    return &Pool{
        config:  config,
        tasks:   make(chan Task, config.QueueSize),
        results: make(chan TaskResult, config.QueueSize),
        ctx:     ctx,
        cancel:  cancel,
    }
}

// Start launches all worker goroutines.
func (p *Pool) Start() {
    p.mu.Lock()
    defer p.mu.Unlock()

    if p.started {
        return
    }
    p.started = true

    for i := 0; i < p.config.Workers; i++ {
        workerID := i + 1
        p.wg.Add(1)
        go p.runWorker(workerID)
    }

    p.config.Logger.Info("thread pool started",
        "workers", p.config.Workers,
        "queue_size", p.config.QueueSize,
    )
}

// Submit adds a task to the pool queue.
// Returns an error if the pool is shutting down or the queue is full.
func (p *Pool) Submit(task Task) error {
    select {
    case <-p.ctx.Done():
        return errors.New("pool is shutting down")
    default:
    }

    select {
    case p.tasks <- task:
        p.metrics.TasksSubmitted.Add(1)
        return nil
    default:
        p.metrics.TasksRejected.Add(1)
        return fmt.Errorf("task queue is full (capacity: %d)", p.config.QueueSize)
    }
}

// SubmitWait adds a task and waits until a slot is available.
// Returns an error if the pool shuts down before the task can enter.
func (p *Pool) SubmitWait(task Task) error {
    select {
    case <-p.ctx.Done():
        return errors.New("pool is shutting down")
    case p.tasks <- task:
        p.metrics.TasksSubmitted.Add(1)
        return nil
    }
}

// Results returns the channel for reading task results.
func (p *Pool) Results() <-chan TaskResult {
    return p.results
}

// Stop shuts the pool down gracefully.
// Waits for all workers to finish processing in-flight tasks.
func (p *Pool) Stop() {
    p.cancel()          // signal all workers to stop
    close(p.tasks)      // close the channel — workers drain remaining tasks then exit
    p.wg.Wait()         // wait for all workers to finish
    close(p.results)    // close the results channel after all workers stop
    p.config.Logger.Info("thread pool stopped",
        "submitted", p.metrics.TasksSubmitted.Load(),
        "completed", p.metrics.TasksCompleted.Load(),
        "failed", p.metrics.TasksFailed.Load(),
        "rejected", p.metrics.TasksRejected.Load(),
    )
}

// GetMetrics returns a snapshot of the pool's current statistics.
func (p *Pool) GetMetrics() map[string]int64 {
    return map[string]int64{
        "submitted":      p.metrics.TasksSubmitted.Load(),
        "completed":      p.metrics.TasksCompleted.Load(),
        "failed":         p.metrics.TasksFailed.Load(),
        "rejected":       p.metrics.TasksRejected.Load(),
        "active_workers": int64(p.metrics.ActiveWorkers.Load()),
        "queue_length":   int64(len(p.tasks)),
    }
}

// runWorker is the goroutine that keeps picking up and executing tasks.
func (p *Pool) runWorker(workerID int) {
    defer p.wg.Done()
    defer p.metrics.ActiveWorkers.Add(-1)
    p.metrics.ActiveWorkers.Add(1)

    p.config.Logger.Debug("worker started", "worker_id", workerID)

    for task := range p.tasks {
        p.executeTask(workerID, task)
    }

    p.config.Logger.Debug("worker stopped", "worker_id", workerID)
}

// executeTask runs a single task with a timeout and panic recovery.
func (p *Pool) executeTask(workerID int, task Task) {
    start := time.Now()

    // Create a context with a timeout if configured
    ctx := p.ctx
    var cancel context.CancelFunc
    if p.config.TaskTimeout > 0 {
        ctx, cancel = context.WithTimeout(p.ctx, p.config.TaskTimeout)
        defer cancel()
    }

    // Execute the task with panic recovery
    var execErr error
    func() {
        defer func() {
            if r := recover(); r != nil {
                execErr = fmt.Errorf("task panicked: %v", r)
            }
        }()
        execErr = task.Execute(ctx)
    }()

    result := TaskResult{
        TaskID:   task.TaskID(),
        Error:    execErr,
        Duration: time.Since(start),
        WorkerID: workerID,
    }

    if execErr != nil {
        p.metrics.TasksFailed.Add(1)
        if p.config.OnError != nil {
            p.config.OnError(result)
        }
    } else {
        p.metrics.TasksCompleted.Add(1)
    }

    // Send the result to the channel (non-blocking — do not block the worker)
    select {
    case p.results <- result:
    default:
        // Results buffer full — log but do not block the worker
        p.config.Logger.Warn("results buffer full, dropping result",
            "task_id", task.TaskID())
    }
}

Concrete Tasks #

package pool

import (
    "context"
    "fmt"
    "time"
)

// HTTPFetchTask fetches a URL and stores the result.
type HTTPFetchTask struct {
    id      string
    url     string
    client  *http.Client
    result  *[]byte
}

func NewHTTPFetchTask(id, url string, client *http.Client, result *[]byte) Task {
    return &HTTPFetchTask{id: id, url: url, client: client, result: result}
}

func (t *HTTPFetchTask) TaskID() string { return t.id }

func (t *HTTPFetchTask) Execute(ctx context.Context) error {
    req, err := http.NewRequestWithContext(ctx, http.MethodGet, t.url, nil)
    if err != nil {
        return fmt.Errorf("failed to build request: %w", err)
    }

    resp, err := t.client.Do(req)
    if err != nil {
        return fmt.Errorf("request failed: %w", err)
    }
    defer resp.Body.Close()

    if resp.StatusCode != http.StatusOK {
        return fmt.Errorf("unexpected status: %d", resp.StatusCode)
    }

    body, err := io.ReadAll(resp.Body)
    if err != nil {
        return fmt.Errorf("read body failed: %w", err)
    }

    *t.result = body
    return nil
}


// EmailTask sends one email.
type EmailTask struct {
    id        string
    recipient string
    subject   string
    body      string
    mailer    EmailSender
}

func NewEmailTask(id, recipient, subject, body string, mailer EmailSender) Task {
    return &EmailTask{
        id: id, recipient: recipient,
        subject: subject, body: body, mailer: mailer,
    }
}

func (t *EmailTask) TaskID() string { return t.id }

func (t *EmailTask) Execute(ctx context.Context) error {
    return t.mailer.Send(ctx, t.recipient, t.subject, t.body)
}


// FuncTask allows using a closure as a task — very flexible.
type FuncTask struct {
    id string
    fn func(ctx context.Context) error
}

func NewFuncTask(id string, fn func(ctx context.Context) error) Task {
    return &FuncTask{id: id, fn: fn}
}

func (t *FuncTask) TaskID() string { return t.id }

func (t *FuncTask) Execute(ctx context.Context) error {
    return t.fn(ctx)
}

Demonstration: Batch URL Fetching #

func fetchURLsConcurrently(urls []string) map[string][]byte {
    config := pool.PoolConfig{
        Workers:     5,
        QueueSize:   len(urls),
        TaskTimeout: 10 * time.Second,
        OnError: func(result pool.TaskResult) {
            log.Printf("Task %s failed after %v: %v",
                result.TaskID, result.Duration, result.Error)
        },
    }

    p := pool.NewPool(config)
    p.Start()
    defer p.Stop()

    results := make(map[string][]byte, len(urls))
    resultBytes := make([][]byte, len(urls))
    var mu sync.Mutex

    // Submit all tasks
    for i, url := range urls {
        idx := i
        urlCopy := url
        task := pool.NewFuncTask(fmt.Sprintf("fetch-%d", idx), func(ctx context.Context) error {
            // ... fetch logic
            mu.Lock()
            results[urlCopy] = resultBytes[idx]
            mu.Unlock()
            return nil
        })
        if err := p.Submit(task); err != nil {
            log.Printf("Cannot submit task for %s: %v", url, err)
        }
    }

    // Read the results
    completed := 0
    for result := range p.Results() {
        completed++
        if completed >= len(urls) {
            break
        }
    }

    return results
}

Graceful Shutdown: Two Strategies #

Graceful shutdown is the most critical aspect of a Thread Pool. There are two different approaches depending on the need.

sequenceDiagram
    participant C as Client
    participant P as Pool
    participant W as Workers

    Note over C,W: Strategy 1: Drain then Stop

    C->>P: Stop()
    P->>P: cancel() — signal to stop accepting new tasks
    P->>P: close(tasks) — close the task channel
    W->>W: drain remaining tasks in the channel
    W->>W: finish → exit
    P->>C: wg.Wait() done

    Note over C,W: Strategy 2: Hard Stop (abandon remaining tasks)
    C->>P: ForceStop()
    P->>P: cancel() — context cancelled
    W->>W: ctx.Done() → exit immediately
    Note over W: Remaining tasks in the queue are ignored

// GracefulPool supports two shutdown modes.
type GracefulPool struct {
    Pool
}

// Stop completes all tasks already in the queue before stopping.
// Does not accept new tasks after Stop is called.
func (p *GracefulPool) Stop() {
    p.cancel()     // stop accepting new tasks
    close(p.tasks) // signal workers to drain and exit
    p.wg.Wait()    // wait for everyone to finish
}

// ForceStop stops all workers immediately, ignoring remaining queued tasks.
func (p *GracefulPool) ForceStop() {
    p.cancel() // cancel the context — workers exit when they check ctx.Done()
    // do not close the tasks channel — workers exit via ctx.Done(), not range
    p.wg.Wait()
}

// Example worker that supports both modes:
func (p *Pool) runWorkerWithCancel(workerID int) {
    defer p.wg.Done()

    for {
        select {
        case task, ok := <-p.tasks:
            if !ok {
                // Channel closed — graceful shutdown: all tasks have been drained
                return
            }
            p.executeTask(workerID, task)

        case <-p.ctx.Done():
            // Context cancelled — force shutdown: drain without executing
            return
        }
    }
}

Dynamic Pool Sizing #

For uneven workloads, the pool size can be adjusted dynamically.

// DynamicPool adjusts the number of workers based on queue length.
type DynamicPool struct {
    Pool
    minWorkers int
    maxWorkers int
    scaleUpAt  int // scale up if the queue exceeds N%
    mu         sync.Mutex
    workerCount int
}

func NewDynamicPool(min, max int, queueSize int) *DynamicPool {
    p := &DynamicPool{
        minWorkers: min,
        maxWorkers: max,
        scaleUpAt:  70, // scale up if the queue > 70% capacity
    }
    p.config.Workers = min
    p.config.QueueSize = queueSize
    return p
}

// AutoScale checks the pool conditions and adds/removes workers.
// Called periodically by a background goroutine.
func (p *DynamicPool) AutoScale() {
    queueUsage := float64(len(p.tasks)) / float64(cap(p.tasks)) * 100

    p.mu.Lock()
    defer p.mu.Unlock()

    if queueUsage > float64(p.scaleUpAt) && p.workerCount < p.maxWorkers {
        // Scale up: add one worker
        p.workerCount++
        p.wg.Add(1)
        go p.runWorker(p.workerCount)
        p.config.Logger.Info("scaled up", "workers", p.workerCount, "queue_usage", queueUsage)
    }
    // Scale down: more complex, needs a signal to a specific worker to exit
}

// StartAutoScaler runs a background goroutine that checks the load periodically.
func (p *DynamicPool) StartAutoScaler(interval time.Duration) {
    go func() {
        ticker := time.NewTicker(interval)
        defer ticker.Stop()
        for {
            select {
            case <-ticker.C:
                p.AutoScale()
            case <-p.ctx.Done():
                return
            }
        }
    }()
}

Monitoring the Pool with Prometheus #

For production, pool metrics need to be exported to a monitoring system.

// PoolMonitor exposes pool metrics as Prometheus gauges.
type PoolMonitor struct {
    pool           *Pool
    queueGauge     prometheus.Gauge
    workersGauge   prometheus.Gauge
    completedCount prometheus.Counter
    failedCount    prometheus.Counter
}

func NewPoolMonitor(pool *Pool, namespace string) *PoolMonitor {
    return &PoolMonitor{
        pool: pool,
        queueGauge: prometheus.NewGauge(prometheus.GaugeOpts{
            Namespace: namespace,
            Name:      "thread_pool_queue_length",
            Help:      "Current number of tasks waiting in queue",
        }),
        workersGauge: prometheus.NewGauge(prometheus.GaugeOpts{
            Namespace: namespace,
            Name:      "thread_pool_active_workers",
            Help:      "Current number of active workers",
        }),
    }
}

// Collect is called by the Prometheus scraper to fetch current metrics.
func (m *PoolMonitor) Collect() {
    metrics := m.pool.GetMetrics()
    m.queueGauge.Set(float64(metrics["queue_length"]))
    m.workersGauge.Set(float64(metrics["active_workers"]))
}

Testing the Thread Pool #

func TestPool_ExecutesAllTasks(t *testing.T) {
    config := pool.DefaultPoolConfig()
    config.Workers = 3
    config.QueueSize = 20

    p := pool.NewPool(config)
    p.Start()
    defer p.Stop()

    const taskCount = 10
    var completed atomic.Int32

    for i := 0; i < taskCount; i++ {
        taskID := fmt.Sprintf("task-%d", i)
        task := pool.NewFuncTask(taskID, func(ctx context.Context) error {
            completed.Add(1)
            return nil
        })
        if err := p.Submit(task); err != nil {
            t.Fatalf("submit failed: %v", err)
        }
    }

    // Collect all the results
    received := 0
    for result := range p.Results() {
        if result.Error != nil {
            t.Errorf("task %s failed: %v", result.TaskID, result.Error)
        }
        received++
        if received >= taskCount {
            break
        }
    }

    if int(completed.Load()) != taskCount {
        t.Errorf("expected %d tasks completed, got %d", taskCount, completed.Load())
    }
}

func TestPool_RejectsWhenFull(t *testing.T) {
    config := pool.PoolConfig{
        Workers:   1,
        QueueSize: 2, // very small to make testing easy
    }
    p := pool.NewPool(config)
    p.Start()
    defer p.Stop()

    // Fill the queue with slow tasks
    slowTask := pool.NewFuncTask("slow", func(ctx context.Context) error {
        time.Sleep(1 * time.Second)
        return nil
    })

    for i := 0; i < 5; i++ {
        _ = p.Submit(slowTask)
    }

    // The next task must be rejected
    err := p.Submit(slowTask)
    if err == nil {
        t.Error("expected error when queue is full")
    }
}

func TestPool_PanicRecovery(t *testing.T) {
    config := pool.DefaultPoolConfig()
    config.Workers = 2

    p := pool.NewPool(config)
    p.Start()
    defer p.Stop()

    // A task that panics
    panicTask := pool.NewFuncTask("panic-task", func(ctx context.Context) error {
        panic("intentional panic for testing")
    })
    normalTask := pool.NewFuncTask("normal-task", func(ctx context.Context) error {
        return nil
    })

    _ = p.Submit(panicTask)
    _ = p.Submit(normalTask)

    var panicResult, normalResult pool.TaskResult
    received := 0
    for result := range p.Results() {
        if result.TaskID == "panic-task" {
            panicResult = result
        } else {
            normalResult = result
        }
        received++
        if received >= 2 {
            break
        }
    }

    // The panic task must produce an error, not crash the pool
    if panicResult.Error == nil {
        t.Error("expected error from panicking task")
    }

    // The normal task must still run after a panic
    if normalResult.Error != nil {
        t.Errorf("normal task should succeed: %v", normalResult.Error)
    }
}

func TestPool_GracefulShutdown(t *testing.T) {
    config := pool.PoolConfig{Workers: 2, QueueSize: 10}
    p := pool.NewPool(config)
    p.Start()

    var completed atomic.Int32
    for i := 0; i < 5; i++ {
        _ = p.Submit(pool.NewFuncTask(fmt.Sprintf("task-%d", i),
            func(ctx context.Context) error {
                time.Sleep(10 * time.Millisecond)
                completed.Add(1)
                return nil
            }))
    }

    // Stop must wait for all tasks to finish
    p.Stop()

    if completed.Load() != 5 {
        t.Errorf("graceful shutdown should complete all queued tasks, got %d/5",
            completed.Load())
    }
}

Thread Pool vs Worker Pool #

Two patterns often considered the same when there is actually a conceptual difference.

Aspect	Thread Pool	Worker Pool
Focus	Manages goroutine lifecycle (create, reuse, destroy)	Distributes tasks to specific workers
Task routing	Any available worker picks up the task	Tasks can be routed to a specific worker
Worker identity	Workers are anonymous and interchangeable	Workers can have their own state
Use case	HTTP request handling, batch processing	Workers with their own DB connections, shard-based processing
Complexity	Simpler	More flexible but more complex

When to Use and When Not to #

USE Thread Pool if:
  ✓ There are many independent tasks that need concurrent execution
  ✓ You need to limit the number of active goroutines to control resources
  ✓ Tasks arrive faster than a single goroutine can process them
  ✓ You need backpressure — reject new tasks when the system is overloaded
  ✓ Tasks are short-lived and homogeneous

AVOID Thread Pool if:
  ✗ There are only a few, infrequent tasks — a plain goroutine is enough
  ✗ Tasks need access to specific state held by only one worker
  ✗ Tasks require guaranteed ordering — use a single goroutine + channel
  ✗ All tasks run for a very long time — the pool will drain quickly; use a Worker Pool instead

Thread Pool Review Checklist #

CONFIGURATION:
  □ The worker count is determined by profiling, not guessing
  □ The queue size is large enough to absorb bursts but not unbounded
  □ A task timeout is configured to prevent workers getting stuck forever
  □ An OnError callback is configured for logging or alerting

SHUTDOWN:
  □ Stop() is called — no goroutines leak
  □ Defer p.Stop() is used in main or tests to ensure cleanup
  □ Graceful vs force shutdown is chosen according to the need
  □ The results channel is drained or closed properly

PANIC SAFETY:
  □ Every task execution is wrapped with recover()
  □ A panic in one task does not crash the whole pool
  □ Panic errors are wrapped as errors and returned as a TaskResult

MONITORING:
  □ Queue length is monitored — alert when approaching capacity
  □ Active worker count is monitored
  □ Task failure rate is monitored

Summary #

Thread Pool bounds the number of active goroutines — resources stay managed, no goroutine explosion during high traffic, and GC pressure drops.
Three key components: a task queue (buffered channel), worker goroutines (looping for range), and a context for shutdown signals.
A buffered channel is the task queue — its size determines how many tasks can queue before Submit starts rejecting; choose it wisely.
Panic recovery is a must — one panicking task must not crash a worker; wrap every Execute with defer recover().
Graceful shutdown: cancel() + close(tasks) + wg.Wait() — this sequence ensures every queued task is processed before workers stop.
The results channel enables non-blocking result collection — workers do not wait for consumers; use an adequate buffer.
Dynamic sizing for uneven loads — scale up when the queue exceeds 70% capacity, scale down when idle.
Monitor queue length — if the queue is often full, the pool needs more workers or faster tasks; if it is always empty, the pool can be shrunk.

← Previous: Interpreter Next: Worker Pool →