A worker pool with typed jobs

Queue jobs for a pool of goroutines. Delegate waits while the queue is full, workers come and go at any time, and Stop finishes the queue before it returns.

Get started View on GitHub

go get github.com/vardius/worker-pool/v3

// A queue of 100 emails, and four workers
// that send them, one email each at a time.
pool := workerpool.New[Email](100)
for range 4 {
  pool.AddWorker(send)
}

for _, e := range emails {
  if err := pool.Delegate(ctx, e); err != nil {
    return err
  }
}
pool.Stop() // sends the queue, then returns

// Wait up to 100 ms for room in the queue,
// then turn the request away.
ctx, cancel := context.WithTimeout(ctx,
  100*time.Millisecond)
defer cancel()

err := pool.Delegate(ctx, email)
if err != nil {
  return errBusy
}

// A worker per connection, each sending
// through its own client.
removes := make(map[*Client]func())
for _, c := range clients {
  removes[c] = pool.AddWorker(c.Send)
}

// A client goes away: its worker finishes
// the email it is sending, then exits.
removes[gone]()

5.6×faster than v2 to run a job through a worker
11.4 nsto queue a job while the queue has room
0allocations to queue a job and run it
100%test coverage, on deterministic time with testing/synctest

Features

What you get

Typed jobs

A Pool[T] queues values of type T, and a worker is a func(T). The compiler checks them: no reflection, no interface{}.

Backpressure

When the queue is full, Delegate waits for room and gives up when its context is done, so a deadline bounds the wait.

Graceful stop

Stop runs the queued jobs on the workers left and returns once they have exited. After it, Delegate returns ErrStopped.

Workers at any time

Add a worker whenever you like, each with its own func. AddWorker returns the func that removes that worker.

Lean

Queueing a job into a queue with room and running it on a worker allocate nothing, and tests guard the allocation counts.

Deterministic tests

Tested in testing/synctest bubbles: fake time, no sleeps, no leaked goroutines, 100% coverage.

Performance

Faster than v2, with no allocations

v2 built a slice of reflect values for every job and called the worker through reflection. v3 sends a typed value on a channel and calls the worker’s func directly. A job through one worker went from 216 ns to 39 ns. With 16 callers at once, the time is wall time per call across all of them, so it measures throughput.

Jobs
Delegate

Time per job, from Delegate to the worker, lower is better

Show as a table

Time per job, from Delegate to the worker: time and allocations, v2 against v3
Scenario	v2	v3
1 worker	215.9 ns2 allocs	38.8 ns0 allocs
16 workers	412.1 ns2 allocs	110.1 ns0 allocs
1 worker, 16 callers at once	372.3 ns2 allocs	86.9 ns0 allocs
16 workers, 16 callers at once	258.3 ns2 allocs	55.4 ns0 allocs

Apple M4 Max, Go 1.27.0, median of 5 runs.

Time per call to Delegate, lower is better

Show as a table

Time per call to Delegate: time and allocations, v2 against v3
Scenario	v2	v3
queue has room	32.6 ns2 allocs	11.4 ns0 allocs
queue has room, 16 callers at once	86.0 ns2 allocs	35.4 ns0 allocs

Apple M4 Max, Go 1.27.0, median of 5 runs.

How these were measuredMethodology, the full v3 suite, and how to run it yourself.

Upgrading

Coming from v2

The idea is the same, with types: a job is a T and a worker a func(T). A worker is removed with the func AddWorker returns, Delegate takes a context, and Stop finishes the queue.

pool := workerpool.New(100)
worker := func(path string) {
  resize(path)
}
_ = pool.AddWorker(worker)

_ = pool.Delegate(path)
_ = pool.RemoveWorker(worker)

pool := workerpool.New[string](100)
remove := pool.AddWorker(resize)

_ = pool.Delegate(ctx, path)
remove()

Migration guideEvery v2 call and its v3 form, and the behaviour changes to know about.

Run your first pool in five minutes

Install the module, add a few workers, and hand them jobs. The guide walks through the queue, workers and stopping.

Get started Read the docs