When to Use a Pool — Specification¶

type Group struct { ... }

// Constructor
func WithContext(ctx context.Context) (*Group, context.Context)

// Bounding
func (g *Group) SetLimit(n int)

// Submission
func (g *Group) Go(f func() error)
func (g *Group) TryGo(f func() error) bool  // returns false if at limit

// Waiting
func (g *Group) Wait() error

type Weighted struct { ... }

func NewWeighted(n int64) *Weighted

func (s *Weighted) Acquire(ctx context.Context, n int64) error
func (s *Weighted) TryAcquire(n int64) bool
func (s *Weighted) Release(n int64)

type Pool struct { ... }

func NewPool(size int, options ...Option) (*Pool, error)

func (p *Pool) Submit(task func()) error

func (p *Pool) Running() int
func (p *Pool) Free() int
func (p *Pool) Waiting() int
func (p *Pool) Cap() int
func (p *Pool) Tune(size int)
func (p *Pool) IsClosed() bool

func (p *Pool) Release()
func (p *Pool) ReleaseTimeout(timeout time.Duration) error
func (p *Pool) Reboot()

type Pool struct { ... }

func NewFunc(n int, f func(any) any) *Pool
func NewCallback(n int, ctor func() Worker) *Pool

func (p *Pool) Process(payload any) any
func (p *Pool) ProcessTimed(payload any, timeout time.Duration) (any, error)
func (p *Pool) ProcessCtx(ctx context.Context, payload any) (any, error)

func (p *Pool) QueueLength() int64
func (p *Pool) GetSize() int
func (p *Pool) SetSize(n int)

func (p *Pool) Close()

type Worker interface {
    Process(any) any
    BlockUntilReady()
    Interrupt()
    Terminate()
}

type WorkerPool struct { ... }

func New(maxWorkers int) *WorkerPool

func (p *WorkerPool) Submit(task func())
func (p *WorkerPool) SubmitWait(task func())

func (p *WorkerPool) WaitingQueueSize() int
func (p *WorkerPool) Stopped() bool

func (p *WorkerPool) Stop()
func (p *WorkerPool) StopWait()

type WorkerPool struct { ... }

func New(maxWorkers, maxCapacity int, options ...Option) *WorkerPool

func (p *WorkerPool) Submit(task func()) Task
func (p *WorkerPool) SubmitBefore(task func(), deadline time.Time) Task
func (p *WorkerPool) SubmitAndWait(task func())

func (p *WorkerPool) RunningWorkers() int
func (p *WorkerPool) IdleWorkers() int
func (p *WorkerPool) MaxWorkers() int
func (p *WorkerPool) SubmittedTasks() uint64
func (p *WorkerPool) WaitingTasks() uint64
func (p *WorkerPool) CompletedTasks() uint64

func (p *WorkerPool) Stop()
func (p *WorkerPool) StopAndWait()

func (p *WorkerPool) Group() TaskGroup

type WrappedPool struct {
    name string
    pool *ants.Pool
    metrics struct {
        submitted, completed, dropped, panicked *prometheus.CounterVec
        taskDuration, submitWait                *prometheus.HistogramVec
        running, capacity, waiting              *prometheus.GaugeVec
    }
}

func (w *WrappedPool) Submit(task func()) error {
    submitStart := time.Now()
    w.metrics.submitted.WithLabelValues(w.name).Inc()
    err := w.pool.Submit(func() {
        w.metrics.submitWait.WithLabelValues(w.name).Observe(time.Since(submitStart).Seconds())
        start := time.Now()
        defer func() {
            w.metrics.taskDuration.WithLabelValues(w.name).Observe(time.Since(start).Seconds())
            w.metrics.completed.WithLabelValues(w.name).Inc()
        }()
        task()
    })
    if err != nil {
        w.metrics.dropped.WithLabelValues(w.name).Inc()
    }
    return err
}

g, ctx := errgroup.WithContext(ctx)
g.SetLimit(K)
for _, x := range xs {
    x := x
    g.Go(func() error { return work(ctx, x) })
}
return g.Wait()

var sem = semaphore.NewWeighted(K)

func op(ctx context.Context, x X) error {
    if err := sem.Acquire(ctx, 1); err != nil { return err }
    defer sem.Release(1)
    return work(ctx, x)
}

pool, _ := ants.NewPool(K, ants.WithPanicHandler(handler))
defer pool.Release()

for msg := range source {
    msg := msg
    pool.Submit(func() { handle(msg) })
}

type myWorker struct { state *State }

func (w *myWorker) Process(p any) any { /* use w.state */ return nil }
func (w *myWorker) BlockUntilReady() {}
func (w *myWorker) Interrupt()       {}
func (w *myWorker) Terminate()       {}

pool := tunny.NewCallback(K, func() tunny.Worker { return &myWorker{state: newState()} })
defer pool.Close()

poolSize, err := strconv.Atoi(os.Getenv("POOL_SIZE"))
if err != nil || poolSize <= 0 {
    poolSize = defaultPoolSize
}
pool, err := ants.NewPool(poolSize, ...)