refactor into cumulative and delta sum types

Author: dashpole

sdk/metric/internal/aggregate/aggregate.go

@@ -110,11 +110,12 @@ func (b Builder[N]) PrecomputedSum(monotonic bool) (Measure[N], ComputeAggregati
 
 // Sum returns a sum aggregate function input and output.
 func (b Builder[N]) Sum(monotonic bool) (Measure[N], ComputeAggregation) {
-	s := newSum[N](monotonic, b.Temporality, b.AggregationLimit, b.resFunc())
 	switch b.Temporality {
 	case metricdata.DeltaTemporality:
+		s := newDeltaSum[N](monotonic, b.AggregationLimit, b.resFunc())
 		return b.filter(s.measure), s.delta
 	default:
+		s := newCumulativeSum[N](monotonic, b.AggregationLimit, b.resFunc())
 		return b.filter(s.measure), s.cumulative
 	}
 }

sdk/metric/internal/aggregate/sum.go

@@ -19,57 +19,34 @@ type sumValue[N int64 | float64] struct {
 	attrs attribute.Set
 }
 
-// valueMap is the storage for sums.
 type valueMap[N int64 | float64] struct {
-	newRes   func(attribute.Set) FilteredExemplarReservoir[N]
-	aggLimit int
-
-	// cumulative sums do not reset values during collection, so in that case
-	// clearValuesOnCollection is false, hcwg is unused, and only values[0]
-	// and len[0] are used. All other aggregations reset on collection, so we
-	// use hcwg to swap between the hot and cold maps and len so measurements
-	// can continue without blocking on collection.
-	//
-	// see hotColdWaitGroup for how this works.
-	clearValuesOnCollection bool
-	hcwg                    hotColdWaitGroup
-	values                  [2]sync.Map
-	len                     [2]atomic.Int64
-}
-
-func newValueMap[N int64 | float64](
-	limit int,
-	r func(attribute.Set) FilteredExemplarReservoir[N],
-	clearValuesOnCollection bool,
-) *valueMap[N] {
-	return &valueMap[N]{
-		newRes:                  r,
-		aggLimit:                limit,
-		clearValuesOnCollection: clearValuesOnCollection,
-	}
+	values sync.Map
+	len    atomic.Int64
 }
 
-func (s *valueMap[N]) measure(ctx context.Context, value N, fltrAttr attribute.Set, droppedAttr []attribute.KeyValue) {
-	hotIdx := uint64(0)
-	if s.clearValuesOnCollection {
-		hotIdx = s.hcwg.start()
-		defer s.hcwg.done(hotIdx)
-	}
-	v, ok := s.values[hotIdx].Load(fltrAttr.Equivalent())
+func (s *valueMap[N]) measure(
+	ctx context.Context,
+	value N,
+	fltrAttr attribute.Set,
+	droppedAttr []attribute.KeyValue,
+	newRes func(attribute.Set) FilteredExemplarReservoir[N],
+	aggLimit int,
+) {
+	v, ok := s.values.Load(fltrAttr.Equivalent())
 	if !ok {
 		// It is possible to exceed the attribute limit if it races with other
 		// new attribute sets. This is an accepted tradeoff to avoid locking
 		// for writes.
-		if s.aggLimit > 0 && s.len[hotIdx].Load() >= int64(s.aggLimit-1) {
+		if aggLimit > 0 && s.len.Load() >= int64(aggLimit-1) {
 			fltrAttr = overflowSet
 		}
 		var loaded bool
-		v, loaded = s.values[hotIdx].LoadOrStore(fltrAttr.Equivalent(), &sumValue[N]{
-			res:   s.newRes(fltrAttr),
+		v, loaded = s.values.LoadOrStore(fltrAttr.Equivalent(), &sumValue[N]{
+			res:   newRes(fltrAttr),
 			attrs: fltrAttr,
 		})
 		if !loaded {
-			s.len[hotIdx].Add(1)
+			s.len.Add(1)
 		}
 	}
 	sv := v.(*sumValue[N])
@@ -80,32 +57,41 @@ func (s *valueMap[N]) measure(ctx context.Context, value N, fltrAttr attribute.S
 	sv.res.Offer(ctx, value, droppedAttr)
 }
 
-// newSum returns an aggregator that summarizes a set of measurements as their
-// arithmetic sum. Each sum is scoped by attributes and the aggregation cycle
-// the measurements were made in.
-func newSum[N int64 | float64](
+// newDeltaSum returns an aggregator that summarizes a set of measurements as
+// their arithmetic sum. Each sum is scoped by attributes and the aggregation
+// cycle the measurements were made in.
+func newDeltaSum[N int64 | float64](
 	monotonic bool,
-	temporality metricdata.Temporality,
 	limit int,
 	r func(attribute.Set) FilteredExemplarReservoir[N],
-) *sum[N] {
-	clearValuesOnCollection := temporality == metricdata.DeltaTemporality
-	return &sum[N]{
-		valueMap:  newValueMap[N](limit, r, clearValuesOnCollection),
+) *deltaSum[N] {
+	return &deltaSum[N]{
+		newRes:    r,
+		aggLimit:  limit,
 		monotonic: monotonic,
 		start:     now(),
 	}
 }
 
-// sum summarizes a set of measurements made as their arithmetic sum.
-type sum[N int64 | float64] struct {
-	*valueMap[N]
+// deltaSum is the storage for sums which resets every collection interval.
+type deltaSum[N int64 | float64] struct {
+	newRes   func(attribute.Set) FilteredExemplarReservoir[N]
+	aggLimit int
 
 	monotonic bool
 	start     time.Time
+
+	hcwg          hotColdWaitGroup
+	hotColdValMap [2]valueMap[N]
 }
 
-func (s *sum[N]) delta(
+func (s *deltaSum[N]) measure(ctx context.Context, value N, fltrAttr attribute.Set, droppedAttr []attribute.KeyValue) {
+	hotIdx := s.hcwg.start()
+	defer s.hcwg.done(hotIdx)
+	s.hotColdValMap[hotIdx].measure(ctx, value, fltrAttr, droppedAttr, s.newRes, s.aggLimit)
+}
+
+func (s *deltaSum[N]) delta(
 	dest *metricdata.Aggregation, //nolint:gocritic // The pointer is needed for the ComputeAggregation interface
 ) int {
 	t := now()
@@ -120,11 +106,11 @@ func (s *sum[N]) delta(
 	readIdx := s.hcwg.swapHotAndWait()
 	// The len will not change while we iterate over values, since we waited
 	// for all writes to finish to the cold values and len.
-	n := int(s.len[readIdx].Load())
+	n := int(s.hotColdValMap[readIdx].len.Load())
 	dPts := reset(sData.DataPoints, n, n)
 
 	var i int
-	s.values[readIdx].Range(func(_, value any) bool {
+	s.hotColdValMap[readIdx].values.Range(func(_, value any) bool {
 		val := value.(*sumValue[N])
 		collectExemplars(&dPts[i].Exemplars, val.res.Collect)
 		dPts[i].Attributes = val.attrs
@@ -134,8 +120,8 @@ func (s *sum[N]) delta(
 		i++
 		return true
 	})
-	s.values[readIdx].Clear()
-	s.len[readIdx].Store(0)
+	s.hotColdValMap[readIdx].values.Clear()
+	s.hotColdValMap[readIdx].len.Store(0)
 	// The delta collection cycle resets.
 	s.start = t
 
@@ -145,7 +131,38 @@ func (s *sum[N]) delta(
 	return i
 }
 
-func (s *sum[N]) cumulative(
+// newCumulativeSum returns an aggregator that summarizes a set of measurements
+// as their arithmetic sum. Each sum is scoped by attributes and the
+// aggregation cycle the measurements were made in.
+func newCumulativeSum[N int64 | float64](
+	monotonic bool,
+	limit int,
+	r func(attribute.Set) FilteredExemplarReservoir[N],
+) *cumulativeSum[N] {
+	return &cumulativeSum[N]{
+		newRes:    r,
+		aggLimit:  limit,
+		monotonic: monotonic,
+		start:     now(),
+	}
+}
+
+// deltaSum is the storage for sums which never reset.
+type cumulativeSum[N int64 | float64] struct {
+	newRes   func(attribute.Set) FilteredExemplarReservoir[N]
+	aggLimit int
+
+	monotonic bool
+	start     time.Time
+
+	valMap valueMap[N]
+}
+
+func (s *cumulativeSum[N]) measure(ctx context.Context, value N, fltrAttr attribute.Set, droppedAttr []attribute.KeyValue) {
+	s.valMap.measure(ctx, value, fltrAttr, droppedAttr, s.newRes, s.aggLimit)
+}
+
+func (s *cumulativeSum[N]) cumulative(
 	dest *metricdata.Aggregation, //nolint:gocritic // The pointer is needed for the ComputeAggregation interface
 ) int {
 	t := now()
@@ -156,13 +173,12 @@ func (s *sum[N]) cumulative(
 	sData.Temporality = metricdata.CumulativeTemporality
 	sData.IsMonotonic = s.monotonic
 
-	readIdx := 0
 	// Values are being concurrently written while we iterate, so only use the
 	// current length for capacity.
-	dPts := reset(sData.DataPoints, 0, int(s.len[readIdx].Load()))
+	dPts := reset(sData.DataPoints, 0, int(s.valMap.len.Load()))
 
 	var i int
-	s.values[readIdx].Range(func(_, value any) bool {
+	s.valMap.values.Range(func(_, value any) bool {
 		val := value.(*sumValue[N])
 		newPt := metricdata.DataPoint[N]{
 			Attributes: val.attrs,
@@ -195,18 +211,13 @@ func newPrecomputedSum[N int64 | float64](
 	r func(attribute.Set) FilteredExemplarReservoir[N],
 ) *precomputedSum[N] {
 	return &precomputedSum[N]{
-		valueMap:  newValueMap[N](limit, r, true),
-		monotonic: monotonic,
-		start:     now(),
+		deltaSum: newDeltaSum(monotonic, limit, r),
 	}
 }
 
 // precomputedSum summarizes a set of observations as their arithmetic sum.
 type precomputedSum[N int64 | float64] struct {
-	*valueMap[N]
-
-	monotonic bool
-	start     time.Time
+	*deltaSum[N]
 
 	reported map[any]N
 }
@@ -227,11 +238,11 @@ func (s *precomputedSum[N]) delta(
 	readIdx := s.hcwg.swapHotAndWait()
 	// The len will not change while we iterate over values, since we waited
 	// for all writes to finish to the cold values and len.
-	n := int(s.len[readIdx].Load())
+	n := int(s.hotColdValMap[readIdx].len.Load())
 	dPts := reset(sData.DataPoints, n, n)
 
 	var i int
-	s.values[readIdx].Range(func(key, value any) bool {
+	s.hotColdValMap[readIdx].values.Range(func(key, value any) bool {
 		val := value.(*sumValue[N])
 		n := val.n.load()
 
@@ -245,8 +256,8 @@ func (s *precomputedSum[N]) delta(
 		i++
 		return true
 	})
-	s.values[readIdx].Clear()
-	s.len[readIdx].Store(0)
+	s.hotColdValMap[readIdx].values.Clear()
+	s.hotColdValMap[readIdx].len.Store(0)
 	s.reported = newReported
 	// The delta collection cycle resets.
 	s.start = t
@@ -272,11 +283,11 @@ func (s *precomputedSum[N]) cumulative(
 	readIdx := s.hcwg.swapHotAndWait()
 	// The len will not change while we iterate over values, since we waited
 	// for all writes to finish to the cold values and len.
-	n := int(s.len[readIdx].Load())
+	n := int(s.hotColdValMap[readIdx].len.Load())
 	dPts := reset(sData.DataPoints, n, n)
 
 	var i int
-	s.values[readIdx].Range(func(_, value any) bool {
+	s.hotColdValMap[readIdx].values.Range(func(_, value any) bool {
 		val := value.(*sumValue[N])
 		collectExemplars(&dPts[i].Exemplars, val.res.Collect)
 		dPts[i].Attributes = val.attrs
@@ -286,8 +297,8 @@ func (s *precomputedSum[N]) cumulative(
 		i++
 		return true
 	})
-	s.values[readIdx].Clear()
-	s.len[readIdx].Store(0)
+	s.hotColdValMap[readIdx].values.Clear()
+	s.hotColdValMap[readIdx].len.Store(0)
 
 	sData.DataPoints = dPts
 	*dest = sData