From 443f97edb0227d80e267e981e6b7e19358d80a78 Mon Sep 17 00:00:00 2001 From: neuronori Date: Mon, 29 Jun 2026 01:38:15 +0000 Subject: [PATCH 1/4] feat(vp8channel): expose tunable wire byte-rate via vp8.max_bytes_per_sec the pacer was pinned at a hardcoded 400kb/s stability compromise, so operators could not reach their sfu's real throughput ceiling without recompiling. the options.maxbytespersec knob already existed but was never wired from yaml. plumb it through config -> session -> transport, keeping 400kb/s as the conservative default. refs #107 --- .../client/client.telemost.vp8channel.yaml | 4 +++ .../server/server.telemost.vp8channel.yaml | 4 +++ docs/settings.md | 3 ++ docs/settings.ru.md | 3 ++ internal/app/session/session.go | 3 ++ internal/app/session/transport_options.go | 5 +-- internal/config/config.go | 7 +++++ internal/transport/vp8channel/options.go | 5 ++- .../vp8channel/transport_unit_test.go | 31 +++++++++++++++++++ 9 files changed, 62 insertions(+), 3 deletions(-) diff --git a/docs/examples/client/client.telemost.vp8channel.yaml b/docs/examples/client/client.telemost.vp8channel.yaml index 83e4984..0c8e2b1 100644 --- a/docs/examples/client/client.telemost.vp8channel.yaml +++ b/docs/examples/client/client.telemost.vp8channel.yaml @@ -33,6 +33,10 @@ socks: vp8: fps: 30 batch_size: 64 + # Потолок скорости на проводе (байт/сек). Дефолт 400000 (~400 КБ/с) - + # консервативно под Telemost. Подними под свой SFU, измерив стабильный + # максимум реальными запусками. 0 = дефолт. + # max_bytes_per_sec: 400000 data: data debug: false diff --git a/docs/examples/server/server.telemost.vp8channel.yaml b/docs/examples/server/server.telemost.vp8channel.yaml index b74c163..bc8ef75 100644 --- a/docs/examples/server/server.telemost.vp8channel.yaml +++ b/docs/examples/server/server.telemost.vp8channel.yaml @@ -34,6 +34,10 @@ socks: vp8: fps: 30 batch_size: 64 + # Потолок скорости на проводе (байт/сек). Дефолт 400000 (~400 КБ/с) - + # консервативно под Telemost. Подними под свой SFU, измерив стабильный + # максимум реальными запусками. 0 = дефолт. + # max_bytes_per_sec: 400000 data: data debug: false diff --git a/docs/settings.md b/docs/settings.md index 40e3a10..1959f94 100644 --- a/docs/settings.md +++ b/docs/settings.md @@ -152,6 +152,9 @@ No extra fields - everything is default. |-----------|----------|:------------:| | `vp8.fps` | VP8 stream FPS | `30` | | `vp8.batch_size` | Frames per tick | `64` | +| `vp8.max_bytes_per_sec` | Wire byte-rate ceiling, bytes/sec | `400000` | + +`vp8.max_bytes_per_sec` caps the byte-rate the pacer feeds to the video track. The default 400000 (400 KB/s) is a conservative ceiling tuned under the Telemost SFU policer knee: above it the SFU starts throttling forwarding and the stream stalls. Other services have a different knee, so raise this once you have measured your own stable maximum with real runs (bump it up until stalls start, then step back). `0` keeps the default. --- diff --git a/docs/settings.ru.md b/docs/settings.ru.md index 5ccf543..1efe2d5 100644 --- a/docs/settings.ru.md +++ b/docs/settings.ru.md @@ -153,6 +153,9 @@ transport. Используй одинаковые traffic-настройки н |-----------|----------|:------------:| | `vp8.fps` | FPS VP8 потока | `30` | | `vp8.batch_size` | Кадров за тик | `64` | +| `vp8.max_bytes_per_sec` | Потолок скорости на проводе, байт/сек | `400000` | + +`vp8.max_bytes_per_sec` ограничивает байтрейт, который пейсер льёт в видеотрек. Дефолт 400000 (400 КБ/с) - консервативный потолок под policer-колено Telemost SFU: выше него SFU начинает резать форвардинг и поток падает. У других сервисов колено другое, поэтому значение можно поднять, измерив свой стабильный максимум реальными запусками (поднимай постепенно, пока не начнутся падения, потом откатись на шаг назад). `0` оставляет дефолт. --- diff --git a/internal/app/session/session.go b/internal/app/session/session.go index 14b17ce..d07bf30 100644 --- a/internal/app/session/session.go +++ b/internal/app/session/session.go @@ -161,6 +161,9 @@ type VideoConfig struct { type VP8Config struct { FPS int BatchSize int + // MaxBytesPerSec caps the wire byte-rate. Zero falls back to the + // transport default. + MaxBytesPerSec int } // SEIConfig holds tunables for the seichannel transport. diff --git a/internal/app/session/transport_options.go b/internal/app/session/transport_options.go index 5f15484..5b7616e 100644 --- a/internal/app/session/transport_options.go +++ b/internal/app/session/transport_options.go @@ -27,8 +27,9 @@ func buildTransportOptions(cfg Config) transport.Options { } case transportVP8: return vp8channel.Options{ - FPS: cfg.VP8.FPS, - BatchSize: cfg.VP8.BatchSize, + FPS: cfg.VP8.FPS, + BatchSize: cfg.VP8.BatchSize, + MaxBytesPerSec: cfg.VP8.MaxBytesPerSec, } case transportSEI: return seichannel.Options{ diff --git a/internal/config/config.go b/internal/config/config.go index dca5863..adbb893 100644 --- a/internal/config/config.go +++ b/internal/config/config.go @@ -136,6 +136,11 @@ type Video struct { type VP8 struct { FPS int `yaml:"fps"` BatchSize int `yaml:"batch_size"` + // MaxBytesPerSec caps the wire byte-rate fed to the video track. The SFU + // policer knee differs per service, so this lets operators dial in the + // real ceiling instead of the conservative built-in default. Zero keeps + // the transport default. + MaxBytesPerSec int `yaml:"max_bytes_per_sec"` } // SEI tunes the seichannel transport. @@ -277,6 +282,7 @@ func Apply(dst session.Config, f File) session.Config { dst.Video.TileRS = pickInt(dst.Video.TileRS, f.Video.TileRS) dst.VP8.FPS = pickInt(dst.VP8.FPS, f.VP8.FPS) dst.VP8.BatchSize = pickInt(dst.VP8.BatchSize, f.VP8.BatchSize) + dst.VP8.MaxBytesPerSec = pickInt(dst.VP8.MaxBytesPerSec, f.VP8.MaxBytesPerSec) dst.SEI.FPS = pickInt(dst.SEI.FPS, f.SEI.FPS) dst.SEI.BatchSize = pickInt(dst.SEI.BatchSize, f.SEI.BatchSize) dst.SEI.FragmentSize = pickInt(dst.SEI.FragmentSize, f.SEI.FragmentSize) @@ -324,6 +330,7 @@ func ApplyProfile(base session.Config, p Profile) session.Config { dst.Video.TileRS = overlayInt(dst.Video.TileRS, p.Video.TileRS) dst.VP8.FPS = overlayInt(dst.VP8.FPS, p.VP8.FPS) dst.VP8.BatchSize = overlayInt(dst.VP8.BatchSize, p.VP8.BatchSize) + dst.VP8.MaxBytesPerSec = overlayInt(dst.VP8.MaxBytesPerSec, p.VP8.MaxBytesPerSec) dst.SEI.FPS = overlayInt(dst.SEI.FPS, p.SEI.FPS) dst.SEI.BatchSize = overlayInt(dst.SEI.BatchSize, p.SEI.BatchSize) dst.SEI.FragmentSize = overlayInt(dst.SEI.FragmentSize, p.SEI.FragmentSize) diff --git a/internal/transport/vp8channel/options.go b/internal/transport/vp8channel/options.go index 4d48681..65e3729 100644 --- a/internal/transport/vp8channel/options.go +++ b/internal/transport/vp8channel/options.go @@ -12,7 +12,10 @@ const ( // defaultMaxBytesPerSec paces the wire byte-rate just under the Telemost // SFU's measured per-slot policer knee. The original 1.2 MiB/s caused the // SFU to throttle subscriber forwarding after ~42s; 400 KB/s stays well - // within the SFU's comfort zone while still giving useful throughput. + // within the SFU's comfort zone while still giving useful throughput. This + // is a conservative floor: the policer knee differs per SFU, so operators + // can raise it via Options.MaxBytesPerSec (vp8.max_bytes_per_sec in YAML) + // once they have measured their own service's stable ceiling. defaultMaxBytesPerSec = 400_000 ) diff --git a/internal/transport/vp8channel/transport_unit_test.go b/internal/transport/vp8channel/transport_unit_test.go index 740fb26..d10238e 100644 --- a/internal/transport/vp8channel/transport_unit_test.go +++ b/internal/transport/vp8channel/transport_unit_test.go @@ -614,6 +614,37 @@ func TestReorderBufferRestoresOrderAndSurvivesLoss(t *testing.T) { } } +// TestMaxBytesPerSecPacing verifies the operator-tunable wire rate cap flows +// into the per-tick byte budget that paces the writer. The default keeps the +// conservative built-in ceiling; an explicit value lets operators dial in +// their own SFU's stable maximum instead of being pinned at the floor. See +// issue #107. +func TestMaxBytesPerSecPacing(t *testing.T) { + cases := []struct { + name string + maxBytes int + fps int + wantPerTick int + }{ + {name: "default floor", maxBytes: 0, fps: 30, wantPerTick: defaultMaxBytesPerSec / 30}, + {name: "explicit raise", maxBytes: 1_000_000, fps: 40, wantPerTick: 1_000_000 / 40}, + {name: "clamped to header", maxBytes: 1, fps: 30, wantPerTick: epochHdrLen}, + } + for _, c := range cases { + t.Run(c.name, func(t *testing.T) { + tr := newStreamTransport( + &engineVideoSession{}, + nil, + transport.Config{}, + Options{FPS: c.fps, BatchSize: 1, MaxBytesPerSec: c.maxBytes}, + ) + if tr.perTickBytes != c.wantPerTick { + t.Fatalf("perTickBytes = %d, want %d", tr.perTickBytes, c.wantPerTick) + } + }) + } +} + func TestSeqLessWrapAround(t *testing.T) { cases := []struct { a, b uint16 From 2dffb5acbeb35b5ac973fbfbf227cd6db3d5e563 Mon Sep 17 00:00:00 2001 From: neuronori Date: Mon, 29 Jun 2026 02:25:28 +0000 Subject: [PATCH 2/4] feat(vp8channel): auto-discover wire rate via delay-based pacer the hardcoded 400kb/s was a stability compromise and the previous knob just moved the guesswork to operators: they had to hand-measure each sfu's policer knee by ramping until stalls. replace both with a delay-based aimd controller that finds the knee at runtime. the sfu policer queues before it drops, so kcp's smoothed rtt inflates ahead of the throughput collapse from #95/#107. the pacer folds one srtt sample per writer tick: probe the rate up additively while delay sits near the path baseline, back off multiplicatively once it inflates. it settles just under the per-sfu knee on its own, both on the client->server writerLoop and each server->client peer pump. vp8.max_bytes_per_sec is now just the probe ceiling (default 1mb/s); a fresh session still starts from the old 400kb/s operating point so a healthy path only ramps up. no per-service hand-tuning left. refs #107 --- .../client/client.telemost.vp8channel.yaml | 8 +- .../server/server.telemost.vp8channel.yaml | 8 +- docs/settings.md | 6 +- docs/settings.ru.md | 6 +- internal/transport/vp8channel/kcp.go | 10 ++ internal/transport/vp8channel/options.go | 36 ++++-- internal/transport/vp8channel/pacer.go | 81 +++++++++++++ internal/transport/vp8channel/pacer_test.go | 110 ++++++++++++++++++ internal/transport/vp8channel/transport.go | 84 ++++++++++--- .../vp8channel/transport_unit_test.go | 29 ++--- 10 files changed, 327 insertions(+), 51 deletions(-) create mode 100644 internal/transport/vp8channel/pacer.go create mode 100644 internal/transport/vp8channel/pacer_test.go diff --git a/docs/examples/client/client.telemost.vp8channel.yaml b/docs/examples/client/client.telemost.vp8channel.yaml index 0c8e2b1..88c19e5 100644 --- a/docs/examples/client/client.telemost.vp8channel.yaml +++ b/docs/examples/client/client.telemost.vp8channel.yaml @@ -33,10 +33,10 @@ socks: vp8: fps: 30 batch_size: 64 - # Потолок скорости на проводе (байт/сек). Дефолт 400000 (~400 КБ/с) - - # консервативно под Telemost. Подними под свой SFU, измерив стабильный - # максимум реальными запусками. 0 = дефолт. - # max_bytes_per_sec: 400000 + # Потолок, до которого адаптивный пейсер может разгоняться (байт/сек). + # Скорость подбирается автоматически по задержке пути - под колено SFU. + # Дефолт 1000000 (~1 МБ/с). Опусти, чтобы жёстко лимитировать. 0 = дефолт. + # max_bytes_per_sec: 1000000 data: data debug: false diff --git a/docs/examples/server/server.telemost.vp8channel.yaml b/docs/examples/server/server.telemost.vp8channel.yaml index bc8ef75..9bf978d 100644 --- a/docs/examples/server/server.telemost.vp8channel.yaml +++ b/docs/examples/server/server.telemost.vp8channel.yaml @@ -34,10 +34,10 @@ socks: vp8: fps: 30 batch_size: 64 - # Потолок скорости на проводе (байт/сек). Дефолт 400000 (~400 КБ/с) - - # консервативно под Telemost. Подними под свой SFU, измерив стабильный - # максимум реальными запусками. 0 = дефолт. - # max_bytes_per_sec: 400000 + # Потолок, до которого адаптивный пейсер может разгоняться (байт/сек). + # Скорость подбирается автоматически по задержке пути - под колено SFU. + # Дефолт 1000000 (~1 МБ/с). Опусти, чтобы жёстко лимитировать. 0 = дефолт. + # max_bytes_per_sec: 1000000 data: data debug: false diff --git a/docs/settings.md b/docs/settings.md index 1959f94..dcdc1e6 100644 --- a/docs/settings.md +++ b/docs/settings.md @@ -152,9 +152,11 @@ No extra fields - everything is default. |-----------|----------|:------------:| | `vp8.fps` | VP8 stream FPS | `30` | | `vp8.batch_size` | Frames per tick | `64` | -| `vp8.max_bytes_per_sec` | Wire byte-rate ceiling, bytes/sec | `400000` | +| `vp8.max_bytes_per_sec` | Wire byte-rate probe ceiling, bytes/sec | `1000000` | -`vp8.max_bytes_per_sec` caps the byte-rate the pacer feeds to the video track. The default 400000 (400 KB/s) is a conservative ceiling tuned under the Telemost SFU policer knee: above it the SFU starts throttling forwarding and the stream stalls. Other services have a different knee, so raise this once you have measured your own stable maximum with real runs (bump it up until stalls start, then step back). `0` keeps the default. +The wire rate is paced by a delay-based adaptive controller that auto-discovers each SFU's policer knee at runtime. The policer queues before it drops, so the path RTT inflates ahead of the throughput collapse; the pacer probes the rate up while RTT stays near the path baseline and backs off once it inflates, settling just under the knee on its own. There is nothing to hand-measure per service. + +`vp8.max_bytes_per_sec` only caps how high the controller may probe (default 1000000, ~1 MB/s). Lower it to hard-limit a path; `0` keeps the built-in cap. A fresh session starts probing from 400 KB/s (the previous conservative default), so a healthy path only ramps up from the old operating point. --- diff --git a/docs/settings.ru.md b/docs/settings.ru.md index 1efe2d5..2446c27 100644 --- a/docs/settings.ru.md +++ b/docs/settings.ru.md @@ -153,9 +153,11 @@ transport. Используй одинаковые traffic-настройки н |-----------|----------|:------------:| | `vp8.fps` | FPS VP8 потока | `30` | | `vp8.batch_size` | Кадров за тик | `64` | -| `vp8.max_bytes_per_sec` | Потолок скорости на проводе, байт/сек | `400000` | +| `vp8.max_bytes_per_sec` | Потолок, до которого пейсер пробует разгоняться, байт/сек | `1000000` | -`vp8.max_bytes_per_sec` ограничивает байтрейт, который пейсер льёт в видеотрек. Дефолт 400000 (400 КБ/с) - консервативный потолок под policer-колено Telemost SFU: выше него SFU начинает резать форвардинг и поток падает. У других сервисов колено другое, поэтому значение можно поднять, измерив свой стабильный максимум реальными запусками (поднимай постепенно, пока не начнутся падения, потом откатись на шаг назад). `0` оставляет дефолт. +Скорость на проводе регулирует адаптивный контроллер на основе задержки: он сам находит policer-колено каждого SFU в рантайме. Policer начинает копить очередь раньше, чем дропать, поэтому RTT пути растёт до того, как обвалится скорость. Пейсер разгоняется, пока RTT держится у базовой линии пути, и сбрасывает скорость, когда RTT раздувается, сам оседая чуть ниже колена. Ничего измерять руками под конкретный сервис не нужно. + +`vp8.max_bytes_per_sec` лишь ограничивает, до какого потолка контроллер может разгоняться (дефолт 1000000, ~1 МБ/с). Опусти, чтобы жёстко лимитировать путь; `0` оставляет встроенный потолок. Свежая сессия стартует с 400 КБ/с (прежний консервативный дефолт), так что здоровый путь только разгоняется от старой рабочей точки. --- diff --git a/internal/transport/vp8channel/kcp.go b/internal/transport/vp8channel/kcp.go index cbba443..566d931 100644 --- a/internal/transport/vp8channel/kcp.go +++ b/internal/transport/vp8channel/kcp.go @@ -164,6 +164,16 @@ func (r *kcpRuntime) send(msg []byte) error { return nil } +// srtt reports KCP's smoothed round-trip estimate in milliseconds, or 0 when +// no sample exists yet. The adaptive pacer reads this to detect the policer +// queueing that precedes a throughput collapse. +func (r *kcpRuntime) srtt() int32 { + if r == nil || r.sess == nil { + return 0 + } + return r.sess.GetSRTT() +} + func (r *kcpRuntime) close() { r.closeOnce.Do(func() { _ = r.sess.Close() diff --git a/internal/transport/vp8channel/options.go b/internal/transport/vp8channel/options.go index 65e3729..333346c 100644 --- a/internal/transport/vp8channel/options.go +++ b/internal/transport/vp8channel/options.go @@ -9,14 +9,34 @@ import ( const ( defaultFPS = 30 defaultBatchSize = 64 - // defaultMaxBytesPerSec paces the wire byte-rate just under the Telemost - // SFU's measured per-slot policer knee. The original 1.2 MiB/s caused the - // SFU to throttle subscriber forwarding after ~42s; 400 KB/s stays well - // within the SFU's comfort zone while still giving useful throughput. This - // is a conservative floor: the policer knee differs per SFU, so operators - // can raise it via Options.MaxBytesPerSec (vp8.max_bytes_per_sec in YAML) - // once they have measured their own service's stable ceiling. - defaultMaxBytesPerSec = 400_000 + // defaultMaxBytesPerSec is the upper bound the adaptive pacer may probe up + // to when no explicit ceiling is configured. The pacer (see pacer.go) + // auto-discovers each SFU's real policer knee at runtime from KCP's + // smoothed RTT, so this is just a sanity cap, not a hand-tuned operating + // point. 1 MiB/s matches the throughput target from issue #107 without + // letting a misbehaving path probe unbounded. + defaultMaxBytesPerSec = 1_000_000 + + // defaultMinProbeRate floors the adaptive rate so a congested path can + // always make forward progress and recover once delay drops. ~120 KB/s + // stays below the worst observed Telemost knee while keeping the control + // plane and keepalives flowing. + defaultMinProbeRate = 120_000 + + // defaultStartRate is where the pacer begins probing from on a fresh + // session: the old conservative 400 KB/s floor, so behaviour on a healthy + // path only ever ramps up from the previously shipped operating point. + defaultStartRate = 400_000 + + // Adaptive pacer tuning. Delay marks are derived from the measured path + // baseline plus a fixed slack so a low-RTT LAN and a high-RTT WAN both get + // a sane congestion threshold. + rateProbeSlackMs = 8 // extra ms below which we probe the rate up + rateCongestionSlackMs = 25 // extra ms above which we back the rate off + rateProbeStepBytes = 50_000 + rateBackoffNum = 4 // multiplicative decrease: rate *= 4/5 + rateBackoffDen = 5 + baseRTTDriftDiv = 64 // baseline rises by 1/64 of the gap per sample ) // Options tunes the vp8channel transport. Zero values fall back to documented defaults. diff --git a/internal/transport/vp8channel/pacer.go b/internal/transport/vp8channel/pacer.go new file mode 100644 index 0000000..66c0a22 --- /dev/null +++ b/internal/transport/vp8channel/pacer.go @@ -0,0 +1,81 @@ +package vp8channel + +// ratePacer is a delay-based AIMD controller that auto-discovers the wire +// byte-rate a policed carrier (the SFU) tolerates, with no operator tuning. +// +// The carrier policer starts queueing before it starts dropping, so KCP's +// smoothed RTT inflates ahead of the throughput collapse seen in issues #95 +// and #107. Each control tick the pacer folds one SRTT sample in: while the +// delay stays near the path baseline it probes the rate up additively, and +// once the delay inflates past the baseline it backs off multiplicatively. +// The rate settles just under the per-SFU policer knee on its own, which is +// the "real run" compromise the fixed 400 KB/s floor only approximated. The +// operator ceiling (vp8.max_bytes_per_sec) just bounds how high it may probe. +type ratePacer struct { + fps int + minRate int + maxRate int + rate int + baseRTT int32 +} + +func newRatePacer(fps, startRate, minRate, maxRate int) *ratePacer { + if fps <= 0 { + fps = defaultFPS + } + if minRate <= 0 { + minRate = defaultMinProbeRate + } + if maxRate < minRate { + maxRate = minRate + } + return &ratePacer{ + fps: fps, + minRate: minRate, + maxRate: maxRate, + rate: max(minRate, min(startRate, maxRate)), + } +} + +// perTick is the current per-frame byte budget. The ticker already paces at +// fps, so a per-tick cap bounds the rate without token bookkeeping. Floor at +// one epoch header so keepalives always fit. +func (rp *ratePacer) perTick() int { + pt := rp.rate / rp.fps + if pt < epochHdrLen { + return epochHdrLen + } + return pt +} + +// observe folds one smoothed-RTT sample (milliseconds, as KCP reports it) into +// the controller and returns the updated per-tick byte budget. A non-positive +// sample means KCP has no RTT estimate yet, so the rate is held steady. +func (rp *ratePacer) observe(srtt int32) int { + if srtt <= 0 { + return rp.perTick() + } + rp.trackBaseline(srtt) + + lowMark := rp.baseRTT + rp.baseRTT/4 + rateProbeSlackMs + highMark := rp.baseRTT + rp.baseRTT/2 + rateCongestionSlackMs + switch { + case srtt >= highMark: + rp.rate = max(rp.minRate, min(rp.rate*rateBackoffNum/rateBackoffDen, rp.maxRate)) + case srtt <= lowMark: + rp.rate = max(rp.minRate, min(rp.rate+rateProbeStepBytes, rp.maxRate)) + } + return rp.perTick() +} + +// trackBaseline anchors baseRTT to the path's uncongested delay: it drops +// instantly to any new minimum and drifts up slowly otherwise, so a genuine +// path change cannot pin the baseline below the real floor forever while a +// transient queueing spike still reads as congestion. +func (rp *ratePacer) trackBaseline(srtt int32) { + if rp.baseRTT == 0 || srtt < rp.baseRTT { + rp.baseRTT = srtt + return + } + rp.baseRTT += (srtt - rp.baseRTT) / baseRTTDriftDiv +} diff --git a/internal/transport/vp8channel/pacer_test.go b/internal/transport/vp8channel/pacer_test.go new file mode 100644 index 0000000..5424987 --- /dev/null +++ b/internal/transport/vp8channel/pacer_test.go @@ -0,0 +1,110 @@ +package vp8channel + +import "testing" + +// TestRatePacerProbesUpOnLowDelay verifies the pacer additively raises the +// rate while the path delay stays near the baseline, up to the ceiling. +func TestRatePacerProbesUpOnLowDelay(t *testing.T) { + rp := newRatePacer(30, defaultStartRate, defaultMinProbeRate, defaultMaxBytesPerSec) + + // First sample establishes the baseline (20ms). Subsequent equal samples + // read as uncongested, so the rate climbs by rateProbeStepBytes each tick. + rp.observe(20) + start := rp.rate + for range 5 { + rp.observe(20) + } + if rp.rate <= start { + t.Fatalf("rate did not climb on low delay: start=%d now=%d", start, rp.rate) + } + if rp.rate > rp.maxRate { + t.Fatalf("rate %d exceeded ceiling %d", rp.rate, rp.maxRate) + } +} + +// TestRatePacerBacksOffOnHighDelay verifies a delay spike past the congestion +// mark triggers a multiplicative decrease. +func TestRatePacerBacksOffOnHighDelay(t *testing.T) { + rp := newRatePacer(30, defaultStartRate, defaultMinProbeRate, defaultMaxBytesPerSec) + rp.observe(20) // baseline 20ms + before := rp.rate + + // 20ms baseline -> highMark = 20 + 10 + 25 = 55ms. A 120ms sample is well + // past it and must shrink the rate. + rp.observe(120) + if rp.rate >= before { + t.Fatalf("rate did not back off on high delay: before=%d after=%d", before, rp.rate) + } + want := before * rateBackoffNum / rateBackoffDen + if rp.rate != want { + t.Fatalf("backoff rate = %d, want %d", rp.rate, want) + } +} + +// TestRatePacerConvergesToKnee drives a synthetic policer: delay stays low +// until the rate crosses a hidden knee, then inflates. The pacer must settle +// in a band around that knee instead of running away or collapsing. +func TestRatePacerConvergesToKnee(t *testing.T) { + const knee = 600_000 + rp := newRatePacer(30, defaultStartRate, defaultMinProbeRate, defaultMaxBytesPerSec) + + srtt := func(rate int) int32 { + // Below the knee the path is idle (20ms). Above it the policer queues, + // inflating delay in proportion to the overshoot. + if rate <= knee { + return 20 + } + return 20 + int32(min((rate-knee)/10_000, 1_000)) //nolint:gosec // G115: bounded by min to 1000 + } + + for range 400 { + rp.observe(srtt(rp.rate)) + } + + // After convergence the rate must sit in a sane band around the knee: + // high enough to beat the old 400 KB/s floor, never pinned at the ceiling. + if rp.rate < defaultStartRate { + t.Fatalf("converged rate %d below start floor %d", rp.rate, defaultStartRate) + } + if rp.rate >= rp.maxRate { + t.Fatalf("rate pinned at ceiling %d, did not detect knee", rp.rate) + } + if rp.rate > knee+4*rateProbeStepBytes { + t.Fatalf("converged rate %d overshoots knee %d", rp.rate, knee) + } +} + +// TestRatePacerHoldsWithoutSample verifies a non-positive SRTT (no estimate +// yet) holds the rate steady rather than probing blind. +func TestRatePacerHoldsWithoutSample(t *testing.T) { + rp := newRatePacer(30, defaultStartRate, defaultMinProbeRate, defaultMaxBytesPerSec) + before := rp.rate + if got := rp.observe(0); got != before/rp.fps { + t.Fatalf("perTick on no-sample = %d, want %d", got, before/rp.fps) + } + if rp.rate != before { + t.Fatalf("rate moved without a sample: before=%d after=%d", before, rp.rate) + } +} + +// TestRatePacerNeverBelowFloor verifies sustained congestion cannot starve the +// rate below the minimum that keeps the control plane and keepalives flowing. +func TestRatePacerNeverBelowFloor(t *testing.T) { + rp := newRatePacer(30, defaultStartRate, defaultMinProbeRate, defaultMaxBytesPerSec) + rp.observe(10) + for range 100 { + rp.observe(5000) // permanent heavy congestion + } + if rp.rate < rp.minRate { + t.Fatalf("rate %d fell below floor %d", rp.rate, rp.minRate) + } +} + +// TestRatePacerPerTickFloor verifies the per-tick budget never drops below one +// epoch header so keepalives always fit even at the rate floor. +func TestRatePacerPerTickFloor(t *testing.T) { + rp := newRatePacer(30, epochHdrLen, epochHdrLen, epochHdrLen) + if got := rp.perTick(); got < epochHdrLen { + t.Fatalf("perTick = %d, want >= %d", got, epochHdrLen) + } +} diff --git a/internal/transport/vp8channel/transport.go b/internal/transport/vp8channel/transport.go index 170ef57..e2c7794 100644 --- a/internal/transport/vp8channel/transport.go +++ b/internal/transport/vp8channel/transport.go @@ -155,7 +155,14 @@ type streamTransport struct { controlKCPOnce sync.Once frameInterval time.Duration batchSize int - perTickBytes int + // Adaptive pacer bounds (bytes/sec). Each writer goroutine builds its own + // ratePacer from these and probes the live wire rate from KCP's SRTT, so + // the operating point auto-tracks each SFU's policer knee. startRate is + // the conservative point a fresh session begins probing from; maxRate is + // the operator ceiling (vp8.max_bytes_per_sec) or the built-in cap. + startRate int + minRate int + maxRate int // localEpoch is stamped into every outgoing VP8 frame. Explicit // upper-layer resets rotate it so the peer can reset its KCP state too. @@ -284,16 +291,13 @@ func newStreamTransport( if batchSize <= 0 { batchSize = defaultBatchSize } - byteRate := opts.MaxBytesPerSec - if byteRate <= 0 { - byteRate = defaultMaxBytesPerSec - } - // Bytes we may emit per frame tick to hold the wire under byteRate. The - // ticker already paces at fps, so a per-tick cap bounds the rate without - // any token bookkeeping. Floor at one epoch header so keepalives fit. - perTickBytes := byteRate / fps - if perTickBytes < epochHdrLen { - perTickBytes = epochHdrLen + // maxRate caps how high the adaptive pacer may probe. An explicit + // vp8.max_bytes_per_sec becomes that ceiling; otherwise the built-in cap + // applies. The pacer auto-discovers the real operating point under it from + // KCP's SRTT, so no value here needs hand-tuning to a specific SFU. + maxRate := opts.MaxBytesPerSec + if maxRate <= 0 { + maxRate = defaultMaxBytesPerSec } tr := &streamTransport{ @@ -307,7 +311,9 @@ func newStreamTransport( writerDone: make(chan struct{}), frameInterval: time.Second / time.Duration(fps), batchSize: batchSize, - perTickBytes: perTickBytes, + startRate: defaultStartRate, + minRate: defaultMinProbeRate, + maxRate: maxRate, bindingToken: bindingToken(cfg.RoomURL), localEpoch: randomEpoch(), peers: make(map[uint32]*kcpRuntime), @@ -335,6 +341,24 @@ func newStreamTransport( return tr } +// fps recovers the frame rate from the configured frame interval. The pacer +// divides the per-second rate by this to get a per-tick byte budget. +func (p *streamTransport) fps() int { + if p.frameInterval <= 0 { + return defaultFPS + } + return max(int(time.Second/p.frameInterval), 1) +} + +// dataSRTT reports the smoothed RTT (ms) of the single-peer data KCP session, +// or 0 when no session or sample exists yet. +func (p *streamTransport) dataSRTT() int32 { + p.kcpMu.RLock() + rt := p.kcp + p.kcpMu.RUnlock() + return rt.srtt() +} + func (p *streamTransport) Connect(ctx context.Context) error { connectCtx, cancel := context.WithTimeout(ctx, defaultConnectTimeout) defer cancel() @@ -719,14 +743,33 @@ func (p *streamTransport) Features() transport.Features { type writerState struct { p *streamTransport keepaliveEvery int - idleTicks int forceKeepaliveEvery int + idleTicks int ticksSinceKeepalive int + // pacer adapts the per-tick byte budget from the live path delay so the + // wire rate tracks the SFU's policer knee without operator tuning. + pacer *ratePacer + // srttFn reports the smoothed RTT (ms) of the KCP session this writer + // feeds, so the pacer can read congestion from the right plane. + srttFn func() int32 // pendingControl holds a control frame that failed WriteSample and must be // retried on the next tick before consuming more frames. pendingControl []byte } +// perTick folds the current path delay into the pacer and returns the byte +// budget for this tick. A nil pacer (defensive) falls back to the start rate. +func (w *writerState) perTick() int { + if w.pacer == nil { + return epochHdrLen + } + var srtt int32 + if w.srttFn != nil { + srtt = w.srttFn() + } + return w.pacer.observe(srtt) +} + func (w *writerState) writeSample(data []byte) bool { return w.p.writeSampleLocked(data) } @@ -798,7 +841,7 @@ func (w *writerState) drainControl() bool { func (w *writerState) drainData() { select { case frame := <-w.p.outbound: - sample := w.p.batchSample(frame, w.p.perTickBytes) + sample := w.p.batchSample(frame, w.perTick()) w.idleTicks = 0 _ = w.writeSample(sample) default: @@ -821,6 +864,8 @@ func (p *streamTransport) writerLoop() { p: p, keepaliveEvery: max(int(keepaliveIdlePeriod/p.frameInterval), 1), forceKeepaliveEvery: max(int((2*time.Second)/p.frameInterval), 1), + pacer: newRatePacer(p.fps(), p.startRate, p.minRate, p.maxRate), + srttFn: p.dataSRTT, } for { @@ -1348,7 +1393,7 @@ func (p *streamTransport) getOrCreatePeerKCP(epoch uint32) *kcpRuntime { logger.Infof("vp8channel: peer session created epoch=0x%08x", epoch) // Pump outbound frames from this peer's queue into the writer. - go p.peerWriterPump(epoch, out) + go p.peerWriterPump(rt, out) return rt } @@ -1361,10 +1406,15 @@ func (p *streamTransport) getOrCreatePeerKCP(epoch uint32) *kcpRuntime { // overran the SFU's policer, drove burst loss, and collapsed throughput to // zero within ~20-40s (issue #95). Stops when the channel is closed or the // transport shuts down. -func (p *streamTransport) peerWriterPump(_ uint32, out chan []byte) { +func (p *streamTransport) peerWriterPump(rt *kcpRuntime, out chan []byte) { ticker := time.NewTicker(p.frameInterval) defer ticker.Stop() + // Each downlink peer paces independently: its own ratePacer reads that + // peer's KCP SRTT and tracks its path's policer knee without affecting + // the others or the client->server path. + pacer := newRatePacer(p.fps(), p.startRate, p.minRate, p.maxRate) + // Inject a decodable VP8 keyframe on the same cadence writerLoop uses for // the client->server path. The server's per-peer bulk path previously // emitted only opaque KCP data frames, which never form a decodable VP8 @@ -1392,7 +1442,7 @@ func (p *streamTransport) peerWriterPump(_ uint32, out chan []byte) { if !ok { return } - sample := p.batchSampleFrom(out, frame, p.perTickBytes) + sample := p.batchSampleFrom(out, frame, pacer.observe(rt.srtt())) _ = p.writeSampleLocked(sample) default: } diff --git a/internal/transport/vp8channel/transport_unit_test.go b/internal/transport/vp8channel/transport_unit_test.go index d10238e..8c42d79 100644 --- a/internal/transport/vp8channel/transport_unit_test.go +++ b/internal/transport/vp8channel/transport_unit_test.go @@ -614,21 +614,19 @@ func TestReorderBufferRestoresOrderAndSurvivesLoss(t *testing.T) { } } -// TestMaxBytesPerSecPacing verifies the operator-tunable wire rate cap flows -// into the per-tick byte budget that paces the writer. The default keeps the -// conservative built-in ceiling; an explicit value lets operators dial in -// their own SFU's stable maximum instead of being pinned at the floor. See -// issue #107. -func TestMaxBytesPerSecPacing(t *testing.T) { +// TestMaxBytesPerSecCeiling verifies vp8.max_bytes_per_sec becomes the upper +// bound the adaptive pacer may probe up to, while zero falls back to the +// built-in cap. The pacer auto-discovers the real operating point under that +// ceiling from KCP's SRTT, so the option is a safety bound, not a hand-tuned +// rate. See issue #107. +func TestMaxBytesPerSecCeiling(t *testing.T) { cases := []struct { name string maxBytes int - fps int - wantPerTick int + wantMaxRate int }{ - {name: "default floor", maxBytes: 0, fps: 30, wantPerTick: defaultMaxBytesPerSec / 30}, - {name: "explicit raise", maxBytes: 1_000_000, fps: 40, wantPerTick: 1_000_000 / 40}, - {name: "clamped to header", maxBytes: 1, fps: 30, wantPerTick: epochHdrLen}, + {name: "default cap", maxBytes: 0, wantMaxRate: defaultMaxBytesPerSec}, + {name: "explicit ceiling", maxBytes: 2_000_000, wantMaxRate: 2_000_000}, } for _, c := range cases { t.Run(c.name, func(t *testing.T) { @@ -636,10 +634,13 @@ func TestMaxBytesPerSecPacing(t *testing.T) { &engineVideoSession{}, nil, transport.Config{}, - Options{FPS: c.fps, BatchSize: 1, MaxBytesPerSec: c.maxBytes}, + Options{FPS: 30, BatchSize: 1, MaxBytesPerSec: c.maxBytes}, ) - if tr.perTickBytes != c.wantPerTick { - t.Fatalf("perTickBytes = %d, want %d", tr.perTickBytes, c.wantPerTick) + if tr.maxRate != c.wantMaxRate { + t.Fatalf("maxRate = %d, want %d", tr.maxRate, c.wantMaxRate) + } + if tr.startRate != defaultStartRate { + t.Fatalf("startRate = %d, want %d", tr.startRate, defaultStartRate) } }) } From 1e270de44a8e62f6600c574a34c333454d340b26 Mon Sep 17 00:00:00 2001 From: neuronori Date: Mon, 29 Jun 2026 05:50:36 +0000 Subject: [PATCH 3/4] fix(vp8channel): drop delay-based pacer, run static rate at 1mb/s the aimd pacer read kcp's GetSRTT as its congestion signal, but the data kcp runs with nc=1 and a fixed 512-segment (~716kb) window it keeps full on purpose (kcp.go: data queues ~2-3s). that self-inflicted bufferbloat dominates srtt no matter where the rate sits relative to the sfu policer knee, so every tick tripped the high-delay mark and the rate decayed to the 120kb/s floor. the result was slower, not faster. the srtt signal is unusable as a knee detector on this carrier, so drop adaptive control and keep the simple static pacer. raise the default to the 1mb/s target from #107 (the old ~42s collapse at 1.2mib/s was the keyframe-starvation drop fixed in #95, not a raw rate ceiling), still tunable via vp8.max_bytes_per_sec. refs #107 --- .../client/client.telemost.vp8channel.yaml | 6 +- .../server/server.telemost.vp8channel.yaml | 6 +- docs/settings.md | 6 +- docs/settings.ru.md | 6 +- internal/transport/vp8channel/kcp.go | 10 -- internal/transport/vp8channel/options.go | 34 ++---- internal/transport/vp8channel/pacer.go | 81 ------------- internal/transport/vp8channel/pacer_test.go | 110 ------------------ internal/transport/vp8channel/transport.go | 84 +++---------- .../vp8channel/transport_unit_test.go | 28 +++-- 10 files changed, 47 insertions(+), 324 deletions(-) delete mode 100644 internal/transport/vp8channel/pacer.go delete mode 100644 internal/transport/vp8channel/pacer_test.go diff --git a/docs/examples/client/client.telemost.vp8channel.yaml b/docs/examples/client/client.telemost.vp8channel.yaml index 88c19e5..d69f912 100644 --- a/docs/examples/client/client.telemost.vp8channel.yaml +++ b/docs/examples/client/client.telemost.vp8channel.yaml @@ -33,9 +33,9 @@ socks: vp8: fps: 30 batch_size: 64 - # Потолок, до которого адаптивный пейсер может разгоняться (байт/сек). - # Скорость подбирается автоматически по задержке пути - под колено SFU. - # Дефолт 1000000 (~1 МБ/с). Опусти, чтобы жёстко лимитировать. 0 = дефолт. + # Потолок скорости на проводе (байт/сек). Дефолт 1000000 (~1 МБ/с). + # Policer-колено у каждого SFU своё: если поток падает - снизь до + # стабильной скорости, на здоровом пути можно поднять. 0 = дефолт. # max_bytes_per_sec: 1000000 data: data diff --git a/docs/examples/server/server.telemost.vp8channel.yaml b/docs/examples/server/server.telemost.vp8channel.yaml index 9bf978d..f951c61 100644 --- a/docs/examples/server/server.telemost.vp8channel.yaml +++ b/docs/examples/server/server.telemost.vp8channel.yaml @@ -34,9 +34,9 @@ socks: vp8: fps: 30 batch_size: 64 - # Потолок, до которого адаптивный пейсер может разгоняться (байт/сек). - # Скорость подбирается автоматически по задержке пути - под колено SFU. - # Дефолт 1000000 (~1 МБ/с). Опусти, чтобы жёстко лимитировать. 0 = дефолт. + # Потолок скорости на проводе (байт/сек). Дефолт 1000000 (~1 МБ/с). + # Policer-колено у каждого SFU своё: если поток падает - снизь до + # стабильной скорости, на здоровом пути можно поднять. 0 = дефолт. # max_bytes_per_sec: 1000000 data: data diff --git a/docs/settings.md b/docs/settings.md index dcdc1e6..b62431e 100644 --- a/docs/settings.md +++ b/docs/settings.md @@ -152,11 +152,9 @@ No extra fields - everything is default. |-----------|----------|:------------:| | `vp8.fps` | VP8 stream FPS | `30` | | `vp8.batch_size` | Frames per tick | `64` | -| `vp8.max_bytes_per_sec` | Wire byte-rate probe ceiling, bytes/sec | `1000000` | +| `vp8.max_bytes_per_sec` | Wire byte-rate ceiling, bytes/sec | `1000000` | -The wire rate is paced by a delay-based adaptive controller that auto-discovers each SFU's policer knee at runtime. The policer queues before it drops, so the path RTT inflates ahead of the throughput collapse; the pacer probes the rate up while RTT stays near the path baseline and backs off once it inflates, settling just under the knee on its own. There is nothing to hand-measure per service. - -`vp8.max_bytes_per_sec` only caps how high the controller may probe (default 1000000, ~1 MB/s). Lower it to hard-limit a path; `0` keeps the built-in cap. A fresh session starts probing from 400 KB/s (the previous conservative default), so a healthy path only ramps up from the old operating point. +`vp8.max_bytes_per_sec` caps the byte-rate the pacer feeds to the video track. The default 1000000 (1 MB/s) matches the issue #107 throughput target. The per-SFU policer knee still varies, so if a service stalls at this rate, lower it until stable; on a healthy path you can raise it. `0` keeps the default. --- diff --git a/docs/settings.ru.md b/docs/settings.ru.md index 2446c27..295c3bf 100644 --- a/docs/settings.ru.md +++ b/docs/settings.ru.md @@ -153,11 +153,9 @@ transport. Используй одинаковые traffic-настройки н |-----------|----------|:------------:| | `vp8.fps` | FPS VP8 потока | `30` | | `vp8.batch_size` | Кадров за тик | `64` | -| `vp8.max_bytes_per_sec` | Потолок, до которого пейсер пробует разгоняться, байт/сек | `1000000` | +| `vp8.max_bytes_per_sec` | Потолок скорости на проводе, байт/сек | `1000000` | -Скорость на проводе регулирует адаптивный контроллер на основе задержки: он сам находит policer-колено каждого SFU в рантайме. Policer начинает копить очередь раньше, чем дропать, поэтому RTT пути растёт до того, как обвалится скорость. Пейсер разгоняется, пока RTT держится у базовой линии пути, и сбрасывает скорость, когда RTT раздувается, сам оседая чуть ниже колена. Ничего измерять руками под конкретный сервис не нужно. - -`vp8.max_bytes_per_sec` лишь ограничивает, до какого потолка контроллер может разгоняться (дефолт 1000000, ~1 МБ/с). Опусти, чтобы жёстко лимитировать путь; `0` оставляет встроенный потолок. Свежая сессия стартует с 400 КБ/с (прежний консервативный дефолт), так что здоровый путь только разгоняется от старой рабочей точки. +`vp8.max_bytes_per_sec` ограничивает байтрейт, который пейсер льёт в видеотрек. Дефолт 1000000 (1 МБ/с) - целевая скорость из issue #107. Policer-колено у каждого SFU своё, поэтому если на этой скорости поток падает - снижай до стабильной; на здоровом пути можно поднять. `0` оставляет дефолт. --- diff --git a/internal/transport/vp8channel/kcp.go b/internal/transport/vp8channel/kcp.go index 566d931..cbba443 100644 --- a/internal/transport/vp8channel/kcp.go +++ b/internal/transport/vp8channel/kcp.go @@ -164,16 +164,6 @@ func (r *kcpRuntime) send(msg []byte) error { return nil } -// srtt reports KCP's smoothed round-trip estimate in milliseconds, or 0 when -// no sample exists yet. The adaptive pacer reads this to detect the policer -// queueing that precedes a throughput collapse. -func (r *kcpRuntime) srtt() int32 { - if r == nil || r.sess == nil { - return 0 - } - return r.sess.GetSRTT() -} - func (r *kcpRuntime) close() { r.closeOnce.Do(func() { _ = r.sess.Close() diff --git a/internal/transport/vp8channel/options.go b/internal/transport/vp8channel/options.go index 333346c..2044da4 100644 --- a/internal/transport/vp8channel/options.go +++ b/internal/transport/vp8channel/options.go @@ -9,34 +9,14 @@ import ( const ( defaultFPS = 30 defaultBatchSize = 64 - // defaultMaxBytesPerSec is the upper bound the adaptive pacer may probe up - // to when no explicit ceiling is configured. The pacer (see pacer.go) - // auto-discovers each SFU's real policer knee at runtime from KCP's - // smoothed RTT, so this is just a sanity cap, not a hand-tuned operating - // point. 1 MiB/s matches the throughput target from issue #107 without - // letting a misbehaving path probe unbounded. + // defaultMaxBytesPerSec paces the wire byte-rate fed to the video track. + // The earlier ~42s collapse at 1.2 MiB/s was the SFU dropping a track with + // no decodable keyframe, not a raw rate ceiling; that keyframe starvation + // is fixed separately (forceKeepalive, issue #95), so the pacer can run at + // the 1 MB/s target from issue #107 instead of the old 400 KB/s stability + // compromise. The policer knee still differs per SFU, so operators can tune + // this via Options.MaxBytesPerSec (vp8.max_bytes_per_sec in YAML). defaultMaxBytesPerSec = 1_000_000 - - // defaultMinProbeRate floors the adaptive rate so a congested path can - // always make forward progress and recover once delay drops. ~120 KB/s - // stays below the worst observed Telemost knee while keeping the control - // plane and keepalives flowing. - defaultMinProbeRate = 120_000 - - // defaultStartRate is where the pacer begins probing from on a fresh - // session: the old conservative 400 KB/s floor, so behaviour on a healthy - // path only ever ramps up from the previously shipped operating point. - defaultStartRate = 400_000 - - // Adaptive pacer tuning. Delay marks are derived from the measured path - // baseline plus a fixed slack so a low-RTT LAN and a high-RTT WAN both get - // a sane congestion threshold. - rateProbeSlackMs = 8 // extra ms below which we probe the rate up - rateCongestionSlackMs = 25 // extra ms above which we back the rate off - rateProbeStepBytes = 50_000 - rateBackoffNum = 4 // multiplicative decrease: rate *= 4/5 - rateBackoffDen = 5 - baseRTTDriftDiv = 64 // baseline rises by 1/64 of the gap per sample ) // Options tunes the vp8channel transport. Zero values fall back to documented defaults. diff --git a/internal/transport/vp8channel/pacer.go b/internal/transport/vp8channel/pacer.go deleted file mode 100644 index 66c0a22..0000000 --- a/internal/transport/vp8channel/pacer.go +++ /dev/null @@ -1,81 +0,0 @@ -package vp8channel - -// ratePacer is a delay-based AIMD controller that auto-discovers the wire -// byte-rate a policed carrier (the SFU) tolerates, with no operator tuning. -// -// The carrier policer starts queueing before it starts dropping, so KCP's -// smoothed RTT inflates ahead of the throughput collapse seen in issues #95 -// and #107. Each control tick the pacer folds one SRTT sample in: while the -// delay stays near the path baseline it probes the rate up additively, and -// once the delay inflates past the baseline it backs off multiplicatively. -// The rate settles just under the per-SFU policer knee on its own, which is -// the "real run" compromise the fixed 400 KB/s floor only approximated. The -// operator ceiling (vp8.max_bytes_per_sec) just bounds how high it may probe. -type ratePacer struct { - fps int - minRate int - maxRate int - rate int - baseRTT int32 -} - -func newRatePacer(fps, startRate, minRate, maxRate int) *ratePacer { - if fps <= 0 { - fps = defaultFPS - } - if minRate <= 0 { - minRate = defaultMinProbeRate - } - if maxRate < minRate { - maxRate = minRate - } - return &ratePacer{ - fps: fps, - minRate: minRate, - maxRate: maxRate, - rate: max(minRate, min(startRate, maxRate)), - } -} - -// perTick is the current per-frame byte budget. The ticker already paces at -// fps, so a per-tick cap bounds the rate without token bookkeeping. Floor at -// one epoch header so keepalives always fit. -func (rp *ratePacer) perTick() int { - pt := rp.rate / rp.fps - if pt < epochHdrLen { - return epochHdrLen - } - return pt -} - -// observe folds one smoothed-RTT sample (milliseconds, as KCP reports it) into -// the controller and returns the updated per-tick byte budget. A non-positive -// sample means KCP has no RTT estimate yet, so the rate is held steady. -func (rp *ratePacer) observe(srtt int32) int { - if srtt <= 0 { - return rp.perTick() - } - rp.trackBaseline(srtt) - - lowMark := rp.baseRTT + rp.baseRTT/4 + rateProbeSlackMs - highMark := rp.baseRTT + rp.baseRTT/2 + rateCongestionSlackMs - switch { - case srtt >= highMark: - rp.rate = max(rp.minRate, min(rp.rate*rateBackoffNum/rateBackoffDen, rp.maxRate)) - case srtt <= lowMark: - rp.rate = max(rp.minRate, min(rp.rate+rateProbeStepBytes, rp.maxRate)) - } - return rp.perTick() -} - -// trackBaseline anchors baseRTT to the path's uncongested delay: it drops -// instantly to any new minimum and drifts up slowly otherwise, so a genuine -// path change cannot pin the baseline below the real floor forever while a -// transient queueing spike still reads as congestion. -func (rp *ratePacer) trackBaseline(srtt int32) { - if rp.baseRTT == 0 || srtt < rp.baseRTT { - rp.baseRTT = srtt - return - } - rp.baseRTT += (srtt - rp.baseRTT) / baseRTTDriftDiv -} diff --git a/internal/transport/vp8channel/pacer_test.go b/internal/transport/vp8channel/pacer_test.go deleted file mode 100644 index 5424987..0000000 --- a/internal/transport/vp8channel/pacer_test.go +++ /dev/null @@ -1,110 +0,0 @@ -package vp8channel - -import "testing" - -// TestRatePacerProbesUpOnLowDelay verifies the pacer additively raises the -// rate while the path delay stays near the baseline, up to the ceiling. -func TestRatePacerProbesUpOnLowDelay(t *testing.T) { - rp := newRatePacer(30, defaultStartRate, defaultMinProbeRate, defaultMaxBytesPerSec) - - // First sample establishes the baseline (20ms). Subsequent equal samples - // read as uncongested, so the rate climbs by rateProbeStepBytes each tick. - rp.observe(20) - start := rp.rate - for range 5 { - rp.observe(20) - } - if rp.rate <= start { - t.Fatalf("rate did not climb on low delay: start=%d now=%d", start, rp.rate) - } - if rp.rate > rp.maxRate { - t.Fatalf("rate %d exceeded ceiling %d", rp.rate, rp.maxRate) - } -} - -// TestRatePacerBacksOffOnHighDelay verifies a delay spike past the congestion -// mark triggers a multiplicative decrease. -func TestRatePacerBacksOffOnHighDelay(t *testing.T) { - rp := newRatePacer(30, defaultStartRate, defaultMinProbeRate, defaultMaxBytesPerSec) - rp.observe(20) // baseline 20ms - before := rp.rate - - // 20ms baseline -> highMark = 20 + 10 + 25 = 55ms. A 120ms sample is well - // past it and must shrink the rate. - rp.observe(120) - if rp.rate >= before { - t.Fatalf("rate did not back off on high delay: before=%d after=%d", before, rp.rate) - } - want := before * rateBackoffNum / rateBackoffDen - if rp.rate != want { - t.Fatalf("backoff rate = %d, want %d", rp.rate, want) - } -} - -// TestRatePacerConvergesToKnee drives a synthetic policer: delay stays low -// until the rate crosses a hidden knee, then inflates. The pacer must settle -// in a band around that knee instead of running away or collapsing. -func TestRatePacerConvergesToKnee(t *testing.T) { - const knee = 600_000 - rp := newRatePacer(30, defaultStartRate, defaultMinProbeRate, defaultMaxBytesPerSec) - - srtt := func(rate int) int32 { - // Below the knee the path is idle (20ms). Above it the policer queues, - // inflating delay in proportion to the overshoot. - if rate <= knee { - return 20 - } - return 20 + int32(min((rate-knee)/10_000, 1_000)) //nolint:gosec // G115: bounded by min to 1000 - } - - for range 400 { - rp.observe(srtt(rp.rate)) - } - - // After convergence the rate must sit in a sane band around the knee: - // high enough to beat the old 400 KB/s floor, never pinned at the ceiling. - if rp.rate < defaultStartRate { - t.Fatalf("converged rate %d below start floor %d", rp.rate, defaultStartRate) - } - if rp.rate >= rp.maxRate { - t.Fatalf("rate pinned at ceiling %d, did not detect knee", rp.rate) - } - if rp.rate > knee+4*rateProbeStepBytes { - t.Fatalf("converged rate %d overshoots knee %d", rp.rate, knee) - } -} - -// TestRatePacerHoldsWithoutSample verifies a non-positive SRTT (no estimate -// yet) holds the rate steady rather than probing blind. -func TestRatePacerHoldsWithoutSample(t *testing.T) { - rp := newRatePacer(30, defaultStartRate, defaultMinProbeRate, defaultMaxBytesPerSec) - before := rp.rate - if got := rp.observe(0); got != before/rp.fps { - t.Fatalf("perTick on no-sample = %d, want %d", got, before/rp.fps) - } - if rp.rate != before { - t.Fatalf("rate moved without a sample: before=%d after=%d", before, rp.rate) - } -} - -// TestRatePacerNeverBelowFloor verifies sustained congestion cannot starve the -// rate below the minimum that keeps the control plane and keepalives flowing. -func TestRatePacerNeverBelowFloor(t *testing.T) { - rp := newRatePacer(30, defaultStartRate, defaultMinProbeRate, defaultMaxBytesPerSec) - rp.observe(10) - for range 100 { - rp.observe(5000) // permanent heavy congestion - } - if rp.rate < rp.minRate { - t.Fatalf("rate %d fell below floor %d", rp.rate, rp.minRate) - } -} - -// TestRatePacerPerTickFloor verifies the per-tick budget never drops below one -// epoch header so keepalives always fit even at the rate floor. -func TestRatePacerPerTickFloor(t *testing.T) { - rp := newRatePacer(30, epochHdrLen, epochHdrLen, epochHdrLen) - if got := rp.perTick(); got < epochHdrLen { - t.Fatalf("perTick = %d, want >= %d", got, epochHdrLen) - } -} diff --git a/internal/transport/vp8channel/transport.go b/internal/transport/vp8channel/transport.go index e2c7794..170ef57 100644 --- a/internal/transport/vp8channel/transport.go +++ b/internal/transport/vp8channel/transport.go @@ -155,14 +155,7 @@ type streamTransport struct { controlKCPOnce sync.Once frameInterval time.Duration batchSize int - // Adaptive pacer bounds (bytes/sec). Each writer goroutine builds its own - // ratePacer from these and probes the live wire rate from KCP's SRTT, so - // the operating point auto-tracks each SFU's policer knee. startRate is - // the conservative point a fresh session begins probing from; maxRate is - // the operator ceiling (vp8.max_bytes_per_sec) or the built-in cap. - startRate int - minRate int - maxRate int + perTickBytes int // localEpoch is stamped into every outgoing VP8 frame. Explicit // upper-layer resets rotate it so the peer can reset its KCP state too. @@ -291,13 +284,16 @@ func newStreamTransport( if batchSize <= 0 { batchSize = defaultBatchSize } - // maxRate caps how high the adaptive pacer may probe. An explicit - // vp8.max_bytes_per_sec becomes that ceiling; otherwise the built-in cap - // applies. The pacer auto-discovers the real operating point under it from - // KCP's SRTT, so no value here needs hand-tuning to a specific SFU. - maxRate := opts.MaxBytesPerSec - if maxRate <= 0 { - maxRate = defaultMaxBytesPerSec + byteRate := opts.MaxBytesPerSec + if byteRate <= 0 { + byteRate = defaultMaxBytesPerSec + } + // Bytes we may emit per frame tick to hold the wire under byteRate. The + // ticker already paces at fps, so a per-tick cap bounds the rate without + // any token bookkeeping. Floor at one epoch header so keepalives fit. + perTickBytes := byteRate / fps + if perTickBytes < epochHdrLen { + perTickBytes = epochHdrLen } tr := &streamTransport{ @@ -311,9 +307,7 @@ func newStreamTransport( writerDone: make(chan struct{}), frameInterval: time.Second / time.Duration(fps), batchSize: batchSize, - startRate: defaultStartRate, - minRate: defaultMinProbeRate, - maxRate: maxRate, + perTickBytes: perTickBytes, bindingToken: bindingToken(cfg.RoomURL), localEpoch: randomEpoch(), peers: make(map[uint32]*kcpRuntime), @@ -341,24 +335,6 @@ func newStreamTransport( return tr } -// fps recovers the frame rate from the configured frame interval. The pacer -// divides the per-second rate by this to get a per-tick byte budget. -func (p *streamTransport) fps() int { - if p.frameInterval <= 0 { - return defaultFPS - } - return max(int(time.Second/p.frameInterval), 1) -} - -// dataSRTT reports the smoothed RTT (ms) of the single-peer data KCP session, -// or 0 when no session or sample exists yet. -func (p *streamTransport) dataSRTT() int32 { - p.kcpMu.RLock() - rt := p.kcp - p.kcpMu.RUnlock() - return rt.srtt() -} - func (p *streamTransport) Connect(ctx context.Context) error { connectCtx, cancel := context.WithTimeout(ctx, defaultConnectTimeout) defer cancel() @@ -743,33 +719,14 @@ func (p *streamTransport) Features() transport.Features { type writerState struct { p *streamTransport keepaliveEvery int - forceKeepaliveEvery int idleTicks int + forceKeepaliveEvery int ticksSinceKeepalive int - // pacer adapts the per-tick byte budget from the live path delay so the - // wire rate tracks the SFU's policer knee without operator tuning. - pacer *ratePacer - // srttFn reports the smoothed RTT (ms) of the KCP session this writer - // feeds, so the pacer can read congestion from the right plane. - srttFn func() int32 // pendingControl holds a control frame that failed WriteSample and must be // retried on the next tick before consuming more frames. pendingControl []byte } -// perTick folds the current path delay into the pacer and returns the byte -// budget for this tick. A nil pacer (defensive) falls back to the start rate. -func (w *writerState) perTick() int { - if w.pacer == nil { - return epochHdrLen - } - var srtt int32 - if w.srttFn != nil { - srtt = w.srttFn() - } - return w.pacer.observe(srtt) -} - func (w *writerState) writeSample(data []byte) bool { return w.p.writeSampleLocked(data) } @@ -841,7 +798,7 @@ func (w *writerState) drainControl() bool { func (w *writerState) drainData() { select { case frame := <-w.p.outbound: - sample := w.p.batchSample(frame, w.perTick()) + sample := w.p.batchSample(frame, w.p.perTickBytes) w.idleTicks = 0 _ = w.writeSample(sample) default: @@ -864,8 +821,6 @@ func (p *streamTransport) writerLoop() { p: p, keepaliveEvery: max(int(keepaliveIdlePeriod/p.frameInterval), 1), forceKeepaliveEvery: max(int((2*time.Second)/p.frameInterval), 1), - pacer: newRatePacer(p.fps(), p.startRate, p.minRate, p.maxRate), - srttFn: p.dataSRTT, } for { @@ -1393,7 +1348,7 @@ func (p *streamTransport) getOrCreatePeerKCP(epoch uint32) *kcpRuntime { logger.Infof("vp8channel: peer session created epoch=0x%08x", epoch) // Pump outbound frames from this peer's queue into the writer. - go p.peerWriterPump(rt, out) + go p.peerWriterPump(epoch, out) return rt } @@ -1406,15 +1361,10 @@ func (p *streamTransport) getOrCreatePeerKCP(epoch uint32) *kcpRuntime { // overran the SFU's policer, drove burst loss, and collapsed throughput to // zero within ~20-40s (issue #95). Stops when the channel is closed or the // transport shuts down. -func (p *streamTransport) peerWriterPump(rt *kcpRuntime, out chan []byte) { +func (p *streamTransport) peerWriterPump(_ uint32, out chan []byte) { ticker := time.NewTicker(p.frameInterval) defer ticker.Stop() - // Each downlink peer paces independently: its own ratePacer reads that - // peer's KCP SRTT and tracks its path's policer knee without affecting - // the others or the client->server path. - pacer := newRatePacer(p.fps(), p.startRate, p.minRate, p.maxRate) - // Inject a decodable VP8 keyframe on the same cadence writerLoop uses for // the client->server path. The server's per-peer bulk path previously // emitted only opaque KCP data frames, which never form a decodable VP8 @@ -1442,7 +1392,7 @@ func (p *streamTransport) peerWriterPump(rt *kcpRuntime, out chan []byte) { if !ok { return } - sample := p.batchSampleFrom(out, frame, pacer.observe(rt.srtt())) + sample := p.batchSampleFrom(out, frame, p.perTickBytes) _ = p.writeSampleLocked(sample) default: } diff --git a/internal/transport/vp8channel/transport_unit_test.go b/internal/transport/vp8channel/transport_unit_test.go index 8c42d79..6c53ae2 100644 --- a/internal/transport/vp8channel/transport_unit_test.go +++ b/internal/transport/vp8channel/transport_unit_test.go @@ -614,19 +614,20 @@ func TestReorderBufferRestoresOrderAndSurvivesLoss(t *testing.T) { } } -// TestMaxBytesPerSecCeiling verifies vp8.max_bytes_per_sec becomes the upper -// bound the adaptive pacer may probe up to, while zero falls back to the -// built-in cap. The pacer auto-discovers the real operating point under that -// ceiling from KCP's SRTT, so the option is a safety bound, not a hand-tuned -// rate. See issue #107. -func TestMaxBytesPerSecCeiling(t *testing.T) { +// TestMaxBytesPerSecPacing verifies the operator-tunable wire rate cap flows +// into the per-tick byte budget that paces the writer. The default uses the +// built-in 1 MB/s target; an explicit value lets operators dial in their own +// SFU's stable maximum. See issue #107. +func TestMaxBytesPerSecPacing(t *testing.T) { cases := []struct { name string maxBytes int - wantMaxRate int + fps int + wantPerTick int }{ - {name: "default cap", maxBytes: 0, wantMaxRate: defaultMaxBytesPerSec}, - {name: "explicit ceiling", maxBytes: 2_000_000, wantMaxRate: 2_000_000}, + {name: "default floor", maxBytes: 0, fps: 30, wantPerTick: defaultMaxBytesPerSec / 30}, + {name: "explicit raise", maxBytes: 1_000_000, fps: 40, wantPerTick: 1_000_000 / 40}, + {name: "clamped to header", maxBytes: 1, fps: 30, wantPerTick: epochHdrLen}, } for _, c := range cases { t.Run(c.name, func(t *testing.T) { @@ -634,13 +635,10 @@ func TestMaxBytesPerSecCeiling(t *testing.T) { &engineVideoSession{}, nil, transport.Config{}, - Options{FPS: 30, BatchSize: 1, MaxBytesPerSec: c.maxBytes}, + Options{FPS: c.fps, BatchSize: 1, MaxBytesPerSec: c.maxBytes}, ) - if tr.maxRate != c.wantMaxRate { - t.Fatalf("maxRate = %d, want %d", tr.maxRate, c.wantMaxRate) - } - if tr.startRate != defaultStartRate { - t.Fatalf("startRate = %d, want %d", tr.startRate, defaultStartRate) + if tr.perTickBytes != c.wantPerTick { + t.Fatalf("perTickBytes = %d, want %d", tr.perTickBytes, c.wantPerTick) } }) } From 4ddb98136699a9d4be76ffa97325408255f90edc Mon Sep 17 00:00:00 2001 From: neuronori Date: Mon, 29 Jun 2026 06:27:32 +0000 Subject: [PATCH 4/4] fix(vp8channel): remove byte-rate pacer that throttled throughput the pacer (52aea2d) and its per-tick byte budget capped the wire well below the sfu's real ceiling: telemost ran ~10mbit before it landed and 2-3mbit after. it was added to fix the #95 collapse, but that collapse was keyframe starvation (sfu dropped the track with no decodable keyframe after ~40s), fixed separately by forceKeepalive in b20554b. the pacer was treating a symptom that no longer exists. drop perTickBytes and the vp8.max_bytes_per_sec knob, and restore the kcp send window (512->4096) and tick (20ms->5ms) that were shrunk alongside the pacer. control liveness no longer depends on a small data window: ping/pong runs on its own isolated kcp session drained with priority by writerLoop, so the original starvation rationale is gone. batchSample still bounds one sample to defaultMaxPayloadSize. refs #107 #95 --- .../client/client.telemost.vp8channel.yaml | 4 -- .../server/server.telemost.vp8channel.yaml | 4 -- docs/settings.md | 3 -- docs/settings.ru.md | 3 -- internal/app/session/session.go | 3 -- internal/app/session/transport_options.go | 5 +- internal/config/config.go | 7 --- internal/transport/vp8channel/kcp.go | 40 +++++++-------- internal/transport/vp8channel/options.go | 11 ----- internal/transport/vp8channel/transport.go | 49 ++++++------------- .../transport/vp8channel/transport_test.go | 2 +- .../vp8channel/transport_unit_test.go | 30 ------------ 12 files changed, 37 insertions(+), 124 deletions(-) diff --git a/docs/examples/client/client.telemost.vp8channel.yaml b/docs/examples/client/client.telemost.vp8channel.yaml index d69f912..83e4984 100644 --- a/docs/examples/client/client.telemost.vp8channel.yaml +++ b/docs/examples/client/client.telemost.vp8channel.yaml @@ -33,10 +33,6 @@ socks: vp8: fps: 30 batch_size: 64 - # Потолок скорости на проводе (байт/сек). Дефолт 1000000 (~1 МБ/с). - # Policer-колено у каждого SFU своё: если поток падает - снизь до - # стабильной скорости, на здоровом пути можно поднять. 0 = дефолт. - # max_bytes_per_sec: 1000000 data: data debug: false diff --git a/docs/examples/server/server.telemost.vp8channel.yaml b/docs/examples/server/server.telemost.vp8channel.yaml index f951c61..b74c163 100644 --- a/docs/examples/server/server.telemost.vp8channel.yaml +++ b/docs/examples/server/server.telemost.vp8channel.yaml @@ -34,10 +34,6 @@ socks: vp8: fps: 30 batch_size: 64 - # Потолок скорости на проводе (байт/сек). Дефолт 1000000 (~1 МБ/с). - # Policer-колено у каждого SFU своё: если поток падает - снизь до - # стабильной скорости, на здоровом пути можно поднять. 0 = дефолт. - # max_bytes_per_sec: 1000000 data: data debug: false diff --git a/docs/settings.md b/docs/settings.md index b62431e..40e3a10 100644 --- a/docs/settings.md +++ b/docs/settings.md @@ -152,9 +152,6 @@ No extra fields - everything is default. |-----------|----------|:------------:| | `vp8.fps` | VP8 stream FPS | `30` | | `vp8.batch_size` | Frames per tick | `64` | -| `vp8.max_bytes_per_sec` | Wire byte-rate ceiling, bytes/sec | `1000000` | - -`vp8.max_bytes_per_sec` caps the byte-rate the pacer feeds to the video track. The default 1000000 (1 MB/s) matches the issue #107 throughput target. The per-SFU policer knee still varies, so if a service stalls at this rate, lower it until stable; on a healthy path you can raise it. `0` keeps the default. --- diff --git a/docs/settings.ru.md b/docs/settings.ru.md index 295c3bf..5ccf543 100644 --- a/docs/settings.ru.md +++ b/docs/settings.ru.md @@ -153,9 +153,6 @@ transport. Используй одинаковые traffic-настройки н |-----------|----------|:------------:| | `vp8.fps` | FPS VP8 потока | `30` | | `vp8.batch_size` | Кадров за тик | `64` | -| `vp8.max_bytes_per_sec` | Потолок скорости на проводе, байт/сек | `1000000` | - -`vp8.max_bytes_per_sec` ограничивает байтрейт, который пейсер льёт в видеотрек. Дефолт 1000000 (1 МБ/с) - целевая скорость из issue #107. Policer-колено у каждого SFU своё, поэтому если на этой скорости поток падает - снижай до стабильной; на здоровом пути можно поднять. `0` оставляет дефолт. --- diff --git a/internal/app/session/session.go b/internal/app/session/session.go index d07bf30..14b17ce 100644 --- a/internal/app/session/session.go +++ b/internal/app/session/session.go @@ -161,9 +161,6 @@ type VideoConfig struct { type VP8Config struct { FPS int BatchSize int - // MaxBytesPerSec caps the wire byte-rate. Zero falls back to the - // transport default. - MaxBytesPerSec int } // SEIConfig holds tunables for the seichannel transport. diff --git a/internal/app/session/transport_options.go b/internal/app/session/transport_options.go index 5b7616e..5f15484 100644 --- a/internal/app/session/transport_options.go +++ b/internal/app/session/transport_options.go @@ -27,9 +27,8 @@ func buildTransportOptions(cfg Config) transport.Options { } case transportVP8: return vp8channel.Options{ - FPS: cfg.VP8.FPS, - BatchSize: cfg.VP8.BatchSize, - MaxBytesPerSec: cfg.VP8.MaxBytesPerSec, + FPS: cfg.VP8.FPS, + BatchSize: cfg.VP8.BatchSize, } case transportSEI: return seichannel.Options{ diff --git a/internal/config/config.go b/internal/config/config.go index adbb893..dca5863 100644 --- a/internal/config/config.go +++ b/internal/config/config.go @@ -136,11 +136,6 @@ type Video struct { type VP8 struct { FPS int `yaml:"fps"` BatchSize int `yaml:"batch_size"` - // MaxBytesPerSec caps the wire byte-rate fed to the video track. The SFU - // policer knee differs per service, so this lets operators dial in the - // real ceiling instead of the conservative built-in default. Zero keeps - // the transport default. - MaxBytesPerSec int `yaml:"max_bytes_per_sec"` } // SEI tunes the seichannel transport. @@ -282,7 +277,6 @@ func Apply(dst session.Config, f File) session.Config { dst.Video.TileRS = pickInt(dst.Video.TileRS, f.Video.TileRS) dst.VP8.FPS = pickInt(dst.VP8.FPS, f.VP8.FPS) dst.VP8.BatchSize = pickInt(dst.VP8.BatchSize, f.VP8.BatchSize) - dst.VP8.MaxBytesPerSec = pickInt(dst.VP8.MaxBytesPerSec, f.VP8.MaxBytesPerSec) dst.SEI.FPS = pickInt(dst.SEI.FPS, f.SEI.FPS) dst.SEI.BatchSize = pickInt(dst.SEI.BatchSize, f.SEI.BatchSize) dst.SEI.FragmentSize = pickInt(dst.SEI.FragmentSize, f.SEI.FragmentSize) @@ -330,7 +324,6 @@ func ApplyProfile(base session.Config, p Profile) session.Config { dst.Video.TileRS = overlayInt(dst.Video.TileRS, p.Video.TileRS) dst.VP8.FPS = overlayInt(dst.VP8.FPS, p.VP8.FPS) dst.VP8.BatchSize = overlayInt(dst.VP8.BatchSize, p.VP8.BatchSize) - dst.VP8.MaxBytesPerSec = overlayInt(dst.VP8.MaxBytesPerSec, p.VP8.MaxBytesPerSec) dst.SEI.FPS = overlayInt(dst.SEI.FPS, p.SEI.FPS) dst.SEI.BatchSize = overlayInt(dst.SEI.BatchSize, p.SEI.BatchSize) dst.SEI.FragmentSize = overlayInt(dst.SEI.FragmentSize, p.SEI.FragmentSize) diff --git a/internal/transport/vp8channel/kcp.go b/internal/transport/vp8channel/kcp.go index cbba443..09d504c 100644 --- a/internal/transport/vp8channel/kcp.go +++ b/internal/transport/vp8channel/kcp.go @@ -27,15 +27,16 @@ const ( // clamped. Stay below that with headroom for KCP overhead (24 bytes). kcpMTU = 1400 - // Send/receive window in segments, sized to the bandwidth-delay product - // of the policed video path. VP8 over QR encoding has very low throughput - // (~0.3 MB/s observed). We use a moderate send window to balance: - // - Large enough for reasonable throughput (don't underutilize the pipe) - // - Small enough that control-plane pongs can get through within liveness timeout - // With 512 segments * 1400 bytes = ~716KB in-flight, and ~0.3 MB/s throughput, - // data sits in queue for ~2-3 seconds, giving control a chance to pass. - kcpSndWnd = 512 - kcpRcvWnd = 1024 + // Send/receive window in segments. Bulk data runs on its own KCP session, + // isolated from the control plane (ping/pong has a separate startKCP and is + // drained with priority by writerLoop), so a large data window no longer + // starves control liveness the way it did before that split (issue #95). + // One VP8 frame can carry many KCP segments and ACKs only trickle back at + // frame cadence, so a generous window is what keeps the policed path full + // and lets throughput reach the SFU's real ceiling (~10 Mbit on Telemost) + // instead of being clamped to a fraction of it. + kcpSndWnd = 4096 + kcpRcvWnd = 4096 // Length prefix for our message framing on top of KCP stream mode. // We use stream mode because UDPSession.Write fragments messages > MSS @@ -74,19 +75,14 @@ func startKCP(out chan<- []byte, onData func([]byte), epochHdr [epochHdrLen]byte } // nodelay=1, interval=5ms, fast resend=2, congestion control OFF (nc=1). - // KCP does NOT regulate the send rate here - the writerLoop byte pacer - // does, fed at a fixed rate just under the carrier's policer knee. KCP's - // own loss-based congestion control is the wrong controller for a hard - // policer: with nc=0 the unavoidable ~4% drops collapsed cwnd and starved - // the wire to ~45 KiB/s. With nc=1 KCP just keeps the BDP-sized window - // full and retransmits the few losses; the pacer caps the rate so we - // never overdrive the policer into its collapse zone. - // nodelay=1, interval=20ms (slower for QR-encoded VP8), fast resend=2, congestion control OFF (nc=1). - // QR-encoded VP8 has very low throughput (~0.3 MB/s), so we use a larger interval - // to allow batching and reduce overhead. KCP's own loss-based congestion control - // is disabled (nc=1) because the carrier has hard bandwidth limits; the writerLoop - // byte pacer handles rate limiting. - sess.SetNoDelay(1, 20, 2, 1) + // The frame ticker already paces emission at the VP8 frame cadence, so the + // 5ms KCP tick just keeps scheduling latency low; a slower tick only adds + // dead time before retransmits and ACKs. nc=1 disables KCP's loss-based + // congestion control because the carrier is a hard policer, not a fair + // queue: with nc=0 the unavoidable ~4% drops collapsed cwnd and starved + // the wire. With nc=1 KCP keeps the window full and retransmits the few + // losses, letting throughput reach the SFU's real ceiling. + sess.SetNoDelay(1, 5, 2, 1) sess.SetWindowSize(kcpSndWnd, kcpRcvWnd) sess.SetMtu(kcpMTU) // Upstream marked SetStreamMode deprecated without providing a replacement; diff --git a/internal/transport/vp8channel/options.go b/internal/transport/vp8channel/options.go index 2044da4..0abde25 100644 --- a/internal/transport/vp8channel/options.go +++ b/internal/transport/vp8channel/options.go @@ -9,23 +9,12 @@ import ( const ( defaultFPS = 30 defaultBatchSize = 64 - // defaultMaxBytesPerSec paces the wire byte-rate fed to the video track. - // The earlier ~42s collapse at 1.2 MiB/s was the SFU dropping a track with - // no decodable keyframe, not a raw rate ceiling; that keyframe starvation - // is fixed separately (forceKeepalive, issue #95), so the pacer can run at - // the 1 MB/s target from issue #107 instead of the old 400 KB/s stability - // compromise. The policer knee still differs per SFU, so operators can tune - // this via Options.MaxBytesPerSec (vp8.max_bytes_per_sec in YAML). - defaultMaxBytesPerSec = 1_000_000 ) // Options tunes the vp8channel transport. Zero values fall back to documented defaults. type Options struct { FPS int BatchSize int - // MaxBytesPerSec caps the wire byte-rate fed to the video track. Zero - // falls back to defaultMaxBytesPerSec. - MaxBytesPerSec int } // TransportOptions marks Options as belonging to the transport options family. diff --git a/internal/transport/vp8channel/transport.go b/internal/transport/vp8channel/transport.go index 170ef57..60003e5 100644 --- a/internal/transport/vp8channel/transport.go +++ b/internal/transport/vp8channel/transport.go @@ -155,7 +155,6 @@ type streamTransport struct { controlKCPOnce sync.Once frameInterval time.Duration batchSize int - perTickBytes int // localEpoch is stamped into every outgoing VP8 frame. Explicit // upper-layer resets rotate it so the peer can reset its KCP state too. @@ -284,18 +283,6 @@ func newStreamTransport( if batchSize <= 0 { batchSize = defaultBatchSize } - byteRate := opts.MaxBytesPerSec - if byteRate <= 0 { - byteRate = defaultMaxBytesPerSec - } - // Bytes we may emit per frame tick to hold the wire under byteRate. The - // ticker already paces at fps, so a per-tick cap bounds the rate without - // any token bookkeeping. Floor at one epoch header so keepalives fit. - perTickBytes := byteRate / fps - if perTickBytes < epochHdrLen { - perTickBytes = epochHdrLen - } - tr := &streamTransport{ stream: stream, track: track, @@ -307,7 +294,6 @@ func newStreamTransport( writerDone: make(chan struct{}), frameInterval: time.Second / time.Duration(fps), batchSize: batchSize, - perTickBytes: perTickBytes, bindingToken: bindingToken(cfg.RoomURL), localEpoch: randomEpoch(), peers: make(map[uint32]*kcpRuntime), @@ -798,7 +784,7 @@ func (w *writerState) drainControl() bool { func (w *writerState) drainData() { select { case frame := <-w.p.outbound: - sample := w.p.batchSample(frame, w.p.perTickBytes) + sample := w.p.batchSample(frame) w.idleTicks = 0 _ = w.writeSample(sample) default: @@ -840,18 +826,16 @@ func (p *streamTransport) writerLoop() { } } -func (p *streamTransport) batchSample(first []byte, maxBytes int) []byte { - return p.batchSampleFrom(p.outbound, first, maxBytes) +func (p *streamTransport) batchSample(first []byte) []byte { + return p.batchSampleFrom(p.outbound, first) } // batchSampleFrom coalesces up to batchSize KCP frames drained from src into a -// single VP8 sample, bounded by maxBytes. The shared writerLoop drains the -// single-peer outbound queue; per-peer pumps drain their own queue through the -// same batching so the server->client path is paced identically to the client. -func (p *streamTransport) batchSampleFrom(src <-chan []byte, first []byte, maxBytes int) []byte { - if maxBytes <= 0 || maxBytes > defaultMaxPayloadSize { - maxBytes = defaultMaxPayloadSize - } +// single VP8 sample, bounded by defaultMaxPayloadSize. The shared writerLoop +// drains the single-peer outbound queue; per-peer pumps drain their own queue +// through the same batching so the server->client path is built identically to +// the client. +func (p *streamTransport) batchSampleFrom(src <-chan []byte, first []byte) []byte { if len(first) <= epochHdrLen || p.batchSize <= 1 { return first } @@ -868,7 +852,7 @@ func (p *streamTransport) batchSampleFrom(src <-chan []byte, first []byte, maxBy continue } payload := frame[epochHdrLen:] - if len(sample)+2+len(payload) > maxBytes { + if len(sample)+2+len(payload) > defaultMaxPayloadSize { return sample } sample = appendBatchPacket(sample, payload) @@ -1354,13 +1338,12 @@ func (p *streamTransport) getOrCreatePeerKCP(epoch uint32) *kcpRuntime { } // peerWriterPump drains a peer's outbound KCP queue and writes frames to the -// shared video track. It paces the server->client path on the same frame -// ticker and per-tick byte budget as writerLoop drives the client->server -// path. Without this, the pump emitted every KCP frame the instant it was -// queued: with KCP congestion control off (nc=1) and a BDP-sized window, that -// overran the SFU's policer, drove burst loss, and collapsed throughput to -// zero within ~20-40s (issue #95). Stops when the channel is closed or the -// transport shuts down. +// shared video track on the same frame ticker writerLoop uses for the +// client->server path, batching queued frames into one VP8 sample per tick. +// Draining on the ticker (rather than emitting each frame the instant it is +// queued) keeps the per-peer writes interleaved with the keyframe injection +// below and lets batchSampleFrom coalesce segments into full samples. Stops +// when the channel is closed or the transport shuts down. func (p *streamTransport) peerWriterPump(_ uint32, out chan []byte) { ticker := time.NewTicker(p.frameInterval) defer ticker.Stop() @@ -1392,7 +1375,7 @@ func (p *streamTransport) peerWriterPump(_ uint32, out chan []byte) { if !ok { return } - sample := p.batchSampleFrom(out, frame, p.perTickBytes) + sample := p.batchSampleFrom(out, frame) _ = p.writeSampleLocked(sample) default: } diff --git a/internal/transport/vp8channel/transport_test.go b/internal/transport/vp8channel/transport_test.go index 075db75..a1735cd 100644 --- a/internal/transport/vp8channel/transport_test.go +++ b/internal/transport/vp8channel/transport_test.go @@ -128,7 +128,7 @@ func TestBatchSampleCarriesMultipleKCPPackets(t *testing.T) { tr.outbound <- packet("three") tr.outbound <- packet("four") - sample := tr.batchSample(packet("one"), defaultMaxPayloadSize) + sample := tr.batchSample(packet("one")) if !bytes.Equal(sample[:epochHdrLen], hdr[:]) { t.Fatalf("sample epoch header = %x, want %x", sample[:epochHdrLen], hdr[:]) } diff --git a/internal/transport/vp8channel/transport_unit_test.go b/internal/transport/vp8channel/transport_unit_test.go index 6c53ae2..740fb26 100644 --- a/internal/transport/vp8channel/transport_unit_test.go +++ b/internal/transport/vp8channel/transport_unit_test.go @@ -614,36 +614,6 @@ func TestReorderBufferRestoresOrderAndSurvivesLoss(t *testing.T) { } } -// TestMaxBytesPerSecPacing verifies the operator-tunable wire rate cap flows -// into the per-tick byte budget that paces the writer. The default uses the -// built-in 1 MB/s target; an explicit value lets operators dial in their own -// SFU's stable maximum. See issue #107. -func TestMaxBytesPerSecPacing(t *testing.T) { - cases := []struct { - name string - maxBytes int - fps int - wantPerTick int - }{ - {name: "default floor", maxBytes: 0, fps: 30, wantPerTick: defaultMaxBytesPerSec / 30}, - {name: "explicit raise", maxBytes: 1_000_000, fps: 40, wantPerTick: 1_000_000 / 40}, - {name: "clamped to header", maxBytes: 1, fps: 30, wantPerTick: epochHdrLen}, - } - for _, c := range cases { - t.Run(c.name, func(t *testing.T) { - tr := newStreamTransport( - &engineVideoSession{}, - nil, - transport.Config{}, - Options{FPS: c.fps, BatchSize: 1, MaxBytesPerSec: c.maxBytes}, - ) - if tr.perTickBytes != c.wantPerTick { - t.Fatalf("perTickBytes = %d, want %d", tr.perTickBytes, c.wantPerTick) - } - }) - } -} - func TestSeqLessWrapAround(t *testing.T) { cases := []struct { a, b uint16