yiming1997/agilepool/skills

agilePool — Adaptive Goroutine Pool

Overview

Package agilepool (github.com/Yiming1997/agilePool/v2, Go 1.23.4) provides an adaptive goroutine pool that dynamically scales worker count using Little's Law. It features:

• Adaptive scaling: Median rate sampling over a sliding window → target = submitMedian × runningWorkers / consumeMedian, with backlog-weighted boosting and exit-rate compensation.
• Chunked overflow buffer: When the handoff channel is full, tasks spool into a linked-list of fixed-size chunks (4096 slots each). Soft cap at 100k total tasks; exceeding it blocks on the channel instead of growing unboundedly.
• Multiple idle containers: LinkedListType (default FIFO), MinHeapType, SliceType, RingQueueType — pluggable via config.
• Configurable locking: muIdle lock can be sync.Mutex (default) or a custom spinLock (three-phase: CAS → exponential backoff → Gosched).
• Task variants: Task interface, TaskFunc (bare func), TaskWithRetry (exponential backoff), context-aware via SubmitCtx, timeout via SubmitBefore.
• Panic-safe: Workers recover from panics and continue processing.
• Graceful shutdown: Close() + Wait() pattern.

Quick Start

import agilepool "github.com/Yiming1997/agilePool/v2"
 
pool := agilepool.NewPool(agilepool.NewConfig(
    agilepool.WithWorkerNumCapacity(1000),
    agilepool.WithTaskQueueSize(10000),
))
defer pool.Close()
 
pool.Submit(agilepool.TaskFunc(func() error {
    // your work here
    return nil
}))
pool.Wait()

API Reference

Pool Creation

pool := agilepool.NewPool(config *Config)

NewPool(nil) uses defaults (unlimited workers, LinkedListType, MutexLock).

Task Interface and Types

// Core interface
type Task interface {
    process()
}
 
// Simple function adapter
// func() error → NewTask → Task
pool.Submit(agilepool.NewTask(func() error { return nil }))
type TaskFunc func() error   // func() error → TaskFunc → Task
pool.Submit(agilepool.TaskFunc(func() error { return nil }))
 
// Retry with backoff
pool.Submit(&agilepool.TaskWithRetry{
    MinBackOff: 1 * time.Second,
    MaxBackOff: 200 * time.Second,
    RetryNum:   3,                    // retries on error, so total attempts = RetryNum + 1
    BackOffStrategy: nil,             // nil → default exponential: min * 2^retryNum, capped at max
    Task: func() error { return errors.New("fail") },
})

Submission Methods

Shutdown

pool.Close()   // Idempotent. Stops cleaner/scaler. New submits become no-ops.
pool.Wait()    // Blocks until all in-flight tasks complete.
// Always: Close() first, then Wait().

Critical: Call Close() before Wait(). Calling Wait() alone blocks forever because the background goroutines never exit.

Query Methods

pool.GetRunningWorkersNum() int64   // currently active goroutines
pool.GetWorkerCreateCount() int64   // lifetime sync.Pool allocations
pool.GetIdleWorkerCount()   int64   // parked in idle container
pool.GetTaskQueueLen()      int     // len(taskQueue) — does NOT include chunked buffer
pool.GetCapacity()          int64   // max worker limit
agilepool.GetDefaultPool() *Pool    // most recently created pool (nil after Close)

Logger

pool.SetLogger(l Logger)  // interface { Printf(string, ...interface{}); Println(...interface{}) }
// Pass *log.Logger, zap.SugaredLogger, etc.

Configuration Options

All via functional options pattern:

agilepool.NewConfig(
    agilepool.WithCleanPeriod(500 * time.Millisecond),       // expired worker cleanup interval
    agilepool.WithTaskQueueSize(10000),                      // handoff channel capacity
    agilepool.WithWorkerNumCapacity(20000),                  // max concurrent workers
    agilepool.WithBlockMode(agilepool.BLOCK),                // BLOCK (default) or NONBLOCK
    agilepool.WithIdleContainerType(agilepool.LinkedListType), // LinkedList/MinHeap/Slice/RingQueue
    agilepool.WithLockType(agilepool.SpinLock),              // MutexLock (default) or SpinLock
    agilepool.WithStatsSamplePeriod(100 * time.Millisecond), // rate sampling interval
    agilepool.WithStatsWindowSize(10),                       // sliding window count for median
    agilepool.WithScalerPeriod(10 * time.Millisecond),       // scaler tick interval
    agilepool.WithBacklogDecayFactor(0.3),                   // queue backlog weight in scaler (0-1)
)

