ewma.go 3.8 KB

123456789101112131415161718192021222324252627282930313233343536373839404142434445464748495051525354555657585960616263646566676869707172737475767778798081828384858687888990919293949596979899100101102103104105106107108109110111112113114115116117118119120121122123124125126127128129130131132133134135136137138
  1. package metrics
  2. import (
  3. "math"
  4. "sync"
  5. "sync/atomic"
  6. )
  7. // EWMAs continuously calculate an exponentially-weighted moving average
  8. // based on an outside source of clock ticks.
  9. type EWMA interface {
  10. Rate() float64
  11. Snapshot() EWMA
  12. Tick()
  13. Update(int64)
  14. }
  15. // NewEWMA constructs a new EWMA with the given alpha.
  16. func NewEWMA(alpha float64) EWMA {
  17. if UseNilMetrics {
  18. return NilEWMA{}
  19. }
  20. return &StandardEWMA{alpha: alpha}
  21. }
  22. // NewEWMA1 constructs a new EWMA for a one-minute moving average.
  23. func NewEWMA1() EWMA {
  24. return NewEWMA(1 - math.Exp(-5.0/60.0/1))
  25. }
  26. // NewEWMA5 constructs a new EWMA for a five-minute moving average.
  27. func NewEWMA5() EWMA {
  28. return NewEWMA(1 - math.Exp(-5.0/60.0/5))
  29. }
  30. // NewEWMA15 constructs a new EWMA for a fifteen-minute moving average.
  31. func NewEWMA15() EWMA {
  32. return NewEWMA(1 - math.Exp(-5.0/60.0/15))
  33. }
  34. // EWMASnapshot is a read-only copy of another EWMA.
  35. type EWMASnapshot float64
  36. // Rate returns the rate of events per second at the time the snapshot was
  37. // taken.
  38. func (a EWMASnapshot) Rate() float64 { return float64(a) }
  39. // Snapshot returns the snapshot.
  40. func (a EWMASnapshot) Snapshot() EWMA { return a }
  41. // Tick panics.
  42. func (EWMASnapshot) Tick() {
  43. panic("Tick called on an EWMASnapshot")
  44. }
  45. // Update panics.
  46. func (EWMASnapshot) Update(int64) {
  47. panic("Update called on an EWMASnapshot")
  48. }
  49. // NilEWMA is a no-op EWMA.
  50. type NilEWMA struct{}
  51. // Rate is a no-op.
  52. func (NilEWMA) Rate() float64 { return 0.0 }
  53. // Snapshot is a no-op.
  54. func (NilEWMA) Snapshot() EWMA { return NilEWMA{} }
  55. // Tick is a no-op.
  56. func (NilEWMA) Tick() {}
  57. // Update is a no-op.
  58. func (NilEWMA) Update(n int64) {}
  59. // StandardEWMA is the standard implementation of an EWMA and tracks the number
  60. // of uncounted events and processes them on each tick. It uses the
  61. // sync/atomic package to manage uncounted events.
  62. type StandardEWMA struct {
  63. uncounted int64 // /!\ this should be the first member to ensure 64-bit alignment
  64. alpha float64
  65. rate uint64
  66. init uint32
  67. mutex sync.Mutex
  68. }
  69. // Rate returns the moving average rate of events per second.
  70. func (a *StandardEWMA) Rate() float64 {
  71. currentRate := math.Float64frombits(atomic.LoadUint64(&a.rate)) * float64(1e9)
  72. return currentRate
  73. }
  74. // Snapshot returns a read-only copy of the EWMA.
  75. func (a *StandardEWMA) Snapshot() EWMA {
  76. return EWMASnapshot(a.Rate())
  77. }
  78. // Tick ticks the clock to update the moving average. It assumes it is called
  79. // every five seconds.
  80. func (a *StandardEWMA) Tick() {
  81. // Optimization to avoid mutex locking in the hot-path.
  82. if atomic.LoadUint32(&a.init) == 1 {
  83. a.updateRate(a.fetchInstantRate())
  84. } else {
  85. // Slow-path: this is only needed on the first Tick() and preserves transactional updating
  86. // of init and rate in the else block. The first conditional is needed below because
  87. // a different thread could have set a.init = 1 between the time of the first atomic load and when
  88. // the lock was acquired.
  89. a.mutex.Lock()
  90. if atomic.LoadUint32(&a.init) == 1 {
  91. // The fetchInstantRate() uses atomic loading, which is unecessary in this critical section
  92. // but again, this section is only invoked on the first successful Tick() operation.
  93. a.updateRate(a.fetchInstantRate())
  94. } else {
  95. atomic.StoreUint32(&a.init, 1)
  96. atomic.StoreUint64(&a.rate, math.Float64bits(a.fetchInstantRate()))
  97. }
  98. a.mutex.Unlock()
  99. }
  100. }
  101. func (a *StandardEWMA) fetchInstantRate() float64 {
  102. count := atomic.LoadInt64(&a.uncounted)
  103. atomic.AddInt64(&a.uncounted, -count)
  104. instantRate := float64(count) / float64(5e9)
  105. return instantRate
  106. }
  107. func (a *StandardEWMA) updateRate(instantRate float64) {
  108. currentRate := math.Float64frombits(atomic.LoadUint64(&a.rate))
  109. currentRate += a.alpha * (instantRate - currentRate)
  110. atomic.StoreUint64(&a.rate, math.Float64bits(currentRate))
  111. }
  112. // Update adds n uncounted events.
  113. func (a *StandardEWMA) Update(n int64) {
  114. atomic.AddInt64(&a.uncounted, n)
  115. }