sfu

package
v0.0.0-...-4d7396a Latest Latest
Warning

This package is not in the latest version of its module.

 Go to latest Published: Apr 4, 2022 License: Apache-2.0 Imports: 27 Imported by: 0

Documentation

Overview

Design of Prober

Probing is used to check for existence of excess channel capacity. This is especially useful in the downstream direction of SFU. SFU forwards audio/video streams from one or more publishers to all the subscribers. But, the downstream channel of a subscriber may not be big enough to carry all the streams. It is also a time varying quantity.

When there is not enough capacity, some streams will be paused. To resume a stream, SFU would need to know that the channel has enough capacity. That's where probing comes in. When conditions are favorable, SFU can send probe packets so that the bandwidth estimator has more data to estimate available channel capacity better. NOTE: What defines `favorable conditions` is implementation dependent.

There are two options for probing

  • Use padding only RTP packets: This one is preferable as probe rate can be controlled more tightly.
  • Resume a paused stream or forward a higher spatial layer: Have to find a stream at probing rate. Also, a stream could get a key frame unexpectedly boosting rate in the probing window.

The strategy used depends on stream allocator implementation. This module can be used if the stream allocator decides to use padding only RTP packets for probing purposes.

Implementation: There are a couple of options

  • Check prober in the forwarding path (pull from prober). This is preferred for scalability reasons. But, this suffers from not being able to probe when all streams are paused (could be due to downstream bandwidth constraints or the corresponding upstream tracks may have paused due to upstream bandwidth constraints). Another issue is not being to have tight control on probing window boundary as the packet forwarding path may not have a packet to forward. But, it should not be a major concern as long as some stream(s) is/are forwarded as there should be a packet at least every 60 ms or so (forwarding only one stream at 15 fps). Usually, it will be serviced much more frequently when there are multiple streams getting forwarded.
  • Run it a go routine. But, that would have to wake up very often to prevent bunching up of probe packets. So, a scalability concern as there is one prober per subscriber peer connection. But, probe windows should be very short (of the order of 100s of ms). So, this approach might be fine.

The implementation here follows the second approach of using a go routine.

Pacing: ------ Ideally, the subscriber peer connection should have a pacer which trickles data out at the estimated channel capacity rate (and estimated channel capacity + probing rate when actively probing).

But, there a few significant challenges

  1. Pacer will require buffering of forwarded packets. That means more memory, more CPU (have to make copy of packets) and more latency in the media stream.
  2. Scalability concern as SFU may be handling hundreds of subscriber peer connections and each one processing the pacing loop at 5ms interval will add up.

So, this module assumes that pacing is inherently provided by the publishers for media streams. That is a reasonable assumption given that publishing clients will run their own pacer and pacing data out at a steady rate.

A further assumption is that if there are multiple publishers for a subscriber peer connection, all the publishers are not pacing in sync, i.e. each publisher's pacer is completely independent and SFU will be receiving the media packets with a good spread and not clumped together.

Given those assumptions, this module monitors media send rate and adjusts probing packet sends accordingly. Although the probing may have a high enough wake up frequency, it is for short windows. For example, probing at 5 Mbps for 1/2 second and sending 1000 byte probe per iteration will wake up every 1.6 ms. That is very high, but should last for 1/2 second or so. 

5 Mbps over 1/2 second = 2.5 Mbps
2.5 Mbps = 312500 bytes = 313 probes at 1000 byte probes
313 probes over 1/2 second = 1.6 ms between probes

A few things to note

  1. When a probe cluster is added, the expected media rate is provided. So, the wake-up interval takes that into account. For example, if probing at 5 Mbps for 1/2 second and if 4 Mbps of it is expected to be provided by media traffic, the wake-up interval becomes 8 ms.
  2. The amount of probing should actually be capped at some value to avoid too much self-induced congestion. It maybe something like 500 kbps. That will increase the wake-up interval to 16 ms in the above example.
  3. In practice, the probing interval may also be shorter. Typically, it can be run for 2 - 3 RTTs to get a good measurement. For the longest hauls, RTT could be 250 ms or so leading to the probing window being long(ish). But, RTT should be much shorter especially if the subscriber peer connection of the client is able to connect to the nearest data center.

Index

Constants

View Source 
const (
	RTPPaddingMaxPayloadSize      = 255
	RTPPaddingEstimatedHeaderSize = 20
	RTPBlankFramesMax             = 6

	FlagStopRTXOnPLI = true
)
View Source 
const (
	FlagPauseOnDowngrade = true
	FlagFilterRTX        = true
)

Forwarder

View Source 
const (
	InvalidLayerSpatial  = int32(-1)
	InvalidLayerTemporal = int32(-1)

	DefaultMaxLayerSpatial  = int32(2)
	DefaultMaxLayerTemporal = int32(3)
)
View Source 
const (
	QuarterResolution = "q"
	HalfResolution    = "h"
	FullResolution    = "f"
)
View Source 
const (
	ChannelCapacityInfinity = 100 * 1000 * 1000 // 100 Mbps

	NumRequiredEstimatesNonProbe = 8
	NumRequiredEstimatesProbe    = 3

	NackRatioThresholdNonProbe = 0.06
	NackRatioThresholdProbe    = 0.04

	NackRatioAttenuator = 0.4 // how much to attenuate NACK ratio while calculating loss adjusted estimate

	ProbeWaitBase      = 5 * time.Second
	ProbeBackoffFactor = 1.5
	ProbeWaitMax       = 30 * time.Second
	ProbeSettleWait    = 250
	ProbeTrendWait     = 2 * time.Second

	ProbePct         = 120
	ProbeMinBps      = 200 * 1000 // 200 kbps
	ProbeMinDuration = 20 * time.Second
	ProbeMaxDuration = 21 * time.Second

	PriorityMin                = uint8(1)
	PriorityMax                = uint8(255)
	PriorityDefaultScreenshare = PriorityMax
	PriorityDefaultVideo       = PriorityMin
)
View Source 
const (
	RtxGateWindow = 2000
)
View Source 
const (
	TransitionCostSpatial = 10
)

