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

Code before:
	// 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.processUnlocked(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)
	})

	// Decrement initialPopulationCount if it is greater than 0 after processing the item, to ensure that anything waiting for the cache
	// to be populated will be unblocked only after the cache has been fully populated.
	if f.initialPopulationCount > 0 {
		f.initialPopulationCount--
	}
	return Deltas{item}, err
}

// processUnlocked processes the given function while unlocking the lock if we are able to, namely if we are emitting atomic events.
func (f *RealFIFO) processUnlocked(process func() error) error {
	// If we are emitting atomic events, we can optimize the processing by unlocking the lock before calling process so that the
	// reflector is not blocked while we process the events. This is because we never access the store in the FIFO, and the store will
	// stay consistent while we process the events.
	if f.emitAtomicEvents {
		f.lock.Unlock()
		defer f.lock.Lock()
	}
	return process()
}

func (f *RealFIFO) PopBatch(processBatch ProcessBatchFunc, process PopProcessFunc) 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 ErrFIFOClosed
		}

		f.cond.Wait()
	}


Review comment (performance):
Let's make a dedicated gate for this, something like `UnlockWhileProcessingFIFO`. Make sure if that is set, AtomicFIFO is also set and KnownObjects is nil.

Rename the function to make it sound less definitive that this is unlocking.

```go
// whileProcessingLocked wraps the actual call to a process or process batch function.
// This allows the RealFIFO to conditionally unlock during the process call if UnlockWhileProcessingFIFO is enabled.
// This function must be called under lock.
// ... maybe link to the issue or description of why this is necessary and possible
func (f *RealFIFO) whileProcessing_locked(process func() error) error {
  if f.unlockWhileProcessing {
    f.lock.Unlock()
    defer f.lock.Lock()
  }
  return process()
}
```


Code after:
	// 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)
	}

	// we wrap in Deltas here to be compatible with preview Pop functions and those interpreting the return value.
	err := process(Deltas{item}, isInInitialList)
	return Deltas{item}, err
}

// batchable stores the delta types that can be batched
var batchable = map[DeltaType]bool{
	Sync:     true,
	Replaced: true,
	Added:    true,
	Updated:  true,
	Deleted:  true,
}

// PopBatch pops as many items as possible to be processed as a batch using processBatch,
// or pop a single item using processSingle if multiple items cannot be batched.
func (f *RealFIFO) PopBatch(processBatch ProcessBatchFunc, processSingle PopProcessFunc) 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 ErrFIFOClosed
		}

		f.cond.Wait()
	}

	isInInitialList := !f.hasSynced_locked()
	unique := sets.NewString()
	deltas := make([]Delta, 0, min(len(f.items), f.batchSize))
	moveDeltaToProcessList := func(i int) {
		deltas = append(deltas, f.items[i])
		// The underlying array still exists and references this object, so the object will not be garbage collected unless we zero the reference.
		f.items[i] = Delta{}
	}