relay/syncer: relay notifies syncer of new write to reduce sync latency

cmd/dm-syncer/main.go

@@ -74,7 +74,7 @@ func main() {
 		log.L().Info("", zap.Stringer("dm-syncer conf", conf))
 	})
 
-	sync := syncer.NewSyncer(conf, nil) // do not support shard DDL for singleton syncer.
+	sync := syncer.NewSyncer(conf, nil, nil) // do not support shard DDL for singleton syncer.
 	sc := make(chan os.Signal, 1)
 	signal.Notify(sc, syscall.SIGHUP, syscall.SIGINT, syscall.SIGTERM, syscall.SIGQUIT)
 

dm/worker/relay.go

@@ -51,6 +51,10 @@ type RelayHolder interface {
 	Result() *pb.ProcessResult
 	// Update updates relay config online
 	Update(ctx context.Context, cfg *config.SourceConfig) error
+	// RegisterListener registers a relay listener
+	RegisterListener(el relay.Listener)
+	// UnRegisterListener unregisters a relay listener
+	UnRegisterListener(el relay.Listener)
 }
 
 // NewRelayHolder is relay holder initializer
@@ -307,6 +311,18 @@ func (h *realRelayHolder) EarliestActiveRelayLog() *streamer.RelayLogInfo {
 	return h.relay.ActiveRelayLog()
 }
 
