workerpool

README

go-worker-pool

Bounded goroutine pool with backpressure and futures for Go

Installation

go get github.com/philiprehberger/go-worker-pool

Usage

Basic Pool

import "github.com/philiprehberger/go-worker-pool"

p := workerpool.New(4) // max 4 concurrent goroutines

for i := 0; i < 100; i++ {
    p.Submit(func() {
        // do work
    })
}

p.Wait() // block until all tasks complete

Future

f := workerpool.Go(p, func() (int, error) {
    return computeExpensiveValue(), nil
})

// do other work...

val, err := f.Get() // block until result is ready

Context-Aware Submit

ctx, cancel := context.WithTimeout(context.Background(), 5*time.Second)
defer cancel()

err := p.SubmitCtx(ctx, func() {
    // do work
})
if err != nil {
    // context was cancelled while waiting for a worker slot
}

Timeout Submit

err := p.SubmitTimeout(func() {
    // do work
}, 2*time.Second)
if errors.Is(err, workerpool.ErrSubmitTimeout) {
    // no worker slot available within 2 seconds
}

// With futures
f, err := workerpool.GoTimeout(p, func() (int, error) {
    return computeValue(), nil
}, 2*time.Second)
if err != nil {
    // timed out waiting for a worker slot
}
val, err := f.Get()

Stats

s := p.Stats()
fmt.Printf("workers=%d active=%d completed=%d\n",
    s.Workers, s.Active, s.Completed)

Resize

p := workerpool.New(4)

// Scale up under load
p.Resize(8)

// Scale down — excess workers drain naturally
p.Resize(2)

Drain

// Wait for all active tasks to finish, then resume accepting work
p.Drain()

API

Function / Type	Description
New(concurrency int) *Pool	Create a new pool with the given concurrency limit
(*Pool) Submit(fn func())	Submit work; blocks if all workers are busy
(*Pool) SubmitCtx(ctx, fn) error	Submit with context; returns ctx.Err() if cancelled while waiting
(*Pool) SubmitTimeout(fn, d) error	Submit with timeout; returns ErrSubmitTimeout if deadline expires
(*Pool) Wait()	Block until all submitted work completes
(*Pool) Stop()	Mark stopped and wait; further submits panic
(*Pool) Running() int	Approximate number of active goroutines
(*Pool) Stats() PoolStats	Snapshot of workers, active, queued, and completed counts
(*Pool) Resize(n int)	Dynamically adjust max concurrency
(*Pool) Drain()	Wait for active tasks to finish, then resume accepting work
Go[T](p, fn) *Future[T]	Submit work that returns a value; returns a Future
GoTimeout[T](p, fn, d) (*Future[T], error)	Submit for result with timeout on submission
(*Future[T]) Get() (T, error)	Block until result is ready
(*Future[T]) Done() bool	Non-blocking check if complete
ErrSubmitTimeout	Sentinel error for timed-out submissions
PoolStats	Struct: Workers, Active, Queued, Completed

Development

go test ./...
go vet ./...

License

MIT

Documentation

Overview

Package workerpool provides a bounded goroutine pool for Go.

Index

• Variables
• type Future
  • func Go[T any](p *Pool, fn func() (T, error)) *Future[T]
  • func GoTimeout[T any](p *Pool, fn func() (T, error), d time.Duration) (*Future[T], error)
  • func (f *Future[T]) Done() bool
  • func (f *Future[T]) Get() (T, error)
• type Pool
  • func New(concurrency int) *Pool
  • func (p *Pool) Drain()
  • func (p *Pool) Resize(n int)
  • func (p *Pool) Running() int
  • func (p *Pool) Stats() PoolStats
  • func (p *Pool) Stop()
  • func (p *Pool) Submit(fn func())
  • func (p *Pool) SubmitCtx(ctx context.Context, fn func()) error
  • func (p *Pool) SubmitTimeout(fn func(), d time.Duration) error
  • func (p *Pool) Wait()
• type PoolStats

Variables

var ErrSubmitTimeout = errors.New("workerpool: submit timed out")

