Fix race condition between Pop and Close FIFO queue

Fixes: kubernetes#90581 (the first part)

When `Close()` is invoked on an empty queue, the control loop inside `Pop()` has a small chance of missing the signal and blocks indefinitely due to a race condition. This PR eliminates the race and allows the control loop inside any blocking `Pop()` to successfully exit after Close() is called.

Kubernetes-commit: d8b90955519d10b99415515f8314dd6d35caae8d

tools/cache/delta_fifo_test.go

@@ -19,6 +19,7 @@ package cache
 import (
 	"fmt"
 	"reflect"
+	"runtime"
 	"testing"
 	"time"
 )
@@ -645,3 +646,36 @@ func TestDeltaFIFO_HasSynced(t *testing.T) {
 		}
 	}
 }
+
+// TestDeltaFIFO_PopShouldUnblockWhenClosed checks that any blocking Pop on an empty queue
+// should unblock and return after Close is called.
+func TestDeltaFIFO_PopShouldUnblockWhenClosed(t *testing.T) {
+	f := NewDeltaFIFOWithOptions(DeltaFIFOOptions{
+		KeyFunction: testFifoObjectKeyFunc,
+		KnownObjects: literalListerGetter(func() []testFifoObject {
+			return []testFifoObject{mkFifoObj("foo", 5)}
+		}),
+	})
+
+	c := make(chan struct{})
+	const jobs = 10
+	for i := 0; i < jobs; i++ {
+		go func() {
+			f.Pop(func(obj interface{}) error {
+				return nil
+			})
+			c <- struct{}{}
+		}()
+	}
+
+	runtime.Gosched()
+	f.Close()
+
+	for i := 0; i < jobs; i++ {
+		select {
+		case <-c:
+		case <-time.After(500 * time.Millisecond):
+			t.Fatalf("timed out waiting for Pop to return after Close")
+		}
+	}
+}