WorkMode

• BLOCK (default): Submit blocks until the task is accepted (channel or chunked buffer).
• NONBLOCK: Submit drops the task when the handoff channel is full. Use TrySubmit to detect rejection.

IdleContainerType

Common Patterns

Graceful Shutdown

pool := agilepool.NewPool(config)
defer pool.Close()        // 1. Stop accepting new work
 
// ... submit tasks ...
 
pool.Wait()               // 2. Wait for completion

Producer-Consumer

pool := agilepool.NewPool(agilepool.NewConfig(
    agilepool.WithWorkerNumCapacity(runtime.NumCPU() * 2),
))
defer pool.Close()
 
var wg sync.WaitGroup
for _, item := range items {
    wg.Add(1)
    it := item
    pool.Submit(agilepool.TaskFunc(func() error {
        defer wg.Done()
        process(it)
        return nil
    }))
}
wg.Wait()
pool.Wait()

Context-Aware with Cancellation

ctx, cancel := context.WithTimeout(context.Background(), 5*time.Second)
defer cancel()
 
pool.SubmitCtx(ctx, agilepool.TaskFunc(func() error {
    select {
    case <-ctx.Done():
        return ctx.Err()
    default:
        return doWork()
    }
}))

Anti-Patterns & Pitfalls

1. Don't call `Wait()` before `Close()` — the pool has background goroutines (cleaner, scaler, sampler). Wait() without Close() blocks forever.

1. Don't `Close()` from within a task without defer — Close() stops the cleaner goroutine, causing idle workers to accumulate. Closing inside a task is safe but use defer pool.Close() only at the outer scope.

1. Nil `*TaskWithRetry` panics — isNilTask checks *TaskWithRetry == nil but not its Task field. Ensure Task is non-nil.

1. `SubmitCtx` with `nil` context — automatically uses context.Background(). An already-canceled context causes the task to be silently dropped.

1. `GetTaskQueueLen()` is incomplete — it only reports len(taskQueue), not the chunked overflow buffer. For true backlog, check your own metrics.

1. Task closure capture — Go loop variable capture applies. Always copy:

``go for _, v := range items { v := v pool.Submit(agilepool.TaskFunc(func() error { use(v); return nil })) } ``

Testing

Tests use package agilepool_test (external test package) — import github.com/Yiming1997/agilePool/v2 and use the public API. Use testing.Short() guards for heavy benchmarks:

if testing.Short() {
    taskCount = 20000  // reduced
}

Internal Architecture (when modifying code)

• `pool.go`: Core Pool struct, NewPool, Submit path (channel → chunked buffer → backpressure), scaler with scaleIfNeeded(), Close/Wait.
• `config.go`: Config struct and With* functional options.
• `worker.go`: worker.run() — three-phase task acquisition loop: (1) channel receive, (2) chunked buffer batch (up to 8 per lock), (3) second channel check before parking.
• `task.go`: Task interface, TaskFunc, TaskWithRetry (exponential backoff default: min × 2^retryNum, capped at max).
• `idle_container.go`: IdleWorkerContainer interface + IdleContainerType enum.
• `idle_linkedlist.go` / `idle_heap.go` / `idle_slice.go` / `idle_ring_queue.go`: Concrete implementations.
• `histogram.go`: Fixed-bucket sliding-window histogram for rate sampling. Used by scaler to compute median submit/consume/exit rates.
• `spinlock.go`: Three-phase spin lock implementing sync.Locker. Used for muIdle when WithLockType(SpinLock).
• `stack_trace.go`: Stack(skip) for panic recovery — reads source files, so not for production hot paths.

Code Modification Notes

• pendingTasks is atomic — decremented in pool.done(), incremented on submit. It covers submitters blocked on the channel, giving the scaler full visibility.
• Workers are pooled via sync.Pool. Do NOT workerPool.Put(w) when the worker is also in idleWorks — that causes dual ownership and data races.
• The scaler runs every scalerPeriod (default 10ms). It reads median rates from histograms, computes target via Little's Law with backlog-weighted boosting.
• taskMu protects the chunked buffer. Workers batch-pop up to 8 tasks per lock acquisition to amortize contention.
• The spinlock does NOT call into the kernel — optimal when hold times are <1µs (e.g., RingQueue index swaps). Use MutexLock for longer or unpredictable hold times.