Multithreaded replication WIP

go/base/context.go

@@ -168,6 +168,9 @@ type MigrationContext struct {
 	CutOverType                  CutOver
 	ReplicaServerId              uint
 
+	// Number of workers used by the trx coordinator
+	NumWorkers int
+
 	Hostname                               string
 	AssumeMasterHostname                   string
 	ApplierTimeZone                        string

go/binlog/gomysql_reader.go

@@ -6,12 +6,10 @@
 package binlog
 
 import (
-	"fmt"
 	"sync"
 
 	"github.com/github/gh-ost/go/base"
 	"github.com/github/gh-ost/go/mysql"
-	"github.com/github/gh-ost/go/sql"
 
 	"time"
 
@@ -76,68 +74,17 @@ func (this *GoMySQLReader) GetCurrentBinlogCoordinates() *mysql.BinlogCoordinate
 	return &returnCoordinates
 }
 
-// StreamEvents
-func (this *GoMySQLReader) handleRowsEvent(ev *replication.BinlogEvent, rowsEvent *replication.RowsEvent, entriesChannel chan<- *BinlogEntry) error {
-	if this.currentCoordinates.IsLogPosOverflowBeyond4Bytes(&this.LastAppliedRowsEventHint) {
-		return fmt.Errorf("Unexpected rows event at %+v, the binlog end_log_pos is overflow 4 bytes", this.currentCoordinates)
-	}
-
-	if this.currentCoordinates.SmallerThanOrEquals(&this.LastAppliedRowsEventHint) {
-		this.migrationContext.Log.Debugf("Skipping handled query at %+v", this.currentCoordinates)
-		return nil
-	}
-
-	dml := ToEventDML(ev.Header.EventType.String())
-	if dml == NotDML {
-		return fmt.Errorf("Unknown DML type: %s", ev.Header.EventType.String())
-	}
-	for i, row := range rowsEvent.Rows {
-		if dml == UpdateDML && i%2 == 1 {
-			// An update has two rows (WHERE+SET)
-			// We do both at the same time
-			continue
-		}
-		binlogEntry := NewBinlogEntryAt(this.currentCoordinates)
-		binlogEntry.DmlEvent = NewBinlogDMLEvent(
-			string(rowsEvent.Table.Schema),
-			string(rowsEvent.Table.Table),
-			dml,
-		)
-		switch dml {
-		case InsertDML:
-			{
-				binlogEntry.DmlEvent.NewColumnValues = sql.ToColumnValues(row)
-			}
-		case UpdateDML:
-			{
-				binlogEntry.DmlEvent.WhereColumnValues = sql.ToColumnValues(row)
-				binlogEntry.DmlEvent.NewColumnValues = sql.ToColumnValues(rowsEvent.Rows[i+1])
-			}
-		case DeleteDML:
-			{
-				binlogEntry.DmlEvent.WhereColumnValues = sql.ToColumnValues(row)
-			}
-		}
-		// The channel will do the throttling. Whoever is reading from the channel
-		// decides whether action is taken synchronously (meaning we wait before
-		// next iteration) or asynchronously (we keep pushing more events)
-		// In reality, reads will be synchronous
-		entriesChannel <- binlogEntry
-	}
-	this.LastAppliedRowsEventHint = this.currentCoordinates
-	return nil
-}
-
-// StreamEvents
-func (this *GoMySQLReader) StreamEvents(canStopStreaming func() bool, entriesChannel chan<- *BinlogEntry) error {
-	if canStopStreaming() {
-		return nil
-	}
+// StreamEvents reads binlog events and sends them to the given channel.
+// It is blocking and should be executed in a goroutine.
+func (this *GoMySQLReader) StreamEvents(ctx context.Context, canStopStreaming func() bool, eventChannel chan<- *replication.BinlogEvent) error {
 	for {
 		if canStopStreaming() {
-			break
+			return nil
+		}
+		if err := ctx.Err(); err != nil {
+			return err
 		}
-		ev, err := this.binlogStreamer.GetEvent(context.Background())
+		ev, err := this.binlogStreamer.GetEvent(ctx)
 		if err != nil {
 			return err
 		}
@@ -147,24 +94,8 @@ func (this *GoMySQLReader) StreamEvents(canStopStreaming func() bool, entriesCha
 			this.currentCoordinates.LogPos = int64(ev.Header.LogPos)
 			this.currentCoordinates.EventSize = int64(ev.Header.EventSize)
 		}()
-
-		switch binlogEvent := ev.Event.(type) {
-		case *replication.RotateEvent:
-			func() {
-				this.currentCoordinatesMutex.Lock()
-				defer this.currentCoordinatesMutex.Unlock()
-				this.currentCoordinates.LogFile = string(binlogEvent.NextLogName)
-			}()
-			this.migrationContext.Log.Infof("rotate to next log from %s:%d to %s", this.currentCoordinates.LogFile, int64(ev.Header.LogPos), binlogEvent.NextLogName)
-		case *replication.RowsEvent:
-			if err := this.handleRowsEvent(ev, binlogEvent, entriesChannel); err != nil {
-				return err
-			}
-		}
+		eventChannel <- ev
 	}
-	this.migrationContext.Log.Debugf("done streaming events")
-
-	return nil
 }
 
 func (this *GoMySQLReader) Close() error {

go/cmd/gh-ost/main.go

@@ -111,6 +111,7 @@ func main() {
 	flag.BoolVar(&migrationContext.PanicOnWarnings, "panic-on-warnings", false, "Panic when SQL warnings are encountered when copying a batch indicating data loss")
 	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) or attempting instant DDL")
 	niceRatio := flag.Float64("nice-ratio", 0, "force being 'nice', imply sleep time per chunk time; range: [0.0..100.0]. Example values: 0 is aggressive. 1: for every 1ms spent copying rows, sleep additional 1ms (effectively doubling runtime); 0.7: for every 10ms spend in a rowcopy chunk, spend 7ms sleeping immediately after")
+	flag.IntVar(&migrationContext.NumWorkers, "workers", 8, "Number of concurrent workers for applying DML events. Each worker uses one goroutine.")
 
 	maxLagMillis := flag.Int64("max-lag-millis", 1500, "replication lag at which to throttle operation")
 	replicationLagQuery := flag.String("replication-lag-query", "", "Deprecated. gh-ost uses an internal, subsecond resolution query")

go/logic/applier.go

@@ -79,12 +79,14 @@ func NewApplier(migrationContext *base.MigrationContext) *Applier {
 	}
 }
 
-func (this *Applier) InitDBConnections() (err error) {
+func (this *Applier) InitDBConnections(maxConns int) (err error) {
 	applierUri := this.connectionConfig.GetDBUri(this.migrationContext.DatabaseName)
 	uriWithMulti := fmt.Sprintf("%s&multiStatements=true", applierUri)
 	if this.db, _, err = mysql.GetDB(this.migrationContext.Uuid, uriWithMulti); err != nil {
 		return err
 	}
+	this.db.SetMaxOpenConns(maxConns)
+	this.db.SetMaxIdleConns(maxConns)
 	singletonApplierUri := fmt.Sprintf("%s&timeout=0", applierUri)
 	if this.singletonDB, _, err = mysql.GetDB(this.migrationContext.Uuid, singletonApplierUri); err != nil {
 		return err

go/logic/applier_test.go

@@ -270,7 +270,7 @@ func (suite *ApplierTestSuite) TestInitDBConnections() {
 	applier := NewApplier(migrationContext)
 	defer applier.Teardown()
 
-	err = applier.InitDBConnections()
+	err = applier.InitDBConnections(8)
 	suite.Require().NoError(err)
 
 	suite.Require().Equal("8.0.40", migrationContext.ApplierMySQLVersion)
@@ -313,7 +313,7 @@ func (suite *ApplierTestSuite) TestApplyDMLEventQueries() {
 	suite.Require().NoError(applier.prepareQueries())
 	defer applier.Teardown()
 
-	err = applier.InitDBConnections()
+	err = applier.InitDBConnections(8)
 	suite.Require().NoError(err)
 
 	dmlEvents := []*binlog.BinlogDMLEvent{
@@ -373,7 +373,7 @@ func (suite *ApplierTestSuite) TestValidateOrDropExistingTables() {
 	applier := NewApplier(migrationContext)
 	defer applier.Teardown()
 
-	err = applier.InitDBConnections()
+	err = applier.InitDBConnections(8)
 	suite.Require().NoError(err)
 
 	err = applier.ValidateOrDropExistingTables()
@@ -408,7 +408,7 @@ func (suite *ApplierTestSuite) TestValidateOrDropExistingTablesWithGhostTableExi
 	applier := NewApplier(migrationContext)
 	defer applier.Teardown()
 
-	err = applier.InitDBConnections()
+	err = applier.InitDBConnections(8)
 	suite.Require().NoError(err)
 
 	err = applier.ValidateOrDropExistingTables()
@@ -442,7 +442,7 @@ func (suite *ApplierTestSuite) TestValidateOrDropExistingTablesWithGhostTableExi
 	applier := NewApplier(migrationContext)
 	defer applier.Teardown()
 
-	err = applier.InitDBConnections()
+	err = applier.InitDBConnections(8)
 	suite.Require().NoError(err)
 
 	err = applier.ValidateOrDropExistingTables()
@@ -483,7 +483,7 @@ func (suite *ApplierTestSuite) TestCreateGhostTable() {
 	applier := NewApplier(migrationContext)
 	defer applier.Teardown()
 
-	err = applier.InitDBConnections()
+	err = applier.InitDBConnections(8)
 	suite.Require().NoError(err)
 
 	err = applier.CreateGhostTable()
@@ -540,7 +540,7 @@ func (suite *ApplierTestSuite) TestPanicOnWarningsInApplyIterationInsertQuerySuc
 	suite.Require().NoError(applier.prepareQueries())
 	defer applier.Teardown()
 
-	err = applier.InitDBConnections()
+	err = applier.InitDBConnections(8)
 	suite.Require().NoError(err)
 
 	_, err = suite.db.ExecContext(ctx, "INSERT INTO test.testing (id, item_id) VALUES (123456, 42);")