Bit more clean up around probe controller refactor (#3227)

* Bit more clean up around probe controller refactor

* consistent order

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() {

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) {

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)
 	}
 }