dandri/livekit
1
0
Fork 0
mirror of https://github.com/livekit/livekit.git synced 2026-08-23 01:09:46 +00:00
Code Issues Packages Projects Releases Wiki Activity

Add method to check for probe goal reached (#3253)

Browse Source 
* Check for early probe end

* use old packets for probing

* fmt
This commit is contained in:
Raja Subramanian
2024-12-15 23:34:07 +05:30
committed by GitHub
parent cfe3178542
commit be65d24333
8 changed files with 54 additions and 22 deletions
Download Patch File Download Diff File Expand all files Collapse all files
pkg/sfu/bwe/bwe.go
+1
View File
@@ -89,6 +89,7 @@ type BWE interface {
ProbeDuration() time.Duration
ProbeClusterStarting(pci ccutils.ProbeClusterInfo)
ProbeClusterDone(pci ccutils.ProbeClusterInfo)
ProbeClusterIsGoalReached() bool
ProbeClusterFinalize() (ccutils.ProbeSignal, int64, bool)
}
pkg/sfu/bwe/null_bwe.go
+4
View File
@@ -62,6 +62,10 @@ func (n *NullBWE) ProbeClusterStarting(_pci ccutils.ProbeClusterInfo) {}
func (n *NullBWE) ProbeClusterDone(_pci ccutils.ProbeClusterInfo) {}
func (n *NullBWE) ProbeClusterIsGoalReached() bool {
return false
}
func (n *NullBWE) ProbeClusterFinalize() (ccutils.ProbeSignal, int64, bool) {
return ccutils.ProbeSignalInconclusive, 0, false
}
pkg/sfu/bwe/remotebwe/probe_controller.go
+6 -2
View File
@@ -62,8 +62,8 @@ var (
ProbeRegulator: ccutils.DefaultProbeRegulatorConfig,
SettleWaitNumRTT: 5,
SettleWaitMin: 500 * time.Millisecond,
SettleWaitMax: 10 * time.Second,
SettleWaitMin: 250 * time.Millisecond,
SettleWaitMax: 5 * time.Second,
}
)
@@ -140,6 +140,10 @@ func (p *probeController) ProbeClusterDone(pci ccutils.ProbeClusterInfo) {
p.setState(probeControllerStateHangover)
}
func (p *probeController) ProbeClusterInfo() ccutils.ProbeClusterInfo {
return p.pci
}
func (p *probeController) MaybeFinalizeProbe() (ccutils.ProbeClusterInfo, bool) {
if p.state != probeControllerStateHangover {
return ccutils.ProbeClusterInfoInvalid, false
pkg/sfu/bwe/remotebwe/remote_bwe.go
+13 -5
View File
@@ -158,11 +158,6 @@ func (r *RemoteBWE) congestionDetectionStateMachine() (bool, bwe.CongestionState
newState := r.congestionState
update := false
trend, reason := r.channelObserver.GetTrend()
if trend == channelTrendCongesting && r.congestionState == bwe.CongestionStateNone {
r.params.Logger.Debugw("remote bwe, channel congesting", "channel", r.channelObserver)
} else if trend == channelTrendClearing && r.congestionState != bwe.CongestionStateNone {
r.params.Logger.Debugw("remote bwe, channel congestion relieving", "channel", r.channelObserver)
}
switch r.congestionState {
case bwe.CongestionStateNone:
@@ -285,6 +280,19 @@ func (r *RemoteBWE) ProbeClusterDone(pci ccutils.ProbeClusterInfo) {
r.probeController.ProbeClusterDone(pci)
}
func (r *RemoteBWE) ProbeClusterIsGoalReached() bool {
r.lock.Lock()
defer r.lock.Unlock()
if !r.probeController.IsInProbe() ||
r.congestionState != bwe.CongestionStateNone ||
!r.channelObserver.HasEnoughEstimateSamples() {
return false
}
return r.channelObserver.GetHighestEstimate() > int64(r.probeController.ProbeClusterInfo().Goal.DesiredBps)
}
func (r *RemoteBWE) ProbeClusterFinalize() (ccutils.ProbeSignal, int64, bool) {
r.lock.Lock()
defer r.lock.Unlock()
pkg/sfu/bwe/sendsidebwe/probe_packet_group.go
+2 -2
View File
@@ -42,8 +42,8 @@ var (
},
SettleWaitNumRTT: 5,
SettleWaitMin: 500 * time.Millisecond,
SettleWaitMax: 10 * time.Second,
SettleWaitMin: 250 * time.Millisecond,
SettleWaitMax: 5 * time.Second,
}
)
pkg/sfu/ccutils/prober.go
-1
View File
@@ -578,7 +578,6 @@ func (c *Cluster) MarshalLogObject(e zapcore.ObjectEncoder) error {
e.AddObject("info", c.info)
e.AddDuration("baseSleepDuration", c.baseSleepDuration)
e.AddInt("numBuckets", len(c.buckets))
e.AddArray("buckets", logger.ObjectSlice(c.buckets))
e.AddInt("bucketIdx", c.bucketIdx)
e.AddInt("probeBytesSent", c.probeBytesSent)
e.AddTime("startTime", c.startTime)
pkg/sfu/streamallocator/streamallocator.go
+27 -11
View File
@@ -202,9 +202,11 @@ type StreamAllocator struct {
isAllocateAllPending bool
rembTrackingSSRC uint32
state streamAllocatorState
isHolding bool
activeProbeClusterId ccutils.ProbeClusterId
state streamAllocatorState
isHolding bool
activeProbeClusterId ccutils.ProbeClusterId
activeProbeGoalReached bool
activeProbeCongesting bool
eventsQueue *utils.TypedOpsQueue[Event]
@@ -679,12 +681,21 @@ func (s *StreamAllocator) handleSignalPeriodicPing(Event) {
s.params.BWE.Reset()
}
// finalize any probe that may have finished/aborted
if s.activeProbeClusterId != ccutils.ProbeClusterIdInvalid {
if !s.activeProbeCongesting && !s.activeProbeGoalReached && s.params.BWE.ProbeClusterIsGoalReached() {
s.params.Logger.Debugw(
"stream allocator: probe goal reached",
"activeProbeClusterId", s.activeProbeClusterId,
)
s.activeProbeGoalReached = true
s.maybeStopProbe()
}
// finalize any probe that may have finished/aborted
if probeSignal, channelCapacity, isFinalized := s.params.BWE.ProbeClusterFinalize(); isFinalized {
s.params.Logger.Debugw(
"stream allocator: probe result",
"probeClusterId", s.activeProbeClusterId,
"activeProbeClusterId", s.activeProbeClusterId,
"probeSignal", probeSignal,
"channelCapacity", channelCapacity,
)
@@ -720,6 +731,8 @@ func (s *StreamAllocator) handleSignalPeriodicPing(Event) {
func (s *StreamAllocator) handleSignalProbeClusterSwitch(event Event) {
pci := event.Data.(ccutils.ProbeClusterInfo)
s.activeProbeClusterId = pci.Id
s.activeProbeGoalReached = false
s.activeProbeCongesting = false
s.params.BWE.ProbeClusterStarting(pci)
@@ -817,12 +830,15 @@ func (s *StreamAllocator) handleSignalCongestionStateChange(event Event) {
if cscd.congestionState == bwe.CongestionStateCongested {
if s.activeProbeClusterId != ccutils.ProbeClusterIdInvalid {
s.params.Logger.Infow(
"stream allocator: channel congestion detected, not updating channel capacity in active probe",
"old(bps)", s.committedChannelCapacity,
"new(bps)", cscd.estimatedAvailableChannelCapacity,
"expectedUsage(bps)", s.getExpectedBandwidthUsage(),
)
if !s.activeProbeCongesting {
s.activeProbeCongesting = true
s.params.Logger.Infow(
"stream allocator: channel congestion detected, not updating channel capacity in active probe",
"old(bps)", s.committedChannelCapacity,
"new(bps)", cscd.estimatedAvailableChannelCapacity,
"expectedUsage(bps)", s.getExpectedBandwidthUsage(),
)
}
} else {
s.params.Logger.Infow(
"stream allocator: channel congestion detected, updating channel capacity",
pkg/sfu/streamallocator/track.go
+1 -1
View File
@@ -135,7 +135,7 @@ func (t *Track) WritePaddingRTP(bytesToSend int) int {
}
func (t *Track) WriteProbePackets(bytesToSend int) int {
return t.downTrack.WriteProbePackets(bytesToSend, true)
return t.downTrack.WriteProbePackets(bytesToSend, false)
}
func (t *Track) AllocateOptimal(allowOvershoot bool, hold bool) sfu.VideoAllocation {