[aggregator] Pool locked aggregations

src/aggregator/aggregation/timer.go

@@ -158,4 +158,6 @@ func (t *Timer) Annotation() []byte {
 }
 
 // Close closes the timer.
-func (t *Timer) Close() { t.stream.Close() }
+func (t *Timer) Close() {
+	t.stream.Close()
+}

src/aggregator/aggregator/elem_base.go

@@ -739,3 +739,76 @@ func newParsedPipeline(pipeline applied.Pipeline) (parsedPipeline, error) {
 		Rollup:                 rollup,
 	}, nil
 }
+
+// Placeholder to make compiler happy about generic elem base.
+// NB: lockedAggregationFromPool and not newLockedAggregation to avoid yet another rename hack in makefile
+func lockedAggregationFromPool(
+	aggregation typeSpecificAggregation,
+	sourcesSeen map[uint32]*bitset.BitSet,
+) *lockedAggregation {
+	return &lockedAggregation{
+		aggregation: aggregation,
+		sourcesSeen: sourcesSeen,
+	}
+}
+
+func (l *lockedAggregation) close() {
+	l.aggregation.Close()
+}
+
+var lockedCounterAggregationPool = sync.Pool{New: func() interface{} { return &lockedCounterAggregation{} }}
+
+func lockedCounterAggregationFromPool(
+	aggregation counterAggregation,
+	sourcesSeen map[uint32]*bitset.BitSet,
+) *lockedCounterAggregation {
+	l := lockedCounterAggregationPool.Get().(*lockedCounterAggregation)
+	l.aggregation = aggregation
+	l.sourcesSeen = sourcesSeen
+
+	return l
+}
+
+func (l *lockedCounterAggregation) close() {
+	l.aggregation.Close()
+	*l = lockedCounterAggregation{}
+	lockedCounterAggregationPool.Put(l)
+}
+
+var lockedGaugeAggregationPool = sync.Pool{New: func() interface{} { return &lockedGaugeAggregation{} }}
+
+func lockedGaugeAggregationFromPool(
+	aggregation gaugeAggregation,
+	sourcesSeen map[uint32]*bitset.BitSet,
+) *lockedGaugeAggregation {
+	l := lockedGaugeAggregationPool.Get().(*lockedGaugeAggregation)
+	l.aggregation = aggregation
+	l.sourcesSeen = sourcesSeen
+
+	return l
+}
+
+func (l *lockedGaugeAggregation) close() {
+	l.aggregation.Close()
+	*l = lockedGaugeAggregation{}
+	lockedGaugeAggregationPool.Put(l)
+}
+
+var lockedTimerAggregationPool = sync.Pool{New: func() interface{} { return &lockedTimerAggregation{} }}
+
+func lockedTimerAggregationFromPool(
+	aggregation timerAggregation,
+	sourcesSeen map[uint32]*bitset.BitSet,
+) *lockedTimerAggregation {
+	l := lockedTimerAggregationPool.Get().(*lockedTimerAggregation)
+	l.aggregation = aggregation
+	l.sourcesSeen = sourcesSeen
+
+	return l
+}
+
+func (l *lockedTimerAggregation) close() {
+	l.aggregation.Close()
+	*l = lockedTimerAggregation{}
+	lockedTimerAggregationPool.Put(l)
+}

src/aggregator/aggregator/generic_elem.go

@@ -126,7 +126,7 @@ type timedAggregation struct {
 
 // close is called when the aggregation has been expired or the element is being closed.
 func (ta *timedAggregation) close() {
-	ta.lockedAgg.aggregation.Close()
+	ta.lockedAgg.close()
 	ta.lockedAgg = nil
 }
 
@@ -744,12 +744,17 @@ func (e *GenericElem) findOrCreate(
 			sourcesSeen = make(map[uint32]*bitset.BitSet)
 		}
 	}
+	// NB(vytenis): lockedAggregation will be returned to pool on timedAggregation close.
+	// this is a bit different from regular pattern of using a pool object due to codegen with Genny limitations,
+	// so we can avoid writing more boilerplate.
+	// timedAggregation itself is always pass-by-value, but lockedAggregation incurs an expensive allocation on heap
+	// in the critical path (30%+, depending on workload as of 2020-05-01): see https://github.com/m3db/m3/pull/4109
 	timedAgg = timedAggregation{
 		startAt: alignedStart,
-		lockedAgg: &lockedAggregation{
-			sourcesSeen: sourcesSeen,
-			aggregation: e.NewAggregation(e.opts, e.aggOpts),
-		},
+		lockedAgg: lockedAggregationFromPool(
+			e.NewAggregation(e.opts, e.aggOpts),
+			sourcesSeen,
+		),
 		inDirtySet: true,
 	}