How it works

Each worker has two parts: one event loop that decides what runs next, and a pool of goroutines that run your function. A job travels from Add, through the queue and the event loop, to a pool goroutine. The scenes below are models of the library. Each one steps through a single case, and you can step through it yourself.

The event loop

Every worker runs one event loop, a single goroutine. It sleeps until something signals it: Add putting a job in the queue, or a pool goroutine finishing work. On each signal it hands jobs to the pool while the queue has work and a slot is free, then goes back to sleep.

A slot is free while the number of busy pool goroutines is below the worker’s concurrency. If no slot is free, the loop waits for a goroutine to come back.

Event loop · step through

step

1 / 8

busy

0

queued

0

done

0

A step-through of one worker. Press Next to move a job from Add, into the queue, through the event loop, to a pool goroutine and back. The concurrency selector switches between a worker that runs one job at a time and one that runs two. Each step names the Go call and says what the loop checks.

queue

event loop

asleep

–signal slot, empty

pool · up to 1

done

Step 1

worker := varmq.NewWorker(process, 1) queue := worker.BindQueue()

BackNextRestart

concurrency

12

Use ← and → to step. A model of the library, not a benchmark.

The worker pool

Every Add, and every job in an AddAll, wakes the event loop. Many Add calls in a row wake it once, and it then dispatches until the queue is empty or no slot is free.

The concurrency limit is an upper bound, not a pool size. The pool starts with one idle goroutine. It starts more while jobs wait, up to the limit, and lets them exit when the queue is empty, until one idle goroutine is left. The scenes label the goroutines Worker 1, Worker 2 and so on. WithIdleWorkerExpiryDuration and WithMinIdleWorkerRatio change how many idle goroutines stay. Concurrency covers both.

Below, six jobs are added to a worker with concurrency 4. Four go out one right after the other. The other two wait in the queue. The pool shrinks as the queue empties.

Worker pool · simulation

loop

asleep

busy

0

queued

0

done

0

A worker pool drains a backlog. Press Add to put jobs in the queue. The jobs arrive as one wake-up for the event loop, which hands them to pool goroutines while a slot is free. The goroutines spawn on demand and the pool shrinks back to one idle goroutine when the queue is empty. Steppers change the concurrency limit and the number of jobs. Play, Pause and Step control the playback.

queue

event loop

asleep

–signal slot, empty

pool · up to 4

done

Press Add to put jobs in the queue.

worker := varmq.NewWorker(process, 4) queue := worker.BindQueue()

Add 6 jobsPauseStepReset

limit

−

4

+

jobs

−

6

+

Each job takes about a second, so every hand-off is one job when the limit is above 1. At limit 1 the one goroutine takes the whole backlog in one batch (batching is below). A model of the library, not a benchmark.

Topics

A Topic feeds the same event loop as a queue, but from a shared log instead of a queue of jobs. Publish appends the payload to the log once, then wakes the event loop of every subscribed worker, as Add does. Each subscription reads the log through its own cursor. So each worker’s loop makes its own job from the payload and hands it to its own pool. The workers run independently. A slow or paused worker does not delay Publish or the other workers.

A payload stays in the log until every subscription has read it.

Topics covers subscribing, capacity and priority.

Batched dispatch

A hand-off costs the event loop the same whether a job takes a nanosecond or a minute. With many short jobs, the loop, not the goroutines, becomes the bottleneck. To avoid that, a goroutine can take several jobs together, as a batch, and run them one after another.

What batching means for your code

• Jobs waiting in a batch are still Queued. The worker’s NumPending counts them and NumProcessing does not. You can still close them.
• On Stop or Pause, the rest of a batch is parked. See what happens to jobs.
• Capacity counts only jobs still in the queue. A bounded queue can have up to 32 more jobs per goroutine pending, held in batches.

Not every source batches:

How big a batch is

The batch size is the smallest of three numbers:

• an even split of the waiting jobs across the goroutines the limit allows;
• enough jobs to keep hand-offs about 32 microseconds apart, sized from the last batch that finished. Until a job has been timed, this is one job, unless concurrency is 1;

With enough jobs waiting, that works out like this:

The scene starts on a worker that has already timed its jobs, with 8 cheap jobs and concurrency 2. The first goroutine takes 4 jobs. The worker spawns a second goroutine for 2 jobs. The last 2 jobs go out one at a time as goroutines become free. A fresh worker with concurrency above 1 gives out one job at a time until its first batch finishes. Pick a row of the table, or set your own numbers, to see the batch size.

Batched dispatch · calculator

jobs per hand-off

1

A calculator for the number of jobs one hand-off carries. Pick a preset or set the concurrency limit, the time one job takes and the number of waiting jobs. The scene shows the three numbers the worker compares: an even split of the waiting jobs, the number that keeps hand-offs about 32 microseconds apart, and the cap of 32. The smallest one is the batch size. For a small limit it also lists the first hand-offs, one per goroutine.

10 goroutines · 3 s jobs24 goroutines · 1 ms jobs3,000 goroutines · 10 ms jobs10,000 goroutines · 10 ms jobsConcurrency 18 cheap jobs · concurrency 2

concurrency limit124810243,00010,000one job takesnot timed yet (first batch)1 µs100 µs1 ms10 ms3 sjobs waiting8100100,0001,000,000

even split0

32 µs apart0

cap32

k := min(even, budget, 32)

Steady state: the pool is full, so alive goroutines equal the limit. The worker takes the smallest of the three numbers. A model of the library, not a benchmark.

Next: worker types.