// collectControlReplicasLag polls all the control replicas to get maximum lag value
collectControlReplicasLag polls all the control replicas to get maximum lag value
// collectControlReplicasLag polls all the control replicas to get maximum lag value
func (this *Throttler) collectControlReplicasLag() {
	readControlReplicasLag := func(replicationLagQuery string) error {
		if (this.migrationContext.TestOnReplica || this.migrationContext.MigrateOnReplica) && (atomic.LoadInt64(&this.migrationContext.AllEventsUpToLockProcessedInjectedFlag) > 0) {
			return nil
		}
		lagResult := mysql.GetMaxReplicationLag(
			this.migrationContext.InspectorConnectionConfig,
			this.migrationContext.GetThrottleControlReplicaKeys(),
			replicationLagQuery,
		)
		this.migrationContext.SetControlReplicasLagResult(lagResult)
		return nil
	}
	aggressiveTicker := time.Tick(100 * time.Millisecond)
	relaxedFactor := 10
	counter := 0
	shouldReadLagAggressively := false
	replicationLagQuery := ""
	for range aggressiveTicker {
		if counter%relaxedFactor == 0 {
			counter = 0
			maxLagMillisecondsThrottleThreshold := atomic.LoadInt64(&this.migrationContext.MaxLagMillisecondsThrottleThreshold)
			replicationLagQuery = this.migrationContext.GetReplicationLagQuery()
			shouldReadLagAggressively = (replicationLagQuery != "" && maxLagMillisecondsThrottleThreshold < 1000)
		}
		if counter == 0 || shouldReadLagAggressively {
			readControlReplicasLag(replicationLagQuery)
		}
		counter++
	}
}
func ( this * Throttler ) collectControlReplicasLag ( ) { readControlReplicasLag := func ( replicationLagQuery string ) error { if ( this . migrationContext . TestOnReplica || this . migrationContext . MigrateOnReplica ) && ( atomic . LoadInt64 ( & this . migrationContext . AllEventsUpToLockProcessedInjectedFlag ) > 0 ) { return nil } lagResult := mysql . GetMaxReplicationLag ( this . migrationContext . InspectorConnectionConfig , this . migrationContext . GetThrottleControlReplicaKeys ( ) , replicationLagQuery , ) this . migrationContext . SetControlReplicasLagResult ( lagResult ) return nil } aggressiveTicker := time . Tick ( 100 * time . Millisecond ) relaxedFactor := 10 counter := 0 shouldReadLagAggressively := false replicationLagQuery := " " for range aggressiveTicker { if counter % relaxedFactor == 0 { counter = 0 maxLagMillisecondsThrottleThreshold := atomic . LoadInt64 ( & this . migrationContext . MaxLagMillisecondsThrottleThreshold ) replicationLagQuery = this . migrationContext . GetReplicationLagQuery ( ) shouldReadLagAggressively = ( replicationLagQuery != " " && maxLagMillisecondsThrottleThreshold < 1000 ) } if counter == 0 || shouldReadLagAggressively { readControlReplicasLag ( replicationLagQuery ) } counter ++ } }
collectControlReplicasLag polls all the control replicas to get maximum lag value
collectControlReplicasLag polls all the control replicas to get maximum lag value
// 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. // We check replication lag every so often, or if we wish to be aggressive