RWLock for sum and last value aggregations

dashpole · dashpole · commit f4f3a8769e1f · 2025-08-14T02:19:43.000Z
diff --git a/sdk/metric/internal/aggregate/exponential_histogram.go b/sdk/metric/internal/aggregate/exponential_histogram.go
@@ -301,7 +301,7 @@ func newExponentialHistogram[N int64 | float64](
 		maxScale: maxScale,
 
 		newRes: r,
-		limit:  newLimiter[*expoHistogramDataPoint[N]](limit),
+		limit:  newLimiter[expoHistogramDataPoint[N]](limit),
 		values: make(map[attribute.Distinct]*expoHistogramDataPoint[N]),
 
 		start: now(),
@@ -317,7 +317,7 @@ type expoHistogram[N int64 | float64] struct {
 	maxScale int32
 
 	newRes   func(attribute.Set) FilteredExemplarReservoir[N]
-	limit    limiter[*expoHistogramDataPoint[N]]
+	limit    limiter[expoHistogramDataPoint[N]]
 	values   map[attribute.Distinct]*expoHistogramDataPoint[N]
 	valuesMu sync.Mutex
 
diff --git a/sdk/metric/internal/aggregate/filtered_reservoir.go b/sdk/metric/internal/aggregate/filtered_reservoir.go
@@ -5,6 +5,7 @@ package aggregate // import "go.opentelemetry.io/otel/sdk/metric/internal/aggreg
 
 import (
 	"context"
+	"sync"
 	"time"
 
 	"go.opentelemetry.io/otel/attribute"
@@ -27,6 +28,7 @@ type FilteredExemplarReservoir[N int64 | float64] interface {
 
 // filteredExemplarReservoir handles the pre-sampled exemplar of measurements made.
 type filteredExemplarReservoir[N int64 | float64] struct {
+	mu        sync.Mutex
 	filter    exemplar.Filter
 	reservoir exemplar.Reservoir
 }
@@ -45,6 +47,8 @@ func NewFilteredExemplarReservoir[N int64 | float64](
 
 func (f *filteredExemplarReservoir[N]) Offer(ctx context.Context, val N, attr []attribute.KeyValue) {
 	if f.filter(ctx) {
+		f.mu.Lock()
+		defer f.mu.Unlock()
 		// only record the current time if we are sampling this measurement.
 		f.reservoir.Offer(ctx, time.Now(), exemplar.NewValue(val), attr)
 	}
diff --git a/sdk/metric/internal/aggregate/histogram.go b/sdk/metric/internal/aggregate/histogram.go
@@ -48,7 +48,7 @@ type histValues[N int64 | float64] struct {
 	bounds []float64
 
 	newRes   func(attribute.Set) FilteredExemplarReservoir[N]
-	limit    limiter[*buckets[N]]
+	limit    limiter[buckets[N]]
 	values   map[attribute.Distinct]*buckets[N]
 	valuesMu sync.Mutex
 }
@@ -69,7 +69,7 @@ func newHistValues[N int64 | float64](
 		noSum:  noSum,
 		bounds: b,
 		newRes: r,
-		limit:  newLimiter[*buckets[N]](limit),
+		limit:  newLimiter[buckets[N]](limit),
 		values: make(map[attribute.Distinct]*buckets[N]),
 	}
 }
diff --git a/sdk/metric/internal/aggregate/lastvalue.go b/sdk/metric/internal/aggregate/lastvalue.go
@@ -6,6 +6,7 @@ package aggregate // import "go.opentelemetry.io/otel/sdk/metric/internal/aggreg
 import (
 	"context"
 	"sync"
+	"sync/atomic"
 	"time"
 
 	"go.opentelemetry.io/otel/attribute"
@@ -15,44 +16,50 @@ import (
 // datapoint is timestamped measurement data.
 type datapoint[N int64 | float64] struct {
 	attrs attribute.Set
-	value N
+	value atomic.Value
 	res   FilteredExemplarReservoir[N]
 }
 
 func newLastValue[N int64 | float64](limit int, r func(attribute.Set) FilteredExemplarReservoir[N]) *lastValue[N] {
 	return &lastValue[N]{
 		newRes: r,
 		limit:  newLimiter[datapoint[N]](limit),
-		values: make(map[attribute.Distinct]datapoint[N]),
+		values: make(map[attribute.Distinct]*datapoint[N]),
 		start:  now(),
 	}
 }
 
 // lastValue summarizes a set of measurements as the last one made.
 type lastValue[N int64 | float64] struct {
-	sync.Mutex
+	sync.RWMutex
 
 	newRes func(attribute.Set) FilteredExemplarReservoir[N]
 	limit  limiter[datapoint[N]]
-	values map[attribute.Distinct]datapoint[N]
+	values map[attribute.Distinct]*datapoint[N]
 	start  time.Time
 }
 
-func (s *lastValue[N]) measure(ctx context.Context, value N, fltrAttr attribute.Set, droppedAttr []attribute.KeyValue) {
-	s.Lock()
-	defer s.Unlock()
-
+func (s *lastValue[N]) getOrCreateDatapoint(fltrAttr attribute.Set, droppedAttr []attribute.KeyValue) *datapoint[N] {
+	s.RLock()
 	attr := s.limit.Attributes(fltrAttr, s.values)
 	d, ok := s.values[attr.Equivalent()]
+	s.RUnlock()
 	if !ok {
-		d.res = s.newRes(attr)
+		d = &datapoint[N]{
+			res:   s.newRes(attr),
+			attrs: attr,
+		}
+		s.Lock()
+		s.values[attr.Equivalent()] = d
+		s.Unlock()
 	}
+	return d
+}
 
-	d.attrs = attr
-	d.value = value
+func (s *lastValue[N]) measure(ctx context.Context, value N, fltrAttr attribute.Set, droppedAttr []attribute.KeyValue) {
+	d := s.getOrCreateDatapoint(fltrAttr, droppedAttr)
 	d.res.Offer(ctx, value, droppedAttr)
-
-	s.values[attr.Equivalent()] = d
+	d.value.Store(value)
 }
 
 func (s *lastValue[N]) delta(
@@ -109,7 +116,7 @@ func (s *lastValue[N]) copyDpts(dest *[]metricdata.DataPoint[N], t time.Time) in
 		(*dest)[i].Attributes = v.attrs
 		(*dest)[i].StartTime = s.start
 		(*dest)[i].Time = t
-		(*dest)[i].Value = v.value
+		(*dest)[i].Value = v.value.Load().(N)
 		collectExemplars(&(*dest)[i].Exemplars, v.res.Collect)
 		i++
 	}
diff --git a/sdk/metric/internal/aggregate/limit.go b/sdk/metric/internal/aggregate/limit.go
@@ -30,7 +30,7 @@ func newLimiter[V any](aggregation int) limiter[V] {
 // aggregation cardinality limit for the existing measurements. If it will,
 // overflowSet is returned. Otherwise, if it will not exceed the limit, or the
 // limit is not set (limit <= 0), attr is returned.
-func (l limiter[V]) Attributes(attrs attribute.Set, measurements map[attribute.Distinct]V) attribute.Set {
+func (l limiter[V]) Attributes(attrs attribute.Set, measurements map[attribute.Distinct]*V) attribute.Set {
 	if l.aggLimit > 0 {
 		_, exists := measurements[attrs.Equivalent()]
 		if !exists && len(measurements) >= l.aggLimit-1 {
diff --git a/sdk/metric/internal/aggregate/limit_test.go b/sdk/metric/internal/aggregate/limit_test.go
@@ -12,7 +12,8 @@ import (
 )
 
 func TestLimiterAttributes(t *testing.T) {
-	m := map[attribute.Distinct]struct{}{alice.Equivalent(): {}}
+	var val struct{}
+	m := map[attribute.Distinct]*struct{}{alice.Equivalent(): &val}
 	t.Run("NoLimit", func(t *testing.T) {
 		l := newLimiter[struct{}](0)
 		assert.Equal(t, alice, l.Attributes(alice, m))
@@ -43,7 +44,8 @@ func TestLimiterAttributes(t *testing.T) {
 var limitedAttr attribute.Set
 
 func BenchmarkLimiterAttributes(b *testing.B) {
-	m := map[attribute.Distinct]struct{}{alice.Equivalent(): {}}
+	var val struct{}
+	m := map[attribute.Distinct]*struct{}{alice.Equivalent(): &val}
 	l := newLimiter[struct{}](2)
 
 	b.ReportAllocs()
diff --git a/sdk/metric/internal/aggregate/sum.go b/sdk/metric/internal/aggregate/sum.go
@@ -5,50 +5,81 @@ package aggregate // import "go.opentelemetry.io/otel/sdk/metric/internal/aggreg
 
 import (
 	"context"
+	"math"
 	"sync"
+	"sync/atomic"
 	"time"
 
 	"go.opentelemetry.io/otel/attribute"
 	"go.opentelemetry.io/otel/sdk/metric/metricdata"
 )
 
 type sumValue[N int64 | float64] struct {
-	n     N
-	res   FilteredExemplarReservoir[N]
-	attrs attribute.Set
+	nFloatBits uint64
+	nInt       uint64
+	value      atomic.Value // Contains N
+	res        FilteredExemplarReservoir[N]
+	attrs      attribute.Set
+}
+
+func (s *sumValue[N]) val() N {
+	fval := math.Float64frombits(atomic.LoadUint64(&s.nFloatBits))
+	ival := atomic.LoadUint64(&s.nInt)
+	return N(fval + float64(ival))
 }
 
 // valueMap is the storage for sums.
 type valueMap[N int64 | float64] struct {
-	sync.Mutex
+	sync.RWMutex
 	newRes func(attribute.Set) FilteredExemplarReservoir[N]
 	limit  limiter[sumValue[N]]
-	values map[attribute.Distinct]sumValue[N]
+	values map[attribute.Distinct]*sumValue[N]
 }
 
 func newValueMap[N int64 | float64](limit int, r func(attribute.Set) FilteredExemplarReservoir[N]) *valueMap[N] {
 	return &valueMap[N]{
 		newRes: r,
 		limit:  newLimiter[sumValue[N]](limit),
-		values: make(map[attribute.Distinct]sumValue[N]),
+		values: make(map[attribute.Distinct]*sumValue[N]),
 	}
 }
 
-func (s *valueMap[N]) measure(ctx context.Context, value N, fltrAttr attribute.Set, droppedAttr []attribute.KeyValue) {
-	s.Lock()
-	defer s.Unlock()
-
+func (s *valueMap[N]) getOrCreateSumValue(fltrAttr attribute.Set, droppedAttr []attribute.KeyValue) *sumValue[N] {
+	s.RLock()
 	attr := s.limit.Attributes(fltrAttr, s.values)
 	v, ok := s.values[attr.Equivalent()]
+	s.RUnlock()
 	if !ok {
-		v.res = s.newRes(attr)
+		v = &sumValue[N]{
+			res:   s.newRes(attr),
+			attrs: attr,
+		}
+		s.Lock()
+		s.values[attr.Equivalent()] = v
+		s.Unlock()
 	}
+	return v
+}
 
-	v.attrs = attr
-	v.n += value
+func (s *valueMap[N]) measure(ctx context.Context, value N, fltrAttr attribute.Set, droppedAttr []attribute.KeyValue) {
+	v := s.getOrCreateSumValue(fltrAttr, droppedAttr)
 	v.res.Offer(ctx, value, droppedAttr)
 
-	s.values[attr.Equivalent()] = v
+	ival := uint64(value)
+	// This case is where the value is an int, or if it is a whole-numbered float.
+	if float64(ival) == float64(value) {
+		atomic.AddUint64(&v.nInt, ival)
+		return
+	}
+
+	// Value must be a float below.
+	for {
+		oldBits := atomic.LoadUint64(&v.nFloatBits)
+		newBits := math.Float64bits(math.Float64frombits(oldBits) + float64(value))
+		if atomic.CompareAndSwapUint64(&v.nFloatBits, oldBits, newBits) {
+			return
+		}
+	}
 }
 
 // newSum returns an aggregator that summarizes a set of measurements as their
@@ -92,7 +123,7 @@ func (s *sum[N]) delta(
 		dPts[i].Attributes = val.attrs
 		dPts[i].StartTime = s.start
 		dPts[i].Time = t
-		dPts[i].Value = val.n
+		dPts[i].Value = val.val()
 		collectExemplars(&dPts[i].Exemplars, val.res.Collect)
 		i++
 	}
@@ -129,7 +160,7 @@ func (s *sum[N]) cumulative(
 		dPts[i].Attributes = value.attrs
 		dPts[i].StartTime = s.start
 		dPts[i].Time = t
-		dPts[i].Value = value.n
+		dPts[i].Value = value.val()
 		collectExemplars(&dPts[i].Exemplars, value.res.Collect)
 		// TODO (#3006): This will use an unbounded amount of memory if there
 		// are unbounded number of attribute sets being aggregated. Attribute
@@ -189,15 +220,15 @@ func (s *precomputedSum[N]) delta(
 
 	var i int
 	for key, value := range s.values {
-		delta := value.n - s.reported[key]
+		delta := value.val() - s.reported[key]
 
 		dPts[i].Attributes = value.attrs
 		dPts[i].StartTime = s.start
 		dPts[i].Time = t
 		dPts[i].Value = delta
 		collectExemplars(&dPts[i].Exemplars, value.res.Collect)
 
-		newReported[key] = value.n
+		newReported[key] = value.val()
 		i++
 	}
 	// Unused attribute sets do not report.
@@ -234,7 +265,7 @@ func (s *precomputedSum[N]) cumulative(
 		dPts[i].Attributes = val.attrs
 		dPts[i].StartTime = s.start
 		dPts[i].Time = t
-		dPts[i].Value = val.n
+		dPts[i].Value = val.val()
 		collectExemplars(&dPts[i].Exemplars, val.res.Collect)
 
 		i++

Original file line number	Diff line number	Diff line change
`@@ -48,7 +48,7 @@ type histValues[N int64 \| float64] struct {`
`48`	`48`	`bounds []float64`
`49`	`49`
`50`	`50`	`newRes func(attribute.Set) FilteredExemplarReservoir[N]`
`51`		`- limit limiter[*buckets[N]]`
	`51`	`+ limit limiter[buckets[N]]`
`52`	`52`	`values map[attribute.Distinct]*buckets[N]`
`53`	`53`	`valuesMu sync.Mutex`
`54`	`54`	`}`
`@@ -69,7 +69,7 @@ func newHistValues[N int64 \| float64](`
`69`	`69`	`noSum: noSum,`
`70`	`70`	`bounds: b,`
`71`	`71`	`newRes: r,`
`72`		`- limit: newLimiter[*buckets[N]](limit),`
	`72`	`+ limit: newLimiter[buckets[N]](limit),`
`73`	`73`	`values: make(map[attribute.Distinct]*buckets[N]),`
`74`	`74`	`}`
`75`	`75`	`}`