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 4d48681..0abde25 100644 --- a/internal/transport/vp8channel/options.go +++ b/internal/transport/vp8channel/options.go @@ -9,20 +9,12 @@ 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. - defaultMaxBytesPerSec = 400_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[:]) }