debug WaitForAllCommitted

- add log
- add test

Author: ffffwh

driver/mysql/applier_gtid_executed.go

@@ -246,11 +246,9 @@ func (a *GtidExecutedCreater) createTableGtidExecutedV4() error {
 }
 
 func (a *ApplierIncr) cleanGtidExecuted(sid uuid.UUID, txSid string) error {
-	a.logger.Debug("incr. cleanup before WaitForExecution")
-	if !a.mtsManager.WaitForAllCommitted() {
+	if !a.mtsManager.WaitForAllCommitted(a.logger.With("txSid", txSid)) {
 		return nil // shutdown
 	}
-	a.logger.Debug("incr. cleanup after WaitForExecution")
 
 	intervalStr := func() string {
 		a.gtidSetLock.RLock()

driver/mysql/applier_incr.go

@@ -347,11 +347,9 @@ func (a *ApplierIncr) handleEntry(entryCtx *common.EntryContext) (err error) {
 	} else {
 		if binlogEntry.Index == 0 {
 			if rotated {
-				a.logger.Debug("binlog rotated. WaitForAllCommitted before", "file", a.replayingBinlogFile)
-				if !a.mtsManager.WaitForAllCommitted() {
+				if !a.mtsManager.WaitForAllCommitted(a.logger.With("rotate", a.replayingBinlogFile)) {
 					return nil // TODO shutdown
 				}
-				a.logger.Debug("binlog rotated. WaitForAllCommitted after", "file", a.replayingBinlogFile)
 				a.mtsManager.lastCommitted = 0
 				a.mtsManager.lastEnqueue = 0
 				a.wsManager.resetCommonParent(0)
@@ -376,12 +374,11 @@ func (a *ApplierIncr) handleEntry(entryCtx *common.EntryContext) (err error) {
 			hasDDL := binlogEntry.HasDDL()
 			inMiddleDDL := hasDDL || a.prevDDL // DDL must be executed separatedly
 			if inMiddleDDL || isBig {
-				a.logger.Info("WaitForAllCommitted",
-					"gno", txGno, "seq", binlogEntry.Coordinates.GetSequenceNumber(),
-					"lc", binlogEntry.Coordinates.GetLastCommit(), "leq", a.mtsManager.lastEnqueue,
+				if !a.mtsManager.WaitForAllCommitted(a.logger.With("gno", txGno,
+					"seq", binlogEntry.Coordinates.GetSequenceNumber(),
+					"lc", binlogEntry.Coordinates.GetLastCommit(),
 					"hasDDL", hasDDL, "prevDDL", a.prevDDL,
-					"bigtx", isBig, "index", binlogEntry.Index)
-				if !a.mtsManager.WaitForAllCommitted() {
+					"bigtx", isBig, "index", binlogEntry.Index)) {
 					return nil // shutdown
 				}
 			}

driver/mysql/applier_incr_test.go

@@ -1,2 +1,44 @@
 package mysql
 
+import (
+	"github.com/hashicorp/go-hclog"
+	"testing"
+	"time"
+)
+
+func TestMtsManager(t *testing.T) {
+	logger := hclog.Default()
+	shutdownCh := make(chan struct{})
+	defer close(shutdownCh)
+
+	mm := NewMtsManager(shutdownCh, logger)
+	go mm.LcUpdater()
+
+	doneCh := make(chan struct{})
+	go func() {
+		mm.WaitForExecution0(1, 0)
+		logger.Info("wait 1 0")
+		mm.Executed0(1)
+		mm.WaitForExecution0(2, 1)
+		logger.Info("wait 2 1")
+		mm.Executed0(2)
+		mm.WaitForExecution0(3, 2)
+		logger.Info("wait 3 2")
+		mm.Executed0(3)
+
+		// BinlogEntry was resent
+		mm.WaitForExecution0(2, 1)
+		mm.Executed0(2)
+
+		mm.WaitForAllCommitted(logger)
+		close(doneCh)
+	}()
+
+	tm := time.NewTimer(10 * time.Second)
+	select {
+	case <-doneCh:
+		t.Log("case finished")
+	case <-tm.C:
+		t.Fatal("might be stuck")
+	}
+}

driver/mysql/applier_mts.go

@@ -4,6 +4,7 @@ import (
 	"container/heap"
 	"hash/fnv"
 	"sync/atomic"
+	"time"
 
 	"github.com/actiontech/dtle/driver/common"
 	"github.com/actiontech/dtle/g"
@@ -56,14 +57,18 @@ func NewMtsManager(shutdownCh chan struct{}, logger g.LoggerType) *MtsManager {
 }
 
 //  This function must be called sequentially.
-func (mm *MtsManager) WaitForAllCommitted() bool {
-	g.Logger.Debug("WaitForAllCommitted", "lc", mm.lastCommitted, "le", mm.lastEnqueue)
+func (mm *MtsManager) WaitForAllCommitted(logger g.LoggerType) bool {
+	t := time.NewTimer(30 * time.Second)
+	defer t.Stop()
+
 	for {
-		if mm.lastCommitted == mm.lastEnqueue {
+		if atomic.LoadInt64(&mm.lastCommitted) == mm.lastEnqueue {
 			return true
 		}
 
 		select {
+		case <-t.C: // this will only be triggered once
+			logger.Warn("WaitForAllCommitted has been stuck for 30s")
 		case <-mm.shutdownCh:
 			return false
 		case <-mm.updated:
@@ -75,15 +80,21 @@ func (mm *MtsManager) WaitForAllCommitted() bool {
 // block for waiting. return true for can_execute, false for abortion.
 //  This function must be called sequentially.
 func (mm *MtsManager) WaitForExecution(binlogEntry *common.DataEntry) bool {
-	mm.lastEnqueue = binlogEntry.Coordinates.GetSequenceNumber()
+	return mm.WaitForExecution0(
+		binlogEntry.Coordinates.GetSequenceNumber(),
+		binlogEntry.Coordinates.(*common.MySQLCoordinateTx).LastCommitted)
+}
+
+func (mm *MtsManager) WaitForExecution0(seq int64, lc int64) bool {
+	mm.lastEnqueue = seq
 
 	if mm.forceMts {
 		return true
 	}
 
 	for {
 		currentLC := atomic.LoadInt64(&mm.lastCommitted)
-		if currentLC >= binlogEntry.Coordinates.(*common.MySQLCoordinateTx).LastCommitted {
+		if currentLC >= lc {
 			return true
 		}
 
@@ -130,10 +141,14 @@ func (mm *MtsManager) LcUpdater() {
 }
 
 func (mm *MtsManager) Executed(binlogEntry *common.DataEntry) {
+	mm.Executed0(binlogEntry.Coordinates.GetSequenceNumber())
+}
+
+func (mm *MtsManager) Executed0(seq int64) {
 	select {
 	case <-mm.shutdownCh:
 		return
-	case mm.chExecuted <- binlogEntry.Coordinates.GetSequenceNumber():
+	case mm.chExecuted <- seq:
 	}
 }