use RWMutex for map access in the metrics SDK

Author: dashpole

CHANGELOG.md

@@ -8,6 +8,10 @@ This project adheres to [Semantic Versioning](https://semver.org/spec/v2.0.0.htm
 
 ## [Unreleased]
 
+### Changed
+
+- Improve performance of concurrent measurements in `go.opentelemetry.io/otel/sdk/metric`. (#7189)
+
 ### Removed
 
 - Drop support for [Go 1.23]. (#7274)

sdk/metric/internal/aggregate/atomic.go

@@ -0,0 +1,43 @@
+// Copyright The OpenTelemetry Authors
+// SPDX-License-Identifier: Apache-2.0
+
+package aggregate // import "go.opentelemetry.io/otel/sdk/metric/internal/aggregate"
+
+import (
+	"math"
+	"sync/atomic"
+)
+
+// counter is an efficient way of adding to a number which is either an
+// int64 or float64.
+type counter[N int64 | float64] struct {
+	// nFloatBits contains only the non-integer portion of the counter.
+	nFloatBits uint64
+	// nInt contains only the integer portion of the counter.
+	nInt uint64
+}
+
+// value returns the float or integer value.
+func (n *counter[N]) value() N {
+	fval := math.Float64frombits(atomic.LoadUint64(&n.nFloatBits))
+	ival := atomic.LoadUint64(&n.nInt)
+	return N(fval + float64(ival))
+}
+
+func (n *counter[N]) add(value N) {
+	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(&n.nInt, ival)
+		return
+	}
+
+	// Value must be a float below.
+	for {
+		oldBits := atomic.LoadUint64(&n.nFloatBits)
+		newBits := math.Float64bits(math.Float64frombits(oldBits) + float64(value))
+		if atomic.CompareAndSwapUint64(&n.nFloatBits, oldBits, newBits) {
+			return
+		}
+	}
+}

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
 

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,9 +47,15 @@ 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)
 	}
 }
 
-func (f *filteredExemplarReservoir[N]) Collect(dest *[]exemplar.Exemplar) { f.reservoir.Collect(dest) }
+func (f *filteredExemplarReservoir[N]) Collect(dest *[]exemplar.Exemplar) {
+	f.mu.Lock()
+	defer f.mu.Unlock()
+	f.reservoir.Collect(dest)
+}

sdk/metric/internal/aggregate/histogram.go

@@ -8,54 +8,66 @@ import (
 	"slices"
 	"sort"
 	"sync"
+	"sync/atomic"
 	"time"
 
 	"go.opentelemetry.io/otel/attribute"
 	"go.opentelemetry.io/otel/sdk/metric/metricdata"
 )
 
 type buckets[N int64 | float64] struct {
+	count    uint64
+	counts   []uint64
+	min, max atomic.Value
+	total    *counter[N]
+
 	attrs attribute.Set
 	res   FilteredExemplarReservoir[N]
-
-	counts   []uint64
-	count    uint64
-	total    N
-	min, max N
 }
 
 // newBuckets returns buckets with n bins.
 func newBuckets[N int64 | float64](attrs attribute.Set, n int) *buckets[N] {
-	return &buckets[N]{attrs: attrs, counts: make([]uint64, n)}
+	return &buckets[N]{attrs: attrs, counts: make([]uint64, n), total: &counter[N]{}}
 }
 
-func (b *buckets[N]) sum(value N) { b.total += value }
+func (b *buckets[N]) bin(idx int) {
+	atomic.AddUint64(&b.counts[idx], 1)
+	atomic.AddUint64(&b.count, 1)
+}
 
