metrics: remove Stop() and arbiter from metrics

metrics/meter.go

@@ -2,7 +2,6 @@ package metrics
 
 import (
 	"sync"
-	"sync/atomic"
 	"time"
 )
 
@@ -16,13 +15,10 @@ type Meter interface {
 	Rate15() float64
 	RateMean() float64
 	Snapshot() Meter
-	Stop()
 }
 
 // GetOrRegisterMeter returns an existing Meter or constructs and registers a
 // new StandardMeter.
-// Be sure to unregister the meter from the registry once it is of no use to
-// allow for garbage collection.
 func GetOrRegisterMeter(name string, r Registry) Meter {
 	if nil == r {
 		r = DefaultRegistry
@@ -32,8 +28,6 @@ func GetOrRegisterMeter(name string, r Registry) Meter {
 
 // GetOrRegisterMeterForced returns an existing Meter or constructs and registers a
 // new StandardMeter no matter the global switch is enabled or not.
-// Be sure to unregister the meter from the registry once it is of no use to
-// allow for garbage collection.
 func GetOrRegisterMeterForced(name string, r Registry) Meter {
 	if nil == r {
 		r = DefaultRegistry
@@ -42,41 +36,21 @@ func GetOrRegisterMeterForced(name string, r Registry) Meter {
 }
 
 // NewMeter constructs a new StandardMeter and launches a goroutine.
-// Be sure to call Stop() once the meter is of no use to allow for garbage collection.
 func NewMeter() Meter {
 	if !Enabled {
 		return NilMeter{}
 	}
-	m := newStandardMeter()
-	arbiter.Lock()
-	defer arbiter.Unlock()
-	arbiter.meters[m] = struct{}{}
-	if !arbiter.started {
-		arbiter.started = true
-		go arbiter.tick()
-	}
-	return m
+	return newStandardMeter()
 }
 
 // NewMeterForced constructs a new StandardMeter and launches a goroutine no matter
 // the global switch is enabled or not.
-// Be sure to call Stop() once the meter is of no use to allow for garbage collection.
 func NewMeterForced() Meter {
-	m := newStandardMeter()
-	arbiter.Lock()
-	defer arbiter.Unlock()
-	arbiter.meters[m] = struct{}{}
-	if !arbiter.started {
-		arbiter.started = true
-		go arbiter.tick()
-	}
-	return m
+	return newStandardMeter()
 }
 
 // NewRegisteredMeter constructs and registers a new StandardMeter
 // and launches a goroutine.
-// Be sure to unregister the meter from the registry once it is of no use to
-// allow for garbage collection.
 func NewRegisteredMeter(name string, r Registry) Meter {
 	c := NewMeter()
 	if nil == r {
@@ -88,8 +62,6 @@ func NewRegisteredMeter(name string, r Registry) Meter {
 
 // NewRegisteredMeterForced constructs and registers a new StandardMeter
 // and launches a goroutine no matter the global switch is enabled or not.
-// Be sure to unregister the meter from the registry once it is of no use to
-// allow for garbage collection.
 func NewRegisteredMeterForced(name string, r Registry) Meter {
 	c := NewMeterForced()
 	if nil == r {
@@ -101,7 +73,6 @@ func NewRegisteredMeterForced(name string, r Registry) Meter {
 
 // MeterSnapshot is a read-only copy of another Meter.
 type MeterSnapshot struct {
-	temp                           atomic.Int64
 	count                          int64
 	rate1, rate5, rate15, rateMean float64
 }
@@ -133,9 +104,6 @@ func (m *MeterSnapshot) RateMean() float64 { return m.rateMean }
 // Snapshot returns the snapshot.
 func (m *MeterSnapshot) Snapshot() Meter { return m }
 
-// Stop is a no-op.
-func (m *MeterSnapshot) Stop() {}
-
 // NilMeter is a no-op Meter.
 type NilMeter struct{}
 
@@ -160,21 +128,16 @@ func (NilMeter) RateMean() float64 { return 0.0 }
 // Snapshot is a no-op.
 func (NilMeter) Snapshot() Meter { return NilMeter{} }
 
-// Stop is a no-op.
-func (NilMeter) Stop() {}
-
 // StandardMeter is the standard implementation of a Meter.
 type StandardMeter struct {
 	lock        sync.RWMutex
-	snapshot    *MeterSnapshot
+	count       int64
 	a1, a5, a15 EWMA
 	startTime   time.Time
-	stopped     atomic.Bool
 }
 
 func newStandardMeter() *StandardMeter {
 	return &StandardMeter{
-		snapshot:  &MeterSnapshot{},
 		a1:        NewEWMA1(),
 		a5:        NewEWMA5(),
 		a15:       NewEWMA15(),
@@ -182,123 +145,53 @@ func newStandardMeter() *StandardMeter {
 	}
 }
 
-// Stop stops the meter, Mark() will be a no-op if you use it after being stopped.
-func (m *StandardMeter) Stop() {
-	stopped := m.stopped.Swap(true)
-	if !stopped {
-		arbiter.Lock()
-		delete(arbiter.meters, m)
-		arbiter.Unlock()
-	}
-}
-
 // Count returns the number of events recorded.
 // It updates the meter to be as accurate as possible
 func (m *StandardMeter) Count() int64 {
-	m.lock.Lock()
-	defer m.lock.Unlock()
-	m.updateMeter()
-	return m.snapshot.count
+	m.lock.RLock()
+	defer m.lock.RUnlock()
+	return m.count
 }
 
 // Mark records the occurrence of n events.
 func (m *StandardMeter) Mark(n int64) {
-	m.snapshot.temp.Add(n)
+	m.lock.Lock()
+	defer m.lock.Unlock()
+	m.count += n
+	m.a1.Update(n)
+	m.a5.Update(n)
+	m.a15.Update(n)
 }
 
 // Rate1 returns the one-minute moving average rate of events per second.
 func (m *StandardMeter) Rate1() float64 {
-	m.lock.RLock()
-	defer m.lock.RUnlock()
-	return m.snapshot.rate1
+	return m.a1.Rate()
 }
 
 // Rate5 returns the five-minute moving average rate of events per second.
 func (m *StandardMeter) Rate5() float64 {
-	m.lock.RLock()
-	defer m.lock.RUnlock()
-	return m.snapshot.rate5
+	return m.a5.Rate()
 }
 
 // Rate15 returns the fifteen-minute moving average rate of events per second.
 func (m *StandardMeter) Rate15() float64 {
-	m.lock.RLock()
-	defer m.lock.RUnlock()
-	return m.snapshot.rate15
+	return m.a15.Rate()
 }
 
 // RateMean returns the meter's mean rate of events per second.
 func (m *StandardMeter) RateMean() float64 {
 	m.lock.RLock()
 	defer m.lock.RUnlock()
-	return m.snapshot.rateMean
+	return float64(m.count) / (1 + time.Since(m.startTime).Seconds())
 }
 
 // Snapshot returns a read-only copy of the meter.
 func (m *StandardMeter) Snapshot() Meter {
-	m.lock.RLock()
-	snapshot := MeterSnapshot{
-		count:    m.snapshot.count,
-		rate1:    m.snapshot.rate1,
-		rate5:    m.snapshot.rate5,
-		rate15:   m.snapshot.rate15,
-		rateMean: m.snapshot.rateMean,
-	}
-	snapshot.temp.Store(m.snapshot.temp.Load())
-	m.lock.RUnlock()
-	return &snapshot
-}
-
-func (m *StandardMeter) updateSnapshot() {
-	// should run with write lock held on m.lock
-	snapshot := m.snapshot
-	snapshot.rate1 = m.a1.Rate()
-	snapshot.rate5 = m.a5.Rate()
-	snapshot.rate15 = m.a15.Rate()
-	snapshot.rateMean = float64(snapshot.count) / time.Since(m.startTime).Seconds()
-}
-
-func (m *StandardMeter) updateMeter() {
-	// should only run with write lock held on m.lock
-	n := m.snapshot.temp.Swap(0)
-	m.snapshot.count += n
-	m.a1.Update(n)
-	m.a5.Update(n)
-	m.a15.Update(n)
-}
-
-func (m *StandardMeter) tick() {
-	m.lock.Lock()
-	defer m.lock.Unlock()
-	m.updateMeter()
-	m.a1.Tick()
-	m.a5.Tick()
-	m.a15.Tick()
-	m.updateSnapshot()
-}
-
-// meterArbiter ticks meters every 5s from a single goroutine.
-// meters are references in a set for future stopping.
-type meterArbiter struct {
-	sync.RWMutex
-	started bool
-	meters  map[*StandardMeter]struct{}
-	ticker  *time.Ticker
-}
-
-var arbiter = meterArbiter{ticker: time.NewTicker(5 * time.Second), meters: make(map[*StandardMeter]struct{})}
-
-// Ticks meters on the scheduled interval
-func (ma *meterArbiter) tick() {
-	for range ma.ticker.C {
-		ma.tickMeters()
-	}
-}
-
-func (ma *meterArbiter) tickMeters() {
-	ma.RLock()
-	defer ma.RUnlock()
-	for meter := range ma.meters {
-		meter.tick()
+	return &MeterSnapshot{
+		count:    m.Count(),
+		rate1:    m.Rate1(),
+		rate5:    m.Rate5(),
+		rate15:   m.Rate15(),
+		rateMean: m.RateMean(),
 	}
 }

metrics/meter_test.go

@@ -1,6 +1,7 @@
 package metrics
 
 import (
+	"math/rand"
 	"testing"
 	"time"
 )
@@ -22,18 +23,10 @@ func TestGetOrRegisterMeter(t *testing.T) {
 }
 
 func TestMeterDecay(t *testing.T) {
-	ma := meterArbiter{
-		ticker: time.NewTicker(time.Millisecond),
-		meters: make(map[*StandardMeter]struct{}),
-	}
-	defer ma.ticker.Stop()
 	m := newStandardMeter()
-	ma.meters[m] = struct{}{}
 	m.Mark(1)
-	ma.tickMeters()
 	rateMean := m.RateMean()
 	time.Sleep(100 * time.Millisecond)
-	ma.tickMeters()
 	if m.RateMean() >= rateMean {
 		t.Error("m.RateMean() didn't decrease")
 	}
@@ -47,22 +40,13 @@ func TestMeterNonzero(t *testing.T) {
 	}
 }
 
-func TestMeterStop(t *testing.T) {
-	l := len(arbiter.meters)
-	m := NewMeter()
-	if l+1 != len(arbiter.meters) {
-		t.Errorf("arbiter.meters: %d != %d\n", l+1, len(arbiter.meters))
-	}
-	m.Stop()
-	if l != len(arbiter.meters) {
-		t.Errorf("arbiter.meters: %d != %d\n", l, len(arbiter.meters))
-	}
-}
-
 func TestMeterSnapshot(t *testing.T) {
+	r := rand.New(rand.NewSource(time.Now().Unix()))
 	m := NewMeter()
-	m.Mark(1)
-	if snapshot := m.Snapshot(); m.RateMean() != snapshot.RateMean() {
+	m.Mark(r.Int63())
+
+	// RateMean() updates every millisecond, so we test Count().
+	if snapshot := m.Snapshot(); m.Count() != snapshot.Count() {
 		t.Fatal(snapshot)
 	}
 }

metrics/prometheus/collector_test.go

@@ -36,12 +36,10 @@ func TestCollector(t *testing.T) {
 	c.addHistogram("test/histogram", histogram)
 
 	meter := metrics.NewMeter()
-	defer meter.Stop()
 	meter.Mark(9999999)
 	c.addMeter("test/meter", meter)
 
 	timer := metrics.NewTimer()
-	defer timer.Stop()
 	timer.Update(20 * time.Millisecond)
 	timer.Update(21 * time.Millisecond)
 	timer.Update(22 * time.Millisecond)

metrics/registry_test.go

@@ -145,19 +145,27 @@ func TestRegistryGetOrRegisterWithLazyInstantiation(t *testing.T) {
 }
 
 func TestRegistryUnregister(t *testing.T) {
-	l := len(arbiter.meters)
+	count := 0
+	counter := func(string, interface{}) {
+		count++
+	}
+
 	r := NewRegistry()
 	r.Register("foo", NewCounter())
 	r.Register("bar", NewMeter())
 	r.Register("baz", NewTimer())
-	if len(arbiter.meters) != l+2 {
-		t.Errorf("arbiter.meters: %d != %d\n", l+2, len(arbiter.meters))
+	r.Each(counter)
+	if count != 3 {
+		t.Errorf("Register() count: %d != %d\n", 3, count)
 	}
+
 	r.Unregister("foo")
 	r.Unregister("bar")
 	r.Unregister("baz")
-	if len(arbiter.meters) != l {
-		t.Errorf("arbiter.meters: %d != %d\n", l+2, len(arbiter.meters))
+	count = 0
+	r.Each(counter)
+	if count != 0 {
+		t.Errorf("Unregister() count: %d != %d\n", 0, count)
 	}
 }
 

metrics/sample_test.go

@@ -296,6 +296,8 @@ func testExpDecaySampleStatistics(t *testing.T, s Sample) {
 }
 
 func testUniformSampleStatistics(t *testing.T, s Sample) {
+	epsilonPercentile := .00000000001
+
 	if count := s.Count(); count != 10000 {
 		t.Errorf("s.Count(): 10000 != %v\n", count)
 	}

metrics/timer.go

@@ -19,7 +19,6 @@ type Timer interface {
 	RateMean() float64
 	Snapshot() Timer
 	StdDev() float64
-	Stop()
 	Sum() int64
 	Time(func())
 	Update(time.Duration)
@@ -29,8 +28,6 @@ type Timer interface {
 
 // GetOrRegisterTimer returns an existing Timer or constructs and registers a
 // new StandardTimer.
-// Be sure to unregister the meter from the registry once it is of no use to
-// allow for garbage collection.
 func GetOrRegisterTimer(name string, r Registry) Timer {
 	if nil == r {
 		r = DefaultRegistry
@@ -39,7 +36,6 @@ func GetOrRegisterTimer(name string, r Registry) Timer {
 }
 
 // NewCustomTimer constructs a new StandardTimer from a Histogram and a Meter.
-// Be sure to call Stop() once the timer is of no use to allow for garbage collection.
 func NewCustomTimer(h Histogram, m Meter) Timer {
 	if !Enabled {
 		return NilTimer{}
@@ -51,8 +47,6 @@ func NewCustomTimer(h Histogram, m Meter) Timer {
 }
 
 // NewRegisteredTimer constructs and registers a new StandardTimer.
-// Be sure to unregister the meter from the registry once it is of no use to
-// allow for garbage collection.
 func NewRegisteredTimer(name string, r Registry) Timer {
 	c := NewTimer()
 	if nil == r {
@@ -64,7 +58,6 @@ func NewRegisteredTimer(name string, r Registry) Timer {
 
 // NewTimer constructs a new StandardTimer using an exponentially-decaying
 // sample with the same reservoir size and alpha as UNIX load averages.
-// Be sure to call Stop() once the timer is of no use to allow for garbage collection.
 func NewTimer() Timer {
 	if !Enabled {
 		return NilTimer{}
@@ -116,9 +109,6 @@ func (NilTimer) Snapshot() Timer { return NilTimer{} }
 // StdDev is a no-op.
 func (NilTimer) StdDev() float64 { return 0.0 }
 
-// Stop is a no-op.
-func (NilTimer) Stop() {}
-
 // Sum is a no-op.
 func (NilTimer) Sum() int64 { return 0 }
 
@@ -208,11 +198,6 @@ func (t *StandardTimer) StdDev() float64 {
 	return t.histogram.StdDev()
 }
 
-// Stop stops the meter.
-func (t *StandardTimer) Stop() {
-	t.meter.Stop()
-}
-
 // Sum returns the sum in the sample.
 func (t *StandardTimer) Sum() int64 {
 	return t.histogram.Sum()
@@ -300,9 +285,6 @@ func (t *TimerSnapshot) Snapshot() Timer { return t }
 // was taken.
 func (t *TimerSnapshot) StdDev() float64 { return t.histogram.StdDev() }
 
-// Stop is a no-op.
-func (t *TimerSnapshot) Stop() {}
-
 // Sum returns the sum at the time the snapshot was taken.
 func (t *TimerSnapshot) Sum() int64 { return t.histogram.Sum() }
 

metrics/timer_test.go

@@ -32,18 +32,6 @@ func TestTimerExtremes(t *testing.T) {
 	}
 }
 
-func TestTimerStop(t *testing.T) {
-	l := len(arbiter.meters)
-	tm := NewTimer()
-	if l+1 != len(arbiter.meters) {
-		t.Errorf("arbiter.meters: %d != %d\n", l+1, len(arbiter.meters))
-	}
-	tm.Stop()
-	if l != len(arbiter.meters) {
-		t.Errorf("arbiter.meters: %d != %d\n", l, len(arbiter.meters))
-	}
-}
-
 func TestTimerFunc(t *testing.T) {
 	var (
 		tm         = NewTimer()