Worker lifecycle

A worker is always in one of seven states. Status() returns the current one as a string, and five Is* methods answer yes or no for some of them.

The diagram below is a worker you can drive. Click a state to read about it. Press a method to call it. The worker has three jobs waiting and runs one at a time, taking every waiting job into one batch. Pause and Stop never interrupt a running job, so press Finish to end it.

Worker lifecycle · state diagram

Status(): Initiated
waiting: 3
running: 0
done: 0

Selected state

Initiated

Log

call from here	result

worker := varmq.NewWorker(process, varmq.WithAutoRun(false))
queue := worker.BindQueue() // then three Add calls

The seven states

State	What it means	`Is*` check
`Initiated`	Created with `WithAutoRun(false)` and not started yet.	none
`Idle`	Started, with no job running.	`IsIdle`, `IsActive`
`Running`	At least one job is running.	`IsRunning`, `IsActive`
`Pausing`	`Pause` was called while jobs were running. They finish, and the worker takes no new ones.	none
`Paused`	No job is running and none will start until `Resume`.	`IsPaused`
`Stopping`	`Stop` was called. Running jobs finish, then the worker’s goroutines are removed.	none
`Stopped`	The event loop has ended and the goroutines are gone. Jobs still in the queue stay there.	`IsStopped`

Idle and Running swap on their own as jobs start and finish, so Running is a moment, not a mode you set. IsActive is true for either of them, which makes it the check for “started and not paused or stopped”.

The program below creates a worker with WithAutoRun(false), adds three jobs, starts it, and pauses it while job 1 runs.

10 collapsed lines
package main

import (
  "fmt"
  "time"

  "github.com/goptics/varmq"
)

func main() {
  worker := varmq.NewWorker(
    func(j varmq.Job[int]) {
      time.Sleep(100 * time.Millisecond)
      fmt.Println("job", j.Data(), "done")
    },
    varmq.WithAutoRun(false),
  )
  queue := worker.BindQueue()

  for i := 1; i <= 3; i++ {
    queue.Add(i)
  }
  fmt.Println("created:", worker.Status())

  worker.Start()
  time.Sleep(10 * time.Millisecond) // job 1 is running
  fmt.Println("started:", worker.Status())

  worker.Pause()
  fmt.Println("Pause:", worker.Status())

  worker.WaitUntilPaused()
  fmt.Println("paused:", worker.Status(), "- pending:", worker.NumPending())

  worker.Resume()
  worker.WaitUntilIdle()
  fmt.Println("drained:", worker.Status())
}

Tab inserts a tab. Esc leaves the editor. Ctrl+Enter runs.

The worker is Initiated until Start. Start makes it Idle, and it is Running a moment later, because jobs were already waiting. Pause returns while job 1 still runs, so the worker is Pausing. Once job 1 is done it is Paused, and jobs 2 and 3 wait until Resume. When they are done the worker is Idle again.

A status is a snapshot. Between reading it and acting on it the worker may have moved on. To know where a worker has landed after Pause or Stop, use the waiting helpers below.

Moving between states

Call	From	Result	Errors from other states
`Start`	`Initiated`, `Stopped`	`Idle`	`ErrRunningWorker` from `Idle` or `Running`; `ErrWorkerPausing`, `ErrWorkerPaused` or `ErrWorkerStopping`
`Pause`	`Idle` becomes `Paused`; `Running` becomes `Pausing`, then `Paused`	already `Pausing` or `Paused`: no error	`ErrNotRunningWorker` from `Initiated`; `ErrWorkerStopping`; `ErrWorkerStopped`
`Resume`	`Paused` becomes `Idle`, and the worker pulls jobs again	already `Idle` or `Running`: no error	`ErrNotRunningWorker` from `Initiated`; `ErrWorkerPausing`; `ErrWorkerStopping`; `ErrWorkerStopped`
`Stop`	`Idle`, `Running`, `Pausing` or `Paused` become `Stopping`, then `Stopped`	already `Stopping` or `Stopped`: no error	`ErrNotRunningWorker` from `Initiated`
`Restart`	any state but `Initiated`	stops, waits for `Stopped`, starts again: `Idle`	`ErrNotRunningWorker` from `Initiated`