Variables

View Source 
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")
	ErrNotVP8                            = errors.New("not VP8")
	ErrOutOfOrderVP8PictureIdCacheMiss   = errors.New("out-of-order VP8 picture id not found in cache")
	ErrFilteredVP8TemporalLayer          = errors.New("filtered VP8 temporal layer")
	ErrTrackAlreadyBind                  = errors.New("already bind")
)
View Source 
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}
)
View Source 
var (
	InvalidLayers = VideoLayers{
					// contains filtered or unexported fields
	}

	DefaultMaxLayers = VideoLayers{
						// contains filtered or unexported fields
	}
)
View Source 
var (
	ErrReceiverClosed        = errors.New("receiver closed")
	ErrDownTrackAlreadyExist = errors.New("DownTrack already exist")
)
View Source 
var (
	ConfigVideo = []StreamTrackerParams{
		{
			SamplesRequired: 1,
			CyclesRequired:  4,
			CycleDuration:   500 * time.Millisecond,
		},
		{
			SamplesRequired: 5,
			CyclesRequired:  60,
			CycleDuration:   500 * time.Millisecond,
		},
		{
			SamplesRequired: 5,
			CyclesRequired:  60,
			CycleDuration:   500 * time.Millisecond,
		},
	}

	// be very forgiving for screen share to account for cases like static screen where there could be only one packet per second
	ConfigScreenshare = []StreamTrackerParams{
		{
			SamplesRequired: 1,
			CyclesRequired:  1,
			CycleDuration:   2 * time.Second,
		},
		{
			SamplesRequired: 1,
			CyclesRequired:  1,
			CycleDuration:   2 * time.Second,
		},
		{
			SamplesRequired: 1,
			CyclesRequired:  1,
			CycleDuration:   2 * time.Second,
		},
	}
)
View Source 
var (
	PacketFactory *sync.Pool
)
View Source 
var (
	VideoAllocationDefault = VideoAllocation{
		// contains filtered or unexported fields
	}
)

Functions

func RidToLayer 

func RidToLayer(rid string) int32

Types

type AddTrackParams 

type AddTrackParams struct {
	Source      livekit.TrackSource
	Priority    uint8
	IsSimulcast bool
	PublisherID livekit.ParticipantID
}

type AudioLevelHandle 

type AudioLevelHandle func(level uint8, duration uint32)

type Bitrates 

type Bitrates [DefaultMaxLayerSpatial + 1][DefaultMaxLayerTemporal + 1]int64

type ChannelObserver 

type ChannelObserver struct {
	// contains filtered or unexported fields
}

func NewChannelObserver 

func NewChannelObserver(
	name string,
	logger logger.Logger,
	estimateRequiredSamples int,
	nackRatioThreshold float64,
) *ChannelObserver

func (*ChannelObserver) AddEstimate 

func (c *ChannelObserver) AddEstimate(estimate int64)

func (*ChannelObserver) AddNack 

func (c *ChannelObserver) AddNack(packets uint32, repeatedNacks uint32)

func (*ChannelObserver) GetHighestEstimate 

func (c *ChannelObserver) GetHighestEstimate() int64

func (*ChannelObserver) GetLowestEstimate 

func (c *ChannelObserver) GetLowestEstimate() int64

func (*ChannelObserver) GetNackRatio 

func (c *ChannelObserver) GetNackRatio() float64

func (*ChannelObserver) GetTrend 

func (c *ChannelObserver) GetTrend() ChannelTrend

func (*ChannelObserver) Reset 

func (c *ChannelObserver) Reset()

func (*ChannelObserver) SeedEstimate 

func (c *ChannelObserver) SeedEstimate(estimate int64)

func (*ChannelObserver) SeedNack 

func (c *ChannelObserver) SeedNack(packets uint32, repeatedNacks uint32)

type ChannelTrend 

type ChannelTrend int
const (
	ChannelTrendNeutral ChannelTrend = iota
	ChannelTrendClearing
	ChannelTrendCongesting
)

func (ChannelTrend) String 

func (c ChannelTrend) String() string

type Cluster 

type Cluster struct {
	// contains filtered or unexported fields
}

func NewCluster 

func NewCluster(id ProbeClusterId, desiredRateBps int, expectedRateBps int, minDuration time.Duration, maxDuration time.Duration) *Cluster

func (*Cluster) GetInfo 

func (c *Cluster) GetInfo() ProbeClusterInfo

func (*Cluster) GetSleepDuration 

func (c *Cluster) GetSleepDuration() time.Duration

func (*Cluster) IsFinished 

func (c *Cluster) IsFinished() bool

func (*Cluster) PacketSent 

func (c *Cluster) PacketSent(size int)

func (*Cluster) ProbeSent 

func (c *Cluster) ProbeSent(size int)

func (*Cluster) Process 

func (c *Cluster) Process(onSendProbe func(bytesToSend int))

func (*Cluster) Start 

func (c *Cluster) Start()

func (*Cluster) String 

func (c *Cluster) String() string

type DownTrack 

type DownTrack struct {
	// contains filtered or unexported fields
}

DownTrack implements TrackLocal, is the track used to write packets to SFU Subscriber, the track handle the packets for simple, simulcast and SVC Publisher.

func NewDownTrack 

func NewDownTrack(
	c webrtc.RTPCodecCapability,
	r TrackReceiver,
	bf *buffer.Factory,
	peerID livekit.ParticipantID,
	mt int,
	logger logger.Logger,
) (*DownTrack, error)

NewDownTrack returns a DownTrack.

func (*DownTrack) AddReceiverReportListener 

func (d *DownTrack) AddReceiverReportListener(listener ReceiverReportListener)

func (*DownTrack) AllocateNextHigher 

func (d *DownTrack) AllocateNextHigher(availableChannelCapacity int64) (VideoAllocation, bool)

func (*DownTrack) AllocateOptimal 

func (d *DownTrack) AllocateOptimal() VideoAllocation