-func (b *buckets[N]) bin(idx int, value N) {
-	b.counts[idx]++
-	b.count++
-	if value < b.min {
-		b.min = value
-	} else if value > b.max {
-		b.max = value
+func (b *buckets[N]) minMax(value N) {
+	for {
+		minLoaded := b.min.Load()
+		if value < minLoaded.(N) && !b.min.CompareAndSwap(minLoaded, value) {
+			// We got a new min value, but lost the race. Try again.
+			continue
+		}
+		maxLoaded := b.max.Load()
+		if value > maxLoaded.(N) && !b.max.CompareAndSwap(maxLoaded, value) {
+			// We got a new max value, but lost the race. Try again.
+			continue
+		}
+		return
 	}
 }
 
 // histValues summarizes a set of measurements as an histValues with
 // explicitly defined buckets.
 type histValues[N int64 | float64] struct {
-	noSum  bool
-	bounds []float64
+	noSum    bool
+	noMinMax bool
+	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
+	valuesMu sync.RWMutex
 }
 
 func newHistValues[N int64 | float64](
 	bounds []float64,
 	noSum bool,
+	noMinMax bool,
 	limit int,
 	r func(attribute.Set) FilteredExemplarReservoir[N],
 ) *histValues[N] {
@@ -66,11 +78,12 @@ func newHistValues[N int64 | float64](
 	b := slices.Clone(bounds)
 	slices.Sort(b)
 	return &histValues[N]{
-		noSum:  noSum,
-		bounds: b,
-		newRes: r,
-		limit:  newLimiter[*buckets[N]](limit),
-		values: make(map[attribute.Distinct]*buckets[N]),
+		noSum:    noSum,
+		noMinMax: noMinMax,
+		bounds:   b,
+		newRes:   r,
+		limit:    newLimiter[buckets[N]](limit),
+		values:   make(map[attribute.Distinct]*buckets[N]),
 	}
 }
 
@@ -89,11 +102,11 @@ func (s *histValues[N]) measure(
 	// (s.bounds[len(s.bounds)-1], +∞).
 	idx := sort.SearchFloat64s(s.bounds, float64(value))
 
-	s.valuesMu.Lock()
-	defer s.valuesMu.Unlock()
+	s.valuesMu.RLock()
 
 	attr := s.limit.Attributes(fltrAttr, s.values)
 	b, ok := s.values[attr.Equivalent()]
+	s.valuesMu.RUnlock()
 	if !ok {
 		// N+1 buckets. For example:
 		//
@@ -106,12 +119,20 @@ func (s *histValues[N]) measure(
 		b.res = s.newRes(attr)
 
 		// Ensure min and max are recorded values (not zero), for new buckets.
-		b.min, b.max = value, value
+		if !s.noMinMax {
+			b.min.Store(value)
+			b.max.Store(value)
+		}
+		s.valuesMu.Lock()
 		s.values[attr.Equivalent()] = b
+		s.valuesMu.Unlock()
+	}
+	b.bin(idx)
+	if !s.noMinMax {
+		b.minMax(value)
 	}
-	b.bin(idx, value)
 	if !s.noSum {
-		b.sum(value)
+		b.total.add(value)
 	}
 	b.res.Offer(ctx, value, droppedAttr)
 }
@@ -125,8 +146,7 @@ func newHistogram[N int64 | float64](
 	r func(attribute.Set) FilteredExemplarReservoir[N],
 ) *histogram[N] {
 	return &histogram[N]{
-		histValues: newHistValues[N](boundaries, noSum, limit, r),
-		noMinMax:   noMinMax,
+		histValues: newHistValues[N](boundaries, noSum, noMinMax, limit, r),
 		start:      now(),
 	}
 }
@@ -136,8 +156,7 @@ func newHistogram[N int64 | float64](
 type histogram[N int64 | float64] struct {
 	*histValues[N]
 
-	noMinMax bool
-	start    time.Time
+	start time.Time
 }
 
 func (s *histogram[N]) delta(
@@ -169,12 +188,12 @@ func (s *histogram[N]) delta(
 		hDPts[i].BucketCounts = val.counts
 
 		if !s.noSum {
-			hDPts[i].Sum = val.total
+			hDPts[i].Sum = val.total.value()
 		}
 
 		if !s.noMinMax {
-			hDPts[i].Min = metricdata.NewExtrema(val.min)
-			hDPts[i].Max = metricdata.NewExtrema(val.max)
+			hDPts[i].Min = metricdata.NewExtrema(val.min.Load().(N))
+			hDPts[i].Max = metricdata.NewExtrema(val.max.Load().(N))
 		}
 
 		collectExemplars(&hDPts[i].Exemplars, val.res.Collect)
@@ -227,12 +246,12 @@ func (s *histogram[N]) cumulative(
 		hDPts[i].BucketCounts = slices.Clone(val.counts)
 
 		if !s.noSum {
-			hDPts[i].Sum = val.total
+			hDPts[i].Sum = val.total.value()
 		}
 
 		if !s.noMinMax {
-			hDPts[i].Min = metricdata.NewExtrema(val.min)
-			hDPts[i].Max = metricdata.NewExtrema(val.max)
+			hDPts[i].Min = metricdata.NewExtrema(val.min.Load().(N))
+			hDPts[i].Max = metricdata.NewExtrema(val.max.Load().(N))
 		}
 
 		collectExemplars(&hDPts[i].Exemplars, val.res.Collect)

sdk/metric/internal/aggregate/histogram_test.go

@@ -278,18 +278,22 @@ func TestBucketsBin(t *testing.T) {
 func testBucketsBin[N int64 | float64]() func(t *testing.T) {
 	return func(t *testing.T) {
 		b := newBuckets[N](alice, 3)
+		b.min.Store(N(0))
+		b.max.Store(N(0))
 		assertB := func(counts []uint64, count uint64, mi, ma N) {
 			t.Helper()
 			assert.Equal(t, counts, b.counts)
 			assert.Equal(t, count, b.count)
-			assert.Equal(t, mi, b.min)
-			assert.Equal(t, ma, b.max)
+			assert.Equal(t, mi, b.min.Load().(N))
+			assert.Equal(t, ma, b.max.Load().(N))
 		}
 
 		assertB([]uint64{0, 0, 0}, 0, 0, 0)
-		b.bin(1, 2)
+		b.bin(1)
+		b.minMax(2)
 		assertB([]uint64{0, 1, 0}, 1, 0, 2)
-		b.bin(0, -1)
+		b.bin(0)
+		b.minMax(-1)
 		assertB([]uint64{1, 1, 0}, 2, -1, 2)
 	}
 }
@@ -304,15 +308,15 @@ func testBucketsSum[N int64 | float64]() func(t *testing.T) {
 		b := newBuckets[N](alice, 3)
 
 		var want N
-		assert.Equal(t, want, b.total)
+		assert.Equal(t, want, b.total.value())
 
-		b.sum(2)
+		b.total.add(2)
 		want = 2
-		assert.Equal(t, want, b.total)
+		assert.Equal(t, want, b.total.value())
 
-		b.sum(-1)
+		b.total.add(-1)
 		want = 1
-		assert.Equal(t, want, b.total)
+		assert.Equal(t, want, b.total.value())
 	}
 }