Merge pull request #108 from neuronori/fix/issue-107-vp8-tunable-rate

fix(vp8channel): remove byte-rate pacer that throttled throughput

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;

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.

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:
 			}

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[:])
 	}