use RWMutex for map access in the metrics SDK

Author: dashpole

CHANGELOG.md

@@ -29,6 +29,7 @@ This project adheres to [Semantic Versioning](https://semver.org/spec/v2.0.0.htm
 - `WithInstrumentationAttributes` in `go.opentelemetry.io/otel/trace` synchronously de-duplicates the passed attributes instead of delegating it to the returned `TracerOption`. (#7266)
 - `WithInstrumentationAttributes` in `go.opentelemetry.io/otel/meter` synchronously de-duplicates the passed attributes instead of delegating it to the returned `MeterOption`. (#7266)
 - `WithInstrumentationAttributes` in `go.opentelemetry.io/otel/log` synchronously de-duplicates the passed attributes instead of delegating it to the returned `LoggerOption`. (#7266)
+- Improve performance of concurrent measurements in `go.opentelemetry.io/otel/sdk/metric`. (#7189)
 
 <!-- Released section -->
 <!-- Don't change this section unless doing release -->

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,58 +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) {
-	b.counts[idx]++
-	b.count++
+	atomic.AddUint64(&b.counts[idx], 1)
+	atomic.AddUint64(&b.count, 1)
 }
 
 func (b *buckets[N]) minMax(value N) {
-	if value < b.min {
-		b.min = value
-	} else if value > b.max {
-		b.max = value
+	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 {
-	noMinMax bool
 	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,
-	noMinMax, noSum bool,
+	noSum bool,
+	noMinMax bool,
 	limit int,
 	r func(attribute.Set) FilteredExemplarReservoir[N],
 ) *histValues[N] {
@@ -70,11 +78,11 @@ func newHistValues[N int64 | float64](
 	b := slices.Clone(bounds)
 	slices.Sort(b)
 	return &histValues[N]{
-		noMinMax: noMinMax,
 		noSum:    noSum,
+		noMinMax: noMinMax,
 		bounds:   b,
 		newRes:   r,
-		limit:    newLimiter[*buckets[N]](limit),
+		limit:    newLimiter[buckets[N]](limit),
 		values:   make(map[attribute.Distinct]*buckets[N]),
 	}
 }
@@ -94,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:
 		//
@@ -111,15 +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)
 	}
 	if !s.noSum {
-		b.sum(value)
+		b.total.add(value)
 	}
 	b.res.Offer(ctx, value, droppedAttr)
 }
@@ -133,7 +146,7 @@ func newHistogram[N int64 | float64](
 	r func(attribute.Set) FilteredExemplarReservoir[N],
 ) *histogram[N] {
 	return &histogram[N]{
-		histValues: newHistValues[N](boundaries, noMinMax, noSum, limit, r),
+		histValues: newHistValues[N](boundaries, noSum, noMinMax, limit, r),
 		start:      now(),
 	}
 }
@@ -175,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)
@@ -233,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,12 +278,14 @@ 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)
@@ -306,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())
 	}
 }
 

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"
@@ -14,45 +15,46 @@ import (
 
 // datapoint is timestamped measurement data.
 type datapoint[N int64 | float64] struct {
+	value atomic.Value
 	attrs attribute.Set
-	value N
 	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()
-
+	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()
 	}
-
-	d.attrs = attr
-	d.value = value
 	d.res.Offer(ctx, value, droppedAttr)
-
-	s.values[attr.Equivalent()] = d
+	d.value.Store(value)
 }
 
 func (s *lastValue[N]) delta(
@@ -109,7 +111,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++
 	}