feat concurrent chunk data

doc/command-line-flags.md

@@ -64,6 +64,9 @@ It is not reliable to parse the `ALTER` statement to determine if it is instant
 ### binlogsyncer-max-reconnect-attempts
 `--binlogsyncer-max-reconnect-attempts=0`, the maximum number of attempts to re-establish a broken inspector connection for sync binlog. `0` or `negative number` means infinite retry, default `0`
 
+### chunk-concurrent-size
+`--chunk-concurrent-size=1`, The number of goroutines to execute chunks concurrently in each copy time slot, default `1`, allowed range `0`-`100`.
+
 ### conf
 
 `--conf=/path/to/my.cnf`: file where credentials are specified. Should be in (or contain) the following format:

go.mod

@@ -11,6 +11,7 @@ require (
 	golang.org/x/net v0.17.0
 	golang.org/x/term v0.13.0
 	golang.org/x/text v0.13.0
+	golang.org/x/sync v0.1.0
 )
 
 require (

go.sum

@@ -94,6 +94,8 @@ golang.org/x/net v0.17.0 h1:pVaXccu2ozPjCXewfr1S7xza/zcXTity9cCdXQYSjIM=
 golang.org/x/net v0.17.0/go.mod h1:NxSsAGuq816PNPmqtQdLE42eU2Fs7NoRIZrHJAlaCOE=
 golang.org/x/sync v0.0.0-20190423024810-112230192c58/go.mod h1:RxMgew5VJxzue5/jJTE5uejpjVlOe/izrB70Jof72aM=
 golang.org/x/sync v0.0.0-20201020160332-67f06af15bc9/go.mod h1:RxMgew5VJxzue5/jJTE5uejpjVlOe/izrB70Jof72aM=
+golang.org/x/sync v0.1.0 h1:wsuoTGHzEhffawBOhz5CYhcrV4IdKZbEyZjBMuTp12o=
+golang.org/x/sync v0.1.0/go.mod h1:RxMgew5VJxzue5/jJTE5uejpjVlOe/izrB70Jof72aM=
 golang.org/x/sys v0.0.0-20190215142949-d0b11bdaac8a/go.mod h1:STP8DvDyc/dI5b8T5hshtkjS+E42TnysNCUPdjciGhY=
 golang.org/x/sys v0.0.0-20190412213103-97732733099d/go.mod h1:h1NjWce9XRLGQEsW7wpKNCjG9DtNlClVuFLEZdDNbEs=
 golang.org/x/sys v0.0.0-20200930185726-fdedc70b468f/go.mod h1:h1NjWce9XRLGQEsW7wpKNCjG9DtNlClVuFLEZdDNbEs=

go/base/context.go

@@ -72,6 +72,14 @@ func NewThrottleCheckResult(throttle bool, reason string, reasonHint ThrottleRea
 	}
 }
 
