Optimize the performance of bidirectional synchronous update table of…

… _drainer_repl_mark (#903) (#909)

drainer/loopbacksync/loopbacksync.go

@@ -16,6 +16,8 @@ package loopbacksync
 const (
  //MarkTableName mark table name
  MarkTableName = "retl._drainer_repl_mark"
+ //ID syncer worker coroutine id
+ ID = "id"
  //ChannelID channel id
  ChannelID = "channel_id"
  //Val val

pkg/loader/executor.go

@@ -18,6 +18,7 @@ import (
  gosql "database/sql"
  "fmt"
  "strings"
+ "sync/atomic"
  "time"
 
  "github.com/pingcap/tidb-binlog/drainer/loopbacksync"
@@ -32,20 +33,26 @@ import (
  "golang.org/x/sync/errgroup"
 )
 
-var defaultBatchSize = 128
+var (
+ defaultBatchSize = 128
+ defaultWorkerCount = 16
+ index int64
+)
 
 type executor struct {
  db *gosql.DB
  batchSize int
+ workerCount int
  info *loopbacksync.LoopBackSync
  queryHistogramVec *prometheus.HistogramVec
  refreshTableInfo func(schema string, table string) (info *tableInfo, err error)
 }
 
 func newExecutor(db *gosql.DB) *executor {
  exe := &executor{
- db: db,
- batchSize: defaultBatchSize,
+ db: db,
+ batchSize: defaultBatchSize,
+ workerCount: defaultWorkerCount,
  }
 
  return exe
@@ -65,6 +72,10 @@ func (e *executor) setSyncInfo(info *loopbacksync.LoopBackSync) {
  e.info = info
 }
 
+func (e *executor) setWorkerCount(workerCount int) {
+ e.workerCount = workerCount
+}
+
 func (e *executor) withQueryHistogramVec(queryHistogramVec *prometheus.HistogramVec) *executor {
  e.queryHistogramVec = queryHistogramVec
  return e
@@ -119,16 +130,36 @@ func (e *executor) updateMark(channel string, tx *tx) error {
  if e.info == nil {
  return nil
  }
- status := 1
- columns := fmt.Sprintf("(%s,%s,%s) VALUES(?,?,?)", loopbacksync.ChannelID, loopbacksync.Val, loopbacksync.ChannelInfo)
  var args []interface{}
- sql := fmt.Sprintf("INSERT INTO %s%s on duplicate key update %s=%s+1;", loopbacksync.MarkTableName, columns, loopbacksync.Val, loopbacksync.Val)
- args = append(args, e.info.ChannelID, status, channel)
- _, err := tx.autoRollbackExec(sql, args...)
+ sql := fmt.Sprintf("update %s set %s=%s+1 where %s=? and %s=? limit 1;", loopbacksync.MarkTableName, loopbacksync.Val, loopbacksync.Val, loopbacksync.ID, loopbacksync.ChannelID)
+ args = append(args, e.addIndex(), e.info.ChannelID)
+ _, err1 := tx.autoRollbackExec(sql, args...)
+ if err1 != nil {
+ return errors.Trace(err1)
+ }
+ return nil
+}
+
+func (e *executor) cleanChannelInfo() error {
+ if e.info == nil {
+ return nil
+ }
+ tx, err := e.begin()
  if err != nil {
  return errors.Trace(err)
  }
- return nil
+ var args []interface{}
+ sql := fmt.Sprintf("delete from %s where %s=? ", loopbacksync.MarkTableName, loopbacksync.ChannelID)
+ args = append(args, e.info.ChannelID)
+ _, err1 := tx.autoRollbackExec(sql, args...)
+ if err1 != nil {
+ return errors.Trace(err1)
+ }
+ err2 := tx.commit()
+ return errors.Trace(err2)
+}
+func (e *executor) addIndex() int64 {
+ return atomic.AddInt64(&index, 1) % ((int64)(e.workerCount))
 }
 
 // return a wrap of sql.Tx

pkg/loader/load.go

@@ -505,16 +505,63 @@ func (s *loaderImpl) createMarkTable() error {
  return nil
 }
 
+func (s *loaderImpl) initMarkTable() error {
+ if err := s.createMarkTable(); err != nil {
+ return errors.Trace(err)
+ }
+ return s.initMarkTableData()
+}
+func (s *loaderImpl) initMarkTableData() error {
+ tx, err := s.db.Begin()
+ if err != nil {
+ return errors.Trace(err)
+ }
+ status := 1
+ channel := ""
+ var builder strings.Builder
+ holder := "(?,?,?,?)"
+ columns := fmt.Sprintf("(%s,%s,%s,%s) ", loopbacksync.ID, loopbacksync.ChannelID, loopbacksync.Val, loopbacksync.ChannelInfo)
+ builder.WriteString("REPLACE INTO " + loopbacksync.MarkTableName + columns + " VALUES ")
+ for i := 0; i < s.workerCount; i++ {
+ if i > 0 {
+ builder.WriteByte(',')
+ }
+ builder.WriteString(holder)
+ }
+ var args []interface{}
+ for id := 0; id < s.workerCount; id++ {
+ args = append(args, id, s.loopBackSyncInfo.ChannelID, status, channel)
+ }
+ query := builder.String()
+ if _, err = tx.Exec(query, args...); err != nil {
+ log.Error("Exec fail, will rollback", zap.String("query", query), zap.Reflect("args", args), zap.Error(err))
+ if rbErr := tx.Rollback(); rbErr != nil {
+ log.Error("Auto rollback", zap.Error(rbErr))
+ }
+ return errors.Trace(err)
+ }
+ if err = tx.Commit(); err != nil {
+ return errors.Trace(err)
+ }
+ return nil
+}
+
+func (s *loaderImpl) cleanChannelInfo() {
+ executor := s.getExecutor()
+ _ = executor.cleanChannelInfo()
+}
+
 // Run will quit when meet any error, or all the txn are drained
 func (s *loaderImpl) Run() error {
  if s.loopBackSyncInfo != nil && s.loopBackSyncInfo.LoopbackControl {
- if err := s.createMarkTable(); err != nil {
+ if err := s.initMarkTable(); err != nil {
  return errors.Trace(err)
  }
  }
  txnManager := newTxnManager(1024, s.input)
  defer func() {
  log.Info("Run()... in Loader quit")
+ s.cleanChannelInfo()
  close(s.successTxn)
  txnManager.Close()
  }()
@@ -624,6 +671,7 @@ func (s *loaderImpl) getExecutor() *executor {
  e = e.withRefreshTableInfo(s.refreshTableInfo)
  }
  e.setSyncInfo(s.loopBackSyncInfo)
+ e.setWorkerCount(s.workerCount)
  if s.metrics != nil && s.metrics.QueryHistogramVec != nil {
  e = e.withQueryHistogramVec(s.metrics.QueryHistogramVec)
  }

pkg/loader/model.go

@@ -192,7 +192,7 @@ func (dml *DML) updateSQL() (sql string, args []interface{}) {
 }
 
 func createMarkTableDDL() string {
- sql := fmt.Sprintf("CREATE TABLE If Not Exists %s ( %s bigint primary key, %s bigint DEFAULT 0, %s varchar(64));", loopbacksync.MarkTableName, loopbacksync.ChannelID, loopbacksync.Val, loopbacksync.ChannelInfo)
+ sql := fmt.Sprintf("CREATE TABLE If Not Exists %s (%s bigint not null,%s bigint not null DEFAULT 0, %s bigint DEFAULT 0, %s varchar(64) ,PRIMARY KEY (%s,%s));", loopbacksync.MarkTableName, loopbacksync.ID, loopbacksync.ChannelID, loopbacksync.Val, loopbacksync.ChannelInfo, loopbacksync.ID, loopbacksync.ChannelID)
  return sql
 }
 

pkg/loader/model_test.go

@@ -242,11 +242,10 @@ func (s *SQLSuite) TestUpdateMarkSQL(c *check.C) {
  db, mock, err := sqlmock.New()
  c.Assert(err, check.IsNil)
  defer db.Close()
- columns := fmt.Sprintf("(%s,%s,%s) VALUES(?,?,?)", loopbacksync.ChannelID, loopbacksync.Val, loopbacksync.ChannelInfo)
- sql := fmt.Sprintf("INSERT INTO %s%s on duplicate key update %s=%s+1;", loopbacksync.MarkTableName, columns, loopbacksync.Val, loopbacksync.Val)
+ sql := fmt.Sprintf("update %s set %s=%s+1 where %s=? and %s=? limit 1;", loopbacksync.MarkTableName, loopbacksync.Val, loopbacksync.Val, loopbacksync.ID, loopbacksync.ChannelID)
  mock.ExpectBegin()
  mock.ExpectExec(regexp.QuoteMeta(sql)).
- WithArgs(100, 1, "").WillReturnResult(sqlmock.NewResult(1, 1))
+ WithArgs(1, 100).WillReturnResult(sqlmock.NewResult(1, 1))
  mock.ExpectCommit()
  e := newExecutor(db)
  tx, err := e.begin()
@@ -261,6 +260,6 @@ func (s *SQLSuite) TestUpdateMarkSQL(c *check.C) {
 }
 func (s *SQLSuite) TestCreateMarkTable(c *check.C) {
  sql := createMarkTableDDL()
- sql1 := fmt.Sprintf("CREATE TABLE If Not Exists %s ( %s bigint primary key, %s bigint DEFAULT 0, %s varchar(64));", loopbacksync.MarkTableName, loopbacksync.ChannelID, loopbacksync.Val, loopbacksync.ChannelInfo)
+ sql1 := fmt.Sprintf("CREATE TABLE If Not Exists %s (%s bigint not null,%s bigint not null DEFAULT 0, %s bigint DEFAULT 0, %s varchar(64) ,PRIMARY KEY (%s,%s));", loopbacksync.MarkTableName, loopbacksync.ID, loopbacksync.ChannelID, loopbacksync.Val, loopbacksync.ChannelInfo, loopbacksync.ID, loopbacksync.ChannelID)
  c.Assert(sql, check.Equals, sql1)
 }