227 lines
7.6 KiB
Go
227 lines
7.6 KiB
Go
package gateway
|
|
|
|
import (
|
|
"sort"
|
|
"testing"
|
|
"time"
|
|
)
|
|
|
|
type syntheticPacerDelivery struct {
|
|
at time.Time
|
|
flow string
|
|
bytes int64
|
|
}
|
|
|
|
func TestFairPacerEightFlowSharesAndCapacitySteps(t *testing.T) {
|
|
start := time.Date(2026, time.January, 1, 0, 0, 0, 0, time.UTC)
|
|
flows := []string{"one", "two", "three", "four", "five", "six", "seven", "eight"}
|
|
pacer := newFairPacer(8000)
|
|
next := make(map[string]time.Time, len(flows))
|
|
baseline := runSyntheticPacer(pacer, start, start.Add(60*time.Second), flows, next)
|
|
assertSyntheticFairness(t, baseline, flows)
|
|
assertSyntheticCap(t, baseline, 1_000_000)
|
|
|
|
pacer.setKbps(6000)
|
|
quarter := runSyntheticPacer(pacer, start.Add(60*time.Second), start.Add(70*time.Second), flows, next)
|
|
assertSyntheticFairness(t, quarter, flows)
|
|
assertSyntheticCap(t, quarter, 750_000)
|
|
|
|
pacer.setKbps(4000)
|
|
half := runSyntheticPacer(pacer, start.Add(70*time.Second), start.Add(80*time.Second), flows, next)
|
|
assertSyntheticFairness(t, half, flows)
|
|
assertSyntheticCap(t, half, 500_000)
|
|
}
|
|
|
|
func TestFairPacerBoundsCatchupAfterHostStall(t *testing.T) {
|
|
start := time.Date(2026, time.January, 1, 0, 0, 0, 0, time.UTC)
|
|
pacer := newFairPacer(8000)
|
|
_ = pacer.reserveAt(start, "one", 1000)
|
|
resumed := start.Add(100 * time.Millisecond)
|
|
next := pacer.reserveAt(resumed, "one", 1000)
|
|
if next.Before(resumed.Add(-fairPacerMaximumCatchup)) || next.After(resumed.Add(10*time.Millisecond)) {
|
|
t.Fatalf("post-stall reservation = %s, want bounded catchup near %s", next, resumed)
|
|
}
|
|
pacer.mu.Lock()
|
|
debt := pacer.flows["one"].debt
|
|
pacer.mu.Unlock()
|
|
if debt <= 0 || debt > nativeApolloVideoQueueLatency-fairPacerMaximumCatchup {
|
|
t.Fatalf("post-stall debt = %s, want bounded valid schedule debt", debt)
|
|
}
|
|
}
|
|
|
|
func TestFairPacerRepaysBoundedDebtAfterHostStall(t *testing.T) {
|
|
start := time.Date(2026, time.January, 1, 0, 0, 0, 0, time.UTC)
|
|
pacer := newFairPacer(8000)
|
|
next := make(map[string]time.Time)
|
|
deliveries := runSyntheticPacerWithStall(
|
|
pacer, start, start.Add(6*time.Second), []string{"one"}, next,
|
|
start.Add(time.Second), 100*time.Millisecond,
|
|
)
|
|
if total := syntheticDeliveryBytes(deliveries); total < 5_990_000 || total > 6_010_000 {
|
|
t.Fatalf("post-stall delivery bytes = %d, want nominal throughput after bounded debt repayment", total)
|
|
}
|
|
pacer.mu.Lock()
|
|
remaining := pacer.flows["one"].debt
|
|
pacer.mu.Unlock()
|
|
if remaining != 0 {
|
|
t.Fatalf("post-stall debt = %s after repayment, want zero", remaining)
|
|
}
|
|
assertSyntheticCap(t, deliveries, 1_000_000)
|
|
t.Logf("single-flow debt repaid: bytes=%d remaining=%s", syntheticDeliveryBytes(deliveries), remaining)
|
|
}
|
|
|
|
func TestFairPacerRepaysSimultaneousEightFlowDebtAcrossCapacitySteps(t *testing.T) {
|
|
start := time.Date(2026, time.January, 1, 0, 0, 0, 0, time.UTC)
|
|
flows := []string{"one", "two", "three", "four", "five", "six", "seven", "eight"}
|
|
|
|
tests := []struct {
|
|
name string
|
|
kbps int64
|
|
bytesPerSecond int64
|
|
minimumBytes int64
|
|
}{
|
|
{name: "baseline", kbps: 8000, bytesPerSecond: 1_000_000, minimumBytes: 9_980_000},
|
|
{name: "quarter", kbps: 6000, bytesPerSecond: 750_000, minimumBytes: 7_480_000},
|
|
{name: "half", kbps: 4000, bytesPerSecond: 500_000, minimumBytes: 4_980_000},
|
|
}
|
|
for _, test := range tests {
|
|
pacer := newFairPacer(8000)
|
|
next := make(map[string]time.Time, len(flows))
|
|
_ = runSyntheticPacer(pacer, start, start.Add(time.Second), flows, next)
|
|
resumed := start.Add(1100 * time.Millisecond)
|
|
for _, flow := range flows {
|
|
next[flow] = pacer.reserveAt(resumed, flow, 1000)
|
|
}
|
|
assertSyntheticDebt(t, pacer, flows, true)
|
|
pacer.setKbps(test.kbps)
|
|
deliveries := runSyntheticPacer(pacer, resumed, resumed.Add(10*time.Second), flows, next)
|
|
if total := syntheticDeliveryBytes(deliveries); total < test.minimumBytes {
|
|
t.Fatalf("%s post-stall delivery bytes = %d, want at least %d", test.name, total, test.minimumBytes)
|
|
}
|
|
assertSyntheticFairness(t, deliveries, flows)
|
|
assertSyntheticCap(t, deliveries, test.bytesPerSecond)
|
|
assertSyntheticDebt(t, pacer, flows, false)
|
|
t.Logf("%s eight-flow debt repaid: bytes=%d cap=%d", test.name, syntheticDeliveryBytes(deliveries), test.bytesPerSecond*5*105/100)
|
|
}
|
|
}
|
|
|
|
func assertSyntheticDebt(t *testing.T, pacer *fairPacer, flows []string, wantDebt bool) {
|
|
t.Helper()
|
|
pacer.mu.Lock()
|
|
defer pacer.mu.Unlock()
|
|
for _, flow := range flows {
|
|
debt := pacer.flows[flow].debt
|
|
if wantDebt && (debt <= 0 || debt > nativeApolloVideoQueueLatency-fairPacerMaximumCatchup) {
|
|
t.Fatalf("flow %s active debt = %s, want bounded nonzero debt", flow, debt)
|
|
}
|
|
if !wantDebt && debt != 0 {
|
|
t.Fatalf("flow %s debt = %s after repayment, want zero", flow, debt)
|
|
}
|
|
}
|
|
}
|
|
|
|
func runSyntheticPacer(pacer *fairPacer, start, end time.Time, flows []string, next map[string]time.Time) []syntheticPacerDelivery {
|
|
const packetBytes = 1000
|
|
for _, flow := range flows {
|
|
if next[flow].IsZero() {
|
|
next[flow] = pacer.reserveAt(start, flow, packetBytes)
|
|
}
|
|
}
|
|
var deliveries []syntheticPacerDelivery
|
|
for {
|
|
flow := ""
|
|
at := end.Add(time.Nanosecond)
|
|
for _, candidate := range flows {
|
|
if next[candidate].Before(at) {
|
|
flow, at = candidate, next[candidate]
|
|
}
|
|
}
|
|
if at.After(end) {
|
|
return deliveries
|
|
}
|
|
deliveries = append(deliveries, syntheticPacerDelivery{at: at, flow: flow, bytes: packetBytes})
|
|
next[flow] = pacer.reserveAt(at, flow, packetBytes)
|
|
}
|
|
}
|
|
|
|
func runSyntheticPacerWithStall(pacer *fairPacer, start, end time.Time, flows []string, next map[string]time.Time, stallAt time.Time, stall time.Duration) []syntheticPacerDelivery {
|
|
const packetBytes = 1000
|
|
for _, flow := range flows {
|
|
if next[flow].IsZero() {
|
|
next[flow] = pacer.reserveAt(start, flow, packetBytes)
|
|
}
|
|
}
|
|
now := start
|
|
stalled := false
|
|
var deliveries []syntheticPacerDelivery
|
|
for {
|
|
flow := ""
|
|
target := end.Add(time.Nanosecond)
|
|
for _, candidate := range flows {
|
|
if next[candidate].Before(target) {
|
|
flow, target = candidate, next[candidate]
|
|
}
|
|
}
|
|
if target.After(end) {
|
|
return deliveries
|
|
}
|
|
if !stalled && !target.Before(stallAt) {
|
|
now = stallAt.Add(stall)
|
|
stalled = true
|
|
}
|
|
if now.Before(target) {
|
|
now = target
|
|
}
|
|
if now.After(end) {
|
|
return deliveries
|
|
}
|
|
deliveries = append(deliveries, syntheticPacerDelivery{at: now, flow: flow, bytes: packetBytes})
|
|
next[flow] = pacer.reserveAt(now, flow, packetBytes)
|
|
}
|
|
}
|
|
|
|
func syntheticDeliveryBytes(deliveries []syntheticPacerDelivery) int64 {
|
|
var total int64
|
|
for _, delivery := range deliveries {
|
|
total += delivery.bytes
|
|
}
|
|
return total
|
|
}
|
|
|
|
func assertSyntheticFairness(t *testing.T, deliveries []syntheticPacerDelivery, flows []string) {
|
|
t.Helper()
|
|
counts := make(map[string]int64, len(flows))
|
|
for _, delivery := range deliveries {
|
|
counts[delivery.flow] += delivery.bytes
|
|
}
|
|
total := int64(0)
|
|
for _, flow := range flows {
|
|
total += counts[flow]
|
|
}
|
|
target := total / int64(len(flows))
|
|
for _, flow := range flows {
|
|
delta := counts[flow] - target
|
|
if delta < 0 {
|
|
delta = -delta
|
|
}
|
|
if target == 0 || float64(delta)/float64(target) > 0.10 {
|
|
t.Fatalf("flow %s share=%d target=%d", flow, counts[flow], target)
|
|
}
|
|
}
|
|
}
|
|
|
|
func assertSyntheticCap(t *testing.T, deliveries []syntheticPacerDelivery, bytesPerSecond int64) {
|
|
t.Helper()
|
|
sort.Slice(deliveries, func(first, second int) bool { return deliveries[first].at.Before(deliveries[second].at) })
|
|
for first, total, last := 0, int64(0), 0; first < len(deliveries); first++ {
|
|
for last < len(deliveries) && deliveries[last].at.Sub(deliveries[first].at) <= 5*time.Second {
|
|
total += deliveries[last].bytes
|
|
last++
|
|
}
|
|
if total > bytesPerSecond*5*105/100 {
|
|
t.Fatalf("five-second egress=%d exceeds cap=%d", total, bytesPerSecond*5)
|
|
}
|
|
total -= deliveries[first].bytes
|
|
}
|
|
}
|