add IsSycPoint as a part of barrier event key (#360)

Author: sdojjy

maintainer/barrier.go

@@ -23,24 +23,29 @@ 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 reporting block event (all dispatchers will report the same event)
+// 2. maintainer wait for all dispatchers reporting block event (all dispatchers must 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, checkpoint ts >= block ts,(means it already finished the write action), (resend logic is needed)
+// 4. maintainer wait for the selected dispatcher reporting event(write) done message (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
+// 6. maintainer wait for all dispatchers reporting event(pass) done message
+// 7. maintainer clear the event
 type Barrier struct {
-	blockedTs  map[uint64]*BarrierEvent
+	blockedTs  map[eventKey]*BarrierEvent
 	controller *Controller
-	// if maintainer is down, the barrier will be re-built, so we can use the dispatcher as the key
-	blockedDispatcher map[common.DispatcherID]*BarrierEvent
+}
+
+// eventKey is the key of the block event,
+// the ddl and sync point are identified by the blockTs and isSyncPoint since they can share the same blockTs
+type eventKey struct {
+	blockTs     uint64
+	isSyncPoint bool
 }
 
 // NewBarrier create a new barrier for the changefeed
 func NewBarrier(controller *Controller) *Barrier {
 	return &Barrier{
-		blockedTs:         make(map[uint64]*BarrierEvent),
-		blockedDispatcher: make(map[common.DispatcherID]*BarrierEvent),
-		controller:        controller,
+		blockedTs:  make(map[eventKey]*BarrierEvent),
+		controller: controller,
 	}
 }
 
@@ -79,14 +84,15 @@ func (b *Barrier) Resend() []*messaging.TargetMessage {
 func (b *Barrier) handleOneStatus(changefeedID string, status *heartbeatpb.TableSpanBlockStatus) *heartbeatpb.DispatcherStatus {
 	dispatcherID := common.NewDispatcherIDFromPB(status.ID)
 	if status.State.EventDone {
-		b.handleEventDone(dispatcherID)
+		b.handleEventDone(dispatcherID, status)
 		return nil
 	}
 	return b.handleBlockState(changefeedID, dispatcherID, status)
 }
 
-func (b *Barrier) handleEventDone(dispatcherID common.DispatcherID) {
-	event, ok := b.blockedDispatcher[dispatcherID]
+func (b *Barrier) handleEventDone(dispatcherID common.DispatcherID, status *heartbeatpb.TableSpanBlockStatus) {
+	key := getEventKey(status.State)
+	event, ok := b.blockedTs[key]
 	// no block event found
 	if !ok {
 		return
@@ -103,10 +109,10 @@ func (b *Barrier) handleEventDone(dispatcherID common.DispatcherID) {
 	}
 
 	// checkpoint ts is advanced, clear the map, so do not need to resend message anymore
-	delete(b.blockedDispatcher, dispatcherID)
-	// all blocked dispatchers are advanced checkpoint ts
-	if len(b.blockedDispatcher) == 0 {
-		delete(b.blockedTs, event.commitTs)
+	event.markDispatcherEventDone(dispatcherID)
+	// all blocked dispatchers are reported event done, we can clean up the event
+	if event.allDispatcherDone() {
+		delete(b.blockedTs, key)
 	}
 }
 
@@ -121,33 +127,38 @@ func (b *Barrier) handleBlockState(changefeedID string,
 				dispatcherID.ToPB(),
 			},
 		},
-		Ack: &heartbeatpb.ACK{CommitTs: blockState.BlockTs},
+		Ack: &heartbeatpb.ACK{CommitTs: blockState.BlockTs, IsSyncPoint: blockState.IsSyncPoint},
 	}
 	if blockState.IsBlocked {
+		key := getEventKey(blockState)
 		// insert an event, or get the old one event check if the event is already tracked
-		event := b.getOrInsertNewEvent(changefeedID, dispatcherID, blockState)
+		event := b.getOrInsertNewEvent(changefeedID, key, blockState)
 		// 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
-		// if ack failed, dispatcher will send a heartbeat again
+		// and the ddl already synced to downstream , e.g.: create table, drop table
+		// if ack failed, dispatcher will send a heartbeat again, so we do not need to care about resend message here
 		NewBlockEvent(changefeedID, b.controller, blockState).scheduleBlockEvent()
 	}
 	return dispatcherStatus
 }
 
-func (b *Barrier) getOrInsertNewEvent(changefeedID string,
-	dispatcherID common.DispatcherID,
+// getOrInsertNewEvent get the block event from the map, if not found, create a new one
+func (b *Barrier) getOrInsertNewEvent(changefeedID string, key eventKey,
 	blockState *heartbeatpb.State) *BarrierEvent {
-	event, ok := b.blockedTs[blockState.BlockTs]
+	event, ok := b.blockedTs[key]
 	if !ok {
 		event = NewBlockEvent(changefeedID, b.controller, blockState)
-		b.blockedTs[blockState.BlockTs] = event
-	}
-	_, ok = b.blockedDispatcher[dispatcherID]
-	if !ok {
-		b.blockedDispatcher[dispatcherID] = event
+		b.blockedTs[key] = event
 	}
 	return event
 }
+
+// getEventKey returns the key of the block event
+func getEventKey(blockState *heartbeatpb.State) eventKey {
+	return eventKey{
+		blockTs:     blockState.BlockTs,
+		isSyncPoint: blockState.IsSyncPoint,
+	}
+}

maintainer/barrier_event.go

@@ -45,7 +45,7 @@ type BarrierEvent struct {
 	schemaIDChange []*heartbeatpb.SchemaIDChange
 	isSyncPoint    bool
 
-	reportedDispatchers map[common.DispatcherID]bool
+	reportedDispatchers map[common.DispatcherID]struct{}
 	lastResendTime      time.Time
 }
 
@@ -60,7 +60,7 @@ func NewBlockEvent(cfID string, scheduler *Controller,
 		newTables:           status.NeedAddedTables,
 		dropDispatchers:     status.NeedDroppedTables,
 		schemaIDChange:      status.UpdatedSchemas,
-		reportedDispatchers: make(map[common.DispatcherID]bool),
+		reportedDispatchers: make(map[common.DispatcherID]struct{}),
 		lastResendTime:      time.Time{},
 		isSyncPoint:         status.IsSyncPoint,
 	}
@@ -117,6 +117,14 @@ func (be *BarrierEvent) scheduleBlockEvent() {
 	}
 }
 
+func (be *BarrierEvent) markDispatcherEventDone(id common.DispatcherID) {
+	delete(be.reportedDispatchers, id)
+}
+
+func (be *BarrierEvent) allDispatcherDone() bool {
+	return len(be.reportedDispatchers) == 0
+}
+
 func (be *BarrierEvent) allDispatcherReported() bool {
 	if be.blockedDispatchers == nil {
 		return true
@@ -188,13 +196,11 @@ func (be *BarrierEvent) dispatcherReachedBlockTs(dispatcherID common.DispatcherI
 	if be.selected {
 		return nil
 	}
-	be.reportedDispatchers[dispatcherID] = true
+	be.reportedDispatchers[dispatcherID] = struct{}{}
 	// all dispatcher reported heartbeat, select the last one to write
 	if be.allDispatcherReported() {
 		be.writerDispatcher = dispatcherID
 		be.selected = true
-		// release memory
-		be.reportedDispatchers = nil
 
 		log.Info("all dispatcher reported heartbeat, select one to write",
 			zap.String("changefeed", be.cfID),

maintainer/barrier_test.go

@@ -53,14 +53,17 @@ func TestOneBlockEvent(t *testing.T) {
 		},
 	})
 	require.NotNil(t, msg)
+	key := eventKey{
+		blockTs:     10,
+		isSyncPoint: true,
+	}
 	resp := msg.Message[0].(*heartbeatpb.HeartBeatResponse)
-	require.Equal(t, barrier.blockedDispatcher[stm.ID], barrier.blockedTs[10])
-	event := barrier.blockedTs[10]
+	event := barrier.blockedTs[key]
 	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.Len(t, event.reportedDispatchers, 1)
 	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)
@@ -72,6 +75,7 @@ func TestOneBlockEvent(t *testing.T) {
 			{
 				ID: stm.ID.ToPB(),
 				State: &heartbeatpb.State{
+					BlockTs:     10,
 					IsBlocked:   true,
 					EventDone:   true,
 					IsSyncPoint: true,
@@ -80,7 +84,6 @@ func TestOneBlockEvent(t *testing.T) {
 		},
 	})
 	require.Len(t, barrier.blockedTs, 0)
-	require.Len(t, barrier.blockedDispatcher, 0)
 }
 
 func TestNormalBlock(t *testing.T) {
@@ -171,11 +174,14 @@ func TestNormalBlock(t *testing.T) {
 		},
 	})
 	require.NotNil(t, msg)
-	require.Equal(t, barrier.blockedDispatcher[selectDispatcherID], barrier.blockedTs[10])
-	event := barrier.blockedTs[10]
+	key := eventKey{
+		blockTs:     10,
+		isSyncPoint: false,
+	}
+	event := barrier.blockedTs[key]
 	require.Equal(t, uint64(10), event.commitTs)
 	require.True(t, event.writerDispatcher == selectDispatcherID)
-	require.Nil(t, event.reportedDispatchers)
+	require.Len(t, event.reportedDispatchers, 3)
 
 	// repeated status
 	barrier.HandleStatus("node1", &heartbeatpb.BlockStatusRequest{
@@ -217,7 +223,7 @@ func TestNormalBlock(t *testing.T) {
 	})
 	require.Equal(t, uint64(10), event.commitTs)
 	require.True(t, event.writerDispatcher == selectDispatcherID)
-	require.Nil(t, event.reportedDispatchers)
+	require.Len(t, event.reportedDispatchers, 3)
 
 	// selected node write done
 	msg = barrier.HandleStatus("node2", &heartbeatpb.BlockStatusRequest{
@@ -234,7 +240,7 @@ func TestNormalBlock(t *testing.T) {
 		},
 	})
 	require.Len(t, barrier.blockedTs, 1)
-	require.Len(t, barrier.blockedDispatcher, 2)
+	require.Len(t, event.reportedDispatchers, 2)
 	msg = barrier.HandleStatus("node1", &heartbeatpb.BlockStatusRequest{
 		ChangefeedID: "test",
 		BlockStatuses: []*heartbeatpb.TableSpanBlockStatus{
@@ -257,7 +263,6 @@ func TestNormalBlock(t *testing.T) {
 		},
 	})
 	require.Len(t, barrier.blockedTs, 0)
-	require.Len(t, barrier.blockedDispatcher, 0)
 }
 
 func TestSchemaBlock(t *testing.T) {
@@ -336,7 +341,8 @@ func TestSchemaBlock(t *testing.T) {
 	require.True(t, resp.DispatcherStatuses[0].Ack.CommitTs == 10)
 	require.True(t, resp.DispatcherStatuses[0].Action.CommitTs == 10)
 	require.True(t, resp.DispatcherStatuses[0].Action.Action == heartbeatpb.Action_Write)
-	event := barrier.blockedTs[10]
+	key := eventKey{blockTs: 10}
+	event := barrier.blockedTs[key]
 	require.Equal(t, uint64(10), event.commitTs)
 	//the last one will be the writer
 	require.Equal(t, event.writerDispatcher.ToPB(), dispatcherIDs[1])
@@ -367,7 +373,7 @@ func TestSchemaBlock(t *testing.T) {
 	resp = msg.Message[0].(*heartbeatpb.HeartBeatResponse)
 	require.Len(t, resp.DispatcherStatuses, 1)
 	require.True(t, resp.DispatcherStatuses[0].Ack.CommitTs == 10)
-	event = barrier.blockedTs[10]
+	event = barrier.blockedTs[key]
 	require.Equal(t, uint64(10), event.commitTs)
 	//the last one will be the writer
 	require.Equal(t, event.writerDispatcher.ToPB(), dispatcherIDs[1])
@@ -395,7 +401,7 @@ func TestSchemaBlock(t *testing.T) {
 		heartbeatpb.Action_Pass)
 	require.Len(t, barrier.blockedTs, 1)
 	// the writer already advanced
-	require.Len(t, barrier.blockedDispatcher, 1)
+	require.Len(t, event.reportedDispatchers, 1)
 	require.Equal(t, 1, len(sche.Absent()))
 	require.Equal(t, 0, len(sche.Commiting()))
 	require.Equal(t, 2, len(sche.Removing()))
@@ -415,7 +421,6 @@ func TestSchemaBlock(t *testing.T) {
 		},
 	})
 	require.Len(t, barrier.blockedTs, 0)
-	require.Len(t, barrier.blockedDispatcher, 0)
 }
 
 func TestSyncPointBlock(t *testing.T) {
@@ -457,6 +462,7 @@ func TestSyncPointBlock(t *testing.T) {
 						TableIDs:      dropTables,
 					},
 					NeedAddedTables: []*heartbeatpb.Table{newSpan},
+					IsSyncPoint:     true,
 				},
 			},
 			{
@@ -473,6 +479,7 @@ func TestSyncPointBlock(t *testing.T) {
 						TableIDs:      dropTables,
 					},
 					NeedAddedTables: []*heartbeatpb.Table{newSpan},
+					IsSyncPoint:     true,
 				},
 			},
 		},
@@ -511,7 +518,8 @@ func TestSyncPointBlock(t *testing.T) {
 	require.True(t, resp.DispatcherStatuses[0].Ack.CommitTs == 10)
 	require.True(t, resp.DispatcherStatuses[0].Action.CommitTs == 10)
 	require.True(t, resp.DispatcherStatuses[0].Action.Action == heartbeatpb.Action_Write)
-	event := barrier.blockedTs[10]
+	key := eventKey{blockTs: 10, isSyncPoint: true}
+	event := barrier.blockedTs[key]
 	require.Equal(t, uint64(10), event.commitTs)
 	//the last one will be the writer
 	require.Equal(t, event.writerDispatcher.ToPB(), dispatcherIDs[2])
@@ -526,7 +534,7 @@ func TestSyncPointBlock(t *testing.T) {
 					IsBlocked:   true,
 					BlockTs:     10,
 					EventDone:   true,
-					IsSyncPoint: false,
+					IsSyncPoint: true,
 				},
 			},
 		},
