Strategies

A worker can have several sources bound to it. A source is a queue, a priority queue or a topic subscription. When a slot becomes free, the worker needs a rule for which source gives the next job. That rule is the strategy. Workers and queues introduced it; this page covers all four.

Choosing a strategy

Each source has a rank: the number you set with WithQueuePriority or WithTopicPriority. A lower rank goes first. The list is every source ordered by rank, lowest first, and by bind order among equal ranks. Every strategy walks the list in that order.

Strategy has four values. WithStrategy sets one when you create the worker, and DefaultStrategy changes the default for workers created after the call.

worker := varmq.NewWorker(process, 2, varmq.WithStrategy(varmq.RoundRobin))

Pick a strategy below and take jobs one at a time. The queues start with 2 alerts, 1 report and 3 emails, the same as the RoundRobin and length programs below. Before each job, the scene says which queue the strategy chooses and why. Picking another strategy starts a new worker, because a worker keeps its strategy.

A few rules hold for all four:

• Empty sources are skipped. A strategy only chooses among sources that have a job waiting.
• The strategy is fixed when the worker is created. There is no call to change it later.
• It decides every job. With two or more sources bound, the worker gives out one job at a time. It picks the source again each time, from the source lengths at that moment. How it works explains why it does not batch in this case. With a single source there is nothing to choose.
• The zero value of Strategy is RoundRobin, not the default. A worker created without WithStrategy uses Priority.

Priority: the default

WithQueuePriority sets the rank of a queue when you bind it. It works on BindQueue and BindPriorityQueue. The programs on this page use a worker with concurrency 1 and WithAutoRun(false). No job starts until every queue is filled, and the worker takes one job at a time. So the jobs print in the order the worker took them.

package main

import (
	"fmt"

	"github.com/goptics/varmq"
)

func main() {
	// Concurrency 1 takes one job at a time, so jobs print in the order they are taken.
	worker := varmq.NewWorker(
		func(j varmq.Job[string]) {
			fmt.Println(j.Data())
		},
		varmq.WithAutoRun(false),
	)

	alerts := worker.BindQueue(varmq.WithQueuePriority(1))
	reports := worker.BindQueue(varmq.WithQueuePriority(2))
	emails := worker.BindQueue() // no priority: it ranks last

	// Fill the lowest-ranked queue first.
	emails.Add("email 1")
	emails.Add("email 2")
	reports.Add("report 1")
	alerts.Add("alert 1")
	alerts.Add("alert 2")

	worker.Start()
	worker.WaitUntilIdle()
}

The emails were added first but run last. Both alerts go before the report, and the report before the emails.

A lower rank goes first. The worker takes from the first source in the list that has a job waiting. A source with no rank set gets the largest possible rank, math.MaxInt, so it comes after every source that has a smaller one. Sources with the same rank go in the order they were bound.

Priority does not share. While a source with a lower rank has jobs waiting, a source with a higher rank gets none. A busy alerts queue can keep emails waiting.

A topic subscription is in the same list. WithTopicPriority sets its rank like WithQueuePriority does. See topics.

RoundRobin

The worker goes through the list one source per job, then starts over. It skips a source that is empty when its turn comes, so once a source is empty, the others share the turns.

package main

import (
	"fmt"

	"github.com/goptics/varmq"
)

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

	alerts := worker.BindQueue()
	reports := worker.BindQueue()
	emails := worker.BindQueue()

	alerts.Add("alert 1")
	alerts.Add("alert 2")
	reports.Add("report 1")
	emails.Add("email 1")
	emails.Add("email 2")
	emails.Add("email 3")

	worker.Start()
	worker.WaitUntilIdle()
}

One job from each queue in turn. reports is empty after its first job, so the later turns go to alerts and emails.

MaxLen and MinLen

Both compare how many jobs are waiting in each source, read again for every job.

• MaxLen takes from the longest source. Each job it takes makes that source shorter. While nothing new arrives, the turns pass to the next-longest source. So the longest sources get shorter together.
• MinLen takes from the shortest source that is not empty. It runs a short source to empty, then moves to the next shortest. A long source waits while any shorter source still has jobs.

On a tie, both take the source that comes first in the list.

package main

import (
	"fmt"

	"github.com/goptics/varmq"
)

func run(name string, strategy varmq.Strategy) {
	worker := varmq.NewWorker(
		func(j varmq.Job[string]) {
			fmt.Print(" ", j.Data())
		},
		varmq.WithStrategy(strategy),
		varmq.WithAutoRun(false),
	)

	alerts := worker.BindQueue()
	reports := worker.BindQueue()
	emails := worker.BindQueue()

	alerts.Add("alert")
	alerts.Add("alert")
	reports.Add("report")
	emails.Add("email")
	emails.Add("email")
	emails.Add("email")

	fmt.Print(name, ":")
	worker.Start()
	worker.WaitUntilIdle()
	fmt.Println()
}

func main() {
	run("MaxLen", varmq.MaxLen)
	run("MinLen", varmq.MinLen)
}

The queues start with 2 alerts, 1 report and 3 emails. MaxLen takes an email from the longest queue, then breaks the 2-2 tie with alerts, which comes first in the list. When reports and emails tie at 1 each, reports goes first, so an email runs last. MinLen empties reports, then alerts, and leaves the long emails queue for last.

“In order” here means the order the jobs leave the queues. With a concurrency above 1 several of them run at once, and their output can interleave.

What priority does under the other strategies

RoundRobin, MaxLen and MinLen do not choose by rank. They still go through the list in rank order. So a rank sets the order of the RoundRobin turns. It also sets which source wins a tie in MaxLen and MinLen. If you give no ranks, the list is the bind order.

Next: Topics.