Context

A worker can follow a context.Context. When that context is done, the worker shuts down. This page covers how to give a worker one, what the shutdown does to running and waiting jobs, and the worker’s own Context().

Giving a worker a context

WithContext sets the context of one worker. DefaultCtx sets the context for every worker you create after the call.

ctx, stop := signal.NotifyContext(context.Background(), os.Interrupt)
defer stop()

worker := varmq.NewWorker(process, 4, varmq.WithContext(ctx))

Without either, the worker follows no context of yours: it behaves as if you had passed context.Background(), and only Stop and Restart cancel its own Context(). A nil context is the same as none.

DefaultCtx is shared: every worker created after it follows that one context, so cancelling it stops all of them. Configuration says when the Default* functions take effect.

The context does not reach your worker function. A Job[T] has only Data() and ID(). A function that needs the context captures it. See stopping a running job early.

What cancelling does

When the context is done, because you called its cancel function or its deadline passed, the worker shuts down the way Stop does. The lifecycle page has the states. For jobs, that means:

• A running job is not interrupted. varmq cannot stop a function. Each running job keeps going until its function returns.
• Jobs that are still waiting do not start. Once the worker has noticed the cancel, it hands out no new job, and a goroutine that holds a batch finishes the job it is on and parks the rest. Between the cancel and the moment the worker notices, it may still hand out a job. A waiting job is not lost, and it is not closed. It stays Queued, either in the queue or, if a goroutine had taken it into a batch, parked on the worker. The worker’s NumPending counts it. See what happens to jobs.
• The worker is Stopping while jobs run, then Stopped. It reaches Stopped once the running jobs have returned and its goroutines are gone.
• Nothing is reported as an error. The cancellation does not reach Errs() or the error handler.
• The queue keeps taking jobs. Add still returns true after the worker has stopped, and the job waits for a worker that will not run it. Close on the queue makes Add refuse jobs, and Purge closes the waiting ones. See queues.

The program below adds job 1 to a worker with concurrency 1, then adds jobs 2 and 3 while job 1 runs, and cancels the context.

context-cancel/main.go

package main

import (
  "context"
  "fmt"
  "time"

  "github.com/goptics/varmq"
)

func main() {
  ctx, cancel := context.WithCancel(context.Background())

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

  queue.Add(1)
  time.Sleep(10 * time.Millisecond) // job 1 is running
  queue.Add(2)
  queue.Add(3)

  cancel()
  time.Sleep(10 * time.Millisecond) // the worker notices on its own goroutine
  fmt.Println("after cancel:", worker.Status(), "- running:", worker.NumProcessing())

  worker.WaitUntilStopped()
  fmt.Println("stopped:", worker.Status(), "- pending:", worker.NumPending(), "- queue Len:", queue.Len())

  _, ok := queue.Add(4)
  fmt.Println("Add after cancel:", ok, "- queue Len:", queue.Len())
}

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

RunEditOpen in Go Playground

Right after the cancel, the worker is Stopping and job 1 is still running. It finishes, and the worker is Stopped. Jobs 2 and 3 never started. They are still in the queue, so NumPending and the queue’s Len() agree. The queue still accepts job 4, which nothing will run.

Jobs 2 and 3 were still in the queue because the program added them after job 1 had started, so the worker held no batch. Had they been added first, the worker could have taken them into a batch with job 1. Jobs in a batch show in NumPending and not in Len(). See batched dispatch.

Cancelling is not the same as calling Stop in two ways. Stop sets the status to Stopping before it returns. A cancel function returns at once, and the worker changes its status on its own goroutine a moment later, so a status read right after it can still show the old one. And Stop can be undone with Restart, while a cancelled context cannot: see starting again after a cancel.

To wait for the shutdown, use WaitUntilStopped. Wait returns as soon as no job is running, which can be while the status is still Stopping. WaitUntilIdle keeps waiting on a stopped worker. Waiting compares them.

Stopping a running job early

Because varmq does not interrupt a function, a job that should end when the worker shuts down has to watch for it. The worker’s Context() is the one to watch: it is done when your context is, and also when Stop or Restart is called. Pass it to anything that accepts a context, or select on its Done channel.

context-stop-early/main.go

package main

import (
  "context"
  "fmt"
  "time"

  "github.com/goptics/varmq"
)

func main() {
  ctx, cancel := context.WithCancel(context.Background())

  // The function reads the worker's Context(), so declare the worker first.
  var worker varmq.IWorkerBinder[int]
  worker = varmq.NewWorker(
    func(j varmq.Job[int]) {
      select {
      case <-time.After(time.Second):
        fmt.Println("job", j.Data(), "done")
      case <-worker.Context().Done():
        fmt.Println("job", j.Data(), "ended early:", worker.Context().Err())
      }
    },
    varmq.WithContext(ctx),
  )
  queue := worker.BindQueue()

  queue.Add(1)
  time.Sleep(100 * time.Millisecond)
  cancel()

  worker.WaitUntilStopped()
  fmt.Println("stopped:", worker.Status())
}

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

RunEditOpen in Go Playground

The job would sleep for a second, but it ends early, as soon as the program cancels the context. Err() reports context canceled. The worker function reads Context() from the worker it belongs to, so the worker is declared first, as an IWorkerBinder.

To varmq, a job that returns early is a finished job. For a plain worker, a normal return counts as successful. With NewErrWorker, return the context’s error if you want it counted as failed. Metrics has the counters.

The worker’s Context()

Context() returns the worker’s own context. It is a child of the one you passed with WithContext or DefaultCtx, or of context.Background() if you passed none. It is done when:

• the context you passed is done;
• Stop is called, including through StopAndWait and WaitAndStop;
• Restart stops the worker.

Its Err() says why, so a deadline that passed shows as context.DeadlineExceeded and a cancel as context.Canceled. A job that waits on Context().Done() ends the same way when Stop is called.

Restart gives the worker a fresh Context(). A Context() you read earlier stays done. Call the method again after a restart instead of keeping the old value.

Starting again after a cancel

A context that is done stays done. Start and Restart give a stopped worker a new internal context, but they derive it from the context you passed, which is already done. The new one is done from the start.

Both calls still return nil, and the worker shuts down again at once. Before it does, it may hand out some of the jobs waiting in the queue, and they run with Context() already cancelled. So a cancelled context ends the worker for good. To run again, create a new worker with a context that is still live. The jobs left in the old queue stay there until you Purge them.

A Stop has no such problem: Restart after a Stop brings the worker back, because the context you passed is still live.

Shutting down on a signal or a deadline

A context from signal.NotifyContext shuts the worker down on Ctrl+C. A context from context.WithTimeout does it when the time is up. In both cases, wait for the running jobs with WaitUntilStopped:

ctx, stop := signal.NotifyContext(context.Background(), os.Interrupt)
defer stop()

worker := varmq.NewWorker(process, 4, varmq.WithContext(ctx))
queue := worker.BindQueue()
// ... add jobs ...

<-ctx.Done()              // Ctrl+C
worker.WaitUntilStopped() // the jobs already running finish first

The context-worker example in the repository shows a worker with concurrency 10 and three-second jobs that stops on Ctrl+C. The jobs that are running finish, and the ones still waiting do not start.