+func (h *realRelayHolder) RegisterListener(el relay.Listener) {
+	h.Lock()
+	defer h.Unlock()
+	h.relay.RegisterListener(el)
+}
+
+func (h *realRelayHolder) UnRegisterListener(el relay.Listener) {
+	h.Lock()
+	defer h.Unlock()
+	h.relay.UnRegisterListener(el)
+}
+
 /******************** dummy relay holder ********************/
 
 type dummyRelayHolder struct {
@@ -424,3 +440,9 @@ func (d *dummyRelayHolder) Stage() pb.Stage {
 	defer d.Unlock()
 	return d.stage
 }
+
+func (d *dummyRelayHolder) RegisterListener(el relay.Listener) {
+}
+
+func (d *dummyRelayHolder) UnRegisterListener(el relay.Listener) {
+}

dm/worker/relay_test.go

@@ -47,6 +47,12 @@ type DummyRelay struct {
 	reloadErr     error
 }
 
+func (d *DummyRelay) RegisterListener(el relay.Listener) {
+}
+
+func (d *DummyRelay) UnRegisterListener(el relay.Listener) {
+}
+
 // NewDummyRelay creates an instance of dummy Relay.
 func NewDummyRelay(cfg *relay.Config) relay.Process {
 	return &DummyRelay{}

dm/worker/server_test.go

@@ -22,6 +22,8 @@ import (
 	"testing"
 	"time"
 
+	"github.com/pingcap/dm/pkg/streamer"
+
 	"github.com/go-mysql-org/go-mysql/mysql"
 	. "github.com/pingcap/check"
 	"github.com/pingcap/failpoint"
@@ -120,7 +122,7 @@ func (t *testServer) TestServer(c *C) {
 		cfg.UseRelay = false
 		return NewRealSubTask(cfg, etcdClient, worker)
 	}
-	createUnits = func(cfg *config.SubTaskConfig, etcdClient *clientv3.Client, worker string) []unit.Unit {
+	createUnits = func(cfg *config.SubTaskConfig, etcdClient *clientv3.Client, worker string, notifier streamer.EventNotifier) []unit.Unit {
 		mockDumper := NewMockUnit(pb.UnitType_Dump)
 		mockLoader := NewMockUnit(pb.UnitType_Load)
 		mockSync := NewMockUnit(pb.UnitType_Sync)

dm/worker/source_worker.go

@@ -353,6 +353,8 @@ func (w *SourceWorker) EnableRelay() (err error) {
 		w.observeRelayStage(w.relayCtx, w.etcdClient, revRelay)
 	}()
 
+	w.relayHolder.RegisterListener(w.subTaskHolder)
+
 	w.relayEnabled.Store(true)
 	w.l.Info("relay enabled")
 	w.subTaskHolder.resetAllSubTasks(true)
@@ -385,6 +387,7 @@ func (w *SourceWorker) DisableRelay() {
 	if w.relayHolder != nil {
 		r := w.relayHolder
 		w.relayHolder = nil
+		r.UnRegisterListener(w.subTaskHolder)
 		r.Close()
 	}
 	if w.relayPurger != nil {

dm/worker/source_worker_test.go

@@ -20,6 +20,8 @@ import (
 	"sync/atomic"
 	"time"
 
+	"github.com/pingcap/dm/pkg/streamer"
+
 	"github.com/DATA-DOG/go-sqlmock"
 	. "github.com/pingcap/check"
 	"github.com/pingcap/failpoint"
@@ -241,7 +243,7 @@ var _ = Suite(&testWorkerFunctionalities{})
 func (t *testWorkerFunctionalities) SetUpSuite(c *C) {
 	NewRelayHolder = NewDummyRelayHolder
 	NewSubTask = NewRealSubTask
-	createUnits = func(cfg *config.SubTaskConfig, etcdClient *clientv3.Client, worker string) []unit.Unit {
+	createUnits = func(cfg *config.SubTaskConfig, etcdClient *clientv3.Client, worker string, notifier streamer.EventNotifier) []unit.Unit {
 		atomic.AddInt32(&t.createUnitCount, 1)
 		mockDumper := NewMockUnit(pb.UnitType_Dump)
 		mockLoader := NewMockUnit(pb.UnitType_Load)
@@ -416,7 +418,7 @@ func (t *testWorkerEtcdCompact) SetUpSuite(c *C) {
 		cfg.UseRelay = false
 		return NewRealSubTask(cfg, etcdClient, worker)
 	}
-	createUnits = func(cfg *config.SubTaskConfig, etcdClient *clientv3.Client, worker string) []unit.Unit {
+	createUnits = func(cfg *config.SubTaskConfig, etcdClient *clientv3.Client, worker string, notifier streamer.EventNotifier) []unit.Unit {
 		mockDumper := NewMockUnit(pb.UnitType_Dump)
 		mockLoader := NewMockUnit(pb.UnitType_Load)
 		mockSync := NewMockUnit(pb.UnitType_Sync)

dm/worker/subtask.go

@@ -18,6 +18,8 @@ import (
 	"sync"
 	"time"
 
+	"github.com/pingcap/dm/pkg/streamer"
+
 	"github.com/go-mysql-org/go-mysql/mysql"
 	"github.com/pingcap/failpoint"
 	"github.com/prometheus/client_golang/prometheus"
@@ -44,12 +46,22 @@ const (
 	waitRelayCatchupTimeout = 30 * time.Second
 )
 
+type relayNotifier struct {
+	// ch with size = 1, we only need to be notified whether binlog file of relay changed, not how many times
+	ch chan interface{}
+}
+
+// Notified implements streamer.EventNotifier.
+func (r relayNotifier) Notified() chan interface{} {
+	return r.ch
+}
+
 // createRealUnits is subtask units initializer
 // it can be used for testing.
 var createUnits = createRealUnits
 
 // createRealUnits creates process units base on task mode.
-func createRealUnits(cfg *config.SubTaskConfig, etcdClient *clientv3.Client, workerName string) []unit.Unit {
+func createRealUnits(cfg *config.SubTaskConfig, etcdClient *clientv3.Client, workerName string, notifier streamer.EventNotifier) []unit.Unit {
 	failpoint.Inject("mockCreateUnitsDumpOnly", func(_ failpoint.Value) {
 		log.L().Info("create mock worker units with dump unit only", zap.String("failpoint", "mockCreateUnitsDumpOnly"))
 		failpoint.Return([]unit.Unit{dumpling.NewDumpling(cfg)})
@@ -64,7 +76,7 @@ func createRealUnits(cfg *config.SubTaskConfig, etcdClient *clientv3.Client, wor
 		} else {
 			us = append(us, loader.NewLightning(cfg, etcdClient, workerName))
 		}
-		us = append(us, syncer.NewSyncer(cfg, etcdClient))
+		us = append(us, syncer.NewSyncer(cfg, etcdClient, notifier))
 	case config.ModeFull:
 		// NOTE: maybe need another checker in the future?
 		us = append(us, dumpling.NewDumpling(cfg))
@@ -74,7 +86,7 @@ func createRealUnits(cfg *config.SubTaskConfig, etcdClient *clientv3.Client, wor
 			us = append(us, loader.NewLightning(cfg, etcdClient, workerName))
 		}
 	case config.ModeIncrement:
-		us = append(us, syncer.NewSyncer(cfg, etcdClient))
+		us = append(us, syncer.NewSyncer(cfg, etcdClient, notifier))
 	default:
 		log.L().Error("unsupported task mode", zap.String("subtask", cfg.Name), zap.String("task mode", cfg.Mode))
 	}
@@ -108,6 +120,8 @@ type SubTask struct {
 	etcdClient *clientv3.Client
 
 	workerName string
+
+	notifier streamer.EventNotifier
 }
 
 // NewSubTask is subtask initializer
@@ -130,14 +144,15 @@ func NewSubTaskWithStage(cfg *config.SubTaskConfig, stage pb.Stage, etcdClient *
 		cancel:     cancel,
 		etcdClient: etcdClient,
 		workerName: workerName,
+		notifier:   &relayNotifier{ch: make(chan interface{}, 1)},
 	}
 	updateTaskMetric(st.cfg.Name, st.cfg.SourceID, st.stage, st.workerName)
 	return &st
 }
 
 // initUnits initializes the sub task processing units.
 func (st *SubTask) initUnits() error {
-	st.units = createUnits(st.cfg, st.etcdClient, st.workerName)
+	st.units = createUnits(st.cfg, st.etcdClient, st.workerName, st.notifier)
 	if len(st.units) < 1 {
 		return terror.ErrWorkerNoAvailUnits.Generate(st.cfg.Name, st.cfg.Mode)
 	}
@@ -724,3 +739,11 @@ func updateTaskMetric(task, sourceID string, stage pb.Stage, workerName string)
 		taskState.WithLabelValues(task, sourceID, workerName).Set(float64(stage))
 	}
 }
+
+func (st *SubTask) relayNotify() {
+	// skip if there's pending notify
+	select {
+	case st.notifier.Notified() <- struct{}{}:
+	default:
+	}
+}

dm/worker/subtask_holder.go

@@ -16,6 +16,8 @@ package worker
 import (
 	"context"
 	"sync"
+
+	"github.com/go-mysql-org/go-mysql/replication"
 )
 
 // subTaskHolder holds subtask instances.
@@ -88,3 +90,14 @@ func (h *subTaskHolder) getAllSubTasks() map[string]*SubTask {
 	}
 	return result
 }
+
+// OnEvent implements relay.Listener
+// only syncer unit of subtask need to be notified, but it's much simpler and less error-prone to manage it here
+// as relay event need to broadcast to every syncer(most subtask have a syncer).
+func (h *subTaskHolder) OnEvent(e *replication.BinlogEvent) {
+	h.mu.RLock()
+	defer h.mu.RUnlock()
+	for _, s := range h.subTasks {
+		s.relayNotify()
+	}
+}

dm/worker/subtask_test.go

@@ -18,6 +18,8 @@ import (
 	"strings"
 	"time"
 
+	"github.com/pingcap/dm/pkg/streamer"
+
 	"github.com/pingcap/dm/dm/config"
 	"github.com/pingcap/dm/dm/pb"
 	"github.com/pingcap/dm/dm/unit"
@@ -47,24 +49,24 @@ func (t *testSubTask) TestCreateUnits(c *C) {
 		Mode: "xxx",
 	}
 	worker := "worker"
-	c.Assert(createUnits(cfg, nil, worker), HasLen, 0)
+	c.Assert(createUnits(cfg, nil, worker, nil), HasLen, 0)
 
 	cfg.Mode = config.ModeFull
-	unitsFull := createUnits(cfg, nil, worker)
+	unitsFull := createUnits(cfg, nil, worker, nil)
 	c.Assert(unitsFull, HasLen, 2)
 	_, ok := unitsFull[0].(*dumpling.Dumpling)
 	c.Assert(ok, IsTrue)
 	_, ok = unitsFull[1].(*loader.Loader)
 	c.Assert(ok, IsTrue)
 
 	cfg.Mode = config.ModeIncrement
-	unitsIncr := createUnits(cfg, nil, worker)
+	unitsIncr := createUnits(cfg, nil, worker, nil)
 	c.Assert(unitsIncr, HasLen, 1)
 	_, ok = unitsIncr[0].(*syncer.Syncer)
 	c.Assert(ok, IsTrue)
 
 	cfg.Mode = config.ModeAll
-	unitsAll := createUnits(cfg, nil, worker)
+	unitsAll := createUnits(cfg, nil, worker, nil)
 	c.Assert(unitsAll, HasLen, 3)
 	_, ok = unitsAll[0].(*dumpling.Dumpling)
 	c.Assert(ok, IsTrue)
@@ -176,7 +178,7 @@ func (t *testSubTask) TestSubTaskNormalUsage(c *C) {
 	defer func() {
 		createUnits = createRealUnits
 	}()
-	createUnits = func(cfg *config.SubTaskConfig, etcdClient *clientv3.Client, worker string) []unit.Unit {
+	createUnits = func(cfg *config.SubTaskConfig, etcdClient *clientv3.Client, worker string, notifier streamer.EventNotifier) []unit.Unit {
 		return nil
 	}
 	st.Run(pb.Stage_Running)
@@ -185,7 +187,7 @@ func (t *testSubTask) TestSubTaskNormalUsage(c *C) {
 
 	mockDumper := NewMockUnit(pb.UnitType_Dump)
 	mockLoader := NewMockUnit(pb.UnitType_Load)
-	createUnits = func(cfg *config.SubTaskConfig, etcdClient *clientv3.Client, worker string) []unit.Unit {
+	createUnits = func(cfg *config.SubTaskConfig, etcdClient *clientv3.Client, worker string, notifier streamer.EventNotifier) []unit.Unit {
 		return []unit.Unit{mockDumper, mockLoader}
 	}
 
@@ -296,7 +298,7 @@ func (t *testSubTask) TestPauseAndResumeSubtask(c *C) {
 	defer func() {
 		createUnits = createRealUnits
 	}()
-	createUnits = func(cfg *config.SubTaskConfig, etcdClient *clientv3.Client, worker string) []unit.Unit {
+	createUnits = func(cfg *config.SubTaskConfig, etcdClient *clientv3.Client, worker string, notifier streamer.EventNotifier) []unit.Unit {
 		return []unit.Unit{mockDumper, mockLoader}
 	}
 
@@ -420,7 +422,7 @@ func (t *testSubTask) TestSubtaskWithStage(c *C) {
 	defer func() {
 		createUnits = createRealUnits
 	}()
-	createUnits = func(cfg *config.SubTaskConfig, etcdClient *clientv3.Client, worker string) []unit.Unit {
+	createUnits = func(cfg *config.SubTaskConfig, etcdClient *clientv3.Client, worker string, notifier streamer.EventNotifier) []unit.Unit {
 		return []unit.Unit{mockDumper, mockLoader}
 	}
 
@@ -445,7 +447,7 @@ func (t *testSubTask) TestSubtaskWithStage(c *C) {
 
 	st = NewSubTaskWithStage(cfg, pb.Stage_Finished, nil, "worker")
 	c.Assert(st.Stage(), DeepEquals, pb.Stage_Finished)
-	createUnits = func(cfg *config.SubTaskConfig, etcdClient *clientv3.Client, worker string) []unit.Unit {
+	createUnits = func(cfg *config.SubTaskConfig, etcdClient *clientv3.Client, worker string, notifier streamer.EventNotifier) []unit.Unit {
 		return []unit.Unit{mockDumper, mockLoader}
 	}