func (*DownTrack) BandwidthRequested 

func (d *DownTrack) BandwidthRequested() int64

func (*DownTrack) Bind 

func (d *DownTrack) Bind(t webrtc.TrackLocalContext) (webrtc.RTPCodecParameters, error)

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 (*DownTrack) Close 

func (d *DownTrack) Close()

func (*DownTrack) CloseWithFlush 

func (d *DownTrack) CloseWithFlush(flush bool)

Close track, flush used to indicate whether send blank frame to flush decoder of client.

  1. When transceiver is reused by other participant's video track, set flush=true to avoid previous video shows before previous 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 flush decoder.

func (*DownTrack) Codec 

func (d *DownTrack) Codec() webrtc.RTPCodecCapability

Codec returns current track codec capability

func (*DownTrack) CreateSenderReport 

func (d *DownTrack) CreateSenderReport() *rtcp.SenderReport

func (*DownTrack) CreateSourceDescriptionChunks 

func (d *DownTrack) CreateSourceDescriptionChunks() []rtcp.SourceDescriptionChunk

func (*DownTrack) DebugInfo 

func (d *DownTrack) DebugInfo() map[string]interface{}

func (*DownTrack) DistanceToDesired 

func (d *DownTrack) DistanceToDesired() int32

func (*DownTrack) GetConnectionScore 

func (d *DownTrack) GetConnectionScore() float32

func (*DownTrack) GetForwardingStatus 

func (d *DownTrack) GetForwardingStatus() ForwardingStatus

func (*DownTrack) GetNackStats 

func (d *DownTrack) GetNackStats() (totalPackets uint32, totalRepeatedNACKs uint32)

func (*DownTrack) GetNextHigherTransition 

func (d *DownTrack) GetNextHigherTransition() (VideoTransition, bool)

func (*DownTrack) GetTrackStats 

func (d *DownTrack) GetTrackStats() map[uint32]*buffer.StreamStatsWithLayers

func (*DownTrack) ID 

func (d *DownTrack) ID() string

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 (*DownTrack) IsDeficient 

func (d *DownTrack) IsDeficient() bool

func (*DownTrack) Kind 

func (d *DownTrack) Kind() webrtc.RTPCodecType

Kind controls if this TrackLocal is audio or video

func (*DownTrack) MaxLayers 

func (d *DownTrack) MaxLayers() VideoLayers

func (*DownTrack) Mute 

func (d *DownTrack) Mute(muted bool)

Mute enables or disables media forwarding

func (*DownTrack) OnAvailableLayersChanged 

func (d *DownTrack) OnAvailableLayersChanged(fn func(dt *DownTrack))

func (*DownTrack) OnBind 

func (d *DownTrack) OnBind(fn func())

func (*DownTrack) OnBitrateAvailabilityChanged 

func (d *DownTrack) OnBitrateAvailabilityChanged(fn func(dt *DownTrack))

func (*DownTrack) OnCloseHandler 

func (d *DownTrack) OnCloseHandler(fn func())

OnCloseHandler method to be called on remote tracked removed

func (*DownTrack) OnMaxLayerChanged 

func (d *DownTrack) OnMaxLayerChanged(fn func(dt *DownTrack, layer int32))

func (*DownTrack) OnPacketSentUnsafe 

func (d *DownTrack) OnPacketSentUnsafe(fn func(dt *DownTrack, size int))

func (*DownTrack) OnPaddingSentUnsafe 

func (d *DownTrack) OnPaddingSentUnsafe(fn func(dt *DownTrack, size int))

func (*DownTrack) OnREMB 

func (d *DownTrack) OnREMB(fn func(dt *DownTrack, remb *rtcp.ReceiverEstimatedMaximumBitrate))

func (*DownTrack) OnRttUpdate 

func (d *DownTrack) OnRttUpdate(fn func(dt *DownTrack, rtt uint32))

func (*DownTrack) OnStatsUpdate 

func (d *DownTrack) OnStatsUpdate(fn func(dt *DownTrack, stat *livekit.AnalyticsStat))

func (*DownTrack) OnSubscribedLayersChanged 

func (d *DownTrack) OnSubscribedLayersChanged(fn func(dt *DownTrack, layers VideoLayers))

func (*DownTrack) OnSubscriptionChanged 

func (d *DownTrack) OnSubscriptionChanged(fn func(dt *DownTrack))

func (*DownTrack) OnTransportCCFeedback 

func (d *DownTrack) OnTransportCCFeedback(fn func(dt *DownTrack, cc *rtcp.TransportLayerCC))

func (*DownTrack) Pause 

func (d *DownTrack) Pause() VideoAllocation

func (*DownTrack) PeerID 

func (d *DownTrack) PeerID() livekit.ParticipantID

func (*DownTrack) ProvisionalAllocate 

func (d *DownTrack) ProvisionalAllocate(availableChannelCapacity int64, layers VideoLayers, allowPause bool) int64

func (*DownTrack) ProvisionalAllocateCommit 

func (d *DownTrack) ProvisionalAllocateCommit() VideoAllocation

func (*DownTrack) ProvisionalAllocateGetBestWeightedTransition 

func (d *DownTrack) ProvisionalAllocateGetBestWeightedTransition() VideoTransition

func (*DownTrack) ProvisionalAllocateGetCooperativeTransition 

func (d *DownTrack) ProvisionalAllocateGetCooperativeTransition() VideoTransition

func (*DownTrack) ProvisionalAllocatePrepare 

func (d *DownTrack) ProvisionalAllocatePrepare()

func (*DownTrack) RID 

func (d *DownTrack) RID() string

RID is required by `webrtc.TrackLocal` interface

func (*DownTrack) Resync 

func (d *DownTrack) Resync()

func (*DownTrack) SSRC 

func (d *DownTrack) SSRC() uint32

func (*DownTrack) SetMaxSpatialLayer 

func (d *DownTrack) SetMaxSpatialLayer(spatialLayer int32)

func (*DownTrack) SetMaxTemporalLayer 

