rawvalue: fix thread safety and sum

Author: elek

go.mod

@@ -1,3 +1,7 @@
 module github.com/spacemonkeygo/monkit/v3
 
-go 1.19
+go 1.23.0
+
+toolchain go1.24.3
+
+require golang.org/x/sync v0.14.0

go.sum

@@ -0,0 +1,2 @@
+golang.org/x/sync v0.14.0 h1:woo0S4Yywslg6hp4eUFjTVOyKt0RookbpAHG4c1HmhQ=
+golang.org/x/sync v0.14.0/go.mod h1:1dzgHSNfp02xaA81J2MS99Qcpr2w7fw1gpm99rleRqA=

val.go

@@ -291,12 +291,28 @@ func (v *RawVal) Observe(val float64) {
 
 // Stats implements the StatSource interface.
 func (v *RawVal) Stats(cb func(key SeriesKey, field string, val float64)) {
+	type datum struct {
+		field string
+		value float64
+	}
+	var data []datum
+	if len(v.stats) < 8 {
+		var buf [8]datum
+		data = buf[:len(v.stats)]
+	} else {
+		data = make([]datum, len(v.stats))
+	}
 	v.mtx.Lock()
-	cb(v.key, "recent", v.value)
-	v.mtx.Unlock()
-	for _, s := range v.stats {
+	for i, s := range v.stats {
 		field, value := s()
-		cb(v.key, field, value)
+		data[i] = datum{field, value}
+	}
+	recent := v.value
+	v.mtx.Unlock()
+
+	cb(v.key, "recent", recent)
+	for _, d := range data {
+		cb(v.key, d.field, d.value)
 	}
 }
 
@@ -320,10 +336,10 @@ func Count() (observe func(val float64), stat func() (field string, val float64)
 
 // Sum is a value aggregator that summarizes the values measured.
 func Sum() (observe func(val float64), stat func() (field string, val float64)) {
-	var sum int
+	var sum float64
 	return func(val float64) {
-			sum += int(val)
+			sum += val
 		}, func() (field string, val float64) {
-			return "sum", float64(sum)
+			return "sum", sum
 		}
 }

val_test.go

@@ -0,0 +1,37 @@
+// Copyright (C) 2015 Space Monkey, Inc.
+//
+// Licensed under the Apache License, Version 2.0 (the "License");
+// you may not use this file except in compliance with the License.
+// You may obtain a copy of the License at
+//
+//   http://www.apache.org/licenses/LICENSE-2.0
+//
+// Unless required by applicable law or agreed to in writing, software
+// distributed under the License is distributed on an "AS IS" BASIS,
+// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+// See the License for the specific language governing permissions and
+// limitations under the License.
+
+package monkit
+
+import (
+	"golang.org/x/sync/errgroup"
+	"testing"
+)
+
+func TestRawValConcurrentSafe(t *testing.T) {
+	rv := NewRawVal(NewSeriesKey("test"), Sum, Count)
+
+	var group errgroup.Group
+	defer func() { _ = group.Wait() }()
+	for i := 0; i < 10; i++ {
+		group.Go(func() error {
+			if i%2 == 0 {
+				rv.Observe(1.0)
+			} else {
+				rv.Stats(func(key SeriesKey, field string, val float64) {})
+			}
+			return nil
+		})
+	}
+}