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

Code before:
	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))
	// only bundle unique items into a batch, always batch atomic replaces separately.
	for i := 0; i < f.batchSize && i < len(f.items); i++ {
		if f.initialPopulationCount > 0 && i >= f.initialPopulationCount {
			break
		}
		item := f.items[i]
		if item.Type == ReplacedList {
			if len(deltas) == 0 {
				deltas = append(deltas, item)
			}
			break
		}
		id, err := f.keyOf(item)
		if err != nil {
			// close the batch here if error happens
			// TODO: log the error when RealFIFOOptions supports passing klog instance like deprecated DeltaFIFO
			// still pop the broken item out of queue to be compatible with the non-batch behavior it should be safe
			// when 1st element is broken, however for Nth broken element, there's possible risk that broken item
			// still can be processed and broke the uniqueness of the batch unexpectedly.
			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{}
			break
		}
		if unique.Has(id) {
			break
		}
		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

Review comment (suggestion):
consider doing this in a defer rather than relocating it to the end of the function to keep it alongside the items truncation and make sure an early return doesn't skip this

`defer func(){ f.initialPopulationCount -= len(deltas) }()`

Code after:
		}

		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{}
	}
	// only bundle unique items into a batch
	for i := 0; i < f.batchSize && i < len(f.items); i++ {
		if f.initialPopulationCount > 0 && i >= f.initialPopulationCount {
			break
		}
		item := f.items[i]
		if !batchable[item.Type] {
			if len(deltas) == 0 {
				// if an unbatchable delta is first in the list, process just that one by itself
				moveDeltaToProcessList(i)
			}
			// close the batch when an unbatchable delta is encountered
			break
		}
		id, err := f.keyOf(item)
		if err != nil {
			// close the batch here if error happens
			// TODO: log the error when RealFIFOOptions supports passing klog instance like deprecated DeltaFIFO
			// still pop the broken item out of queue to be compatible with the non-batch behavior it should be safe
			// when 1st element is broken, however for Nth broken element, there's possible risk that broken item
			// still can be processed and broke the uniqueness of the batch unexpectedly.
			moveDeltaToProcessList(i)
			break
		}
		if unique.Has(id) {
			// close the batch if a duplicate item is encountered
			break
		}
		unique.Insert(id)
		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)