diff --git a/pkg/sfu/redreceiver.go b/pkg/sfu/redreceiver.go index ad0c54b8a..1788c9ef3 100644 --- a/pkg/sfu/redreceiver.go +++ b/pkg/sfu/redreceiver.go @@ -17,6 +17,7 @@ package sfu import ( "encoding/binary" "fmt" + "slices" "go.uber.org/atomic" @@ -50,7 +51,11 @@ type RedReceiver struct { logger logger.Logger closed atomic.Bool pktBuff [maxRedCount]*rtp.Packet - redPayloadBuf [mtuSize]byte + // copies that pktBuff points at, as a forwarded packet's payload is in a read buffer reused for the next packet, + // one more slot than pktBuff so that a new copy never overwrites a redundant block of the packet being encoded + histPkts [maxRedCount + 1]rtp.Packet + histPayloads [maxRedCount + 1][maxRedPayload]byte + redPayloadBuf [mtuSize]byte // forwarded packet, reused like redPayloadBuf since ForwardRTP runs on one goroutine // and down tracks do not keep the packet past WriteRTP redExtPkt buffer.ExtPacket @@ -189,17 +194,22 @@ func (r *RedReceiver) encodeRedForPrimary(pkt *rtp.Packet, redPayload []byte) (i redPkts = append(redPkts, prev) } - // insert primary packet in history buffer - // NOTE: packet is copied from retransmission buffer and used in forwarding path. So, not making another - // copy here and just maintaining pointer to the packet as the forwarding path should not alter the packet. + // insert a copy of the primary packet in history buffer for i := redLength - 1; i >= 0; i-- { if r.pktBuff[i] == nil || // history is empty pkt.SequenceNumber-r.pktBuff[i].SequenceNumber < (1<<15) { // received packet has more recent sequence number + // pick the slot before aging out, as the aged out packet can be a redundant block of this packet + slot := r.freeHistSlot() // age out older ones for j := 0; j < i; j++ { r.pktBuff[j] = r.pktBuff[j+1] } - r.pktBuff[i] = pkt + n := copy(r.histPayloads[slot][:], pkt.Payload) + r.histPkts[slot] = rtp.Packet{ + Header: rtp.Header{SequenceNumber: pkt.SequenceNumber, Timestamp: pkt.Timestamp}, + Payload: r.histPayloads[slot][:n], + } + r.pktBuff[i] = &r.histPkts[slot] break } } @@ -207,6 +217,16 @@ func (r *RedReceiver) encodeRedForPrimary(pkt *rtp.Packet, redPayload []byte) (i return encodeRedForPrimary(redPkts, pkt, redPayload) } +// freeHistSlot returns a history slot that pktBuff does not point at, there is always one as there is a spare slot +func (r *RedReceiver) freeHistSlot() int { + for slot := range r.histPkts { + if !slices.Contains(r.pktBuff[:], &r.histPkts[slot]) { + return slot + } + } + return 0 +} + func encodeRedForPrimary(redPkts []*rtp.Packet, primary *rtp.Packet, redPayload []byte) (int, error) { payloadSize := len(primary.Payload) + 1 for _, p := range redPkts { diff --git a/pkg/sfu/redreceiver_test.go b/pkg/sfu/redreceiver_test.go index 06df4974c..f98824d0a 100644 --- a/pkg/sfu/redreceiver_test.go +++ b/pkg/sfu/redreceiver_test.go @@ -15,6 +15,7 @@ package sfu import ( + "bytes" "testing" "github.com/pion/rtp" @@ -22,6 +23,7 @@ import ( "github.com/stretchr/testify/require" "github.com/livekit/livekit-server/pkg/sfu/buffer" + "github.com/livekit/mediatransportutil/pkg/bucket" "github.com/livekit/protocol/logger" ) @@ -234,6 +236,55 @@ func TestRedReceiver(t *testing.T) { }) } +// reads every packet into one reused buffer, the way ReceiverBase.forwardRTP does +func TestRedReceiverReusedReadBuffer(t *testing.T) { + opus := webrtc.RTPCodecParameters{ + RTPCodecCapability: webrtc.RTPCodecCapability{MimeType: webrtc.MimeTypeOpus, ClockRate: 48000, Channels: 2}, + PayloadType: 111, + } + buff := buffer.NewBuffer(0x5678, 500, 200) + require.NoError(t, buff.Bind(webrtc.RTPParameters{Codecs: []webrtc.RTPCodecParameters{opus}}, opus.RTPCodecCapability, 0)) + + w := &WebRTCReceiver{ + ReceiverBase: &ReceiverBase{ + params: ReceiverBaseParams{ + Kind: webrtc.RTPCodecTypeAudio, + Logger: logger.GetLogger(), + }, + }, + } + red := w.GetRedReceiver().(*RedReceiver) + dt := &dummyDowntrack{TrackSender: &DownTrack{}} + require.NoError(t, red.AddDownTrack(dt)) + + var payloads [][]byte + pktBuf := make([]byte, bucket.RTPMaxPktSize) + for i := range 6 { + payload := bytes.Repeat([]byte{byte(0x10 * (i + 1))}, 60) + payloads = append(payloads, payload) + pkt := rtp.Packet{ + Header: rtp.Header{Version: 2, PayloadType: 111, SequenceNumber: uint16(100 + i), Timestamp: uint32(960 * i), SSRC: 0x5678}, + Payload: payload, + } + raw, err := pkt.Marshal() + require.NoError(t, err) + _, err = buff.Write(raw) + require.NoError(t, err) + extPkt, err := buff.ReadExtended(pktBuf) + require.NoError(t, err) + red.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) + require.NoError(t, err) + require.Len(t, blocks, 3) + for i, block := range blocks { + require.Equal(t, payloads[3+i], block.Payload) + } +} + func verifyRedEncodings(t *testing.T, red *rtp.Packet, redPkts []*rtp.Packet) { solidPkts := make([]*rtp.Packet, 0, len(redPkts)) for _, pkt := range redPkts {