From 0e7bdeabcb10679e951a96b880669651a0160313 Mon Sep 17 00:00:00 2001
From: Raja Subramanian <raja.gobi@tutanota.com>
Date: Sat, 10 Jun 2023 02:07:28 +0530
Subject: [PATCH] Simplify probe done handling. (#1782)

* Simplify probe done handling.

Seeing a case where the channel abserver is not re-created after
an aborted probe. Simplifying probe done (no callbacks, making it
synchronous).

* log more
---
 pkg/sfu/streamallocator/probe_controller.go | 52 +++++++++------------
 pkg/sfu/streamallocator/prober.go           |  2 +-
 pkg/sfu/streamallocator/streamallocator.go  | 13 ++++--
 3 files changed, 31 insertions(+), 36 deletions(-)

diff --git a/pkg/sfu/streamallocator/probe_controller.go b/pkg/sfu/streamallocator/probe_controller.go
index edb0dfefe..519d31832 100644
--- a/pkg/sfu/streamallocator/probe_controller.go
+++ b/pkg/sfu/streamallocator/probe_controller.go
@@ -38,8 +38,6 @@ type ProbeController struct {
 	abortedProbeClusterId ProbeClusterId
 	probeTrendObserved    bool
 	probeEndTime          time.Time
-
-	onProbeDone func(isSuccessful bool)
 }
 
 func NewProbeController(params ProbeControllerParams) *ProbeController {
@@ -51,13 +49,6 @@ func NewProbeController(params ProbeControllerParams) *ProbeController {
 	return p
 }
 
-func (p *ProbeController) OnProbeDone(f func(isSuccessful bool)) {
-	p.lock.Lock()
-	defer p.lock.Unlock()
-
-	p.onProbeDone = f
-}
-
 func (p *ProbeController) Reset() {
 	p.lock.Lock()
 	defer p.lock.Unlock()
@@ -69,23 +60,18 @@ func (p *ProbeController) Reset() {
 	p.clearProbeLocked()
 }
 
-func (p *ProbeController) ProbeClusterDone(info ProbeClusterInfo, lowestEstimate int64) {
+func (p *ProbeController) ProbeClusterDone(info ProbeClusterInfo, lowestEstimate int64) bool {
 	p.lock.Lock()
+	defer p.lock.Unlock()
+
 	if p.probeClusterId != info.Id {
-		p.lock.Unlock()
-		return
+		p.params.Logger.Infow("not expected probe cluster", "probeClusterId", p.probeClusterId, "resetProbeClusterId", info.Id)
+		return false
 	}
 
 	if p.abortedProbeClusterId == ProbeClusterIdInvalid {
 		// successful probe, finalize
-		isSuccessful := p.finalizeProbeLocked()
-		onProbeDone := p.onProbeDone
-		p.lock.Unlock()
-
-		if onProbeDone != nil {
-			onProbeDone(isSuccessful)
-		}
-		return
+		return p.finalizeProbeLocked()
 	}
 
 	// ensure probe queue is flushed
@@ -97,7 +83,15 @@ func (p *ProbeController) ProbeClusterDone(info ProbeClusterInfo, lowestEstimate
 	}
 	queueWait := time.Duration(queueTime+float64(ProbeSettleWait)) * time.Millisecond
 	p.probeEndTime = p.lastProbeStartTime.Add(queueWait)
-	p.lock.Unlock()
+	p.params.Logger.Infow(
+		"setting probe end time",
+		"probeClusterId", p.probeClusterId,
+		"expectedDuration", expectedDuration,
+		"queueTime", queueTime,
+		"queueWait", queueWait,
+		"probeEndTime", p.probeEndTime,
+	)
+	return false
 }
 
 func (p *ProbeController) CheckProbe(trend ChannelTrend, highestEstimate int64) {
@@ -139,19 +133,15 @@ func (p *ProbeController) CheckProbe(trend ChannelTrend, highestEstimate int64)
 	}
 }
 
-func (p *ProbeController) MaybeFinalizeProbe() {
+func (p *ProbeController) MaybeFinalizeProbe() (isHandled bool, isSuccessful bool) {
 	p.lock.Lock()
-	var onProbeDone func(bool)
-	isSuccessful := false
-	if p.isInProbeLocked() && !p.probeEndTime.IsZero() && time.Now().After(p.probeEndTime) {
-		isSuccessful = p.finalizeProbeLocked()
-		onProbeDone = p.onProbeDone
-	}
-	p.lock.Unlock()
+	defer p.lock.Unlock()
 
-	if onProbeDone != nil {
-		onProbeDone(isSuccessful)
+	if p.isInProbeLocked() && !p.probeEndTime.IsZero() && time.Now().After(p.probeEndTime) {
+		return true, p.finalizeProbeLocked()
 	}
+
+	return false, false
 }
 
 func (p *ProbeController) DoesProbeNeedFinalize() bool {
diff --git a/pkg/sfu/streamallocator/prober.go b/pkg/sfu/streamallocator/prober.go
index a1b1a3205..66ef47b8a 100644
--- a/pkg/sfu/streamallocator/prober.go
+++ b/pkg/sfu/streamallocator/prober.go
@@ -175,7 +175,7 @@ func (p *Prober) Reset() {
 
 	p.clustersMu.Lock()
 	if p.activeCluster != nil {
-		p.logger.Debugw("resetting active cluster", "cluster", p.activeCluster.String())
+		p.logger.Infow("prober: resetting active cluster", "cluster", p.activeCluster.String())
 		reset = true
 		info = p.activeCluster.GetInfo()
 	}
diff --git a/pkg/sfu/streamallocator/streamallocator.go b/pkg/sfu/streamallocator/streamallocator.go
index edea5051c..d13141e7d 100644
--- a/pkg/sfu/streamallocator/streamallocator.go
+++ b/pkg/sfu/streamallocator/streamallocator.go
@@ -200,7 +200,6 @@ func NewStreamAllocator(params StreamAllocatorParams) *StreamAllocator {
 		Prober: s.prober,
 		Logger: params.Logger,
 	})
-	s.probeController.OnProbeDone(s.onProbeDone)
 
 	s.resetState()
 
@@ -492,7 +491,7 @@ func (s *StreamAllocator) OnSendProbe(bytesToSend int) {
 	})
 }
 
-// called when prober wants to send packet(s)
+// called when prober finishes a probe cluster, could be called when prober is reset which stops an active cluster
 func (s *StreamAllocator) OnProbeClusterDone(info ProbeClusterInfo) {
 	s.postEvent(Event{
 		Signal: streamAllocatorSignalProbeClusterDone,
@@ -640,7 +639,10 @@ func (s *StreamAllocator) handleSignalEstimate(event *Event) {
 
 func (s *StreamAllocator) handleSignalPeriodicPing(event *Event) {
 	// finalize probe if necessary
-	s.probeController.MaybeFinalizeProbe()
+	isHandled, isSuccessful := s.probeController.MaybeFinalizeProbe()
+	if isHandled {
+		s.onProbeDone(isSuccessful)
+	}
 
 	// probe if necessary and timing is right
 	if s.state == streamAllocatorStateDeficient {
@@ -673,7 +675,10 @@ func (s *StreamAllocator) handleSignalSendProbe(event *Event) {
 
 func (s *StreamAllocator) handleSignalProbeClusterDone(event *Event) {
 	info, _ := event.Data.(ProbeClusterInfo)
-	s.probeController.ProbeClusterDone(info, int64(math.Min(float64(s.committedChannelCapacity), float64(s.channelObserver.GetLowestEstimate()))))
+	isHandled := s.probeController.ProbeClusterDone(info, int64(math.Min(float64(s.committedChannelCapacity), float64(s.channelObserver.GetLowestEstimate()))))
+	if isHandled {
+		s.onProbeDone(true)
+	}
 }
 
 func (s *StreamAllocator) handleSignalResume(event *Event) {