README
¶
adaptive
A CPU-aware worker pool whose primary goal is to NOT saturate the host CPU.
Unlike a fixed-size pool that maximizes throughput, adaptive scales the worker
count DOWN when host CPU usage exceeds a target (default 0.75) so
latency-critical paths keep their headroom. When there is spare CPU and a queued
backlog, it scales back UP, bounded by MinWorkers and MaxWorkers.
The do-no-harm contract: this pool treats CPU headroom as a shared resource and
yields it when contested. It is intentionally NOT a throughput-maximizing pool.
Use workerpool for fixed-concurrency throughput; use adaptive when the
workload is opportunistic and must back off the host under load.
Autoscale model
Every SampleInterval the autoscaler samples host CPU and adjusts the worker
count toward the target ceiling:
- CPU >
TargetCPU: shrink by one (down toMinWorkers) to free headroom. This is the do-no-harm behavior: yield CPU when contested. - CPU <
TargetCPUand queued backlog: grow by one (up toMaxWorkers) to drain it. No backlog means no grow, so workers are not added to idle on an empty queue. - Otherwise: hold steady.
Shrink is non-preemptive: a signaled worker finishes its current job, then exits. Remaining queued jobs stay for the surviving workers. No task is ever dropped by the autoscaler.
API
| Symbol | Behavior |
|---|---|
New[Job](work, opts...) (*Pool[Job], error) |
Build, start MinWorkers, launch autoscaler; errors on bad config |
WithMinWorkers(n) / WithMaxWorkers(n) |
Worker bounds (default 1 / NumCPU()*2) |
WithTargetCPU(f) |
CPU fraction ceiling (default 0.75); must be in (0,1) |
WithSampleInterval(d) |
Autoscaler tick / CPU sample interval (default 1s) |
WithQueueSize(n) |
Job queue cap (default MaxWorkers); full queue is backpressure |
WithLoadMonitor(m) |
Inject a LoadMonitor (tests pass a fake; default samples gopsutil) |
(*Pool).Submit(ctx, j) error |
Enqueue; ErrClosed after close, ErrFull on full queue / ctx done |
(*Pool).TrySubmit(j) (bool, error) |
Non-blocking enqueue |
(*Pool).Workers() int |
Current live worker count (atomic; lags the last decision by up to SampleInterval) |
(*Pool).Close() error |
Stop autoscaler, reject new submits, drain queued jobs with surviving workers, wait. Idempotent |
Close waits for in-flight jobs to finish, so a Work func that blocks
indefinitely (stuck network call, held lock, infinite loop) will hang Close.
Work MUST be non-blocking or honor a context/deadline internally: Go cannot
preempt it. Close's wait bound is (in-flight jobs) * max(Work duration).
Sentinel errors: ErrClosed (Submit after Close), ErrFull (queue full).
Both are testable with errors.Is.
Example
var processed atomic.Int64
// A fake monitor makes the autoscaler deterministic; omit WithLoadMonitor in
// production and the pool samples gopsutil directly.
pool, err := adaptive.New[int](
func(j int) { processed.Add(1) },
adaptive.WithMinWorkers[int](1),
adaptive.WithMaxWorkers[int](4),
adaptive.WithTargetCPU[int](0.75),
adaptive.WithSampleInterval[int](10*time.Millisecond),
adaptive.WithLoadMonitor[int](fakeLoadMonitor{frac: 0.4}),
)
if err != nil {
log.Fatal(err)
}
for i := 0; i < 16; i++ {
_ = pool.Submit(context.Background(), i)
}
pool.Close() // graceful: drains queued jobs and waits for workers
fmt.Println("processed:", processed.Load())
Testing
No live server is required. The autoscaler's CPU decisions are driven by an
injected fakeMonitor so they are deterministic; the package-var cpuPercent
sampler is swapped to cover the gopsutil error and empty-slice branches. A
single smoke test reads real kernel CPU counters via the default monitor (a
~5ms sample, no workload) and runs under -short.
go test -race -cover ./adaptive/...
Documentation
¶
Overview ¶
Package adaptive implements a CPU-aware worker pool whose primary goal is to NOT saturate the host CPU.
Unlike a fixed-size pool that maximizes throughput, adaptive scales the worker count DOWN when host CPU usage exceeds a target (default 0.75) so that latency-critical paths keep their headroom — e.g. an RTB bidder with a ~100ms timeout must not be starved by background work running in the same pool. When there is spare CPU and a queued backlog, it scales back UP, bounded by MinWorkers and MaxWorkers.
The do-no-harm contract: this pool treats CPU headroom as a shared resource and yields it when contested. It is intentionally NOT a throughput-maximizing pool. Use workerpool for fixed-concurrency throughput; use adaptive when the workload is opportunistic and must back off the host under load.
Worker add/remove is safe under concurrent Submit. Shrink signals a worker to exit after its current job (no hard preemption), so an in-flight job always runs to completion.
Index ¶
- Variables
- type LoadMonitor
- type Option
- func WithLoadMonitor[Job any](m LoadMonitor) Option[Job]
- func WithMaxWorkers[Job any](n int) Option[Job]
- func WithMinWorkers[Job any](n int) Option[Job]
- func WithQueueSize[Job any](n int) Option[Job]
- func WithSampleInterval[Job any](d time.Duration) Option[Job]
- func WithTargetCPU[Job any](f float64) Option[Job]
- type Options
- type Pool
Examples ¶
Constants ¶
This section is empty.
Variables ¶
var ( // ErrClosed is returned by Submit after Close. ErrClosed = errors.New("adaptive: pool closed") // ErrFull is returned by Submit when the job queue is full (backpressure). ErrFull = errors.New("adaptive: queue full") )
Sentinel errors. Use errors.Is to test.
Functions ¶
This section is empty.
Types ¶
type LoadMonitor ¶
LoadMonitor reports the host CPU load as a fraction in [0,1]. The default implementation samples gopsutil cpu.Percent over SampleInterval. Tests inject a fake to drive the autoscaler deterministically (real CPU is flaky).
type Option ¶
Option configures the Pool.
func WithLoadMonitor ¶
func WithLoadMonitor[Job any](m LoadMonitor) Option[Job]
WithLoadMonitor injects a LoadMonitor (tests pass a fake). Without it, the pool samples gopsutil cpu.Percent over SampleInterval.
func WithMaxWorkers ¶
WithMaxWorkers sets the maximum worker count (default runtime.NumCPU()*2). Must be >= MinWorkers.
func WithMinWorkers ¶
WithMinWorkers sets the minimum worker count (default 1). Must be >= 1 and <= MaxWorkers.
func WithQueueSize ¶
WithQueueSize sets the job queue capacity (default MaxWorkers). A larger queue absorbs bursts at the cost of memory and latency; a full queue returns ErrFull from Submit (backpressure).
func WithSampleInterval ¶
WithSampleInterval sets the autoscaler tick / CPU sample interval (default 1s). Must be > 0.
func WithTargetCPU ¶
WithTargetCPU sets the CPU fraction ceiling (default 0.75) above which the pool shrinks. Must be in (0,1).
type Options ¶
type Options[Job any] struct { MinWorkers int MaxWorkers int TargetCPU float64 SampleInterval time.Duration QueueSize int Work func(Job) // contains filtered or unexported fields }
Options holds the pool configuration. Zero values fall back to defaults in New: MinWorkers=1, MaxWorkers=runtime.NumCPU()*2, TargetCPU=0.75, SampleInterval=1s, QueueSize=MaxWorkers.
type Pool ¶
type Pool[Job any] struct { // contains filtered or unexported fields }
Pool is a CPU-aware autoscaling worker pool over an arbitrary job type.
Submit enqueues jobs on a bounded queue; a pool of workers drains it. An autoscaler goroutine samples host CPU every SampleInterval and adjusts the worker count toward the TargetCPU ceiling, clamped to [MinWorkers,MaxWorkers].
func New ¶
New builds an adaptive pool, starts MinWorkers, and launches the autoscaler goroutine. work must be non-nil.
Returns an error (rather than panicking) for invalid configuration: MinWorkers < 1, MaxWorkers < MinWorkers, TargetCPU not in (0,1), SampleInterval <= 0, or a nil work function.
Example ¶
ExampleNew shows an adaptive pool that processes jobs while keeping host CPU under a target. A fake monitor is injected so the example is deterministic; in production, omit WithLoadMonitor and the pool samples gopsutil directly.
This example has no // Output: comment because worker scheduling and the autoscaler tick are timing-dependent, so go test compiles but does not execute it.
package main
import (
"context"
"fmt"
"sync/atomic"
"time"
"github.com/v8fg/kit4go/adaptive"
)
// fakeLoadMonitor is a deterministic LoadMonitor for the example. Real CPU is
// intentionally NOT used here so the example is reproducible.
type fakeLoadMonitor struct {
frac float64
}
func (m fakeLoadMonitor) CPU() (float64, error) { return m.frac, nil }
// ExampleNew shows an adaptive pool that processes jobs while keeping host CPU
// under a target. A fake monitor is injected so the example is deterministic;
// in production, omit WithLoadMonitor and the pool samples gopsutil directly.
//
// This example has no // Output: comment because worker scheduling and the
// autoscaler tick are timing-dependent, so go test compiles but does not
// execute it.
func main() {
var processed atomic.Int64
// CPU reported at 0.4 (under the 0.75 target): the pool keeps headroom for
// latency-critical paths and only grows when there is queued backlog.
pool, err := adaptive.New[int](
func(j int) { processed.Add(1) },
adaptive.WithMinWorkers[int](1),
adaptive.WithMaxWorkers[int](4),
adaptive.WithTargetCPU[int](0.75),
adaptive.WithSampleInterval[int](10*time.Millisecond),
adaptive.WithLoadMonitor[int](fakeLoadMonitor{frac: 0.4}),
)
if err != nil {
fmt.Println("new pool:", err)
return
}
for i := range 16 {
_ = pool.Submit(context.Background(), i)
}
pool.Close() // graceful: drains queued jobs and waits for workers
fmt.Println("processed:", processed.Load())
}
Output:
func (*Pool[Job]) Close ¶
Close stops the autoscaler, rejects new submissions, and drains remaining queued jobs with the surviving workers, then waits for them to exit. Bounded: drains only jobs already queued at close time (Submit after Close is rejected). Idempotent and safe to call concurrently.
Close waits for in-flight jobs to finish, so a Work func that blocks indefinitely (stuck network call, held lock, infinite loop) will hang Close. Work MUST be non-blocking or honor a context/deadline internally — Go cannot preempt it. Close's wait bound is: (in-flight jobs) x max(Work duration).
Accepted-but-not-yet-consumed jobs are guaranteed to run: after the workers exit, Close does a final non-blocking drain of the queue and runs any stragglers synchronously. This closes the Submit/Close race where a job is enqueued (Submit returns nil) after the last worker's drainQueue has already finished — without this drain such a job would be silently abandoned.
func (*Pool[Job]) Recovered ¶ added in v0.7.0
Recovered returns the number of user work() panics recovered by workers since the pool was created.
func (*Pool[Job]) SetOnPanic ¶ added in v0.7.0
SetOnPanic installs a hook fired (non-blocking) whenever a worker recovers a panic from the user work function. The hook runs on a worker goroutine, so it must not block. Set to nil to disable. A recovered work panic is otherwise silent — the worker keeps draining and the only trace is Recovered() climbing — so wire an alert here for a misbehaving job type.
func (*Pool[Job]) Submit ¶
Submit enqueues a job. It returns ErrClosed if the pool is shutting down, ErrFull if the queue is full (backpressure), or blocks up to ctx deadline.
Backpressure model: Submit blocks while the queue is full (bounded by ctx). For non-blocking submit, use Pool.TrySubmit.
Source Files