optimize the big tx DML pipeline #1018.

Do not re-creating goroutine & channels.

Author: ffffwh

driver/mysql/applier.go

@@ -1075,9 +1075,8 @@ func (a *Applier) Shutdown() error {
 	close(a.shutdownCh)
 
 	if a.ai != nil {
-		a.ai.wg.Wait()
+		a.ai.Shutdown()
 	}
-	a.logger.Debug("Shutdown. a.ai.wg.Wait. after")
 	a.wg.Wait()
 	a.logger.Debug("Shutdown. a.wg.Wait. after")
 

driver/mysql/applier_incr.go

@@ -16,7 +16,6 @@ import (
 	sql "github.com/actiontech/dtle/driver/mysql/sql"
 	"github.com/actiontech/dtle/g"
 	gomysql "github.com/go-mysql-org/go-mysql/mysql"
-	"github.com/hashicorp/go-hclog"
 	"github.com/pkg/errors"
 	uuid "github.com/satori/go.uuid"
 )
@@ -62,6 +61,8 @@ type ApplierIncr struct {
 
 	wg                    sync.WaitGroup
 	SkipGtidExecutedTable bool
+	logTxCommit           bool
+	noBigTxDMLPipe        bool
 
 	mtsManager  *MtsManager
 	wsManager   *WritesetManager
@@ -71,8 +72,9 @@ type ApplierIncr struct {
 	sourceType  string
 	tableSpecs  []*common.TableSpec
 
-	inBigTx     bool
-	logTxCommit bool
+	inBigTx         bool
+	bigTxEventQueue chan *dmlExecItem
+	bigTxEventWg    sync.WaitGroup
 }
 
 func NewApplierIncr(applier *Applier, sourcetype string) (*ApplierIncr, error) {
@@ -94,6 +96,7 @@ func NewApplierIncr(applier *Applier, sourcetype string) (*ApplierIncr, error) {
 		gtidSetLock:           applier.gtidSetLock,
 		tableItems:            make(mapSchemaTableItems),
 		sourceType:            sourcetype,
+		bigTxEventQueue:       make(chan *dmlExecItem, 16),
 	}
 
 	if g.EnvIsTrue(g.ENV_SKIP_GTID_EXECUTED_TABLE) {
@@ -103,6 +106,10 @@ func NewApplierIncr(applier *Applier, sourcetype string) (*ApplierIncr, error) {
 	if g.EnvIsTrue(g.ENV_DTLE_LOG_TX_COMMIT) {
 		a.logTxCommit = true
 	}
+	if g.EnvIsTrue(g.ENV_DTLE_NO_BIG_TX_DML_PIPE) {
+		a.logger.Info("found DTLE_NO_BIG_TX_DML_PIPE")
+		a.noBigTxDMLPipe = true
+	}
 
 	a.timestampCtx = NewTimestampContext(a.shutdownCh, a.logger, func() bool {
 		return len(a.binlogEntryQueue) == 0 && len(a.applyBinlogMtsTxQueue) == 0
@@ -186,11 +193,32 @@ func (a *ApplierIncr) Run() (err error) {
 	return nil
 }
 
+func (a *ApplierIncr) bigTxQueueExecutor() {
+	for {
+		item := <-a.bigTxEventQueue
+		if item == nil {
+			break
+		}
+
+		if !a.HasShutdown() {
+			err := a.prepareIfNilAndExecute(item, 0)
+			if err != nil {
+				a.OnError(common.TaskStateDead, err)
+			}
+		}
+		a.bigTxEventWg.Done()
+	}
+}
+
 func (a *ApplierIncr) MtsWorker(workerIndex int) {
 	keepLoop := true
 
 	logger := a.logger.With("worker", workerIndex)
 
+	if workerIndex == 0 {
+		go a.bigTxQueueExecutor()
+	}
+
 	t := time.NewTicker(pingInterval)
 	defer t.Stop()
 	hasEntry := false
@@ -445,6 +473,38 @@ func (a *ApplierIncr) HasShutdown() bool {
 		return false
 	}
 }
+func (a *ApplierIncr) prepareIfNilAndExecute(item *dmlExecItem, workerIdx int) (err error) {
+	// hasUK bool, pstmt **gosql.Stmt, query string, args []interface{}
+	var r gosql.Result
+
+	if item.hasUK {
+		if *item.pstmt == nil {
+			a.logger.Debug("buildDMLEventQuery prepare query", "query", item.query)
+			*item.pstmt, err = a.dbs[workerIdx].Db.PrepareContext(a.ctx, item.query)
+			if err != nil {
+				a.logger.Error("buildDMLEventQuery prepare query", "query", item.query, "err", err)
+				return err
+			}
+		}
+
+		r, err = (*item.pstmt).ExecContext(a.ctx, item.args...)
+	} else {
+		r, err = a.dbs[workerIdx].Db.ExecContext(a.ctx, item.query, item.args...)
+	}
+
+	if err != nil {
+		a.logger.Error("error at exec", "gno", item.gno, "err", err)
+		return err
+	}
+
+	nr, err := r.RowsAffected()
+	if err != nil {
+		a.logger.Error("RowsAffected error", "gno", item.gno, "event", 0, "err", err)
+	} else {
+		a.logger.Debug("RowsAffected.after", "gno", item.gno, "event", 0, "nr", nr)
+	}
+	return nil
+}
 
 // ApplyEventQueries applies multiple DML queries onto the dest table
 func (a *ApplierIncr) ApplyBinlogEvent(workerIdx int, binlogEntryCtx *common.EntryContext) error {
@@ -455,7 +515,7 @@ func (a *ApplierIncr) ApplyBinlogEvent(workerIdx int, binlogEntryCtx *common.Ent
 
 	var err error
 	var timestamp uint32
-	txSid := binlogEntry.Coordinates.GetSid()
+	gno := binlogEntry.Coordinates.GetGNO()
 
 	dbApplier.DbMutex.Lock()
 	if dbApplier.Tx == nil {
@@ -469,79 +529,17 @@ func (a *ApplierIncr) ApplyBinlogEvent(workerIdx int, binlogEntryCtx *common.Ent
 		atomic.AddInt64(a.memory2, -int64(binlogEntry.Size()))
 	}()
 
-	type execItem struct {
-		hasUK bool
-		pstmt **gosql.Stmt
-		query string
-		args []interface{}
-	}
-	prepareIfNilAndExecute := func(hasUK bool, pstmt **gosql.Stmt, query string, args []interface{}) (err error) {
-		var r gosql.Result
-
-		if hasUK {
-			if *pstmt == nil {
-				a.logger.Debug("buildDMLEventQuery prepare query", "query", query)
-				*pstmt, err = a.dbs[workerIdx].Db.PrepareContext(a.ctx, query)
-				if err != nil {
-					a.logger.Error("buildDMLEventQuery prepare query", "query", query, "err", err)
-					return err
-				}
-			}
-
-			r, err = (*pstmt).ExecContext(a.ctx, args...)
-		} else {
-			r, err = a.dbs[workerIdx].Db.ExecContext(a.ctx, query, args...)
-		}
-
-		if err != nil {
-			logger.Error("error at exec", "gtid", hclog.Fmt("%s:%d", txSid, binlogEntry.Coordinates.GetGNO()),
-				"err", err)
-			return err
-		}
-
-		nr, err := r.RowsAffected()
-		if err != nil {
-			logger.Error("RowsAffected error", "gno", binlogEntry.Coordinates.GetGNO(), "event", 0, "err", err)
-		} else {
-			logger.Debug("RowsAffected.after", "gno", binlogEntry.Coordinates.GetGNO(), "event", 0, "nr", nr)
-		}
-		return nil
-	}
-
-	somequeue := make(chan *execItem, 16)
-	queueOrExec := func(item *execItem) error {
-		if a.inBigTx {
-			somequeue <- item
+	queueOrExec := func(item *dmlExecItem) error {
+		// TODO check if shutdown?
+		if !a.noBigTxDMLPipe && a.inBigTx {
+			a.bigTxEventWg.Add(1)
+			a.bigTxEventQueue <- item
 			return nil
 		} else {
-			return prepareIfNilAndExecute(item.hasUK, item.pstmt, item.query, item.args)
+			return a.prepareIfNilAndExecute(item, workerIdx)
 		}
 	}
 
-	executionLoopExitedCh := make(chan struct{})
-	if a.inBigTx {
-		go func() {
-			defer close(executionLoopExitedCh)
-			for {
-				select {
-				case item := <-somequeue:
-					if item == nil {
-						return
-					}
-
-					err := prepareIfNilAndExecute(item.hasUK, item.pstmt, item.query, item.args)
-					if err != nil {
-						a.OnError(common.TaskStateDead, err)
-					}
-				case <-a.shutdownCh:
-					return
-				}
-			}
-		}()
-	} else {
-		close(executionLoopExitedCh)
-	}
-
 	for i, event := range binlogEntry.Events {
 		if a.HasShutdown() {
 			break
@@ -680,7 +678,7 @@ func (a *ApplierIncr) ApplyBinlogEvent(workerIdx int, binlogEntryCtx *common.Ent
 					}
 					a.logger.Debug("BuildDMLInsertQuery", "query", query)
 
-					err = queueOrExec(&execItem{true, pstmt, query, sharedArgs})
+					err = queueOrExec(&dmlExecItem{true, pstmt, query, sharedArgs, gno})
 					if err != nil {
 						return err
 					}
@@ -695,7 +693,7 @@ func (a *ApplierIncr) ApplyBinlogEvent(workerIdx int, binlogEntryCtx *common.Ent
 					}
 					a.logger.Debug("BuildDMLDeleteQuery", "query", query)
 
-					err = queueOrExec(&execItem{hasUK, pstmt, query, uniqueKeyArgs})
+					err = queueOrExec(&dmlExecItem{hasUK, pstmt, query, uniqueKeyArgs, gno})
 					if err != nil {
 						return err
 					}
@@ -722,7 +720,7 @@ func (a *ApplierIncr) ApplyBinlogEvent(workerIdx int, binlogEntryCtx *common.Ent
 							return err
 						}
 
-						err = queueOrExec(&execItem{true, pstmt, query, sharedArgs})
+						err = queueOrExec(&dmlExecItem{true, pstmt, query, sharedArgs, gno})
 						if err != nil {
 							return err
 						}
@@ -735,7 +733,7 @@ func (a *ApplierIncr) ApplyBinlogEvent(workerIdx int, binlogEntryCtx *common.Ent
 						}
 						a.logger.Debug("BuildDMLDeleteQuery", "query", query)
 
-						err = queueOrExec(&execItem{hasUK, pstmt, query, uniqueKeyArgs})
+						err = queueOrExec(&dmlExecItem{hasUK, pstmt, query, uniqueKeyArgs, gno})
 						if err != nil {
 							return err
 						}
@@ -750,7 +748,7 @@ func (a *ApplierIncr) ApplyBinlogEvent(workerIdx int, binlogEntryCtx *common.Ent
 						args = append(args, sharedArgs...)
 						args = append(args, uniqueKeyArgs...)
 
-						err = queueOrExec(&execItem{hasUK, pstmt, query, args})
+						err = queueOrExec(&dmlExecItem{hasUK, pstmt, query, args, gno})
 						if err != nil {
 							return err
 						}
@@ -769,8 +767,10 @@ func (a *ApplierIncr) ApplyBinlogEvent(workerIdx int, binlogEntryCtx *common.Ent
 		timestamp = event.Timestamp
 		atomic.AddUint64(&a.appliedQueryCount, uint64(1))
 	}
-	close(somequeue)
-	<-executionLoopExitedCh
+	a.bigTxEventWg.Wait()
+	if a.HasShutdown() {
+		return fmt.Errorf("ApplyBinlogEvent: applier has been shutdown. gno %v", gno)
+	}
 
 	if binlogEntry.Final {
 		if !a.SkipGtidExecutedTable && a.sourceType == "mysql" {
@@ -887,3 +887,17 @@ func (a *ApplierIncr) handleEntryOracle(entryCtx *common.EntryContext) (err erro
 	}
 	return nil
 }
+
+func (a *ApplierIncr) Shutdown() {
+	close(a.bigTxEventQueue)
+	a.wg.Wait()
+	a.logger.Debug("Shutdown. ApplierIncr.wg.Wait. after")
+}
+
+type dmlExecItem struct {
+	hasUK bool
+	pstmt **gosql.Stmt
+	query string
+	args []interface{}
+	gno  int64 // for log only
+}

g/g.go

@@ -44,6 +44,7 @@ const (
 	ENV_SKIP_GTID_EXECUTED_TABLE = "DTLE_SKIP_GTID_EXECUTED_TABLE"
 	ENV_FORCE_MTS                = "DTLE_FORCE_MTS"
 	ENV_DTLE_LOG_TX_COMMIT       = "DTLE_LOG_TX_COMMIT"
+	ENV_DTLE_NO_BIG_TX_DML_PIPE  = "DTLE_NO_BIG_TX_DML_PIPE"
 
 	NatsMaxPayload        = 64 * 1024 * 1024