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

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
@@ -315,6 +319,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 {
@@ -432,3 +448,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/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/subtask.go

@@ -100,6 +100,7 @@ type SubTask struct {
 	units    []unit.Unit // units do job one by one
 	currUnit unit.Unit
 	prevUnit unit.Unit
+	syncer   *syncer.Syncer
 	resultWg sync.WaitGroup
 
 	stage  pb.Stage          // stage of current sub task
@@ -190,11 +191,23 @@ func (st *SubTask) initUnits() error {
 
 	needCloseUnits = st.units[:skipIdx]
 	st.units = st.units[skipIdx:]
+	st.postInitSyncer()
 
 	st.setCurrUnit(st.units[0])
 	return nil
 }
 
+func (st *SubTask) postInitSyncer() {
+	// TODO, right now initUnits create units first and then remove unnecessary units(before first non fresh unit)
+	// maybe can be refactored into check first, then create, so we don't need to loop all units to get syncer here
+	for _, u := range st.units {
+		if s, ok := u.(*syncer.Syncer); ok {
+			st.syncer = s
+			break
+		}
+	}
+}
+
 // Run runs the sub task.
 // TODO: check concurrent problems.
 func (st *SubTask) Run(expectStage pb.Stage) {
@@ -724,3 +737,13 @@ func updateTaskMetric(task, sourceID string, stage pb.Stage, workerName string)
 		taskState.WithLabelValues(task, sourceID, workerName).Set(float64(stage))
 	}
 }
+
+func (st *SubTask) relayNotify() {
+	if st.syncer != nil {
+		// skip if there's pending notify
+		select {
+		case st.syncer.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()
+	}
+}

loader/lightning.go

@@ -18,16 +18,7 @@ import (
 	"path/filepath"
 	"sync"
 
-	"github.com/docker/go-units"
-	"github.com/pingcap/failpoint"
-	"github.com/pingcap/parser/mysql"
 	"github.com/pingcap/tidb-tools/pkg/dbutil"
-	"github.com/pingcap/tidb/br/pkg/lightning"
-	"github.com/pingcap/tidb/br/pkg/lightning/common"
-	lcfg "github.com/pingcap/tidb/br/pkg/lightning/config"
-	"go.etcd.io/etcd/clientv3"
-	"go.uber.org/atomic"
-	"go.uber.org/zap"
 
 	"github.com/pingcap/dm/dm/config"
 	"github.com/pingcap/dm/dm/pb"
@@ -37,6 +28,16 @@ import (
 	tcontext "github.com/pingcap/dm/pkg/context"
 	"github.com/pingcap/dm/pkg/log"
 	"github.com/pingcap/dm/pkg/utils"
+
+	"github.com/docker/go-units"
+	"github.com/pingcap/failpoint"
+	"github.com/pingcap/parser/mysql"
+	"github.com/pingcap/tidb/br/pkg/lightning"
+	"github.com/pingcap/tidb/br/pkg/lightning/common"
+	lcfg "github.com/pingcap/tidb/br/pkg/lightning/config"
+	"go.etcd.io/etcd/clientv3"
+	"go.uber.org/atomic"
+	"go.uber.org/zap"
 )
 
 const (