nodebatcher: Allow batcher pre-allocations to reduce on new batch (#419)

* Allow batcher pre-allocations to reduce on new batch

Currently, we re-use the previous capacity size when pre-allocating
space for the new batch in nodebatcher. This doesn't work well when a
node has had high throughput and then stops sending traffic, or maybe
the process that was sending a lot of traffic shifts to another node.
This PR reduces allocation space based not just on current cap, but also
current size to fix this.

* Move comment placement

* Refactor marginally

* PR comments

Author: Steven Karis

internal/collector/processor/nodebatcher/node_batcher.go

@@ -37,7 +37,7 @@ import (
 )
 
 const (
-	initialBatchCapacity     = uint32(1024)
+	initialBatchCapacity     = uint32(512)
 	nodeStatusDead           = uint32(1)
 	batchStatusClosed        = uint32(1)
 	tickerPendingNodesBuffer = 16
@@ -237,13 +237,20 @@ func (nb *nodeBatcher) add(spans []*tracepb.Span) {
 
 // cutBatch cuts the provided batch, and sets a new batch on this nodeBatcher
 func (nb *nodeBatcher) cutBatch(b *batch) {
+	initialCap := b.getCurrentCap()
+	currSize := b.getCurrentItemCount()
+	reducedCap := initialCap >> 1
+	for currSize < reducedCap && reducedCap > initialBatchCapacity {
+		initialCap = reducedCap
+		reducedCap = reducedCap >> 1
+	}
 	// atomic.CompareAndSwapPointer only takes unsafe.Pointer interfaces. We do not use unsafe
 	// to skirt around the golang type system.
 	currBatchPtr := (*unsafe.Pointer)(unsafe.Pointer(&nb.currBatch))
 	swapped := atomic.CompareAndSwapPointer(
 		currBatchPtr,
 		unsafe.Pointer(b),
-		unsafe.Pointer(newBatch(b.getCurrentCap(), nb.sendBatchSize)),
+		unsafe.Pointer(newBatch(initialCap, nb.sendBatchSize)),
 	)
 	// Since we are doing an atomic compare and swap, this batch will only be sent once.
 	if swapped {