logservice: fix dynamic stream handler and optimize eventstore cpu usage (#365)

* avoid heap allocation

* avoid scan ddl in handle

* some rename

* hack

* fix

* notify max event commit ts as original way

* small fix

* Revert "avoid heap allocation"

This reverts commit 0ea7907.

* try optimize

* increase batch size

* increase ch size

* only send resolved ts whn advance

* small improvement

* small fix

* change event processor num

* small fix

* try less processor

* increase processor num

* use original processor num

* try fix dead lock

Author: lidezhu

logservice/eventstore/event_store.go

@@ -47,6 +47,9 @@ type spanProgress struct {
 	resolvedTsUpdated atomic.Int64
 	resolvedTs        atomic.Uint64
 
+	// tag is supplied at subscription time and is passed to the consume function.
+	tag interface{}
+
 	consume struct {
 		// This lock is used to prevent the table progress from being
 		// removed while consuming events.
@@ -74,7 +77,7 @@ func (p *spanProgress) resolveLock(currentTime time.Time) {
 type LogPuller struct {
 	client  *SubscriptionClient
 	pdClock pdutil.Clock
-	consume func(context.Context, *common.RawKVEntry, SubscriptionID) error
+	consume func(context.Context, *common.RawKVEntry, SubscriptionID, interface{}) error
 
 	subscriptions struct {
 		sync.RWMutex
@@ -89,7 +92,7 @@ type LogPuller struct {
 func NewLogPuller(
 	client *SubscriptionClient,
 	pdClock pdutil.Clock,
-	consume func(context.Context, *common.RawKVEntry, SubscriptionID) error,
+	consume func(context.Context, *common.RawKVEntry, SubscriptionID, interface{}) error,
 ) *LogPuller {
 	puller := &LogPuller{
 		client:  client,
@@ -155,6 +158,7 @@ func (p *LogPuller) Close(ctx context.Context) error {
 func (p *LogPuller) Subscribe(
 	span heartbeatpb.TableSpan,
 	startTs uint64,
+	tag interface{},
 ) SubscriptionID {
 	p.subscriptions.Lock()
 
@@ -163,6 +167,7 @@ func (p *LogPuller) Subscribe(
 	progress := &spanProgress{
 		span:  span,
 		subID: subID,
+		tag:   tag,
 	}
 
 	progress.consume.f = func(
@@ -173,7 +178,7 @@ func (p *LogPuller) Subscribe(
 		progress.consume.RLock()
 		defer progress.consume.RUnlock()
 		if !progress.consume.removed {
-			return p.consume(ctx, raw, subID)
+			return p.consume(ctx, raw, subID, progress.tag)
 		}
 		return nil
 	}

logservice/logpuller/log_puller.go

@@ -72,10 +72,7 @@ func NewLogPullerMultiSpan(
 	}
 
 	// consumeWrapper may be called concurrently
-	consumeWrapper := func(ctx context.Context, entry *common.RawKVEntry, subID SubscriptionID) error {
-		if entry == nil {
-			return nil
-		}
+	consumeWrapper := func(ctx context.Context, entry *common.RawKVEntry, subID SubscriptionID, _ interface{}) error {
 		if entry.IsResolved() {
 			pullerWrapper.tryUpdatePendingResolvedTs(subID, entry.CRTs)
 			return nil
@@ -85,7 +82,7 @@ func NewLogPullerMultiSpan(
 
 	pullerWrapper.innerPuller = NewLogPuller(client, pdClock, consumeWrapper)
 	for _, span := range spans {
-		subID := pullerWrapper.innerPuller.Subscribe(span, startTs)
+		subID := pullerWrapper.innerPuller.Subscribe(span, startTs, nil)
 		item := &resolvedTsItem{
 			resolvedTs: 0,
 		}

logservice/logpuller/log_puller_multi_span.go

@@ -92,20 +92,16 @@ func TestMultiplexingPullerResolvedForward(t *testing.T) {
 	subID1 := subIDs[1]
 
 	puller.innerPuller.client.consume(ctx, LogEvent{
-		regionFeedEvent: regionFeedEvent{
-			Val: &common.RawKVEntry{
-				OpType: common.OpTypeResolved,
-				CRTs:   uint64(1000),
-			},
+		Val: &common.RawKVEntry{
+			OpType: common.OpTypeResolved,
+			CRTs:   uint64(1000),
 		},
 		SubscriptionID: subID0,
 	})
 	puller.innerPuller.client.consume(ctx, LogEvent{
-		regionFeedEvent: regionFeedEvent{
-			Val: &common.RawKVEntry{
-				OpType: common.OpTypeResolved,
-				CRTs:   uint64(1005),
-			},
+		Val: &common.RawKVEntry{
+			OpType: common.OpTypeResolved,
+			CRTs:   uint64(1005),
 		},
 		SubscriptionID: subID1,
 	})
@@ -119,11 +115,9 @@ func TestMultiplexingPullerResolvedForward(t *testing.T) {
 	}
 
 	puller.innerPuller.client.consume(ctx, LogEvent{
-		regionFeedEvent: regionFeedEvent{
-			Val: &common.RawKVEntry{
-				OpType: common.OpTypeResolved,
-				CRTs:   uint64(1002),
-			},
+		Val: &common.RawKVEntry{
+			OpType: common.OpTypeResolved,
+			CRTs:   uint64(1002),
 		},
 		SubscriptionID: subID0,
 	})
@@ -137,11 +131,9 @@ func TestMultiplexingPullerResolvedForward(t *testing.T) {
 	}
 
 	puller.innerPuller.client.consume(ctx, LogEvent{
-		regionFeedEvent: regionFeedEvent{
-			Val: &common.RawKVEntry{
-				OpType: common.OpTypeResolved,
-				CRTs:   uint64(1008),
-			},
+		Val: &common.RawKVEntry{
+			OpType: common.OpTypeResolved,
+			CRTs:   uint64(1008),
 		},
 		SubscriptionID: subID0,
 	})

logservice/logpuller/log_puller_multi_span_test.go

@@ -159,9 +159,9 @@ func (w *changeEventProcessor) processEvent(ctx context.Context, event statefulE
 // NOTE: context.Canceled won't be treated as an error.
 func (w *changeEventProcessor) handleEventEntry(ctx context.Context, x *cdcpb.Event_Entries_, state *regionFeedState) error {
 	startTs := state.region.subscribedSpan.startTs
-	emit := func(assembled regionFeedEvent) error {
+	emit := func(val *common.RawKVEntry) error {
 		// TODO: add a metric to indicate whether the event is sent successfully.
-		e := newLogEvent(assembled, state.region.subscribedSpan)
+		e := newLogEvent(val, state.region.subscribedSpan)
 		return w.client.consume(ctx, e)
 	}
 	tableID := state.region.subscribedSpan.span.TableID
@@ -177,7 +177,7 @@ func (w *changeEventProcessor) doHandle(
 	x *cdcpb.Event_Entries_,
 	startTs uint64,
 	state *regionFeedState,
-	emit func(assembled regionFeedEvent) error,
+	emit func(val *common.RawKVEntry) error,
 	tableID common.TableID,
 ) error {
 	regionID, _, _ := state.getRegionMeta()
@@ -269,30 +269,25 @@ func (w *changeEventProcessor) doHandle(
 	return nil
 }
 
-func (w *changeEventProcessor) assembleRowEvent(regionID uint64, entry *cdcpb.Event_Row) (regionFeedEvent, error) {
+func (w *changeEventProcessor) assembleRowEvent(regionID uint64, entry *cdcpb.Event_Row) (*common.RawKVEntry, error) {
 	var opType common.OpType
 	switch entry.GetOpType() {
 	case cdcpb.Event_Row_DELETE:
 		opType = common.OpTypeDelete
 	case cdcpb.Event_Row_PUT:
 		opType = common.OpTypePut
 	default:
-		return regionFeedEvent{}, cerror.ErrUnknownKVEventType.GenWithStackByArgs(entry.GetOpType(), entry)
+		return &common.RawKVEntry{}, cerror.ErrUnknownKVEventType.GenWithStackByArgs(entry.GetOpType(), entry)
 	}
-
-	return regionFeedEvent{
+	return &common.RawKVEntry{
+		OpType:   opType,
+		Key:      entry.Key,
+		Value:    entry.GetValue(),
+		StartTs:  entry.StartTs,
+		CRTs:     entry.CommitTs,
 		RegionID: regionID,
-		Val: &common.RawKVEntry{
-			OpType:   opType,
-			Key:      entry.Key,
-			Value:    entry.GetValue(),
-			StartTs:  entry.StartTs,
-			CRTs:     entry.CommitTs,
-			RegionID: regionID,
-			OldValue: entry.GetOldValue(),
-		},
+		OldValue: entry.GetOldValue(),
 	}, nil
-
 }
 
 func (w *changeEventProcessor) handleResolvedTs(ctx context.Context, batch resolvedTsBatch) {
@@ -324,20 +319,16 @@ func (w *changeEventProcessor) advanceTableSpan(ctx context.Context, batch resol
 		state.updateResolvedTs(batch.ts)
 	}
 
-	table := batch.regions[0].region.subscribedSpan
+	span := batch.regions[0].region.subscribedSpan
 	now := time.Now().UnixMilli()
-	lastAdvance := table.lastAdvanceTime.Load()
-	if now-lastAdvance > int64(w.client.config.AdvanceResolvedTsIntervalInMs) && table.lastAdvanceTime.CompareAndSwap(lastAdvance, now) {
-		ts := table.rangeLock.ResolvedTs()
-		// TODO: only send ts when ts is larger than previous ts
-		if ts > table.startTs {
-			revent := regionFeedEvent{
-				Val: &common.RawKVEntry{
-					OpType: common.OpTypeResolved,
-					CRTs:   ts,
-				},
-			}
-			e := newLogEvent(revent, table)
+	lastAdvance := span.lastAdvanceTime.Load()
+	if now-lastAdvance > int64(w.client.config.AdvanceResolvedTsIntervalInMs) && span.lastAdvanceTime.CompareAndSwap(lastAdvance, now) {
+		ts := span.rangeLock.ResolvedTs()
+		if ts > span.startTs {
+			e := newLogEvent(&common.RawKVEntry{
+				OpType: common.OpTypeResolved,
+				CRTs:   ts,
+			}, span)
 			if err := w.client.consume(ctx, e); err != nil {
 				return
 			}

logservice/logpuller/region_change_event_processor.go

@@ -84,14 +84,14 @@ type regionFeedEvent struct {
 
 // LogEvent wrap a region event with subscriptionID to indicate which subscription it belongs to.
 type LogEvent struct {
-	regionFeedEvent
+	Val *common.RawKVEntry
 	SubscriptionID
 }
 
-func newLogEvent(e regionFeedEvent, span *subscribedSpan) LogEvent {
+func newLogEvent(val *common.RawKVEntry, span *subscribedSpan) LogEvent {
 	return LogEvent{
-		regionFeedEvent: e,
-		SubscriptionID:  span.subID,
+		Val:            val,
+		SubscriptionID: span.subID,
 	}
 }
 
@@ -868,12 +868,12 @@ func (e *errCache) dispatch(ctx context.Context) error {
 	ticker := time.NewTicker(10 * time.Millisecond)
 	sendToErrCh := func() {
 		e.Lock()
-		defer e.Unlock()
 		if len(e.cache) == 0 {
 			return
 		}
 		errInfo := e.cache[0]
 		e.cache = e.cache[1:]
+		e.Unlock()
 		e.errCh <- errInfo
 	}
 	for {

logservice/logpuller/subscription_client.go

@@ -236,8 +236,8 @@ func TestSubscriptionWithFailedTiKV(t *testing.T) {
 	// it should auto switch to other stores and fetch events finally.
 	select {
 	case event := <-consumeCh:
-		require.Equal(t, common.OpTypeResolved, event.regionFeedEvent.Val.OpType)
-		require.Equal(t, ts, event.regionFeedEvent.Val.CRTs)
+		require.Equal(t, common.OpTypeResolved, event.Val.OpType)
+		require.Equal(t, ts, event.Val.CRTs)
 	case <-time.After(5 * time.Second):
 		require.True(t, false, "reconnection not succeed in 5 second")
 	}
@@ -252,8 +252,8 @@ func TestSubscriptionWithFailedTiKV(t *testing.T) {
 	// it should auto switch to other stores and fetch events finally.
 	select {
 	case event := <-consumeCh:
-		require.Equal(t, common.OpTypeResolved, event.regionFeedEvent.Val.OpType)
-		require.Equal(t, ts, event.regionFeedEvent.Val.CRTs)
+		require.Equal(t, common.OpTypeResolved, event.Val.OpType)
+		require.Equal(t, ts, event.Val.CRTs)
 	case <-time.After(5 * time.Second):
 		require.True(t, false, "reconnection not succeed in 5 second")
 	}

logservice/logpuller/subscription_client_test.go

@@ -91,10 +91,6 @@ func (p *ddlJobFetcher) close(ctx context.Context) error {
 }
 
 func (p *ddlJobFetcher) input(ctx context.Context, rawEvent *common.RawKVEntry) error {
-	if rawEvent == nil {
-		return nil
-	}
-
 	if rawEvent.IsResolved() {
 		p.advanceResolvedTs(uint64(rawEvent.CRTs))
 		return nil

logservice/schemastore/ddl_job_fetcher.go

@@ -292,6 +292,22 @@ func (p *persistentStorage) getTableInfo(tableID int64, ts uint64) (*common.Tabl
 	return store.getTableInfo(ts)
 }
 
+// TODO: this may consider some shouldn't be send ddl, like create table, does it matter?
+func (p *persistentStorage) getMaxEventCommitTs(tableID int64, ts uint64) uint64 {
+	p.mu.RLock()
+	defer p.mu.RUnlock()
+	if len(p.tablesDDLHistory[tableID]) == 0 {
+		return 0
+	}
+	index := sort.Search(len(p.tablesDDLHistory[tableID]), func(i int) bool {
+		return p.tablesDDLHistory[tableID][i] > ts
+	})
+	if index == 0 {
+		return 0
+	}
+	return p.tablesDDLHistory[tableID][index-1]
+}
+
 // TODO: not all ddl in p.tablesDDLHistory should be sent to the dispatcher, verify dispatcher will set the right range
 func (p *persistentStorage) fetchTableDDLEvents(tableID int64, tableFilter filter.Filter, start, end uint64) ([]commonEvent.DDLEvent, error) {
 	// TODO: check a dispatcher from created table start ts > finish ts of create table

logservice/schemastore/persist_storage.go

@@ -32,14 +32,22 @@ type SchemaStore interface {
 	// return table info with largest version <= ts
 	GetTableInfo(tableID int64, ts uint64) (*common.TableInfo, error)
 
+	// TODO: how to respect tableFilter
+	GetTableDDLEventState(tableID int64) DDLEventState
+
 	// FetchTableDDLEvents returns the next ddl events which finishedTs are within the range (start, end]
-	// and it returns a timestamp which means there will be no more ddl events before(<=) it
+	// The caller must ensure end <= current resolvedTs
 	// TODO: add a parameter limit
-	FetchTableDDLEvents(tableID int64, tableFilter filter.Filter, start, end uint64) ([]commonEvent.DDLEvent, uint64, error)
+	FetchTableDDLEvents(tableID int64, tableFilter filter.Filter, start, end uint64) ([]commonEvent.DDLEvent, error)
 
 	FetchTableTriggerDDLEvents(tableFilter filter.Filter, start uint64, limit int) ([]commonEvent.DDLEvent, uint64, error)
 }
 
+type DDLEventState struct {
+	ResolvedTs       uint64
+	MaxEventCommitTs uint64
+}
+
 type schemaStore struct {
 	pdClock pdutil.Clock
 
@@ -225,19 +233,27 @@ func (s *schemaStore) GetTableInfo(tableID int64, ts uint64) (*common.TableInfo,
 	return s.dataStorage.getTableInfo(tableID, ts)
 }
 
-func (s *schemaStore) FetchTableDDLEvents(tableID int64, tableFilter filter.Filter, start, end uint64) ([]commonEvent.DDLEvent, uint64, error) {
-	currentResolvedTs := s.resolvedTs.Load()
-	if currentResolvedTs <= start {
-		return nil, currentResolvedTs, nil
+func (s *schemaStore) GetTableDDLEventState(tableID int64) DDLEventState {
+	resolvedTs := s.resolvedTs.Load()
+	maxEventCommitTs := s.dataStorage.getMaxEventCommitTs(tableID, resolvedTs)
+	return DDLEventState{
+		ResolvedTs:       resolvedTs,
+		MaxEventCommitTs: maxEventCommitTs,
 	}
-	if currentResolvedTs < end {
-		end = currentResolvedTs
+}
+
+func (s *schemaStore) FetchTableDDLEvents(tableID int64, tableFilter filter.Filter, start, end uint64) ([]commonEvent.DDLEvent, error) {
+	currentResolvedTs := s.resolvedTs.Load()
+	if end > currentResolvedTs {
+		log.Panic("end should not be greater than current resolved ts",
+			zap.Uint64("end", end),
+			zap.Uint64("currentResolvedTs", currentResolvedTs))
 	}
 	events, err := s.dataStorage.fetchTableDDLEvents(tableID, tableFilter, start, end)
 	if err != nil {
-		return nil, 0, err
+		return nil, err
 	}
-	return events, end, nil
+	return events, nil
 }
 
 // FetchTableTriggerDDLEvents returns the next ddl events which finishedTs are within the range (start, end]