dwrr

README

dwrr

github.com/version2beta/dwrr is a generic cost-aware Deficit Weighted Round Robin scheduler for FIFO queues. Requires Go 1.22.2 or newer.

Install

go get github.com/version2beta/dwrr@v0.10.1

Example

package main

import (
	"fmt"

	"github.com/version2beta/dwrr"
)

func main() {
	s := dwrr.NewDWRR[string](2)
	if err := s.SetQuantum(0, 2); err != nil {
		panic(err)
	}
	if err := s.SetQuantum(1, 3); err != nil {
		panic(err)
	}
	if err := s.Enqueue(0, []dwrr.Item[string]{{Value: "a", Cost: 2}, {Value: "b", Cost: 2}}); err != nil {
		panic(err)
	}
	if err := s.Enqueue(1, []dwrr.Item[string]{{Value: "x", Cost: 4}, {Value: "y", Cost: 1}}); err != nil {
		panic(err)
	}

	fmt.Println(s.Take(1)) // [{0 a 2}]; queue 0's visit remains active
	fmt.Println(s.Take(1)) // [{0 b 2}]; queue 1 gets its first quantum and blocks at cost 4
	fmt.Println(s.Take(1)) // [{1 x 4}]; queue 1 accumulated enough deficit
	fmt.Println(s.Take(1)) // [{1 y 1}]
}

(The result formatting above is illustrative; the Scheduled fields are Queue, Value, and Cost.)

Scheduling

NewDWRR[T](count) creates count queues, each with quantum 1. Queue indices start at 0, stay stable while a queue exists, and are fixed in count except for AddQueue and RemoveQueue (which removes only the last queue). A removed-last index may be reused by a later AddQueue; indices are not permanent IDs and are best suited to fixed or slowly changing lanes. AddQueue() appends an empty queue at quantum 1 and returns its index. The zero value is unsupported; use the constructor. Item[T] has a Value and a positive Cost; Scheduled[T] reports the source queue, value, and cost. SetQuantum(queue, quantum) configures a positive quantum and returns ErrZeroQuantum for zero. Setting the current quantum is idempotent and changes no scheduler state. Enqueue returns ErrZeroCost if any item has zero cost and validates the full input before changing the queue. It copies each item record, so an accepted cost cannot change if the caller later modifies its input slice; values of type T are shallow-copied. Invalid queue indices panic as slice indexing does.

The scheduler visits queues cyclically. At the start of a visit it adds that queue's quantum to its deficit, then serves affordable FIFO heads consecutively, subtracting each cost. An unaffordable head blocks later items and ends that visit. Unused deficit carries forward while the queue remains nonempty: this is what lets costs larger than one quantum eventually become eligible. Empty queues reset their deficit. A Take limit is only a retrieval limit, not a fairness cap: if it interrupts a visit, that visit and its remaining deficit persist, and the next call continues without awarding another quantum. Output is item-by-item in scheduler service order; queue identity remains in each result.

Take(n) returns up to n scheduled items in service order; it returns fewer only after the queued work is exhausted. It returns a nonnil empty slice for n=0, no queues, or no available work; negative n panics. Take(0) does not change scheduling state. The result is not preallocated from n, and its slice storage does not alias internal queues. Internal visits may process several items and cross multiple queue rounds in one call. A retrieval limit never ends a visit or caps a queue's fairness share.

Costs and quanta are limited to MaxCost and MaxQuantum, both uint64(1<<63 - 1). Enqueue returns ErrCostTooLarge for a larger cost and validates the whole batch before changing the queue. SetQuantum checks the queue index first, then returns ErrZeroQuantum for zero or ErrQuantumTooLarge above the limit, without changing state. Deficits use exact uint64 arithmetic: credit is below 2⁶³ outside an active visit, a grant adds at most MaxQuantum, and fast-forward keeps credit below the head cost. Pausing never grants credit. Once a complete cycle produces no item, the scheduler uses exact division to skip only whole unproductive cycles, crediting every nonempty queue equally before resuming visits in order.

Queue operations and mutation semantics

All methods synchronize scheduler state; do not copy a DWRR after first use. Take selects and removes work; the caller owns downstream execution and its concurrency.

• Enqueue appends FIFO items. Appending to the active queue joins that same visit at its tail without another quantum.
• SetQuantum keeps leftover credit and any active visit. A reweight takes effect at the next visit's quantum grant, not when a paused visit resumes; an unchanged quantum is a true no-op. Credit carried from a completed visit is less than the head cost at that time, so this transition effect is bounded and brief. A paused visit keeps the credit it already received.
• Take selects and removes items in service order. RemoveHead directly removes the FIFO head without scheduling it, returning an item and true or the zero item and false when empty. If items remain, it preserves deficit and the active visit; emptying the queue resets deficit and cancels its active visit. DrainQueue directly removes all items, resets deficit, and cancels its active visit. Emptying via Take or RemoveHead retains the queue's remaining storage for reuse; DrainQueue and removing the queue release that storage.
• AddQueue appends an empty queue at quantum 1 and returns its index. Existing indices and an active visit remain unchanged.
• RemoveQueue drops the last queue and its contents; removing the active queue closes its visit. Other indices remain unchanged and the cursor stays valid. Removing from no queues is a no-op.

