diff --git a/pkg/sfu/bwe/remotebwe/remote_bwe.go b/pkg/sfu/bwe/remotebwe/remote_bwe.go index a868afa9e..e7e36ed46 100644 --- a/pkg/sfu/bwe/remotebwe/remote_bwe.go +++ b/pkg/sfu/bwe/remotebwe/remote_bwe.go @@ -278,7 +278,7 @@ func (r *RemoteBWE) newChannelObserver() { }, r.params.Logger, ) - r.channelObserver.SeedEstimate(r.lastReceivedEstimate) + r.channelObserver.SeedEstimate(r.committedChannelCapacity) } else { r.channelObserver = newChannelObserver( channelObserverParams{ @@ -321,6 +321,7 @@ func (r *RemoteBWE) ProbeClusterDone(_pci ccutils.ProbeClusterInfo) (bool, int64 "lastReceived", r.lastReceivedEstimate, "expectedBandwidthUsage", r.lastExpectedBandwidthUsage, "channel", pco, + "isSignalValid", pco.HasEnoughEstimateSamples(), ) if !pco.HasEnoughEstimateSamples() { diff --git a/pkg/sfu/streamallocator/probe_controller.go b/pkg/sfu/streamallocator/probe_controller.go index 42f44be94..40ed41e61 100644 --- a/pkg/sfu/streamallocator/probe_controller.go +++ b/pkg/sfu/streamallocator/probe_controller.go @@ -19,9 +19,7 @@ import ( "sync" "time" - "github.com/livekit/livekit-server/pkg/sfu/bwe" "github.com/livekit/livekit-server/pkg/sfu/ccutils" - "github.com/livekit/livekit-server/pkg/sfu/pacer" "github.com/livekit/protocol/logger" "github.com/livekit/protocol/utils/mono" ) @@ -96,9 +94,6 @@ var ( type ProbeControllerParams struct { Config ProbeControllerConfig - Prober *ccutils.Prober - BWE bwe.BWE - Pacer pacer.Pacer Logger logger.Logger } @@ -201,18 +196,10 @@ func (p *ProbeController) ProbeClusterDone(pci ccutils.ProbeClusterInfo) { p.lock.Lock() defer p.lock.Unlock() - if p.state != ProbeControllerStateProbing { - p.params.Logger.Warnw("unexpected probe controller state", nil, "state", p.state) + if p.pci.Id == pci.Id { + p.pci.Result = pci.Result + p.setState(ProbeControllerStateHangover) } - - if p.pci.Id != pci.Id { - p.params.Logger.Warnw("not expected probe cluster", nil, "expectedId", p.pci.Id, "doneId", pci.Id) - } - - p.pci.Result = pci.Result - p.params.Prober.ClusterDone(pci) - - p.setState(ProbeControllerStateHangover) } func (p *ProbeController) MaybeFinalizeProbe() (ccutils.ProbeClusterInfo, bool) { diff --git a/pkg/sfu/streamallocator/streamallocator.go b/pkg/sfu/streamallocator/streamallocator.go index 67959906e..0b0ce7905 100644 --- a/pkg/sfu/streamallocator/streamallocator.go +++ b/pkg/sfu/streamallocator/streamallocator.go @@ -236,9 +236,6 @@ func NewStreamAllocator(params StreamAllocatorParams, enabled bool, allowPause b s.probeController = NewProbeController(ProbeControllerParams{ Config: s.params.Config.ProbeController, - Prober: s.prober, - BWE: s.params.BWE, - Pacer: s.params.Pacer, Logger: params.Logger, }) @@ -769,6 +766,7 @@ func (s *StreamAllocator) handleSignalPacerProbeObserverClusterComplete(event Ev probeClusterId, _ := event.Data.(ccutils.ProbeClusterId) pci := s.params.Pacer.EndProbeCluster(probeClusterId) s.probeController.ProbeClusterDone(pci) + s.prober.ClusterDone(pci) } func (s *StreamAllocator) handleSignalResume(event Event) { @@ -1069,8 +1067,8 @@ func (s *StreamAllocator) maybeStopProbe() { activeProbeClusterId := s.probeController.GetActiveProbeClusterId() if activeProbeClusterId != ccutils.ProbeClusterIdInvalid { pci := s.params.Pacer.EndProbeCluster(activeProbeClusterId) - s.prober.Reset(pci) s.probeController.ProbeClusterDone(pci) + s.prober.Reset(pci) } }