dandri/livekit
1
0
Fork 0
mirror of https://github.com/livekit/livekit.git synced 2026-09-16 06:22:36 +00:00
Code Issues Packages Projects Releases Wiki Activity
Files
5ed3a5ae955cf67eab8970bc9cf419fc99af1f85
livekit/pkg/sfu/downtrack.go
T
Raja Subramanian 1ab2bf043b Clean up packet size logging (#4536) 
Reverting
- https://github.com/livekit/livekit/pull/4521
- https://github.com/livekit/livekit/pull/4525

There are TWCC feedback packets that are larger than MTU. Seems to
happen under a couple of conditions
1. Bad client data, i. e. severely out-of-order packets, bad sequence
   numbers, etc.
2. On an ICE restart - this is rare, but it seemed to be flaky network
   with some packets arriving and some not and causing a lot of gaps.

Either case, not much to do. If fargmentation/re-assembly back to
publisher works, the feedback will make it through. If not, feedbacks
will be missed and clients have to work with some missing data which is
not unexpected and the protocol is designed to handle.

However, filed pion/interceptor issue just in case - https://github.com/pion/interceptor/issues/416
2026-05-20 23:58:05 +05:30

2683 lines
81 KiB
Go
Raw Blame History

// Copyright 2023 LiveKit, Inc.
//
// Licensed under the Apache License, Version 2.0 (the "License");
// you may not use this file except in compliance with the License.
// You may obtain a copy of the License at
//
// http://www.apache.org/licenses/LICENSE-2.0
//
// Unless required by applicable law or agreed to in writing, software
// distributed under the License is distributed on an "AS IS" BASIS,
// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
// See the License for the specific language governing permissions and
// limitations under the License.
package sfu
import (
"encoding/binary"
"errors"
"fmt"
"io"
"math"
"math/rand"
"strings"
"sync"
"time"
"github.com/pion/rtcp"
"github.com/pion/rtp"
"github.com/pion/sdp/v3"
"github.com/pion/transport/v4/packetio"
"github.com/pion/webrtc/v4"
"go.uber.org/atomic"
"go.uber.org/zap/zapcore"
"github.com/livekit/protocol/codecs/mime"
"github.com/livekit/protocol/livekit"
"github.com/livekit/protocol/logger"
"github.com/livekit/protocol/utils/mono"
"github.com/livekit/livekit-server/pkg/sfu/buffer"
"github.com/livekit/livekit-server/pkg/sfu/bwe"
"github.com/livekit/livekit-server/pkg/sfu/ccutils"
"github.com/livekit/livekit-server/pkg/sfu/connectionquality"
"github.com/livekit/livekit-server/pkg/sfu/pacer"
"github.com/livekit/livekit-server/pkg/sfu/packettrailer"
act "github.com/livekit/livekit-server/pkg/sfu/rtpextension/abscapturetime"
dd "github.com/livekit/livekit-server/pkg/sfu/rtpextension/dependencydescriptor"
pd "github.com/livekit/livekit-server/pkg/sfu/rtpextension/playoutdelay"
"github.com/livekit/livekit-server/pkg/sfu/rtpstats"
"github.com/livekit/livekit-server/pkg/sfu/utils"
)
// TrackSender defines an interface send media to remote peer
type TrackSender interface {
UpTrackLayersChange()
UpTrackBitrateAvailabilityChange()
UpTrackMaxPublishedLayerChange(maxPublishedLayer int32)
UpTrackMaxTemporalLayerSeenChange(maxTemporalLayerSeen int32)
UpTrackBitrateReport(availableLayers []int32, bitrates Bitrates)
WriteRTP(p *buffer.ExtPacket, layer int32) int32
Close()
IsClosed() bool
// ID is the globally unique identifier for this Track.
ID() string
SubscriberID() livekit.ParticipantID
HandleRTCPSenderReportData(
payloadType webrtc.PayloadType,
layer int32,
publisherSRData *livekit.RTCPSenderReportState,
) error
Resync()
SetReceiver(TrackReceiver)
ReceiverRestart(TrackReceiver)
}
// -------------------------------------------------------------------
const (
RTPPaddingMaxPayloadSize = 255
RTPPaddingEstimatedHeaderSize = 20
RTPBlankFramesMuteSeconds = float32(1.0)
RTPBlankFramesCloseSeconds = float32(0.2)
FlagStopRTXOnPLI = true
keyFrameIntervalMin = 200
keyFrameIntervalMax = 1000
flushTimeout = 1 * time.Second
waitBeforeSendPaddingOnMute = 100 * time.Millisecond
maxPaddingOnMuteDuration = 5 * time.Second
paddingOnMuteInterval = 100 * time.Millisecond
)
// -------------------------------------------------------------------
var (
errUnknownKind = errors.New("unknown kind of codec")
errOutOfOrderSequenceNumberCacheMiss = errors.New("out-of-order sequence number not found in cache")
errPaddingOnlyPacket = errors.New("padding only packet that need not be forwarded")
errDuplicatePacket = errors.New("duplicate packet")
errPaddingNotOnFrameBoundary = errors.New("padding cannot send on non-frame boundary")
errDownTrackAlreadyBound = errors.New("already bound")
errPayloadOverflow = errors.New("payload overflow")
)
var (
VP8KeyFrame8x8 = []byte{
0x10, 0x02, 0x00, 0x9d, 0x01, 0x2a, 0x08, 0x00,
0x08, 0x00, 0x00, 0x47, 0x08, 0x85, 0x85, 0x88,
0x85, 0x84, 0x88, 0x02, 0x02, 0x00, 0x0c, 0x0d,
0x60, 0x00, 0xfe, 0xff, 0xab, 0x50, 0x80,
}
H264KeyFrame2x2SPS = []byte{
0x67, 0x42, 0xc0, 0x1f, 0x0f, 0xd9, 0x1f, 0x88,
0x88, 0x84, 0x00, 0x00, 0x03, 0x00, 0x04, 0x00,
0x00, 0x03, 0x00, 0xc8, 0x3c, 0x60, 0xc9, 0x20,
}
H264KeyFrame2x2PPS = []byte{
0x68, 0x87, 0xcb, 0x83, 0xcb, 0x20,
}
H264KeyFrame2x2IDR = []byte{
0x65, 0x88, 0x84, 0x0a, 0xf2, 0x62, 0x80, 0x00,
0xa7, 0xbe,
}
H264KeyFrame2x2 = [][]byte{H264KeyFrame2x2SPS, H264KeyFrame2x2PPS, H264KeyFrame2x2IDR}
OpusSilenceFrame = []byte{
0xf8, 0xff, 0xfe, 0x00, 0x00, 0x00, 0x00, 0x00,
0x00, 0x00, 0x00, 0x00, 0x00, 0x00, 0x00, 0x00,
0x00, 0x00, 0x00, 0x00, 0x00, 0x00, 0x00, 0x00,
0x00, 0x00, 0x00, 0x00, 0x00, 0x00, 0x00, 0x00,
0x00, 0x00, 0x00, 0x00, 0x00, 0x00, 0x00, 0x00,
0x00, 0x00, 0x00, 0x00, 0x00, 0x00, 0x00, 0x00,
0x00, 0x00, 0x00, 0x00, 0x00, 0x00, 0x00, 0x00,
0x00, 0x00, 0x00, 0x00, 0x00, 0x00, 0x00, 0x00,
0x00, 0x00, 0x00, 0x00, 0x00, 0x00, 0x00, 0x00,
0x00, 0x00, 0x00, 0x00, 0x00, 0x00, 0x00, 0x00,
}
// PCMU (G.711 µ-law) silence frame - 0xff represents zero amplitude
// 160 samples = 20ms at 8kHz sample rate
PCMUSilenceFrame = []byte{
0xff, 0xff, 0xff, 0xff, 0xff, 0xff, 0xff, 0xff,
0xff, 0xff, 0xff, 0xff, 0xff, 0xff, 0xff, 0xff,
0xff, 0xff, 0xff, 0xff, 0xff, 0xff, 0xff, 0xff,
0xff, 0xff, 0xff, 0xff, 0xff, 0xff, 0xff, 0xff,
0xff, 0xff, 0xff, 0xff, 0xff, 0xff, 0xff, 0xff,
0xff, 0xff, 0xff, 0xff, 0xff, 0xff, 0xff, 0xff,
0xff, 0xff, 0xff, 0xff, 0xff, 0xff, 0xff, 0xff,
0xff, 0xff, 0xff, 0xff, 0xff, 0xff, 0xff, 0xff,
0xff, 0xff, 0xff, 0xff, 0xff, 0xff, 0xff, 0xff,
0xff, 0xff, 0xff, 0xff, 0xff, 0xff, 0xff, 0xff,
0xff, 0xff, 0xff, 0xff, 0xff, 0xff, 0xff, 0xff,
0xff, 0xff, 0xff, 0xff, 0xff, 0xff, 0xff, 0xff,
0xff, 0xff, 0xff, 0xff, 0xff, 0xff, 0xff, 0xff,
0xff, 0xff, 0xff, 0xff, 0xff, 0xff, 0xff, 0xff,
0xff, 0xff, 0xff, 0xff, 0xff, 0xff, 0xff, 0xff,
0xff, 0xff, 0xff, 0xff, 0xff, 0xff, 0xff, 0xff,
0xff, 0xff, 0xff, 0xff, 0xff, 0xff, 0xff, 0xff,
0xff, 0xff, 0xff, 0xff, 0xff, 0xff, 0xff, 0xff,
0xff, 0xff, 0xff, 0xff, 0xff, 0xff, 0xff, 0xff,
0xff, 0xff, 0xff, 0xff, 0xff, 0xff, 0xff, 0xff,
}
// PCMA (G.711 A-law) silence frame - 0xd5 represents zero amplitude
// 160 samples = 20ms at 8kHz sample rate
PCMASilenceFrame = []byte{
0xd5, 0xd5, 0xd5, 0xd5, 0xd5, 0xd5, 0xd5, 0xd5,
0xd5, 0xd5, 0xd5, 0xd5, 0xd5, 0xd5, 0xd5, 0xd5,
0xd5, 0xd5, 0xd5, 0xd5, 0xd5, 0xd5, 0xd5, 0xd5,
0xd5, 0xd5, 0xd5, 0xd5, 0xd5, 0xd5, 0xd5, 0xd5,
0xd5, 0xd5, 0xd5, 0xd5, 0xd5, 0xd5, 0xd5, 0xd5,
0xd5, 0xd5, 0xd5, 0xd5, 0xd5, 0xd5, 0xd5, 0xd5,
0xd5, 0xd5, 0xd5, 0xd5, 0xd5, 0xd5, 0xd5, 0xd5,
0xd5, 0xd5, 0xd5, 0xd5, 0xd5, 0xd5, 0xd5, 0xd5,
0xd5, 0xd5, 0xd5, 0xd5, 0xd5, 0xd5, 0xd5, 0xd5,
0xd5, 0xd5, 0xd5, 0xd5, 0xd5, 0xd5, 0xd5, 0xd5,
0xd5, 0xd5, 0xd5, 0xd5, 0xd5, 0xd5, 0xd5, 0xd5,
0xd5, 0xd5, 0xd5, 0xd5, 0xd5, 0xd5, 0xd5, 0xd5,
0xd5, 0xd5, 0xd5, 0xd5, 0xd5, 0xd5, 0xd5, 0xd5,
0xd5, 0xd5, 0xd5, 0xd5, 0xd5, 0xd5, 0xd5, 0xd5,
0xd5, 0xd5, 0xd5, 0xd5, 0xd5, 0xd5, 0xd5, 0xd5,
0xd5, 0xd5, 0xd5, 0xd5, 0xd5, 0xd5, 0xd5, 0xd5,
0xd5, 0xd5, 0xd5, 0xd5, 0xd5, 0xd5, 0xd5, 0xd5,
0xd5, 0xd5, 0xd5, 0xd5, 0xd5, 0xd5, 0xd5, 0xd5,
0xd5, 0xd5, 0xd5, 0xd5, 0xd5, 0xd5, 0xd5, 0xd5,
0xd5, 0xd5, 0xd5, 0xd5, 0xd5, 0xd5, 0xd5, 0xd5,
}
dummyAbsSendTimeExt, _ = rtp.NewAbsSendTimeExtension(mono.Now()).Marshal()
dummyTransportCCExt, _ = rtp.TransportCCExtension{TransportSequence: 12345}.Marshal()
)
// -------------------------------------------------------------------
type DownTrackState struct {
RTPStats *rtpstats.RTPStatsSender
DeltaStatsSenderSnapshotId uint32
RTPStatsRTX *rtpstats.RTPStatsSender
DeltaStatsRTXSenderSnapshotId uint32
ForwarderState *livekit.RTPForwarderState
PlayoutDelayControllerState PlayoutDelayControllerState
}
func (d DownTrackState) MarshalLogObject(e zapcore.ObjectEncoder) error {
e.AddObject("RTPStats", d.RTPStats)
e.AddUint32("DeltaStatsSenderSnapshotId", d.DeltaStatsSenderSnapshotId)
e.AddObject("RTPStatsRTX", d.RTPStatsRTX)
e.AddUint32("DeltaStatsRTXSenderSnapshotId", d.DeltaStatsRTXSenderSnapshotId)
e.AddObject("ForwarderState", logger.Proto(d.ForwarderState))
e.AddObject("PlayoutDelayControllerState", d.PlayoutDelayControllerState)
return nil
}
// -------------------------------------------------------------------
type DownTrackStreamAllocatorListener interface {
// RTCP received
OnREMB(dt *DownTrack, remb *rtcp.ReceiverEstimatedMaximumBitrate)
OnTransportCCFeedback(dt *DownTrack, cc *rtcp.TransportLayerCC)
// video layer availability changed
OnAvailableLayersChanged(dt *DownTrack)
// video layer bitrate availability changed
OnBitrateAvailabilityChanged(dt *DownTrack)
// max published spatial layer changed
OnMaxPublishedSpatialChanged(dt *DownTrack)
// max published temporal layer changed
OnMaxPublishedTemporalChanged(dt *DownTrack)
// subscription changed - mute/unmute
OnSubscriptionChanged(dt *DownTrack)
// subscribed max video layer changed
OnSubscribedLayerChanged(dt *DownTrack, layers buffer.VideoLayer)
// stream resumed
OnResume(dt *DownTrack)
// check if track should participate in BWE
IsBWEEnabled(dt *DownTrack) bool
// get the BWE type in use
BWEType() bwe.BWEType
// check if subscription mute can be applied
IsSubscribeMutable(dt *DownTrack) bool
}
// -------------------------------------------------------------------
type DownTrackListener interface {
OnBindAndConnected()
OnStatsUpdate(stat *livekit.AnalyticsStat)
OnMaxSubscribedLayerChanged(layer int32)
OnRttUpdate(rtt uint32)
OnCodecNegotiated(webrtc.RTPCodecCapability)
OnDownTrackClose(isExpectedToResume bool)
OnStreamStarted()
}
// -------------------------------------------------------------------
type bindState int
const (
bindStateUnbound bindState = iota
// downtrack negotiated, but waiting for receiver to be ready to start forwarding
bindStateWaitForReceiverReady
// downtrack is bound and ready to forward
bindStateBound
)
func (bs bindState) String() string {
switch bs {
case bindStateUnbound:
return "unbound"
case bindStateWaitForReceiverReady:
return "waitForReceiverReady"
case bindStateBound:
return "bound"
}
return "unknown"
}
// -------------------------------------------------------------------
var _ TrackSender = (*DownTrack)(nil)
type ReceiverReportListener func(dt *DownTrack, report *rtcp.ReceiverReport)
type DownTrackParams struct {
Codecs []webrtc.RTPCodecParameters
IsEncrypted bool
Source livekit.TrackSource
Receiver TrackReceiver
BufferFactory *buffer.Factory
SubID livekit.ParticipantID
StreamID string
MaxTrack int
PlayoutDelayLimit *livekit.PlayoutDelay
Pacer pacer.Pacer
Logger logger.Logger
Trailer []byte
RTCPWriter func([]rtcp.Packet) error
DisableSenderReportPassThrough bool
SupportsCodecChange bool
StripPacketTrailer bool
Listener DownTrackListener
}
// DownTrack implements webrtc.TrackLocal, is the track used to write packets
// to SFU Subscriber, the track handle the packets for simple, simulcast
// and SVC Publisher.
// A DownTrack has the following lifecycle
// - new
// - bound / unbound
// - closed
// once closed, a DownTrack cannot be re-used.
type DownTrack struct {
params DownTrackParams
id livekit.TrackID
kind webrtc.RTPCodecType
ssrc uint32
ssrcRTX uint32
payloadType atomic.Uint32
payloadTypeRTX atomic.Uint32
sequencer *sequencer
rtxSequenceNumber atomic.Uint64
receiverLock sync.RWMutex
receiver TrackReceiver
forwarder *Forwarder
upstreamCodecs []webrtc.RTPCodecParameters
codec atomic.Value // webrtc.RTPCodecCapability
clockRate uint32
negotiatedCodecParameters []webrtc.RTPCodecParameters
// payload types for red codec only
isRED bool
upstreamPrimaryPT uint8
primaryPT uint8
absSendTimeExtID int
transportWideExtID int
dependencyDescriptorExtID int
playoutDelayExtID int
absCaptureTimeExtID int
transceiver atomic.Pointer[webrtc.RTPTransceiver]
writeStream webrtc.TrackLocalWriter
rtcpReader *buffer.RTCPReader
rtcpReaderRTX *buffer.RTCPReader
listenerLock sync.RWMutex
receiverReportListeners []ReceiverReportListener
bindLock sync.Mutex
bindState atomic.Value
onBinding func(error)
bindOnReceiverReady func()
isClosed atomic.Bool
connected atomic.Bool
bindAndConnectedOnce atomic.Bool
writable atomic.Bool
writeStopped atomic.Bool
isReceiverReady bool
rtpStats *rtpstats.RTPStatsSender
deltaStatsSenderSnapshotId uint32
rtpStatsRTX *rtpstats.RTPStatsSender
deltaStatsRTXSenderSnapshotId uint32
totalRepeatedNACKs atomic.Uint32
blankFramesGeneration atomic.Uint32
connectionStats *connectionquality.ConnectionStats
onStatsUpdate atomic.Value // func(d *DownTrack, stat *livekit.AnalyticsStat)
isNACKThrottled atomic.Bool
activePaddingOnMuteUpTrack atomic.Bool
streamAllocatorLock sync.RWMutex
streamAllocatorListener DownTrackStreamAllocatorListener
probeClusterId atomic.Uint32
playoutDelay *PlayoutDelayController
pacer pacer.Pacer
maxLayerNotifierChMu sync.RWMutex
maxLayerNotifierCh chan string
maxLayerNotifierChClosed bool
keyFrameRequesterChMu sync.RWMutex
keyFrameRequesterCh chan struct{}
keyFrameRequesterChClosed bool
createdAt int64
}
// NewDownTrack returns a DownTrack.
func NewDownTrack(params DownTrackParams) (*DownTrack, error) {
mimeType := mime.NormalizeMimeType(params.Codecs[0].MimeType)
var kind webrtc.RTPCodecType
switch {
case mime.IsMimeTypeAudio(mimeType):
kind = webrtc.RTPCodecTypeAudio
case mime.IsMimeTypeVideo(mimeType):
kind = webrtc.RTPCodecTypeVideo
default:
kind = webrtc.RTPCodecType(0)
}
codec := params.Codecs[0].RTPCodecCapability
d := &DownTrack{
params: params,
id: params.Receiver.TrackID(),
upstreamCodecs: params.Codecs,
kind: kind,
clockRate: codec.ClockRate,
pacer: params.Pacer,
maxLayerNotifierCh: make(chan string, 1),
keyFrameRequesterCh: make(chan struct{}, 1),
createdAt: time.Now().UnixNano(),
receiver: params.Receiver,
}
d.codec.Store(codec)
d.bindState.Store(bindStateUnbound)
d.params.Logger = params.Logger.WithValues(
"subscriberID", d.SubscriberID(),
)
var mdCacheSize, mdCacheSizeRTX int
if d.kind == webrtc.RTPCodecTypeVideo {
mdCacheSize, mdCacheSizeRTX = 8192, 8192
} else {
mdCacheSize, mdCacheSizeRTX = 8192, 1024
}
d.rtpStats = rtpstats.NewRTPStatsSender(rtpstats.RTPStatsParams{}, mdCacheSize)
// clock rate will be set on bind or codec change with matching codec's clock rate
d.rtpStats.SetLogger(d.params.Logger.WithValues("stream", "primary"))
d.deltaStatsSenderSnapshotId = d.rtpStats.NewSenderSnapshotId()
d.rtpStatsRTX = rtpstats.NewRTPStatsSender(rtpstats.RTPStatsParams{IsRTX: true}, mdCacheSizeRTX)
// clock rate will be set on bind or codec change with matching codec's clock rate
d.rtpStatsRTX.SetLogger(d.params.Logger.WithValues("stream", "rtx"))
d.deltaStatsRTXSenderSnapshotId = d.rtpStatsRTX.NewSenderSnapshotId()
d.forwarder = NewForwarder(
d.kind,
d.params.Logger,
false, // skipReferenceTS
false, // disableOpportunisticAllocation
d.rtpStats,
)
d.connectionStats = connectionquality.NewConnectionStats(connectionquality.ConnectionStatsParams{
SenderProvider: d,
Logger: d.params.Logger.WithValues("direction", "down"),
})
d.connectionStats.OnStatsUpdate(func(_cs *connectionquality.ConnectionStats, stat *livekit.AnalyticsStat) {
d.params.Listener.OnStatsUpdate(stat)
if fn, ok := d.onStatsUpdate.Load().(func(*DownTrack, *livekit.AnalyticsStat)); ok && fn != nil {
fn(d, stat)
}
})
if d.kind == webrtc.RTPCodecTypeVideo {
if delay := params.PlayoutDelayLimit; delay.GetEnabled() {
var err error
d.playoutDelay, err = NewPlayoutDelayController(delay.GetMin(), delay.GetMax(), params.Logger, d.rtpStats)
if err != nil {
return nil, err
}
}
go d.maxLayerNotifierWorker()
go d.keyFrameRequester()
}
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
}
// Bind is called by the PeerConnection after negotiation is complete
// This asserts that the code requested is supported by the remote peer.
// If so it sets up all the state (SSRC and PayloadType) to have a call
func (d *DownTrack) Bind(t webrtc.TrackLocalContext) (webrtc.RTPCodecParameters, error) {
d.bindLock.Lock()
if d.bindState.Load() != bindStateUnbound {
d.bindLock.Unlock()
return webrtc.RTPCodecParameters{}, errDownTrackAlreadyBound
}
// the TrackLocalContext's codec parameters will be set to the bound codec after Bind returns,
// so keep a copy of the codec parameters here to use it later
d.negotiatedCodecParameters = append([]webrtc.RTPCodecParameters{}, t.CodecParameters()...)
var codec, matchedUpstreamCodec webrtc.RTPCodecParameters
for _, c := range d.upstreamCodecs {
matchCodec, err := utils.CodecParametersFuzzySearch(c, d.negotiatedCodecParameters)
if err == nil {
codec = matchCodec
matchedUpstreamCodec = c
break
} else {
// for encrypyted tracks, should match on primary codec,
// i. e. codec at index 0 if the combination of upstream codecs is opus and RED
if d.params.IsEncrypted {
isRedAndOpus := true
for _, u := range d.upstreamCodecs {
if !mime.IsMimeTypeStringOpus(u.MimeType) || !mime.IsMimeTypeStringRED(u.MimeType) {
isRedAndOpus = false
break
}
}
if isRedAndOpus {
break
}
}
}
}
if codec.MimeType == "" {
err := webrtc.ErrUnsupportedCodec
onBinding := d.onBinding
d.bindLock.Unlock()
d.params.Logger.Infow(
"bind error for unsupported codec",
"codecs", d.upstreamCodecs,
"remoteParameters", d.negotiatedCodecParameters,
)
if onBinding != nil {
onBinding(err)
}
// don't return error here, as pion will not start transports if Bind fails at first answer
return webrtc.RTPCodecParameters{}, nil
}
// if a downtrack is closed before bind, it already unsubscribed from client, don't do subsequent operation and return here.
if d.IsClosed() {
d.params.Logger.Debugw("DownTrack closed before bind")
d.bindLock.Unlock()
return codec, nil
}
// Bind is called under RTPSender.mu lock,
// call the RTPSender.GetParameters (which setRTPHeaderExtensions invokes)
// in goroutine to avoid deadlock
go d.setRTPHeaderExtensions()
doBind := func() {
d.bindLock.Lock()
if d.IsClosed() {
d.bindLock.Unlock()
d.params.Logger.Debugw("DownTrack closed before bind")
return
}
isFECEnabled := false
if mime.IsMimeTypeStringRED(matchedUpstreamCodec.MimeType) {
d.isRED = true
for _, c := range d.upstreamCodecs {
isFECEnabled = strings.Contains(strings.ToLower(c.SDPFmtpLine), "useinbandfec=1")
// assume upstream primary codec is opus since we only support it for audio now
if mime.IsMimeTypeStringOpus(c.MimeType) {
d.upstreamPrimaryPT = uint8(c.PayloadType)
break
}
}
if d.upstreamPrimaryPT == 0 {
d.params.Logger.Errorw(
"failed to find upstream primary opus payload type for RED", nil,
"matchedCodec", codec,
"upstreamCodec", d.upstreamCodecs,
)
}
var primaryPT, secondaryPT int
if n, err := fmt.Sscanf(codec.SDPFmtpLine, "%d/%d", &primaryPT, &secondaryPT); err != nil || n != 2 {
d.params.Logger.Errorw(
"failed to parse primary and secondary payload type for RED", err,
"matchedCodec", codec,
)
}
d.primaryPT = uint8(primaryPT)
} else if mime.IsMimeTypeStringAudio(matchedUpstreamCodec.MimeType) {
isFECEnabled = strings.Contains(strings.ToLower(matchedUpstreamCodec.SDPFmtpLine), "fec")
}
logFields := []any{
"codecs", d.upstreamCodecs,
"matchCodec", codec,
"ssrc", t.SSRC(),
"ssrcRTX", t.SSRCRetransmission(),
"isFECEnabled", isFECEnabled,
}
if d.isRED {
logFields = append(
logFields,
"isRED", d.isRED,
"upstreamPrimaryPT", d.upstreamPrimaryPT,
"primaryPT", d.primaryPT,
)
}
d.ssrc = uint32(t.SSRC())
d.ssrcRTX = uint32(t.SSRCRetransmission())
d.payloadType.Store(uint32(codec.PayloadType))
d.payloadTypeRTX.Store(uint32(utils.FindRTXPayloadType(codec.PayloadType, d.negotiatedCodecParameters)))
logFields = append(
logFields,
"payloadType", d.payloadType.Load(),
"payloadTypeRTX", d.payloadTypeRTX.Load(),
"codecParameters", d.negotiatedCodecParameters,
)
d.params.Logger.Debugw("DownTrack.Bind", logFields...)
d.writeStream = t.WriteStream()
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 d.ssrcRTX != 0 {
if rr := d.params.BufferFactory.GetOrNew(packetio.RTCPBufferPacket, d.ssrcRTX).(*buffer.RTCPReader); rr != nil {
rr.OnPacket(func(pkt []byte) {
d.handleRTCPRTX(pkt)
})
d.rtcpReaderRTX = rr
}
}
d.sequencer = newSequencer(d.params.MaxTrack, d.kind == webrtc.RTPCodecTypeVideo, d.params.Logger)
d.codec.Store(codec.RTPCodecCapability)
d.rtpStats.SetClockRate(codec.RTPCodecCapability.ClockRate)
d.rtpStatsRTX.SetClockRate(codec.RTPCodecCapability.ClockRate)
if d.onBinding != nil {
d.onBinding(nil)
}
d.setBindStateLocked(bindStateBound)
d.bindLock.Unlock()
receiver := d.Receiver()
d.forwarder.DetermineCodec(codec.RTPCodecCapability, receiver.HeaderExtensions(), receiver.VideoLayerMode())
d.connectionStats.Start(d.Mime(), isFECEnabled)
d.params.Logger.Debugw("downtrack bound")
}
isReceiverReady := d.isReceiverReady
if !isReceiverReady {
d.params.Logger.Debugw("downtrack bound: receiver not ready", "codec", codec)
d.bindOnReceiverReady = doBind
d.setBindStateLocked(bindStateWaitForReceiverReady)
}
d.bindLock.Unlock()
d.params.Listener.OnCodecNegotiated(codec.RTPCodecCapability)
if isReceiverReady {
doBind()
}
return codec, nil
}
func (d *DownTrack) setBindStateLocked(state bindState) {
if d.bindState.Swap(state) == state {
return
}
if state == bindStateBound || state == bindStateUnbound {
d.bindOnReceiverReady = nil
d.onBindAndConnectedChange()
}
}
func (d *DownTrack) handleReceiverReady() {
d.bindLock.Lock()
if d.isReceiverReady {
d.bindLock.Unlock()
return
}
d.params.Logger.Debugw("downtrack receiver ready")
d.isReceiverReady = true
doBind := d.bindOnReceiverReady
d.bindOnReceiverReady = nil
d.bindLock.Unlock()
if doBind != nil {
doBind()
}
}
func (d *DownTrack) handleUpstreamCodecChange(mimeType string) {
d.bindLock.Lock()
existingMimeType := d.codec.Load().(webrtc.RTPCodecCapability).MimeType
if mime.IsMimeTypeStringEqual(existingMimeType, mimeType) {
d.bindLock.Unlock()
return
}
if !d.params.SupportsCodecChange {
d.bindLock.Unlock()
d.params.Logger.Infow("client doesn't support codec change, renegotiate new codec")
go d.Close()
return
}
oldPT, oldRtxPT, oldCodec := d.payloadType.Load(), d.payloadTypeRTX.Load(), d.codec.Load().(webrtc.RTPCodecCapability)
var codec webrtc.RTPCodecParameters
for _, c := range d.upstreamCodecs {
if !mime.IsMimeTypeStringEqual(c.MimeType, mimeType) {
continue
}
matchCodec, err := utils.CodecParametersFuzzySearch(c, d.negotiatedCodecParameters)
if err == nil {
codec = matchCodec
break
}
}
if codec.MimeType == "" {
// codec not found, should not happen since the upstream codec should only fall back to higher compatibility (vp8)
d.params.Logger.Errorw(
"can't find matched codec for new upstream payload type", nil,
"upstreamCodecs", d.upstreamCodecs,
"remoteParameters", d.negotiatedCodecParameters,
"mime", mimeType,
)
d.bindLock.Unlock()
return
}
d.payloadType.Store(uint32(codec.PayloadType))
d.payloadTypeRTX.Store(uint32(utils.FindRTXPayloadType(codec.PayloadType, d.negotiatedCodecParameters)))
d.codec.Store(codec.RTPCodecCapability)
d.rtpStats.SetClockRate(codec.RTPCodecCapability.ClockRate)
d.rtpStatsRTX.SetClockRate(codec.RTPCodecCapability.ClockRate)
isFECEnabled := strings.Contains(strings.ToLower(codec.SDPFmtpLine), "fec")
d.bindLock.Unlock()
d.params.Logger.Infow(
"upstream codec changed",
"oldPT", oldPT, "newPT", d.payloadType.Load(),
"oldRTXPT", oldRtxPT, "newRTXPT", d.payloadTypeRTX.Load(),
"oldCodec", oldCodec, "newCodec", codec.RTPCodecCapability,
)
receiver := d.Receiver()
d.forwarder.Restart()
d.forwarder.DetermineCodec(codec.RTPCodecCapability, receiver.HeaderExtensions(), receiver.VideoLayerMode())
d.connectionStats.UpdateCodec(d.Mime(), isFECEnabled)
}
// Unbind implements the teardown logic when the track is no longer needed. This happens
// because a track has been stopped.
func (d *DownTrack) Unbind(_ webrtc.TrackLocalContext) error {
d.bindLock.Lock()
d.setBindStateLocked(bindStateUnbound)
d.bindLock.Unlock()
return nil
}
func (d *DownTrack) SetStreamAllocatorListener(listener DownTrackStreamAllocatorListener) {
d.streamAllocatorLock.Lock()
d.streamAllocatorListener = listener
d.streamAllocatorLock.Unlock()
d.setRTPHeaderExtensions()
if listener != nil {
// kick off a gratuitous allocation
listener.OnSubscriptionChanged(d)
}
}
func (d *DownTrack) getStreamAllocatorListener() DownTrackStreamAllocatorListener {
d.streamAllocatorLock.RLock()
defer d.streamAllocatorLock.RUnlock()
return d.streamAllocatorListener
}
func (d *DownTrack) SetProbeClusterId(probeClusterId ccutils.ProbeClusterId) {
d.probeClusterId.Store(uint32(probeClusterId))
}
func (d *DownTrack) SwapProbeClusterId(match ccutils.ProbeClusterId, swap ccutils.ProbeClusterId) {
d.probeClusterId.CompareAndSwap(uint32(match), uint32(swap))
}
// ID is the unique identifier for this Track. This should be unique for the
// stream, but doesn't have to globally unique. A common example would be 'audio' or 'video'
// and StreamID would be 'desktop' or 'webcam'
func (d *DownTrack) ID() string { return string(d.id) }
// Codec returns current track codec capability
func (d *DownTrack) Codec() webrtc.RTPCodecCapability {
return d.codec.Load().(webrtc.RTPCodecCapability)
}
func (d *DownTrack) Mime() mime.MimeType {
return mime.NormalizeMimeType(d.codec.Load().(webrtc.RTPCodecCapability).MimeType)
}
// StreamID is the group this track belongs too. This must be unique
func (d *DownTrack) StreamID() string { return d.params.StreamID }
func (d *DownTrack) SubscriberID() livekit.ParticipantID {
// add `createdAt` to ensure repeated subscriptions from same subscriber to same publisher does not collide
return livekit.ParticipantID(fmt.Sprintf("%s:%d", d.params.SubID, d.createdAt))
}
func (d *DownTrack) Receiver() TrackReceiver {
d.receiverLock.RLock()
defer d.receiverLock.RUnlock()
return d.receiver
}
func (d *DownTrack) SetReceiver(r TrackReceiver) {
d.params.Logger.Debugw("downtrack set receiver", "codec", r.Codec())
d.bindLock.Lock()
if d.IsClosed() {
d.bindLock.Unlock()
return
}
d.receiverLock.Lock()
old := d.receiver
d.receiver = r
d.receiverLock.Unlock()
old.DeleteDownTrack(d.SubscriberID())
d.bindLock.Unlock()
r.AddOnReady(d.handleReceiverReady)
d.handleUpstreamCodecChange(r.Codec().MimeType)
d.bindLock.Lock()
if err := r.AddDownTrack(d); err != nil {
d.params.Logger.Warnw("failed to add downtrack to receiver", err)
}
d.bindLock.Unlock()
if sal := d.getStreamAllocatorListener(); sal != nil {
sal.OnSubscribedLayerChanged(d, d.forwarder.MaxLayer())
}
}
// Sets RTP header extensions for this track
func (d *DownTrack) setRTPHeaderExtensions() {
isBWEEnabled := false
bweType := bwe.BWETypeNone
sal := d.getStreamAllocatorListener()
if sal != nil {
isBWEEnabled = sal.IsBWEEnabled(d)
bweType = sal.BWEType()
}
tr := d.transceiver.Load()
if tr == nil {
return
}
var extensions []webrtc.RTPHeaderExtensionParameter
if sender := tr.Sender(); sender != nil {
extensions = sender.GetParameters().HeaderExtensions
d.params.Logger.Debugw("negotiated downtrack extensions", "extensions", extensions)
}
d.bindLock.Lock()
for _, ext := range extensions {
switch ext.URI {
case sdp.ABSSendTimeURI:
if sal != nil {
if isBWEEnabled && bweType == bwe.BWETypeRemote {
if d.absSendTimeExtID != 0 && d.absSendTimeExtID != ext.ID {
d.params.Logger.Infow("absSendTimeExtID mismatch", "current", d.absSendTimeExtID, "negotiated", ext.ID)
}
d.absSendTimeExtID = ext.ID
} else {
if d.absSendTimeExtID != 0 {
d.params.Logger.Infow("absSendTimeExtID disabled unexpectedly", "negotiated", ext.ID)
}
d.absSendTimeExtID = 0
}
}
case dd.ExtensionURI:
if d.dependencyDescriptorExtID != 0 && d.dependencyDescriptorExtID != ext.ID {
d.params.Logger.Infow("dependencyDescriptorExtID mismatch", "current", d.dependencyDescriptorExtID, "negotiated", ext.ID)
}
d.dependencyDescriptorExtID = ext.ID
case pd.PlayoutDelayURI:
if d.playoutDelayExtID != 0 && d.playoutDelayExtID != ext.ID {
d.params.Logger.Infow("playoutDelayExtID mismatch", "current", d.playoutDelayExtID, "negotiated", ext.ID)
}
d.playoutDelayExtID = ext.ID
case sdp.TransportCCURI:
if sal != nil {
if isBWEEnabled && bweType == bwe.BWETypeSendSide {
if d.transportWideExtID != 0 && d.transportWideExtID != ext.ID {
d.params.Logger.Infow("transportWideExtID mismatch", "current", d.transportWideExtID, "negotiated", ext.ID)
}
d.transportWideExtID = ext.ID
} else {
if d.transportWideExtID != 0 {
d.params.Logger.Infow("transportWideExtID disabled unexpectedly", "negotiated", ext.ID)
}
d.transportWideExtID = 0
}
}
case act.AbsCaptureTimeURI:
if d.absCaptureTimeExtID != 0 && d.absCaptureTimeExtID != ext.ID {
d.params.Logger.Infow("absCaptureTimeExtID mismatch", "current", d.absCaptureTimeExtID, "negotiated", ext.ID)
}
d.absCaptureTimeExtID = ext.ID
}
}
d.params.Logger.Debugw(
"negotiated extension ids",
"absSendTimeExtID", d.absSendTimeExtID,
"dependencyDescriptorExtID", d.dependencyDescriptorExtID,
"playoutDelayExtID", d.playoutDelayExtID,
"transportWideExtID", d.transportWideExtID,
"absCaptureTimeExtID", d.absCaptureTimeExtID,
)
d.bindLock.Unlock()
}
// Kind controls if this TrackLocal is audio or video
func (d *DownTrack) Kind() webrtc.RTPCodecType {
return d.kind
}
// RID is required by `webrtc.TrackLocal` interface
func (d *DownTrack) RID() string {
return ""
}
func (d *DownTrack) SSRC() uint32 {
return d.ssrc
}
func (d *DownTrack) SSRCRTX() uint32 {
return d.ssrcRTX
}
func (d *DownTrack) SetTransceiver(transceiver *webrtc.RTPTransceiver) {
d.transceiver.Store(transceiver)
d.setRTPHeaderExtensions()
}
func (d *DownTrack) GetTransceiver() *webrtc.RTPTransceiver {
return d.transceiver.Load()
}
func (d *DownTrack) postKeyFrameRequestEvent() {
if d.kind != webrtc.RTPCodecTypeVideo {
return
}
d.keyFrameRequesterChMu.RLock()
if !d.keyFrameRequesterChClosed {
select {
case d.keyFrameRequesterCh <- struct{}{}:
default:
}
}
d.keyFrameRequesterChMu.RUnlock()
}
func (d *DownTrack) keyFrameRequester() {
getInterval := func() time.Duration {
interval := min(max(2*d.rtpStats.GetRtt(), keyFrameIntervalMin), keyFrameIntervalMax)
return time.Duration(interval) * time.Millisecond
}
timer := time.NewTimer(math.MaxInt64)
timer.Stop()
defer timer.Stop()
for !d.IsClosed() {
timer.Reset(getInterval())
select {
case _, more := <-d.keyFrameRequesterCh:
if !more {
return
}
if !timer.Stop() {
<-timer.C
}
case <-timer.C:
}
locked, layer := d.forwarder.CheckSync()
if !locked && layer != buffer.InvalidLayerSpatial && d.writable.Load() {
d.params.Logger.Debugw("sending PLI for layer lock", "layer", layer)
d.Receiver().SendPLI(layer, false)
d.rtpStats.UpdateLayerLockPliAndTime(1)
}
}
}
func (d *DownTrack) postMaxLayerNotifierEvent(event string) {
if d.kind != webrtc.RTPCodecTypeVideo {
return
}
d.maxLayerNotifierChMu.RLock()
if !d.maxLayerNotifierChClosed {
select {
case d.maxLayerNotifierCh <- event:
default:
d.params.Logger.Debugw("max layer notifier channel busy", "event", event)
}
}
d.maxLayerNotifierChMu.RUnlock()
}
func (d *DownTrack) maxLayerNotifierWorker() {
for event := range d.maxLayerNotifierCh {
maxLayerSpatial := d.forwarder.GetMaxSubscribedSpatial()
d.params.Logger.Debugw("max subscribed layer processed", "layer", maxLayerSpatial, "event", event)
d.params.Logger.Debugw(
"notifying max subscribed layer",
"layer", maxLayerSpatial,
"event", event,
)
d.params.Listener.OnMaxSubscribedLayerChanged(maxLayerSpatial)
}
d.params.Logger.Debugw(
"notifying max subscribed layer",
"layer", buffer.InvalidLayerSpatial,
"event", "close",
)
d.params.Listener.OnMaxSubscribedLayerChanged(buffer.InvalidLayerSpatial)
}
// WriteRTP writes an RTP Packet to the DownTrack
func (d *DownTrack) WriteRTP(extPkt *buffer.ExtPacket, layer int32) int32 {
if !d.writable.Load() {
return 0
}
tp, err := d.forwarder.GetTranslationParams(extPkt, layer)
if tp.shouldDrop {
if err != nil {
d.params.Logger.Errorw("could not get translation params", err)
}
return 0
}
poolEntity := PacketFactory.Get().(*[]byte)
payload := *poolEntity
copy(payload, tp.codecBytes)
n := copy(payload[len(tp.codecBytes):], extPkt.Packet.Payload[tp.incomingHeaderSize:])
if n != len(extPkt.Packet.Payload[tp.incomingHeaderSize:]) {
d.params.Logger.Errorw(
"payload overflow", errPayloadOverflow,
"want", len(extPkt.Packet.Payload[tp.incomingHeaderSize:]),
"have", n,
)
PacketFactory.Put(poolEntity)
return 0
}
payload = payload[:len(tp.codecBytes)+n]
if d.params.StripPacketTrailer {
if strip := packettrailer.StripTrailer(payload, tp.marker); strip > 0 {
payload = payload[:len(payload)-strip]
}
}
// translate RTP header
hdr := RTPHeaderFactory.Get().(*rtp.Header)
*hdr = rtp.Header{
Version: extPkt.Packet.Version,
Padding: extPkt.Packet.Padding,
Marker: tp.marker,
PayloadType: d.getTranslatedPayloadType(extPkt.Packet.PayloadType),
SequenceNumber: uint16(tp.rtp.extSequenceNumber),
Timestamp: uint32(tp.rtp.extTimestamp),
SSRC: d.ssrc,
}
// add extensions
if d.dependencyDescriptorExtID != 0 && tp.ddBytes != nil {
hdr.SetExtension(uint8(d.dependencyDescriptorExtID), tp.ddBytes)
}
if d.playoutDelayExtID != 0 && d.playoutDelay != nil {
if val := d.playoutDelay.GetDelayExtension(hdr.SequenceNumber); val != nil {
hdr.SetExtension(uint8(d.playoutDelayExtID), val)
// NOTE: play out delay extension is not cached in sequencer,
// i. e. they will not be added to retransmitted packet.
// But, it is okay as the extension is added till a RTCP Receiver Report for
// the corresponding sequence number is received.
// The extreme case is all packets containing the play out delay are lost and
// all of them retransmitted and an RTCP Receiver Report received for those
// retransmitted sequence numbers. But, that is highly improbable, if not impossible.
}
}
var actBytes []byte
if extPkt.AbsCaptureTimeExt != nil && d.absCaptureTimeExtID != 0 {
// normalize capture time to SFU clock.
// NOTE: even if there is estimated offset populated, just re-map the
// absolute capture time stamp as it should be the same RTCP sender report
// clock domain of publisher. SFU is normalising sender reports of publisher
// to SFU clock before sending to subscribers. So, capture time should be
// normalized to the same clock. Clear out any offset.
_, _, _, refSenderReport := d.forwarder.GetSenderReportParams()
if refSenderReport != nil {
actExtCopy := *extPkt.AbsCaptureTimeExt
if err = actExtCopy.Rewrite(
rtpstats.RTCPSenderReportPropagationDelay(
refSenderReport,
!d.params.DisableSenderReportPassThrough,
),
); err == nil {
actBytes, err = actExtCopy.Marshal()
if err == nil {
hdr.SetExtension(uint8(d.absCaptureTimeExtID), actBytes)
}
}
}
}
d.addDummyExtensions(hdr)
if d.sequencer != nil {
d.sequencer.push(
extPkt.Arrival,
extPkt.ExtSequenceNumber,
tp.rtp.extSequenceNumber,
tp.rtp.extTimestamp,
hdr.Marker,
int8(layer),
payload[:len(tp.codecBytes)],
tp.incomingHeaderSize,
tp.ddBytes,
actBytes,
)
}
headerSize := hdr.MarshalSize()
d.rtpStats.Update(
extPkt.Arrival,
tp.rtp.extSequenceNumber,
tp.rtp.extTimestamp,
hdr.Marker,
headerSize,
len(payload),
0,
extPkt.IsOutOfOrder,
)
pacerPacket := pacer.PacketFactory.Get().(*pacer.Packet)
*pacerPacket = pacer.Packet{
Header: hdr,
HeaderPool: RTPHeaderFactory,
HeaderSize: headerSize,
Payload: payload,
ProbeClusterId: ccutils.ProbeClusterId(d.probeClusterId.Load()),
AbsSendTimeExtID: uint8(d.absSendTimeExtID),
TransportWideExtID: uint8(d.transportWideExtID),
WriteStream: d.writeStream,
Pool: PacketFactory,
PoolEntity: poolEntity,
}
d.pacer.Enqueue(pacerPacket)
if extPkt.IsKeyFrame {
d.isNACKThrottled.Store(false)
d.rtpStats.UpdateKeyFrame(1)
d.params.Logger.Debugw(
"forwarded key frame",
"layer", layer,
"rtpsn", tp.rtp.extSequenceNumber,
"rtpts", tp.rtp.extTimestamp,
)
}
if tp.isSwitching {
d.postMaxLayerNotifierEvent("switching")
}
if tp.isResuming {
if sal := d.getStreamAllocatorListener(); sal != nil {
sal.OnResume(d)
}
}
if tp.isStarting {
d.params.Listener.OnStreamStarted()
}
return 1
}
// WritePaddingRTP tries to write as many padding only RTP packets as necessary
// to satisfy given size to the DownTrack
func (d *DownTrack) WritePaddingRTP(bytesToSend int, paddingOnMute bool, forceMarker bool) int {
if !d.writable.Load() {
return 0
}
if !paddingOnMute {
if !d.rtpStats.IsActive() {
return 0
}
// Ideally should look at header extensions negotiated for
// track and decide if padding can be sent. But, browsers behave
// in unexpected ways when using audio for bandwidth estimation and
// padding is mainly used to probe for excess available bandwidth.
// So, to be safe, limit to video tracks
if d.kind == webrtc.RTPCodecTypeAudio {
return 0
}
// LK-TODO-START
// Potentially write padding even if muted. Given that padding
// can be sent only on frame boundaries, writing on disabled tracks
// will give more options.
// LK-TODO-END
if d.forwarder.IsMuted() {
return 0
}
// Hold sending padding packets till first RTCP-RR is received for this RTP stream.
// That is definitive proof that the remote side knows about this RTP stream.
if d.rtpStats.LastReceiverReportTime() == 0 {
return 0
}
}
// RTP padding maximum is 255 bytes. Break it up.
// Use 20 byte as estimate of RTP header size (12 byte header + 8 byte extension)
num := (bytesToSend + RTPPaddingMaxPayloadSize + RTPPaddingEstimatedHeaderSize - 1) / (RTPPaddingMaxPayloadSize + RTPPaddingEstimatedHeaderSize)
if num == 0 {
return 0
}
frameRate := uint32(0)
if paddingOnMute {
// advance timestamps when sending dummy padding packets to start a stream
// to ensure receiver sees proper timestamp and starts the stream
frameRate = uint32(time.Second / paddingOnMuteInterval)
}
snts, err := d.forwarder.GetSnTsForPadding(num, frameRate, forceMarker)
if err != nil {
return 0
}
//
// Register with sequencer as padding only so that NACKs for these can be filtered out.
// Retransmission is probably a sign of network congestion/badness.
// So, retransmitting padding only packets is only going to make matters worse.
//
if d.sequencer != nil {
d.sequencer.pushPadding(snts[0].extSequenceNumber, snts[len(snts)-1].extSequenceNumber)
}
bytesSent := 0
payloads := make([]byte, RTPPaddingMaxPayloadSize*len(snts))
for i := range snts {
hdr := RTPHeaderFactory.Get().(*rtp.Header)
*hdr = rtp.Header{
Version: 2,
Padding: true,
Marker: false,
PayloadType: uint8(d.payloadType.Load()),
SequenceNumber: uint16(snts[i].extSequenceNumber),
Timestamp: uint32(snts[i].extTimestamp),
SSRC: d.ssrc,
}
d.addDummyExtensions(hdr)
payload := payloads[i*RTPPaddingMaxPayloadSize : (i+1)*RTPPaddingMaxPayloadSize : (i+1)*RTPPaddingMaxPayloadSize]
// last byte of padding has padding size including that byte
payload[RTPPaddingMaxPayloadSize-1] = byte(RTPPaddingMaxPayloadSize)
hdrSize := hdr.MarshalSize()
payloadSize := len(payload)
d.rtpStats.Update(
mono.UnixNano(),
snts[i].extSequenceNumber,
snts[i].extTimestamp,
hdr.Marker,
hdrSize,
0,
payloadSize,
false,
)
pacerPacket := pacer.PacketFactory.Get().(*pacer.Packet)
*pacerPacket = pacer.Packet{
Header: hdr,
HeaderPool: RTPHeaderFactory,
HeaderSize: hdrSize,
Payload: payload,
ProbeClusterId: ccutils.ProbeClusterId(d.probeClusterId.Load()),
IsProbe: true,
AbsSendTimeExtID: uint8(d.absSendTimeExtID),
TransportWideExtID: uint8(d.transportWideExtID),
WriteStream: d.writeStream,
}
d.pacer.Enqueue(pacerPacket)
bytesSent += hdrSize + payloadSize
}
return bytesSent
}
// Mute enables or disables media forwarding - subscriber triggered
func (d *DownTrack) Mute(muted bool) {
isSubscribeMutable := true
if sal := d.getStreamAllocatorListener(); sal != nil {
isSubscribeMutable = sal.IsSubscribeMutable(d)
}
changed := d.forwarder.Mute(muted, isSubscribeMutable)
d.handleMute(muted, changed)
}
// PubMute enables or disables media forwarding - publisher side
func (d *DownTrack) PubMute(pubMuted bool) {
changed := d.forwarder.PubMute(pubMuted)
d.handleMute(pubMuted, changed)
}
func (d *DownTrack) handleMute(muted bool, changed bool) {
if !changed {
return
}
d.connectionStats.UpdateMute(d.forwarder.IsAnyMuted())
//
// Subscriber mute changes trigger a max layer notification.
// That could result in encoding layers getting turned on/off on publisher side
// (depending on aggregate layer requirements of all subscribers of the track).
//
// Publisher mute changes should not trigger notification.
// If publisher turns off all layers because of subscribers indicating
// no layers required due to publisher mute (bit of circular dependency),
// there will be a delay in layers turning back on when unmute happens.
// Unmute path will require
// 1. unmute signalling out-of-band from publisher received by downtrack(s)
// 2. downtrack(s) notifying max layer
// 3. out-of-band notification about max layer sent back to the publisher
// 4. publisher starts layer(s)
// Ideally, on publisher mute, whatever layers were active remain active and
// can be restarted by publisher immediately on unmute.
//
// Note that while publisher mute is active, subscriber changes can also happen
// and that could turn on/off layers on publisher side.
//
d.postMaxLayerNotifierEvent("mute")
if sal := d.getStreamAllocatorListener(); sal != nil {
sal.OnSubscriptionChanged(d)
}
// when muting, send a few silence frames to ensure residual noise does not
// put the comfort noise generator on decoder side in a bad state where it
// generates noise that is not so comfortable.
//
// One possibility is not to inject blank frames when publisher is muted
// and let forwarding continue. When publisher is muted, unless the media
// stream is stopped, publisher will send silence frames which should have
// comfort noise information. But, in case the publisher stops at an
// inopportune frame (due to media stream stop or injecting audio from a file),
// the decoder could be in a noisy state. So, inject blank frames on publisher
// mute too.
d.blankFramesGeneration.Inc()
if d.kind == webrtc.RTPCodecTypeAudio && muted {
d.writeBlankFrameRTP(RTPBlankFramesMuteSeconds, d.blankFramesGeneration.Load())
}
}
func (d *DownTrack) IsClosed() bool {
return d.isClosed.Load()
}
func (d *DownTrack) Close() {
d.CloseWithFlush(true, true)
}
// CloseWithFlush - `flush` used to indicate whether send blank frame to flush
// decoder of client.
// 1. When transceiver of this track is reused by some other participant's video track,
// set flush=true to avoid previous video shows before new stream is displayed.
// 2. in case of session migration, participant migrate from other node, video track should
// be resumed with same participant, set flush=false since we don't need to flush decoder.
func (d *DownTrack) CloseWithFlush(flush bool, isEnding bool) {
d.bindLock.Lock()
if d.isClosed.Swap(true) {
// already closed
d.bindLock.Unlock()
return
}
d.params.Logger.Debugw("close downtrack", "flushBlankFrame", flush, "isEnding", isEnding)
if d.bindState.Load() == bindStateBound {
d.forwarder.Mute(true, true)
// write blank frames after disabling so that other frames do not interfere.
// Idea here is to send blank key frames to flush the decoder buffer at the remote end.
// Otherwise, with transceiver re-use last frame from previous stream is held in the
// display buffer and there could be a brief moment where the previous stream is displayed.
if flush {
doneFlushing := d.writeBlankFrameRTP(RTPBlankFramesCloseSeconds, d.blankFramesGeneration.Inc())
// wait a limited time to flush
timer := time.NewTimer(flushTimeout)
defer timer.Stop()
select {
case <-doneFlushing:
case <-timer.C:
d.blankFramesGeneration.Inc() // in case flush is still running
}
}
d.params.Logger.Debugw("closing sender", "kind", d.kind)
}
d.setBindStateLocked(bindStateUnbound)
d.Receiver().DeleteDownTrack(d.SubscriberID())
if d.rtcpReader != nil && isEnding {
d.params.Logger.Debugw("downtrack close rtcp reader")
d.rtcpReader.Close()
d.rtcpReader.OnPacket(nil)
}
if d.rtcpReaderRTX != nil && isEnding {
d.params.Logger.Debugw("downtrack close rtcp rtx reader")
d.rtcpReaderRTX.Close()
d.rtcpReaderRTX.OnPacket(nil)
}
d.bindLock.Unlock()
d.connectionStats.Close()
d.rtpStats.Stop()
d.rtpStatsRTX.Stop()
d.params.Logger.Debugw(
"rtp stats",
"direction", "downstream",
"mime", d.Mime().String(),
"ssrc", d.ssrc,
"stats", d.rtpStats,
"statsRTX", d.rtpStatsRTX,
)
d.maxLayerNotifierChMu.Lock()
d.maxLayerNotifierChClosed = true
close(d.maxLayerNotifierCh)
d.maxLayerNotifierChMu.Unlock()
d.keyFrameRequesterChMu.Lock()
d.keyFrameRequesterChClosed = true
close(d.keyFrameRequesterCh)
d.keyFrameRequesterChMu.Unlock()
d.params.Listener.OnDownTrackClose(!flush)
}
func (d *DownTrack) SetMaxSpatialLayer(spatialLayer int32) {
changed, maxLayer := d.forwarder.SetMaxSpatialLayer(spatialLayer)
if !changed {
return
}
d.postMaxLayerNotifierEvent("max-subscribed")
d.postKeyFrameRequestEvent()
if sal := d.getStreamAllocatorListener(); sal != nil {
sal.OnSubscribedLayerChanged(d, maxLayer)
}
}
func (d *DownTrack) SetMaxTemporalLayer(temporalLayer int32) {
changed, maxLayer := d.forwarder.SetMaxTemporalLayer(temporalLayer)
if !changed {
return
}
if sal := d.getStreamAllocatorListener(); sal != nil {
sal.OnSubscribedLayerChanged(d, maxLayer)
}
}
func (d *DownTrack) MaxLayer() buffer.VideoLayer {
return d.forwarder.MaxLayer()
}
func (d *DownTrack) GetState() DownTrackState {
dts := DownTrackState{
RTPStats: d.rtpStats,
DeltaStatsSenderSnapshotId: d.deltaStatsSenderSnapshotId,
RTPStatsRTX: d.rtpStatsRTX,
DeltaStatsRTXSenderSnapshotId: d.deltaStatsRTXSenderSnapshotId,
ForwarderState: d.forwarder.GetState(),
}
if d.playoutDelay != nil {
dts.PlayoutDelayControllerState = d.playoutDelay.GetState()
}
return dts
}
func (d *DownTrack) SeedState(state DownTrackState) {
if d.writable.Load() {
return
}
if state.RTPStats != nil || state.ForwarderState != nil {
d.params.Logger.Debugw("seeding downtrack state", "state", state)
}
if state.RTPStats != nil {
d.rtpStats.Seed(state.RTPStats)
d.deltaStatsSenderSnapshotId = state.DeltaStatsSenderSnapshotId
if d.playoutDelay != nil {
d.playoutDelay.SeedState(state.PlayoutDelayControllerState)
}
}
if state.RTPStatsRTX != nil {
d.rtpStatsRTX.Seed(state.RTPStatsRTX)
d.deltaStatsRTXSenderSnapshotId = state.DeltaStatsRTXSenderSnapshotId
d.rtxSequenceNumber.Store(d.rtpStatsRTX.ExtHighestSequenceNumber())
}
d.forwarder.SeedState(state.ForwarderState)
}
func (d *DownTrack) StopWriteAndGetState() DownTrackState {
d.params.Logger.Debugw("stopping write")
d.bindLock.Lock()
d.writable.Store(false)
d.writeStopped.Store(true)
d.bindLock.Unlock()
return d.GetState()
}
func (d *DownTrack) UpTrackLayersChange() {
if sal := d.getStreamAllocatorListener(); sal != nil {
sal.OnAvailableLayersChanged(d)
}
}
func (d *DownTrack) UpTrackBitrateAvailabilityChange() {
if sal := d.getStreamAllocatorListener(); sal != nil {
sal.OnBitrateAvailabilityChanged(d)
}
}
func (d *DownTrack) UpTrackMaxPublishedLayerChange(maxPublishedLayer int32) {
if d.forwarder.SetMaxPublishedLayer(maxPublishedLayer) {
if sal := d.getStreamAllocatorListener(); sal != nil {
sal.OnMaxPublishedSpatialChanged(d)
}
}
}
func (d *DownTrack) UpTrackMaxTemporalLayerSeenChange(maxTemporalLayerSeen int32) {
if d.forwarder.SetMaxTemporalLayerSeen(maxTemporalLayerSeen) {
if sal := d.getStreamAllocatorListener(); sal != nil {
sal.OnMaxPublishedTemporalChanged(d)
}
}
}
func (d *DownTrack) maybeAddTransition(bitrate int64, distance float64, pauseReason VideoPauseReason) {
if d.kind == webrtc.RTPCodecTypeAudio {
return
}
if pauseReason == VideoPauseReasonBandwidth {
d.connectionStats.UpdatePause(true)
} else {
d.connectionStats.UpdatePause(false)
d.connectionStats.AddLayerTransition(distance)
d.connectionStats.AddBitrateTransition(bitrate)
}
}
func (d *DownTrack) UpTrackBitrateReport(availableLayers []int32, bitrates Bitrates) {
d.maybeAddTransition(
d.forwarder.GetOptimalBandwidthNeeded(bitrates),
d.forwarder.DistanceToDesired(availableLayers, bitrates),
d.forwarder.PauseReason(),
)
}
func (d *DownTrack) OnBinding(fn func(error)) {
d.bindLock.Lock()
defer d.bindLock.Unlock()
d.onBinding = fn
}
func (d *DownTrack) AddReceiverReportListener(listener ReceiverReportListener) {
d.listenerLock.Lock()
defer d.listenerLock.Unlock()
d.receiverReportListeners = append(d.receiverReportListeners, listener)
}
func (d *DownTrack) IsDeficient() bool {
return d.forwarder.IsDeficient()
}
func (d *DownTrack) BandwidthRequested() int64 {
_, brs := d.Receiver().GetLayeredBitrate()
return d.forwarder.BandwidthRequested(brs)
}
func (d *DownTrack) DistanceToDesired() float64 {
al, brs := d.Receiver().GetLayeredBitrate()
return d.forwarder.DistanceToDesired(al, brs)
}
func (d *DownTrack) AllocateOptimal(allowOvershoot bool, hold bool) VideoAllocation {
al, brs := d.Receiver().GetLayeredBitrate()
allocation := d.forwarder.AllocateOptimal(al, brs, allowOvershoot, hold)
d.postKeyFrameRequestEvent()
d.maybeAddTransition(allocation.BandwidthNeeded, allocation.DistanceToDesired, allocation.PauseReason)
return allocation
}
func (d *DownTrack) ProvisionalAllocatePrepare() {
al, brs := d.Receiver().GetLayeredBitrate()
d.forwarder.ProvisionalAllocatePrepare(al, brs)
}
func (d *DownTrack) ProvisionalAllocateReset() {
d.forwarder.ProvisionalAllocateReset()
}
func (d *DownTrack) ProvisionalAllocate(availableChannelCapacity int64, layers buffer.VideoLayer, allowPause bool, allowOvershoot bool) (bool, int64) {
return d.forwarder.ProvisionalAllocate(availableChannelCapacity, layers, allowPause, allowOvershoot)
}
func (d *DownTrack) ProvisionalAllocateGetCooperativeTransition(allowOvershoot bool) VideoTransition {
transition, availableLayers, brs := d.forwarder.ProvisionalAllocateGetCooperativeTransition(allowOvershoot)
d.params.Logger.Debugw(
"stream: cooperative transition",
"transition", &transition,
"availableLayers", availableLayers,
"bitrates", brs,
)
return transition
}
func (d *DownTrack) ProvisionalAllocateGetBestWeightedTransition() VideoTransition {
transition, availableLayers, brs := d.forwarder.ProvisionalAllocateGetBestWeightedTransition()
d.params.Logger.Debugw(
"stream: best weighted transition",
"transition", &transition,
"availableLayers", availableLayers,
"bitrates", brs,
)
return transition
}
func (d *DownTrack) ProvisionalAllocateCommit() VideoAllocation {
allocation := d.forwarder.ProvisionalAllocateCommit()
d.postKeyFrameRequestEvent()
d.maybeAddTransition(allocation.BandwidthNeeded, allocation.DistanceToDesired, allocation.PauseReason)
return allocation
}
func (d *DownTrack) AllocateNextHigher(availableChannelCapacity int64, allowOvershoot bool) (VideoAllocation, bool) {
al, brs := d.Receiver().GetLayeredBitrate()
allocation, available := d.forwarder.AllocateNextHigher(availableChannelCapacity, al, brs, allowOvershoot)
d.postKeyFrameRequestEvent()
d.maybeAddTransition(allocation.BandwidthNeeded, allocation.DistanceToDesired, allocation.PauseReason)
return allocation, available
}
func (d *DownTrack) GetNextHigherTransition(allowOvershoot bool) (VideoTransition, bool) {
availableLayers, brs := d.Receiver().GetLayeredBitrate()
transition, available := d.forwarder.GetNextHigherTransition(brs, allowOvershoot)
d.params.Logger.Debugw(
"stream: get next higher layer",
"transition", transition,
"available", available,
"availableLayers", availableLayers,
"bitrates", brs,
)
return transition, available
}
func (d *DownTrack) Pause() VideoAllocation {
al, brs := d.Receiver().GetLayeredBitrate()
allocation := d.forwarder.Pause(al, brs)
d.maybeAddTransition(allocation.BandwidthNeeded, allocation.DistanceToDesired, allocation.PauseReason)
return allocation
}
func (d *DownTrack) Resync() {
d.forwarder.Resync()
}
func (d *DownTrack) ReceiverRestart(rcvr TrackReceiver) {
if rcvr.Mime() != d.Receiver().Mime() {
d.params.Logger.Infow("upstream receiver restart - skipped", "mime", d.Receiver().Mime().String(), "newMime", rcvr.Mime().String())
return
}
d.bindLock.Lock()
codec := d.codec.Load().(webrtc.RTPCodecCapability)
d.bindLock.Unlock()
receiver := d.Receiver()
d.params.Logger.Infow("upstream receiver restart", "mime", receiver.Mime().String())
d.forwarder.Restart()
d.forwarder.DetermineCodec(codec, receiver.HeaderExtensions(), receiver.VideoLayerMode())
}
func (d *DownTrack) CreateSourceDescriptionChunks() []rtcp.SourceDescriptionChunk {
transceiver := d.transceiver.Load()
if d.bindState.Load() != bindStateBound || transceiver == nil {
return nil
}
return []rtcp.SourceDescriptionChunk{
{
Source: d.ssrc,
Items: []rtcp.SourceDescriptionItem{
{
Type: rtcp.SDESCNAME,
Text: d.params.StreamID,
},
{
Type: rtcp.SDESType(15),
Text: transceiver.Mid(),
},
},
},
}
}
func (d *DownTrack) CreateSenderReport() *rtcp.SenderReport {
if d.bindState.Load() != bindStateBound {
return nil
}
_, _, tsOffset, refSenderReport := d.forwarder.GetSenderReportParams()
return d.rtpStats.GetRtcpSenderReport(d.ssrc, refSenderReport, tsOffset, !d.params.DisableSenderReportPassThrough)
// not sending RTCP Sender Report for RTX
}
func (d *DownTrack) writeBlankFrameRTP(duration float32, generation uint32) chan struct{} {
done := make(chan struct{})
go func() {
// don't send if not writable OR nothing has been sent
if !d.writable.Load() || !d.rtpStats.IsActive() {
close(done)
return
}
mimeType := d.Mime()
var getBlankFrame func(bool) ([]byte, error)
switch mimeType {
case mime.MimeTypeOpus:
getBlankFrame = d.getAudioBlankFrameFunc(OpusSilenceFrame)
case mime.MimeTypeRED:
getBlankFrame = d.getOpusRedBlankFrame
case mime.MimeTypePCMU:
getBlankFrame = d.getAudioBlankFrameFunc(PCMUSilenceFrame)
case mime.MimeTypePCMA:
getBlankFrame = d.getAudioBlankFrameFunc(PCMASilenceFrame)
case mime.MimeTypeVP8:
getBlankFrame = d.getVP8BlankFrame
case mime.MimeTypeH264:
getBlankFrame = d.getH264BlankFrame
default:
close(done)
return
}
frameRate := uint32(30)
if mimeType == mime.MimeTypeOpus || mimeType == mime.MimeTypeRED ||
mimeType == mime.MimeTypePCMU || mimeType == mime.MimeTypePCMA {
frameRate = 50
}
// send a number of blank frames just in case there is loss.
// Intentionally ignoring check for mute or bandwidth constrained mute
// as this is used to clear client side buffer.
numFrames := int(float32(frameRate) * duration)
frameDuration := time.Duration(1000/frameRate) * time.Millisecond
ticker := time.NewTicker(frameDuration)
defer ticker.Stop()
for {
if generation != d.blankFramesGeneration.Load() || numFrames <= 0 || !d.writable.Load() || !d.rtpStats.IsActive() {
close(done)
return
}
snts, frameEndNeeded, err := d.forwarder.GetSnTsForBlankFrames(frameRate, 1)
if err != nil {
d.params.Logger.Warnw("could not get SN/TS for blank frame", err)
close(done)
return
}
for i := range snts {
hdr := &rtp.Header{
Version: 2,
Padding: false,
Marker: true,
PayloadType: uint8(d.payloadType.Load()),
SequenceNumber: uint16(snts[i].extSequenceNumber),
Timestamp: uint32(snts[i].extTimestamp),
SSRC: d.ssrc,
}
d.addDummyExtensions(hdr)
payload, err := getBlankFrame(frameEndNeeded)
if err != nil {
d.params.Logger.Warnw("could not get blank frame", err)
close(done)
return
}
headerSize := hdr.MarshalSize()
d.rtpStats.Update(
mono.UnixNano(),
snts[i].extSequenceNumber,
snts[i].extTimestamp,
hdr.Marker,
headerSize,
len(payload),
0,
false,
)
pacerPacket := pacer.PacketFactory.Get().(*pacer.Packet)
*pacerPacket = pacer.Packet{
Header: hdr,
HeaderSize: headerSize,
Payload: payload,
ProbeClusterId: ccutils.ProbeClusterId(d.probeClusterId.Load()),
AbsSendTimeExtID: uint8(d.absSendTimeExtID),
TransportWideExtID: uint8(d.transportWideExtID),
WriteStream: d.writeStream,
}
d.pacer.Enqueue(pacerPacket)
// only the first frame will need frameEndNeeded to close out the
// previous picture, rest are small key frames (for the video case)
frameEndNeeded = false
}
numFrames--
<-ticker.C
}
}()
return done
}
func (d *DownTrack) maybeAddTrailer(buf []byte) int {
if len(buf) < len(d.params.Trailer) {
d.params.Logger.Warnw("trailer too big", nil, "bufLen", len(buf), "trailerLen", len(d.params.Trailer))
return 0
}
copy(buf, d.params.Trailer)
return len(d.params.Trailer)
}
func (d *DownTrack) getAudioBlankFrameFunc(silentPayload []byte) func(_frameEndNeeded bool) ([]byte, error) {
// silence frame
// Used shortly after muting to ensure residual noise does not keep
// generating noise at the decoder after the stream is stopped
// i. e. comfort noise generation actually not producing something comfortable.
return func(_frameEndNeeded bool) ([]byte, error) {
payload := make([]byte, 1000)
copy(payload[0:], silentPayload)
trailerLen := d.maybeAddTrailer(payload[len(silentPayload):])
return payload[:len(silentPayload)+trailerLen], nil
}
}
func (d *DownTrack) getOpusRedBlankFrame(_frameEndNeeded bool) ([]byte, error) {
// primary only silence frame for opus/red, there is no need to contain redundant silent frames
payload := make([]byte, 1000)
// primary header
// 0 1 2 3 4 5 6 7
// +-+-+-+-+-+-+-+-+
// |0| Block PT |
// +-+-+-+-+-+-+-+-+
payload[0] = opusPT
copy(payload[1:], OpusSilenceFrame)
trailerLen := d.maybeAddTrailer(payload[1+len(OpusSilenceFrame):])
return payload[:1+len(OpusSilenceFrame)+trailerLen], nil
}
func (d *DownTrack) getVP8BlankFrame(frameEndNeeded bool) ([]byte, error) {
// 8x8 key frame
// Used even when closing out a previous frame. Looks like receivers
// do not care about content (it will probably end up being an undecodable
// frame, but that should be okay as there are key frames following)
header, err := d.forwarder.GetPadding(frameEndNeeded)
if err != nil {
return nil, err
}
payload := make([]byte, 1000)
copy(payload, header)
copy(payload[len(header):], VP8KeyFrame8x8)
trailerLen := d.maybeAddTrailer(payload[len(header)+len(VP8KeyFrame8x8):])
return payload[:len(header)+len(VP8KeyFrame8x8)+trailerLen], nil
}
func (d *DownTrack) getH264BlankFrame(_frameEndNeeded bool) ([]byte, error) {
// TODO - Jie Zeng
// now use STAP-A to compose sps, pps, idr together, most decoder support packetization-mode 1.
// if client only support packetization-mode 0, use single nalu unit packet
buf := make([]byte, 1000)
offset := 0
buf[0] = 0x18 // STAP-A
offset++
for _, payload := range H264KeyFrame2x2 {
binary.BigEndian.PutUint16(buf[offset:], uint16(len(payload)))
offset += 2
copy(buf[offset:offset+len(payload)], payload)
offset += len(payload)
}
offset += d.maybeAddTrailer(buf[offset:])
return buf[:offset], nil
}
func (d *DownTrack) handleRTCP(bytes []byte) {
pkts, err := rtcp.Unmarshal(bytes)
if err != nil {
d.params.Logger.Errorw("could not unmarshal rtcp receiver packet", err)
return
}
pliOnce := true
sendPliOnce := func() {
_, layer := d.forwarder.CheckSync()
if pliOnce {
if layer != buffer.InvalidLayerSpatial {
d.params.Logger.Debugw("sending PLI RTCP", "layer", layer)
d.Receiver().SendPLI(layer, false)
d.isNACKThrottled.Store(true)
d.rtpStats.UpdatePliTime()
pliOnce = false
}
}
}
rttToReport := uint32(0)
var numNACKs uint32
var numPLIs uint32
var numFIRs uint32
for _, pkt := range pkts {
switch p := pkt.(type) {
case *rtcp.PictureLossIndication:
if p.MediaSSRC == d.ssrc {
numPLIs++
sendPliOnce()
}
case *rtcp.FullIntraRequest:
if p.MediaSSRC == d.ssrc {
numFIRs++
sendPliOnce()
}
case *rtcp.ReceiverEstimatedMaximumBitrate:
if sal := d.getStreamAllocatorListener(); sal != nil {
sal.OnREMB(d, p)
}
case *rtcp.ReceiverReport:
// create new receiver report w/ only valid reception reports
rr := &rtcp.ReceiverReport{
SSRC: p.SSRC,
ProfileExtensions: p.ProfileExtensions,
}
for _, r := range p.Reports {
if r.SSRC != d.ssrc {
continue
}
rr.Reports = append(rr.Reports, r)
rtt, isRttChanged := d.rtpStats.UpdateFromReceiverReport(r)
if isRttChanged {
rttToReport = rtt
}
if d.playoutDelay != nil {
d.playoutDelay.OnSeqAcked(uint16(r.LastSequenceNumber))
// screen share track has inaccuracy jitter due to its low frame rate and bursty traffic
if d.params.Source != livekit.TrackSource_SCREEN_SHARE {
jitterMs := uint64(r.Jitter*1e3) / uint64(d.clockRate)
d.playoutDelay.SetJitter(uint32(jitterMs))
}
}
}
// RTX-TODO: This is used for media loss proxying only as of 2024-12-15.
// Ideally, this should keep deltas between previous RTCP Receiver Report
// and current report, calculate the loss in the window and reconcile it with
// data in a similar window from RTX stream (to ensure losses are discounted
// for NACKs), but keeping this simple for several reasons
// - media loss proxying is a configurable setting and could be disabled
// - media loss proxying is used for audio only and audio may not have NACKing
// - to keep it simple
if len(rr.Reports) > 0 {
d.listenerLock.RLock()
rrListeners := d.receiverReportListeners
d.listenerLock.RUnlock()
for _, l := range rrListeners {
l(d, rr)
}
}
case *rtcp.TransportLayerNack:
if p.MediaSSRC == d.ssrc {
var nacks []uint16
for _, pair := range p.Nacks {
packetList := pair.PacketList()
numNACKs += uint32(len(packetList))
nacks = append(nacks, packetList...)
}
go d.retransmitPackets(nacks)
}
case *rtcp.TransportLayerCC:
if p.MediaSSRC == d.ssrc {
if sal := d.getStreamAllocatorListener(); sal != nil {
sal.OnTransportCCFeedback(d, p)
}
}
case *rtcp.ExtendedReport:
// SFU only responds with the DLRRReport for the track has the sender SSRC, the behavior is different with
// browser's implementation, which includes all sent tracks. It is ok since all the tracks
// use the same connection, and server-sdk-go can get the rtt from the first DLRRReport
// (libwebrtc/browsers don't send XR to calculate rtt, it only responds)
var lastRR uint32
for _, report := range p.Reports {
if rr, ok := report.(*rtcp.ReceiverReferenceTimeReportBlock); ok {
lastRR = uint32(rr.NTPTimestamp >> 16)
break
}
}
if lastRR > 0 {
d.params.RTCPWriter([]rtcp.Packet{&rtcp.ExtendedReport{
SenderSSRC: d.ssrc,
Reports: []rtcp.ReportBlock{
&rtcp.DLRRReportBlock{
Reports: []rtcp.DLRRReport{{
SSRC: p.SenderSSRC,
LastRR: lastRR,
DLRR: 0, // no delay
}},
},
},
}})
}
}
}
d.rtpStats.UpdateNack(numNACKs)
d.rtpStats.UpdatePli(numPLIs)
d.rtpStats.UpdateFir(numFIRs)
if rttToReport != 0 {
if d.sequencer != nil {
d.sequencer.setRTT(rttToReport)
}
d.params.Listener.OnRttUpdate(rttToReport)
}
}
func (d *DownTrack) handleRTCPRTX(bytes []byte) {
pkts, err := rtcp.Unmarshal(bytes)
if err != nil {
d.params.Logger.Errorw("could not unmarshal rtcp rtx receiver packet", err)
return
}
for _, pkt := range pkts {
switch p := pkt.(type) {
case *rtcp.ReceiverReport:
for _, r := range p.Reports {
if r.SSRC != d.ssrcRTX {
continue
}
d.rtpStatsRTX.UpdateFromReceiverReport(r)
}
case *rtcp.ReceiverEstimatedMaximumBitrate:
if sal := d.getStreamAllocatorListener(); sal != nil {
sal.OnREMB(d, p)
}
case *rtcp.TransportLayerCC:
if p.MediaSSRC == d.ssrcRTX {
if sal := d.getStreamAllocatorListener(); sal != nil {
sal.OnTransportCCFeedback(d, p)
}
}
}
}
}
func (d *DownTrack) SetConnected() {
d.bindLock.Lock()
if !d.connected.Swap(true) {
d.onBindAndConnectedChange()
}
d.params.Logger.Debugw("downtrack connected")
d.bindLock.Unlock()
}
// SetActivePaddingOnMuteUpTrack will enable padding on the track when its uptrack is muted.
// Pion will not fire OnTrack event until it receives packet for the track,
// so we send padding packets to help pion client (go-sdk) to fire the event.
func (d *DownTrack) SetActivePaddingOnMuteUpTrack() {
d.activePaddingOnMuteUpTrack.Store(true)
}
func (d *DownTrack) retransmitPacket(epm *extPacketMeta, sourcePkt []byte, isProbe bool) (int, error) {
var pkt rtp.Packet
if err := pkt.Unmarshal(sourcePkt); err != nil {
d.params.Logger.Errorw("could not unmarshal rtp packet to send via RTX", err)
return 0, err
}
hdr := RTPHeaderFactory.Get().(*rtp.Header)
*hdr = rtp.Header{
Version: pkt.Header.Version,
Padding: pkt.Header.Padding,
Marker: epm.marker,
PayloadType: d.getTranslatedPayloadType(pkt.Header.PayloadType),
SequenceNumber: epm.targetSeqNo,
Timestamp: epm.timestamp,
SSRC: d.ssrc,
}
rtxOffset := 0
var rtxExtSequenceNumber uint64
if rtxPT := d.payloadTypeRTX.Load(); rtxPT != 0 && d.ssrcRTX != 0 {
rtxExtSequenceNumber = d.rtxSequenceNumber.Inc()
rtxOffset = 2
hdr.PayloadType = uint8(rtxPT)
hdr.SequenceNumber = uint16(rtxExtSequenceNumber)
hdr.SSRC = d.ssrcRTX
}
if d.dependencyDescriptorExtID != 0 {
var ddBytes []byte
if len(epm.ddBytesSlice) != 0 {
ddBytes = epm.ddBytesSlice
} else {
ddBytes = epm.ddBytes[:epm.ddBytesSize]
}
if len(ddBytes) != 0 {
hdr.SetExtension(uint8(d.dependencyDescriptorExtID), ddBytes)
}
}
if d.absCaptureTimeExtID != 0 && len(epm.actBytes) != 0 {
hdr.SetExtension(uint8(d.absCaptureTimeExtID), epm.actBytes)
}
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)]
}
if d.params.StripPacketTrailer {
if strip := packettrailer.StripTrailer(payload[rtxOffset:], epm.marker); strip > 0 {
payload = payload[:len(payload)-strip]
}
}
headerSize := hdr.MarshalSize()
var (
payloadSize, paddingSize int
isOutOfOrder bool
)
if isProbe {
// although not padding only packets, marking it as padding for accounting as padding is used to signify probing,
// also not marking them as out-of-order although sequence numbers in packets are out-of-order because of re-sending packets
payloadSize, paddingSize, isOutOfOrder = 0, len(payload), false
} else {
payloadSize, paddingSize, isOutOfOrder = len(payload), 0, true
}
if hdr.SSRC == d.ssrcRTX {
d.rtpStatsRTX.Update(
mono.UnixNano(),
rtxExtSequenceNumber,
0,
hdr.Marker,
headerSize,
payloadSize,
paddingSize,
isOutOfOrder,
)
} else {
d.rtpStats.Update(
mono.UnixNano(),
epm.extSequenceNumber,
epm.extTimestamp,
hdr.Marker,
headerSize,
payloadSize,
paddingSize,
isOutOfOrder,
)
}
pacerPacket := pacer.PacketFactory.Get().(*pacer.Packet)
*pacerPacket = pacer.Packet{
Header: hdr,
HeaderPool: RTPHeaderFactory,
HeaderSize: headerSize,
Payload: payload,
ProbeClusterId: ccutils.ProbeClusterId(d.probeClusterId.Load()),
IsProbe: isProbe,
IsRTX: !isProbe,
AbsSendTimeExtID: uint8(d.absSendTimeExtID),
TransportWideExtID: uint8(d.transportWideExtID),
WriteStream: d.writeStream,
Pool: PacketFactory,
PoolEntity: poolEntity,
}
d.pacer.Enqueue(pacerPacket)
return headerSize + len(payload), nil
}
func (d *DownTrack) retransmitPackets(nacks []uint16) {
if d.sequencer == nil {
return
}
if FlagStopRTXOnPLI && d.isNACKThrottled.Load() {
return
}
filtered, disallowedLayers := d.forwarder.FilterRTX(nacks)
if len(filtered) == 0 {
return
}
src := PacketFactory.Get().(*[]byte)
defer PacketFactory.Put(src)
receiver := d.Receiver()
nackAcks := uint32(0)
nackMisses := uint32(0)
numRepeatedNACKs := uint32(0)
for _, epm := range d.sequencer.getExtPacketMetas(filtered) {
if disallowedLayers[epm.layer] {
continue
}
nackAcks++
pktBuff := *src
n, err := receiver.ReadRTP(pktBuff, uint8(epm.layer), epm.sourceSeqNo)
if err != nil {
if err == io.EOF {
break
}
nackMisses++
continue
}
if epm.nacked > 1 {
numRepeatedNACKs++
}
d.retransmitPacket(&epm, pktBuff[:n], false)
}
d.totalRepeatedNACKs.Add(numRepeatedNACKs)
d.rtpStats.UpdateNackProcessed(nackAcks, nackMisses, numRepeatedNACKs)
}
func (d *DownTrack) WriteProbePackets(bytesToSend int, usePadding bool) int {
rtxPT := uint8(d.payloadTypeRTX.Load())
if rtxPT == 0 || d.ssrcRTX == 0 {
return d.WritePaddingRTP(bytesToSend, false, false)
}
if !d.writable.Load() ||
!d.rtpStats.IsActive() ||
(d.absSendTimeExtID == 0 && d.transportWideExtID == 0) ||
d.rtpStats.LastReceiverReportTime() == 0 ||
d.sequencer == nil {
return 0
}
bytesSent := 0
if usePadding {
num := (bytesToSend + RTPPaddingMaxPayloadSize + RTPPaddingEstimatedHeaderSize - 1) / (RTPPaddingMaxPayloadSize + RTPPaddingEstimatedHeaderSize)
if num == 0 {
return 0
}
payloads := make([]byte, RTPPaddingMaxPayloadSize*num)
for i := range num {
rtxExtSequenceNumber := d.rtxSequenceNumber.Inc()
hdr := RTPHeaderFactory.Get().(*rtp.Header)
*hdr = rtp.Header{
Version: 2,
Padding: true,
Marker: false,
PayloadType: rtxPT,
SequenceNumber: uint16(rtxExtSequenceNumber),
Timestamp: 0,
SSRC: d.ssrcRTX,
}
d.addDummyExtensions(hdr)
payload := payloads[i*RTPPaddingMaxPayloadSize : (i+1)*RTPPaddingMaxPayloadSize : (i+1)*RTPPaddingMaxPayloadSize]
// last byte of padding has padding size including that byte
payload[RTPPaddingMaxPayloadSize-1] = byte(RTPPaddingMaxPayloadSize)
hdrSize := hdr.MarshalSize()
payloadSize := len(payload)
d.rtpStatsRTX.Update(
mono.UnixNano(),
rtxExtSequenceNumber,
0,
hdr.Marker,
hdrSize,
0,
payloadSize,
false,
)
pacerPacket := pacer.PacketFactory.Get().(*pacer.Packet)
*pacerPacket = pacer.Packet{
Header: hdr,
HeaderPool: RTPHeaderFactory,
HeaderSize: hdrSize,
Payload: payload,
ProbeClusterId: ccutils.ProbeClusterId(d.probeClusterId.Load()),
IsProbe: true,
AbsSendTimeExtID: uint8(d.absSendTimeExtID),
TransportWideExtID: uint8(d.transportWideExtID),
WriteStream: d.writeStream,
}
d.pacer.Enqueue(pacerPacket)
bytesSent += hdrSize + payloadSize
}
} else {
src := PacketFactory.Get().(*[]byte)
defer PacketFactory.Put(src)
receiver := d.Receiver()
endExtHighestSequenceNumber := d.rtpStats.ExtHighestSequenceNumber()
startExtHighestSequenceNumber := endExtHighestSequenceNumber - 5
for esn := startExtHighestSequenceNumber; esn <= endExtHighestSequenceNumber; esn++ {
epm := d.sequencer.lookupExtPacketMeta(esn)
if epm == nil {
continue
}
pktBuff := *src
n, err := receiver.ReadRTP(pktBuff, uint8(epm.layer), epm.sourceSeqNo)
if err != nil {
if err == io.EOF {
break
}
continue
}
sent, _ := d.retransmitPacket(epm, pktBuff[:n], true)
bytesSent += sent
if bytesSent >= bytesToSend {
break
}
}
}
return bytesSent
}
func (d *DownTrack) addDummyExtensions(hdr *rtp.Header) {
// add dummy extensions (actual ones will be filed by pacer) to get header size
if d.absSendTimeExtID != 0 {
hdr.SetExtension(uint8(d.absSendTimeExtID), dummyAbsSendTimeExt)
}
if d.transportWideExtID != 0 {
hdr.SetExtension(uint8(d.transportWideExtID), dummyTransportCCExt)
}
}
func (d *DownTrack) getTranslatedPayloadType(srcPT uint8) uint8 {
// send primary codec to subscriber if the publisher sent primary codec when red is negotiated,
// this will happen when the payload is too large to encode into red payload (exceeds mtu).
if d.isRED && srcPT == d.upstreamPrimaryPT && d.primaryPT != 0 {
return d.primaryPT
}
return uint8(d.payloadType.Load())
}
func (d *DownTrack) DebugInfo() map[string]any {
stats := map[string]any{
"LastPli": d.rtpStats.LastPli(),
}
stats["RTPMunger"] = d.forwarder.RTPMungerDebugInfo()
senderReport := d.CreateSenderReport()
if senderReport != nil {
stats["NTPTime"] = senderReport.NTPTime
stats["RTPTime"] = senderReport.RTPTime
stats["PacketCount"] = senderReport.PacketCount
}
return map[string]any{
"SubscriberID": d.params.SubID,
"TrackID": d.id,
"StreamID": d.params.StreamID,
"SSRC": d.ssrc,
"MimeType": d.Mime().String(),
"BindState": d.bindState.Load().(bindState),
"Muted": d.forwarder.IsMuted(),
"PubMuted": d.forwarder.IsPubMuted(),
"CurrentSpatialLayer": d.forwarder.CurrentLayer().Spatial,
"Stats": stats,
}
}
func (d *DownTrack) GetConnectionScoreAndQuality() (float32, livekit.ConnectionQuality) {
return d.connectionStats.GetScoreAndQuality()
}
// OnStatsUpdate registers an additional callback that fires alongside the
// configured DownTrackListener whenever connection-quality stats are produced.
// Intended for tests and observers; the production listener path is unaffected.
func (d *DownTrack) OnStatsUpdate(fn func(d *DownTrack, stat *livekit.AnalyticsStat)) {
d.onStatsUpdate.Store(fn)
}
func (d *DownTrack) GetTrackStats() *livekit.RTPStats {
return rtpstats.ReconcileRTPStatsWithRTX(d.rtpStats.ToProto(), d.rtpStatsRTX.ToProto())
}
func (d *DownTrack) deltaStats(ds *rtpstats.RTPDeltaInfo, dsrv *rtpstats.RTPDeltaInfo) map[uint32]*buffer.StreamStatsWithLayers {
if ds == nil && dsrv == nil {
return nil
}
streamStats := make(map[uint32]*buffer.StreamStatsWithLayers, 1)
streamStats[d.ssrc] = &buffer.StreamStatsWithLayers{
RTPStats: ds,
RTPStatsRemoteView: dsrv,
Layers: map[int32]*rtpstats.RTPDeltaInfo{
0: ds,
},
}
return streamStats
}
func (d *DownTrack) GetDeltaStatsSender() map[uint32]*buffer.StreamStatsWithLayers {
ds, dsrv := d.rtpStats.DeltaInfoSender(d.deltaStatsSenderSnapshotId)
dsRTX, dsrvRTX := d.rtpStatsRTX.DeltaInfoSender(d.deltaStatsRTXSenderSnapshotId)
return d.deltaStats(
rtpstats.ReconcileRTPDeltaInfoWithRTX(ds, dsRTX),
rtpstats.ReconcileRTPDeltaInfoWithRTX(dsrv, dsrvRTX),
)
}
func (d *DownTrack) GetPrimaryStreamLastReceiverReportTime() time.Time {
return time.Unix(0, d.rtpStats.LastReceiverReportTime())
}
func (d *DownTrack) GetPrimaryStreamPacketsSent() uint64 {
return d.rtpStats.GetPacketsSeenMinusPadding()
}
func (d *DownTrack) GetNackStats() (totalPackets uint32, totalRepeatedNACKs uint32) {
totalPackets = uint32(d.rtpStats.GetPacketsSeenMinusPadding())
totalRepeatedNACKs = d.totalRepeatedNACKs.Load()
return
}
func (d *DownTrack) onBindAndConnectedChange() {
if d.writeStopped.Load() {
return
}
d.writable.Store(d.connected.Load() && d.bindState.Load() == bindStateBound)
if d.connected.Load() && d.bindState.Load() == bindStateBound && !d.bindAndConnectedOnce.Swap(true) {
go d.params.Listener.OnBindAndConnected()
if d.activePaddingOnMuteUpTrack.Load() {
go d.sendPaddingOnMute()
}
// kick off PLI request if allocation is pending
d.postKeyFrameRequestEvent()
}
}
func (d *DownTrack) sendPaddingOnMute() {
// let uptrack have chance to send packet before we send padding
time.Sleep(waitBeforeSendPaddingOnMute)
if d.kind == webrtc.RTPCodecTypeVideo {
d.sendPaddingOnMuteForVideo()
} else {
d.sendSilentFrameOnMuteForAudio()
}
}
func (d *DownTrack) sendPaddingOnMuteForVideo() {
numPackets := maxPaddingOnMuteDuration / paddingOnMuteInterval
for i := range int(numPackets) {
if d.rtpStats.IsActive() || d.IsClosed() {
return
}
if i == 0 {
d.params.Logger.Debugw("sending padding on mute")
}
d.WritePaddingRTP(20, true, true)
time.Sleep(paddingOnMuteInterval)
}
}
func (d *DownTrack) sendSilentFrameOnMuteForAudio() {
var (
payload []byte
err error
)
switch d.Mime() {
case mime.MimeTypeOpus:
payload, err = d.getAudioBlankFrameFunc(OpusSilenceFrame)(false)
case mime.MimeTypeRED:
payload, err = d.getOpusRedBlankFrame(false)
case mime.MimeTypePCMU:
payload, err = d.getAudioBlankFrameFunc(PCMUSilenceFrame)(false)
case mime.MimeTypePCMA:
payload, err = d.getAudioBlankFrameFunc(PCMASilenceFrame)(false)
default:
d.params.Logger.Infow("unsupported mime type for silent frame on mute", "mimeType", d.Mime())
return
}
if err != nil {
d.params.Logger.Warnw("could not get blank frame", err)
return
}
frameRate := uint32(50)
frameDuration := time.Duration(1000/frameRate) * time.Millisecond
numFrames := frameRate * uint32(maxPaddingOnMuteDuration/time.Second)
first := true
for {
if d.rtpStats.IsActive() || d.IsClosed() || numFrames <= 0 {
return
}
if first {
first = false
d.params.Logger.Debugw("sending padding on mute")
}
snts, _, err := d.forwarder.GetSnTsForBlankFrames(frameRate, 1)
if err != nil {
d.params.Logger.Warnw("could not get SN/TS for blank frame", err)
return
}
for i := range len(snts) {
hdr := &rtp.Header{
Version: 2,
Padding: false,
Marker: true,
PayloadType: uint8(d.payloadType.Load()),
SequenceNumber: uint16(snts[i].extSequenceNumber),
Timestamp: uint32(snts[i].extTimestamp),
SSRC: d.ssrc,
}
d.addDummyExtensions(hdr)
headerSize := hdr.MarshalSize()
d.rtpStats.Update(
mono.UnixNano(),
snts[i].extSequenceNumber,
snts[i].extTimestamp,
hdr.Marker,
headerSize,
0,
len(payload), // although this is using empty frames, mark as padding as these are used to trigger Pion OnTrack only
false,
)
pacerPacket := pacer.PacketFactory.Get().(*pacer.Packet)
*pacerPacket = pacer.Packet{
Header: hdr,
HeaderSize: headerSize,
Payload: payload,
ProbeClusterId: ccutils.ProbeClusterId(d.probeClusterId.Load()),
AbsSendTimeExtID: uint8(d.absSendTimeExtID),
TransportWideExtID: uint8(d.transportWideExtID),
WriteStream: d.writeStream,
}
d.pacer.Enqueue(pacerPacket)
}
numFrames--
time.Sleep(frameDuration)
}
}
func (d *DownTrack) HandleRTCPSenderReportData(
_payloadType webrtc.PayloadType,
layer int32,
publisherSRData *livekit.RTCPSenderReportState,
) error {
d.forwarder.SetRefSenderReport(layer, publisherSRData)
currentLayer, isSingleStream, tsOffset, refSenderReport := d.forwarder.GetSenderReportParams()
if layer == currentLayer || (layer == 0 && isSingleStream) {
d.handleRTCPSenderReportData(refSenderReport, tsOffset)
}
return nil
}
func (d *DownTrack) handleRTCPSenderReportData(publisherSRData *livekit.RTCPSenderReportState, tsOffset uint64) {
d.rtpStats.MaybeAdjustFirstPacketTime(publisherSRData, tsOffset)
}
// -------------------------------------------------------------------------------
Reference in New Issue View Git Blame Copy Permalink