123456789101112131415161718192021222324252627282930313233343536373839404142434445464748495051525354555657585960616263646566676869707172737475767778798081828384858687888990919293949596979899100101102103104105106107108109110111112113114115116117118119120121122123124125126127128129130131132133134135136137138 |
- 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)
- }
|