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

Code before:
		}
		unique.Insert(id)
		deltas = append(deltas, item)
		// 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{}
	}
	f.items = f.items[len(deltas):]

	// 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 (with a max of 1 second for the whole batch)
	// 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(deltas[0])
		trace := utiltrace.New("RealFIFO PopBatch 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"},
			utiltrace.Field{Key: "BatchSize", Value: len(deltas)})
		defer trace.LogIfLong(min(100*time.Millisecond*time.Duration(len(deltas)), time.Second))
	}

	var err error
	if len(deltas) == 1 {
		err = f.processUnlocked(func() error {
			return process(Deltas{deltas[0]}, isInInitialList)
		})
	} else {
		err = f.processUnlocked(func() error {
			return processBatch(deltas, isInInitialList)
		})
	}
	// Decrement initialPopulationCount if it is greater than 0 after processing the batch, 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 -= len(deltas)
	}
	return err
}

// Replace
// 1. finds those items in f.items that are not in newItems and creates synthetic deletes for them
// 2. finds items in knownObjects that are not in newItems and creates synthetic deletes for them
// 3. adds the newItems to the queue
func (f *RealFIFO) Replace(newItems []interface{}, resourceVersion string) error {
	f.lock.Lock()
	defer f.lock.Unlock()

	if f.emitAtomicEvents {
		f.items = f.items[:0]

Review comment (suggestion):
this can go back to `return f.processUnlocked(...` if we handle the initialPopulationCount with a defer

Code after:
		moveDeltaToProcessList(i)
	}
	if f.initialPopulationCount > 0 {
		f.initialPopulationCount -= len(deltas)
	}
	f.items = f.items[len(deltas):]

	// 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 (with a max of 1 second for the whole batch)
	// 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(deltas[0])
		trace := utiltrace.New("RealFIFO PopBatch 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"},
			utiltrace.Field{Key: "BatchSize", Value: len(deltas)})
		defer trace.LogIfLong(min(100*time.Millisecond*time.Duration(len(deltas)), time.Second))
	}

	if len(deltas) == 1 {
		return processSingle(Deltas{deltas[0]}, isInInitialList)
	}
	return processBatch(deltas, isInInitialList)
}

// Replace
// 1. finds those items in f.items that are not in newItems and creates synthetic deletes for them
// 2. finds items in knownObjects that are not in newItems and creates synthetic deletes for them
// 3. adds the newItems to the queue
func (f *RealFIFO) Replace(newItems []interface{}, resourceVersion string) error {
	f.lock.Lock()
	defer f.lock.Unlock()

	var err error
	if f.emitAtomicEvents {
		err = f.addReplaceToItemsLocked(newItems, resourceVersion)
	} else {
		err = reconcileReplacement(f.items, f.knownObjects, newItems, f.keyOf,
			func(obj DeletedFinalStateUnknown) error {
				return f.addToItems_locked(Deleted, true, obj)
			},
			func(obj interface{}) error {
				return f.addToItems_locked(Replaced, false, obj)
			})
	}
	if err != nil {
		return err
	}