func (d *DownTrack) SetMaxTemporalLayer(temporalLayer int32)

func (*DownTrack) SetRTPHeaderExtensions 

func (d *DownTrack) SetRTPHeaderExtensions(rtpHeaderExtensions []webrtc.RTPHeaderExtensionParameter)

Sets RTP header extensions for this track

func (*DownTrack) SetTransceiver 

func (d *DownTrack) SetTransceiver(transceiver *webrtc.RTPTransceiver)

func (*DownTrack) Stop 

func (d *DownTrack) Stop() error

func (*DownTrack) StreamID 

func (d *DownTrack) StreamID() string

StreamID is the group this track belongs too. This must be unique

func (*DownTrack) Unbind 

func (d *DownTrack) Unbind(_ webrtc.TrackLocalContext) error

Unbind implements the teardown logic when the track is no longer needed. This happens because a track has been stopped.

func (*DownTrack) UpTrackBitrateAvailabilityChange 

func (d *DownTrack) UpTrackBitrateAvailabilityChange()

func (*DownTrack) UpTrackLayersChange 

func (d *DownTrack) UpTrackLayersChange(availableLayers []int32)

func (*DownTrack) WritePaddingRTP 

func (d *DownTrack) WritePaddingRTP(bytesToSend int) int

WritePaddingRTP tries to write as many padding only RTP packets as necessary to satisfy given size to the DownTrack

func (*DownTrack) WriteRTP 

func (d *DownTrack) WriteRTP(extPkt *buffer.ExtPacket, layer int32) error

WriteRTP writes an RTP Packet to the DownTrack

type Event 

type Event struct {
	Signal    Signal
	DownTrack *DownTrack
	Data      interface{}
}

func (Event) String 

func (e Event) String() string

type Forwarder 

type Forwarder struct {
	// contains filtered or unexported fields
}

func NewForwarder 

func NewForwarder(codec webrtc.RTPCodecCapability, kind webrtc.RTPCodecType, logger logger.Logger) *Forwarder

func (*Forwarder) AllocateNextHigher 

func (f *Forwarder) AllocateNextHigher(availableChannelCapacity int64, brs Bitrates) (VideoAllocation, bool)

func (*Forwarder) AllocateOptimal 

func (f *Forwarder) AllocateOptimal(brs Bitrates) VideoAllocation

func (*Forwarder) BandwidthRequested 

func (f *Forwarder) BandwidthRequested(brs Bitrates) int64

func (*Forwarder) CheckSync 

func (f *Forwarder) CheckSync() (locked bool, layer int32)

func (*Forwarder) CurrentLayers 

func (f *Forwarder) CurrentLayers() VideoLayers

func (*Forwarder) DistanceToDesired 

func (f *Forwarder) DistanceToDesired() int32

func (*Forwarder) FilterRTX 

func (f *Forwarder) FilterRTX(nacks []uint16) (filtered []uint16, disallowedLayers [DefaultMaxLayerSpatial + 1]bool)

func (*Forwarder) GetForwardingStatus 

func (f *Forwarder) GetForwardingStatus() ForwardingStatus

func (*Forwarder) GetNextHigherTransition 

func (f *Forwarder) GetNextHigherTransition(brs Bitrates) (VideoTransition, bool)

func (*Forwarder) GetPaddingVP8 

func (f *Forwarder) GetPaddingVP8(frameEndNeeded bool) *buffer.VP8

func (*Forwarder) GetRTPMungerParams 

func (f *Forwarder) GetRTPMungerParams() RTPMungerParams

func (*Forwarder) GetSnTsForBlankFrames 

func (f *Forwarder) GetSnTsForBlankFrames() ([]SnTs, bool, error)

func (*Forwarder) GetSnTsForPadding 

func (f *Forwarder) GetSnTsForPadding(num int) ([]SnTs, error)

func (*Forwarder) GetTranslationParams 

func (f *Forwarder) GetTranslationParams(extPkt *buffer.ExtPacket, layer int32) (*TranslationParams, error)

func (*Forwarder) IsDeficient 

func (f *Forwarder) IsDeficient() bool

func (*Forwarder) IsMuted 

func (f *Forwarder) IsMuted() bool

func (*Forwarder) MaxLayers 

func (f *Forwarder) MaxLayers() VideoLayers

func (*Forwarder) Mute 

func (f *Forwarder) Mute(val bool) (bool, VideoLayers)

func (*Forwarder) Pause 

func (f *Forwarder) Pause(brs Bitrates) VideoAllocation

func (*Forwarder) ProvisionalAllocate 

func (f *Forwarder) ProvisionalAllocate(availableChannelCapacity int64, layers VideoLayers, allowPause bool) int64

func (*Forwarder) ProvisionalAllocateCommit 

func (f *Forwarder) ProvisionalAllocateCommit() VideoAllocation

func (*Forwarder) ProvisionalAllocateGetBestWeightedTransition 

func (f *Forwarder) ProvisionalAllocateGetBestWeightedTransition() VideoTransition

func (*Forwarder) ProvisionalAllocateGetCooperativeTransition 

func (f *Forwarder) ProvisionalAllocateGetCooperativeTransition() VideoTransition

func (*Forwarder) ProvisionalAllocatePrepare 

func (f *Forwarder) ProvisionalAllocatePrepare(bitrates Bitrates)

func (*Forwarder) Resync 

func (f *Forwarder) Resync()

func (*Forwarder) SetMaxSpatialLayer 

func (f *Forwarder) SetMaxSpatialLayer(spatialLayer int32) (bool, VideoLayers, VideoLayers)

func (*Forwarder) SetMaxTemporalLayer 

func (f *Forwarder) SetMaxTemporalLayer(temporalLayer int32) (bool, VideoLayers, VideoLayers)

func (*Forwarder) TargetLayers 

func (f *Forwarder) TargetLayers() VideoLayers

func (*Forwarder) UpTrackLayersChange 

func (f *Forwarder) UpTrackLayersChange(availableLayers []int32)

type ForwardingStatus 

