Repository: kubernetes/kubernetes
File: staging/src/k8s.io/client-go/tools/cache/shared_informer_test.go (Go)

Code before:
	// We must potentially advance time to unblock sharedProcess.run,
	// otherwise synctest.Wait() below returns without anything
	// being done.
	time.Sleep(startupDelay)

	// Ensure all listeners got the initial after the initial processing, at the initial start time.
	//
	// synctest.Wait doesn't detect deadlocks involving mutexes: they are considered not durably
	// blocking, so in case of such a deadlock it'll just keep waiting until the entire unit test
	// times out. The backtraces then show the deadlock.
	synctest.Wait()
	for _, listener := range listeners {
		if !listener.satisfiedExpectations() {
			t.Errorf("%s: %s: expected %v, got %v", time.Since(start), listener.name, listener.expectedItemNames, listener.receivedItemNames)
		}
	}

	// reset
	for _, listener := range listeners {
		listener.receivedItemNames = []string{}
	}

	// advance so listener2 gets a resync
	time.Sleep(2*time.Second - startupDelay)
	synctest.Wait()

	// make sure listener2 got the resync
	if !listener2.satisfiedExpectations() {
		t.Errorf("%s: %s: expected %v, got %v", time.Since(start), listener2.name, listener2.expectedItemNames, listener2.receivedItemNames)
	}

	// make sure listeners 1 and 3 got nothing
	if len(listener1.receivedItemNames) != 0 {
		t.Errorf("%s: listener1: should not have resynced (got %d)", time.Since(start), len(listener1.receivedItemNames))
	}
	if len(listener3.receivedItemNames) != 0 {
		t.Errorf("%s: listener3: should not have resynced (got %d)", time.Since(start), len(listener3.receivedItemNames))
	}

	// reset
	for _, listener := range listeners {
		listener.receivedItemNames = []string{}
	}

	// advance so listener3 gets a resync
	time.Sleep(1 * time.Second)
	synctest.Wait()

	// make sure listener3 got the resync
	if !listener3.satisfiedExpectations() {
		t.Errorf("%s: %s: expected %v, got %v", time.Since(start), listener3.name, listener3.expectedItemNames, listener3.receivedItemNames)

Review comment (bug):
With the `for !p.listenersStarted && len(p.listeners) > 0` commented out:

```console
go test -run=TestListenerResyncPeriods -v -timeout=1s .
=== RUN   TestListenerResyncPeriods
    shared_informer_test.go:236: 0s: listener3: handle: pod1
    shared_informer_test.go:236: 0s: listener1: handle: pod1
    shared_informer_test.go:236: 0s: listener2: handle: pod1
    shared_informer_test.go:236: 0s: listener2: handle: pod2
    shared_informer_test.go:236: 0s: listener1: handle: pod2
    shared_informer_test.go:236: 0s: listener3: handle: pod2
    shared_informer_test.go:236: 2s: listener2: handle: pod1
    shared_informer_test.go:236: 2s: listener2: handle: pod2
    shared_informer_test.go:236: 3s: listener3: handle: pod1
    shared_informer_test.go:236: 3s: listener3: handle: pod2
--- PASS: TestListenerResyncPeriods (0.00s)
=== RUN   TestListenerResyncPeriodsDelayed
panic: test timed out after 1s
	running tests:
		TestListenerResyncPeriodsDelayed (1s)

goroutine 21 [running]:
testing.(*M).startAlarm.func1()
	/nvme/gopath/go-1.25.2/src/testing/testing.go:2682 +0x345
created by time.goFunc
	/nvme/gopath/go-1.25.2/src/time/sleep.go:215 +0x2d

goroutine 1 [chan receive]:
testing.(*T).Run(0xc0000e9180, {0xe23472?, 0xc00019da88?}, 0xe902f0)
	/nvme/gopath/go-1.25.2/src/testing/testing.go:2005 +0x485
testing.runTests.func1(0xc0000e9180)
	/nvme/gopath/go-1.25.2/src/testing/testing.go:2477 +0x37
testing.tRunner(0xc0000e9180, 0xc00019dbc8)
	/nvme/gopath/go-1.25.2/src/testing/testing.go:1934 +0xea
testing.runTests(0xc000012b40, {0x155bb20, 0x9b, 0x9b}, {0xc00024a860?, 0xd31040?, 0x15612a0?})
	/nvme/gopath/go-1.25.2/src/testing/testing.go:2475 +0x4b4
testing.(*M).Run(0xc000230b40)
	/nvme/gopath/go-1.25.2/src/testing/testing.go:2337 +0x63a
go.uber.org/goleak.VerifyTestMain({0xf86620?, 0xc000230b40?}, {0x0, 0x0, 0x0})
	/nvme/gopath/src/k8s.io/kubernetes/vendor/go.uber.org/goleak/testmain.go:53 +0x5a
k8s.io/client-go/tools/cache.TestMain(...)
	/nvme/gopath/src/k8s.io/kubernetes/staging/src/k8s.io/client-go/tools/cache/main_test.go:26
main.main()
	_testmain.go:391 +0xb7

goroutine 50 [synctest.Run (durable), synctest bubble 2]:
internal/synctest.Run(0xc000314120)
	/nvme/gopath/go-1.25.2/src/runtime/synctest.go:218 +0x1e6
testing/synctest.Test(0xc00032e000, 0xe90960)
	/nvme/gopath/go-1.25.2/src/testing/synctest/synctest.go:282 +0x90
k8s.io/client-go/tools/cache.TestListenerResyncPeriodsDelayed(0xc00032e000?)
	/nvme/gopath/src/k8s.io/kubernetes/staging/src/k8s.io/client-go/tools/cache/shared_informer_test.go:229 +0x1a
testing.tRunner(0xc00032e000, 0xe902f0)
	/nvme/gopath/go-1.25.2/src/testing/testing.go:1934 +0xea
created by testing.(*T).Run in goroutine 1
	/nvme/gopath/go-1.25.2/src/testing/testing.go:1997 +0x465

goroutine 51 [chan receive (durable), synctest bubble 2]:
testing/synctest.testingSynctestTest(0xc00032e000, 0xe90960)
	/nvme/gopath/go-1.25.2/src/testing/testing.go:2047 +0x288
testing/synctest.Test.func1()
	/nvme/gopath/go-1.25.2/src/testing/synctest/synctest.go:283 +0x25
created by testing/synctest.Test in goroutine 50
	/nvme/gopath/go-1.25.2/src/testing/synctest/synctest.go:282 +0x90

goroutine 52 [synctest.Wait (durable), synctest bubble 2]:
internal/synctest.Wait()
	/nvme/gopath/go-1.25.2/src/runtime/synctest.go:298 +0x78
testing/synctest.Wait(...)
	/nvme/gopath/go-1.25.2/src/testing/synctest/synctest.go:302
k8s.io/client-go/tools/cache.testListenerResyncPeriods(0xc00032e1c0, 0x3b9aca00)
	/nvme/gopath/src/k8s.io/kubernetes/staging/src/k8s.io/client-go/tools/cache/shared_informer_test.go:282 +0x755
k8s.io/client-go/tools/cache.TestListenerResyncPeriodsDelayed.func1(0xc00032e1c0?)
	/nvme/gopath/src/k8s.io/kubernetes/staging/src/k8s.io/client-go/tools/cache/shared_informer_test.go:229 +0x18
testing.tRunner(0xc00032e1c0, 0xe90960)
	/nvme/gopath/go-1.25.2/src/testing/testing.go:1934 +0xea
created by testing/synctest.testingSynctestTest in goroutine 51
	/nvme/gopath/go-1.25.2/src/testing/testing.go:2046 +0x26d

goroutine 53 [chan receive (durable), synctest bubble 2]:
k8s.io/apimachinery/pkg/watch.(*Broadcaster).loop(0xc0003280a0)
	/nvme/gopath/src/k8s.io/kubernetes/staging/src/k8s.io/apimachinery/pkg/watch/mux.go:268 +0x66
created by k8s.io/apimachinery/pkg/watch.NewBroadcaster in goroutine 52
	/nvme/gopath/src/k8s.io/kubernetes/staging/src/k8s.io/apimachinery/pkg/watch/mux.go:77 +0x125

goroutine 54 [chan send (durable), synctest bubble 2]:
k8s.io/client-go/tools/cache.(*processorListener).add(...)
	/nvme/gopath/src/k8s.io/kubernetes/staging/src/k8s.io/client-go/tools/cache/shared_informer.go:1044
k8s.io/client-go/tools/cache.(*sharedProcessor).distribute(0xd60fe0?, {0xd60fe0, 0xc0003122e8}, 0x0)
	/nvme/gopath/src/k8s.io/kubernetes/staging/src/k8s.io/client-go/tools/cache/shared_informer.go:887 +0x1d0
k8s.io/client-go/tools/cache.(*sharedIndexInformer).OnAdd(0xc00033a000, {0xdfcda0, 0xc000334f08}, 0x1)
	/nvme/gopath/src/k8s.io/kubernetes/staging/src/k8s.io/client-go/tools/cache/shared_informer.go:744 +0x7b
k8s.io/client-go/tools/cache.processReplacedAllInfo({0xf8c7a0, 0xc00033a000}, {{0xe28739, 0x1}, {0xc000314380, 0x2, 0x2}}, {0xf92760, 0xc000314160}, 0x1, ...)
	/nvme/gopath/src/k8s.io/kubernetes/staging/src/k8s.io/client-go/tools/cache/controller.go:762 +0x33c
k8s.io/client-go/tools/cache.processDeltas({0xf8c7a0, 0xc00033a000}, {0xf92760, 0xc000314160}, {0xc0003143e0?, 0x0?, 0x0?}, 0x1, 0xe907e8)
	/nvme/gopath/src/k8s.io/kubernetes/staging/src/k8s.io/client-go/tools/cache/controller.go:599 +0x407
k8s.io/client-go/tools/cache.(*sharedIndexInformer).HandleDeltas(0xc00033a000, {0xd1f160?, 0xc0003122d0?}, 0x1)
	/nvme/gopath/src/k8s.io/kubernetes/staging/src/k8s.io/client-go/tools/cache/shared_informer.go:728 +0x1a7
k8s.io/client-go/tools/cache.(*RealFIFO).PopBatch.func3()
	/nvme/gopath/src/k8s.io/kubernetes/staging/src/k8s.io/client-go/tools/cache/the_real_fifo.go:472 +0xad
k8s.io/client-go/tools/cache.(*RealFIFO).whileProcessing_locked(0x122e200?, 0x10e05d8?)
	/nvme/gopath/src/k8s.io/kubernetes/staging/src/k8s.io/client-go/tools/cache/the_real_fifo.go:369 +0x90
k8s.io/client-go/tools/cache.(*RealFIFO).PopBatch(0xc00031c0b0, 0xc0003002b0, 0xc0003002a0)
	/nvme/gopath/src/k8s.io/kubernetes/staging/src/k8s.io/client-go/tools/cache/the_real_fifo.go:471 +0xc55
k8s.io/client-go/tools/cache.(*controller).processLoop(0xc00031c160, {0xf8df28, 0xc00032a3f0})
	/nvme/gopath/src/k8s.io/kubernetes/staging/src/k8s.io/client-go/tools/cache/controller.go:234 +0x165
k8s.io/apimachinery/pkg/util/wait.BackoffUntilWithContext.func1({0xf8df28?, 0xc00032a3f0?}, 0xc00030a390?)
	/nvme/gopath/src/k8s.io/kubernetes/staging/src/k8s.io/apimachinery/pkg/util/wait/backoff.go:255 +0x51
k8s.io/apimachinery/pkg/util/wait.BackoffUntilWithContext({0xf8df28, 0xc00032a3f0}, 0xc00019fdc0, {0xf87060, 0xc00030a390}, 0x1)
	/nvme/gopath/src/k8s.io/kubernetes/staging/src/k8s.io/apimachinery/pkg/util/wait/backoff.go:256 +0xe5
k8s.io/apimachinery/pkg/util/wait.JitterUntilWithContext({0xf8df28, 0xc00032a3f0}, 0xc0000a4dc0, 0x3b9aca00, 0x0, 0x1)
	/nvme/gopath/src/k8s.io/kubernetes/staging/src/k8s.io/apimachinery/pkg/util/wait/backoff.go:223 +0x8f
k8s.io/apimachinery/pkg/util/wait.UntilWithContext(...)
	/nvme/gopath/src/k8s.io/kubernetes/staging/src/k8s.io/apimachinery/pkg/util/wait/backoff.go:172
k8s.io/client-go/tools/cache.(*controller).RunWithContext(0xc00031c160, {0xf8df28, 0xc00032a3f0})
	/nvme/gopath/src/k8s.io/kubernetes/staging/src/k8s.io/client-go/tools/cache/controller.go:199 +0x41e
k8s.io/client-go/tools/cache.(*sharedIndexInformer).RunWithContext(0xc00033a000, {0xf8df28, 0xc00032a3f0})
	/nvme/gopath/src/k8s.io/kubernetes/staging/src/k8s.io/client-go/tools/cache/shared_informer.go:586 +0x39a
k8s.io/client-go/tools/cache.(*sharedIndexInformer).Run(...)
	/nvme/gopath/src/k8s.io/kubernetes/staging/src/k8s.io/client-go/tools/cache/shared_informer.go:533
k8s.io/client-go/tools/cache.testListenerResyncPeriods.(*Group).StartWithChannel.func3()
	/nvme/gopath/src/k8s.io/kubernetes/staging/src/k8s.io/apimachinery/pkg/util/wait/wait.go:55 +0x1b
k8s.io/apimachinery/pkg/util/wait.(*Group).Start.func1()
	/nvme/gopath/src/k8s.io/kubernetes/staging/src/k8s.io/apimachinery/pkg/util/wait/wait.go:72 +0x4c
created by k8s.io/apimachinery/pkg/util/wait.(*Group).Start in goroutine 52
	/nvme/gopath/src/k8s.io/kubernetes/staging/src/k8s.io/apimachinery/pkg/util/wait/wait.go:70 +0x73

goroutine 58 [select (durable), synctest bubble 2]:
k8s.io/client-go/tools/cache.handleAnyWatch({0xf8df28, 0xc00032a3f0}, {0x0?, 0x0?, 0x15612a0?}, {0xf8a4f8, 0xc00030a5a0}, {0xf8f2c0, 0xc00031c0b0}, {0xf9b160, ...}, ...)
	/nvme/gopath/src/k8s.io/kubernetes/staging/src/k8s.io/client-go/tools/cache/reflector.go:903 +0x2b3
k8s.io/client-go/tools/cache.handleWatch({0xf8df28?, 0xc00032a3f0?}, {0xc000304190?, 0x0?, 0x15612a0?}, {0xf8a4f8?, 0xc00030a5a0?}, {0xf8f2c0?, 0xc00031c0b0?}, {0xf9b160, ...}, ...)
	/nvme/gopath/src/k8s.io/kubernetes/staging/src/k8s.io/client-go/tools/cache/reflector.go:861 +0xea
k8s.io/client-go/tools/cache.(*Reflector).watch(0xc00030e690, {0xf8df28, 0xc00032a3f0}, {0xf8a4f8, 0xc00030a5a0}, 0xc0003163f0)
	/nvme/gopath/src/k8s.io/kubernetes/staging/src/k8s.io/client-go/tools/cache/reflector.go:551 +0x60e
k8s.io/client-go/tools/cache.(*Reflector).watchWithResync(0xc00030e690, {0xf8df28, 0xc00032a3f0}, {0x0, 0x0})
	/nvme/gopath/src/k8s.io/kubernetes/staging/src/k8s.io/client-go/tools/cache/reflector.go:491 +0x186
k8s.io/client-go/tools/cache.(*Reflector).ListAndWatchWithContext(0xc00030e690, {0xf8df28, 0xc00032a3f0})
	/nvme/gopath/src/k8s.io/kubernetes/staging/src/k8s.io/client-go/tools/cache/reflector.go:446 +0x472
k8s.io/client-go/tools/cache.(*Reflector).RunWithContext.func1()
	/nvme/gopath/src/k8s.io/kubernetes/staging/src/k8s.io/client-go/tools/cache/reflector.go:368 +0x2e
k8s.io/apimachinery/pkg/util/wait.BackoffUntil.func1({0x0?, 0x0?})
	/nvme/gopath/src/k8s.io/kubernetes/staging/src/k8s.io/apimachinery/pkg/util/wait/backoff.go:233 +0x13
k8s.io/apimachinery/pkg/util/wait.BackoffUntilWithContext.func1({0xf8df28?, 0xc00032a3f0?}, 0x41e900?)
	/nvme/gopath/src/k8s.io/kubernetes/staging/src/k8s.io/apimachinery/pkg/util/wait/backoff.go:255 +0x51
k8s.io/apimachinery/pkg/util/wait.BackoffUntilWithContext({0xf8df28, 0xc00032a3f0}, 0xc000199ea8, {0xf87080, 0xc000328190}, 0x1)
	/nvme/gopath/src/k8s.io/kubernetes/staging/src/k8s.io/apimachinery/pkg/util/wait/backoff.go:256 +0xe5
k8s.io/apimachinery/pkg/util/wait.BackoffUntil(0xf905f0?, {0xf87080?, 0xc000328190?}, 0x48?, 0x12?)
	/nvme/gopath/src/k8s.io/kubernetes/staging/src/k8s.io/apimachinery/pkg/util/wait/backoff.go:233 +0x46
k8s.io/client-go/tools/cache.(*Reflector).RunWithContext(0xc00030e690, {0xf8df28, 0xc00032a3f0})
	/nvme/gopath/src/k8s.io/kubernetes/staging/src/k8s.io/client-go/tools/cache/reflector.go:367 +0x22b
k8s.io/client-go/tools/cache.(*controller).RunWithContext.(*Group).StartWithContext.func3()
	/nvme/gopath/src/k8s.io/kubernetes/staging/src/k8s.io/apimachinery/pkg/util/wait/wait.go:63 +0x1f
k8s.io/apimachinery/pkg/util/wait.(*Group).Start.func1()
	/nvme/gopath/src/k8s.io/kubernetes/staging/src/k8s.io/apimachinery/pkg/util/wait/wait.go:72 +0x4c
created by k8s.io/apimachinery/pkg/util/wait.(*Group).Start in goroutine 54
	/nvme/gopath/src/k8s.io/kubernetes/staging/src/k8s.io/apimachinery/pkg/util/wait/wait.go:70 +0x73

goroutine 60 [select (durable), synctest bubble 2]:
context.(*cancelCtx).propagateCancel.func2()
	/nvme/gopath/go-1.25.2/src/context/context.go:523 +0x98
created by context.(*cancelCtx).propagateCancel in goroutine 58
	/nvme/gopath/go-1.25.2/src/context/context.go:522 +0x409

goroutine 56 [sync.RWMutex.RLock, synctest bubble 2]:
sync.runtime_SemacquireRWMutexR(0x3b9aca00?, 0x0?, 0xb8eec0?)
	/nvme/gopath/go-1.25.2/src/runtime/sema.go:100 +0x25
sync.(*RWMutex).RLock(...)
	/nvme/gopath/go-1.25.2/src/sync/rwmutex.go:74
k8s.io/client-go/tools/cache.(*sharedProcessor).run.func1(0xc0003280f0)
	/nvme/gopath/src/k8s.io/kubernetes/staging/src/k8s.io/client-go/tools/cache/shared_informer.go:904 +0x65
k8s.io/client-go/tools/cache.(*sharedProcessor).run(0xc0003280f0, {0xf8e010, 0xc000328140})
	/nvme/gopath/src/k8s.io/kubernetes/staging/src/k8s.io/client-go/tools/cache/shared_informer.go:912 +0x46
k8s.io/client-go/tools/cache.(*sharedIndexInformer).RunWithContext.(*Group).StartWithContext.func4()
	/nvme/gopath/src/k8s.io/kubernetes/staging/src/k8s.io/apimachinery/pkg/util/wait/wait.go:63 +0x1f
k8s.io/apimachinery/pkg/util/wait.(*Group).Start.func1()
	/nvme/gopath/src/k8s.io/kubernetes/staging/src/k8s.io/apimachinery/pkg/util/wait/wait.go:72 +0x4c
created by k8s.io/apimachinery/pkg/util/wait.(*Group).Start in goroutine 54
	/nvme/gopath/src/k8s.io/kubernetes/staging/src/k8s.io/apimachinery/pkg/util/wait/wait.go:70 +0x73

goroutine 57 [chan receive (durable), synctest bubble 2]:
k8s.io/client-go/tools/cache.(*controller).RunWithContext.func1()
	/nvme/gopath/src/k8s.io/kubernetes/staging/src/k8s.io/client-go/tools/cache/controller.go:167 +0x2c
created by k8s.io/client-go/tools/cache.(*controller).RunWithContext in goroutine 54
	/nvme/gopath/src/k8s.io/kubernetes/staging/src/k8s.io/client-go/tools/cache/controller.go:166 +0xdf

goroutine 61 [sync.RWMutex.Lock, synctest bubble 2]:
sync.runtime_SemacquireRWMutex(0x2?, 0x78?, 0x2c00000000000070?)
	/nvme/gopath/go-1.25.2/src/runtime/sema.go:105 +0x25
sync.(*RWMutex).Lock(0x2?)
	/nvme/gopath/go-1.25.2/src/sync/rwmutex.go:155 +0x6b
k8s.io/client-go/tools/cache.(*sharedProcessor).shouldResync(0xc0003280f0)
	/nvme/gopath/src/k8s.io/kubernetes/staging/src/k8s.io/client-go/tools/cache/shared_informer.go:937 +0x49
k8s.io/client-go/tools/cache.(*Reflector).startResync(0xc00030e690, {0xf8e010, 0xc000328280}, 0xc0003163f0)
	/nvme/gopath/src/k8s.io/kubernetes/staging/src/k8s.io/client-go/tools/cache/reflector.go:464 +0x156
k8s.io/client-go/tools/cache.(*Reflector).watchWithResync.func2()
	/nvme/gopath/src/k8s.io/kubernetes/staging/src/k8s.io/client-go/tools/cache/reflector.go:489 +0x25
k8s.io/apimachinery/pkg/util/wait.(*Group).Start.func1()
	/nvme/gopath/src/k8s.io/kubernetes/staging/src/k8s.io/apimachinery/pkg/util/wait/wait.go:72 +0x4c
created by k8s.io/apimachinery/pkg/util/wait.(*Group).Start in goroutine 58
	/nvme/gopath/src/k8s.io/kubernetes/staging/src/k8s.io/apimachinery/pkg/util/wait/wait.go:70 +0x73
FAIL	k8s.io/client-go/tools/cache	1.011s
FAIL
```

Code after:
	informer.AddEventHandlerWithResyncPeriod(listener2, listener2.resyncPeriod)

	// listener 3, resync every 3s
	listener3 := newTestListener("listener3", 3*time.Second, "pod1", "pod2")
	listener3.printlnFunc = println
	informer.AddEventHandlerWithResyncPeriod(listener3, listener3.resyncPeriod)
	listeners := []*testListener{listener1, listener2, listener3}

	var wg wait.Group
	stop := make(chan struct{})
	wg.StartWithChannel(stop, informer.Run)
	defer func() {
		close(stop)
		wg.Wait()
	}()

	// We must potentially advance time to unblock sharedProcess.run,
	// otherwise synctest.Wait() below returns without anything
	// being done.
	time.Sleep(startupDelay)

	// Ensure all listeners got the initial after the initial processing, at the initial start time.
	//
	// synctest.Wait doesn't detect deadlocks involving mutexes: they are considered not durably
	// blocking, so in case of such a deadlock it'll just keep waiting until the entire unit test
	// times out. The backtraces then show the deadlock.
	synctest.Wait()
	for _, listener := range listeners {
		if !listener.satisfiedExpectations() {
			t.Errorf("%s: %s: expected %v, got %v", time.Since(start), listener.name, listener.expectedItemNames, listener.receivedItemNames)
		}
	}

	// reset
	for _, listener := range listeners {
		listener.reset()
	}

	// advance so listener2 gets a resync
	time.Sleep(2*time.Second - startupDelay)
	synctest.Wait()

	// make sure listener2 got the resync
	if !listener2.satisfiedExpectations() {
		t.Errorf("%s: %s: expected %v, got %v", time.Since(start), listener2.name, listener2.expectedItemNames, listener2.receivedItemNames)
	}

	// make sure listeners 1 and 3 got nothing
	if len(listener1.receivedItemNames) != 0 {
		t.Errorf("%s: listener1: should not have resynced (got %d)", time.Since(start), len(listener1.receivedItemNames))
	}