sechmachine/VerseVDI-Data-Plane
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

fix(gateway): preserve qualification wire pacing

Browse Source
This commit is contained in:
sechmachine
2026-08-09 15:59:15 +07:00
parent 0b7e7b8b31
commit aa4f948fbc
4 changed files with 24 additions and 16 deletions
Download Patch File Download Diff File Expand all files Collapse all files
gateway/qualification_contract_test.go
+10 -4
View File
@@ -75,9 +75,15 @@ func TestQualificationApolloFixturePacesSourceShapedVideo(t *testing.T) {
if len(packets) != 1000 || len(packets[0]) != 1072 {
t.Fatalf("source vector = %d packets of %d bytes, want 1000 packets of 1072 bytes", len(packets), len(packets[0]))
}
packetsPerMillisecond, batchSize := qualificationApolloVideoPacing(len(packets[0]))
if packetsPerMillisecond != 93 || batchSize != 61 {
t.Fatalf("Apollo pacing vector = %d packets/ms, batch %d; want 93 and 61", packetsPerMillisecond, batchSize)
packetsPerMillisecond, batchSize := qualificationApolloVideoPacing(apolloVideoRawPacketSize)
if apolloVideoRawPacketSize != 1040 || packetsPerMillisecond != 96 || batchSize != 63 {
t.Fatalf("Apollo raw pacing vector = %d bytes, %d packets/ms, batch %d; want 1040, 96, and 63", apolloVideoRawPacketSize, packetsPerMillisecond, batchSize)
}
wantOffsets := []time.Duration{0, 656250 * time.Nanosecond, 1312500 * time.Nanosecond, 10416666 * time.Nanosecond}
for index, sent := range []int{0, 63, 126, 1000} {
if got := qualificationApolloVideoOffset(sent, packetsPerMillisecond); got != wantOffsets[index] {
t.Fatalf("Apollo pacing offset after %d packets = %s, want %s", sent, got, wantOffsets[index])
}
}
receiver, err := net.ListenUDP("udp", &net.UDPAddr{IP: net.ParseIP("127.0.0.1")})
@@ -113,7 +119,7 @@ func TestQualificationApolloFixturePacesSourceShapedVideo(t *testing.T) {
t.Fatal(err)
}
elapsed := time.Since(started)
wantCarry := 10 * time.Millisecond // floor(1000 / 93) ms at Apollo's pinned 80%-of-1-Gbps rate.
wantCarry := qualificationApolloVideoOffset(len(packets), packetsPerMillisecond)
if elapsed < wantCarry {
t.Fatalf("source fixture sent the next frame after %s, before Apollo pacing carry %s", elapsed, wantCarry)
}
gateway/qualification_harness_test.go
+10 -10
View File
@@ -762,7 +762,7 @@ func (f *qualificationApolloFixture) sendVideo(ctx context.Context, packets [][]
if len(packets) == 0 {
return nil
}
packetsPerMillisecond, batchSize := qualificationApolloVideoPacing(len(packets[0]))
packetsPerMillisecond, batchSize := qualificationApolloVideoPacing(apolloVideoRawPacketSize)
if packetsPerMillisecond == 0 || batchSize == 0 {
return ErrProviderMalformed
}
@@ -773,14 +773,11 @@ func (f *qualificationApolloFixture) sendVideo(ctx context.Context, packets [][]
if f.videoNext.After(frameStart) {
frameStart = f.videoNext
}
framePackets, groupPackets := 0, 0
framePackets := 0
for batchStart := 0; batchStart < len(packets); batchStart += batchSize {
if framePackets == 0 || groupPackets >= packetsPerMillisecond {
due := frameStart.Add(time.Millisecond * time.Duration(framePackets) / time.Duration(packetsPerMillisecond))
if err := qualificationWaitContext(ctx, due); err != nil {
return err
}
groupPackets = 0
due := frameStart.Add(qualificationApolloVideoOffset(framePackets, packetsPerMillisecond))
if err := qualificationWaitContext(ctx, due); err != nil {
return err
}
batchEnd := min(batchStart+batchSize, len(packets))
for _, packet := range packets[batchStart:batchEnd] {
@@ -794,9 +791,8 @@ func (f *qualificationApolloFixture) sendVideo(ctx context.Context, packets [][]
}
currentBatch := batchEnd - batchStart
framePackets += currentBatch
groupPackets += currentBatch
}
f.videoNext = frameStart.Add(time.Millisecond * time.Duration(framePackets) / time.Duration(packetsPerMillisecond))
f.videoNext = frameStart.Add(qualificationApolloVideoOffset(framePackets, packetsPerMillisecond))
return nil
}
@@ -809,6 +805,10 @@ func qualificationApolloVideoPacing(packetBytes int) (packetsPerMillisecond, bat
return packetsPerMillisecond, batchSize
}
func qualificationApolloVideoOffset(packets, packetsPerMillisecond int) time.Duration {
return time.Millisecond * time.Duration(packets) / time.Duration(packetsPerMillisecond)
}
func qualificationWaitContext(ctx context.Context, due time.Time) error {
delay := time.Until(due)
if delay <= 0 {