Queues

Workers and queues showed how a worker gets a queue. This page covers the in-memory queues: the first-in first-out queue and the priority queue, how to bound them, and the calls that inspect and empty them.

FIFO and priority queues

Method	Binds	Jobs leave in
`BindQueue`	a first-in first-out queue	the order they were added
`BindPriorityQueue`	a priority queue	priority order

Add on a FIFO queue takes the data. Add on a priority queue takes the data and then an int priority:

9 collapsed lines
package main

import (
  "fmt"

  "github.com/goptics/varmq"
)

func main() {
  // Not started, so all three jobs wait in the queue together.
  worker := varmq.NewWorker(
    func(j varmq.Job[string]) {
      fmt.Println("running", j.Data())
    },
    varmq.WithAutoRun(false),
  )

  queue := worker.BindPriorityQueue()

  queue.Add("report", 10)
  queue.Add("alert", 1)
  queue.Add("digest", 10)

  worker.Start()
  worker.WaitUntilIdle()
}

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

The worker is created with WithAutoRun(false), so all three jobs wait until Start. alert runs first although it was added second, and report runs before digest because they share a priority and report was added first.

Each queue below has its own worker. Add jobs to both queues, then start the workers. The rail shows the order the jobs leave in.

Queue order · simulation

BindQueue()

jobs leave in the order they were added

Waiting0 jobs

worker

Ran, in order

BindPriorityQueue()

lower number leaves first

Waiting0 jobs

worker

Ran, in order

queue.Add("report") pq.Add("report", 10)

priority10

A lower number runs first. Among the jobs that are waiting, priority 1 goes before priority 10, whatever order they were added in. Priorities can be negative. Jobs with the same priority run in the order they were added.

Priority orders waiting jobs only. A job that is already running is not interrupted. A worker takes one job at a time from a priority queue, so a high-priority job added later does not wait behind jobs that were already handed out in a batch. How it works explains batches.

An Item in AddAll carries its own priority. See adding many jobs.

Capacity

A queue has no limit unless you set one. WithQueueCapacity bounds how many jobs can wait in it. Pass it when you bind the queue. A value below 1 means no limit.

When the queue is full, Add does not block. It returns a nil handle and false. Treat that as backpressure: retry later, drop the job, or tell your caller to slow down. A full queue, a closed queue and a duplicate job ID all give the same false. IsFull() tells you whether the queue is at its capacity.

9 collapsed lines
package main

import (
  "fmt"

  "github.com/goptics/varmq"
)

func main() {
  // Not started, so nothing leaves the queue to make room.
  worker := varmq.NewWorker(
    func(j varmq.Job[string]) {
      fmt.Println("running", j.Data())
    },
    varmq.WithAutoRun(false),
  )

  queue := worker.BindQueue(varmq.WithQueueCapacity(2))

  _, ok := queue.Add("a")
  fmt.Println("add a:", ok)
  _, ok = queue.Add("b")
  fmt.Println("add b:", ok)

  job, ok := queue.Add("c")
  fmt.Println("add c:", ok, "- handle is nil:", job == nil)
  fmt.Println("Len:", queue.Len(), "- IsFull:", queue.IsFull())

  worker.Start()
  worker.WaitUntilIdle()
}

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

Two jobs fit, and Add refuses the third with a nil handle. The worker is not started, so nothing leaves the queue to make room. Without WithAutoRun(false) the worker would take the first jobs at once and the queue would have room again.

Set a capacity below and add jobs until the queue refuses one. Then start the worker: it takes the waiting jobs into a batch, and the queue has room again. Purge() and Close() are covered in the next section.

Queue handle · simulation

In the queueopen

worker

Taken into a batchnot in the queue

Len(): 0
IsFull(): false
Add(): not called

PurgedStatus() reads Closed

Ran

queue := worker.BindQueue(varmq.WithQueueCapacity(2))

capacity2

Capacity bounds the jobs still in the queue. On a FIFO queue a worker can take several jobs into a batch, and those are no longer in the queue, so more jobs than the capacity can be waiting overall. A batch holds at most 32 jobs. How it works has the details. A priority queue is not batched.

If you bind your own queue, the capacity reaches it only when it implements CapacitySetter, and only then does the queue refuse jobs. Without it, Add keeps accepting jobs, although IsFull() still compares Len() with the number you passed.

Len, Purge and Close

These three work on the queue handle that Bind... returns.

Len() is the number of jobs waiting in the queue. It leaves out a job a worker has taken into a batch, and a job the worker parked when you paused or stopped it. The worker’s NumPending counts those. A job you closed while it waited still counts until the worker reaches it. See closing a queued job. The queue handle also has a NumPending method. It is deprecated: use Len.

Purge() removes every job that has not started and closes each one. Its Status() reads Closed, Wait returns, Err returns nil and Result returns the zero value. Jobs a worker had already taken into a batch, or parked, are closed too: the worker closes a batched job as it reaches it. A job that is running is not affected. Purging also frees the IDs of the jobs it removes.

Close() stops the queue taking jobs: every later Add returns false. With the built-in queues, jobs already in the queue stay there and still run. The queue stays closed, and varmq has no call to reopen it. The worker is not stopped, and neither are other queues bound to it. For the worker, see the lifecycle. The built-in queues return a nil error from Close.

9 collapsed lines
package main

import (
  "fmt"

  "github.com/goptics/varmq"
)

func main() {
  worker := varmq.NewWorker(
    func(j varmq.Job[string]) {
      fmt.Println("running", j.Data())
    },
    varmq.WithAutoRun(false),
  )

  queue := worker.BindQueue()

  a, _ := queue.Add("a")
  b, _ := queue.Add("b")
  fmt.Println("Len:", queue.Len())

  queue.Purge()
  fmt.Println("after Purge, Len:", queue.Len(), "- a:", a.Status(), "- b:", b.Status())

  queue.Add("c")
  queue.Close()
  _, ok := queue.Add("d")
  fmt.Println("after Close, add d:", ok)

  // "c" was added before Close, so it still runs.
  worker.Start()
  worker.WaitUntilIdle()
}

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

Purge empties the queue and closes both jobs, so their status reads Closed. After Close, Add refuses d, but c, added before the queue closed, still runs once the worker starts.

Your own queue

WithQueue and WithPriorityQueue bind a queue you wrote, instead of an in-memory one. It has to implement IQueue or IPriorityQueue.

The worker calls Enqueue when you call Add, and Dequeue when it has a free slot. The items are the jobs themselves, so store them and return them unchanged. Purge uses Values to find the jobs to close, so Values has to return what is in the queue.

The worker batches a queue passed to WithQueue like the built-in FIFO queue, so it can take several jobs at once, unless the queue also implements IAcknowledgeable. A queue passed to WithPriorityQueue is handled like the built-in priority queue, one job at a time.

Two optional interfaces add behaviour: CapacitySetter, above, for WithQueueCapacity, and IDedupQueue with IDedupPriorityQueue for job ID checks. Adapters lists every queue interface. For queues that survive a restart or are shared between processes, see persistence.

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