syncer: refactor sync dml

dm/config/task.go

@@ -243,8 +243,9 @@ type SyncerConfig struct {
 	MaxRetry int `yaml:"max-retry" toml:"max-retry" json:"max-retry"`
 
 	// refine following configs to top level configs?
-	AutoFixGTID      bool `yaml:"auto-fix-gtid" toml:"auto-fix-gtid" json:"auto-fix-gtid"`
-	EnableGTID       bool `yaml:"enable-gtid" toml:"enable-gtid" json:"enable-gtid"`
+	AutoFixGTID bool `yaml:"auto-fix-gtid" toml:"auto-fix-gtid" json:"auto-fix-gtid"`
+	EnableGTID  bool `yaml:"enable-gtid" toml:"enable-gtid" json:"enable-gtid"`
+	// deprecated
 	DisableCausality bool `yaml:"disable-detect" toml:"disable-detect" json:"disable-detect"`
 	SafeMode         bool `yaml:"safe-mode" toml:"safe-mode" json:"safe-mode"`
 	// deprecated, use `ansi-quotes` in top level config instead
@@ -620,6 +621,9 @@ func (c *TaskConfig) adjust() error {
 		if inst.Syncer.EnableANSIQuotes {
 			log.L().Warn("DM could discover proper ANSI_QUOTES, `enable-ansi-quotes` is no longer take effect")
 		}
+		if inst.Syncer.DisableCausality {
+			log.L().Warn("`disable-causality` is no longer take effect")
+		}
 
 		for _, name := range inst.ExpressionFilters {
 			if _, ok := c.ExprFilter[name]; !ok {

dm/config/task_test.go

@@ -700,7 +700,6 @@ func (t *testConfig) TestGenAndFromSubTaskConfigs(c *C) {
 				MaxRetry:                10,
 				AutoFixGTID:             true,
 				EnableGTID:              true,
-				DisableCausality:        false,
 				SafeMode:                true,
 			},
 			CleanDumpFile:    true,

dm/worker/status.go

@@ -17,7 +17,7 @@ import (
 	"encoding/json"
 	"sort"
 
-	"github.com/golang/protobuf/jsonpb"
+	"github.com/gogo/protobuf/jsonpb"
 	"go.uber.org/zap"
 
 	"github.com/pingcap/dm/dm/common"

syncer/causality.go

@@ -14,31 +14,106 @@
 package syncer
 
 import (
-	"github.com/pingcap/dm/pkg/terror"
+	"sync"
+	"time"
+
+	"go.uber.org/zap"
+
+	"github.com/pingcap/dm/pkg/log"
+	"github.com/pingcap/dm/syncer/metrics"
 )
 
 // causality provides a simple mechanism to improve the concurrency of SQLs execution under the premise of ensuring correctness.
 // causality groups sqls that maybe contain causal relationships, and syncer executes them linearly.
-// if some conflicts exist in more than one groups, then syncer waits all SQLs that are grouped be executed and reset causality.
+// if some conflicts exist in more than one groups, causality generate a conflict job and reset.
 // this mechanism meets quiescent consistency to ensure correctness.
 type causality struct {
 	relations map[string]string
+	outCh     chan *job
+	inCh      chan *job
+	logger    log.Logger
+	chanSize  int
+
+	// for metrics
+	task   string
+	source string
 }
 
-func newCausality() *causality {
-	return &causality{
+// newCausality creates a causality instance.
+func newCausality(chanSize int, task, source string, pLogger *log.Logger) *causality {
+	causality := &causality{
 		relations: make(map[string]string),
+		task:      task,
+		chanSize:  chanSize,
+		source:    source,
+		logger:    pLogger.WithFields(zap.String("component", "causality")),
 	}
+	return causality
 }
 
-func (c *causality) add(keys []string) error {
-	if len(keys) == 0 {
-		return nil
+// run runs a causality instance.
+func (c *causality) run(inCh chan *job) chan *job {
+	c.inCh = inCh
+	c.outCh = make(chan *job, c.chanSize)
+
+	var wg sync.WaitGroup
+
+	wg.Add(1)
+	go func() {
+		defer wg.Done()
+		c.runcausality()
+	}()
+
+	go func() {
+		defer c.close()
+		wg.Wait()
+	}()
+	return c.outCh
+}
+
+// runcausality receives dml jobs and returns causality jobs
+// When meet conflict, returns a conflict job.
+func (c *causality) runcausality() {
+	for j := range c.outCh {
+		metrics.QueueSizeGauge.WithLabelValues(c.task, "causality_input", c.source).Set(float64(len(c.outCh)))
+
+		startTime := time.Now()
+		if j.tp == flush {
+			c.reset()
+		} else {
+			keys := j.keys
+			var key string
+			if len(keys) > 0 {
+				key = keys[0]
+			}
+
+			// detectConflict before add
+			if c.detectConflict(keys) {
+				c.logger.Debug("meet causality key, will generate a conflict job to flush all sqls", zap.Strings("keys", keys))
+				c.outCh <- newConflictJob()
+				c.reset()
+			}
+			c.add(keys)
+			j.key = c.get(key)
+			c.logger.Debug("key for keys", zap.String("key", key), zap.Strings("keys", keys))
+		}
+		metrics.ConflictDetectDurationHistogram.WithLabelValues(c.task, c.source).Observe(time.Since(startTime).Seconds())
+
+		c.outCh <- j
 	}
+}
+
+// close closes outer channel.
+func (c *causality) close() {
+	close(c.outCh)
+}
 
-	if c.detectConflict(keys) {
-		return terror.ErrSyncUnitCausalityConflict.Generate()
+// add adds keys relation.
+func (c *causality) add(keys []string) {
+	if len(keys) == 0 {
+		return
 	}
+
 	// find causal key
 	selectedRelation := keys[0]
 	var nonExistKeys []string
@@ -53,13 +128,14 @@ func (c *causality) add(keys []string) error {
 	for _, key := range nonExistKeys {
 		c.relations[key] = selectedRelation
 	}
-	return nil
 }
 
+// get gets relation for a key.
 func (c *causality) get(key string) string {
 	return c.relations[key]
 }
 
+// reset resets relations.
 func (c *causality) reset() {
 	c.relations = make(map[string]string)
 }

syncer/causality_test.go

@@ -14,25 +14,97 @@
 package syncer
 
 import (
+	"time"
+
 	. "github.com/pingcap/check"
+	"github.com/pingcap/parser"
+	"github.com/pingcap/tidb-tools/pkg/filter"
+	"github.com/pingcap/tidb/util/mock"
+
+	"github.com/pingcap/dm/pkg/binlog"
+	"github.com/pingcap/dm/pkg/log"
+	"github.com/pingcap/dm/pkg/utils"
 )
 
-func (s *testSyncerSuite) TestCausality(c *C) {
-	ca := newCausality()
+func (s *testSyncerSuite) TestDetectConflict(c *C) {
+	ca := &causality{
+		relations: make(map[string]string),
+	}
 	caseData := []string{"test_1", "test_2", "test_3"}
 	excepted := map[string]string{
 		"test_1": "test_1",
 		"test_2": "test_1",
 		"test_3": "test_1",
 	}
-	c.Assert(ca.add(caseData), IsNil)
+	c.Assert(ca.detectConflict(caseData), IsFalse)
+	ca.add(caseData)
 	c.Assert(ca.relations, DeepEquals, excepted)
-	c.Assert(ca.add([]string{"test_4"}), IsNil)
+	c.Assert(ca.detectConflict([]string{"test_4"}), IsFalse)
+	ca.add([]string{"test_4"})
 	excepted["test_4"] = "test_4"
 	c.Assert(ca.relations, DeepEquals, excepted)
 	conflictData := []string{"test_4", "test_3"}
 	c.Assert(ca.detectConflict(conflictData), IsTrue)
-	c.Assert(ca.add(conflictData), NotNil)
 	ca.reset()
 	c.Assert(ca.relations, HasLen, 0)
 }
+
+func (s *testSyncerSuite) TestCasuality(c *C) {
+	p := parser.New()
+	se := mock.NewContext()
+	schema := "create table tb(a int primary key, b int unique);"
+	ti, err := createTableInfo(p, se, int64(0), schema)
+	c.Assert(err, IsNil)
+
+	jobCh := make(chan *job, 10)
+	logger := log.L()
+	causality := newCausality(1024, "task", "source", &logger)
+	causalityCh := causality.run(jobCh)
+	testCases := []struct {
+		op   opType
+		vals [][]interface{}
+	}{
+		{
+			op:   insert,
+			vals: [][]interface{}{{1, 2}},
+		},
+		{
+			op:   insert,
+			vals: [][]interface{}{{2, 3}},
+		},
+		{
+			op:   update,
+			vals: [][]interface{}{{2, 3}, {3, 4}},
+		},
+		{
+			op:   del,
+			vals: [][]interface{}{{1, 2}},
+		},
+		{
+			op:   insert,
+			vals: [][]interface{}{{1, 3}},
+		},
+	}
+	results := []opType{insert, insert, update, del, conflict, insert}
+	table := &filter.Table{Schema: "test", Name: "t1"}
+	location := binlog.NewLocation("")
+	ec := &eventContext{startLocation: &location, currentLocation: &location, lastLocation: &location}
+
+	for _, tc := range testCases {
+		var keys []string
+		for _, val := range tc.vals {
+			keys = append(keys, genMultipleKeys(ti, val, "tb")...)
+		}
+		job := newDMLJob(tc.op, table, table, "", nil, keys, ec)
+		jobCh <- job
+	}
+
+	c.Assert(utils.WaitSomething(30, 100*time.Millisecond, func() bool {
+		return len(causality.outCh) == len(results)
+	}), IsTrue)
+
+	for _, op := range results {
+		job := <-causalityCh
+		c.Assert(job.tp, Equals, op)
+	}
+}