diff --git a/pkg/config/config.go b/pkg/config/config.go index 54401ce7b..afc27c2a6 100644 --- a/pkg/config/config.go +++ b/pkg/config/config.go @@ -333,11 +333,12 @@ var DefaultConfig = Config{ AutoCreate: true, EnabledCodecs: []CodecSpec{ {Mime: webrtc.MimeTypeOpus}, - {Mime: "audio/red"}, + {Mime: sfu.MimeTypeAudioRed}, {Mime: webrtc.MimeTypeVP8}, {Mime: webrtc.MimeTypeH264}, {Mime: webrtc.MimeTypeVP9}, {Mime: webrtc.MimeTypeAV1}, + {Mime: webrtc.MimeTypeRTX}, }, EmptyTimeout: 5 * 60, DepartureTimeout: 20, diff --git a/pkg/rtc/mediaengine.go b/pkg/rtc/mediaengine.go index 62c3007c9..88261c5fa 100644 --- a/pkg/rtc/mediaengine.go +++ b/pkg/rtc/mediaengine.go @@ -24,10 +24,6 @@ import ( "github.com/livekit/protocol/livekit" ) -const ( - videoRTXMimeType = "video/rtx" -) - var OpusCodecCapability = webrtc.RTPCodecCapability{ MimeType: webrtc.MimeTypeOpus, ClockRate: 48000, @@ -41,7 +37,7 @@ var RedCodecCapability = webrtc.RTPCodecCapability{ SDPFmtpLine: "111/111", } var videoRTX = webrtc.RTPCodecCapability{ - MimeType: videoRTXMimeType, + MimeType: webrtc.MimeTypeRTX, ClockRate: 90000, } @@ -137,7 +133,7 @@ func registerCodecs(me *webrtc.MediaEngine, codecs []*livekit.Codec, rtcpFeedbac if filterOutH264HighProfile && codec.RTPCodecCapability.SDPFmtpLine == h264HighProfileFmtp { continue } - if codec.MimeType == videoRTXMimeType { + if strings.EqualFold(codec.MimeType, webrtc.MimeTypeRTX) { continue } if IsCodecEnabled(codecs, codec.RTPCodecCapability) { @@ -147,7 +143,7 @@ func registerCodecs(me *webrtc.MediaEngine, codecs []*livekit.Codec, rtcpFeedbac if rtxEnabled { if err := me.RegisterCodec(webrtc.RTPCodecParameters{ RTPCodecCapability: webrtc.RTPCodecCapability{ - MimeType: videoRTXMimeType, + MimeType: webrtc.MimeTypeRTX, ClockRate: 90000, SDPFmtpLine: fmt.Sprintf("apt=%d", codec.PayloadType), }, diff --git a/pkg/rtc/participant.go b/pkg/rtc/participant.go index 661ee1aef..7bf90347c 100644 --- a/pkg/rtc/participant.go +++ b/pkg/rtc/participant.go @@ -3017,7 +3017,7 @@ func (p *ParticipantImpl) setupEnabledCodecs(publishEnabledCodecs []*livekit.Cod func (p *ParticipantImpl) GetEnabledPublishCodecs() []*livekit.Codec { codecs := make([]*livekit.Codec, 0, len(p.enabledPublishCodecs)) for _, c := range p.enabledPublishCodecs { - if c.Mime == "video/rtx" { + if strings.EqualFold(c.Mime, webrtc.MimeTypeRTX) { continue } codecs = append(codecs, c) diff --git a/pkg/sfu/downtrack.go b/pkg/sfu/downtrack.go index ab285f69e..5dc790240 100644 --- a/pkg/sfu/downtrack.go +++ b/pkg/sfu/downtrack.go @@ -20,6 +20,7 @@ import ( "fmt" "io" "math" + "math/rand" "strings" "sync" "time" @@ -217,13 +218,16 @@ type DowntrackParams struct { // - closed // once closed, a DownTrack cannot be re-used. type DownTrack struct { - params DowntrackParams - id livekit.TrackID - kind webrtc.RTPCodecType - mime string - ssrc uint32 - payloadType uint8 - sequencer *sequencer + params DowntrackParams + id livekit.TrackID + kind webrtc.RTPCodecType + mime string + ssrc uint32 + ssrcRTX uint32 + payloadType uint8 + payloadTypeRTX uint8 + sequencer *sequencer + rtxSequenceNumber atomic.Uint64 forwarder *Forwarder @@ -244,6 +248,7 @@ type DownTrack struct { transceiver atomic.Pointer[webrtc.RTPTransceiver] writeStream webrtc.TrackLocalWriter rtcpReader *buffer.RTCPReader + rtcpReaderRTX *buffer.RTCPReader listenerLock sync.RWMutex receiverReportListeners []ReceiverReportListener @@ -387,6 +392,7 @@ func NewDownTrack(params DowntrackParams) (*DownTrack, error) { } d.params.Receiver.AddOnReady(d.handleReceiverReady) + d.rtxSequenceNumber.Store(uint64(rand.Intn(1<<14)) + uint64(1<<15)) // a random number in third quartile of sequence number space d.params.Logger.Debugw("downtrack created", "upstreamCodecs", d.upstreamCodecs) return d, nil @@ -490,6 +496,7 @@ func (d *DownTrack) Bind(t webrtc.TrackLocalContext) (webrtc.RTPCodecParameters, "codecs", d.upstreamCodecs, "matchCodec", codec, "ssrc", t.SSRC(), + "ssrcRTX", t.SSRCRetransmission(), "isFECEnabled", isFECEnabled, } if d.isRED { @@ -504,15 +511,23 @@ func (d *DownTrack) Bind(t webrtc.TrackLocalContext) (webrtc.RTPCodecParameters, ) d.ssrc = uint32(t.SSRC()) + d.ssrcRTX = uint32(t.SSRCRetransmission()) d.payloadType = uint8(codec.PayloadType) + d.payloadTypeRTX = uint8(utils.FindRTXPayloadType(codec.PayloadType, t.CodecParameters())) d.writeStream = t.WriteStream() d.mime = strings.ToLower(codec.MimeType) - if rr := d.params.BufferFactory.GetOrNew(packetio.RTCPBufferPacket, uint32(t.SSRC())).(*buffer.RTCPReader); rr != nil { + if rr := d.params.BufferFactory.GetOrNew(packetio.RTCPBufferPacket, d.ssrc).(*buffer.RTCPReader); rr != nil { rr.OnPacket(func(pkt []byte) { d.handleRTCP(pkt) }) d.rtcpReader = rr } + if rr := d.params.BufferFactory.GetOrNew(packetio.RTCPBufferPacket, d.ssrcRTX).(*buffer.RTCPReader); rr != nil { + rr.OnPacket(func(pkt []byte) { + d.handleRTCP(pkt) + }) + d.rtcpReaderRTX = rr + } d.sequencer = newSequencer(d.params.MaxTrack, d.kind == webrtc.RTPCodecTypeVideo, d.params.Logger) @@ -1154,6 +1169,11 @@ func (d *DownTrack) CloseWithFlush(flush bool) { d.rtcpReader.Close() d.rtcpReader.OnPacket(nil) } + if d.rtcpReaderRTX != nil && flush { + d.params.Logger.Debugw("downtrack close rtcp rtx reader") + d.rtcpReaderRTX.Close() + d.rtcpReaderRTX.OnPacket(nil) + } d.bindLock.Unlock() d.connectionStats.Close() @@ -1761,7 +1781,7 @@ func (d *DownTrack) handleRTCP(bytes []byte) { } case *rtcp.TransportLayerCC: - if p.MediaSSRC == d.ssrc { + if p.MediaSSRC == d.ssrc || p.MediaSSRC == d.ssrcRTX { if sal := d.getStreamAllocatorListener(); sal != nil { sal.OnTransportCCFeedback(d, p) } @@ -1883,17 +1903,13 @@ func (d *DownTrack) retransmitPackets(nacks []uint16) { Timestamp: epm.timestamp, SSRC: d.ssrc, } + rtxOffset := 0 + if d.payloadTypeRTX != 0 && d.ssrcRTX != 0 { + rtxOffset = 2 - poolEntity := PacketFactory.Get().(*[]byte) - payload := *poolEntity - if len(epm.codecBytesSlice) != 0 { - n := copy(payload, epm.codecBytesSlice) - m := copy(payload[n:], pkt.Payload[epm.numCodecBytesIn:]) - payload = payload[:n+m] - } else { - copy(payload, epm.codecBytes[:epm.numCodecBytesOut]) - copy(payload[epm.numCodecBytesOut:], pkt.Payload[epm.numCodecBytesIn:]) - payload = payload[:int(epm.numCodecBytesOut)+len(pkt.Payload)-int(epm.numCodecBytesIn)] + hdr.PayloadType = d.payloadTypeRTX + hdr.SequenceNumber = uint16(d.rtxSequenceNumber.Inc()) + hdr.SSRC = d.ssrcRTX } if d.dependencyDescriptorExtID != 0 { @@ -1912,6 +1928,22 @@ func (d *DownTrack) retransmitPackets(nacks []uint16) { } d.addDummyExtensions(hdr) + poolEntity := PacketFactory.Get().(*[]byte) + payload := *poolEntity + if rtxOffset != 0 { + // write OSN (Original Sequence Number) + binary.BigEndian.PutUint16(payload[0:2], epm.targetSeqNo) + } + if len(epm.codecBytesSlice) != 0 { + n := copy(payload[rtxOffset:], epm.codecBytesSlice) + m := copy(payload[rtxOffset+n:], pkt.Payload[epm.numCodecBytesIn:]) + payload = payload[:rtxOffset+n+m] + } else { + copy(payload[rtxOffset:], epm.codecBytes[:epm.numCodecBytesOut]) + copy(payload[rtxOffset+int(epm.numCodecBytesOut):], pkt.Payload[epm.numCodecBytesIn:]) + payload = payload[:rtxOffset+int(epm.numCodecBytesOut)+len(pkt.Payload)-int(epm.numCodecBytesIn)] + } + headerSize := hdr.MarshalSize() d.rtpStats.Update( mono.UnixNano(), diff --git a/pkg/sfu/utils/helpers.go b/pkg/sfu/utils/helpers.go index 79dc2a0a2..09990c183 100644 --- a/pkg/sfu/utils/helpers.go +++ b/pkg/sfu/utils/helpers.go @@ -45,6 +45,18 @@ func CodecParametersFuzzySearch(needle webrtc.RTPCodecParameters, haystack []web return webrtc.RTPCodecParameters{}, webrtc.ErrCodecNotFound } +// Given a CodecParameters find the RTX CodecParameters if one exists +func FindRTXPayloadType(needle webrtc.PayloadType, haystack []webrtc.RTPCodecParameters) webrtc.PayloadType { + aptStr := fmt.Sprintf("apt=%d", needle) + for _, c := range haystack { + if aptStr == c.SDPFmtpLine { + return c.PayloadType + } + } + + return webrtc.PayloadType(0) +} + // GetHeaderExtensionID returns the ID of a header extension, or 0 if not found func GetHeaderExtensionID(extensions []interceptor.RTPHeaderExtension, extension webrtc.RTPHeaderExtensionCapability) int { for _, h := range extensions { diff --git a/test/client/client.go b/test/client/client.go index 78ae601d5..977616c86 100644 --- a/test/client/client.go +++ b/test/client/client.go @@ -42,6 +42,7 @@ import ( "github.com/livekit/livekit-server/pkg/rtc" "github.com/livekit/livekit-server/pkg/rtc/transport/transportfakes" "github.com/livekit/livekit-server/pkg/rtc/types" + "github.com/livekit/livekit-server/pkg/sfu/buffer" ) type SignalRequestHandler func(msg *livekit.SignalRequest) error @@ -172,6 +173,8 @@ func NewRTCClient(conn *websocket.Conn, opts *Options) (*RTCClient, error) { } conf.SettingEngine.SetLite(false) conf.SettingEngine.SetAnsweringDTLSRole(webrtc.DTLSRoleClient) + ff := buffer.NewFactoryOfBufferFactory(500, 200) + conf.SetBufferFactory(ff.CreateBufferFactory()) var codecs []*livekit.Codec for _, codec := range []*livekit.Codec{ {