diff --git a/internal/transport/vp8channel/concurrency_test.go b/internal/transport/vp8channel/concurrency_test.go new file mode 100644 index 0000000..454b9da --- /dev/null +++ b/internal/transport/vp8channel/concurrency_test.go @@ -0,0 +1,65 @@ +package vp8channel + +import ( + "sync" + "testing" + "time" +) + +// TestWriteSampleLockedSerializesConcurrentWriters is the regression guard for +// issue #95. The server writes bulk data through per-peer pumps while +// writerLoop writes control/keepalive frames, both calling writeSampleLocked +// concurrently. pion's TrackLocalStaticSample.WriteSample is not safe for +// concurrent use: a frame's packetize step and its RTP emit are not atomic, so +// unsynchronized callers interleave packets on the wire and the remote VP8 +// reassembler drops the corrupted frames. This test installs a sampleWriter +// that fails if two calls are ever in flight at once, then hammers it from many +// goroutines. +func TestWriteSampleLockedSerializesConcurrentWriters(t *testing.T) { + var ( + inFlight int32 + mu sync.Mutex + overlaps int + ) + + p := &streamTransport{frameInterval: time.Second / 30} + p.sampleWriter = func([]byte) bool { + mu.Lock() + inFlight++ + if inFlight > 1 { + overlaps++ + } + mu.Unlock() + + // Hold the "wire" briefly so any missing serialization manifests as + // an observed overlap rather than a lucky interleave. + time.Sleep(50 * time.Microsecond) + + mu.Lock() + inFlight-- + mu.Unlock() + return true + } + + const ( + writers = 16 + perWriter = 64 + ) + var wg sync.WaitGroup + wg.Add(writers) + for range writers { + go func() { + defer wg.Done() + payload := []byte{1, 2, 3, 4} + for range perWriter { + p.writeSampleLocked(payload) + } + }() + } + wg.Wait() + + if overlaps != 0 { + t.Fatalf("writeSampleLocked allowed %d concurrent track writes; "+ + "WriteSample must be fully serialized (issue #95 regression)", overlaps) + } +} diff --git a/internal/transport/vp8channel/transport.go b/internal/transport/vp8channel/transport.go index 6adf8fd..7b820e9 100644 --- a/internal/transport/vp8channel/transport.go +++ b/internal/transport/vp8channel/transport.go @@ -111,6 +111,16 @@ type videoSession interface { type streamTransport struct { stream videoSession track *webrtc.TrackLocalStaticSample + // writeMu serializes all track.WriteSample calls. pion's WriteSample is + // not safe for concurrent use (see writeSampleLocked); the server writes + // bulk data from per-peer pumps while writerLoop writes control frames + // and keepalives, so both paths must funnel through this lock. + writeMu sync.Mutex + // sampleWriter, when set, replaces the real track.WriteSample call. + // Tests inject a writer here to observe the exact byte stream that + // reaches the track and to assert that writeSampleLocked serializes + // concurrent callers. Always invoked under writeMu. + sampleWriter func([]byte) bool onData func([]byte) onPeerData func(peerID string, data []byte) // onControlData is called with every reassembled message from the @@ -647,9 +657,30 @@ type writerState struct { } func (w *writerState) writeSample(data []byte) bool { - return w.p.track.WriteSample(media.Sample{ + return w.p.writeSampleLocked(data) +} + +// writeSampleLocked serializes every WriteSample call on the shared video +// track behind a single mutex. pion's TrackLocalStaticSample.WriteSample is +// NOT safe for concurrent use: it packetizes under its own lock but then +// releases that lock before pushing the resulting RTP packets onto the wire. +// Two concurrent callers therefore each reserve a contiguous block of RTP +// sequence numbers and then race to emit their packets, interleaving them on +// the wire. The receiver's VP8 reassembler enforces strict sequence +// contiguity, so any interleaved frame is discarded - which is exactly the +// server->client bulk-data stall in issue #95 (the server runs a per-peer +// peerWriterPump for data plus writerLoop for control/keepalive, both hitting +// this track at once). Funneling all writes through this mutex makes each +// sample's packetize+send atomic and keeps sequence numbers monotonic. +func (p *streamTransport) writeSampleLocked(data []byte) bool { + p.writeMu.Lock() + defer p.writeMu.Unlock() + if p.sampleWriter != nil { + return p.sampleWriter(data) + } + return p.track.WriteSample(media.Sample{ Data: data, - Duration: w.p.frameInterval, + Duration: p.frameInterval, }) == nil } @@ -1066,10 +1097,7 @@ func (p *streamTransport) peerWriterPump(_ uint32, out chan []byte) { if !ok { return } - _ = p.track.WriteSample(media.Sample{ - Data: frame, - Duration: p.frameInterval, - }) + _ = p.writeSampleLocked(frame) } } }