fix barrier block logic (#358)

* fix barrier block logic

* fix barrier block logic

Author: sdojjy

maintainer/barrier.go

@@ -23,9 +23,9 @@ import (
 // Barrier manage the block events for the changefeed
 // the block event processing logic:
 // 1. dispatcher report an event to maintainer, like ddl, sync point
-// 2. maintainer wait for all dispatchers to reach the same commit ts (all dispatchers will report the same event)
+// 2. maintainer wait for all dispatchers reporting block event (all dispatchers will report the same event)
 // 3. maintainer choose one dispatcher to write(tack an action) the event to downstream, (resend logic is needed)
-// 4. maintainer wait for the selected dispatcher advance its checkpoint ts,(means it already finished the write action), (resend logic is needed)
+// 4. maintainer wait for the selected dispatcher advance its checkpoint ts, checkpoint ts >= block ts,(means it already finished the write action), (resend logic is needed)
 // 5. maintainer send pass action to all other dispatchers. (resend logic is needed)
 // 6. maintainer wait for all dispatchers advance checkpoints, and cleanup memory
 type Barrier struct {
@@ -90,25 +90,10 @@ func (b *Barrier) handleNoStateHeartbeat(dispatcherID common.DispatcherID, check
 	if !ok {
 		return nil
 	}
-	// no block event send ,but reached the block point
-	if checkpointTs == event.commitTs {
-		action := event.dispatcherReachedBlockTs(dispatcherID)
-		// all dispatcher reported heartbeat, select one to write
-		if action != nil {
-			dispatcherStatus := &heartbeatpb.DispatcherStatus{
-				InfluencedDispatchers: &heartbeatpb.InfluencedDispatchers{
-					InfluenceType: heartbeatpb.InfluenceType_Normal,
-					DispatcherIDs: []*heartbeatpb.DispatcherID{event.writerDispatcher.ToPB()},
-				},
-				Action: action,
-			}
-			return dispatcherStatus
-		}
-	}
 
 	// there is a block event and the dispatcher advanced its checkpoint ts
 	// which means we have sent pass or write action to it
-	if checkpointTs > event.commitTs {
+	if checkpointTs >= event.commitTs {
 		// the writer already synced ddl to downstream
 		if event.writerDispatcher == dispatcherID {
 			// schedule new and removed tasks
@@ -143,10 +128,8 @@ func (b *Barrier) handleStateHeartbeat(changefeedID string,
 	if blockState.IsBlocked {
 		// insert an event, or get the old one event check if the event is already tracked
 		event := b.getOrInsertNewEvent(changefeedID, dispatcherID, blockState)
-		// the dispatcher already reached the block event block ts, check whether we need to send write action
-		if status.CheckpointTs == blockState.BlockTs {
-			dispatcherStatus.Action = event.dispatcherReachedBlockTs(dispatcherID)
-		}
+		// check if all dispatchers already reported the block event, and check whether we need to send write action
+		dispatcherStatus.Action = event.dispatcherReachedBlockTs(dispatcherID)
 	} else {
 		// it's not a blocked event, it must be sent by table event trigger dispatcher
 		// the ddl already synced to downstream , e.g.: create table, drop table

maintainer/barrier_event.go

@@ -43,7 +43,7 @@ type BarrierEvent struct {
 	newTables      []*heartbeatpb.Table
 	schemaIDChange []*heartbeatpb.SchemaIDChange
 
-	advancedDispatchers map[common.DispatcherID]bool
+	reportedDispatchers map[common.DispatcherID]bool
 	lastResendTime      time.Time
 }
 
@@ -58,7 +58,7 @@ func NewBlockEvent(cfID string, scheduler *Controller,
 		newTables:           status.NeedAddedTables,
 		dropDispatchers:     status.NeedDroppedTables,
 		schemaIDChange:      status.UpdatedSchemas,
-		advancedDispatchers: make(map[common.DispatcherID]bool),
+		reportedDispatchers: make(map[common.DispatcherID]bool),
 		lastResendTime:      time.Time{},
 	}
 	if event.blockedDispatchers != nil && event.blockedDispatchers.InfluenceType == heartbeatpb.InfluenceType_Normal {
@@ -122,12 +122,12 @@ func (be *BarrierEvent) allDispatcherReported() bool {
 	// todo: we should check ddl dispatcher checkpoint ts here, because the size may affect by ddls?
 	switch be.blockedDispatchers.InfluenceType {
 	case heartbeatpb.InfluenceType_DB:
-		return len(be.advancedDispatchers) >=
+		return len(be.reportedDispatchers) >=
 			len(be.scheduler.GetTasksBySchemaID(be.blockedDispatchers.SchemaID))
 	case heartbeatpb.InfluenceType_All:
-		return len(be.advancedDispatchers) >= be.scheduler.TaskSize()
+		return len(be.reportedDispatchers) >= be.scheduler.TaskSize()
 	case heartbeatpb.InfluenceType_Normal:
-		return len(be.advancedDispatchers) >= len(be.blockedTasks)
+		return len(be.reportedDispatchers) >= len(be.blockedTasks)
 	}
 	return false
 }
@@ -179,18 +179,19 @@ func (be *BarrierEvent) sendPassAction() []*messaging.TargetMessage {
 	return msgs
 }
 
-// dispatcherReachedBlockTs check if all the dispatchers reached the block ts
+// dispatcherReachedBlockTs check if all the dispatchers reported the block events,
+// if so, select one dispatcher to write, currently choose the last one
 func (be *BarrierEvent) dispatcherReachedBlockTs(dispatcherID common.DispatcherID) *heartbeatpb.DispatcherAction {
 	if be.selected {
 		return nil
 	}
-	be.advancedDispatchers[dispatcherID] = true
+	be.reportedDispatchers[dispatcherID] = true
 	// all dispatcher reported heartbeat, select the last one to write
 	if be.allDispatcherReported() {
 		be.writerDispatcher = dispatcherID
 		be.selected = true
 		// release memory
-		be.advancedDispatchers = nil
+		be.reportedDispatchers = nil
 
 		log.Info("all dispatcher reported heartbeat, select one to write",
 			zap.String("changefeed", be.cfID),
@@ -214,7 +215,7 @@ func (be *BarrierEvent) resend() []*messaging.TargetMessage {
 		return nil
 	}
 	be.lastResendTime = time.Now()
-	// we select a dispatcher as the writer, still waiting for that dispatcher advance it's checkpoint ts
+	// we select a dispatcher as the writer, still waiting for that dispatcher advance its checkpoint ts
 	if !be.writerDispatcherAdvanced {
 		//resend write action
 		stm := be.scheduler.GetTask(be.writerDispatcher)

maintainer/barrier_test.go

@@ -26,6 +26,57 @@ import (
 	"go.uber.org/zap"
 )
 
+func TestOneBlockEvent(t *testing.T) {
+	sche := NewController("test", 1, nil, nil, nil, 1000, 0)
+	sche.AddNewNode("node1")
+	sche.AddNewTable(common.Table{1, 1}, 0)
+	stm := sche.GetTasksByTableIDs(1)[0]
+	stm.Primary = "node1"
+	stm.State = scheduler.SchedulerStatusWorking
+	sche.tryMoveTask(stm.ID, stm, scheduler.SchedulerStatusAbsent, "", true)
+	barrier := NewBarrier(sche)
+	msg := barrier.HandleStatus("node1", &heartbeatpb.HeartBeatRequest{
+		ChangefeedID: "test",
+		Statuses: []*heartbeatpb.TableSpanStatus{
+			{
+				ID: stm.ID.ToPB(),
+				State: &heartbeatpb.State{
+					IsBlocked: true,
+					BlockTs:   10,
+					BlockTables: &heartbeatpb.InfluencedTables{
+						InfluenceType: heartbeatpb.InfluenceType_All,
+					},
+				},
+				CheckpointTs: 0,
+			},
+		},
+	})
+	require.NotNil(t, msg)
+	resp := msg.Message[0].(*heartbeatpb.HeartBeatResponse)
+	require.Equal(t, barrier.blockedDispatcher[stm.ID], barrier.blockedTs[10])
+	event := barrier.blockedTs[10]
+	require.Equal(t, uint64(10), event.commitTs)
+	require.True(t, event.writerDispatcher == stm.ID)
+	require.True(t, event.selected)
+	require.False(t, event.writerDispatcherAdvanced)
+	require.Nil(t, event.reportedDispatchers)
+	require.Equal(t, resp.DispatcherStatuses[0].Ack.CommitTs, uint64(10))
+	require.Equal(t, resp.DispatcherStatuses[0].Action.CommitTs, uint64(10))
+	require.Equal(t, resp.DispatcherStatuses[0].Action.Action, heartbeatpb.Action_Write)
+
+	msg = barrier.HandleStatus("node1", &heartbeatpb.HeartBeatRequest{
+		ChangefeedID: "test",
+		Statuses: []*heartbeatpb.TableSpanStatus{
+			{
+				ID:           stm.ID.ToPB(),
+				CheckpointTs: 10,
+			},
+		},
+	})
+	require.Len(t, barrier.blockedTs, 0)
+	require.Len(t, barrier.blockedDispatcher, 0)
+}
+
 func TestNormalBlock(t *testing.T) {
 	sche := NewController("test", 1, nil, nil, nil, 1000, 0)
 	sche.AddNewNode("node1")
@@ -121,7 +172,7 @@ func TestNormalBlock(t *testing.T) {
 	event := barrier.blockedTs[10]
 	require.Equal(t, uint64(10), event.commitTs)
 	require.True(t, event.writerDispatcher == selectDispatcherID)
-	require.Nil(t, event.advancedDispatchers)
+	require.Nil(t, event.reportedDispatchers)
 
 	// repeated status
 	barrier.HandleStatus("node1", &heartbeatpb.HeartBeatRequest{
@@ -165,7 +216,7 @@ func TestNormalBlock(t *testing.T) {
 	})
 	require.Equal(t, uint64(10), event.commitTs)
 	require.True(t, event.writerDispatcher == selectDispatcherID)
-	require.Nil(t, event.advancedDispatchers)
+	require.Nil(t, event.reportedDispatchers)
 
 	// selected node write done
 	msg = barrier.HandleStatus("node2", &heartbeatpb.HeartBeatRequest{