type ForwardingStatus int
const (
	ForwardingStatusOff ForwardingStatus = iota
	ForwardingStatusPartial
	ForwardingStatusOptimal
)

type MaxDistanceSorter 

type MaxDistanceSorter []*Track

func (MaxDistanceSorter) Len 

func (m MaxDistanceSorter) Len() int

func (MaxDistanceSorter) Less 

func (m MaxDistanceSorter) Less(i, j int) bool

func (MaxDistanceSorter) Swap 

func (m MaxDistanceSorter) Swap(i, j int)

type MinDistanceSorter 

type MinDistanceSorter []*Track

func (MinDistanceSorter) Len 

func (m MinDistanceSorter) Len() int

func (MinDistanceSorter) Less 

func (m MinDistanceSorter) Less(i, j int) bool

func (MinDistanceSorter) Swap 

func (m MinDistanceSorter) Swap(i, j int)

type ProbeClusterId 

type ProbeClusterId uint32
const (
	ProbeClusterIdInvalid ProbeClusterId = 0
)

type ProbeClusterInfo 

type ProbeClusterInfo struct {
	Id        ProbeClusterId
	BytesSent int
	Duration  time.Duration
}

type Prober 

type Prober struct {
	// contains filtered or unexported fields
}

func NewProber 

func NewProber(params ProberParams) *Prober

func (*Prober) AddCluster 

func (p *Prober) AddCluster(desiredRateBps int, expectedRateBps int, minDuration time.Duration, maxDuration time.Duration) ProbeClusterId

func (*Prober) IsRunning 

func (p *Prober) IsRunning() bool

func (*Prober) OnProbeClusterDone 

func (p *Prober) OnProbeClusterDone(f func(info ProbeClusterInfo))

func (*Prober) OnSendProbe 

func (p *Prober) OnSendProbe(f func(bytesToSend int))

func (*Prober) PacketSent 

func (p *Prober) PacketSent(size int)

func (*Prober) ProbeSent 

func (p *Prober) ProbeSent(size int)

func (*Prober) Reset 

func (p *Prober) Reset()

type ProberParams 

type ProberParams struct {
	Logger logger.Logger
}

type RTPMunger 

type RTPMunger struct {
	RTPMungerParams
	// contains filtered or unexported fields
}

func NewRTPMunger 

func NewRTPMunger(logger logger.Logger) *RTPMunger

func (*RTPMunger) FilterRTX 

func (r *RTPMunger) FilterRTX(nacks []uint16) []uint16

func (*RTPMunger) GetParams 

func (r *RTPMunger) GetParams() RTPMungerParams

func (*RTPMunger) IsOnFrameBoundary 

func (r *RTPMunger) IsOnFrameBoundary() bool

func (*RTPMunger) PacketDropped 

func (r *RTPMunger) PacketDropped(extPkt *buffer.ExtPacket)

func (*RTPMunger) SetLastSnTs 

func (r *RTPMunger) SetLastSnTs(extPkt *buffer.ExtPacket)

func (*RTPMunger) UpdateAndGetPaddingSnTs 

func (r *RTPMunger) UpdateAndGetPaddingSnTs(num int, clockRate uint32, frameRate uint32, forceMarker bool) ([]SnTs, error)

func (*RTPMunger) UpdateAndGetSnTs 

func (r *RTPMunger) UpdateAndGetSnTs(extPkt *buffer.ExtPacket) (*TranslationParamsRTP, error)

func (*RTPMunger) UpdateSnTsOffsets 

func (r *RTPMunger) UpdateSnTsOffsets(extPkt *buffer.ExtPacket, snAdjust uint16, tsAdjust uint32)

type RTPMungerParams 

type RTPMungerParams struct {
	// contains filtered or unexported fields
}

type ReceiverOpts 

type ReceiverOpts func(w *WebRTCReceiver) *WebRTCReceiver

func WithLoadBalanceThreshold 

func WithLoadBalanceThreshold(downTracks int) ReceiverOpts

WithLoadBalanceThreshold enables parallelization of packet writes when downTracks exceeds threshold Value should be between 3 and 150. For a server handling a few large rooms, use a smaller value (required to handle very large (250+ participant) rooms). For a server handling many small rooms, use a larger value or disable. Set to 0 (disabled) by default.

func WithPliThrottle 

func WithPliThrottle(pliThrottleConfig config.PLIThrottleConfig) ReceiverOpts

WithPliThrottle indicates minimum time(ms) between sending PLIs

func WithStreamTrackers 

func WithStreamTrackers() ReceiverOpts

WithStreamTrackers enables StreamTracker use for simulcast

type ReceiverReportListener 

type ReceiverReportListener func(dt *DownTrack, report *rtcp.ReceiverReport)

type SequenceNumberOrdering 

type SequenceNumberOrdering int

RTPMunger 

const (
	SequenceNumberOrderingContiguous SequenceNumberOrdering = iota
	SequenceNumberOrderingOutOfOrder
	SequenceNumberOrderingGap
	SequenceNumberOrderingDuplicate
)

type Signal 

type Signal int
const (
	SignalAddTrack Signal = iota
	SignalRemoveTrack
	SignalSetTrackPriority
	SignalEstimate
	SignalTargetBitrate
	SignalAvailableLayersChange
	SignalBitrateAvailabilityChange
	SignalSubscriptionChange
	SignalSubscribedLayersChange
	SignalPeriodicPing
	SignalSendProbe
	SignalProbeClusterDone
)

func (Signal) String 

func (s Signal) String() string

type SnTs 

type SnTs struct {
	// contains filtered or unexported fields
}

type State 

type State int
const (
	StateStable State = iota
	StateDeficient
)

func (State) String 

func (s State) String() string

type StreamAllocator 

type StreamAllocator struct {
	// contains filtered or unexported fields
}

func NewStreamAllocator 

func NewStreamAllocator(params StreamAllocatorParams) *StreamAllocator

func (*StreamAllocator) AddTrack 

func (s *StreamAllocator) AddTrack(downTrack *DownTrack, params AddTrackParams)