ErrSubmitTimeout is returned when a task submission exceeds its timeout.

Types

type Future

type Future[T any] struct {
	// contains filtered or unexported fields
}

Future represents a value that will be available at some point in the future. It is safe to call Get and Done from multiple goroutines concurrently.

func Go

func Go[T any](p *Pool, fn func() (T, error)) *Future[T]

Go submits a function that returns a value and an error to the pool, and returns a Future that can be used to retrieve the result. The function runs asynchronously in the pool. Use Get to block until the result is ready.

func GoTimeout

func GoTimeout[T any](p *Pool, fn func() (T, error), d time.Duration) (*Future[T], error)

GoTimeout submits a function that returns a value and an error to the pool, waiting at most duration d for a worker slot. It returns a Future and nil error on success, or nil and ErrSubmitTimeout if no slot becomes available within the deadline.

func (*Future[T]) Done

func (f *Future[T]) Done() bool

Done reports whether the future's result is available without blocking. It returns true if the submitted function has completed.

func (*Future[T]) Get

func (f *Future[T]) Get() (T, error)

Get blocks until the future's result is available and returns the value and error produced by the submitted function.

type Pool

type Pool struct {
	// contains filtered or unexported fields
}

Pool is a bounded goroutine pool that limits the number of concurrently running goroutines. It uses a buffered channel as a semaphore to enforce backpressure — when all worker slots are occupied, Submit blocks until a slot becomes available.

func New

func New(concurrency int) *Pool

New creates a new Pool with the given concurrency limit. The concurrency parameter controls how many goroutines can run simultaneously. It panics if concurrency is less than 1.

func (*Pool) Drain

func (p *Pool) Drain()

Drain removes all pending tasks from the queue by draining and refilling the semaphore channel. Active tasks are not interrupted and will complete normally. Because Submit blocks on the semaphore channel, any goroutines blocked in Submit will eventually proceed once Drain returns.

func (*Pool) Resize

func (p *Pool) Resize(n int)

Resize adjusts the pool's maximum concurrency to n. If n is greater than the current capacity, new slots are made available immediately. If n is smaller, excess workers are allowed to drain naturally — no goroutines are interrupted. Resize panics if n is less than 1.

func (*Pool) Running

func (p *Pool) Running() int

Running returns the approximate number of goroutines currently executing work. Because goroutines start and finish concurrently, the value is an approximation.

func (*Pool) Stats

func (p *Pool) Stats() PoolStats

Stats returns a snapshot of the pool's current activity. Workers is the maximum concurrency, Active is the number of goroutines currently running tasks, Queued is the number of occupied semaphore slots (an approximation that includes active workers), and Completed is the total number of tasks that have finished since the pool was created.

func (*Pool) Stop

func (p *Pool) Stop()

Stop marks the pool as stopped and waits for all in-flight work to complete. After Stop returns, any call to Submit or SubmitCtx will panic.

func (*Pool) Submit

func (p *Pool) Submit(fn func())

Submit sends a function to the pool for execution. It blocks if all worker slots are currently occupied (backpressure). The function runs in its own goroutine once a slot is available. Submit panics if the pool has been stopped.

func (*Pool) SubmitCtx

func (p *Pool) SubmitCtx(ctx context.Context, fn func()) error

SubmitCtx sends a function to the pool for execution with context support. It returns ctx.Err() if the context is cancelled or times out while waiting for a worker slot. If a slot is acquired, the function runs in its own goroutine and nil is returned. SubmitCtx panics if the pool has been stopped.

func (*Pool) SubmitTimeout

func (p *Pool) SubmitTimeout(fn func(), d time.Duration) error

SubmitTimeout sends a function to the pool for execution, waiting at most duration d for a worker slot. It returns ErrSubmitTimeout if no slot becomes available within the deadline. SubmitTimeout panics if the pool has been stopped.

func (*Pool) Wait

func (p *Pool) Wait()

Wait blocks until all submitted work has completed.

type PoolStats

type PoolStats struct {
	Workers   int
	Active    int
	Queued    int
	Completed int64
}

PoolStats contains a snapshot of pool activity.