Fix data race in Meter::Reset

meter.go

@@ -59,7 +59,7 @@ func (m *Meter) Reset() {
 	defer globalSweeper.snapshotMu.Unlock()
 	atomic.StoreUint64(&m.accumulator, 0)
 	m.snapshot.Rate = 0
-	m.snapshot.Total = 0
+	atomic.StoreUint64(&m.snapshot.Total, 0)
 	m.snapshot.LastUpdate = time.Now()
 }
 

meter_test.go

@@ -3,6 +3,7 @@ package flow
 import (
 	"fmt"
 	"math"
+	"sync"
 	"testing"
 	"time"
 )
@@ -48,6 +49,25 @@ func TestResetMeter(t *testing.T) {
 	}
 }
 
+func TestMarkResetMeterMulti(t *testing.T) {
+	var wg sync.WaitGroup
+	wg.Add(2)
+
+	meter := new(Meter)
+	go func(meter *Meter) {
+		meter.Mark(30)
+		meter.Mark(30)
+		wg.Done()
+	}(meter)
+
+	go func(meter *Meter) {
+		meter.Reset()
+		wg.Done()
+	}(meter)
+
+	wg.Wait()
+}
+
 func roundTens(x float64) int64 {
 	return int64(math.Floor(x/10+0.5)) * 10
 }

sweeper.go

@@ -45,7 +45,7 @@ func (sw *sweeper) run() {
 func (sw *sweeper) register(m *Meter) {
 	// Add back the snapshot total. If we unregistered this
 	// one, we set it to zero.
-	atomic.AddUint64(&m.accumulator, m.snapshot.Total)
+	atomic.AddUint64(&m.accumulator, atomic.LoadUint64(&m.snapshot.Total))
 	sw.meters = append(sw.meters, m)
 }