syncer: improve performance

dm/config/task.go

@@ -55,8 +55,10 @@ var (
 	defaultPoolSize = 16
 	defaultDir      = "./dumped_data"
 	// SyncerConfig
-	defaultWorkerCount = 16
-	defaultBatch       = 100
+	defaultWorkerCount             = 16
+	defaultBatch                   = 100
+	defaultQueueSize               = 5120
+	defaultCheckpointFlushInterval = 30
 )
 
 // Meta represents binlog's meta pos
@@ -193,6 +195,11 @@ type SyncerConfig struct {
 	MetaFile    string `yaml:"meta-file" toml:"meta-file" json:"meta-file"` // meta filename, used only when load SubConfig directly
 	WorkerCount int    `yaml:"worker-count" toml:"worker-count" json:"worker-count"`
 	Batch       int    `yaml:"batch" toml:"batch" json:"batch"`
+	QueueSize   int    `yaml:"queue-size" toml:"queue-size" json:"queue-size"`
+
+	// checkpoint flush interval in seconds.
+	CheckpointFlushInterval int `yaml:"checkpoint-flush-interval" toml:"checkpoint-flush-interval" json:"checkpoint-flush-interval"`
+
 	// deprecated
 	MaxRetry int `yaml:"max-retry" toml:"max-retry" json:"max-retry"`
 
@@ -206,8 +213,10 @@ type SyncerConfig struct {
 
 func defaultSyncerConfig() SyncerConfig {
 	return SyncerConfig{
-		WorkerCount: defaultWorkerCount,
-		Batch:       defaultBatch,
+		WorkerCount:             defaultWorkerCount,
+		Batch:                   defaultBatch,
+		QueueSize:               defaultQueueSize,
+		CheckpointFlushInterval: defaultCheckpointFlushInterval,
 	}
 }
 

syncer/checkpoint.go

@@ -46,8 +46,6 @@ var (
 	globalCpTable        = "" // global checkpoint's cp_table
 	maxCheckPointTimeout = "1m"
 	minCheckpoint        = mysql.Position{Pos: 4}
-
-	maxCheckPointSaveTime = 30 * time.Second
 )
 
 // NOTE: now we sync from relay log, so not add GTID support yet
@@ -453,7 +451,7 @@ func (cp *RemoteCheckPoint) String() string {
 func (cp *RemoteCheckPoint) CheckGlobalPoint() bool {
 	cp.RLock()
 	defer cp.RUnlock()
-	return time.Since(cp.globalPointSaveTime) >= maxCheckPointSaveTime
+	return time.Since(cp.globalPointSaveTime) >= time.Duration(cp.cfg.CheckpointFlushInterval)*time.Second
 }
 
 // Rollback implements CheckPoint.Rollback

syncer/metrics.go

@@ -64,6 +64,15 @@ var (
 			Buckets:   prometheus.ExponentialBuckets(0.00005, 2, 21),
 		}, []string{"task"})
 
+	addJobDurationHistogram = prometheus.NewHistogramVec(
+		prometheus.HistogramOpts{
+			Namespace: "dm",
+			Subsystem: "syncer",
+			Name:      "add_job_duration",
+			Help:      "bucketed histogram of add a job to the queue time (s)",
+			Buckets:   prometheus.ExponentialBuckets(0.00005, 2, 21),
+		}, []string{"type", "task", "queueNo"})
+
 	binlogSkippedEventsTotal = prometheus.NewCounterVec(
 		prometheus.CounterOpts{
 			Namespace: "dm",
@@ -205,6 +214,7 @@ func RegisterMetrics(registry *prometheus.Registry) {
 	registry.MustRegister(binlogEventSizeHistogram)
 	registry.MustRegister(binlogEvent)
 	registry.MustRegister(conflictDetectDurationHistogram)
+	registry.MustRegister(addJobDurationHistogram)
 	registry.MustRegister(binlogSkippedEventsTotal)
 	registry.MustRegister(addedJobsTotal)
 	registry.MustRegister(finishedJobsTotal)

syncer/syncer.go

@@ -275,7 +275,7 @@ func (s *Syncer) newJobChans(count int) {
 	s.closeJobChans()
 	s.jobs = make([]chan *job, 0, count)
 	for i := 0; i < count; i++ {
-		s.jobs = append(s.jobs, make(chan *job, 1000))
+		s.jobs = append(s.jobs, make(chan *job, s.cfg.QueueSize))
 	}
 	s.jobsClosed.Set(false)
 }
@@ -757,7 +757,10 @@ func (s *Syncer) addJob(job *job) error {
 		// ugly code addJob and sync, refine it later
 		s.jobWg.Add(s.cfg.WorkerCount)
 		for i := 0; i < s.cfg.WorkerCount; i++ {
+			startTime := time.Now()
 			s.jobs[i] <- job
+			// flush for every DML queue
+			addJobDurationHistogram.WithLabelValues("flush", s.cfg.Name, s.queueBucketMapping[i]).Observe(time.Since(startTime).Seconds())
 		}
 		s.jobWg.Wait()
 		finishedJobsTotal.WithLabelValues("flush", s.cfg.Name, adminQueueName).Inc()
@@ -767,15 +770,19 @@ func (s *Syncer) addJob(job *job) error {
 		addedJobsTotal.WithLabelValues("ddl", s.cfg.Name, adminQueueName).Inc()
 		s.jobWg.Add(1)
 		queueBucket = s.cfg.WorkerCount
+		startTime := time.Now()
 		s.jobs[queueBucket] <- job
+		addJobDurationHistogram.WithLabelValues("ddl", s.cfg.Name, adminQueueName).Observe(time.Since(startTime).Seconds())
 		if job.ddlExecItem != nil {
 			execDDLReq = job.ddlExecItem.req
 		}
 	case insert, update, del:
 		s.jobWg.Add(1)
 		queueBucket = int(utils.GenHashKey(job.key)) % s.cfg.WorkerCount
 		s.addCount(false, s.queueBucketMapping[queueBucket], job.tp, 1)
+		startTime := time.Now()
 		s.jobs[queueBucket] <- job
+		addJobDurationHistogram.WithLabelValues(job.tp.String(), s.cfg.Name, s.queueBucketMapping[queueBucket]).Observe(time.Since(startTime).Seconds())
 	}
 
 	if s.tracer.Enable() {