gf/third/github.com/rcrowley/go-metrics/ewma.go

139 lines
3.8 KiB
Go
Raw Normal View History

package metrics
import (
"math"
"sync"
"sync/atomic"
)
// EWMAs continuously calculate an exponentially-weighted moving average
// based on an outside source of clock ticks.
type EWMA interface {
Rate() float64
Snapshot() EWMA
Tick()
Update(int64)
}
// NewEWMA constructs a new EWMA with the given alpha.
func NewEWMA(alpha float64) EWMA {
if UseNilMetrics {
return NilEWMA{}
}
return &StandardEWMA{alpha: alpha}
}
// NewEWMA1 constructs a new EWMA for a one-minute moving average.
func NewEWMA1() EWMA {
return NewEWMA(1 - math.Exp(-5.0/60.0/1))
}
// NewEWMA5 constructs a new EWMA for a five-minute moving average.
func NewEWMA5() EWMA {
return NewEWMA(1 - math.Exp(-5.0/60.0/5))
}
// NewEWMA15 constructs a new EWMA for a fifteen-minute moving average.
func NewEWMA15() EWMA {
return NewEWMA(1 - math.Exp(-5.0/60.0/15))
}
// EWMASnapshot is a read-only copy of another EWMA.
type EWMASnapshot float64
// Rate returns the rate of events per second at the time the snapshot was
// taken.
func (a EWMASnapshot) Rate() float64 { return float64(a) }
// Snapshot returns the snapshot.
func (a EWMASnapshot) Snapshot() EWMA { return a }
// Tick panics.
func (EWMASnapshot) Tick() {
panic("Tick called on an EWMASnapshot")
}
// Update panics.
func (EWMASnapshot) Update(int64) {
panic("Update called on an EWMASnapshot")
}
// NilEWMA is a no-op EWMA.
type NilEWMA struct{}
// Rate is a no-op.
func (NilEWMA) Rate() float64 { return 0.0 }
// Snapshot is a no-op.
func (NilEWMA) Snapshot() EWMA { return NilEWMA{} }
// Tick is a no-op.
func (NilEWMA) Tick() {}
// Update is a no-op.
func (NilEWMA) Update(n int64) {}
// StandardEWMA is the standard implementation of an EWMA and tracks the number
// of uncounted events and processes them on each tick. It uses the
// sync/atomic package to manage uncounted events.
type StandardEWMA struct {
uncounted int64 // /!\ this should be the first member to ensure 64-bit alignment
alpha float64
rate uint64
init uint32
mutex sync.Mutex
}
// Rate returns the moving average rate of events per second.
func (a *StandardEWMA) Rate() float64 {
currentRate := math.Float64frombits(atomic.LoadUint64(&a.rate)) * float64(1e9)
return currentRate
}
// Snapshot returns a read-only copy of the EWMA.
func (a *StandardEWMA) Snapshot() EWMA {
return EWMASnapshot(a.Rate())
}
// Tick ticks the clock to update the moving average. It assumes it is called
// every five seconds.
func (a *StandardEWMA) Tick() {
// Optimization to avoid mutex locking in the hot-path.
if atomic.LoadUint32(&a.init) == 1 {
a.updateRate(a.fetchInstantRate())
} else {
// Slow-path: this is only needed on the first Tick() and preserves transactional updating
// of init and rate in the else block. The first conditional is needed below because
// a different thread could have set a.init = 1 between the time of the first atomic load and when
// the lock was acquired.
a.mutex.Lock()
if atomic.LoadUint32(&a.init) == 1 {
// The fetchInstantRate() uses atomic loading, which is unecessary in this critical section
// but again, this section is only invoked on the first successful Tick() operation.
a.updateRate(a.fetchInstantRate())
} else {
atomic.StoreUint32(&a.init, 1)
atomic.StoreUint64(&a.rate, math.Float64bits(a.fetchInstantRate()))
}
a.mutex.Unlock()
}
}
func (a *StandardEWMA) fetchInstantRate() float64 {
count := atomic.LoadInt64(&a.uncounted)
atomic.AddInt64(&a.uncounted, -count)
instantRate := float64(count) / float64(5e9)
return instantRate
}
func (a *StandardEWMA) updateRate(instantRate float64) {
currentRate := math.Float64frombits(atomic.LoadUint64(&a.rate))
currentRate += a.alpha * (instantRate - currentRate)
atomic.StoreUint64(&a.rate, math.Float64bits(currentRate))
}
// Update adds n uncounted events.
func (a *StandardEWMA) Update(n int64) {
atomic.AddInt64(&a.uncounted, n)
}