The queue

func New[T any](queueLength int) *Pool[T]
func (p *Pool[T]) Delegate(ctx context.Context, job T) error

A pool has one queue. New takes its length, and the type parameter is the type of a job: any value, often a struct with what the job needs.

type Email struct {
  To, Subject, Body string
}

pool := workerpool.New[Email](100)

Delegate adds a job to the queue. The next free worker takes it, in the order jobs were queued.

When there is room

Delegate queues the job and returns nil at once. It doesn’t wait for the job to run.

A queue of length 0 holds nothing: Delegate waits until a worker takes the job from it. Use it when the caller should never get ahead of the workers.

When the queue is full

Delegate waits for a worker to take a job and make room. It gives up when its context is done, and returns the context’s error: pass a context with a deadline to bound the wait.

ctx, cancel := context.WithTimeout(ctx, 100*time.Millisecond)
defer cancel()

if err := pool.Delegate(ctx, email); err != nil {
  return fmt.Errorf("queue email to %s: %w", email.To, err)
}

With context.Background(), Delegate waits as long as it takes, or until the pool is stopped.

Delegate from a job

A job may call Delegate on its own pool. If every worker does that while the queue is full, none is left to make room, and they wait for each other. Give those calls a deadline, or queue the follow-up work elsewhere.

When the context is done first

If the context is already done, Delegate returns its error and doesn’t queue the job, even when there is room. Cancelling a producer’s context stops it queueing at once.

When the pool is stopped

Delegate returns ErrStopped. Calls that were waiting for room when Stop was called return too.

if errors.Is(err, workerpool.ErrStopped) {
  // shutting down: turn the work away
}

A job that Delegate queued, with a nil error, runs before Stop returns, as long as a worker is left to run it.

Before any worker

Jobs wait in the queue until a worker is free, so you can queue them before adding a worker. Once the queue is full, Delegate waits as it always does.

Example: answering 503 when the queue stays full

This server queues each request’s body as a job. When the queue stays full for 100 ms, it answers 503 Service Unavailable with a Retry-After header, so clients back off instead of piling up. On Ctrl-C it stops taking requests, then stops the pool, which finishes the queued jobs:

examples/server/main.go

// Command server queues the body of each POST /jobs request for four background workers.
// It answers 202 Accepted once the job is queued, and 503 Service Unavailable when the queue stays full
// for 100 ms, so that clients back off. On Ctrl-C it stops taking requests and finishes the queued jobs.
//
//  go run ./examples/server
//  curl -d 'resize photo.jpg' localhost:8080/jobs
package main

import (
  "context"
  "io"
  "log"
  "net/http"
  "os"
  "os/signal"
  "time"

  "github.com/vardius/worker-pool/v3"
)

// enqueue queues each request's body as a job, waiting up to wait for room in the queue.
func enqueue(pool *workerpool.Pool[string], wait time.Duration) http.HandlerFunc {
  return func(w http.ResponseWriter, r *http.Request) {
    body, err := io.ReadAll(http.MaxBytesReader(w, r.Body, 1<<20))
    if err != nil {
      http.Error(w, err.Error(), http.StatusBadRequest)
      return
    }

    ctx, cancel := context.WithTimeout(r.Context(), wait)
    defer cancel()
    if err := pool.Delegate(ctx, string(body)); err != nil {
      // The queue stayed full, the client went away, or the server is shutting down.
      w.Header().Set("Retry-After", "1")
      http.Error(w, "busy, try again later", http.StatusServiceUnavailable)
      return
    }
    w.WriteHeader(http.StatusAccepted)
  }
}

func process(job string) {
  time.Sleep(time.Second) // stands in for real work
  log.Printf("done: %s", job)
}

func main() {
  pool := workerpool.New[string](100)
  for range 4 {
    pool.AddWorker(process)
  }

  mux := http.NewServeMux()
  mux.HandleFunc("POST /jobs", enqueue(pool, 100*time.Millisecond))
  srv := &http.Server{Addr: "localhost:8080", Handler: mux, ReadHeaderTimeout: 5 * time.Second}

  ctx, stop := signal.NotifyContext(context.Background(), os.Interrupt)
  defer stop()
  serveErr := make(chan error, 1)
  go func() { serveErr <- srv.ListenAndServe() }()
  select {
  case err := <-serveErr:
    log.Fatal(err)
  case <-ctx.Done():
  }

  // Stop taking requests first, so that no handler is left queueing when the pool stops.
  log.Print("shutting down, finishing the queued jobs")
  shutdownCtx, cancel := context.WithTimeout(context.Background(), 5*time.Second)
  defer cancel()
  if err := srv.Shutdown(shutdownCtx); err != nil {
    log.Print(err)
  }
  pool.Stop()
}