README
¶
go-adaptive-pool
A production-grade adaptive worker pool for Go that handles dynamic scaling, backpressure, metrics and safe shutdown under load. It's built to keep your system stable when traffic spikes by not letting goroutines grow out of control.
Features
- Bounded Concurrency: Fixed queue size prevents unbounded memory growth
- Explicit Backpressure: Context-aware blocking when queue is full
- Adaptive Scaling: Workers scale up/down based on queue utilization
- Safe Shutdown: Graceful draining with deterministic worker cleanup
- Prometheus Metrics: Built-in observability for queue depth, throughput, and latency
- Zero Global State: Multiple pool instances with isolated metrics
- Production Ready: Comprehensive tests with race detector and goroutine leak detection
Installation
go get github.com/iyashjayesh/go-adaptive-pool
Quick Start
package main
import (
"context"
"log"
"time"
"github.com/iyashjayesh/go-adaptive-pool"
)
func main() {
// creating pool
pool, err := adaptivepool.New(
adaptivepool.WithMinWorkers(4),
adaptivepool.WithMaxWorkers(32),
adaptivepool.WithQueueSize(1000),
)
if err != nil {
log.Fatal(err)
}
defer pool.Shutdown(context.Background())
// Submit job
job := func(ctx context.Context) error {
// Your work here
time.Sleep(100 * time.Millisecond)
return nil
}
if err := pool.Submit(context.Background(), job); err != nil {
log.Printf("Failed to Submit: %v", err)
}
}
Configuration Options
pool, err := adaptivepool.New(
// Minimum workers (default: 1)
adaptivepool.WithMinWorkers(4),
// Maximum workers (default: runtime.NumCPU())
adaptivepool.WithMaxWorkers(32),
// Queue capacity (default: 1000)
adaptivepool.WithQueueSize(5000),
// Queue % to trigger scale-up (default: 0.7)
adaptivepool.WithScaleUpThreshold(0.6),
// Idle time before scale-down (default: 30s)
adaptivepool.WithScaleDownIdleDuration(20*time.Second),
// Cooldown between scaling operations (default: 5s)
adaptivepool.WithScaleCooldown(3*time.Second),
// Enable/disable metrics (default: true)
adaptivepool.WithMetricsEnabled(true),
)
Backpressure Handling
The pool enforces backpressure when the queue is full:
// Block until capacity is available
ctx := context.Background()
err := pool.Submit(ctx, job)
// Timeout after 5 seconds
ctx, cancel := context.WithTimeout(context.Background(), 5*time.Second)
defer cancel()
err := pool.Submit(ctx, job)
if err == adaptivepool.ErrTimeout {
// Handle overload
}
// Return immediately if full
ctx, cancel := context.WithTimeout(context.Background(), 0)
defer cancel()
err := pool.Submit(ctx, job)
Metrics
Access pool metrics for observability:
metrics := pool.Metrics()
fmt.Printf("Queue Depth: %d\n", metrics.QueueDepth())
fmt.Printf("Active Workers: %d\n", metrics.ActiveWorkers())
fmt.Printf("Jobs Processed: %d\n", metrics.JobsProcessed())
fmt.Printf("Jobs Rejected: %d\n", metrics.JobsRejected())
fmt.Printf("Avg Latency: %v\n", metrics.AvgJobLatency())
Graceful Shutdown
// Shutdown with timeout
ctx, cancel := context.WithTimeout(context.Background(), 30*time.Second)
defer cancel()
if err := pool.Shutdown(ctx); err != nil {
log.Printf("Shutdown error: %v", err)
}
Shutdown behavior:
- Stops accepting new jobs (returns
ErrPoolShutdown) - Drains in-flight jobs within timeout
- Terminates all workers deterministically
- Returns error if jobs were dropped
Examples
HTTP Server with Backpressure
See examples/http_server for a complete HTTP server that:
- Processes background jobs via the pool
- Returns 503 when overloaded
- Exposes metrics endpoint
- Handles graceful shutdown
cd examples/http_server
go run main.go
Batch Processing
See examples/batch_processor for processing large batches with:
- Adaptive worker scaling
- Real-time progress tracking
- Throughput metrics
cd examples/batch_processor
go run main.go
1 Million RPS Stress Test
We performed an extreme pressure test (1M RPS target for 30s with 500KB tasks) to compare the adaptive pool against naive goroutine spawning.
| Metric | Naive (No Pool) | Adaptive Pool |
|---|---|---|
| Peak RAM Usage | 50.86 GB | 1.41 GB |
| Average Latency | 2,063 ms | 161 ms |
| Peak Goroutines | 104,473 | 5,002 |
| Reliability | Fails Under Load | Rock Solid |
Why the Pool Wins:
Under extreme load, the Naive approach causes a "Goroutine Explosion" and "Memory Bomb" that forces the Go runtime into constant Garbage Collection, leading to unusable 2-second latencies. The go-adaptive-pool shields your system by enforcing backpressure and resource caps.
For a detailed deep-dive into this test and the mechanics of the pool, check out the blog post: Scaling to 1 Million RPS.
Run the comparison yourself:
make run-comparison
Benchmarks
Run standard micro-benchmarks:
go test -bench=. -benchmem -benchtime=10s
Sample micro-benchmark results:
BenchmarkPoolThroughput/workers=8-10 500000 2341 ns/op 128 B/op 3 allocs/op
BenchmarkAdaptivePool-10 300000 3892 ns/op 256 B/op 5 allocs/op
BenchmarkNaiveGoroutines-10 200000 8234 ns/op 512 B/op 12 allocs/op
Design Principles
- Bounded > Unbounded: Fixed limits prevent cascading failures
- Explicit > Implicit: Backpressure forces correct overload handling
- Simple APIs > Clever Internals: Easy to use, hard to misuse
- Correct Shutdown > Fast Shutdown: No goroutine leaks, ever
- Metrics are Mandatory: Observability is not optional
Architecture
Producer → Submit(ctx) → Bounded Queue → Dispatcher → Adaptive Workers → Job Execution
↓
Backpressure
Key components:
- Bounded Queue: Buffered channel with fixed capacity
- Dispatcher: Routes jobs to workers and monitors scaling
- Workers: Execute jobs with panic recovery and metrics tracking
- Scaling Logic: Adjusts worker count based on queue utilization
Comparison with Other Libraries
| Feature | go-adaptive-pool | ants | pond | errgroup |
|---|---|---|---|---|
| Adaptive Scaling | Yes | No | Yes | No |
| Explicit Backpressure | Yes | Partial | Partial | No |
| Prometheus Metrics | Yes | No | No | No |
| Graceful Shutdown | Yes | Yes | Yes | Yes |
| Context Support | Yes | Partial | Yes | Yes |
| Zero Global State | Yes | No | Yes | Yes |
Testing
Run tests with race detector:
go test -race -v ./...
Run with coverage:
go test -race -coverprofile=coverage.out ./...
go tool cover -html=coverage.out
Contributing
Contributions are welcome! Please see CONTRIBUTING.md for guidelines.
License
MIT License - see LICENSE for details.
Acknowledgments
Inspired by:
- ants - High-performance goroutine pool
- pond - Minimalistic worker pool
- asynq - Distributed task queue
Related Documentation
- Examples - Complete working examples
- Benchmarks - Performance comparisons
Star History
Documentation
¶
Index ¶
Constants ¶
This section is empty.
Variables ¶
var ( // ErrPoolShutdown is returned when attempting to Submit a job to a shutdown pool ErrPoolShutdown = errors.New("pool is shutdown") // ErrQueueFull is returned when the job queue is at capacity and cannot accept new jobs ErrQueueFull = errors.New("queue is full") // ErrTimeout is returned when a job submission times out ErrTimeout = errors.New("submission timeout") // ErrInvalidConfig is returned when pool configuration is invalid ErrInvalidConfig = errors.New("invalid pool configuration") )
Functions ¶
This section is empty.
Types ¶
type Config ¶
type Config struct {
// contains filtered or unexported fields
}
Config holds the configuration for the pool
type Metrics ¶
type Metrics interface {
// QueueDepth returns the current number of jobs in the queue
QueueDepth() int
// ActiveWorkers returns the current number of active workers
ActiveWorkers() int
// JobsProcessed returns the total number of jobs processed
JobsProcessed() int64
// JobsRejected returns the total number of jobs rejected
JobsRejected() int64
// AvgJobLatency returns the average job latency
AvgJobLatency() time.Duration
}
Metrics provides observability into pool behavior
type Option ¶
type Option func(*Config)
Option is a functional option for configuring the pool
func WithMaxWorkers ¶
WithMaxWorkers sets the maximum number of workers Default: runtime.NumCPU()
func WithMetricsEnabled ¶
WithMetricsEnabled enables or disables metrics collection Default: true
func WithMinWorkers ¶
WithMinWorkers sets the minimum number of workers Default: 1
func WithQueueSize ¶
WithQueueSize sets the bounded queue capacity Default: 1000
func WithScaleCooldown ¶
WithScaleCooldown sets the minimum duration between scaling operations Default: 5s
func WithScaleDownIdleDuration ¶
WithScaleDownIdleDuration sets the idle duration before scaling down workers Default: 30s
func WithScaleUpThreshold ¶
WithScaleUpThreshold sets the queue utilization percentage (0.0 to 1.0) that triggers scale-up Default: 0.7 (70%)
type Pool ¶
type Pool interface {
// Submit Submits a job to the pool for execution
// Returns ErrPoolShutdown if the pool is shutdown
// Blocks until the job is enqueued or context is cancelled
Submit(ctx context.Context, job Job) error
// Shutdown gracefully shuts down the pool
// Waits for in-flight jobs to complete within the context timeout
Shutdown(ctx context.Context) error
// Metrics returns the pool metrics
Metrics() Metrics
}
Pool represents a worker pool that can execute jobs concurrently
Source Files
¶
Directories
¶
| Path | Synopsis |
|---|---|
|
examples
|
|
|
batch_processor
command
Package main provides a batch processing example using the adaptive pool.
|
Package main provides a batch processing example using the adaptive pool. |
|
comparison
command
Package main provides a simple benchmark comparison tool for the adaptive pool.
|
Package main provides a simple benchmark comparison tool for the adaptive pool. |
|
http_server
command
Package main provides an HTTP server example using the adaptive pool.
|
Package main provides an HTTP server example using the adaptive pool. |
|
one_million_simulator/naive
command
Package main provides a naive simulator that spawns a goroutine for every task.
|
Package main provides a naive simulator that spawns a goroutine for every task. |
|
one_million_simulator/with_pool
command
Package main provides a 1 Million RPS simulator to demonstrate the adaptive pool's behavior.
|
Package main provides a 1 Million RPS simulator to demonstrate the adaptive pool's behavior. |