+type IterationRangeValues struct {
+	Min                *sql.ColumnValues
+	Max                *sql.ColumnValues
+	Size               int64
+	IsIncludeMinValues bool
+	HasFurtherRange    bool
+}
+
 // MigrationContext has the general, global state of migration. It is used by
 // all components throughout the migration process.
 type MigrationContext struct {
@@ -119,6 +127,7 @@ type MigrationContext struct {
 	HeartbeatIntervalMilliseconds       int64
 	defaultNumRetries                   int64
 	ChunkSize                           int64
+	ChunkConcurrentSize                 int64
 	niceRatio                           float64
 	MaxLagMillisecondsThrottleThreshold int64
 	throttleControlReplicaKeys          *mysql.InstanceKeyMap
@@ -210,25 +219,26 @@ type MigrationContext struct {
 	InCutOverCriticalSectionFlag           int64
 	PanicAbort                             chan error
 
-	OriginalTableColumnsOnApplier    *sql.ColumnList
-	OriginalTableColumns             *sql.ColumnList
-	OriginalTableVirtualColumns      *sql.ColumnList
-	OriginalTableUniqueKeys          [](*sql.UniqueKey)
-	OriginalTableAutoIncrement       uint64
-	GhostTableColumns                *sql.ColumnList
-	GhostTableVirtualColumns         *sql.ColumnList
-	GhostTableUniqueKeys             [](*sql.UniqueKey)
-	UniqueKey                        *sql.UniqueKey
-	SharedColumns                    *sql.ColumnList
-	ColumnRenameMap                  map[string]string
-	DroppedColumnsMap                map[string]bool
-	MappedSharedColumns              *sql.ColumnList
-	MigrationRangeMinValues          *sql.ColumnValues
-	MigrationRangeMaxValues          *sql.ColumnValues
-	Iteration                        int64
-	MigrationIterationRangeMinValues *sql.ColumnValues
-	MigrationIterationRangeMaxValues *sql.ColumnValues
-	ForceTmpTableName                string
+	OriginalTableColumnsOnApplier            *sql.ColumnList
+	OriginalTableColumns                     *sql.ColumnList
+	OriginalTableVirtualColumns              *sql.ColumnList
+	OriginalTableUniqueKeys                  [](*sql.UniqueKey)
+	OriginalTableAutoIncrement               uint64
+	GhostTableColumns                        *sql.ColumnList
+	GhostTableVirtualColumns                 *sql.ColumnList
+	GhostTableUniqueKeys                     [](*sql.UniqueKey)
+	UniqueKey                                *sql.UniqueKey
+	SharedColumns                            *sql.ColumnList
+	ColumnRenameMap                          map[string]string
+	DroppedColumnsMap                        map[string]bool
+	MappedSharedColumns                      *sql.ColumnList
+	MigrationRangeMinValues                  *sql.ColumnValues
+	MigrationRangeMaxValues                  *sql.ColumnValues
+	Iteration                                int64
+	MigrationIterationRangeMinValues         *sql.ColumnValues
+	MigrationIterationRangeMaxValues         *sql.ColumnValues
+	CalculateNextIterationRangeEndValuesLock *sync.Mutex
+	ForceTmpTableName                        string
 
 	recentBinlogCoordinates mysql.BinlogCoordinates
 
@@ -269,26 +279,27 @@ type ContextConfig struct {
 
 func NewMigrationContext() *MigrationContext {
 	return &MigrationContext{
-		Uuid:                                uuid.NewString(),
-		defaultNumRetries:                   60,
-		ChunkSize:                           1000,
-		InspectorConnectionConfig:           mysql.NewConnectionConfig(),
-		ApplierConnectionConfig:             mysql.NewConnectionConfig(),
-		MaxLagMillisecondsThrottleThreshold: 1500,
-		CutOverLockTimeoutSeconds:           3,
-		DMLBatchSize:                        10,
-		etaNanoseonds:                       ETAUnknown,
-		maxLoad:                             NewLoadMap(),
-		criticalLoad:                        NewLoadMap(),
-		throttleMutex:                       &sync.Mutex{},
-		throttleHTTPMutex:                   &sync.Mutex{},
-		throttleControlReplicaKeys:          mysql.NewInstanceKeyMap(),
-		configMutex:                         &sync.Mutex{},
-		pointOfInterestTimeMutex:            &sync.Mutex{},
-		lastHeartbeatOnChangelogMutex:       &sync.Mutex{},
-		ColumnRenameMap:                     make(map[string]string),
-		PanicAbort:                          make(chan error),
-		Log:                                 NewDefaultLogger(),
+		Uuid:                                     uuid.NewString(),
+		defaultNumRetries:                        60,
+		ChunkSize:                                1000,
+		InspectorConnectionConfig:                mysql.NewConnectionConfig(),
+		ApplierConnectionConfig:                  mysql.NewConnectionConfig(),
+		MaxLagMillisecondsThrottleThreshold:      1500,
+		CutOverLockTimeoutSeconds:                3,
+		DMLBatchSize:                             10,
+		etaNanoseonds:                            ETAUnknown,
+		maxLoad:                                  NewLoadMap(),
+		criticalLoad:                             NewLoadMap(),
+		throttleMutex:                            &sync.Mutex{},
+		throttleHTTPMutex:                        &sync.Mutex{},
+		throttleControlReplicaKeys:               mysql.NewInstanceKeyMap(),
+		configMutex:                              &sync.Mutex{},
+		pointOfInterestTimeMutex:                 &sync.Mutex{},
+		lastHeartbeatOnChangelogMutex:            &sync.Mutex{},
+		CalculateNextIterationRangeEndValuesLock: &sync.Mutex{},
+		ColumnRenameMap:                          make(map[string]string),
+		PanicAbort:                               make(chan error),
+		Log:                                      NewDefaultLogger(),
 	}
 }
 
@@ -616,6 +627,16 @@ func (this *MigrationContext) SetChunkSize(chunkSize int64) {
 	atomic.StoreInt64(&this.ChunkSize, chunkSize)
 }
 
+func (this *MigrationContext) SetChunkConcurrentSize(chunkConcurrentSize int64) {
+	if chunkConcurrentSize < 1 {
+		chunkConcurrentSize = 1
+	}
+	if chunkConcurrentSize > 100 {
+		chunkConcurrentSize = 100
+	}
+	atomic.StoreInt64(&this.ChunkConcurrentSize, chunkConcurrentSize)
+}
+
 func (this *MigrationContext) SetDMLBatchSize(batchSize int64) {
 	if batchSize < 1 {
 		batchSize = 1

go/cmd/gh-ost/main.go

@@ -104,6 +104,7 @@ func main() {
 	flag.BoolVar(&migrationContext.CutOverExponentialBackoff, "cut-over-exponential-backoff", false, "Wait exponentially longer intervals between failed cut-over attempts. Wait intervals obey a maximum configurable with 'exponential-backoff-max-interval').")
 	exponentialBackoffMaxInterval := flag.Int64("exponential-backoff-max-interval", 64, "Maximum number of seconds to wait between attempts when performing various operations with exponential backoff.")
 	chunkSize := flag.Int64("chunk-size", 1000, "amount of rows to handle in each iteration (allowed range: 10-100,000)")
+	chunkConcurrentSize := flag.Int64("chunk-concurrent-size", 1, "The number of goroutines to execute chunks concurrently in each copy time slot, default 1 (allowed range: 1-100)")
 	dmlBatchSize := flag.Int64("dml-batch-size", 10, "batch size for DML events to apply in a single transaction (range 1-100)")
 	defaultRetries := flag.Int64("default-retries", 60, "Default number of retries for various operations before panicking")
 	cutOverLockTimeoutSeconds := flag.Int64("cut-over-lock-timeout-seconds", 3, "Max number of seconds to hold locks on tables while attempting to cut-over (retry attempted when lock exceeds timeout)")
@@ -294,6 +295,7 @@ func main() {
 	migrationContext.SetHeartbeatIntervalMilliseconds(*heartbeatIntervalMillis)
 	migrationContext.SetNiceRatio(*niceRatio)
 	migrationContext.SetChunkSize(*chunkSize)
+	migrationContext.SetChunkConcurrentSize(*chunkConcurrentSize)
 	migrationContext.SetDMLBatchSize(*dmlBatchSize)
 	migrationContext.SetMaxLagMillisecondsThrottleThreshold(*maxLagMillis)
 	migrationContext.SetThrottleQuery(*throttleQuery)

go/logic/applier.go

@@ -76,6 +76,10 @@ func (this *Applier) InitDBConnections() (err error) {
 	if this.db, _, err = mysql.GetDB(this.migrationContext.Uuid, applierUri); err != nil {
 		return err
 	}
+	concurrentSize := atomic.LoadInt64(&this.migrationContext.ChunkConcurrentSize)
+	if concurrentSize > mysql.MaxDBPoolConnections {
+		this.db.SetMaxOpenConns(int(concurrentSize))
+	}
 	singletonApplierUri := fmt.Sprintf("%s&timeout=0", applierUri)
 	if this.singletonDB, _, err = mysql.GetDB(this.migrationContext.Uuid, singletonApplierUri); err != nil {
 		return err
@@ -561,11 +565,18 @@ func (this *Applier) ReadMigrationRangeValues() error {
 // which will be used for copying the next chunk of rows. Ir returns "false" if there is
 // no further chunk to work through, i.e. we're past the last chunk and are done with
 // iterating the range (and this done with copying row chunks)
-func (this *Applier) CalculateNextIterationRangeEndValues() (hasFurtherRange bool, err error) {
-	this.migrationContext.MigrationIterationRangeMinValues = this.migrationContext.MigrationIterationRangeMaxValues
-	if this.migrationContext.MigrationIterationRangeMinValues == nil {
-		this.migrationContext.MigrationIterationRangeMinValues = this.migrationContext.MigrationRangeMinValues
+func (this *Applier) CalculateNextIterationRangeEndValues() (values *base.IterationRangeValues, err error) {
+	this.migrationContext.CalculateNextIterationRangeEndValuesLock.Lock()
+	defer this.migrationContext.CalculateNextIterationRangeEndValuesLock.Unlock()
+
+	iterationRangeValues := &base.IterationRangeValues{}
+
+	iterationRangeValues.Min = this.migrationContext.MigrationIterationRangeMaxValues
+	if iterationRangeValues.Min == nil {
+		iterationRangeValues.Min = this.migrationContext.MigrationRangeMinValues
+		iterationRangeValues.IsIncludeMinValues = true
 	}
+
 	for i := 0; i < 2; i++ {
 		buildFunc := sql.BuildUniqueKeyRangeEndPreparedQueryViaOffset
 		if i == 1 {
@@ -575,46 +586,48 @@ func (this *Applier) CalculateNextIterationRangeEndValues() (hasFurtherRange boo
 			this.migrationContext.DatabaseName,
 			this.migrationContext.OriginalTableName,
 			&this.migrationContext.UniqueKey.Columns,
-			this.migrationContext.MigrationIterationRangeMinValues.AbstractValues(),
+			iterationRangeValues.Min.AbstractValues(),
 			this.migrationContext.MigrationRangeMaxValues.AbstractValues(),
 			atomic.LoadInt64(&this.migrationContext.ChunkSize),
-			this.migrationContext.GetIteration() == 0,
+			iterationRangeValues.IsIncludeMinValues,
 			fmt.Sprintf("iteration:%d", this.migrationContext.GetIteration()),
 		)
 		if err != nil {
-			return hasFurtherRange, err
+			return iterationRangeValues, err
 		}
 
 		rows, err := this.db.Query(query, explodedArgs...)
 		if err != nil {
-			return hasFurtherRange, err
+			return iterationRangeValues, err
 		}
 		defer rows.Close()
 
 		iterationRangeMaxValues := sql.NewColumnValues(this.migrationContext.UniqueKey.Len())
 		for rows.Next() {
 			if err = rows.Scan(iterationRangeMaxValues.ValuesPointers...); err != nil {
-				return hasFurtherRange, err
+				return iterationRangeValues, err
 			}
-			hasFurtherRange = true
+			iterationRangeValues.HasFurtherRange = true
 		}
 		if err = rows.Err(); err != nil {
-			return hasFurtherRange, err
+			return iterationRangeValues, err
 		}
-		if hasFurtherRange {
-			this.migrationContext.MigrationIterationRangeMaxValues = iterationRangeMaxValues
-			return hasFurtherRange, nil
+		if iterationRangeValues.HasFurtherRange {
+			iterationRangeValues.Max = iterationRangeMaxValues
+			this.migrationContext.MigrationIterationRangeMinValues = iterationRangeValues.Min
+			this.migrationContext.MigrationIterationRangeMaxValues = iterationRangeValues.Max
+			return iterationRangeValues, nil
 		}
 	}
 	this.migrationContext.Log.Debugf("Iteration complete: no further range to iterate")
-	return hasFurtherRange, nil
+	return iterationRangeValues, nil
 }
 
 // ApplyIterationInsertQuery issues a chunk-INSERT query on the ghost table. It is where
 // data actually gets copied from original table.
-func (this *Applier) ApplyIterationInsertQuery() (chunkSize int64, rowsAffected int64, duration time.Duration, err error) {
+func (this *Applier) ApplyIterationInsertQuery(iterationRangeValues *base.IterationRangeValues) (chunkSize int64, rowsAffected int64, duration time.Duration, err error) {
 	startTime := time.Now()
-	chunkSize = atomic.LoadInt64(&this.migrationContext.ChunkSize)
+	chunkSize = iterationRangeValues.Size
 
 	query, explodedArgs, err := sql.BuildRangeInsertPreparedQuery(
 		this.migrationContext.DatabaseName,
@@ -624,9 +637,9 @@ func (this *Applier) ApplyIterationInsertQuery() (chunkSize int64, rowsAffected
 		this.migrationContext.MappedSharedColumns.Names(),
 		this.migrationContext.UniqueKey.Name,
 		&this.migrationContext.UniqueKey.Columns,
-		this.migrationContext.MigrationIterationRangeMinValues.AbstractValues(),
-		this.migrationContext.MigrationIterationRangeMaxValues.AbstractValues(),
-		this.migrationContext.GetIteration() == 0,
+		iterationRangeValues.Min.AbstractValues(),
+		iterationRangeValues.Max.AbstractValues(),
+		iterationRangeValues.IsIncludeMinValues,
 		this.migrationContext.IsTransactionalTable(),
 	)
 	if err != nil {
@@ -663,8 +676,8 @@ func (this *Applier) ApplyIterationInsertQuery() (chunkSize int64, rowsAffected
 	duration = time.Since(startTime)
 	this.migrationContext.Log.Debugf(
 		"Issued INSERT on range: [%s]..[%s]; iteration: %d; chunk-size: %d",
-		this.migrationContext.MigrationIterationRangeMinValues,
-		this.migrationContext.MigrationIterationRangeMaxValues,
+		iterationRangeValues.Min,
+		iterationRangeValues.Max,
 		this.migrationContext.GetIteration(),
 		chunkSize)
 	return chunkSize, rowsAffected, duration, nil