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().

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.

With WithContext, a cancel leaves the queued jobs in the queue. To finish every queued job on SIGTERM, do not tie the worker to the signal context. Wait for the signal, then call WaitAndStop:

ctx, stop := signal.NotifyContext(context.Background(), syscall.SIGTERM)
defer stop()

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

<-ctx.Done()              // SIGTERM
worker.WaitAndStop()      // queued jobs run unless the worker is paused, then the stop starts
worker.WaitUntilStopped() // only if you need the Stopped state

WaitAndStop returns once the stop has started, while the worker can still be Stopping. A job added after its wait ends can stay queued, so stop adding jobs before you call it. See combined calls.

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(). 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 continues until its function returns.
• Jobs that are still waiting do not start. Between the cancel and the moment the worker notices it, the worker may still hand out a job. A waiting job is not lost or closed. It stays Queued, and 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. The job then 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. While job 1 runs, it adds jobs 2 and 3 and cancels the context.

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())
}

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.

A job that a worker has already taken into a batch counts in NumPending but 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. The worker changes its status on its own goroutine a moment later. So a status read right after the cancel can still show the old status. 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.

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())
}

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 permanently. 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.