Throttler teardown

Author: nikhilmat

go/logic/throttler.go

@@ -41,16 +41,18 @@ const frenoMagicHint = "freno"
 // Throttler collects metrics related to throttling and makes informed decision
 // whether throttling should take place.
 type Throttler struct {
-	migrationContext *base.MigrationContext
-	applier          *Applier
-	inspector        *Inspector
+	migrationContext  *base.MigrationContext
+	applier           *Applier
+	inspector         *Inspector
+	finishedMigrating int64
 }
 
 func NewThrottler(migrationContext *base.MigrationContext, applier *Applier, inspector *Inspector) *Throttler {
 	return &Throttler{
-		migrationContext: migrationContext,
-		applier:          applier,
-		inspector:        inspector,
+		migrationContext:  migrationContext,
+		applier:           applier,
+		inspector:         inspector,
+		finishedMigrating: 0,
 	}
 }
 
@@ -159,6 +161,9 @@ func (this *Throttler) collectReplicationLag(firstThrottlingCollected chan<- boo
 
 	ticker := time.Tick(time.Duration(this.migrationContext.HeartbeatIntervalMilliseconds) * time.Millisecond)
 	for range ticker {
+		if atomic.LoadInt64(&this.finishedMigrating) > 0 {
+			return
+		}
 		go collectFunc()
 	}
 }
@@ -233,6 +238,9 @@ func (this *Throttler) collectControlReplicasLag() {
 	shouldReadLagAggressively := false
 
 	for range aggressiveTicker {
+		if atomic.LoadInt64(&this.finishedMigrating) > 0 {
+			return
+		}
 		if counter%relaxedFactor == 0 {
 			// we only check if we wish to be aggressive once per second. The parameters for being aggressive
 			// do not typically change at all throughout the migration, but nonetheless we check them.
@@ -285,6 +293,10 @@ func (this *Throttler) collectThrottleHTTPStatus(firstThrottlingCollected chan<-
 
 	ticker := time.Tick(100 * time.Millisecond)
 	for range ticker {
+		if atomic.LoadInt64(&this.finishedMigrating) > 0 {
+			return
+		}
+
 		if sleep, _ := collectFunc(); sleep {
 			time.Sleep(1 * time.Second)
 		}
@@ -393,6 +405,10 @@ func (this *Throttler) initiateThrottlerCollection(firstThrottlingCollected chan
 
 		throttlerMetricsTick := time.Tick(1 * time.Second)
 		for range throttlerMetricsTick {
+			if atomic.LoadInt64(&this.finishedMigrating) > 0 {
+				return
+			}
+
 			this.collectGeneralThrottleMetrics()
 		}
 	}()
@@ -419,6 +435,9 @@ func (this *Throttler) initiateThrottlerChecks() error {
 	}
 	throttlerFunction()
 	for range throttlerTick {
+		if atomic.LoadInt64(&this.finishedMigrating) > 0 {
+			return nil
+		}
 		throttlerFunction()
 	}
 
@@ -440,3 +459,8 @@ func (this *Throttler) throttle(onThrottled func()) {
 		time.Sleep(250 * time.Millisecond)
 	}
 }
+
+func (this *Throttler) Teardown() {
+	this.migrationContext.Log.Debugf("Tearing down...")
+	atomic.StoreInt64(&this.finishedMigrating, 1)
+}