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

Code before:
	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)
	})
	return Deltas{item}, err
}

// whileProcessing_locked processes the given function while unlocking the lock if we are able to, namely if we are emitting atomic events.
func (f *RealFIFO) whileProcessing_locked(process func() error) error {
	// If we are emitting atomic events and we have the , 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.
	//
	// We only do this if we have less than 2 full batches of items in the queue, to prevent the number of items in the queue from growing
	// unboundedly.
	if f.unlockWhileProcessing && len(f.items) < f.batchSize*2 {
		f.lock.Unlock()
		defer f.lock.Lock()
	}
	return process()
}

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

Review comment (performance):
Our intent here is to unlock _only_ so calls to _add_ to the queue can unblock

However, a side-effect is that any pending concurrent calls to Pop/PopBatch would also unlock and run concurrently. We don't actually expect those from any in-tree code (`controller#processLoop` is single-threaded), but I don't love opening up the possibility of that.

A few things we could do:

1. Document on `Queue#Pop`, `QueueWithBatch#PopBatch`, `RealFIFO#{Pop,PopBatch}` and on `RealFIFOOptions#UnlockWhileProcessing` that calls to Pop or PopBatch must be single-threaded

2. Move the helper function that makes it easy to invoke Pop ad-hoc to be test-only and reproduce it privately in the one spot we have a unit test calling it ... move this bit to fifo_test.go to avoid encouraging people to call Pop directly:
```diff
-// Pop is helper function for popping from Queue.
-// WARNING: Do NOT use this function in non-test code to avoid races
-// unless you really really really really know what you are doing.
-//
-// NOTE: This function is deprecated and may be removed in the future without
-// additional warning.
-func Pop(queue Queue) interface{} {
-       var result interface{}
-       queue.Pop(func(obj interface{}, isInInitialList bool) error {
-               result = obj
-               return nil
-       })
-       return result
-}
-
```

3. We *could* actively prevent concurrent calls to Pop/PopBatch by having a popLock we acquire / release, though I don't really think that is necessary

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

	// 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)
	})
	return Deltas{item}, err
}

// whileProcessing_locked calls the `process` function.
// The lock must be held before calling `whileProcessing_locked`, and is held when `whileProcessing_locked` returns.
// whileProcessing_locked releases the lock during the call to `process` if f.unlockWhileProcessing is true and the f.items queue is not too long.
func (f *RealFIFO) whileProcessing_locked(process func() error) error {
	// Unlock before calling `process` so new items can be enqueued during processing.
	// Only do this if the queue contains less than 2 full batches of items,
	// to prevent the queue from growing unboundedly.
	if f.unlockWhileProcessing && len(f.items) < f.batchSize*2 {
		f.lock.Unlock()
		defer f.lock.Lock()
	}
	return process()
}

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