The errors are values in the package, listed under Variables. Two details:

Pause, Resume and Stop return at once. Restart does not: it waits for running jobs to finish before it starts the worker again. If another goroutine restarts the worker first, it returns ErrRunningWorker.
A stopped worker cannot be resumed. Resume returns ErrWorkerStopped. Use Restart, or Start, which also accepts a stopped worker.

The program below stops a worker, tries to resume it, adds a job while it is stopped, then restarts it.

9 collapsed lines
package main

import (
  "fmt"

  "github.com/goptics/varmq"
)

func main() {
  worker := varmq.NewWorker(func(j varmq.Job[int]) {
    fmt.Println("job", j.Data(), "done")
  })
  queue := worker.BindQueue()

  fmt.Println("Start:", worker.Start())

  worker.StopAndWait()
  fmt.Println("stopped:", worker.Status())
  fmt.Println("Resume:", worker.Resume())

  queue.Add(1)
  fmt.Println("added while stopped - pending:", worker.NumPending())

  worker.Restart()
  worker.WaitUntilIdle()
  fmt.Println("restarted:", worker.Status())
}

Tab inserts a tab. Esc leaves the editor. Ctrl+Enter runs.

Start fails on a worker that is already running, which a worker is by default (see WithAutoRun). Resume refuses the stopped worker. The job added while it was stopped waits in the queue, and runs after Restart.

What happens to jobs

Pause and stop never interrupt a running job. Each running job finishes first.

Jobs that have not started stay queued and are not lost. A worker may have taken several jobs from the queue at once (see how it works). On pause or stop it finishes the job it is running and parks the rest of the batch. Parked jobs are still Queued, and the worker’s NumPending counts them, but the queue’s Len() does not. They run first, in their original order, after Resume or Restart. Purge closes them.

A stop also cancels the worker’s Context(). A restart gives the worker a fresh one.

Waiting for a state

Pause, Resume and Stop only start the change. These block until it is done:

PauseAndWait is Pause followed by WaitUntilPaused.
StopAndWait is Stop followed by WaitUntilStopped.
WaitUntilPaused and WaitUntilStopped wait for one state, and never return if the worker does not reach it.
WaitUntilIdle waits for Idle with no pending jobs, so on a paused or stopped worker it keeps waiting until Resume, Restart or Start brings it back to Idle and it drains.

WaitAndPause and WaitAndStop go the other way: they call Wait first, then pause or stop. Waiting compares all of them.

Settings

`WithAutoRun`

WithAutoRun decides whether the worker starts when you create it. The default is true: NewWorker returns a worker that is already Idle, and Start on it returns ErrRunningWorker.

Pass WithAutoRun(false) to get an Initiated worker instead. You can bind queues and add jobs to it right away. The jobs wait until you call Start. Use it when you want to finish setting things up before any job runs.

`WithStoppedRetention`

This setting matters only for a worker with a name (WithName or a string argument to NewWorker). Start adds a named worker to WorkerRegistry, the map that the REST Handler reads. A worker without a name is never registered.

When a registered worker reaches Stopped, it stays in the registry for a while, so it still shows up as Stopped. WithStoppedRetention sets how long. When the time is up and the worker is still stopped, it is removed from the registry.

The default is one minute. DefaultStoppedRetention changes the default for workers created after the call.
0 turns the cleanup off: a stopped worker stays in the registry for good.
Starting the worker again cancels the timer, and re-registers the worker if it had already been removed.

worker := varmq.NewWorker(process, 2,
  varmq.WithName("emails"),
  varmq.WithStoppedRetention(10*time.Minute),
)

VarMQ is MIT licensed. Benchmark figures come from github.com/goptics/benchmarks.