support for millisecond throttling

- `--max-lag-millis` is at least `100ms`
- `--heartbeat-interval-millis` introduced; defaults `500ms`, can range `100ms` - `1s`
- Control replicas lag calculated asynchronously to throttle test
  - aggressive when `max-lag-millis < 1000` and when `replication-lag-query` is given

doc/interactive-commands.md

@@ -18,7 +18,7 @@ Both interfaces may serve at the same time. Both respond to simple text command,
 - `status`: returns a detailed status summary of migration progress and configuration
 - `sup`: returns a brief status summary of migration progress
 - `chunk-size=<newsize>`: modify the `chunk-size`; applies on next running copy-iteration
-- `max-lag-millis=<max-lag>`: modify the maximum replication lag threshold (milliseconds, minimum value is `1000`, i.e. 1 second)
+- `max-lag-millis=<max-lag>`: modify the maximum replication lag threshold (milliseconds, minimum value is `100`, i.e. `0.1` second)
 - `max-load=<max-load-thresholds>`: modify the `max-load` config; applies on next running copy-iteration
   The `max-load` format must be: `some_status=<numeric-threshold>[,some_status=<numeric-threshold>...]`. For example: `Threads_running=50,threads_connected=1000`, and you would then write/echo `max-load=Threads_running=50,threads_connected=1000` to the socket.
 - `critical-load=<load>`: change critical load setting (exceeding given thresholds causes panic and abort)

go/base/context.go