func (*StreamAllocator) OnStreamStateChange 

func (s *StreamAllocator) OnStreamStateChange(f func(update *StreamStateUpdate) error)

func (*StreamAllocator) RemoveTrack 

func (s *StreamAllocator) RemoveTrack(downTrack *DownTrack)

func (*StreamAllocator) SetBandwidthEstimator 

func (s *StreamAllocator) SetBandwidthEstimator(bwe cc.BandwidthEstimator)

func (*StreamAllocator) SetTrackPriority 

func (s *StreamAllocator) SetTrackPriority(downTrack *DownTrack, priority uint8)

func (*StreamAllocator) Start 

func (s *StreamAllocator) Start()

func (*StreamAllocator) Stop 

func (s *StreamAllocator) Stop()

type StreamAllocatorParams 

type StreamAllocatorParams struct {
	Config config.CongestionControlConfig
	Logger logger.Logger
}

type StreamState 

type StreamState int
const (
	StreamStateActive StreamState = iota
	StreamStatePaused
)

type StreamStateInfo 

type StreamStateInfo struct {
	ParticipantID livekit.ParticipantID
	TrackID       livekit.TrackID
	State         StreamState
}

type StreamStateUpdate 

type StreamStateUpdate struct {
	StreamStates []*StreamStateInfo
}

func NewStreamStateUpdate 

func NewStreamStateUpdate() *StreamStateUpdate

func (*StreamStateUpdate) Empty 

func (s *StreamStateUpdate) Empty() bool

func (*StreamStateUpdate) HandleStreamingChange 

func (s *StreamStateUpdate) HandleStreamingChange(change VideoStreamingChange, track *Track)

type StreamStatus 

type StreamStatus int32
const (
	StreamStatusStopped StreamStatus = 0
	StreamStatusActive  StreamStatus = 1
)

func (StreamStatus) String 

func (s StreamStatus) String() string

type StreamTracker 

type StreamTracker struct {
	// contains filtered or unexported fields
}

StreamTracker keeps track of packet flow and ensures a particular up track is consistently producing It runs its own goroutine for detection, and fires OnStatusChanged callback

func NewStreamTracker 

func NewStreamTracker(params StreamTrackerParams) *StreamTracker

func (*StreamTracker) BitrateTemporalCumulative 

func (s *StreamTracker) BitrateTemporalCumulative() [4]int64

BitrateTemporalCumulative returns the current stream bitrate temporal layer accumulated with lower temporal layers.

func (*StreamTracker) Observe 

func (s *StreamTracker) Observe(sn uint16, temporalLayer int32, pktSize int)

Observe a packet that's received

func (*StreamTracker) OnBitrateAvailable 

func (s *StreamTracker) OnBitrateAvailable(f func())

func (*StreamTracker) OnStatusChanged 

func (s *StreamTracker) OnStatusChanged(f func(status StreamStatus))

func (*StreamTracker) Reset 

func (s *StreamTracker) Reset()

func (*StreamTracker) SetPaused 

func (s *StreamTracker) SetPaused(paused bool)

func (*StreamTracker) Start 

func (s *StreamTracker) Start()

func (*StreamTracker) Status 

func (s *StreamTracker) Status() StreamStatus

func (*StreamTracker) Stop 

func (s *StreamTracker) Stop()

type StreamTrackerManager 

type StreamTrackerManager struct {
	// contains filtered or unexported fields
}

func NewStreamTrackerManager 

func NewStreamTrackerManager(logger logger.Logger, source livekit.TrackSource) *StreamTrackerManager

func (*StreamTrackerManager) AddTracker 

func (s *StreamTrackerManager) AddTracker(layer int32)

func (*StreamTrackerManager) GetAvailableLayers 

func (s *StreamTrackerManager) GetAvailableLayers() []int32

func (*StreamTrackerManager) GetBitrateTemporalCumulative 

func (s *StreamTrackerManager) GetBitrateTemporalCumulative() Bitrates

func (*StreamTrackerManager) GetTracker 

func (s *StreamTrackerManager) GetTracker(layer int32) *StreamTracker

func (*StreamTrackerManager) HasSpatialLayer 

func (s *StreamTrackerManager) HasSpatialLayer(layer int32) bool

func (*StreamTrackerManager) IsReducedQuality 

func (s *StreamTrackerManager) IsReducedQuality() bool

func (*StreamTrackerManager) OnAvailableLayersChanged 

func (s *StreamTrackerManager) OnAvailableLayersChanged(f func(availableLayers []int32))

func (*StreamTrackerManager) OnBitrateAvailabilityChanged 

func (s *StreamTrackerManager) OnBitrateAvailabilityChanged(f func())

func (*StreamTrackerManager) RemoveAllTrackers 

func (s *StreamTrackerManager) RemoveAllTrackers()

func (*StreamTrackerManager) RemoveTracker 

func (s *StreamTrackerManager) RemoveTracker(layer int32)

func (*StreamTrackerManager) SetMaxExpectedSpatialLayer 

func (s *StreamTrackerManager) SetMaxExpectedSpatialLayer(layer int32)

func (*StreamTrackerManager) SetPaused 

func (s *StreamTrackerManager) SetPaused(paused bool)

type StreamTrackerParams 

type StreamTrackerParams struct {
	// number of samples needed per cycle
	SamplesRequired uint32

	// number of cycles needed to be active
	CyclesRequired uint32

	CycleDuration time.Duration

	Logger logger.Logger
}

type Track 

type Track struct {
	// contains filtered or unexported fields
}

func (*Track) AllocateNextHigher 

func (t *Track) AllocateNextHigher(availableChannelCapacity int64) (VideoAllocation, bool)

func (*Track) AllocateOptimal 

func (t *Track) AllocateOptimal() VideoAllocation

func (*Track) BandwidthRequested 

func (t *Track) BandwidthRequested() int64

func (*Track) DistanceToDesired 

func (t *Track) DistanceToDesired() int32

func (*Track) DownTrack 

func (t *Track) DownTrack() *DownTrack

