Skip to content

Commit 9d124ef

Browse files
sushanbsushanb
authored
feat(bigtable): add PeakEwma continuous time-decay latency tracker (#20187)
## Summary Adds `PeakEwma`, a thread-safe continuous-time exponentially-weighted moving average of latency samples. The upcoming AFE picker (Session subsystem) uses one instance per bucket to score candidates; landing the type standalone lets it merge ahead of its consumers. - **`peak_ewma.go`** — `PeakEwma` struct, `NewPeakEwma(tau)`, `NewPeakEwmaSeeded(tau, seed)`, `Update(latency)`, `Value()`. Decay weight is `e^(-dt/tau)` computed on every Update. First Update snaps value to the sample; subsequent updates blend prior and new proportionally. - **`NewPeakEwmaSeeded`** — Java parity: `SessionList.java` seeds transport at 500µs and e2e at 1ms so a brand-new AFE doesn't win the least-latency picker by looking free-cost. The seed is authoritative only until the first `Update`. No callers on `main` yet — follow-up PRs for the AFE picker and `sessionList` will consume it. ## Test plan - [x] `go build ./bigtable/...` - [x] `go test ./bigtable/internal/transport/ -run '^TestPeakEwma' -count=10 -race` — passes (1.6s) - [x] `gofmt -l bigtable/internal/transport/peak_ewma*.go` — clean - [x] `go vet ./bigtable/internal/transport/` — clean Test coverage: - Unseeded / seeded initial `Value()` - First `Update` overrides the seed (pins the documented behavior) - Constant-sample invariant — `w*L + (1-w)*L == L` for any weight, so the EWMA is exact on a flat stream - Convex-combination bound — Value stays within `[min, max]` of samples seen - `Value()` is a pure read (no state mutation) - Wide-gap dominance — `dt >> tau` ⇒ weight → 0 ⇒ Value tracks the newest sample (skipped under `-short`) - Mutex race check with mixed reader/writer goroutines (meaningful under `-race`)
1 parent 02e3c6d commit 9d124ef

2 files changed

Lines changed: 402 additions & 0 deletions

File tree

‎bigtable/internal/transport/peak_ewma.go‎

Lines changed: 108 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -0,0 +1,108 @@
1+
// Copyright 2026 Google LLC
2+
//
3+
// Licensed under the Apache License, Version 2.0 (the "License");
4+
// you may not use this file except in compliance with the License.
5+
// You may obtain a copy of the License at
6+
//
7+
// http://www.apache.org/licenses/LICENSE-2.0
8+
//
9+
// Unless required by applicable law or agreed to in writing, software
10+
// distributed under the License is distributed on an "AS IS" BASIS,
11+
// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
12+
// See the License for the specific language governing permissions and
13+
// limitations under the License.
14+
15+
package internal
16+
17+
import (
18+
"math"
19+
"sync"
20+
"time"
21+
)
22+
23+
// PeakEwma is a thread-safe latency tracker that rises immediately on
24+
// higher samples (the "peak" step) and decays exponentially toward
25+
// lower ones (the EWMA step). The AFE picker uses one instance per
26+
// bucket to score candidates; see sessionList.
27+
//
28+
// The peak-step is what prevents a new AFE from stealing traffic just
29+
// because its latest vRPC happened to be fast — a plain symmetric
30+
// EWMA would blend a single low sample into the average, briefly
31+
// making the bucket look cheapest and steering pickers at it.
32+
type PeakEwma struct {
33+
mu sync.Mutex
34+
tau time.Duration
35+
value float64
36+
lastUpdate time.Time
37+
}
38+
39+
// NewPeakEwma creates a PeakEwma with the given decay time constant tau.
40+
// The initial value is zero — the first positive Update peak-snaps to
41+
// its sample. lastUpdate is stamped at construction so subsequent
42+
// updates see a valid dt.
43+
func NewPeakEwma(tau time.Duration) *PeakEwma {
44+
return &PeakEwma{
45+
tau: tau,
46+
lastUpdate: time.Now(),
47+
}
48+
}
49+
50+
// NewPeakEwmaSeeded returns a PeakEwma pre-seeded with the given cost.
51+
// The seed participates in the first Update's decay/blend normally —
52+
// it is NOT overwritten on the first sample. SessionList seeds
53+
// transport at 500µs and e2e at 1ms so a brand-new AFE doesn't win
54+
// the least-latency picker by looking free-cost.
55+
func NewPeakEwmaSeeded(tau, seed time.Duration) *PeakEwma {
56+
return &PeakEwma{
57+
tau: tau,
58+
value: float64(seed),
59+
lastUpdate: time.Now(),
60+
}
61+
}
62+
63+
// Update folds a new latency sample into the tracker. Two distinct
64+
// steps:
65+
//
66+
// - Peak-step: if the sample is higher than the current value, snap
67+
// up immediately. No decay, no blend — the higher observation
68+
// supersedes.
69+
// - Decay-step: otherwise, apply e^(-dt/tau) time-decay to the
70+
// current value and blend the sample in proportionally.
71+
//
72+
// Non-positive samples are ignored; a backward clock (dt < 0) is
73+
// clamped to 0 so the blend weight stays in [0,1]; a non-positive tau
74+
// collapses to zero decay (new sample fully replaces) so tau=0/dt=0
75+
// doesn't produce NaN via exp(-0/0).
76+
func (e *PeakEwma) Update(latency time.Duration) {
77+
if latency <= 0 {
78+
return
79+
}
80+
e.mu.Lock()
81+
defer e.mu.Unlock()
82+
now := time.Now()
83+
latencyNs := float64(latency)
84+
if e.value < latencyNs {
85+
e.value = latencyNs
86+
e.lastUpdate = now
87+
return
88+
}
89+
dt := now.Sub(e.lastUpdate)
90+
if dt < 0 {
91+
dt = 0
92+
}
93+
e.lastUpdate = now
94+
var decay float64
95+
if e.tau > 0 {
96+
decay = math.Exp(-float64(dt) / float64(e.tau))
97+
}
98+
e.value = e.value*decay + latencyNs*(1-decay)
99+
}
100+
101+
// Value returns the current tracker value in the same units as the
102+
// samples fed into Update (nanoseconds, as float64 — the picker
103+
// consumes this raw).
104+
func (e *PeakEwma) Value() float64 {
105+
e.mu.Lock()
106+
defer e.mu.Unlock()
107+
return e.value
108+
}

0 commit comments

Comments
 (0)