@@ -536,7 +544,7 @@ func TestSyncPointBlock(t *testing.T) {
 	require.Len(t, msgs, 2)
 	require.Len(t, barrier.blockedTs, 1)
 	// the writer already advanced
-	require.Len(t, barrier.blockedDispatcher, 2)
+	require.Len(t, event.reportedDispatchers, 2)
 	// other dispatcher advanced checkpoint ts
 	msg = barrier.HandleStatus("node1", &heartbeatpb.BlockStatusRequest{
 		ChangefeedID: "test",
@@ -547,7 +555,7 @@ func TestSyncPointBlock(t *testing.T) {
 					IsBlocked:   true,
 					BlockTs:     10,
 					EventDone:   true,
-					IsSyncPoint: false,
+					IsSyncPoint: true,
 				},
 			},
 			{
@@ -556,13 +564,12 @@ func TestSyncPointBlock(t *testing.T) {
 					IsBlocked:   true,
 					BlockTs:     10,
 					EventDone:   true,
-					IsSyncPoint: false,
+					IsSyncPoint: true,
 				},
 			},
 		},
 	})
 	require.Len(t, barrier.blockedTs, 0)
-	require.Len(t, barrier.blockedDispatcher, 0)
 }
 
 func TestNonBlocked(t *testing.T) {
@@ -601,7 +608,6 @@ func TestNonBlocked(t *testing.T) {
 	require.True(t, heartbeatpb.InfluenceType_Normal == resp.DispatcherStatuses[0].InfluencedDispatchers.InfluenceType)
 	require.Equal(t, resp.DispatcherStatuses[0].InfluencedDispatchers.DispatcherIDs[0], blockedDispatcherIDS[0])
 	require.Len(t, barrier.blockedTs, 0)
-	require.Len(t, barrier.blockedDispatcher, 0)
 	require.Len(t, barrier.controller.Absent(), 2)
 }
 
@@ -630,6 +636,7 @@ func TestSyncPointBlockPerf(t *testing.T) {
 					InfluenceType: heartbeatpb.InfluenceType_All,
 					SchemaID:      1,
 				},
+				IsSyncPoint: true,
 			},
 		})
 	}