eventservice: backport adaptive scan window to release-8.5

downstreamadapter/dispatcher/basic_dispatcher.go

@@ -528,7 +528,8 @@ func (d *BasicDispatcher) handleEvents(dispatcherEvents []DispatcherEvent, wakeC
 				zap.Stringer("dispatcher", d.id),
 				zap.String("query", ddl.Query),
 				zap.Any("tableSpan", d.GetTableSpan()),
-				zap.Int64("table", ddl.GetTableID()),
+				zap.Int64("oldTableID", d.tableSpan.GetTableID()),
+				zap.Int64("currentTableID", ddl.GetTableID()),
 				zap.Uint64("commitTs", event.GetCommitTs()),
 				zap.Uint64("seq", event.GetSeq()))
 			now := time.Now()

downstreamadapter/eventcollector/event_collector.go

@@ -84,6 +84,7 @@ type changefeedStat struct {
 	metricMemoryUsageMaxRedo  prometheus.Gauge
 	metricMemoryUsageUsedRedo prometheus.Gauge
 	dispatcherCount           atomic.Int32
+	memoryReleaseCount        atomic.Uint32
 }
 
 func newChangefeedStat(changefeedID common.ChangeFeedID) *changefeedStat {
@@ -440,11 +441,17 @@ func (c *EventCollector) processDSFeedback(ctx context.Context) error {
 			return context.Cause(ctx)
 		case feedback := <-c.ds.Feedback():
 			if feedback.FeedbackType == dynstream.ReleasePath {
+				if v, ok := c.changefeedMap.Load(feedback.Area); ok {
+					v.(*changefeedStat).memoryReleaseCount.Add(1)
+				}
 				log.Info("release dispatcher memory in DS", zap.Any("dispatcherID", feedback.Path))
 				c.ds.Release(feedback.Path)
 			}
 		case feedback := <-c.redoDs.Feedback():
 			if feedback.FeedbackType == dynstream.ReleasePath {
+				if v, ok := c.changefeedMap.Load(feedback.Area); ok {
+					v.(*changefeedStat).memoryReleaseCount.Add(1)
+				}
 				log.Info("release dispatcher memory in redo DS", zap.Any("dispatcherID", feedback.Path))
 				c.redoDs.Release(feedback.Path)
 			}
@@ -617,9 +624,24 @@ func (c *EventCollector) controlCongestion(ctx context.Context) error {
 }
 
 func (c *EventCollector) newCongestionControlMessages() map[node.ID]*event.CongestionControl {
+	changefeedMemoryReleaseCount := make(map[common.ChangeFeedID]uint32)
+	getAndResetMemoryReleaseCount := func(changefeedID common.ChangeFeedID) uint32 {
+		if count, ok := changefeedMemoryReleaseCount[changefeedID]; ok {
+			return count
+		}
+		v, ok := c.changefeedMap.Load(changefeedID.ID())
+		if !ok {
+			return 0
+		}
+		count := v.(*changefeedStat).memoryReleaseCount.Swap(0)
+		changefeedMemoryReleaseCount[changefeedID] = count
+		return count
+	}
+
 	// collect path-level available memory and total available memory for each changefeed
 	changefeedPathMemory := make(map[common.ChangeFeedID]map[common.DispatcherID]uint64)
 	changefeedTotalMemory := make(map[common.ChangeFeedID]uint64)
+	changefeedUsageRatio := make(map[common.ChangeFeedID]float64)
 
 	// collect from main dynamic stream
 	for _, quota := range c.ds.GetMetrics().MemoryControl.AreaMemoryMetrics {
@@ -637,6 +659,7 @@ func (c *EventCollector) newCongestionControlMessages() map[node.ID]*event.Conge
 		}
 		// store total available memory from AreaMemoryMetric
 		changefeedTotalMemory[cfID] = uint64(quota.AvailableMemory())
+		changefeedUsageRatio[cfID] = calcUsageRatio(quota.MemoryUsage(), quota.MaxMemory())
 	}
 
 	// collect from redo dynamic stream and take minimum
@@ -658,11 +681,9 @@ func (c *EventCollector) newCongestionControlMessages() map[node.ID]*event.Conge
 			}
 		}
 		// take minimum total available memory between main and redo streams
-		if existing, exists := changefeedTotalMemory[cfID]; exists {
-			changefeedTotalMemory[cfID] = min(existing, uint64(quota.AvailableMemory()))
-		} else {
-			changefeedTotalMemory[cfID] = uint64(quota.AvailableMemory())
-		}
+		updateMinUint64MapValue(changefeedTotalMemory, cfID, uint64(quota.AvailableMemory()))
+		// take maximum usage ratio between main and redo streams
+		changefeedUsageRatio[cfID] = max(changefeedUsageRatio[cfID], calcUsageRatio(quota.MemoryUsage(), quota.MaxMemory()))
 	}
 
 	if len(changefeedPathMemory) == 0 {
@@ -699,30 +720,64 @@ func (c *EventCollector) newCongestionControlMessages() map[node.ID]*event.Conge
 	// build congestion control messages for each node
 	result := make(map[node.ID]*event.CongestionControl)
 	for nodeID, changefeedDispatchers := range nodeDispatcherMemory {
-		congestionControl := event.NewCongestionControl()
+		congestionControl := event.NewCongestionControlWithVersion(event.CongestionControlVersion2)
 
 		for changefeedID, dispatcherMemory := range changefeedDispatchers {
 			if len(dispatcherMemory) == 0 {
 				continue
 			}
 
 			// get total available memory directly from AreaMemoryMetric
-			totalAvailable := uint64(changefeedTotalMemory[changefeedID])
-			congestionControl.AddAvailableMemoryWithDispatchers(
+			totalAvailable, ok := changefeedTotalMemory[changefeedID]
+			if !ok {
+				continue
+			}
+			congestionControl.AddAvailableMemoryWithDispatchersAndUsageAndReleaseCount(
 				changefeedID.ID(),
 				totalAvailable,
+				changefeedUsageRatio[changefeedID],
 				dispatcherMemory,
+				getAndResetMemoryReleaseCount(changefeedID),
 			)
 		}
 
 		if len(congestionControl.GetAvailables()) > 0 {
 			result[nodeID] = congestionControl
 		}
 	}
-
 	return result
 }
 
+func updateMinUint64MapValue(m map[common.ChangeFeedID]uint64, key common.ChangeFeedID, value uint64) {
+	if existing, exists := m[key]; exists {
+		m[key] = min(existing, value)
+	} else {
+		m[key] = value
+	}
+}
+
+func updateMaxUint64MapValue(m map[common.ChangeFeedID]uint64, key common.ChangeFeedID, value uint64) {
+	if existing, exists := m[key]; exists {
+		m[key] = max(existing, value)
+	} else {
+		m[key] = value
+	}
+}
+
+func calcUsageRatio(usedMemory int64, maxMemory int64) float64 {
+	if maxMemory <= 0 {
+		return 0
+	}
+	ratio := float64(usedMemory) / float64(maxMemory)
+	if ratio < 0 {
+		return 0
+	}
+	if ratio > 1 {
+		return 1
+	}
+	return ratio
+}
+
 func (c *EventCollector) updateMetrics(ctx context.Context) error {
 	ticker := time.NewTicker(5 * time.Second)
 	defer ticker.Stop()