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

Code before:
// 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) {
	f.lock.Lock()
	defer f.lock.Unlock()

	for len(f.items) == 0 {
		// 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().
		if f.closed {
			return nil, ErrFIFOClosed
		}

		f.cond.Wait()
	}

	isInInitialList := !f.hasSynced_locked()
	item := f.items[0]
	// The underlying array still exists and references this object, so the object will not be garbage collected unless we zero the reference.
	f.items[0] = Delta{}
	f.items = f.items[1:]
	// Decrement initialPopulationCount if it is greater than 0 after processing, to ensure that anything waiting for the cache
	// to be populated will be unblocked only after the cache has been fully populated. This is required to ensure that functions such as
	// hasSynced will not return true before the cache has been fully populated since it also waits for the initialPopulationCount to be 0.
	defer func() {
		if f.initialPopulationCount > 0 {
			f.initialPopulationCount--
		}
	}()

	// Only log traces if the queue depth is greater than 10 and it takes more than
	// 100 milliseconds to process one item from the queue.
	// Queue depth never goes high because processing an item is locking the queue,
	// and new items can't be added until processing finish.
	// https://github.com/kubernetes/kubernetes/issues/103789
	if len(f.items) > 10 {
		id, _ := f.keyOf(item)
		trace := utiltrace.New("RealFIFO Pop Process",
			utiltrace.Field{Key: "ID", Value: id},
			utiltrace.Field{Key: "Depth", Value: len(f.items)},
			utiltrace.Field{Key: "Reason", Value: "slow event handlers blocking the queue"})
		defer trace.LogIfLong(100 * time.Millisecond)
	}

	// Process the item, this may unlock the lock, and allow other goroutines to add items to the queue.
	err := f.whileProcessing_locked(func() error {
		// we wrap in Deltas here to be compatible with preview Pop functions and those interpreting the return value.
		return process(Deltas{item}, isInInitialList)
	})

Review comment (bug):
can probably simplify to something like this (here and in PopBatch):

```go
// Decrement initialPopulationCount if needed.
// This is done in a defer so we only do this *after* processing is complete,
// so concurrent calls to hasSynced will not incorrectly return true while processing is still happening.
```


Code after:
// ready, they are returned in the order in which they were added/updated. The
// item is removed from the queue (and the store) before it is processed. The
// process function is only guaranteed to be called under lock if
// UnlockWhileProcessing is false. If the process function is updating data
// structures that need to be in sync with the queue, ensure
// UnlockWhileProcessing is false. It is expected that the caller of Pop will be
// a single threaded consumer since otherwise it is possible for multiple
// PopProcessFuncs to be running simultaneously.
func (f *RealFIFO) Pop(process PopProcessFunc) (interface{}, error) {
	f.lock.Lock()
	defer f.lock.Unlock()

	for len(f.items) == 0 {
		// 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().
		if f.closed {
			return nil, ErrFIFOClosed
		}

		f.cond.Wait()
	}

	isInInitialList := !f.hasSynced_locked()
	item := f.items[0]
	// The underlying array still exists and references this object, so the object will not be garbage collected unless we zero the reference.
	f.items[0] = Delta{}
	f.items = f.items[1:]
	// Decrement initialPopulationCount if needed.
	// This is done in a defer so we only do this *after* processing is complete,
	// so concurrent calls to hasSynced will not incorrectly return true while processing is still happening.
	defer func() {
		if f.initialPopulationCount > 0 {
			f.initialPopulationCount--
		}
	}()

	// Only log traces if the queue depth is greater than 10 and it takes more than
	// 100 milliseconds to process one item from the queue.
	// Queue depth never goes high because processing an item is locking the queue,
	// and new items can't be added until processing finish.
	// https://github.com/kubernetes/kubernetes/issues/103789
	if len(f.items) > 10 {
		id, _ := f.keyOf(item)
		trace := utiltrace.New("RealFIFO Pop Process",
			utiltrace.Field{Key: "ID", Value: id},
			utiltrace.Field{Key: "Depth", Value: len(f.items)},
			utiltrace.Field{Key: "Reason", Value: "slow event handlers blocking the queue"})
		defer trace.LogIfLong(100 * time.Millisecond)
	}