From 40c01b99a7bac05f6de438d67dd20a60333a0ac4 Mon Sep 17 00:00:00 2001 From: Michael Aspinwall Date: Fri, 16 Jan 2026 00:46:52 +0000 Subject: [PATCH 1/2] Ensure that processing does not block queue writers --- pkg/scheduler/schedule_one_test.go | 12 ++- .../client-go/features/known_features.go | 9 ++ .../src/k8s.io/client-go/tools/cache/fifo.go | 23 ++--- .../k8s.io/client-go/tools/cache/fifo_test.go | 15 +++ .../client-go/tools/cache/the_real_fifo.go | 98 +++++++++++++++---- .../reference/feature_list.md | 1 + 6 files changed, 120 insertions(+), 38 deletions(-) diff --git a/pkg/scheduler/schedule_one_test.go b/pkg/scheduler/schedule_one_test.go index 592ff01a84e..b85189d35a6 100644 --- a/pkg/scheduler/schedule_one_test.go +++ b/pkg/scheduler/schedule_one_test.go @@ -4742,7 +4742,7 @@ func setupTestScheduler(ctx context.Context, t *testing.T, client clientset.Inte nodeInfoSnapshot: snapshot, percentageOfNodesToScore: schedulerapi.DefaultPercentageOfNodesToScore, NextPod: func(logger klog.Logger) (*framework.QueuedPodInfo, error) { - return &framework.QueuedPodInfo{PodInfo: mustNewPodInfo(t, clientcache.Pop(queuedPodStore).(*v1.Pod))}, nil + return &framework.QueuedPodInfo{PodInfo: mustNewPodInfo(t, pop(queuedPodStore).(*v1.Pod))}, nil }, SchedulingQueue: schedulingQueue, APIDispatcher: apiDispatcher, @@ -4825,3 +4825,13 @@ func mustNewPodInfo(t *testing.T, pod *v1.Pod) *framework.PodInfo { } return podInfo } + +func pop(queue clientcache.Queue) interface{} { + obj, err := queue.Pop(func(obj interface{}, isInInitialList bool) error { + return nil + }) + if err != nil { + return nil + } + return obj +} diff --git a/staging/src/k8s.io/client-go/features/known_features.go b/staging/src/k8s.io/client-go/features/known_features.go index ad5a3d75db4..56785991d81 100644 --- a/staging/src/k8s.io/client-go/features/known_features.go +++ b/staging/src/k8s.io/client-go/features/known_features.go @@ -76,6 +76,12 @@ const ( // GA: v1.35 InformerResourceVersion Feature = "InformerResourceVersion" + // owner: @michaelasp + // beta: v1.36 + // + // Allow the FIFO to unlock while processing items to allow other goroutines to add items to the queue. + UnlockWhileProcessingFIFO Feature = "UnlockWhileProcessingFIFO" + // owner: @p0lyn0mial // beta: v1.30 // @@ -108,6 +114,9 @@ var defaultVersionedKubernetesFeatureGates = map[Feature]VersionedSpecs{ {Version: version.MustParse("1.30"), Default: false, PreRelease: Alpha}, {Version: version.MustParse("1.35"), Default: true, PreRelease: GA}, }, + UnlockWhileProcessingFIFO: { + {Version: version.MustParse("1.36"), Default: true, PreRelease: Beta}, + }, WatchListClient: { {Version: version.MustParse("1.30"), Default: false, PreRelease: Beta}, {Version: version.MustParse("1.35"), Default: true, PreRelease: Beta}, diff --git a/staging/src/k8s.io/client-go/tools/cache/fifo.go b/staging/src/k8s.io/client-go/tools/cache/fifo.go index 6c0e3ed8072..44d006b6987 100644 --- a/staging/src/k8s.io/client-go/tools/cache/fifo.go +++ b/staging/src/k8s.io/client-go/tools/cache/fifo.go @@ -47,7 +47,9 @@ type Queue interface { // may return an ErrRequeue{inner} and in this case Pop will (a) // return that (key, accumulator) association to the Queue as part // of the atomic processing and (b) return the inner error from - // Pop. + // Pop. It is expected that the caller of Pop will be a single + // threaded consumer since otherwise it is possible for multiple + // PopProcessFuncs to be running simultaneously. Pop(PopProcessFunc) (interface{}, error) // HasSynced returns true if the first batch of keys have all been @@ -77,25 +79,12 @@ type QueueWithBatch interface { // is called when a batch is ready to be processed. The PopProcessFunc // is called when a singleton item is ready to be processed. The // ProcessBatchFunc and PopProcessFunc must do the same processing to - // ensure consistent behavior. + // ensure consistent behavior. It is expected that the caller of PopBatch + // will be a single threaded consumer since otherwise it is possible for + // multiple ProcessBatchFuncs/PopProcessFuncs to be running simultaneously. PopBatch(processBatch ProcessBatchFunc, processSingle PopProcessFunc) error } -// Pop is helper function for popping from Queue. -// WARNING: Do NOT use this function in non-test code to avoid races -// unless you really really really really know what you are doing. -// -// NOTE: This function is deprecated and may be removed in the future without -// additional warning. -func Pop(queue Queue) interface{} { - var result interface{} - queue.Pop(func(obj interface{}, isInInitialList bool) error { - result = obj - return nil - }) - return result -} - // FIFO is a Queue in which (a) each accumulator is simply the most // recently provided object and (b) the collection of keys to process // is a FIFO. The accumulators all start out empty, and deleting an diff --git a/staging/src/k8s.io/client-go/tools/cache/fifo_test.go b/staging/src/k8s.io/client-go/tools/cache/fifo_test.go index 39c27c6d9c5..41bdee72702 100644 --- a/staging/src/k8s.io/client-go/tools/cache/fifo_test.go +++ b/staging/src/k8s.io/client-go/tools/cache/fifo_test.go @@ -67,6 +67,21 @@ func (f *FIFO) GetByKey(key string) (item interface{}, exists bool, err error) { return item, exists, nil } +// Pop is helper function for popping from Queue. +// WARNING: Do NOT use this function in non-test code to avoid races +// unless you really really really really know what you are doing. +// +// NOTE: This function is deprecated and may be removed in the future without +// additional warning. +func Pop(queue Queue) interface{} { + var result interface{} + queue.Pop(func(obj interface{}, isInInitialList bool) error { + result = obj + return nil + }) + return result +} + func testFifoObjectKeyFunc(obj interface{}) (string, error) { return obj.(testFifoObject).name, nil } diff --git a/staging/src/k8s.io/client-go/tools/cache/the_real_fifo.go b/staging/src/k8s.io/client-go/tools/cache/the_real_fifo.go index 76a68e17f1e..96d282f2caa 100644 --- a/staging/src/k8s.io/client-go/tools/cache/the_real_fifo.go +++ b/staging/src/k8s.io/client-go/tools/cache/the_real_fifo.go @@ -50,6 +50,12 @@ type RealFIFOOptions struct { // atomically for Replace and Resync operations. // If AtomicEvents is true, KnownObjects must be nil. AtomicEvents bool + + // UnlockWhileProcessing is used to specify whether the RealFIFO can unlock + // the lock while processing events. If it is set, the lock can be unlocked + // while processing events to allow other goroutines to add items to the queue. + // If UnlockWhileProcessing is true, AtomicEvents must be true as well. + UnlockWhileProcessing bool } const ( @@ -102,6 +108,11 @@ type RealFIFO struct { // * a single ReplacedAll event will be emitted instead of multiple Replace events // * a single SyncAll event will be emitted instead of multiple Sync events emitAtomicEvents bool + + // unlockWhileProcessing defines whether we can unlock while processing events. + // This may only be set if emitAtomicEvents is true. If unlockWhileProcessing is true, + // Pop and PopBatch must be called from a single threaded consumer. + unlockWhileProcessing bool } // ReplacedAllInfo is the object associated with a Delta of type=ReplacedAll @@ -285,10 +296,14 @@ func (f *RealFIFO) IsClosed() bool { } // Pop waits until an item is ready and processes it. If multiple items are -// ready, they are returned in the order in which they were added/updated. -// The item is removed from the queue (and the store) before it is processed. -// process function is called under lock, so it is safe -// update data structures in it that need to be in sync with the queue. +// ready, they are returned in the order in which they were added/updated. The +// item is removed from the queue (and the store) before it is processed. The +// process function is only guaranteed to be called under lock if +// UnlockWhileProcessing is false. If the process function is updating data +// structures that need to be in sync with the queue, ensure +// UnlockWhileProcessing is false. It is expected that the caller of Pop will be +// a single threaded consumer since otherwise it is possible for multiple +// PopProcessFuncs to be running simultaneously. func (f *RealFIFO) Pop(process PopProcessFunc) (interface{}, error) { f.lock.Lock() defer f.lock.Unlock() @@ -309,9 +324,14 @@ func (f *RealFIFO) Pop(process PopProcessFunc) (interface{}, error) { // The underlying array still exists and references this object, so the object will not be garbage collected unless we zero the reference. f.items[0] = Delta{} f.items = f.items[1:] - if f.initialPopulationCount > 0 { - f.initialPopulationCount-- - } + // Decrement initialPopulationCount if needed. + // This is done in a defer so we only do this *after* processing is complete, + // so concurrent calls to hasSynced will not incorrectly return true while processing is still happening. + defer func() { + if f.initialPopulationCount > 0 { + f.initialPopulationCount-- + } + }() // Only log traces if the queue depth is greater than 10 and it takes more than // 100 milliseconds to process one item from the queue. @@ -327,11 +347,28 @@ func (f *RealFIFO) Pop(process PopProcessFunc) (interface{}, error) { defer trace.LogIfLong(100 * time.Millisecond) } - // we wrap in Deltas here to be compatible with preview Pop functions and those interpreting the return value. - err := process(Deltas{item}, isInInitialList) + // Process the item, this may unlock the lock, and allow other goroutines to add items to the queue. + err := f.whileProcessing_locked(func() error { + // we wrap in Deltas here to be compatible with preview Pop functions and those interpreting the return value. + return process(Deltas{item}, isInInitialList) + }) return Deltas{item}, err } +// whileProcessing_locked calls the `process` function. +// The lock must be held before calling `whileProcessing_locked`, and is held when `whileProcessing_locked` returns. +// whileProcessing_locked releases the lock during the call to `process` if f.unlockWhileProcessing is true and the f.items queue is not too long. +func (f *RealFIFO) whileProcessing_locked(process func() error) error { + // Unlock before calling `process` so new items can be enqueued during processing. + // Only do this if the queue contains less than 2 full batches of items, + // to prevent the queue from growing unboundedly. + if f.unlockWhileProcessing && len(f.items) < f.batchSize*2 { + f.lock.Unlock() + defer f.lock.Lock() + } + return process() +} + // batchable stores the delta types that can be batched var batchable = map[DeltaType]bool{ Sync: true, @@ -343,6 +380,13 @@ var batchable = map[DeltaType]bool{ // PopBatch pops as many items as possible to be processed as a batch using processBatch, // or pop a single item using processSingle if multiple items cannot be batched. +// +// The processBatch and processSingle functions are only guaranteed to be called +// under lock if UnlockWhileProcessing is false. If the process functions are +// updating data structures that need to be in sync with the queue, ensure +// UnlockWhileProcessing is false. It is expected that the caller of PopBatch +// will be a single threaded consumer, since otherwise it is possible for +// multiple ProcessBatchFunc or PopProcessFunc's to be running simultaneously. func (f *RealFIFO) PopBatch(processBatch ProcessBatchFunc, processSingle PopProcessFunc) error { f.lock.Lock() defer f.lock.Unlock() @@ -397,10 +441,16 @@ func (f *RealFIFO) PopBatch(processBatch ProcessBatchFunc, processSingle PopProc unique.Insert(id) moveDeltaToProcessList(i) } - if f.initialPopulationCount > 0 { - f.initialPopulationCount -= len(deltas) - } + f.items = f.items[len(deltas):] + // Decrement initialPopulationCount if needed. + // This is done in a defer so we only do this *after* processing is complete, + // so concurrent calls to hasSynced will not incorrectly return true while processing is still happening. + defer func() { + if f.initialPopulationCount > 0 { + f.initialPopulationCount -= len(deltas) + } + }() // Only log traces if the queue depth is greater than 10 and it takes more than // 100 milliseconds to process one item from the queue (with a max of 1 second for the whole batch) @@ -418,9 +468,13 @@ func (f *RealFIFO) PopBatch(processBatch ProcessBatchFunc, processSingle PopProc } if len(deltas) == 1 { - return processSingle(Deltas{deltas[0]}, isInInitialList) + return f.whileProcessing_locked(func() error { + return processSingle(Deltas{deltas[0]}, isInInitialList) + }) } - return processBatch(deltas, isInInitialList) + return f.whileProcessing_locked(func() error { + return processBatch(deltas, isInInitialList) + }) } // Replace @@ -641,18 +695,22 @@ func NewRealFIFOWithOptions(opts RealFIFOOptions) *RealFIFO { panic("coding error: knownObjects must not be provided when AtomicEvents is true") } } else { + if opts.UnlockWhileProcessing { + panic("coding error: UnlockWhileProcessing must be false when AtomicEvents is false") + } if opts.KnownObjects == nil { panic("coding error: knownObjects must be provided when AtomicEvents is false") } } f := &RealFIFO{ - items: make([]Delta, 0, 10), - keyFunc: opts.KeyFunction, - knownObjects: opts.KnownObjects, - transformer: opts.Transformer, - batchSize: defaultBatchSize, - emitAtomicEvents: opts.AtomicEvents, + items: make([]Delta, 0, 10), + keyFunc: opts.KeyFunction, + knownObjects: opts.KnownObjects, + transformer: opts.Transformer, + batchSize: defaultBatchSize, + emitAtomicEvents: opts.AtomicEvents, + unlockWhileProcessing: opts.UnlockWhileProcessing, } f.cond.L = &f.lock diff --git a/test/compatibility_lifecycle/reference/feature_list.md b/test/compatibility_lifecycle/reference/feature_list.md index b6bac32a865..e19eca60fd0 100644 --- a/test/compatibility_lifecycle/reference/feature_list.md +++ b/test/compatibility_lifecycle/reference/feature_list.md @@ -205,6 +205,7 @@ | TranslateStreamCloseWebsocketRequests | :ballot_box_with_check: 1.30+ | | 1.29 | 1.30– | | | | [code](https://cs.k8s.io/?q=%5CbTranslateStreamCloseWebsocketRequests%5Cb&i=nope&files=&excludeFiles=CHANGELOG&repos=kubernetes/kubernetes) [KEPs](https://cs.k8s.io/?q=%5CbTranslateStreamCloseWebsocketRequests%5Cb&i=nope&files=&excludeFiles=CHANGELOG&repos=kubernetes/enhancements) | | UnauthenticatedHTTP2DOSMitigation | :ballot_box_with_check: 1.29+ | | | 1.25– | | | | [code](https://cs.k8s.io/?q=%5CbUnauthenticatedHTTP2DOSMitigation%5Cb&i=nope&files=&excludeFiles=CHANGELOG&repos=kubernetes/kubernetes) [KEPs](https://cs.k8s.io/?q=%5CbUnauthenticatedHTTP2DOSMitigation%5Cb&i=nope&files=&excludeFiles=CHANGELOG&repos=kubernetes/enhancements) | | UnknownVersionInteroperabilityProxy | | | 1.28– | | | | APIServerIdentity | [code](https://cs.k8s.io/?q=%5CbUnknownVersionInteroperabilityProxy%5Cb&i=nope&files=&excludeFiles=CHANGELOG&repos=kubernetes/kubernetes) [KEPs](https://cs.k8s.io/?q=%5CbUnknownVersionInteroperabilityProxy%5Cb&i=nope&files=&excludeFiles=CHANGELOG&repos=kubernetes/enhancements) | +| UnlockWhileProcessingFIFO | :ballot_box_with_check: 1.36+ | | | 1.36– | | | | [code](https://cs.k8s.io/?q=%5CbUnlockWhileProcessingFIFO%5Cb&i=nope&files=&excludeFiles=CHANGELOG&repos=kubernetes/kubernetes) [KEPs](https://cs.k8s.io/?q=%5CbUnlockWhileProcessingFIFO%5Cb&i=nope&files=&excludeFiles=CHANGELOG&repos=kubernetes/enhancements) | | UserNamespacesHostNetworkSupport | | | 1.35– | | | | UserNamespacesSupport | [code](https://cs.k8s.io/?q=%5CbUserNamespacesHostNetworkSupport%5Cb&i=nope&files=&excludeFiles=CHANGELOG&repos=kubernetes/kubernetes) [KEPs](https://cs.k8s.io/?q=%5CbUserNamespacesHostNetworkSupport%5Cb&i=nope&files=&excludeFiles=CHANGELOG&repos=kubernetes/enhancements) | | UserNamespacesSupport | :ballot_box_with_check: 1.33+ | | 1.25–1.29 | 1.30– | | | | [code](https://cs.k8s.io/?q=%5CbUserNamespacesSupport%5Cb&i=nope&files=&excludeFiles=CHANGELOG&repos=kubernetes/kubernetes) [KEPs](https://cs.k8s.io/?q=%5CbUserNamespacesSupport%5Cb&i=nope&files=&excludeFiles=CHANGELOG&repos=kubernetes/enhancements) | | VolumeAttributesClass | :ballot_box_with_check: 1.34+ | :closed_lock_with_key: 1.36+ | 1.29–1.30 | 1.31–1.33 | 1.34– | | | [code](https://cs.k8s.io/?q=%5CbVolumeAttributesClass%5Cb&i=nope&files=&excludeFiles=CHANGELOG&repos=kubernetes/kubernetes) [KEPs](https://cs.k8s.io/?q=%5CbVolumeAttributesClass%5Cb&i=nope&files=&excludeFiles=CHANGELOG&repos=kubernetes/enhancements) | From 082f38d97bb68e77347b48dd5b29023b1a357373 Mon Sep 17 00:00:00 2001 From: Michael Aspinwall Date: Fri, 16 Jan 2026 00:47:09 +0000 Subject: [PATCH 2/2] Add benchmarks for adding objects with slow handlers --- .../tools/cache/controller_bench_test.go | 110 ++++++++++++++++++ 1 file changed, 110 insertions(+) create mode 100644 staging/src/k8s.io/client-go/tools/cache/controller_bench_test.go diff --git a/staging/src/k8s.io/client-go/tools/cache/controller_bench_test.go b/staging/src/k8s.io/client-go/tools/cache/controller_bench_test.go new file mode 100644 index 00000000000..754ab3f1236 --- /dev/null +++ b/staging/src/k8s.io/client-go/tools/cache/controller_bench_test.go @@ -0,0 +1,110 @@ +/* +Copyright The Kubernetes Authors. + +Licensed under the Apache License, Version 2.0 (the "License"); +you may not use this file except in compliance with the License. +You may obtain a copy of the License at + + http://www.apache.org/licenses/LICENSE-2.0 + +Unless required by applicable law or agreed to in writing, software +distributed under the License is distributed on an "AS IS" BASIS, +WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. +See the License for the specific language governing permissions and +limitations under the License. +*/ + +package cache + +import ( + "context" + "errors" + "fmt" + "testing" + "time" + + v1 "k8s.io/api/core/v1" + metav1 "k8s.io/apimachinery/pkg/apis/meta/v1" + fcache "k8s.io/client-go/tools/cache/testing" +) + +const handlerWaitTime = time.Millisecond + +func BenchmarkAddWithSlowHandlers(b *testing.B) { + for _, unlockWhileProcessing := range []bool{false, true} { + b.Run(fmt.Sprintf("unlockWhileProcessing=%t", unlockWhileProcessing), func(b *testing.B) { + ctx, cancel := context.WithCancel(context.Background()) + source := fcache.NewFakeControllerSource() + b.Cleanup(func() { + cancel() + source.Shutdown() + }) + testIDs := []string{"a-hello"} + source.Add(&v1.Pod{ObjectMeta: metav1.ObjectMeta{Name: testIDs[0]}}) + + store := NewStore(DeletionHandlingMetaNamespaceKeyFunc) + fifoOptions := RealFIFOOptions{} + if unlockWhileProcessing { + fifoOptions.UnlockWhileProcessing = true + fifoOptions.AtomicEvents = true + } else { + fifoOptions.KnownObjects = store + } + fifo := NewRealFIFOWithOptions(fifoOptions) + handler := ResourceEventHandlerFuncs{ + UpdateFunc: func(oldObj, newObj interface{}) { + time.Sleep(handlerWaitTime) + }, + } + + cfg := &Config{ + Queue: fifo, + ListerWatcher: source, + ObjectType: &v1.Pod{}, + FullResyncPeriod: 0, + + Process: func(obj interface{}, isInInitialList bool) error { + if deltas, ok := obj.(Deltas); ok { + return processDeltas(handler, store, deltas, isInInitialList, DeletionHandlingMetaNamespaceKeyFunc) + } + return errors.New("object given as Process argument is not Deltas") + }, + ProcessBatch: func(deltaList []Delta, isInInitialList bool) error { + return processDeltasInBatch(handler, store, deltaList, isInInitialList, DeletionHandlingMetaNamespaceKeyFunc) + }, + } + c := New(cfg) + go c.RunWithContext(ctx) + if !WaitForCacheSync(ctx.Done(), c.HasSynced) { + b.Fatal("Timed out waiting for cache sync") + } + + // Producer: Modify object as fast as the handler can process it. This should ensure that the process func is always running. + go func() { + // Stop when the benchmark context is cancelled. + ticker := time.NewTicker(handlerWaitTime) + defer ticker.Stop() + for { + select { + case <-ctx.Done(): + return + case <-ticker.C: + source.Modify(&v1.Pod{ObjectMeta: metav1.ObjectMeta{Name: testIDs[0]}}) + } + } + }() + benchmarkReflectorWithSlowHandlers(b, fifo) + }) + } +} + +func benchmarkReflectorWithSlowHandlers(b *testing.B, fifo *RealFIFO) { + b.ResetTimer() + // Try adding an object to the queue, while the controller is processing other events. + for i := 0; i < b.N; i++ { + if err := fifo.Add(&v1.Pod{ObjectMeta: metav1.ObjectMeta{Name: "b-hello"}}); err != nil { + b.Fatal(err) + } + } + b.StopTimer() +}