diff --git a/pkg/controller/priorityqueue/priorityqueue.go b/pkg/controller/priorityqueue/priorityqueue.go index dd267cd45a..aa9ab44e22 100644 --- a/pkg/controller/priorityqueue/priorityqueue.go +++ b/pkg/controller/priorityqueue/priorityqueue.go @@ -537,20 +537,26 @@ func (w *priorityqueue[T]) logState() { if !w.log.V(5).Enabled() { continue } - w.lock.Lock() - items := make([]*item[T], 0, len(w.items)) - w.waiting.Ascend(func(item *item[T]) bool { - items = append(items, item) - return true - }) - w.ready.Ascend(func(item *item[T]) bool { - items = append(items, item) - return true - }) - w.lock.Unlock() - w.log.V(5).Info("workqueue_items", "items", items) + w.log.V(5).Info("workqueue_items", "items", w.cloneItems()) + } +} + +// cloneItems returns a deep copy of all queued items, taken under the lock, so that callers +// like logState can use them after it is released without racing with writers of ReadyAt. +func (w *priorityqueue[T]) cloneItems() []item[T] { + w.lock.Lock() + defer w.lock.Unlock() + + items := make([]item[T], 0, len(w.items)) + appendItem := func(it *item[T]) bool { + items = append(items, it.clone()) + return true } + w.waiting.Ascend(appendItem) + w.ready.Ascend(appendItem) + + return items } func lessWaiting[T comparable](a, b *item[T]) bool { @@ -574,6 +580,15 @@ type item[T comparable] struct { ReadyAt *time.Time `json:"readyAt,omitempty"` } +// clone returns a copy of the item that shares no memory with it. +func (i *item[T]) clone() item[T] { + clone := *i + if i.ReadyAt != nil { + clone.ReadyAt = new(*i.ReadyAt) + } + return clone +} + func (w *priorityqueue[T]) updateUnfinishedWorkLoop() { t := time.Tick(500 * time.Millisecond) // borrowed from workqueue: https://github.com/kubernetes/kubernetes/blob/67a807bf142c7a2a5ecfdb2a5d24b4cdea4cc79c/staging/src/k8s.io/client-go/util/workqueue/queue.go#L182 for { diff --git a/pkg/controller/priorityqueue/priorityqueue_test.go b/pkg/controller/priorityqueue/priorityqueue_test.go index bb6b658c86..798c671f6f 100644 --- a/pkg/controller/priorityqueue/priorityqueue_test.go +++ b/pkg/controller/priorityqueue/priorityqueue_test.go @@ -18,19 +18,24 @@ package priorityqueue import ( "fmt" + "math" "math/rand/v2" + "reflect" "strconv" "sync" "testing" "testing/synctest" "time" + "github.com/go-logr/logr/funcr" + "github.com/google/go-cmp/cmp" fuzz "github.com/google/gofuzz" . "github.com/onsi/ginkgo/v2" . "github.com/onsi/gomega" "k8s.io/apimachinery/pkg/util/sets" "k8s.io/client-go/util/workqueue" "k8s.io/utils/ptr" + "sigs.k8s.io/randfill" ) var _ = Describe("Controllerworkqueue", func() { @@ -614,11 +619,11 @@ func TestHighPriorityItemsAreReturnedBeforeLowPriorityItemMultipleTimes(t *testi } } -func newQueue() (*priorityqueue[string], *fakeMetricsProvider) { +func newQueue(opts ...Opt[string]) (*priorityqueue[string], *fakeMetricsProvider) { metrics := newFakeMetricsProvider() - q := New("test", func(o *Opts[string]) { + q := New("test", append([]Opt[string]{func(o *Opts[string]) { o.MetricProvider = metrics - }) + }}, opts...)...) q.(*priorityqueue[string]).waiting = &btreeInteractionValidator{ bTree: q.(*priorityqueue[string]).waiting, shouldHaveReadyAt: true, @@ -1068,3 +1073,70 @@ func TestPriorityQueueDoesNotDeadlockOnShutdownWhileHandingOutItems(t *testing.T } } } + +func TestPriorityQueueLogStateDoesNotRaceOnReadyAt(t *testing.T) { + t.Parallel() + synctest.Test(t, func(t *testing.T) { + q, _ := newQueue(func(o *Opts[string]) { + o.Log = funcr.New(func(string, string) {}, funcr.Options{Verbosity: 5}) + }) + defer q.ShutDown() + + // Re-adding an item with a lower After updates its ReadyAt, and every + // sleep lets logState log the item once, concurrently to the update. + for i := range 10 { + q.AddWithOpts(AddOpts{After: time.Hour - time.Duration(i)*time.Minute}, "foo") + time.Sleep(10 * time.Second) + } + }) +} + +func TestItemCloneCopiesAllFields(t *testing.T) { + t.Parallel() + + seed := time.Now().UnixNano() + t.Logf("seed: %d", seed) + filler := newItemFiller(seed) + + for range 100 { + original := item[string]{} + filler.Fill(&original) + + if diff := cmp.Diff(original, original.clone()); diff != "" { + t.Errorf("clone doesn't match the original item: %s", diff) + } + } +} + +func TestItemCloneDoesNotShareMemoryWithOriginal(t *testing.T) { + t.Parallel() + + seed := time.Now().UnixNano() + t.Logf("seed: %d", seed) + filler := newItemFiller(seed) + + for range 100 { + original := item[string]{} + filler.Fill(&original) + + originalValue, cloneValue := reflect.ValueOf(original), reflect.ValueOf(original.clone()) + for i := range cloneValue.NumField() { + field := cloneValue.Field(i) + if kind := field.Kind(); kind != reflect.Pointer && kind != reflect.Map && kind != reflect.Slice { + continue + } + + if !field.IsNil() && field.Pointer() == originalValue.Field(i).Pointer() { + t.Fatalf("field %s of the clone is shared with the original item", cloneValue.Type().Field(i).Name) + } + } + } +} + +// newItemFiller returns a filler that also fills time.Time, which randfill +// leaves at its zero value as all of its fields are unexported. +func newItemFiller(seed int64) *randfill.Filler { + return randfill.NewWithSeed(seed).NilChance(0).Funcs(func(t *time.Time, c randfill.Continue) { + *t = time.Unix(c.Int63n(math.MaxInt32), c.Int63n(int64(time.Second))) + }) +}