From 708b8ea6baa7a41e786a9c74a28a1203cef63b4e Mon Sep 17 00:00:00 2001 From: David Colburn Date: Wed, 14 Oct 2026 15:57:01 -0500 Subject: [PATCH] sfu: use mediatransportutil/pkg/red in RedPrimaryReceiver (#5014) RED depacketizing now lives in mediatransportutil so the SFU and replay export share one implementation. Drop the local copies; the encoder round-trip tests now exercise the shared package. --- go.mod | 2 +- go.sum | 4 +- pkg/sfu/redprimaryreceiver.go | 191 +--------------------------------- pkg/sfu/redreceiver_test.go | 15 +-- 4 files changed, 15 insertions(+), 197 deletions(-) diff --git a/go.mod b/go.mod index 806263a4a..5e5d3d78c 100644 --- a/go.mod +++ b/go.mod @@ -20,7 +20,7 @@ require ( github.com/jellydator/ttlcache/v3 v3.4.1 github.com/jxskiss/base62 v1.1.0 github.com/livekit/mageutil v0.0.0-20250511045019-0f1ff63f7731 - github.com/livekit/mediatransportutil v0.0.0-20260821083140-f234b534b095 + github.com/livekit/mediatransportutil v0.0.0-20261014192331-dd8e31a0adc0 github.com/livekit/protocol v1.52.2-0.20261012065025-9bb708cdb819 github.com/livekit/psrpc v0.8.2 github.com/mackerelio/go-osstat v0.2.8 diff --git a/go.sum b/go.sum index f7599ec1d..33f0341ad 100644 --- a/go.sum +++ b/go.sum @@ -164,8 +164,8 @@ github.com/livekit/ice/v4 v4.4.0-warp.3 h1:iSAtEiLeCi/ZKEKOsRrDlBg/BRcyUKpV/NuAX github.com/livekit/ice/v4 v4.4.0-warp.3/go.mod h1:obAyD+J+Hzs7QA7Y8YXHp5uIn6gb7z87pKedXZkrcFU= github.com/livekit/mageutil v0.0.0-20250511045019-0f1ff63f7731 h1:9x+U2HGLrSw5ATTo469PQPkqzdoU7be46ryiCDO3boc= github.com/livekit/mageutil v0.0.0-20250511045019-0f1ff63f7731/go.mod h1:Rs3MhFwutWhGwmY1VQsygw28z5bWcnEYmS1OG9OxjOQ= -github.com/livekit/mediatransportutil v0.0.0-20260821083140-f234b534b095 h1:BcliKAXoMhl/nWmzQweQ5kmh4Qqagxl4s3Z5pvM/7AY= -github.com/livekit/mediatransportutil v0.0.0-20260821083140-f234b534b095/go.mod h1:o8CFmAdrVwzJNOCsQCLUzXRjokkufNshnQHOe4fRaqU= +github.com/livekit/mediatransportutil v0.0.0-20261014192331-dd8e31a0adc0 h1:BSLyHr3Lsl71bHGHsfiYNmbohugR2yoaf9HjAZeFJnc= +github.com/livekit/mediatransportutil v0.0.0-20261014192331-dd8e31a0adc0/go.mod h1:o8CFmAdrVwzJNOCsQCLUzXRjokkufNshnQHOe4fRaqU= github.com/livekit/protocol v1.52.2-0.20261012065025-9bb708cdb819 h1:oZ/dI5RG22t0GalRDU1BJWCMtNAEQVPU+odrRgycKDE= github.com/livekit/protocol v1.52.2-0.20261012065025-9bb708cdb819/go.mod h1:k7v95/DNhMkslMeBxdQ+lQ6gKc8hMNGYB7t0+6sEGU4= github.com/livekit/psrpc v0.8.2 h1:7tbkhBJoYlK5Csfkdv1XYbZjyAQeaEZgupvIXNCv7Dc= diff --git a/pkg/sfu/redprimaryreceiver.go b/pkg/sfu/redprimaryreceiver.go index fdef27c81..fce6ece40 100644 --- a/pkg/sfu/redprimaryreceiver.go +++ b/pkg/sfu/redprimaryreceiver.go @@ -15,9 +15,6 @@ package sfu import ( - "encoding/binary" - "errors" - "go.uber.org/atomic" "github.com/pion/rtp" @@ -25,17 +22,13 @@ import ( "github.com/livekit/livekit-server/pkg/sfu/buffer" "github.com/livekit/livekit-server/pkg/sfu/utils" + "github.com/livekit/mediatransportutil/pkg/red" "github.com/livekit/protocol/livekit" "github.com/livekit/protocol/logger" ) var _ REDTransformer = (*RedPrimaryReceiver)(nil) -var ( - ErrIncompleteRedHeader = errors.New("incomplete red block header") - ErrIncompleteRedBlock = errors.New("incomplete red block payload") -) - type RedPrimaryReceiver struct { TrackReceiver downTrackSpreader *utils.DownTrackSpreader[TrackSender] @@ -43,14 +36,10 @@ type RedPrimaryReceiver struct { closed atomic.Bool redPT uint8 - firstPktReceived bool - lastSeq uint16 + depacketizer red.Depacketizer highestForwardedExtSequenceNumber uint64 highestForwardedExtTimestamp uint64 - // bitset for upstream packet receive history [lastSeq-8, lastSeq-1], bit 1 represents packet received - pktHistory byte - // forwarded packet, reused since ForwardRTP runs on one goroutine and // down tracks do not keep the packet past WriteRTP sendExtPkt buffer.ExtPacket @@ -80,7 +69,7 @@ func (r *RedPrimaryReceiver) ForwardRTP(pkt *buffer.ExtPacket, spatialLayer int3 return utils.BroadcastRTP(r.downTrackSpreader, pkt, spatialLayer) } - pkts, err := r.getSendPktsFromRed(pkt.Packet) + pkts, err := r.depacketizer.Depacketize(pkt.Packet) if err != nil { r.logger.Errorw("get encoding for red failed", err, "payloadtype", pkt.Packet.PayloadType) return 0 @@ -115,10 +104,8 @@ func (r *RedPrimaryReceiver) ForwardRTP(pkt *buffer.ExtPacket, spatialLayer int3 "packetESN", pPkt.ExtSequenceNumber, "packetExtractedTS", pkts[i].Timestamp, "packetETS", pPkt.ExtTimestamp, - "pktHistory", r.pktHistory, "redHeader", pkt.Packet.Payload[:min(len(pkt.Packet.Payload), 10)], "payloadSize", len(pkt.Packet.Payload), - "lastSeq", r.lastSeq, "highestFowardedExtSequenceNumber", r.highestForwardedExtSequenceNumber, "highestFowardedExtTimestamp", r.highestForwardedExtTimestamp, ) @@ -207,7 +194,7 @@ func (r *RedPrimaryReceiver) ReadRTP(buf []byte, layer uint8, esn uint64) (int, var pkt rtp.Packet pkt.Unmarshal(buf[:n]) - payload, err := extractPrimaryEncodingForRED(pkt.Payload) + payload, err := red.ExtractPrimaryEncoding(pkt.Payload) if err != nil { return 0, err } @@ -215,173 +202,3 @@ func (r *RedPrimaryReceiver) ReadRTP(buf []byte, layer uint8, esn uint64) (int, return pkt.MarshalTo(buf) } - -func (r *RedPrimaryReceiver) getSendPktsFromRed(rtp *rtp.Packet) ([]*rtp.Packet, error) { - var needRecover bool - if !r.firstPktReceived { - r.lastSeq = rtp.SequenceNumber - r.pktHistory = 0 - r.firstPktReceived = true - } else { - diff := rtp.SequenceNumber - r.lastSeq - switch { - case diff == 0: // duplicate - break - - case diff > 0x8000: // unorder - // in history - if 65535-diff < 8 { - r.pktHistory |= 1 << (65535 - diff) - needRecover = true - } - - case diff > 8: // long jump - r.lastSeq = rtp.SequenceNumber - r.pktHistory = 0 - needRecover = true - - default: - r.lastSeq = rtp.SequenceNumber - r.pktHistory = (r.pktHistory << byte(diff)) | 1<<(diff-1) - needRecover = true - } - } - - var recoverBits byte - if needRecover { - bitIndex := r.lastSeq - rtp.SequenceNumber - for i := range maxRedCount { - if bitIndex > 7 { - break - } - if r.pktHistory&byte(1<>= 10 - tsOffset := blockHead & 0x3FFF - blockHead >>= 14 - pt := uint8(blockHead & 0x7F) - - blocks = append(blocks, block{pt: pt, length: length, tsOffset: tsOffset}) - - blockLength += length - payload = payload[4:] - } - } - - if len(payload) < blockLength { - return nil, ErrIncompleteRedBlock - } - - pkts := make([]*rtp.Packet, 0, len(blocks)) - for i, b := range blocks { - if b.primary { - header := redPkt.Header - header.PayloadType = b.pt - pkts = append(pkts, &rtp.Packet{Header: header, Payload: payload}) - break - } - - recoverIndex := len(blocks) - i - 1 - if recoverIndex < 1 || recoverBits&(1<<(recoverIndex-1)) == 0 { - // skip past packet/block that does not need recovery - payload = payload[b.length:] - continue - } - - // recover missing packet - header := redPkt.Header - header.SequenceNumber -= uint16(recoverIndex) - header.Timestamp -= b.tsOffset - header.PayloadType = b.pt - pkts = append(pkts, &rtp.Packet{Header: header, Payload: payload[:b.length]}) - - payload = payload[b.length:] - } - - return pkts, nil -} - -func extractPrimaryEncodingForRED(payload []byte) ([]byte, error) { - - /* RED payload https://datatracker.ietf.org/doc/html/rfc2198#section-3 - 0 1 2 3 - 0 1 2 3 4 5 6 7 8 9 0 1 2 3 4 5 6 7 8 9 0 1 2 3 4 5 6 7 8 9 0 1 - +-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+ - |F| block PT | timestamp offset | block length | - +-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+ - F: 1 bit First bit in header indicates whether another header block - follows. If 1 further header blocks follow, if 0 this is the - last header block. - */ - - var blockLength int - for { - if len(payload) < 1 { - // illegal data, need at least one byte for primary encoding - return nil, ErrIncompleteRedHeader - } - - if payload[0]&0x80 == 0 { - // last block is primary encoding data - payload = payload[1:] - break - } else { - if len(payload) < 4 { - // illegal data - return nil, ErrIncompleteRedHeader - } - - blockLength += int(binary.BigEndian.Uint16(payload[2:]) & 0x03FF) - payload = payload[4:] - } - } - - if len(payload) < blockLength { - return nil, ErrIncompleteRedBlock - } - - return payload[blockLength:], nil -} diff --git a/pkg/sfu/redreceiver_test.go b/pkg/sfu/redreceiver_test.go index f98824d0a..8082b0b17 100644 --- a/pkg/sfu/redreceiver_test.go +++ b/pkg/sfu/redreceiver_test.go @@ -24,6 +24,7 @@ import ( "github.com/livekit/livekit-server/pkg/sfu/buffer" "github.com/livekit/mediatransportutil/pkg/bucket" + "github.com/livekit/mediatransportutil/pkg/red" "github.com/livekit/protocol/logger" ) @@ -253,9 +254,9 @@ func TestRedReceiverReusedReadBuffer(t *testing.T) { }, }, } - red := w.GetRedReceiver().(*RedReceiver) + rr := w.GetRedReceiver().(*RedReceiver) dt := &dummyDowntrack{TrackSender: &DownTrack{}} - require.NoError(t, red.AddDownTrack(dt)) + require.NoError(t, rr.AddDownTrack(dt)) var payloads [][]byte pktBuf := make([]byte, bucket.RTPMaxPktSize) @@ -272,12 +273,12 @@ func TestRedReceiverReusedReadBuffer(t *testing.T) { require.NoError(t, err) extPkt, err := buff.ReadExtended(pktBuf) require.NoError(t, err) - red.ForwardRTP(extPkt, 0) + rr.ForwardRTP(extPkt, 0) buffer.ReleaseExtPacket(extPkt) } // the last RED packet carries the two packets before it as redundant blocks - blocks, err := extractPktsFromRed(dt.lastReceivedPkt, 0xFF) + blocks, err := red.ExtractPackets(dt.lastReceivedPkt, 0xFF) require.NoError(t, err) require.Len(t, blocks, 3) for i, block := range blocks { @@ -285,14 +286,14 @@ func TestRedReceiverReusedReadBuffer(t *testing.T) { } } -func verifyRedEncodings(t *testing.T, red *rtp.Packet, redPkts []*rtp.Packet) { +func verifyRedEncodings(t *testing.T, redPkt *rtp.Packet, redPkts []*rtp.Packet) { solidPkts := make([]*rtp.Packet, 0, len(redPkts)) for _, pkt := range redPkts { if pkt != nil { solidPkts = append(solidPkts, pkt) } } - pktsFromRed, err := extractPktsFromRed(red, 0xFF) + pktsFromRed, err := red.ExtractPackets(redPkt, 0xFF) require.NoError(t, err) require.Len(t, pktsFromRed, len(solidPkts)) for i, pkt := range pktsFromRed { @@ -493,7 +494,7 @@ func TestExtractPrimaryEncodingForRED(t *testing.T) { primaryPkts := make([]*rtp.Packet, 0, len(redPkts)) for _, redPkt := range redPkts { - payload, err := extractPrimaryEncodingForRED(redPkt.Payload) + payload, err := red.ExtractPrimaryEncoding(redPkt.Payload) require.NoError(t, err) primaryHeader := redPkt.Header primaryHeader.PayloadType = 111