func (*Track) GetNackDelta 

func (t *Track) GetNackDelta() (uint32, uint32)

func (*Track) GetNextHigherTransition 

func (t *Track) GetNextHigherTransition() (VideoTransition, bool)

func (*Track) ID 

func (t *Track) ID() livekit.TrackID

func (*Track) IsDeficient 

func (t *Track) IsDeficient() bool

func (*Track) IsManaged 

func (t *Track) IsManaged() bool

func (*Track) Pause 

func (t *Track) Pause() VideoAllocation

func (*Track) Priority 

func (t *Track) Priority() uint8

func (*Track) ProvisionalAllocate 

func (t *Track) ProvisionalAllocate(availableChannelCapacity int64, layers VideoLayers, allowPause bool) int64

func (*Track) ProvisionalAllocateCommit 

func (t *Track) ProvisionalAllocateCommit() VideoAllocation

func (*Track) ProvisionalAllocateGetBestWeightedTransition 

func (t *Track) ProvisionalAllocateGetBestWeightedTransition() VideoTransition

func (*Track) ProvisionalAllocateGetCooperativeTransition 

func (t *Track) ProvisionalAllocateGetCooperativeTransition() VideoTransition

func (*Track) ProvisionalAllocatePrepare 

func (t *Track) ProvisionalAllocatePrepare()

func (*Track) PublisherID 

func (t *Track) PublisherID() livekit.ParticipantID

func (*Track) SetPriority 

func (t *Track) SetPriority(priority uint8) bool

func (*Track) UpdateMaxLayers 

func (t *Track) UpdateMaxLayers(layers VideoLayers)

func (*Track) WritePaddingRTP 

func (t *Track) WritePaddingRTP(bytesToSend int) int

type TrackReceiver 

type TrackReceiver interface {
	TrackID() livekit.TrackID
	StreamID() string
	Codec() webrtc.RTPCodecCapability

	ReadRTP(buf []byte, layer uint8, sn uint16) (int, error)
	GetBitrateTemporalCumulative() Bitrates

	SendPLI(layer int32)

	SetUpTrackPaused(paused bool)
	SetMaxExpectedSpatialLayer(layer int32)

	AddDownTrack(track TrackSender) error
	DeleteDownTrack(peerID livekit.ParticipantID)

	DebugInfo() map[string]interface{}
}

TrackReceiver defines an interface receive media from remote peer

type TrackSender 

type TrackSender interface {
	UpTrackLayersChange(availableLayers []int32)
	UpTrackBitrateAvailabilityChange()
	WriteRTP(p *buffer.ExtPacket, layer int32) error
	Close()
	// ID is the globally unique identifier for this Track.
	ID() string
	Codec() webrtc.RTPCodecCapability
	PeerID() livekit.ParticipantID
}

TrackSender defines an interface send media to remote peer

type TrackSorter 

type TrackSorter []*Track

func (TrackSorter) Len 

func (t TrackSorter) Len() int

func (TrackSorter) Less 

func (t TrackSorter) Less(i, j int) bool

func (TrackSorter) Swap 

func (t TrackSorter) Swap(i, j int)

type TranslationParams 

type TranslationParams struct {
	// contains filtered or unexported fields
}

type TranslationParamsRTP 

type TranslationParamsRTP struct {
	// contains filtered or unexported fields
}

type TranslationParamsVP8 

type TranslationParamsVP8 struct {
	// contains filtered or unexported fields
}

VP8 munger

type TrendDetector 

type TrendDetector struct {
	// contains filtered or unexported fields
}

func NewTrendDetector 

func NewTrendDetector(name string, logger logger.Logger, requiredSamples int) *TrendDetector

func (*TrendDetector) AddValue 

func (t *TrendDetector) AddValue(value int64)

func (*TrendDetector) GetDirection 

func (t *TrendDetector) GetDirection() TrendDirection

func (*TrendDetector) GetHighest 

func (t *TrendDetector) GetHighest() int64

func (*TrendDetector) GetLowest 

func (t *TrendDetector) GetLowest() int64

func (*TrendDetector) GetValues 

func (t *TrendDetector) GetValues() []int64

func (*TrendDetector) Reset 

func (t *TrendDetector) Reset()

func (*TrendDetector) Seed 

func (t *TrendDetector) Seed(value int64)

type TrendDirection 

type TrendDirection int
const (
	TrendDirectionNeutral TrendDirection = iota
	TrendDirectionUpward
	TrendDirectionDownward
)

func (TrendDirection) String 

func (t TrendDirection) String() string

type VP8Munger 

type VP8Munger struct {
	VP8MungerParams
	// contains filtered or unexported fields
}

func NewVP8Munger 

func NewVP8Munger(logger logger.Logger) *VP8Munger

func (*VP8Munger) PictureIdOffset 

func (v *VP8Munger) PictureIdOffset(extPictureId int32) (int32, bool)

for testing only

func (*VP8Munger) SetLast 

func (v *VP8Munger) SetLast(extPkt *buffer.ExtPacket)

func (*VP8Munger) UpdateAndGet 

func (v *VP8Munger) UpdateAndGet(extPkt *buffer.ExtPacket, ordering SequenceNumberOrdering, maxTemporalLayer int32) (*TranslationParamsVP8, error)

func (*VP8Munger) UpdateAndGetPadding 

func (v *VP8Munger) UpdateAndGetPadding(newPicture bool) *buffer.VP8

func (*VP8Munger) UpdateOffsets 

func (v *VP8Munger) UpdateOffsets(extPkt *buffer.ExtPacket)

type VP8MungerParams 

type VP8MungerParams struct {
	// contains filtered or unexported fields
}

type VP8PictureIdWrapHandler 

type VP8PictureIdWrapHandler struct {
	// contains filtered or unexported fields
}

func (*VP8PictureIdWrapHandler) Init 

func (v *VP8PictureIdWrapHandler) Init(extPictureId int32, mBit bool)

func (*VP8PictureIdWrapHandler) MaxPictureId 

