Skip to content

Commit a1da5fe

Browse files
aponysusaponysus
authored
Add latency aware hedging and respect policy.Hedge.MaxHedges. (#2)
* Add latency aware hedging and respect policy.Hedge.MaxHedges. * Fix triggers test.
1 parent b4001d2 commit a1da5fe

9 files changed

Lines changed: 563 additions & 22 deletions

File tree

‎hedge/fixed_delay.go‎

Lines changed: 9 additions & 8 deletions
Original file line numberDiff line numberDiff line change
@@ -8,16 +8,17 @@ type FixedDelayTrigger struct {
88
}
99

1010
func (t FixedDelayTrigger) ShouldSpawnHedge(state HedgeState) (bool, time.Duration) {
11-
// If we haven't reached the delay yet, wait until we do.
12-
if state.Elapsed < t.Delay {
13-
return false, t.Delay - state.Elapsed
11+
// We stop if we've reached the maximum number of attempts (Primary + MaxHedges).
12+
if state.AttemptsLaunched >= 1+state.MaxHedges {
13+
return false, 0
1414
}
1515

16-
// FixedDelay spawns a single hedge after the specified delay.
17-
// If we've already launched more than 1 attempt (primary), we stop.
18-
19-
if state.AttemptsLaunched > 1 {
20-
return false, 0 // No more hedges from this trigger
16+
// For multiple hedges, we space them out by the delay.
17+
// Primary (1) -> Wait Delay -> Hedge 1 (2) -> Wait Delay -> Hedge 2 (3) ...
18+
// Target elapsed time for the *next* hedge is Delay * AttemptsLaunched.
19+
target := t.Delay * time.Duration(state.AttemptsLaunched)
20+
if state.Elapsed < target {
21+
return false, target - state.Elapsed
2122
}
2223

2324
return true, 0

‎hedge/tracker.go‎

Lines changed: 109 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -0,0 +1,109 @@
1+
package hedge
2+
3+
import (
4+
"sort"
5+
"sync"
6+
"time"
7+
)
8+
9+
// LatencySnapshot contains latency quantiles.
10+
type LatencySnapshot struct {
11+
P50 time.Duration
12+
P90 time.Duration
13+
P95 time.Duration
14+
P99 time.Duration
15+
}
16+
17+
// LatencyTracker tracks recent latency samples and calculates quantiles.
18+
type LatencyTracker interface {
19+
// Observe records a duration sample.
20+
Observe(d time.Duration)
21+
// Snapshot returns the current latency snapshot.
22+
Snapshot() LatencySnapshot
23+
}
24+
25+
// RingBufferTracker implements LatencyTracker using a fixed-size ring buffer.
26+
// It is safe for concurrent use.
27+
type RingBufferTracker struct {
28+
mu sync.RWMutex
29+
samples []time.Duration
30+
idx int
31+
full bool
32+
}
33+
34+
// NewRingBufferTracker creates a new tracker with the specified size.
35+
// Size must be greater than 0.
36+
func NewRingBufferTracker(size int) *RingBufferTracker {
37+
if size <= 0 {
38+
size = 256 // Default safe size
39+
}
40+
return &RingBufferTracker{
41+
samples: make([]time.Duration, size),
42+
}
43+
}
44+
45+
// Observe records a duration sample.
46+
func (t *RingBufferTracker) Observe(d time.Duration) {
47+
t.mu.Lock()
48+
defer t.mu.Unlock()
49+
50+
t.samples[t.idx] = d
51+
t.idx++
52+
if t.idx >= len(t.samples) {
53+
t.idx = 0
54+
t.full = true
55+
}
56+
}
57+
58+
// Snapshot returns the current latency snapshot.
59+
func (t *RingBufferTracker) Snapshot() LatencySnapshot {
60+
t.mu.RLock()
61+
defer t.mu.RUnlock()
62+
63+
count := t.idx
64+
if t.full {
65+
count = len(t.samples)
66+
}
67+
68+
if count == 0 {
69+
return LatencySnapshot{}
70+
}
71+
72+
// Copy samples to avoid holding lock during sort
73+
// We only copy valid samples
74+
sorted := make([]time.Duration, count)
75+
if t.full {
76+
copy(sorted, t.samples)
77+
} else {
78+
copy(sorted, t.samples[:count])
79+
}
80+
81+
sort.Slice(sorted, func(i, j int) bool {
82+
return sorted[i] < sorted[j]
83+
})
84+
85+
return LatencySnapshot{
86+
P50: quantile(sorted, 0.50),
87+
P90: quantile(sorted, 0.90),
88+
P95: quantile(sorted, 0.95),
89+
P99: quantile(sorted, 0.99),
90+
}
91+
}
92+
93+
func quantile(sorted []time.Duration, q float64) time.Duration {
94+
if len(sorted) == 0 {
95+
return 0
96+
}
97+
// Use (N-1)*q to interpret index in 0-based array.
98+
// For N=100 (indices 0-99):
99+
// q=0.5 -> 49.5 -> 49 (Value 50)
100+
// q=0.99 -> 98.01 -> 98 (Value 99)
101+
idx := int(float64(len(sorted)-1) * q)
102+
if idx >= len(sorted) {
103+
idx = len(sorted) - 1
104+
}
105+
if idx < 0 {
106+
idx = 0
107+
}
108+
return sorted[idx]
109+
}

‎hedge/tracker_test.go‎

Lines changed: 86 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -0,0 +1,86 @@
1+
package hedge
2+
3+
import (
4+
"testing"
5+
"time"
6+
)
7+
8+
func TestRingBufferTracker_Empty(t *testing.T) {
9+
tracker := NewRingBufferTracker(10)
10+
snap := tracker.Snapshot()
11+
12+
if snap.P50 != 0 || snap.P90 != 0 || snap.P99 != 0 {
13+
t.Errorf("expected zero stats for empty tracker, got %+v", snap)
14+
}
15+
}
16+
17+
func TestRingBufferTracker_Simple(t *testing.T) {
18+
tracker := NewRingBufferTracker(100)
19+
20+
// Add 100 samples: 1ms, 2ms, ... 100ms
21+
for i := 1; i <= 100; i++ {
22+
tracker.Observe(time.Duration(i) * time.Millisecond)
23+
}
24+
25+
snap := tracker.Snapshot()
26+
27+
// P50 should be ~50ms
28+
if snap.P50 != 50*time.Millisecond {
29+
t.Errorf("expected P50=50ms, got %v", snap.P50)
30+
}
31+
// P90 should be ~90ms
32+
if snap.P90 != 90*time.Millisecond {
33+
t.Errorf("expected P90=90ms, got %v", snap.P90)
34+
}
35+
// P99 should be ~99ms
36+
if snap.P99 != 99*time.Millisecond {
37+
t.Errorf("expected P99=99ms, got %v", snap.P99)
38+
}
39+
}
40+
41+
func TestRingBufferTracker_Rollover(t *testing.T) {
42+
tracker := NewRingBufferTracker(5)
43+
44+
// Fill with low values
45+
for i := 0; i < 5; i++ {
46+
tracker.Observe(1 * time.Millisecond)
47+
}
48+
49+
// Overwrite with high values
50+
tracker.Observe(100 * time.Millisecond)
51+
tracker.Observe(100 * time.Millisecond)
52+
53+
// Buffer: [100, 100, 1, 1, 1] (sorted: 1, 1, 1, 100, 100)
54+
// P50 (index 2) = 1ms
55+
// P90 (index 4) = 100ms
56+
57+
snap := tracker.Snapshot()
58+
if snap.P50 != 1*time.Millisecond {
59+
t.Errorf("expected P50=1ms, got %v", snap.P50)
60+
}
61+
if snap.P90 != 100*time.Millisecond {
62+
t.Errorf("expected P90=100ms, got %v", snap.P90)
63+
}
64+
}
65+
66+
func TestRingBufferTracker_Concurrent(t *testing.T) {
67+
tracker := NewRingBufferTracker(1000)
68+
done := make(chan bool)
69+
70+
go func() {
71+
for i := 0; i < 1000; i++ {
72+
tracker.Observe(time.Duration(i) * time.Millisecond)
73+
}
74+
done <- true
75+
}()
76+
77+
go func() {
78+
for i := 0; i < 100; i++ {
79+
tracker.Snapshot()
80+
}
81+
done <- true
82+
}()
83+
84+
<-done
85+
<-done
86+
}

‎hedge/triggers.go‎

Lines changed: 51 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -0,0 +1,51 @@
1+
package hedge
2+
3+
import (
4+
"strings"
5+
"time"
6+
)
7+
8+
// LatencyTrigger spawns a hedge if the elapsed time exceeds a dynamic threshold.
9+
type LatencyTrigger struct {
10+
Percentile string // "p50", "p90", "p95", "p99"
11+
}
12+
13+
// ShouldSpawnHedge checks if the hedge should be spawned based on latency stats.
14+
func (t LatencyTrigger) ShouldSpawnHedge(state HedgeState) (bool, time.Duration) {
15+
threshold := time.Duration(0)
16+
17+
// Since we haven't updated HedgeState yet, I will write the logic assuming state has Snapshot.
18+
// We will update types.go in the next step.
19+
20+
switch strings.ToLower(t.Percentile) {
21+
case "p50":
22+
threshold = state.Snapshot.P50
23+
case "p90":
24+
threshold = state.Snapshot.P90
25+
case "p95":
26+
threshold = state.Snapshot.P95
27+
case "p99":
28+
threshold = state.Snapshot.P99
29+
default:
30+
// Default to P95 or similar if unknown? Or invalid config.
31+
// Safe fallback: 0 means never (since elapsed always > 0).
32+
return false, 0
33+
}
34+
35+
if threshold <= 0 {
36+
return false, 0
37+
}
38+
39+
if state.Elapsed > threshold {
40+
// Spawn!
41+
// But only if we haven't already reached the limit.
42+
if state.AttemptsLaunched >= 1+state.MaxHedges {
43+
return false, 0
44+
}
45+
return true, 0
46+
}
47+
48+
// Wait remaining
49+
remaining := threshold - state.Elapsed
50+
return false, remaining
51+
}

‎hedge/triggers_test.go‎

Lines changed: 112 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -0,0 +1,112 @@
1+
package hedge
2+
3+
import (
4+
"testing"
5+
"time"
6+
)
7+
8+
func TestLatencyTrigger_ShouldSpawnHedge(t *testing.T) {
9+
snap := LatencySnapshot{
10+
P50: 10 * time.Millisecond,
11+
P90: 50 * time.Millisecond,
12+
P99: 100 * time.Millisecond,
13+
}
14+
15+
tests := []struct {
16+
name string
17+
percentile string
18+
elapsed time.Duration
19+
attempts int
20+
maxHedges int
21+
want bool
22+
wantWait time.Duration
23+
}{
24+
{
25+
name: "P50 Trigger - Below Threshold",
26+
percentile: "p50",
27+
elapsed: 5 * time.Millisecond,
28+
attempts: 1,
29+
maxHedges: 1,
30+
want: false,
31+
wantWait: 5 * time.Millisecond, // 10 - 5
32+
},
33+
{
34+
name: "P50 Trigger - Above Threshold",
35+
percentile: "p50",
36+
elapsed: 11 * time.Millisecond,
37+
attempts: 1,
38+
maxHedges: 1,
39+
want: true,
40+
wantWait: 0,
41+
},
42+
{
43+
name: "P99 Trigger - Below Threshold",
44+
percentile: "p99",
45+
elapsed: 90 * time.Millisecond,
46+
attempts: 1,
47+
maxHedges: 1,
48+
want: false,
49+
wantWait: 10 * time.Millisecond,
50+
},
51+
{
52+
name: "P99 Trigger - Above Threshold",
53+
percentile: "p99",
54+
elapsed: 101 * time.Millisecond,
55+
attempts: 1,
56+
maxHedges: 1,
57+
want: true,
58+
wantWait: 0,
59+
},
60+
{
61+
name: "Already Hedged - Should Stop",
62+
percentile: "p50",
63+
elapsed: 20 * time.Millisecond,
64+
attempts: 2,
65+
maxHedges: 1,
66+
want: false,
67+
wantWait: 0,
68+
},
69+
{
70+
name: "Unknown Percentile",
71+
percentile: "pXX",
72+
elapsed: 1000 * time.Millisecond,
73+
attempts: 1,
74+
maxHedges: 1,
75+
want: false,
76+
wantWait: 0,
77+
},
78+
{
79+
name: "Zero Stats",
80+
percentile: "p50",
81+
elapsed: 100 * time.Millisecond,
82+
attempts: 1,
83+
maxHedges: 1,
84+
want: false, // Threshold is 0, so <= 0 check returns false?
85+
// Logic: threshold <= 0 returns false.
86+
wantWait: 0,
87+
},
88+
}
89+
90+
for _, tt := range tests {
91+
t.Run(tt.name, func(t *testing.T) {
92+
trigger := LatencyTrigger{Percentile: tt.percentile}
93+
state := HedgeState{
94+
Elapsed: tt.elapsed,
95+
AttemptsLaunched: tt.attempts,
96+
MaxHedges: tt.maxHedges,
97+
Snapshot: snap,
98+
}
99+
if tt.name == "Zero Stats" {
100+
state.Snapshot = LatencySnapshot{}
101+
}
102+
103+
got, gotWait := trigger.ShouldSpawnHedge(state)
104+
if got != tt.want {
105+
t.Errorf("got %v, want %v", got, tt.want)
106+
}
107+
if gotWait != tt.wantWait {
108+
t.Errorf("gotWait %v, want %v", gotWait, tt.wantWait)
109+
}
110+
})
111+
}
112+
}

‎hedge/types.go‎

Lines changed: 2 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -15,6 +15,8 @@ type HedgeState struct {
1515
MaxHedges int
1616
// Elapsed is the time elapsed since AttemptStart.
1717
Elapsed time.Duration
18+
// Snapshot contains the current latency statistics for the operation.
19+
Snapshot LatencySnapshot
1820
}
1921

2022
// Trigger decides when to spawn a hedged attempt.

0 commit comments

Comments
 (0)