cleanup

src/aggregator/aggregation/counter.go

@@ -136,4 +136,6 @@ func (c *Counter) Annotation() []byte {
 }
 
 // Close closes the counter.
-func (c *Counter) Close() {}
+func (c *Counter) Close() {
+	c.annotation = nil
+}

src/aggregator/aggregation/gauge.go

@@ -170,4 +170,6 @@ func (g *Gauge) Annotation() []byte {
 }
 
 // Close closes the gauge.
-func (g *Gauge) Close() {}
+func (g *Gauge) Close() {
+	g.annotation = nil
+}

src/aggregator/aggregation/timer.go

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

src/aggregator/aggregator/elem_base.go

@@ -732,7 +732,7 @@ func newParsedPipeline(pipeline applied.Pipeline) (parsedPipeline, error) {
 	}, nil
 }
 
-// placeholder to make compiler happy about generic elem base
+// 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,
@@ -757,21 +757,13 @@ func lockedCounterAggregationFromPool(
 	l := lockedCounterAggregationPool.Get().(*lockedCounterAggregation)
 	l.aggregation = aggregation
 	l.sourcesSeen = sourcesSeen
-	l.lastUpdatedAt = 0
-	l.dirty = false
-	l.closed = false
-	l.resendEnabled = false
+
 	return l
 }
 
 func (l *lockedCounterAggregation) close() {
 	l.aggregation.Close()
-	l.aggregation = counterAggregation{}
-	l.sourcesSeen = nil
-	l.lastUpdatedAt = 0
-	l.dirty = false
-	l.closed = false
-	l.resendEnabled = false
+	*l = lockedCounterAggregation{}
 	lockedCounterAggregationPool.Put(l)
 }
 
@@ -784,17 +776,13 @@ func lockedGaugeAggregationFromPool(
 	l := lockedGaugeAggregationPool.Get().(*lockedGaugeAggregation)
 	l.aggregation = aggregation
 	l.sourcesSeen = sourcesSeen
+
 	return l
 }
 
 func (l *lockedGaugeAggregation) close() {
 	l.aggregation.Close()
-	l.aggregation = gaugeAggregation{}
-	l.sourcesSeen = nil
-	l.lastUpdatedAt = 0
-	l.dirty = false
-	l.closed = false
-	l.resendEnabled = false
+	*l = lockedGaugeAggregation{}
 	lockedGaugeAggregationPool.Put(l)
 }
 
@@ -807,16 +795,12 @@ func lockedTimerAggregationFromPool(
 	l := lockedTimerAggregationPool.Get().(*lockedTimerAggregation)
 	l.aggregation = aggregation
 	l.sourcesSeen = sourcesSeen
+
 	return l
 }
 
 func (l *lockedTimerAggregation) close() {
 	l.aggregation.Close()
-	l.aggregation = timerAggregation{}
-	l.sourcesSeen = nil
-	l.lastUpdatedAt = 0
-	l.dirty = false
-	l.closed = false
-	l.resendEnabled = false
+	*l = lockedTimerAggregation{}
 	lockedTimerAggregationPool.Put(l)
 }

src/aggregator/aggregator/generic_elem.go

@@ -742,7 +742,11 @@ func (e *GenericElem) findOrCreate(
 			sourcesSeen = make(map[uint32]*bitset.BitSet)
 		}
 	}
-	// lockedAggregation will be returned to pool on timedAggregation close
+	// 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: lockedAggregationFromPool(