func (v *VP8PictureIdWrapHandler) MaxPictureId() int32

func (*VP8PictureIdWrapHandler) Unwrap 

func (v *VP8PictureIdWrapHandler) Unwrap(pictureId uint16, mBit bool) int32

unwrap picture id and update the maxPictureId. return unwrapped value

func (*VP8PictureIdWrapHandler) UpdateMaxPictureId 

func (v *VP8PictureIdWrapHandler) UpdateMaxPictureId(extPictureId int32, mBit bool)

type VideoAllocation 

type VideoAllocation struct {
	// contains filtered or unexported fields
}

func (VideoAllocation) String 

func (v VideoAllocation) String() string

type VideoAllocationProvisional 

type VideoAllocationProvisional struct {
	// contains filtered or unexported fields
}

type VideoAllocationState 

type VideoAllocationState int
const (
	VideoAllocationStateNone VideoAllocationState = iota
	VideoAllocationStateMuted
	VideoAllocationStateFeedDry
	VideoAllocationStateAwaitingMeasurement
	VideoAllocationStateOptimal
	VideoAllocationStateDeficient
)

func (VideoAllocationState) String 

func (v VideoAllocationState) String() string

type VideoLayers 

type VideoLayers struct {
	// contains filtered or unexported fields
}

func (VideoLayers) GreaterThan 

func (v VideoLayers) GreaterThan(v2 VideoLayers) bool

func (VideoLayers) SpatialGreaterThanOrEqual 

func (v VideoLayers) SpatialGreaterThanOrEqual(v2 VideoLayers) bool

func (VideoLayers) String 

func (v VideoLayers) String() string

type VideoStreamingChange 

type VideoStreamingChange int
const (
	VideoStreamingChangeNone VideoStreamingChange = iota
	VideoStreamingChangePausing
	VideoStreamingChangeResuming
)

func (VideoStreamingChange) String 

func (v VideoStreamingChange) String() string

type VideoTransition 

type VideoTransition struct {
	// contains filtered or unexported fields
}

type WebRTCReceiver 

type WebRTCReceiver struct {
	// contains filtered or unexported fields
}

WebRTCReceiver receives a media track

func NewWebRTCReceiver 

func NewWebRTCReceiver(
	receiver *webrtc.RTPReceiver,
	track *webrtc.TrackRemote,
	pid livekit.ParticipantID,
	source livekit.TrackSource,
	logger logger.Logger,
	opts ...ReceiverOpts,
) *WebRTCReceiver

NewWebRTCReceiver creates a new webrtc track receiver

func (*WebRTCReceiver) AddDownTrack 

func (w *WebRTCReceiver) AddDownTrack(track TrackSender) error

func (*WebRTCReceiver) AddUpTrack 

func (w *WebRTCReceiver) AddUpTrack(track *webrtc.TrackRemote, buff *buffer.Buffer)

func (*WebRTCReceiver) Codec 

func (w *WebRTCReceiver) Codec() webrtc.RTPCodecCapability

func (*WebRTCReceiver) DebugInfo 

func (w *WebRTCReceiver) DebugInfo() map[string]interface{}

func (*WebRTCReceiver) DeleteDownTrack 

func (w *WebRTCReceiver) DeleteDownTrack(peerID livekit.ParticipantID)

DeleteDownTrack removes a DownTrack from a Receiver

func (*WebRTCReceiver) GetBitrateTemporalCumulative 

func (w *WebRTCReceiver) GetBitrateTemporalCumulative() Bitrates

func (*WebRTCReceiver) GetConnectionScore 

func (w *WebRTCReceiver) GetConnectionScore() float32

func (*WebRTCReceiver) GetTrackStats 

func (w *WebRTCReceiver) GetTrackStats() map[uint32]*buffer.StreamStatsWithLayers

func (*WebRTCReceiver) Kind 

func (w *WebRTCReceiver) Kind() webrtc.RTPCodecType

func (*WebRTCReceiver) OnCloseHandler 

func (w *WebRTCReceiver) OnCloseHandler(fn func())

OnCloseHandler method to be called on remote tracked removed

func (*WebRTCReceiver) OnStatsUpdate 

func (w *WebRTCReceiver) OnStatsUpdate(fn func(w *WebRTCReceiver, stat *livekit.AnalyticsStat))

func (*WebRTCReceiver) ReadRTP 

func (w *WebRTCReceiver) ReadRTP(buf []byte, layer uint8, sn uint16) (int, error)

func (*WebRTCReceiver) SSRC 

func (w *WebRTCReceiver) SSRC(layer int) uint32

func (*WebRTCReceiver) SendPLI 

func (w *WebRTCReceiver) SendPLI(layer int32)

func (*WebRTCReceiver) SetMaxExpectedSpatialLayer 

func (w *WebRTCReceiver) SetMaxExpectedSpatialLayer(layer int32)

func (*WebRTCReceiver) SetRTCPCh 

func (w *WebRTCReceiver) SetRTCPCh(ch chan []rtcp.Packet)

func (*WebRTCReceiver) SetRTT 

func (w *WebRTCReceiver) SetRTT(rtt uint32)

func (*WebRTCReceiver) SetTrackMeta 

func (w *WebRTCReceiver) SetTrackMeta(trackID livekit.TrackID, streamID string)

func (*WebRTCReceiver) SetUpTrackPaused 

func (w *WebRTCReceiver) SetUpTrackPaused(paused bool)

SetUpTrackPaused indicates upstream will not be sending any data. this will reflect the "muted" status and will pause streamtracker to ensure we don't turn off the layer

func (*WebRTCReceiver) StreamID 

func (w *WebRTCReceiver) StreamID() string

func (*WebRTCReceiver) TrackID 

func (w *WebRTCReceiver) TrackID() livekit.TrackID

Source Files

View all Source files

Directories

Path Synopsis

Jump to

Keyboard shortcuts

? : This menu
/ : Search site
f or F : Jump to
y or Y : Canonical URL
go.dev uses cookies from Google to deliver and enhance the quality of its services and to analyze traffic. Learn more.