From f67d30b3529ea970dd5fd069eaddfc7f61d74169 Mon Sep 17 00:00:00 2001 From: Keisuke Ishigami Date: Thu, 29 May 2025 00:34:43 +0900 Subject: [PATCH 1/6] handle context in process loop --- .../k8s.io/client-go/tools/cache/controller.go | 16 +++++++++------- 1 file changed, 9 insertions(+), 7 deletions(-) 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 1497700d818..f0a483b6f5b 100644 --- a/staging/src/k8s.io/client-go/tools/cache/controller.go +++ b/staging/src/k8s.io/client-go/tools/cache/controller.go @@ -204,13 +204,15 @@ func (c *controller) LastSyncResourceVersion() string { // concurrently. func (c *controller) processLoop(ctx context.Context) { for { - // TODO: Plumb through the ctx so that this can - // actually exit when the controller is stopped. Or just give up on this stuff - // ever being stoppable. - _, err := c.config.Queue.Pop(PopProcessFunc(c.config.Process)) - if err != nil { - if err == ErrFIFOClosed { - return + select { + case <-ctx.Done(): + return + default: + _, err := c.config.Queue.Pop(PopProcessFunc(c.config.Process)) + if err != nil { + if err == ErrFIFOClosed { + return + } } } } From 1c33d98762511c10f89c40358a1935250b03b0c8 Mon Sep 17 00:00:00 2001 From: Keisuke Ishigami Date: Sat, 14 Jun 2025 21:59:56 +0900 Subject: [PATCH 2/6] pop respects the context --- .../client-go/tools/cache/controller.go | 4 ++-- .../client-go/tools/cache/delta_fifo.go | 8 ++++++- .../src/k8s.io/client-go/tools/cache/fifo.go | 21 +++++++++++++------ .../client-go/tools/cache/the_real_fifo.go | 8 ++++++- 4 files changed, 31 insertions(+), 10 deletions(-) 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 f0a483b6f5b..8bafcb67bc0 100644 --- a/staging/src/k8s.io/client-go/tools/cache/controller.go +++ b/staging/src/k8s.io/client-go/tools/cache/controller.go @@ -208,9 +208,9 @@ func (c *controller) processLoop(ctx context.Context) { case <-ctx.Done(): return default: - _, err := c.config.Queue.Pop(PopProcessFunc(c.config.Process)) + _, err := c.config.Queue.Pop(ctx, PopProcessFunc(c.config.Process)) if err != nil { - if err == ErrFIFOClosed { + if errors.Is(err, ErrFIFOClosed) || errors.Is(err, ErrCtxDone) { return } } diff --git a/staging/src/k8s.io/client-go/tools/cache/delta_fifo.go b/staging/src/k8s.io/client-go/tools/cache/delta_fifo.go index 264d7559a05..57872a1e1a8 100644 --- a/staging/src/k8s.io/client-go/tools/cache/delta_fifo.go +++ b/staging/src/k8s.io/client-go/tools/cache/delta_fifo.go @@ -17,6 +17,7 @@ limitations under the License. package cache import ( + "context" "errors" "fmt" "sync" @@ -490,11 +491,16 @@ func (f *DeltaFIFO) IsClosed() bool { // // Pop returns a 'Deltas', which has a complete list of all the things // that happened to the object (deltas) while it was sitting in the queue. -func (f *DeltaFIFO) Pop(process PopProcessFunc) (interface{}, error) { +func (f *DeltaFIFO) Pop(ctx context.Context, process PopProcessFunc) (interface{}, error) { f.lock.Lock() defer f.lock.Unlock() for { for len(f.queue) == 0 { + // if the length of the queue is 0 and ctx is done, return here + if ctx.Err() != nil { + return nil, ErrCtxDone + } + // When the queue is empty, invocation of Pop() is blocked until new item is enqueued. // When Close() is called, the f.closed is set and the condition is broadcasted. // Which causes this loop to continue and return from the Pop(). 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 5c2ca90084d..fde34edc5b2 100644 --- a/staging/src/k8s.io/client-go/tools/cache/fifo.go +++ b/staging/src/k8s.io/client-go/tools/cache/fifo.go @@ -17,6 +17,7 @@ limitations under the License. package cache import ( + "context" "errors" "sync" @@ -30,6 +31,9 @@ type PopProcessFunc func(obj interface{}, isInInitialList bool) error // ErrFIFOClosed used when FIFO is closed var ErrFIFOClosed = errors.New("DeltaFIFO: manipulating with closed queue") +// ErrCtxDone is used when context is done +var ErrCtxDone = errors.New("DeltaFIFO: context done") + // Queue extends ReflectorStore with a collection of Store keys to "process". // Every Add, Update, or Delete may put the object's key in that collection. // A Queue has a way to derive the corresponding key given an accumulator. @@ -38,8 +42,8 @@ var ErrFIFOClosed = errors.New("DeltaFIFO: manipulating with closed queue") type Queue interface { ReflectorStore - // Pop blocks until there is at least one key to process or the - // Queue is closed. In the latter case Pop returns with an error. + // Pop blocks until there is at least one key to process, ctx is done, + // or the Queue is closed. In the latter case Pop returns with an error. // In the former case Pop atomically picks one key to process, // removes that (key, accumulator) association from the Store, and // processes the accumulator. Pop returns the accumulator that @@ -48,7 +52,7 @@ type Queue interface { // return that (key, accumulator) association to the Queue as part // of the atomic processing and (b) return the inner error from // Pop. - Pop(PopProcessFunc) (interface{}, error) + Pop(context.Context, PopProcessFunc) (interface{}, error) // HasSynced returns true if the first batch of keys have all been // popped. The first batch of keys are those of the first Replace @@ -66,9 +70,9 @@ type Queue interface { // // NOTE: This function is deprecated and may be removed in the future without // additional warning. -func Pop(queue Queue) interface{} { +func Pop(ctx context.Context, queue Queue) interface{} { var result interface{} - queue.Pop(func(obj interface{}, isInInitialList bool) error { + queue.Pop(ctx, func(obj interface{}, isInInitialList bool) error { result = obj return nil }) @@ -191,11 +195,16 @@ func (f *FIFO) IsClosed() bool { // so if you don't successfully process it, it should be added back with // AddIfNotPresent(). process function is called under lock, so it is safe // update data structures in it that need to be in sync with the queue. -func (f *FIFO) Pop(process PopProcessFunc) (interface{}, error) { +func (f *FIFO) Pop(ctx context.Context, process PopProcessFunc) (interface{}, error) { f.lock.Lock() defer f.lock.Unlock() for { for len(f.queue) == 0 { + // if the length of the queue is 0 and ctx is done, return here + if ctx.Err() != nil { + return nil, ErrCtxDone + } + // When the queue is empty, invocation of Pop() is blocked until new item is enqueued. // When Close() is called, the f.closed is set and the condition is broadcasted. // Which causes this loop to continue and return from the Pop(). 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 9be14ff3811..337e3a8fabe 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 @@ -17,6 +17,7 @@ limitations under the License. package cache import ( + "context" "fmt" utilruntime "k8s.io/apimachinery/pkg/util/runtime" "k8s.io/apimachinery/pkg/util/sets" @@ -191,11 +192,16 @@ func (f *RealFIFO) IsClosed() bool { // 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. -func (f *RealFIFO) Pop(process PopProcessFunc) (interface{}, error) { +func (f *RealFIFO) Pop(ctx context.Context, process PopProcessFunc) (interface{}, error) { f.lock.Lock() defer f.lock.Unlock() for len(f.items) == 0 { + // if the length of the queue is 0 and ctx is done, return here + if ctx.Err() != nil { + return nil, ErrCtxDone + } + // When the queue is empty, invocation of Pop() is blocked until new item is enqueued. // When Close() is called, the f.closed is set and the condition is broadcasted. // Which causes this loop to continue and return from the Pop(). From 2cd5dbbdaab98020b37df7125f622db7a6dd885a Mon Sep 17 00:00:00 2001 From: Keisuke Ishigami Date: Sat, 14 Jun 2025 22:04:22 +0900 Subject: [PATCH 3/6] modify tests --- pkg/scheduler/schedule_one_test.go | 2 +- .../client-go/tools/cache/delta_fifo_test.go | 43 ++++++++++--------- .../k8s.io/client-go/tools/cache/fifo_test.go | 23 +++++----- .../client-go/tools/cache/reflector_test.go | 2 +- .../tools/cache/the_real_fifo_test.go | 33 +++++++------- 5 files changed, 53 insertions(+), 50 deletions(-) diff --git a/pkg/scheduler/schedule_one_test.go b/pkg/scheduler/schedule_one_test.go index 689bddbaac4..4629878122c 100644 --- a/pkg/scheduler/schedule_one_test.go +++ b/pkg/scheduler/schedule_one_test.go @@ -4199,7 +4199,7 @@ func setupTestScheduler(ctx context.Context, t *testing.T, queuedPodStore *clien 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, clientcache.Pop(context.Background(), queuedPodStore).(*v1.Pod))}, nil }, SchedulingQueue: schedulingQueue, Profiles: profile.Map{testSchedulerName: fwk}, diff --git a/staging/src/k8s.io/client-go/tools/cache/delta_fifo_test.go b/staging/src/k8s.io/client-go/tools/cache/delta_fifo_test.go index 9c67012cc1c..b3e80897ad0 100644 --- a/staging/src/k8s.io/client-go/tools/cache/delta_fifo_test.go +++ b/staging/src/k8s.io/client-go/tools/cache/delta_fifo_test.go @@ -17,6 +17,7 @@ limitations under the License. package cache import ( + "context" "fmt" "reflect" "runtime" @@ -86,12 +87,12 @@ func (f *DeltaFIFO) GetByKey(key string) (item interface{}, exists bool, err err // helper function to reduce stuttering func testPop(f *DeltaFIFO) testFifoObject { - return Pop(f).(Deltas).Newest().Object.(testFifoObject) + return Pop(context.Background(), f).(Deltas).Newest().Object.(testFifoObject) } // testPopIfAvailable returns `{}, false` if Pop returns a nil object func testPopIfAvailable(f *DeltaFIFO) (testFifoObject, bool) { - obj := Pop(f) + obj := Pop(context.Background(), f) if obj == nil { return testFifoObject{}, false } @@ -179,7 +180,7 @@ func TestDeltaFIFO_replaceWithDeleteDeltaIn(t *testing.T) { f.Delete(oldObj) f.Replace([]interface{}{newObj}, "") - actualDeltas := Pop(f) + actualDeltas := Pop(context.Background(), f) expectedDeltas := Deltas{ Delta{Type: Deleted, Object: oldObj}, Delta{Type: Sync, Object: newObj}, @@ -289,7 +290,7 @@ func TestDeltaFIFOW_ReplaceMakesDeletionsForObjectsOnlyInQueue(t *testing.T) { }), }) tt.operations(fWithKnownObjects) - actualDeltasWithKnownObjects := Pop(fWithKnownObjects) + actualDeltasWithKnownObjects := Pop(context.Background(), fWithKnownObjects) if !reflect.DeepEqual(tt.expectedDeltas, actualDeltasWithKnownObjects) { t.Errorf("expected %#v, got %#v", tt.expectedDeltas, actualDeltasWithKnownObjects) } @@ -302,7 +303,7 @@ func TestDeltaFIFOW_ReplaceMakesDeletionsForObjectsOnlyInQueue(t *testing.T) { KeyFunction: testFifoObjectKeyFunc, }) tt.operations(fWithoutKnownObjects) - actualDeltasWithoutKnownObjects := Pop(fWithoutKnownObjects) + actualDeltasWithoutKnownObjects := Pop(context.Background(), fWithoutKnownObjects) if !reflect.DeepEqual(tt.expectedDeltas, actualDeltasWithoutKnownObjects) { t.Errorf("expected %#v, got %#v", tt.expectedDeltas, actualDeltasWithoutKnownObjects) } @@ -402,7 +403,7 @@ func TestDeltaFIFO_transformer(t *testing.T) { } for i := 0; i < 2; i++ { - obj, err := f.Pop(func(o interface{}, isInInitialList bool) error { return nil }) + obj, err := f.Pop(context.Background(), func(o interface{}, isInInitialList bool) error { return nil }) if err != nil { t.Fatalf("got nothing on try %v?", i) } @@ -592,7 +593,7 @@ func TestDeltaFIFO_ReplaceMakesDeletions(t *testing.T) { } for _, expected := range expectedList { - cur := Pop(f).(Deltas) + cur := Pop(context.Background(), f).(Deltas) if e, a := expected, cur; !reflect.DeepEqual(e, a) { t.Errorf("Expected %#v, got %#v", e, a) } @@ -618,7 +619,7 @@ func TestDeltaFIFO_ReplaceMakesDeletions(t *testing.T) { } for _, expected := range expectedList { - cur := Pop(f).(Deltas) + cur := Pop(context.Background(), f).(Deltas) if e, a := expected, cur; !reflect.DeepEqual(e, a) { t.Errorf("Expected %#v, got %#v", e, a) } @@ -648,7 +649,7 @@ func TestDeltaFIFO_ReplaceMakesDeletions(t *testing.T) { } for _, expected := range expectedList { - cur := Pop(f).(Deltas) + cur := Pop(context.Background(), f).(Deltas) if e, a := expected, cur; !reflect.DeepEqual(e, a) { t.Errorf("Expected %#v, got %#v", e, a) } @@ -679,7 +680,7 @@ func TestDeltaFIFO_ReplaceMakesDeletions(t *testing.T) { } for _, expected := range expectedList { - cur := Pop(f).(Deltas) + cur := Pop(context.Background(), f).(Deltas) if e, a := expected, cur; !reflect.DeepEqual(e, a) { t.Errorf("Expected %#v, got %#v", e, a) } @@ -697,7 +698,7 @@ func TestDeltaFIFO_ReplaceMakesDeletions(t *testing.T) { } for _, expected := range expectedList { - cur := Pop(f).(Deltas) + cur := Pop(context.Background(), f).(Deltas) if e, a := expected, cur; !reflect.DeepEqual(e, a) { t.Errorf("Expected %#v, got %#v", e, a) } @@ -727,7 +728,7 @@ func TestDeltaFIFO_ReplaceMakesDeletionsReplaced(t *testing.T) { } for _, expected := range expectedList { - cur := Pop(f).(Deltas) + cur := Pop(context.Background(), f).(Deltas) if e, a := expected, cur; !reflect.DeepEqual(e, a) { t.Errorf("Expected %#v, got %#v", e, a) } @@ -751,7 +752,7 @@ func TestDeltaFIFO_ReplaceDeltaType(t *testing.T) { } for _, expected := range expectedList { - cur := Pop(f).(Deltas) + cur := Pop(context.Background(), f).(Deltas) if e, a := expected, cur; !reflect.DeepEqual(e, a) { t.Errorf("Expected %#v, got %#v", e, a) } @@ -773,18 +774,18 @@ func TestDeltaFIFO_UpdateResyncRace(t *testing.T) { } for _, expected := range expectedList { - cur := Pop(f).(Deltas) + cur := Pop(context.Background(), f).(Deltas) if e, a := expected, cur; !reflect.DeepEqual(e, a) { t.Errorf("Expected %#v, got %#v", e, a) } } } -// pop2 captures both parameters, unlike Pop(). +// pop2 captures both parameters, unlike Pop(context.Background(), ). func pop2[T any](queue Queue) (T, bool) { var result interface{} var isList bool - queue.Pop(func(obj interface{}, isInInitialList bool) error { + queue.Pop(context.Background(), func(obj interface{}, isInInitialList bool) error { result = obj isList = isInInitialList return nil @@ -909,15 +910,15 @@ func TestDeltaFIFO_HasSynced(t *testing.T) { { actions: []func(f *DeltaFIFO){ func(f *DeltaFIFO) { f.Replace([]interface{}{mkFifoObj("a", 1), mkFifoObj("b", 2)}, "0") }, - func(f *DeltaFIFO) { Pop(f) }, + func(f *DeltaFIFO) { Pop(context.Background(), f) }, }, expectedSynced: false, }, { actions: []func(f *DeltaFIFO){ func(f *DeltaFIFO) { f.Replace([]interface{}{mkFifoObj("a", 1), mkFifoObj("b", 2)}, "0") }, - func(f *DeltaFIFO) { Pop(f) }, - func(f *DeltaFIFO) { Pop(f) }, + func(f *DeltaFIFO) { Pop(context.Background(), f) }, + func(f *DeltaFIFO) { Pop(context.Background(), f) }, }, expectedSynced: true, }, @@ -926,7 +927,7 @@ func TestDeltaFIFO_HasSynced(t *testing.T) { // there cannot be duplicate keys in the list or apiserver is broken. actions: []func(f *DeltaFIFO){ func(f *DeltaFIFO) { f.Replace([]interface{}{mkFifoObj("a", 1), mkFifoObj("a", 2)}, "0") }, - func(f *DeltaFIFO) { Pop(f) }, + func(f *DeltaFIFO) { Pop(context.Background(), f) }, }, expectedSynced: true, }, @@ -958,7 +959,7 @@ func TestDeltaFIFO_PopShouldUnblockWhenClosed(t *testing.T) { const jobs = 10 for i := 0; i < jobs; i++ { go func() { - f.Pop(func(obj interface{}, isInInitialList bool) error { + f.Pop(context.Background(), func(obj interface{}, isInInitialList bool) error { return nil }) c <- struct{}{} 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..327f9f2d226 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 @@ -17,6 +17,7 @@ limitations under the License. package cache import ( + "context" "reflect" "runtime" "testing" @@ -97,7 +98,7 @@ func TestFIFO_basic(t *testing.T) { lastInt := int(0) lastUint := uint64(0) for i := 0; i < amount*2; i++ { - switch obj := Pop(f).(testFifoObject).val.(type) { + switch obj := Pop(context.Background(), f).(testFifoObject).val.(type) { case int: if obj <= lastInt { t.Errorf("got %v (int) out of order, last was %v", obj, lastInt) @@ -131,7 +132,7 @@ func TestFIFO_addUpdate(t *testing.T) { got := make(chan testFifoObject, 2) go func() { for { - obj := Pop(f) + obj := Pop(context.Background(), f) if obj == nil { return } @@ -162,7 +163,7 @@ func TestFIFO_addReplace(t *testing.T) { got := make(chan testFifoObject, 2) go func() { for { - obj := Pop(f) + obj := Pop(context.Background(), f) if obj == nil { return } @@ -194,21 +195,21 @@ func TestFIFO_detectLineJumpers(t *testing.T) { f.Add(mkFifoObj("foo", 13)) f.Add(mkFifoObj("zab", 30)) - if e, a := 13, Pop(f).(testFifoObject).val; a != e { + if e, a := 13, Pop(context.Background(), f).(testFifoObject).val; a != e { t.Fatalf("expected %d, got %d", e, a) } f.Add(mkFifoObj("foo", 14)) // ensure foo doesn't jump back in line - if e, a := 1, Pop(f).(testFifoObject).val; a != e { + if e, a := 1, Pop(context.Background(), f).(testFifoObject).val; a != e { t.Fatalf("expected %d, got %d", e, a) } - if e, a := 30, Pop(f).(testFifoObject).val; a != e { + if e, a := 30, Pop(context.Background(), f).(testFifoObject).val; a != e { t.Fatalf("expected %d, got %d", e, a) } - if e, a := 14, Pop(f).(testFifoObject).val; a != e { + if e, a := 14, Pop(context.Background(), f).(testFifoObject).val; a != e { t.Fatalf("expected %d, got %d", e, a) } } @@ -243,15 +244,15 @@ func TestFIFO_HasSynced(t *testing.T) { { actions: []func(f *FIFO){ func(f *FIFO) { f.Replace([]interface{}{mkFifoObj("a", 1), mkFifoObj("b", 2)}, "0") }, - func(f *FIFO) { Pop(f) }, + func(f *FIFO) { Pop(context.Background(), f) }, }, expectedSynced: false, }, { actions: []func(f *FIFO){ func(f *FIFO) { f.Replace([]interface{}{mkFifoObj("a", 1), mkFifoObj("b", 2)}, "0") }, - func(f *FIFO) { Pop(f) }, - func(f *FIFO) { Pop(f) }, + func(f *FIFO) { Pop(context.Background(), f) }, + func(f *FIFO) { Pop(context.Background(), f) }, }, expectedSynced: true, }, @@ -278,7 +279,7 @@ func TestFIFO_PopShouldUnblockWhenClosed(t *testing.T) { const jobs = 10 for i := 0; i < jobs; i++ { go func() { - f.Pop(func(obj interface{}, isInInitialList bool) error { + f.Pop(context.Background(), func(obj interface{}, isInInitialList bool) error { return nil }) c <- struct{}{} diff --git a/staging/src/k8s.io/client-go/tools/cache/reflector_test.go b/staging/src/k8s.io/client-go/tools/cache/reflector_test.go index c1c3fef0a30..0fe2ff2bd4f 100644 --- a/staging/src/k8s.io/client-go/tools/cache/reflector_test.go +++ b/staging/src/k8s.io/client-go/tools/cache/reflector_test.go @@ -600,7 +600,7 @@ func TestReflectorListAndWatch(t *testing.T) { // Verify we received the right objects with the right resource versions. for _, expectedObj := range tc.expectedStore { - storeObj := Pop(s).(metav1.Object) + storeObj := Pop(context.Background(), s).(metav1.Object) assert.Equal(t, expectedObj.GetName(), storeObj.GetName()) assert.Equal(t, expectedObj.GetResourceVersion(), storeObj.GetResourceVersion()) } 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 649ea368723..7aa16863cc9 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 @@ -17,6 +17,7 @@ limitations under the License. package cache import ( + "context" "fmt" "reflect" "runtime" @@ -38,7 +39,7 @@ const closedFIFOName = "FIFO WAS CLOSED" func popN(queue Queue, count int) []interface{} { result := []interface{}{} for i := 0; i < count; i++ { - queue.Pop(func(obj interface{}, isInInitialList bool) error { + queue.Pop(context.Background(), func(obj interface{}, isInInitialList bool) error { result = append(result, obj) return nil }) @@ -48,7 +49,7 @@ func popN(queue Queue, count int) []interface{} { // helper function to reduce stuttering func testRealFIFOPop(f *RealFIFO) testFifoObject { - val := Pop(f) + val := Pop(context.Background(), f) if val == nil { return testFifoObject{name: closedFIFOName} } @@ -404,7 +405,7 @@ func TestRealFIFO_transformer(t *testing.T) { } for i := 0; i < len(expected1); i++ { - obj, err := f.Pop(func(o interface{}, isInInitialList bool) error { return nil }) + obj, err := f.Pop(context.Background(), func(o interface{}, isInInitialList bool) error { return nil }) if err != nil { t.Fatalf("got nothing on try %v?", i) } @@ -593,7 +594,7 @@ func TestRealFIFO_ReplaceMakesDeletions(t *testing.T) { } for _, expected := range expectedList { - cur := Pop(f).(Deltas) + cur := Pop(context.Background(), f).(Deltas) if e, a := expected, cur; !reflect.DeepEqual(e, a) { t.Errorf("Expected %#v, got %#v", e, a) } @@ -622,7 +623,7 @@ func TestRealFIFO_ReplaceMakesDeletions(t *testing.T) { } for _, expected := range expectedList { - cur := Pop(f).(Deltas) + cur := Pop(context.Background(), f).(Deltas) if e, a := expected, cur; !reflect.DeepEqual(e, a) { t.Errorf("Expected %#v, got %#v", e, a) } @@ -653,7 +654,7 @@ func TestRealFIFO_ReplaceMakesDeletions(t *testing.T) { } for _, expected := range expectedList { - cur := Pop(f).(Deltas) + cur := Pop(context.Background(), f).(Deltas) if e, a := expected, cur; !reflect.DeepEqual(e, a) { t.Errorf("Expected %#v, got %#v", e, a) } @@ -685,7 +686,7 @@ func TestRealFIFO_ReplaceMakesDeletions(t *testing.T) { } for i, expected := range expectedList { - cur := Pop(f).(Deltas) + cur := Pop(context.Background(), f).(Deltas) if e, a := expected, cur; !reflect.DeepEqual(e, a) { t.Errorf("%d Expected %#v, got %#v", i, e, a) } @@ -707,7 +708,7 @@ func TestRealFIFO_ReplaceMakesDeletions(t *testing.T) { } for _, expected := range expectedList { - cur := Pop(f).(Deltas) + cur := Pop(context.Background(), f).(Deltas) if e, a := expected, cur; !reflect.DeepEqual(e, a) { t.Errorf("Expected %#v, got %#v", e, a) } @@ -738,7 +739,7 @@ func TestRealFIFO_ReplaceMakesDeletionsReplaced(t *testing.T) { } for _, expected := range expectedList { - cur := Pop(f).(Deltas) + cur := Pop(context.Background(), f).(Deltas) if e, a := expected, cur; !reflect.DeepEqual(e, a) { t.Errorf("Expected %#v, got %#v", e, a) } @@ -764,7 +765,7 @@ func TestRealFIFO_UpdateResyncRace(t *testing.T) { } for _, expected := range expectedList { - cur := Pop(f).(Deltas) + cur := Pop(context.Background(), f).(Deltas) if e, a := expected, cur; !reflect.DeepEqual(e, a) { t.Errorf("Expected %#v, got %#v", e, a) } @@ -900,15 +901,15 @@ func TestRealFIFO_HasSynced(t *testing.T) { { actions: []func(f *RealFIFO){ func(f *RealFIFO) { f.Replace([]interface{}{mkFifoObj("a", 1), mkFifoObj("b", 2)}, "0") }, - func(f *RealFIFO) { Pop(f) }, + func(f *RealFIFO) { Pop(context.Background(), f) }, }, expectedSynced: false, }, { actions: []func(f *RealFIFO){ func(f *RealFIFO) { f.Replace([]interface{}{mkFifoObj("a", 1), mkFifoObj("b", 2)}, "0") }, - func(f *RealFIFO) { Pop(f) }, - func(f *RealFIFO) { Pop(f) }, + func(f *RealFIFO) { Pop(context.Background(), f) }, + func(f *RealFIFO) { Pop(context.Background(), f) }, }, expectedSynced: true, }, @@ -917,9 +918,9 @@ func TestRealFIFO_HasSynced(t *testing.T) { // there cannot be duplicate keys in the list or apiserver is broken. actions: []func(f *RealFIFO){ func(f *RealFIFO) { f.Replace([]interface{}{mkFifoObj("a", 1), mkFifoObj("a", 2)}, "0") }, - func(f *RealFIFO) { Pop(f) }, + func(f *RealFIFO) { Pop(context.Background(), f) }, // ATTENTION: difference with delta_fifo_test, every event is delivered, so a is listed twice and must be popped twice to remove both - func(f *RealFIFO) { Pop(f) }, + func(f *RealFIFO) { Pop(context.Background(), f) }, }, expectedSynced: true, }, @@ -956,7 +957,7 @@ func TestRealFIFO_PopShouldUnblockWhenClosed(t *testing.T) { const jobs = 10 for i := 0; i < jobs; i++ { go func() { - f.Pop(func(obj interface{}, isInInitialList bool) error { + f.Pop(context.Background(), func(obj interface{}, isInInitialList bool) error { return nil }) c <- struct{}{} From 2dcce93336e52eb16aab595fcc75254f4abfa5ae Mon Sep 17 00:00:00 2001 From: Keisuke Ishigami Date: Thu, 26 Jun 2025 23:49:25 +0900 Subject: [PATCH 4/6] Revert "modify tests" This reverts commit 2cd5dbbdaab98020b37df7125f622db7a6dd885a. --- pkg/scheduler/schedule_one_test.go | 2 +- .../client-go/tools/cache/delta_fifo_test.go | 43 +++++++++---------- .../k8s.io/client-go/tools/cache/fifo_test.go | 23 +++++----- .../client-go/tools/cache/reflector_test.go | 2 +- .../tools/cache/the_real_fifo_test.go | 33 +++++++------- 5 files changed, 50 insertions(+), 53 deletions(-) diff --git a/pkg/scheduler/schedule_one_test.go b/pkg/scheduler/schedule_one_test.go index 4629878122c..689bddbaac4 100644 --- a/pkg/scheduler/schedule_one_test.go +++ b/pkg/scheduler/schedule_one_test.go @@ -4199,7 +4199,7 @@ func setupTestScheduler(ctx context.Context, t *testing.T, queuedPodStore *clien nodeInfoSnapshot: snapshot, percentageOfNodesToScore: schedulerapi.DefaultPercentageOfNodesToScore, NextPod: func(logger klog.Logger) (*framework.QueuedPodInfo, error) { - return &framework.QueuedPodInfo{PodInfo: mustNewPodInfo(t, clientcache.Pop(context.Background(), queuedPodStore).(*v1.Pod))}, nil + return &framework.QueuedPodInfo{PodInfo: mustNewPodInfo(t, clientcache.Pop(queuedPodStore).(*v1.Pod))}, nil }, SchedulingQueue: schedulingQueue, Profiles: profile.Map{testSchedulerName: fwk}, diff --git a/staging/src/k8s.io/client-go/tools/cache/delta_fifo_test.go b/staging/src/k8s.io/client-go/tools/cache/delta_fifo_test.go index b3e80897ad0..9c67012cc1c 100644 --- a/staging/src/k8s.io/client-go/tools/cache/delta_fifo_test.go +++ b/staging/src/k8s.io/client-go/tools/cache/delta_fifo_test.go @@ -17,7 +17,6 @@ limitations under the License. package cache import ( - "context" "fmt" "reflect" "runtime" @@ -87,12 +86,12 @@ func (f *DeltaFIFO) GetByKey(key string) (item interface{}, exists bool, err err // helper function to reduce stuttering func testPop(f *DeltaFIFO) testFifoObject { - return Pop(context.Background(), f).(Deltas).Newest().Object.(testFifoObject) + return Pop(f).(Deltas).Newest().Object.(testFifoObject) } // testPopIfAvailable returns `{}, false` if Pop returns a nil object func testPopIfAvailable(f *DeltaFIFO) (testFifoObject, bool) { - obj := Pop(context.Background(), f) + obj := Pop(f) if obj == nil { return testFifoObject{}, false } @@ -180,7 +179,7 @@ func TestDeltaFIFO_replaceWithDeleteDeltaIn(t *testing.T) { f.Delete(oldObj) f.Replace([]interface{}{newObj}, "") - actualDeltas := Pop(context.Background(), f) + actualDeltas := Pop(f) expectedDeltas := Deltas{ Delta{Type: Deleted, Object: oldObj}, Delta{Type: Sync, Object: newObj}, @@ -290,7 +289,7 @@ func TestDeltaFIFOW_ReplaceMakesDeletionsForObjectsOnlyInQueue(t *testing.T) { }), }) tt.operations(fWithKnownObjects) - actualDeltasWithKnownObjects := Pop(context.Background(), fWithKnownObjects) + actualDeltasWithKnownObjects := Pop(fWithKnownObjects) if !reflect.DeepEqual(tt.expectedDeltas, actualDeltasWithKnownObjects) { t.Errorf("expected %#v, got %#v", tt.expectedDeltas, actualDeltasWithKnownObjects) } @@ -303,7 +302,7 @@ func TestDeltaFIFOW_ReplaceMakesDeletionsForObjectsOnlyInQueue(t *testing.T) { KeyFunction: testFifoObjectKeyFunc, }) tt.operations(fWithoutKnownObjects) - actualDeltasWithoutKnownObjects := Pop(context.Background(), fWithoutKnownObjects) + actualDeltasWithoutKnownObjects := Pop(fWithoutKnownObjects) if !reflect.DeepEqual(tt.expectedDeltas, actualDeltasWithoutKnownObjects) { t.Errorf("expected %#v, got %#v", tt.expectedDeltas, actualDeltasWithoutKnownObjects) } @@ -403,7 +402,7 @@ func TestDeltaFIFO_transformer(t *testing.T) { } for i := 0; i < 2; i++ { - obj, err := f.Pop(context.Background(), func(o interface{}, isInInitialList bool) error { return nil }) + obj, err := f.Pop(func(o interface{}, isInInitialList bool) error { return nil }) if err != nil { t.Fatalf("got nothing on try %v?", i) } @@ -593,7 +592,7 @@ func TestDeltaFIFO_ReplaceMakesDeletions(t *testing.T) { } for _, expected := range expectedList { - cur := Pop(context.Background(), f).(Deltas) + cur := Pop(f).(Deltas) if e, a := expected, cur; !reflect.DeepEqual(e, a) { t.Errorf("Expected %#v, got %#v", e, a) } @@ -619,7 +618,7 @@ func TestDeltaFIFO_ReplaceMakesDeletions(t *testing.T) { } for _, expected := range expectedList { - cur := Pop(context.Background(), f).(Deltas) + cur := Pop(f).(Deltas) if e, a := expected, cur; !reflect.DeepEqual(e, a) { t.Errorf("Expected %#v, got %#v", e, a) } @@ -649,7 +648,7 @@ func TestDeltaFIFO_ReplaceMakesDeletions(t *testing.T) { } for _, expected := range expectedList { - cur := Pop(context.Background(), f).(Deltas) + cur := Pop(f).(Deltas) if e, a := expected, cur; !reflect.DeepEqual(e, a) { t.Errorf("Expected %#v, got %#v", e, a) } @@ -680,7 +679,7 @@ func TestDeltaFIFO_ReplaceMakesDeletions(t *testing.T) { } for _, expected := range expectedList { - cur := Pop(context.Background(), f).(Deltas) + cur := Pop(f).(Deltas) if e, a := expected, cur; !reflect.DeepEqual(e, a) { t.Errorf("Expected %#v, got %#v", e, a) } @@ -698,7 +697,7 @@ func TestDeltaFIFO_ReplaceMakesDeletions(t *testing.T) { } for _, expected := range expectedList { - cur := Pop(context.Background(), f).(Deltas) + cur := Pop(f).(Deltas) if e, a := expected, cur; !reflect.DeepEqual(e, a) { t.Errorf("Expected %#v, got %#v", e, a) } @@ -728,7 +727,7 @@ func TestDeltaFIFO_ReplaceMakesDeletionsReplaced(t *testing.T) { } for _, expected := range expectedList { - cur := Pop(context.Background(), f).(Deltas) + cur := Pop(f).(Deltas) if e, a := expected, cur; !reflect.DeepEqual(e, a) { t.Errorf("Expected %#v, got %#v", e, a) } @@ -752,7 +751,7 @@ func TestDeltaFIFO_ReplaceDeltaType(t *testing.T) { } for _, expected := range expectedList { - cur := Pop(context.Background(), f).(Deltas) + cur := Pop(f).(Deltas) if e, a := expected, cur; !reflect.DeepEqual(e, a) { t.Errorf("Expected %#v, got %#v", e, a) } @@ -774,18 +773,18 @@ func TestDeltaFIFO_UpdateResyncRace(t *testing.T) { } for _, expected := range expectedList { - cur := Pop(context.Background(), f).(Deltas) + cur := Pop(f).(Deltas) if e, a := expected, cur; !reflect.DeepEqual(e, a) { t.Errorf("Expected %#v, got %#v", e, a) } } } -// pop2 captures both parameters, unlike Pop(context.Background(), ). +// pop2 captures both parameters, unlike Pop(). func pop2[T any](queue Queue) (T, bool) { var result interface{} var isList bool - queue.Pop(context.Background(), func(obj interface{}, isInInitialList bool) error { + queue.Pop(func(obj interface{}, isInInitialList bool) error { result = obj isList = isInInitialList return nil @@ -910,15 +909,15 @@ func TestDeltaFIFO_HasSynced(t *testing.T) { { actions: []func(f *DeltaFIFO){ func(f *DeltaFIFO) { f.Replace([]interface{}{mkFifoObj("a", 1), mkFifoObj("b", 2)}, "0") }, - func(f *DeltaFIFO) { Pop(context.Background(), f) }, + func(f *DeltaFIFO) { Pop(f) }, }, expectedSynced: false, }, { actions: []func(f *DeltaFIFO){ func(f *DeltaFIFO) { f.Replace([]interface{}{mkFifoObj("a", 1), mkFifoObj("b", 2)}, "0") }, - func(f *DeltaFIFO) { Pop(context.Background(), f) }, - func(f *DeltaFIFO) { Pop(context.Background(), f) }, + func(f *DeltaFIFO) { Pop(f) }, + func(f *DeltaFIFO) { Pop(f) }, }, expectedSynced: true, }, @@ -927,7 +926,7 @@ func TestDeltaFIFO_HasSynced(t *testing.T) { // there cannot be duplicate keys in the list or apiserver is broken. actions: []func(f *DeltaFIFO){ func(f *DeltaFIFO) { f.Replace([]interface{}{mkFifoObj("a", 1), mkFifoObj("a", 2)}, "0") }, - func(f *DeltaFIFO) { Pop(context.Background(), f) }, + func(f *DeltaFIFO) { Pop(f) }, }, expectedSynced: true, }, @@ -959,7 +958,7 @@ func TestDeltaFIFO_PopShouldUnblockWhenClosed(t *testing.T) { const jobs = 10 for i := 0; i < jobs; i++ { go func() { - f.Pop(context.Background(), func(obj interface{}, isInInitialList bool) error { + f.Pop(func(obj interface{}, isInInitialList bool) error { return nil }) c <- struct{}{} 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 327f9f2d226..39c27c6d9c5 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 @@ -17,7 +17,6 @@ limitations under the License. package cache import ( - "context" "reflect" "runtime" "testing" @@ -98,7 +97,7 @@ func TestFIFO_basic(t *testing.T) { lastInt := int(0) lastUint := uint64(0) for i := 0; i < amount*2; i++ { - switch obj := Pop(context.Background(), f).(testFifoObject).val.(type) { + switch obj := Pop(f).(testFifoObject).val.(type) { case int: if obj <= lastInt { t.Errorf("got %v (int) out of order, last was %v", obj, lastInt) @@ -132,7 +131,7 @@ func TestFIFO_addUpdate(t *testing.T) { got := make(chan testFifoObject, 2) go func() { for { - obj := Pop(context.Background(), f) + obj := Pop(f) if obj == nil { return } @@ -163,7 +162,7 @@ func TestFIFO_addReplace(t *testing.T) { got := make(chan testFifoObject, 2) go func() { for { - obj := Pop(context.Background(), f) + obj := Pop(f) if obj == nil { return } @@ -195,21 +194,21 @@ func TestFIFO_detectLineJumpers(t *testing.T) { f.Add(mkFifoObj("foo", 13)) f.Add(mkFifoObj("zab", 30)) - if e, a := 13, Pop(context.Background(), f).(testFifoObject).val; a != e { + if e, a := 13, Pop(f).(testFifoObject).val; a != e { t.Fatalf("expected %d, got %d", e, a) } f.Add(mkFifoObj("foo", 14)) // ensure foo doesn't jump back in line - if e, a := 1, Pop(context.Background(), f).(testFifoObject).val; a != e { + if e, a := 1, Pop(f).(testFifoObject).val; a != e { t.Fatalf("expected %d, got %d", e, a) } - if e, a := 30, Pop(context.Background(), f).(testFifoObject).val; a != e { + if e, a := 30, Pop(f).(testFifoObject).val; a != e { t.Fatalf("expected %d, got %d", e, a) } - if e, a := 14, Pop(context.Background(), f).(testFifoObject).val; a != e { + if e, a := 14, Pop(f).(testFifoObject).val; a != e { t.Fatalf("expected %d, got %d", e, a) } } @@ -244,15 +243,15 @@ func TestFIFO_HasSynced(t *testing.T) { { actions: []func(f *FIFO){ func(f *FIFO) { f.Replace([]interface{}{mkFifoObj("a", 1), mkFifoObj("b", 2)}, "0") }, - func(f *FIFO) { Pop(context.Background(), f) }, + func(f *FIFO) { Pop(f) }, }, expectedSynced: false, }, { actions: []func(f *FIFO){ func(f *FIFO) { f.Replace([]interface{}{mkFifoObj("a", 1), mkFifoObj("b", 2)}, "0") }, - func(f *FIFO) { Pop(context.Background(), f) }, - func(f *FIFO) { Pop(context.Background(), f) }, + func(f *FIFO) { Pop(f) }, + func(f *FIFO) { Pop(f) }, }, expectedSynced: true, }, @@ -279,7 +278,7 @@ func TestFIFO_PopShouldUnblockWhenClosed(t *testing.T) { const jobs = 10 for i := 0; i < jobs; i++ { go func() { - f.Pop(context.Background(), func(obj interface{}, isInInitialList bool) error { + f.Pop(func(obj interface{}, isInInitialList bool) error { return nil }) c <- struct{}{} diff --git a/staging/src/k8s.io/client-go/tools/cache/reflector_test.go b/staging/src/k8s.io/client-go/tools/cache/reflector_test.go index 0fe2ff2bd4f..c1c3fef0a30 100644 --- a/staging/src/k8s.io/client-go/tools/cache/reflector_test.go +++ b/staging/src/k8s.io/client-go/tools/cache/reflector_test.go @@ -600,7 +600,7 @@ func TestReflectorListAndWatch(t *testing.T) { // Verify we received the right objects with the right resource versions. for _, expectedObj := range tc.expectedStore { - storeObj := Pop(context.Background(), s).(metav1.Object) + storeObj := Pop(s).(metav1.Object) assert.Equal(t, expectedObj.GetName(), storeObj.GetName()) assert.Equal(t, expectedObj.GetResourceVersion(), storeObj.GetResourceVersion()) } 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 7aa16863cc9..649ea368723 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 @@ -17,7 +17,6 @@ limitations under the License. package cache import ( - "context" "fmt" "reflect" "runtime" @@ -39,7 +38,7 @@ const closedFIFOName = "FIFO WAS CLOSED" func popN(queue Queue, count int) []interface{} { result := []interface{}{} for i := 0; i < count; i++ { - queue.Pop(context.Background(), func(obj interface{}, isInInitialList bool) error { + queue.Pop(func(obj interface{}, isInInitialList bool) error { result = append(result, obj) return nil }) @@ -49,7 +48,7 @@ func popN(queue Queue, count int) []interface{} { // helper function to reduce stuttering func testRealFIFOPop(f *RealFIFO) testFifoObject { - val := Pop(context.Background(), f) + val := Pop(f) if val == nil { return testFifoObject{name: closedFIFOName} } @@ -405,7 +404,7 @@ func TestRealFIFO_transformer(t *testing.T) { } for i := 0; i < len(expected1); i++ { - obj, err := f.Pop(context.Background(), func(o interface{}, isInInitialList bool) error { return nil }) + obj, err := f.Pop(func(o interface{}, isInInitialList bool) error { return nil }) if err != nil { t.Fatalf("got nothing on try %v?", i) } @@ -594,7 +593,7 @@ func TestRealFIFO_ReplaceMakesDeletions(t *testing.T) { } for _, expected := range expectedList { - cur := Pop(context.Background(), f).(Deltas) + cur := Pop(f).(Deltas) if e, a := expected, cur; !reflect.DeepEqual(e, a) { t.Errorf("Expected %#v, got %#v", e, a) } @@ -623,7 +622,7 @@ func TestRealFIFO_ReplaceMakesDeletions(t *testing.T) { } for _, expected := range expectedList { - cur := Pop(context.Background(), f).(Deltas) + cur := Pop(f).(Deltas) if e, a := expected, cur; !reflect.DeepEqual(e, a) { t.Errorf("Expected %#v, got %#v", e, a) } @@ -654,7 +653,7 @@ func TestRealFIFO_ReplaceMakesDeletions(t *testing.T) { } for _, expected := range expectedList { - cur := Pop(context.Background(), f).(Deltas) + cur := Pop(f).(Deltas) if e, a := expected, cur; !reflect.DeepEqual(e, a) { t.Errorf("Expected %#v, got %#v", e, a) } @@ -686,7 +685,7 @@ func TestRealFIFO_ReplaceMakesDeletions(t *testing.T) { } for i, expected := range expectedList { - cur := Pop(context.Background(), f).(Deltas) + cur := Pop(f).(Deltas) if e, a := expected, cur; !reflect.DeepEqual(e, a) { t.Errorf("%d Expected %#v, got %#v", i, e, a) } @@ -708,7 +707,7 @@ func TestRealFIFO_ReplaceMakesDeletions(t *testing.T) { } for _, expected := range expectedList { - cur := Pop(context.Background(), f).(Deltas) + cur := Pop(f).(Deltas) if e, a := expected, cur; !reflect.DeepEqual(e, a) { t.Errorf("Expected %#v, got %#v", e, a) } @@ -739,7 +738,7 @@ func TestRealFIFO_ReplaceMakesDeletionsReplaced(t *testing.T) { } for _, expected := range expectedList { - cur := Pop(context.Background(), f).(Deltas) + cur := Pop(f).(Deltas) if e, a := expected, cur; !reflect.DeepEqual(e, a) { t.Errorf("Expected %#v, got %#v", e, a) } @@ -765,7 +764,7 @@ func TestRealFIFO_UpdateResyncRace(t *testing.T) { } for _, expected := range expectedList { - cur := Pop(context.Background(), f).(Deltas) + cur := Pop(f).(Deltas) if e, a := expected, cur; !reflect.DeepEqual(e, a) { t.Errorf("Expected %#v, got %#v", e, a) } @@ -901,15 +900,15 @@ func TestRealFIFO_HasSynced(t *testing.T) { { actions: []func(f *RealFIFO){ func(f *RealFIFO) { f.Replace([]interface{}{mkFifoObj("a", 1), mkFifoObj("b", 2)}, "0") }, - func(f *RealFIFO) { Pop(context.Background(), f) }, + func(f *RealFIFO) { Pop(f) }, }, expectedSynced: false, }, { actions: []func(f *RealFIFO){ func(f *RealFIFO) { f.Replace([]interface{}{mkFifoObj("a", 1), mkFifoObj("b", 2)}, "0") }, - func(f *RealFIFO) { Pop(context.Background(), f) }, - func(f *RealFIFO) { Pop(context.Background(), f) }, + func(f *RealFIFO) { Pop(f) }, + func(f *RealFIFO) { Pop(f) }, }, expectedSynced: true, }, @@ -918,9 +917,9 @@ func TestRealFIFO_HasSynced(t *testing.T) { // there cannot be duplicate keys in the list or apiserver is broken. actions: []func(f *RealFIFO){ func(f *RealFIFO) { f.Replace([]interface{}{mkFifoObj("a", 1), mkFifoObj("a", 2)}, "0") }, - func(f *RealFIFO) { Pop(context.Background(), f) }, + func(f *RealFIFO) { Pop(f) }, // ATTENTION: difference with delta_fifo_test, every event is delivered, so a is listed twice and must be popped twice to remove both - func(f *RealFIFO) { Pop(context.Background(), f) }, + func(f *RealFIFO) { Pop(f) }, }, expectedSynced: true, }, @@ -957,7 +956,7 @@ func TestRealFIFO_PopShouldUnblockWhenClosed(t *testing.T) { const jobs = 10 for i := 0; i < jobs; i++ { go func() { - f.Pop(context.Background(), func(obj interface{}, isInInitialList bool) error { + f.Pop(func(obj interface{}, isInInitialList bool) error { return nil }) c <- struct{}{} From 74af3ac8ad1122528bb9971c3a2d282eff529beb Mon Sep 17 00:00:00 2001 From: Keisuke Ishigami Date: Thu, 26 Jun 2025 23:49:34 +0900 Subject: [PATCH 5/6] Revert "pop respects the context" This reverts commit 1c33d98762511c10f89c40358a1935250b03b0c8. --- .../client-go/tools/cache/controller.go | 4 ++-- .../client-go/tools/cache/delta_fifo.go | 8 +------ .../src/k8s.io/client-go/tools/cache/fifo.go | 21 ++++++------------- .../client-go/tools/cache/the_real_fifo.go | 8 +------ 4 files changed, 10 insertions(+), 31 deletions(-) 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 8bafcb67bc0..f0a483b6f5b 100644 --- a/staging/src/k8s.io/client-go/tools/cache/controller.go +++ b/staging/src/k8s.io/client-go/tools/cache/controller.go @@ -208,9 +208,9 @@ func (c *controller) processLoop(ctx context.Context) { case <-ctx.Done(): return default: - _, err := c.config.Queue.Pop(ctx, PopProcessFunc(c.config.Process)) + _, err := c.config.Queue.Pop(PopProcessFunc(c.config.Process)) if err != nil { - if errors.Is(err, ErrFIFOClosed) || errors.Is(err, ErrCtxDone) { + if err == ErrFIFOClosed { return } } diff --git a/staging/src/k8s.io/client-go/tools/cache/delta_fifo.go b/staging/src/k8s.io/client-go/tools/cache/delta_fifo.go index 57872a1e1a8..264d7559a05 100644 --- a/staging/src/k8s.io/client-go/tools/cache/delta_fifo.go +++ b/staging/src/k8s.io/client-go/tools/cache/delta_fifo.go @@ -17,7 +17,6 @@ limitations under the License. package cache import ( - "context" "errors" "fmt" "sync" @@ -491,16 +490,11 @@ func (f *DeltaFIFO) IsClosed() bool { // // Pop returns a 'Deltas', which has a complete list of all the things // that happened to the object (deltas) while it was sitting in the queue. -func (f *DeltaFIFO) Pop(ctx context.Context, process PopProcessFunc) (interface{}, error) { +func (f *DeltaFIFO) Pop(process PopProcessFunc) (interface{}, error) { f.lock.Lock() defer f.lock.Unlock() for { for len(f.queue) == 0 { - // if the length of the queue is 0 and ctx is done, return here - if ctx.Err() != nil { - return nil, ErrCtxDone - } - // When the queue is empty, invocation of Pop() is blocked until new item is enqueued. // When Close() is called, the f.closed is set and the condition is broadcasted. // Which causes this loop to continue and return from the Pop(). 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 fde34edc5b2..5c2ca90084d 100644 --- a/staging/src/k8s.io/client-go/tools/cache/fifo.go +++ b/staging/src/k8s.io/client-go/tools/cache/fifo.go @@ -17,7 +17,6 @@ limitations under the License. package cache import ( - "context" "errors" "sync" @@ -31,9 +30,6 @@ type PopProcessFunc func(obj interface{}, isInInitialList bool) error // ErrFIFOClosed used when FIFO is closed var ErrFIFOClosed = errors.New("DeltaFIFO: manipulating with closed queue") -// ErrCtxDone is used when context is done -var ErrCtxDone = errors.New("DeltaFIFO: context done") - // Queue extends ReflectorStore with a collection of Store keys to "process". // Every Add, Update, or Delete may put the object's key in that collection. // A Queue has a way to derive the corresponding key given an accumulator. @@ -42,8 +38,8 @@ var ErrCtxDone = errors.New("DeltaFIFO: context done") type Queue interface { ReflectorStore - // Pop blocks until there is at least one key to process, ctx is done, - // or the Queue is closed. In the latter case Pop returns with an error. + // Pop blocks until there is at least one key to process or the + // Queue is closed. In the latter case Pop returns with an error. // In the former case Pop atomically picks one key to process, // removes that (key, accumulator) association from the Store, and // processes the accumulator. Pop returns the accumulator that @@ -52,7 +48,7 @@ type Queue interface { // return that (key, accumulator) association to the Queue as part // of the atomic processing and (b) return the inner error from // Pop. - Pop(context.Context, PopProcessFunc) (interface{}, error) + Pop(PopProcessFunc) (interface{}, error) // HasSynced returns true if the first batch of keys have all been // popped. The first batch of keys are those of the first Replace @@ -70,9 +66,9 @@ type Queue interface { // // NOTE: This function is deprecated and may be removed in the future without // additional warning. -func Pop(ctx context.Context, queue Queue) interface{} { +func Pop(queue Queue) interface{} { var result interface{} - queue.Pop(ctx, func(obj interface{}, isInInitialList bool) error { + queue.Pop(func(obj interface{}, isInInitialList bool) error { result = obj return nil }) @@ -195,16 +191,11 @@ func (f *FIFO) IsClosed() bool { // so if you don't successfully process it, it should be added back with // AddIfNotPresent(). process function is called under lock, so it is safe // update data structures in it that need to be in sync with the queue. -func (f *FIFO) Pop(ctx context.Context, process PopProcessFunc) (interface{}, error) { +func (f *FIFO) Pop(process PopProcessFunc) (interface{}, error) { f.lock.Lock() defer f.lock.Unlock() for { for len(f.queue) == 0 { - // if the length of the queue is 0 and ctx is done, return here - if ctx.Err() != nil { - return nil, ErrCtxDone - } - // When the queue is empty, invocation of Pop() is blocked until new item is enqueued. // When Close() is called, the f.closed is set and the condition is broadcasted. // Which causes this loop to continue and return from the Pop(). 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 337e3a8fabe..9be14ff3811 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 @@ -17,7 +17,6 @@ limitations under the License. package cache import ( - "context" "fmt" utilruntime "k8s.io/apimachinery/pkg/util/runtime" "k8s.io/apimachinery/pkg/util/sets" @@ -192,16 +191,11 @@ func (f *RealFIFO) IsClosed() bool { // 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. -func (f *RealFIFO) Pop(ctx context.Context, process PopProcessFunc) (interface{}, error) { +func (f *RealFIFO) Pop(process PopProcessFunc) (interface{}, error) { f.lock.Lock() defer f.lock.Unlock() for len(f.items) == 0 { - // if the length of the queue is 0 and ctx is done, return here - if ctx.Err() != nil { - return nil, ErrCtxDone - } - // When the queue is empty, invocation of Pop() is blocked until new item is enqueued. // When Close() is called, the f.closed is set and the condition is broadcasted. // Which causes this loop to continue and return from the Pop(). From 5cca03792748714547c39330ac9cb8cb9c7c60ae Mon Sep 17 00:00:00 2001 From: Keisuke Ishigami Date: Fri, 27 Jun 2025 00:49:02 +0900 Subject: [PATCH 6/6] resolve linter check --- staging/src/k8s.io/client-go/tools/cache/controller.go | 4 ++-- 1 file changed, 2 insertions(+), 2 deletions(-) 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 f0a483b6f5b..5f983b6b63a 100644 --- a/staging/src/k8s.io/client-go/tools/cache/controller.go +++ b/staging/src/k8s.io/client-go/tools/cache/controller.go @@ -208,9 +208,9 @@ func (c *controller) processLoop(ctx context.Context) { case <-ctx.Done(): return default: - _, err := c.config.Queue.Pop(PopProcessFunc(c.config.Process)) + _, err := c.config.Pop(PopProcessFunc(c.config.Process)) if err != nil { - if err == ErrFIFOClosed { + if errors.Is(err, ErrFIFOClosed) { return } }