✨ Priorityqueue: Optionally return items fairly

This change makes it possible for the priorityqueue to return items
fairly across one or more fairness dimensions. These fairness dimensions
are extracted from items using a configurable extractor func. Fairness
is only ensured within a given priority, as the purpose of priorities is
to return higher-priority items before lower-priority items.

An example use-case is to use the namespace as fairness dimension, which
then ensures that one very busy namespace can not starve off other
namespaces.

An example use-case for more than one dimension are multi-cluster
controllers, where fairness should first be achieved across clusters and
then across namespaces within each cluster.

By default, no extractor func is configured so this doesn't change any
behavior. We can revisit this in the future.

Naively, the whole problem looks like it could be solved by sorting the
items in the queue appropriately, but that doesn't really work because:

* Most (all?) sorting algorithm assume that the order of two elements
  can be determined just by looking at those two elements, but for
  fairness we have to consider the items in front of the current
  elements as well as the items that we already handed out
* We can not hand out locked items and have to ignore them for the
  purpose of fairness. This in turn means that whenever an item gets
  unlocked, we'd have to redo our sorting

This change solves the problem by introducing a `fairQueue` structure
which holds all items that are ready to be handed out and in the
highest priority bracket that has ready items. The priorityqueue then
hands out items from its internal `fairQueue` and manages its content.

The `fairQueue` works by using a slice of all the fairness dinensions
it has ever observed and a map of fairness dimension to item slice.
Whenever it hands out an item, it iterates over this slice and checks
the corresponding map for items until it found an item to hand out.

Author: alvaroaleman

pkg/controller/priorityqueue/fairqueue.go

@@ -0,0 +1,96 @@
+package priorityqueue
+
+import (
+	"slices"
+
+	"k8s.io/utils/ptr"
+)
+
+type fairq[T comparable] struct {
+	children map[string]*fairq[T]
+	items    []*item[T]
+	// order is the order in which to dequeue. Nil is used
+	// as a magic value to indicate we should dequeue from
+	// items instead of a child next.
+	order []*string
+}
+
+func (q *fairq[T]) add(i *item[T], dimensions ...string) {
+	if len(dimensions) == 0 {
+		if len(q.order) == len(q.children) {
+			q.order = append([]*string{nil}, q.order...)
+		}
+		q.items = append(q.items, i)
+
+		// Sort here so items that get unlocked can be added, rather
+		// than having to completely drain and rebuild the fairQueue.
+		slices.SortFunc(q.items, func(a, b *item[T]) int {
+			if less(a, b) {
+				return -1
+			}
+			return 1
+		})
+
+		return
+	}
+
+	dimension := dimensions[0]
+	dimensions = dimensions[1:]
+
+	if q.children == nil {
+		q.children = make(map[string]*fairq[T], 1)
+	}
+
+	_, exists := q.children[dimension]
+	if !exists {
+		q.children[dimension] = &fairq[T]{}
+		q.order = append([]*string{ptr.To(dimension)}, q.order...)
+	}
+
+	q.children[dimension].add(i, dimensions...)
+}
+
+func (q *fairq[T]) dequeue() (*item[T], bool) {
+	var item *item[T]
+	var hasItem bool
+
+	for idx, dimension := range q.order {
+		if dimension != nil { // child element
+			item, hasItem = q.children[*dimension].dequeue()
+			if !hasItem {
+				continue
+			}
+		} else if len(q.items) > 0 { // leaf element
+			item = q.items[0]
+			q.items = q.items[1:]
+		} else {
+			continue
+		}
+
+		q.order = append(q.order[:idx], q.order[idx+1:]...)
+		q.order = append(q.order, dimension)
+
+		return item, true
+	}
+
+	return item, false
+}
+
+func (q *fairq[T]) drain() {
+	for _, child := range q.children {
+		child.drain()
+	}
+	q.items = nil
+}
+
+func (q *fairq[T]) isEmpty() bool {
+	if len(q.items) > 0 {
+		return false
+	}
+	for _, child := range q.children {
+		if !child.isEmpty() {
+			return false
+		}
+	}
+	return true
+}

pkg/controller/priorityqueue/fairqueue_test.go

@@ -0,0 +1,87 @@
+package priorityqueue
+
+import (
+	"strings"
+
+	. "github.com/onsi/ginkgo/v2"
+	. "github.com/onsi/gomega"
+)
+
+var _ = Describe("FairQueue", func() {
+	DescribeTable("Ordering tests",
+		Entry("simple",
+			[]string{"foo", "foo", "bar"},
+			[]string{"bar", "foo", "foo"},
+		),
+		Entry("balanced",
+			[]string{"foo", "foo", "bar", "bar"},
+			[]string{"bar", "foo", "bar", "foo"},
+		),
+		Entry("mixed-dimensional, multi-dimensional comes first",
+			[]string{"foo-foo", "foo-foo", "foo"},
+			[]string{"foo", "foo-foo", "foo-foo"},
+		),
+		Entry("mixed-dimensional, single dimension comes first",
+			[]string{"foo", "foo-foo", "foo-foo"},
+			[]string{"foo-foo", "foo", "foo-foo"},
+		),
+		Entry("complex mixed-dimensional",
+			[]string{"bar", "foo", "foo-foo", "foo-foo", "foo-bar"},
+			// First the foo with the latest-added second dimension, then the bar,
+			// then the remaining foos based on the second dimension.
+			[]string{"foo-bar", "bar", "foo-foo", "foo", "foo-foo"},
+		),
+
+		func(adds []string, expected []string) {
+			q := &fairq[string]{}
+			for idx, add := range adds {
+				q.add(&item[string]{Key: add, AddedCounter: uint64(idx)}, strings.Split(add, "-")...)
+			}
+
+			actual := make([]string, 0, len(expected))
+			for range len(expected) {
+				item, didGetItem := q.dequeue()
+				Expect(didGetItem).To(BeTrue())
+				actual = append(actual, item.Key)
+			}
+
+			_, didGetItem := q.dequeue()
+			Expect(didGetItem).To(BeFalse())
+			Expect(actual).To(Equal(expected))
+		},
+	)
+
+	It("retains fairness across queue operations", func() {
+		q := &fairq[string]{}
+		q.add(&item[string]{Key: "foo"}, "foo")
+		_, _ = q.dequeue()
+		q.add(&item[string]{Key: "bar", AddedCounter: 1}, "bar")
+		q.add(&item[string]{Key: "foo", AddedCounter: 2}, "foo")
+
+		item, _ := q.dequeue()
+		Expect(item.Key).To(Equal("bar"))
+		Expect(q.isEmpty()).To(BeFalse())
+	})
+
+	It("sorts by added counter", func() {
+		q := &fairq[string]{}
+		q.add(&item[string]{Key: "foo", AddedCounter: 2})
+		q.add(&item[string]{Key: "bar", AddedCounter: 1})
+
+		item, _ := q.dequeue()
+		Expect(item.Key).To(Equal("bar"))
+		Expect(q.isEmpty()).To(BeFalse())
+	})
+
+	It("drains all items", func() {
+		q := &fairq[string]{}
+		q.add(&item[string]{Key: "foo"}, "foo")
+		q.add(&item[string]{Key: "bar"}, "bar")
+		q.add(&item[string]{Key: "baz"}, "baz")
+
+		q.drain()
+		_, gotItem := q.dequeue()
+		Expect(gotItem).To(BeFalse())
+		Expect(q.isEmpty()).To(BeTrue())
+	})
+})