@@ -64,6 +64,7 @@ type MigrationContext struct {
 	CliUser     string
 	CliPassword string
 
+	HeartbeatIntervalMilliseconds       int64
 	defaultNumRetries                   int64
 	ChunkSize                           int64
 	niceRatio                           float64
@@ -111,6 +112,7 @@ type MigrationContext struct {
 	pointOfInterestTime       time.Time
 	pointOfInterestTimeMutex  *sync.Mutex
 	CurrentLag                int64
+	controlReplicasLagResult  mysql.ReplicationLagResult
 	TotalRowsCopied           int64
 	TotalDMLEventsApplied     int64
 	isThrottled               bool
@@ -323,6 +325,16 @@ func (this *MigrationContext) TimeSincePointOfInterest() time.Duration {
 	return time.Since(this.pointOfInterestTime)
 }
 
+func (this *MigrationContext) SetHeartbeatIntervalMilliseconds(heartbeatIntervalMilliseconds int64) {
+	if heartbeatIntervalMilliseconds < 100 {
+		heartbeatIntervalMilliseconds = 100
+	}
+	if heartbeatIntervalMilliseconds > 1000 {
+		heartbeatIntervalMilliseconds = 1000
+	}
+	this.HeartbeatIntervalMilliseconds = heartbeatIntervalMilliseconds
+}
+
 func (this *MigrationContext) SetMaxLagMillisecondsThrottleThreshold(maxLagMillisecondsThrottleThreshold int64) {
 	if maxLagMillisecondsThrottleThreshold < 100 {
 		maxLagMillisecondsThrottleThreshold = 100
@@ -451,6 +463,20 @@ func (this *MigrationContext) ReadCriticalLoad(criticalLoadList string) error {
 	return nil
 }
 
+func (this *MigrationContext) GetControlReplicasLagResult() mysql.ReplicationLagResult {
+	this.throttleMutex.Lock()
+	defer this.throttleMutex.Unlock()
+
+	lagResult := this.controlReplicasLagResult
+	return lagResult
+}
+
+func (this *MigrationContext) SetControlReplicasLagResult(lagResult *mysql.ReplicationLagResult) {
+	this.throttleMutex.Lock()
+	defer this.throttleMutex.Unlock()
+	this.controlReplicasLagResult = *lagResult
+}
+
 func (this *MigrationContext) GetThrottleControlReplicaKeys() *mysql.InstanceKeyMap {
 	this.throttleMutex.Lock()
 	defer this.throttleMutex.Unlock()

go/cmd/gh-ost/main.go

@@ -81,6 +81,7 @@ func main() {
 	replicationLagQuery := flag.String("replication-lag-query", "", "Query that detects replication lag in seconds. Result can be a floating point (by default gh-ost issues SHOW SLAVE STATUS and reads Seconds_behind_master). If you're using pt-heartbeat, query would be something like: SELECT ROUND(UNIX_TIMESTAMP() - MAX(UNIX_TIMESTAMP(ts))) AS delay FROM my_schema.heartbeat")
 	throttleControlReplicas := flag.String("throttle-control-replicas", "", "List of replicas on which to check for lag; comma delimited. Example: myhost1.com:3306,myhost2.com,myhost3.com:3307")
 	throttleQuery := flag.String("throttle-query", "", "when given, issued (every second) to check if operation should throttle. Expecting to return zero for no-throttle, >0 for throttle. Query is issued on the migrated server. Make sure this query is lightweight")
+	heartbeatIntervalMillis := flag.Int64("heartbeat-interval-millis", 500, "how frequently would gh-ost inject a heartbeat value")
 	flag.StringVar(&migrationContext.ThrottleFlagFile, "throttle-flag-file", "", "operation pauses when this file exists; hint: use a file that is specific to the table being altered")
 	flag.StringVar(&migrationContext.ThrottleAdditionalFlagFile, "throttle-additional-flag-file", "/tmp/gh-ost.throttle", "operation pauses when this file exists; hint: keep default, use for throttling multiple gh-ost operations")
 	flag.StringVar(&migrationContext.PostponeCutOverFlagFile, "postpone-cut-over-flag-file", "", "while this file exists, migration will postpone the final stage of swapping tables, and will keep on syncing the ghost table. Cut-over/swapping would be ready to perform the moment the file is deleted.")
@@ -181,6 +182,7 @@ func main() {
 	if migrationContext.ServeSocketFile == "" {
 		migrationContext.ServeSocketFile = fmt.Sprintf("/tmp/gh-ost.%s.%s.sock", migrationContext.DatabaseName, migrationContext.OriginalTableName)
 	}
+	migrationContext.SetHeartbeatIntervalMilliseconds(*heartbeatIntervalMillis)
 	migrationContext.SetNiceRatio(*niceRatio)
 	migrationContext.SetChunkSize(*chunkSize)
 	migrationContext.SetMaxLagMillisecondsThrottleThreshold(*maxLagMillis)

go/logic/applier.go

@@ -258,7 +258,7 @@ func (this *Applier) WriteChangelogState(value string) (string, error) {
 
 // InitiateHeartbeat creates a heartbeat cycle, writing to the changelog table.
 // This is done asynchronously
-func (this *Applier) InitiateHeartbeat(heartbeatIntervalMilliseconds int64) {
+func (this *Applier) InitiateHeartbeat() {
 	var numSuccessiveFailures int64
 	injectHeartbeat := func() error {
 		if _, err := this.WriteChangelog("heartbeat", time.Now().Format(time.RFC3339Nano)); err != nil {
@@ -273,10 +273,10 @@ func (this *Applier) InitiateHeartbeat(heartbeatIntervalMilliseconds int64) {
 	}
 	injectHeartbeat()
 
-	heartbeatTick := time.Tick(time.Duration(heartbeatIntervalMilliseconds) * time.Millisecond)
+	heartbeatTick := time.Tick(time.Duration(this.migrationContext.HeartbeatIntervalMilliseconds) * time.Millisecond)
 	for range heartbeatTick {
 		// Generally speaking, we would issue a goroutine, but I'd actually rather
-		// have this blocked rather than spam the master in the event something
+		// have this block the loop rather than spam the master in the event something
 		// goes wrong
 		if err := injectHeartbeat(); err != nil {
 			return

go/logic/migrator.go

@@ -36,8 +36,7 @@ const (
 type tableWriteFunc func() error
 
 const (
-	applyEventsQueueBuffer        = 100
-	heartbeatIntervalMilliseconds = 1000
+	applyEventsQueueBuffer = 100
 )
 
 type PrintStatusRule int
@@ -159,7 +158,7 @@ func (this *Migrator) shouldThrottle() (result bool, reason string) {
 		checkThrottleControlReplicas = false
 	}
 	if checkThrottleControlReplicas {
-		lagResult := mysql.GetMaxReplicationLag(this.migrationContext.InspectorConnectionConfig, this.migrationContext.GetThrottleControlReplicaKeys(), this.migrationContext.GetReplicationLagQuery())
+		lagResult := this.migrationContext.GetControlReplicasLagResult()
 		if lagResult.Err != nil {
 			return true, fmt.Sprintf("%+v %+v", lagResult.Key, lagResult.Err)
 		}
@@ -328,8 +327,8 @@ func (this *Migrator) onChangelogStateEvent(dmlEvent *binlog.BinlogDMLEvent) (er
 	return nil
 }
 
-// onChangelogHeartbeat is called when a heartbeat event is intercepted
-func (this *Migrator) onChangelogHeartbeat(heartbeatValue string) (err error) {
+// parseChangelogHeartbeat is called when a heartbeat event is intercepted
+func (this *Migrator) parseChangelogHeartbeat(heartbeatValue string) (err error) {
 	heartbeatTime, err := time.Parse(time.RFC3339Nano, heartbeatValue)
 	if err != nil {
 		return log.Errore(err)
@@ -427,7 +426,8 @@ func (this *Migrator) Migrate() (err error) {
 	if err := this.addDMLEventsListener(); err != nil {
 		return err
 	}
-	go this.initiateHeartbeatListener()
+	go this.initiateHeartbeatReader()
+	go this.initiateControlReplicasReader()
 
 	if err := this.applier.ReadMigrationRangeValues(); err != nil {
 		return err
@@ -1045,11 +1045,11 @@ func (this *Migrator) printStatus(rule PrintStatusRule, writers ...io.Writer) {
 	fmt.Fprintln(w, status)
 }
 
-// initiateHeartbeatListener listens for heartbeat events. gh-ost implements its own
+// initiateHeartbeatReader listens for heartbeat events. gh-ost implements its own
 // heartbeat mechanism, whether your DB has or hasn't an existing heartbeat solution.
 // Heartbeat is supplied via the changelog table
-func (this *Migrator) initiateHeartbeatListener() {
-	ticker := time.Tick((heartbeatIntervalMilliseconds * time.Millisecond) / 2)
+func (this *Migrator) initiateHeartbeatReader() {
+	ticker := time.Tick(time.Duration(this.migrationContext.HeartbeatIntervalMilliseconds) * time.Millisecond)
 	for range ticker {
 		go func() error {
 			if atomic.LoadInt64(&this.cleanupImminentFlag) > 0 {
@@ -1059,19 +1059,47 @@ func (this *Migrator) initiateHeartbeatListener() {
 			if err != nil {
 				return log.Errore(err)
 			}
-			for hint, value := range changelogState {
-				switch hint {
-				case "heartbeat":
-					{
-						this.onChangelogHeartbeat(value)
-					}
-				}
+			if heartbeatValue, ok := changelogState["heartbeat"]; ok {
+				this.parseChangelogHeartbeat(heartbeatValue)
 			}
 			return nil
 		}()
 	}
 }
 
+// initiateControlReplicasReader
+func (this *Migrator) initiateControlReplicasReader() {
+	readControlReplicasLag := func(replicationLagQuery string) error {
+		if (this.migrationContext.TestOnReplica || this.migrationContext.MigrateOnReplica) && (atomic.LoadInt64(&this.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 {
+			// 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.
+			counter = 0
+			maxLagMillisecondsThrottleThreshold := atomic.LoadInt64(&this.migrationContext.MaxLagMillisecondsThrottleThreshold)
+			replicationLagQuery = this.migrationContext.GetReplicationLagQuery()
+			shouldReadLagAggressively = (replicationLagQuery != "" && maxLagMillisecondsThrottleThreshold < 1000)
+		}
+		if counter == 0 || shouldReadLagAggressively {
+			// We check replication lag every so often, or if we wish to be aggressive
+			readControlReplicasLag(replicationLagQuery)
+		}
+		counter++
+	}
+}
+
 // initiateStreaming begins treaming of binary log events and registers listeners for such events
 func (this *Migrator) initiateStreaming() error {
 	this.eventsStreamer = NewEventsStreamer()
@@ -1139,7 +1167,7 @@ func (this *Migrator) initiateApplier() error {
 	}
 
 	this.applier.WriteChangelogState(string(TablesInPlace))
-	go this.applier.InitiateHeartbeat(heartbeatIntervalMilliseconds)
+	go this.applier.InitiateHeartbeat()
 	return nil
 }