// shouldThrottle performs checks to see whether we should currently be throttling.
// It merely observes the metrics collected by other components, it does not issue
// its own metric collection.
shouldThrottle performs checks to see whether we should currently be throttling.
It merely observes the metrics collected by other components, it does not issue
its own metric collection.
// shouldThrottle performs checks to see whether we should currently be throttling . // It merely observes the metrics collected by other components it does not issue // its own metric collection .
func (this *Throttler) shouldThrottle() (result bool, reason string) {
	generalCheckResult := this.migrationContext.GetThrottleGeneralCheckResult()
	if generalCheckResult.ShouldThrottle {
		return generalCheckResult.ShouldThrottle, generalCheckResult.Reason
	}
	maxLagMillisecondsThrottleThreshold := atomic.LoadInt64(&this.migrationContext.MaxLagMillisecondsThrottleThreshold)
	lag := atomic.LoadInt64(&this.migrationContext.CurrentLag)
	if time.Duration(lag) > time.Duration(maxLagMillisecondsThrottleThreshold)*time.Millisecond {
		return true, fmt.Sprintf("lag=%fs", time.Duration(lag).Seconds())
	}
	checkThrottleControlReplicas := true
	if (this.migrationContext.TestOnReplica || this.migrationContext.MigrateOnReplica) && (atomic.LoadInt64(&this.migrationContext.AllEventsUpToLockProcessedInjectedFlag) > 0) {
		checkThrottleControlReplicas = false
	}
	if checkThrottleControlReplicas {
		lagResult := this.migrationContext.GetControlReplicasLagResult()
		if lagResult.Err != nil {
			return true, fmt.Sprintf("%+v %+v", lagResult.Key, lagResult.Err)
		}
		if lagResult.Lag > time.Duration(maxLagMillisecondsThrottleThreshold)*time.Millisecond {
			return true, fmt.Sprintf("%+v replica-lag=%fs", lagResult.Key, lagResult.Lag.Seconds())
		}
	}
	return false, ""
}
func ( this * Throttler ) shouldThrottle ( ) ( result bool , reason string ) { generalCheckResult := this . migrationContext . GetThrottleGeneralCheckResult ( ) if generalCheckResult . ShouldThrottle { return generalCheckResult . ShouldThrottle , generalCheckResult . Reason } maxLagMillisecondsThrottleThreshold := atomic . LoadInt64 ( & this . migrationContext . MaxLagMillisecondsThrottleThreshold ) lag := atomic . LoadInt64 ( & this . migrationContext . CurrentLag ) if time . Duration ( lag ) > time . Duration ( maxLagMillisecondsThrottleThreshold ) * time . Millisecond { return true , fmt . Sprintf ( " " , time . Duration ( lag ) . Seconds ( ) ) } checkThrottleControlReplicas := true if ( this . migrationContext . TestOnReplica || this . migrationContext . MigrateOnReplica ) && ( atomic . LoadInt64 ( & this . migrationContext . AllEventsUpToLockProcessedInjectedFlag ) > 0 ) { checkThrottleControlReplicas = false } if checkThrottleControlReplicas { lagResult := this . migrationContext . GetControlReplicasLagResult ( ) if lagResult . Err != nil { return true , fmt . Sprintf ( " " , lagResult . Key , lagResult . Err ) } if lagResult . Lag > time . Duration ( maxLagMillisecondsThrottleThreshold ) * time . Millisecond { return true , fmt . Sprintf ( " " , lagResult . Key , lagResult . Lag . Seconds ( ) ) } } return false , " " }
shouldThrottle performs checks to see whether we should currently be throttling.
shouldThrottle performs checks to see whether we should currently be throttling .
// Replication lag throttle // Got here? No metrics indicates we need throttling.