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

Code before:
	synctest.Test(t, func(t *testing.T) { testListenerResyncPeriods(t, 0) })
}

func TestListenerResyncPeriodsDelayed(t *testing.T) {
	// sharedProcessor used to deadlock when sharedProcessor.run happened
	// to be reached after the first sharedProcessor.distribute.
	// This artififical delay simulates that situation.
	//
	// Must not run in parallel to other tests, but it doesn't need to:
	// it doesn't sleep in real-world time and therefore completes quickly.
	synctest.Test(t, func(t *testing.T) { testListenerResyncPeriods(t, time.Second) })
}

func testListenerResyncPeriods(t *testing.T, startupDelay time.Duration) {
	start := time.Now()
	println := func(msg string) {
		delta := time.Since(start)
		t.Logf("%s: %s", delta, msg)
	}

	sharedProcessorRunDelay.Store(&startupDelay)
	defer sharedProcessorRunDelay.Store(nil)

	// source simulates an apiserver object endpoint.
	source := newFakeControllerSource(t)
	source.Add(&v1.Pod{ObjectMeta: metav1.ObjectMeta{Name: "pod1"}})
	source.Add(&v1.Pod{ObjectMeta: metav1.ObjectMeta{Name: "pod2"}})

	// create the shared informer and resync every 1s
	informer := NewSharedInformer(source, &v1.Pod{}, 1*time.Second).(*sharedIndexInformer)

	// listener 1, never resync
	listener1 := newTestListener("listener1", 0, "pod1", "pod2")
	listener1.printlnFunc = println
	informer.AddEventHandlerWithResyncPeriod(listener1, listener1.resyncPeriod)

	// listener 2, resync every 2s
	listener2 := newTestListener("listener2", 2*time.Second, "pod1", "pod2")
	listener2.printlnFunc = println
	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() {

Review comment (bug):
This was already broken before (`receivedItemNames` is read and written in a different goroutine), but now the race detector noticed.

Code after:
	synctest.Test(t, func(t *testing.T) { testListenerResyncPeriods(t, 0) })
}

func TestListenerResyncPeriodsDelayed(t *testing.T) {
	// sharedProcessor used to deadlock when sharedProcessor.run happened
	// to be reached after the first sharedProcessor.distribute.
	// This artififical delay simulates that situation.
	//
	// Must not run in parallel to other tests, but it doesn't need to:
	// it doesn't sleep in real-world time and therefore completes quickly.
	synctest.Test(t, func(t *testing.T) { testListenerResyncPeriods(t, time.Second) })
}

func testListenerResyncPeriods(t *testing.T, startupDelay time.Duration) {
	start := time.Now()
	println := func(msg string) {
		delta := time.Since(start)
		t.Logf("%s: %s", delta, msg)
	}

	if startupDelay > 0 {
		hook := func() {
			time.Sleep(startupDelay)
		}
		sharedProcessorRunHook.Store(&hook)
		defer sharedProcessorRunHook.Store(nil)
	}

	// source simulates an apiserver object endpoint.
	source := newFakeControllerSource(t)
	source.Add(&v1.Pod{ObjectMeta: metav1.ObjectMeta{Name: "pod1"}})
	source.Add(&v1.Pod{ObjectMeta: metav1.ObjectMeta{Name: "pod2"}})

	// create the shared informer and resync every 1s
	informer := NewSharedInformer(source, &v1.Pod{}, 1*time.Second).(*sharedIndexInformer)

	// listener 1, never resync
	listener1 := newTestListener("listener1", 0, "pod1", "pod2")
	listener1.printlnFunc = println
	informer.AddEventHandlerWithResyncPeriod(listener1, listener1.resyncPeriod)

	// listener 2, resync every 2s
	listener2 := newTestListener("listener2", 2*time.Second, "pod1", "pod2")
	listener2.printlnFunc = println
	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}