Cost fairness is fairness over application-declared costs, not execution time, bytes, or wall-clock resource use. Synchronization protects scheduler state, not concurrent mutation of values stored in T.

Documentation

Overview

Package dwrr implements a cost-aware Deficit Weighted Round Robin scheduler. Quantum and item costs are positive uint64 values. Scheduling order is determined by queue visits and FIFO heads, not by downstream execution time.

Index

• Constants
• Variables
• type DWRR
  • func NewDWRR[T any](count uint) *DWRR[T]
  • func (d *DWRR[T]) AddQueue() uint
  • func (d *DWRR[T]) DrainQueue(queue uint) []Item[T]
  • func (d *DWRR[T]) Enqueue(queue uint, items []Item[T]) error
  • func (d *DWRR[T]) RemoveHead(queue uint) (Item[T], bool)
  • func (d *DWRR[T]) RemoveQueue()
  • func (d *DWRR[T]) SetQuantum(queue uint, quantum uint64) error
  • func (d *DWRR[T]) Take(n int) []Scheduled[T]
• type Item
• type Scheduled

Constants

const (
	// MaxCost is the largest accepted item cost.
	MaxCost = uint64(1<<63 - 1)
	// MaxQuantum is the largest accepted queue quantum.
	MaxQuantum = uint64(1<<63 - 1)
)

Variables

var (
	// ErrZeroQuantum is returned when SetQuantum is given a zero quantum.
	ErrZeroQuantum = errors.New("dwrr: quantum must be positive")
	// ErrZeroCost is returned when Enqueue receives an item with zero cost.
	ErrZeroCost = errors.New("dwrr: item cost must be positive")
	// ErrCostTooLarge is returned when Enqueue receives a cost above MaxCost.
	ErrCostTooLarge = errors.New("dwrr: item cost exceeds MaxCost")
	// ErrQuantumTooLarge is returned when SetQuantum receives a value above MaxQuantum.
	ErrQuantumTooLarge = errors.New("dwrr: quantum exceeds MaxQuantum")
)

Types

type DWRR

type DWRR[T any] struct {
	// contains filtered or unexported fields
}

DWRR is a concurrent-safe cost-aware deficit round-robin scheduler. Do not copy it after first use. Its zero value is unusable; use NewDWRR.

func NewDWRR

func NewDWRR[T any](count uint) *DWRR[T]

NewDWRR creates count empty queues, each with quantum 1. Queue indices start at zero and remain stable while their queues exist; AddQueue and RemoveQueue change the count, with RemoveQueue affecting only the last queue.

func (*DWRR[T]) AddQueue

func (d *DWRR[T]) AddQueue() uint

AddQueue appends an empty queue with quantum 1 and returns its index. The index is stable while the queue exists; if the last queue is removed, a later AddQueue may reuse its index.

func (*DWRR[T]) DrainQueue

func (d *DWRR[T]) DrainQueue(queue uint) []Item[T]

DrainQueue removes and returns a copy of every queued item, resets deficit, and cancels an active visit. Invalid indices panic.

func (*DWRR[T]) Enqueue

func (d *DWRR[T]) Enqueue(queue uint, items []Item[T]) error

Enqueue validates the complete input before appending it atomically. New items join an active visit at the queue tail. Item records, including costs, are copied; values of T are shallow-copied. Zero cost returns ErrZeroCost; a cost above MaxCost returns ErrCostTooLarge. Invalid queue indices panic.

func (*DWRR[T]) RemoveHead

func (d *DWRR[T]) RemoveHead(queue uint) (Item[T], bool)

RemoveHead removes and returns a copy of the FIFO head. It returns the zero Item and false when empty. If items remain, deficit and active-visit state are preserved; emptying the queue resets deficit and cancels an active visit. Invalid indices panic.

func (*DWRR[T]) RemoveQueue

func (d *DWRR[T]) RemoveQueue()

RemoveQueue removes the last queue and its contents. Existing queue indices remain stable. Removing the active queue cancels that visit; cursor is kept in range (or reset to zero when no queues remain). Empty removal is a no-op.

func (*DWRR[T]) SetQuantum

func (d *DWRR[T]) SetQuantum(queue uint, quantum uint64) error

SetQuantum changes a queue's quantum while keeping its deficit and any active visit. The new quantum is used at the next grant, not when a paused visit resumes. Setting the current quantum is a no-op. Zero quantum returns ErrZeroQuantum; a quantum above MaxQuantum returns ErrQuantumTooLarge. Both errors leave state unchanged. Invalid indices panic first.

func (*DWRR[T]) Take

func (d *DWRR[T]) Take(n int) []Scheduled[T]

Take returns up to n items in service order. A limit can pause an active visit; its credit and position persist for the next call. Take(0) returns a nonnil empty slice without changing state. Negative n panics. The returned slice is always nonnil and never aliases internal queue storage.

type Item

type Item[T any] struct {
	Value T      // Payload returned by Take.
	Cost  uint64 // Positive scheduling cost, fixed when Enqueue succeeds.
}

Item is a payload and its positive scheduling cost. Enqueue copies the item record, so later changes to the caller's Cost do not affect scheduling.

type Scheduled

type Scheduled[T any] struct {
	Queue uint   // Index of the source queue.
	Value T      // Scheduled payload.
	Cost  uint64 // The item's declared scheduling cost.
}

Scheduled is an item selected from Queue in scheduler service order.