*: improve performance

dm/config/subtask.go

@@ -287,6 +287,13 @@ func (c *SubTaskConfig) Adjust(verifyDecryptPassword bool) error {
 		c.LoaderConfig.Dir += dirSuffix
 	}
 
+	if c.SyncerConfig.QueueSize == 0 {
+		c.SyncerConfig.QueueSize = defaultQueueSize
+	}
+	if c.SyncerConfig.CheckpointFlushInterval == 0 {
+		c.SyncerConfig.CheckpointFlushInterval = defaultCheckpointFlushInterval
+	}
+
 	c.From.Adjust()
 	c.To.Adjust()
 

dm/config/task.go

@@ -61,8 +61,10 @@ var (
 	defaultPoolSize = 16
 	defaultDir      = "./dumped_data"
 	// SyncerConfig
-	defaultWorkerCount = 16
-	defaultBatch       = 100
+	defaultWorkerCount             = 16
+	defaultBatch                   = 100
+	defaultQueueSize               = 1024 // do not give too large default value to avoid OOM
+	defaultCheckpointFlushInterval = 30   // in seconds
 )
 
 // Meta represents binlog's meta pos
@@ -204,6 +206,10 @@ 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"`
 
@@ -217,8 +223,10 @@ type SyncerConfig struct {
 
 func defaultSyncerConfig() SyncerConfig {
 	return SyncerConfig{
-		WorkerCount: defaultWorkerCount,
-		Batch:       defaultBatch,
+		WorkerCount:             defaultWorkerCount,
+		Batch:                   defaultBatch,
+		QueueSize:               defaultQueueSize,
+		CheckpointFlushInterval: defaultCheckpointFlushInterval,
 	}
 }