diff --git a/staging/src/k8s.io/client-go/tools/cache/controller.go b/staging/src/k8s.io/client-go/tools/cache/controller.go index cf4f0b0c1c5..16e277a3143 100644 --- a/staging/src/k8s.io/client-go/tools/cache/controller.go +++ b/staging/src/k8s.io/client-go/tools/cache/controller.go @@ -231,7 +231,7 @@ func (c *controller) processLoop(ctx context.Context) { default: var err error if useBatchProcess { - err = batchQueue.PopBatch(c.config.ProcessBatch) + err = batchQueue.PopBatch(c.config.ProcessBatch, PopProcessFunc(c.config.Process)) } else { // otherwise fallback to non-batch process behavior _, err = c.config.Pop(PopProcessFunc(c.config.Process)) 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 eb3ef589bd8..6c0e3ed8072 100644 --- a/staging/src/k8s.io/client-go/tools/cache/fifo.go +++ b/staging/src/k8s.io/client-go/tools/cache/fifo.go @@ -73,8 +73,12 @@ type QueueWithBatch interface { // PopBatch behaves similarly to Queue#Pop, but processes multiple keys // as a batch. The implementation determines the batching strategy, - // such as the number of keys to include per batch. - PopBatch(ProcessBatchFunc) error + // such as the number of keys to include per batch. The ProcessBatchFunc + // 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. + PopBatch(processBatch ProcessBatchFunc, processSingle PopProcessFunc) error } // Pop is helper function for popping from Queue. 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 bffdbeb0a1c..3afd79af59a 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 @@ -264,7 +264,9 @@ func (f *RealFIFO) Pop(process PopProcessFunc) (interface{}, error) { return Deltas{item}, err } -func (f *RealFIFO) PopBatch(process ProcessBatchFunc) error { +// 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. +func (f *RealFIFO) PopBatch(processBatch ProcessBatchFunc, processSingle PopProcessFunc) error { f.lock.Lock() defer f.lock.Unlock() @@ -304,6 +306,7 @@ func (f *RealFIFO) PopBatch(process ProcessBatchFunc) error { break } if unique.Has(id) { + // close the batch if a duplicate item is encountered break } unique.Insert(id) @@ -329,8 +332,10 @@ func (f *RealFIFO) PopBatch(process ProcessBatchFunc) error { defer trace.LogIfLong(min(100*time.Millisecond*time.Duration(len(deltas)), time.Second)) } - err := process(deltas, isInInitialList) - return err + if len(deltas) == 1 { + return processSingle(Deltas{deltas[0]}, isInInitialList) + } + return processBatch(deltas, isInInitialList) } // Replace diff --git a/staging/src/k8s.io/client-go/tools/cache/the_real_fifo_test.go b/staging/src/k8s.io/client-go/tools/cache/the_real_fifo_test.go index bfd3bcfb928..a5d26ab4a3f 100644 --- a/staging/src/k8s.io/client-go/tools/cache/the_real_fifo_test.go +++ b/staging/src/k8s.io/client-go/tools/cache/the_real_fifo_test.go @@ -1141,7 +1141,13 @@ func TestRealFIFO_PopMultipleDeltaInBatch(t *testing.T) { receivedInitial <- obj } return nil - }) + }, PopProcessFunc(func(obj interface{}, isInInitialList bool) error { + received <- []Delta{*obj.(Deltas).Newest()} + if isInInitialList { + receivedInitial <- []Delta{*obj.(Deltas).Newest()} + } + return nil + })) }() timer := time.NewTimer(time.Millisecond * 50) select { @@ -1265,7 +1271,10 @@ func TestRealFIFO_PopBrokenItemsInBatch(t *testing.T) { _ = f.PopBatch(func(obj []Delta, isInInitialList bool) error { received <- obj return nil - }) + }, PopProcessFunc(func(obj interface{}, isInInitialList bool) error { + received <- []Delta{*obj.(Deltas).Newest()} + return nil + })) }() timer := time.NewTimer(time.Millisecond * 50) select {