From bed0bf61f07e4cfc1f59021796fb993b4a55fe86 Mon Sep 17 00:00:00 2001 From: dongmen <414110582@qq.com> Date: Tue, 11 Aug 2026 21:11:19 +0800 Subject: [PATCH] eventservice: replace scan window with fair memory admission --- metrics/grafana/ticdc_new_arch.json | 206 ---- .../ticdc_new_arch_next_gen.json | 206 ---- .../ticdc_new_arch_with_keyspace_name.json | 206 ---- pkg/config/debug.go | 2 + pkg/eventservice/dispatcher_stat.go | 63 +- pkg/eventservice/event_broker.go | 277 +++--- pkg/eventservice/event_broker_test.go | 336 ++----- pkg/eventservice/scan_admission.go | 241 +++++ pkg/eventservice/scan_admission_test.go | 223 +++++ pkg/eventservice/scan_window.go | 893 ------------------ pkg/eventservice/scan_window_test.go | 486 ---------- pkg/metrics/event_service.go | 72 -- utils/dynstream/memory_control.go | 14 +- utils/dynstream/memory_control_test.go | 7 + .../dynstream/parallel_dynamic_stream_test.go | 67 ++ utils/dynstream/stream.go | 63 +- utils/dynstream/stream_test.go | 5 +- 17 files changed, 839 insertions(+), 2528 deletions(-) create mode 100644 pkg/eventservice/scan_admission.go create mode 100644 pkg/eventservice/scan_admission_test.go delete mode 100644 pkg/eventservice/scan_window.go delete mode 100644 pkg/eventservice/scan_window_test.go diff --git a/metrics/grafana/ticdc_new_arch.json b/metrics/grafana/ticdc_new_arch.json index 0a3f4bbd37..d82d6bd5d0 100644 --- a/metrics/grafana/ticdc_new_arch.json +++ b/metrics/grafana/ticdc_new_arch.json @@ -12359,212 +12359,6 @@ }, "id": 20007, "panels": [ - { - "aliasColors": {}, - "bars": false, - "dashLength": 10, - "dashes": false, - "datasource": "${DS_TEST-CLUSTER}", - "description": "The lag between changefeed checkpoint ts and the lac1 ts of upstream TiDB.", - "fieldConfig": { - "defaults": { - "links": [] - }, - "overrides": [] - }, - "fill": 1, - "fillGradient": 0, - "gridPos": { - "h": 6, - "w": 12, - "x": 0, - "y": 12 - }, - "hiddenSeries": false, - "id": 60029, - "legend": { - "alignAsTable": true, - "avg": false, - "current": true, - "max": true, - "min": false, - "rightSide": false, - "show": true, - "total": false, - "values": true - }, - "lines": true, - "linewidth": 1, - "links": [], - "nullPointMode": "null", - "options": { - "alertThreshold": true - }, - "paceLength": 10, - "percentage": false, - "pluginVersion": "7.5.17", - "pointradius": 2, - "points": false, - "renderer": "flot", - "seriesOverrides": [], - "spaceLength": 10, - "stack": false, - "steppedLine": false, - "targets": [ - { - "exemplar": true, - "expr": "max(ticdc_event_service_scan_window_interval{k8s_cluster=\"$k8s_cluster\", tidb_cluster=\"$tidb_cluster\", namespace=~\"$namespace\", changefeed=~\"$changefeed\", instance=~\"$ticdc_instance\"}) by (namespace, changefeed,instance)", - "format": "time_series", - "interval": "", - "intervalFactor": 1, - "legendFormat": "{{instance}}{{namespace}}-{{changefeed}}", - "refId": "A" - } - ], - "thresholds": [], - "timeFrom": null, - "timeRegions": [], - "timeShift": null, - "title": "Scan window interval", - "tooltip": { - "shared": true, - "sort": 0, - "value_type": "individual" - }, - "type": "graph", - "xaxis": { - "buckets": null, - "mode": "time", - "name": null, - "show": true, - "values": [] - }, - "yaxes": [ - { - "format": "s", - "logBase": 1, - "min": "0", - "show": true - }, - { - "format": "short", - "logBase": 1, - "show": false - } - ], - "yaxis": { - "align": false - } - }, - { - "aliasColors": {}, - "bars": true, - "cacheTimeout": null, - "dashLength": 10, - "dashes": false, - "datasource": "${DS_TEST-CLUSTER}", - "description": "", - "fieldConfig": { - "defaults": {}, - "overrides": [] - }, - "fill": 0, - "fillGradient": 0, - "gridPos": { - "h": 6, - "w": 12, - "x": 12, - "y": 12 - }, - "hiddenSeries": false, - "id": 60030, - "legend": { - "alignAsTable": true, - "avg": false, - "current": true, - "max": false, - "min": false, - "rightSide": false, - "show": true, - "sideWidth": null, - "total": false, - "values": true - }, - "lines": true, - "linewidth": 2, - "links": [], - "nullPointMode": "null", - "options": { - "alertThreshold": true - }, - "percentage": false, - "pluginVersion": "7.5.17", - "pointradius": 2, - "points": false, - "renderer": "flot", - "seriesOverrides": [ - { - "alias": "/approximate current time.*/", - "bars": false - } - ], - "spaceLength": 10, - "stack": false, - "steppedLine": false, - "targets": [ - { - "exemplar": true, - "expr": "max(ticdc_event_service_scan_window_base_ts{k8s_cluster=\"$k8s_cluster\", tidb_cluster=\"$tidb_cluster\", namespace=~\"$namespace\", changefeed=~\"$changefeed\"}) by (namespace,changefeed,instance)", - "format": "time_series", - "hide": false, - "interval": "", - "intervalFactor": 1, - "legendFormat": "{{instance}}{{namespace}}-{{changefeed}}", - "refId": "B" - } - ], - "thresholds": [], - "timeFrom": null, - "timeRegions": [], - "timeShift": null, - "title": "Scan window base ts", - "tooltip": { - "shared": true, - "sort": 0, - "value_type": "individual" - }, - "type": "graph", - "xaxis": { - "buckets": null, - "max": null, - "mode": "time", - "name": null, - "show": true, - "values": [] - }, - "yaxes": [ - { - "format": "none", - "label": null, - "logBase": 1, - "max": null, - "min": null, - "show": true - }, - { - "format": "short", - "label": null, - "logBase": 1, - "max": null, - "min": null, - "show": false - } - ], - "yaxis": { - "align": false, - "alignLevel": null - } - }, { "cards": { "cardPadding": 0, diff --git a/metrics/nextgengrafana/ticdc_new_arch_next_gen.json b/metrics/nextgengrafana/ticdc_new_arch_next_gen.json index f920ac2ccd..2c07c0b799 100644 --- a/metrics/nextgengrafana/ticdc_new_arch_next_gen.json +++ b/metrics/nextgengrafana/ticdc_new_arch_next_gen.json @@ -12359,212 +12359,6 @@ }, "id": 20007, "panels": [ - { - "aliasColors": {}, - "bars": false, - "dashLength": 10, - "dashes": false, - "datasource": "${DS_TEST-CLUSTER}", - "description": "The lag between changefeed checkpoint ts and the lac1 ts of upstream TiDB.", - "fieldConfig": { - "defaults": { - "links": [] - }, - "overrides": [] - }, - "fill": 1, - "fillGradient": 0, - "gridPos": { - "h": 6, - "w": 12, - "x": 0, - "y": 12 - }, - "hiddenSeries": false, - "id": 60029, - "legend": { - "alignAsTable": true, - "avg": false, - "current": true, - "max": true, - "min": false, - "rightSide": false, - "show": true, - "total": false, - "values": true - }, - "lines": true, - "linewidth": 1, - "links": [], - "nullPointMode": "null", - "options": { - "alertThreshold": true - }, - "paceLength": 10, - "percentage": false, - "pluginVersion": "7.5.17", - "pointradius": 2, - "points": false, - "renderer": "flot", - "seriesOverrides": [], - "spaceLength": 10, - "stack": false, - "steppedLine": false, - "targets": [ - { - "exemplar": true, - "expr": "max(ticdc_event_service_scan_window_interval{k8s_cluster=\"$k8s_cluster\", sharedpool_id=\"$tidb_cluster\", keyspace_name=~\"$keyspace_name\", changefeed=~\"$changefeed\", instance=~\"$ticdc_instance\"}) by (keyspace_name, changefeed,instance)", - "format": "time_series", - "interval": "", - "intervalFactor": 1, - "legendFormat": "{{instance}}{{keyspace_name}}-{{changefeed}}", - "refId": "A" - } - ], - "thresholds": [], - "timeFrom": null, - "timeRegions": [], - "timeShift": null, - "title": "Scan window interval", - "tooltip": { - "shared": true, - "sort": 0, - "value_type": "individual" - }, - "type": "graph", - "xaxis": { - "buckets": null, - "mode": "time", - "name": null, - "show": true, - "values": [] - }, - "yaxes": [ - { - "format": "s", - "logBase": 1, - "min": "0", - "show": true - }, - { - "format": "short", - "logBase": 1, - "show": false - } - ], - "yaxis": { - "align": false - } - }, - { - "aliasColors": {}, - "bars": true, - "cacheTimeout": null, - "dashLength": 10, - "dashes": false, - "datasource": "${DS_TEST-CLUSTER}", - "description": "", - "fieldConfig": { - "defaults": {}, - "overrides": [] - }, - "fill": 0, - "fillGradient": 0, - "gridPos": { - "h": 6, - "w": 12, - "x": 12, - "y": 12 - }, - "hiddenSeries": false, - "id": 60030, - "legend": { - "alignAsTable": true, - "avg": false, - "current": true, - "max": false, - "min": false, - "rightSide": false, - "show": true, - "sideWidth": null, - "total": false, - "values": true - }, - "lines": true, - "linewidth": 2, - "links": [], - "nullPointMode": "null", - "options": { - "alertThreshold": true - }, - "percentage": false, - "pluginVersion": "7.5.17", - "pointradius": 2, - "points": false, - "renderer": "flot", - "seriesOverrides": [ - { - "alias": "/approximate current time.*/", - "bars": false - } - ], - "spaceLength": 10, - "stack": false, - "steppedLine": false, - "targets": [ - { - "exemplar": true, - "expr": "max(ticdc_event_service_scan_window_base_ts{k8s_cluster=\"$k8s_cluster\", sharedpool_id=\"$tidb_cluster\", keyspace_name=~\"$keyspace_name\", changefeed=~\"$changefeed\"}) by (keyspace_name,changefeed,instance)", - "format": "time_series", - "hide": false, - "interval": "", - "intervalFactor": 1, - "legendFormat": "{{instance}}{{keyspace_name}}-{{changefeed}}", - "refId": "B" - } - ], - "thresholds": [], - "timeFrom": null, - "timeRegions": [], - "timeShift": null, - "title": "Scan window base ts", - "tooltip": { - "shared": true, - "sort": 0, - "value_type": "individual" - }, - "type": "graph", - "xaxis": { - "buckets": null, - "max": null, - "mode": "time", - "name": null, - "show": true, - "values": [] - }, - "yaxes": [ - { - "format": "none", - "label": null, - "logBase": 1, - "max": null, - "min": null, - "show": true - }, - { - "format": "short", - "label": null, - "logBase": 1, - "max": null, - "min": null, - "show": false - } - ], - "yaxis": { - "align": false, - "alignLevel": null - } - }, { "cards": { "cardPadding": 0, diff --git a/metrics/nextgengrafana/ticdc_new_arch_with_keyspace_name.json b/metrics/nextgengrafana/ticdc_new_arch_with_keyspace_name.json index f71e45145a..fe3e0740c4 100644 --- a/metrics/nextgengrafana/ticdc_new_arch_with_keyspace_name.json +++ b/metrics/nextgengrafana/ticdc_new_arch_with_keyspace_name.json @@ -3932,212 +3932,6 @@ }, "id": 20007, "panels": [ - { - "aliasColors": {}, - "bars": false, - "dashLength": 10, - "dashes": false, - "datasource": "${DS_TEST-CLUSTER}", - "description": "The lag between changefeed checkpoint ts and the lac1 ts of upstream TiDB.", - "fieldConfig": { - "defaults": { - "links": [] - }, - "overrides": [] - }, - "fill": 1, - "fillGradient": 0, - "gridPos": { - "h": 6, - "w": 12, - "x": 0, - "y": 12 - }, - "hiddenSeries": false, - "id": 60029, - "legend": { - "alignAsTable": true, - "avg": false, - "current": true, - "max": true, - "min": false, - "rightSide": false, - "show": true, - "total": false, - "values": true - }, - "lines": true, - "linewidth": 1, - "links": [], - "nullPointMode": "null", - "options": { - "alertThreshold": true - }, - "paceLength": 10, - "percentage": false, - "pluginVersion": "7.5.17", - "pointradius": 2, - "points": false, - "renderer": "flot", - "seriesOverrides": [], - "spaceLength": 10, - "stack": false, - "steppedLine": false, - "targets": [ - { - "exemplar": true, - "expr": "max(ticdc_event_service_scan_window_interval{k8s_cluster=\"$k8s_cluster\", tidb_cluster=\"$tidb_cluster\", keyspace_name=~\"$keyspace_name\", changefeed=~\"$changefeed\", instance=~\"$ticdc_instance\"}) by (keyspace_name, changefeed,instance)", - "format": "time_series", - "interval": "", - "intervalFactor": 1, - "legendFormat": "{{instance}}{{keyspace_name}}-{{changefeed}}", - "refId": "A" - } - ], - "thresholds": [], - "timeFrom": null, - "timeRegions": [], - "timeShift": null, - "title": "Scan window interval", - "tooltip": { - "shared": true, - "sort": 0, - "value_type": "individual" - }, - "type": "graph", - "xaxis": { - "buckets": null, - "mode": "time", - "name": null, - "show": true, - "values": [] - }, - "yaxes": [ - { - "format": "s", - "logBase": 1, - "min": "0", - "show": true - }, - { - "format": "short", - "logBase": 1, - "show": false - } - ], - "yaxis": { - "align": false - } - }, - { - "aliasColors": {}, - "bars": true, - "cacheTimeout": null, - "dashLength": 10, - "dashes": false, - "datasource": "${DS_TEST-CLUSTER}", - "description": "", - "fieldConfig": { - "defaults": {}, - "overrides": [] - }, - "fill": 0, - "fillGradient": 0, - "gridPos": { - "h": 6, - "w": 12, - "x": 12, - "y": 12 - }, - "hiddenSeries": false, - "id": 60030, - "legend": { - "alignAsTable": true, - "avg": false, - "current": true, - "max": false, - "min": false, - "rightSide": false, - "show": true, - "sideWidth": null, - "total": false, - "values": true - }, - "lines": true, - "linewidth": 2, - "links": [], - "nullPointMode": "null", - "options": { - "alertThreshold": true - }, - "percentage": false, - "pluginVersion": "7.5.17", - "pointradius": 2, - "points": false, - "renderer": "flot", - "seriesOverrides": [ - { - "alias": "/approximate current time.*/", - "bars": false - } - ], - "spaceLength": 10, - "stack": false, - "steppedLine": false, - "targets": [ - { - "exemplar": true, - "expr": "max(ticdc_event_service_scan_window_base_ts{k8s_cluster=\"$k8s_cluster\", tidb_cluster=\"$tidb_cluster\", keyspace_name=~\"$keyspace_name\", changefeed=~\"$changefeed\"}) by (keyspace_name,changefeed,instance)", - "format": "time_series", - "hide": false, - "interval": "", - "intervalFactor": 1, - "legendFormat": "{{instance}}{{keyspace_name}}-{{changefeed}}", - "refId": "B" - } - ], - "thresholds": [], - "timeFrom": null, - "timeRegions": [], - "timeShift": null, - "title": "Scan window base ts", - "tooltip": { - "shared": true, - "sort": 0, - "value_type": "individual" - }, - "type": "graph", - "xaxis": { - "buckets": null, - "max": null, - "mode": "time", - "name": null, - "show": true, - "values": [] - }, - "yaxes": [ - { - "format": "none", - "label": null, - "logBase": 1, - "max": null, - "min": null, - "show": true - }, - { - "format": "short", - "label": null, - "logBase": 1, - "max": null, - "min": null, - "show": false - } - ], - "yaxis": { - "align": false, - "alignLevel": null - } - }, { "aliasColors": {}, "bars": false, diff --git a/pkg/config/debug.go b/pkg/config/debug.go index 54a5ba868c..8d78bd9574 100644 --- a/pkg/config/debug.go +++ b/pkg/config/debug.go @@ -167,6 +167,8 @@ type EventServiceConfig struct { // and split update insert spill. LargeTxnThresholdInBytes int64 `toml:"large-txn-threshold-in-bytes" json:"large_txn_threshold_in_bytes"` + // EnableScanWindow is retained for configuration compatibility and is ignored. + // Scan admission is controlled by downstream memory credits instead of a ts window. EnableScanWindow bool `toml:"enable-scan-window" json:"enable_scan_window"` // FIXME: For now we found cdc may OOM when there is a large amount of events to be sent to event collector from a remote event service. diff --git a/pkg/eventservice/dispatcher_stat.go b/pkg/eventservice/dispatcher_stat.go index ffad7257d1..bd70d7df38 100644 --- a/pkg/eventservice/dispatcher_stat.go +++ b/pkg/eventservice/dispatcher_stat.go @@ -140,6 +140,12 @@ type dispatcherStat struct { // this dispatcher. isTaskScanning atomic.Bool + // Scan admission state is fixed-size per dispatcher. A zero waiting timestamp + // means the dispatcher is not waiting for scan resources. + scanPending atomic.Bool + scanAdmissionWaitingSince atomic.Int64 + lastScanGrantEpoch atomic.Uint64 + // activeScanMu protects activeScan and serializes scan registration with // markRemoved. activeScan lets reset/remove cancel the in-flight scan before // cleaning up its large-transaction state, including interrupting TiKV/KMS @@ -368,7 +374,10 @@ func (a *dispatcherStat) resetScanLimit() { a.lastUpdateScanLimitTime.Store(time.Now()) } -type scanTask = *dispatcherStat +type scanTask struct { + *dispatcherStat + grant scanAdmissionGrant +} func (t scanTask) GetKey() common.DispatcherID { return t.id @@ -505,48 +514,32 @@ type changefeedStatus struct { dispatchers sync.Map // common.DispatcherID -> *atomic.Pointer[dispatcherStat] - availableMemoryQuota sync.Map // nodeID -> atomic.Uint64 (memory quota in bytes) - minSentTs atomic.Uint64 - scanInterval atomic.Int64 - reportBandState atomic.Int32 - fastBandState atomic.Int32 - slowBandState atomic.Int32 - - scanWindowController *adaptiveScanWindowController - syncPointInterval time.Duration + scanAdmission *scanAdmissionController + syncPointInterval time.Duration } func newChangefeedStatus(changefeedID common.ChangeFeedID, syncPointInterval time.Duration) *changefeedStatus { - return newChangefeedStatusWithScanWindow( - changefeedID, - syncPointInterval, - isScanWindowEnabled(), - ) + return &changefeedStatus{ + changefeedID: changefeedID, + syncPointInterval: syncPointInterval, + scanAdmission: newScanAdmissionController(), + } } -func isScanWindowEnabled() bool { - cfg := config.GetGlobalServerConfig() - if cfg == nil || cfg.Debug == nil || cfg.Debug.EventService == nil { - return true - } - return cfg.Debug.EventService.EnableScanWindow +func (a *dispatcherStat) markScanAdmissionWaiting(now time.Time) { + a.scanAdmissionWaitingSince.CompareAndSwap(0, now.UnixNano()) } -func newChangefeedStatusWithScanWindow( - changefeedID common.ChangeFeedID, - syncPointInterval time.Duration, - enableScanWindow bool, -) *changefeedStatus { - status := &changefeedStatus{ - changefeedID: changefeedID, - syncPointInterval: syncPointInterval, - } - if enableScanWindow { - status.scanWindowController = newAdaptiveScanWindowController(time.Now()) - } - status.scanInterval.Store(int64(defaultScanInterval)) +func (a *dispatcherStat) clearScanAdmissionWait() { + a.scanAdmissionWaitingSince.Store(0) +} - return status +func (a *dispatcherStat) isScanAdmissionProtected(now time.Time) bool { + if a.enableSyncPoint && a.receivedResolvedTs.Load() > a.nextSyncPoint.Load() { + return true + } + waitingSince := a.scanAdmissionWaitingSince.Load() + return waitingSince > 0 && now.Sub(time.Unix(0, waitingSince)) >= scanAdmissionWaitThreshold } func (c *changefeedStatus) addDispatcher(id common.DispatcherID, dispatcher *atomic.Pointer[dispatcherStat]) { diff --git a/pkg/eventservice/event_broker.go b/pkg/eventservice/event_broker.go index b996871fce..545db81070 100644 --- a/pkg/eventservice/event_broker.go +++ b/pkg/eventservice/event_broker.go @@ -52,8 +52,7 @@ const ( maxReadyEventIntervalSeconds = 10 // defaultSendResolvedTsInterval use to control whether to send a resolvedTs event to the dispatcher when its scan is skipped. - defaultSendResolvedTsInterval = time.Second * 2 - defaultRefreshMinSentResolvedTsInterval = time.Second * 1 + defaultSendResolvedTsInterval = time.Second * 2 ) // eventBroker get event from the eventStore, and send the event to the dispatchers. @@ -190,7 +189,7 @@ func newEventBroker( }) g.Go(func() error { - return c.refreshMinSentResolvedTs(ctx) + return c.runScanAdmissionScheduler(ctx) }) g.Go(func() error { @@ -269,23 +268,6 @@ func (c *eventBroker) sendDDL(ctx context.Context, remoteID node.ID, e *event.DD zap.Uint64("seq", e.Seq), zap.Int64("mode", d.info.GetMode())) } -func (c *eventBroker) refreshMinSentResolvedTs(ctx context.Context) error { - ticker := time.NewTicker(defaultRefreshMinSentResolvedTsInterval) - defer ticker.Stop() - for { - select { - case <-ctx.Done(): - return context.Cause(ctx) - case <-ticker.C: - c.changefeedMap.Range(func(key, value interface{}) bool { - status := value.(*changefeedStatus) - status.refreshMinSentResolvedTs() - return true - }) - } - } -} - func (c *eventBroker) sendSignalResolvedTs(d *dispatcherStat) { // Can't send resolvedTs if there was a interrupted scan task happened before. // A non-zero scan-progress start-ts indicates that there was an interrupted scan task before. @@ -417,7 +399,7 @@ func (c *eventBroker) logUninitializedDispatchers(ctx context.Context) error { // It checks various conditions (dispatcher status, DDL state, max commit ts of dml event) // to decide whether scanning is needed and returns the appropriate time range. // If no valid range is found, it returns an empty ScanRequest. -func (c *eventBroker) getScanTaskRequest(task scanTask) (bool, eventstore.ScanRequest) { +func (c *eventBroker) getScanTaskRequest(task *dispatcherStat) (bool, eventstore.ScanRequest) { // 1. Get the range and resume cursor of the dispatcher. request, needScan := task.getScanRequest() if !needScan { @@ -438,69 +420,8 @@ func (c *eventBroker) getScanTaskRequest(task scanTask) (bool, eventstore.ScanRe return false, eventstore.ScanRequest{} } dataRange.CommitTsEnd = min(dataRange.CommitTsEnd, ddlState.ResolvedTs) - commitTsEndBeforeWindow := dataRange.CommitTsEnd - // If the latest ddl commit ts is in current resolved range and larger than current scan start, - // this dispatcher still has pending ddl to catch up. - hasPendingDDLEventInCurrentRange := dataRange.CommitTsStart < ddlState.MaxEventCommitTs && - ddlState.MaxEventCommitTs <= commitTsEndBeforeWindow - nextSyncPointTs := task.nextSyncPoint.Load() - hasPendingSyncPointEventInCurrentRange := task.enableSyncPoint && commitTsEndBeforeWindow > nextSyncPointTs - scanMaxTs := task.changefeedStat.getScanMaxTs() - if scanMaxTs > 0 { - dataRange.CommitTsEnd = min(dataRange.CommitTsEnd, scanMaxTs) - if dataRange.CommitTsEnd < commitTsEndBeforeWindow { - log.Debug("scan window capped", - zap.Stringer("changefeedID", task.changefeedStat.changefeedID), - zap.Stringer("dispatcherID", task.id), - zap.Uint64("baseTs", task.changefeedStat.minSentTs.Load()), - zap.Uint64("scanMaxTs", scanMaxTs), - zap.Uint64("beforeEndTs", commitTsEndBeforeWindow), - zap.Uint64("afterEndTs", dataRange.CommitTsEnd), - zap.Duration("scanInterval", time.Duration(task.changefeedStat.scanInterval.Load())), - ) - } - } - - if dataRange.CommitTsEnd <= dataRange.CommitTsStart && - (hasPendingDDLEventInCurrentRange || hasPendingSyncPointEventInCurrentRange) { - // Global scan window base can be pinned by other lagging dispatchers. - // For a table with pending ddl or syncpoint in current range, use a local bounded step - // to keep this dispatcher making forward progress, so barrier coverage can eventually complete. - interval := time.Duration(task.changefeedStat.scanInterval.Load()) - if interval <= 0 { - interval = defaultScanInterval - } - localScanMaxTs := oracle.GoTimeToTS(oracle.GetTimeFromTS(dataRange.CommitTsStart).Add(interval)) - if hasPendingSyncPointEventInCurrentRange && nextSyncPointTs >= dataRange.CommitTsStart && - localScanMaxTs <= nextSyncPointTs { - localScanMaxTs = nextSyncPointTs + 1 - } - dataRange.CommitTsEnd = min(commitTsEndBeforeWindow, localScanMaxTs) - if dataRange.CommitTsEnd > dataRange.CommitTsStart { - log.Info("scan window local advance due to pending barrier event", - zap.Stringer("changefeedID", task.changefeedStat.changefeedID), - zap.Stringer("dispatcherID", task.id), - zap.Uint64("startTs", dataRange.CommitTsStart), - zap.Uint64("globalScanMaxTs", scanMaxTs), - zap.Uint64("localScanMaxTs", localScanMaxTs), - zap.Bool("hasPendingDDL", hasPendingDDLEventInCurrentRange), - zap.Uint64("ddlCommitTs", ddlState.MaxEventCommitTs), - zap.Bool("hasPendingSyncPoint", hasPendingSyncPointEventInCurrentRange), - zap.Uint64("nextSyncPointTs", nextSyncPointTs), - zap.Uint64("newEndTs", dataRange.CommitTsEnd)) - } - } hasRowResume := len(request.Cursor.Position) != 0 - // A published row cursor at C came from an earlier scan whose DDL and received - // resolved-ts bounds had already reached C. Since those bounds do not regress, - // only the adaptive scan window can move CommitTsEnd behind C. For example, if - // C=100 and the window caps the end at 80, restore the effective range to - // [100, 100] so scanning resumes after Position inside that transaction. - if hasRowResume && dataRange.CommitTsEnd < dataRange.CommitTsStart { - dataRange.CommitTsEnd = dataRange.CommitTsStart - } - if dataRange.CommitTsEnd <= dataRange.CommitTsStart { // A cursor makes [C, C] meaningful: Position resumes rows inside a // transaction, while TxnStartTs resumes later transactions at the same C. @@ -510,7 +431,6 @@ func (c *eventBroker) getScanTaskRequest(task scanTask) (bool, eventstore.ScanRe return true, request } updateMetricEventServiceSkipResolvedTsCount(task.info.GetMode()) - // Scan range can become empty after applying capping (for example, scan window). // Send a signal resolved-ts event (rate limited) to keep downstream responsive, // but do not advance the watermark here. c.sendSignalResolvedTs(task) @@ -541,7 +461,7 @@ func (c *eventBroker) getScanTaskRequest(task scanTask) (bool, eventstore.ScanRe // // Note: A true return value only indicates potential scanning need, // final determination occurs when the scanTask is actully processed. -func (c *eventBroker) scanReady(task scanTask) bool { +func (c *eventBroker) scanReady(task *dispatcherStat) bool { span := task.info.GetTableSpan() if span.Equal(common.KeyspaceDDLSpan(span.KeyspaceID)) { return false @@ -566,7 +486,7 @@ func (c *eventBroker) scanReady(task scanTask) bool { return ok } -func (c *eventBroker) checkAndSendReady(task scanTask) bool { +func (c *eventBroker) checkAndSendReady(task *dispatcherStat) bool { // only dispatcher with epoch 0 need send ready event. if task.epoch == 0 { now := time.Now().Unix() @@ -593,7 +513,7 @@ func (c *eventBroker) checkAndSendReady(task scanTask) bool { return true } -func (c *eventBroker) sendHandshakeIfNeed(task scanTask) { +func (c *eventBroker) sendHandshakeIfNeed(task *dispatcherStat) { // Fast path. if task.isHandshaked() { return @@ -647,7 +567,7 @@ func (c *eventBroker) emitSyncPointEventIfNeeded(ts uint64, d *dispatcherStat, r } } -func (c *eventBroker) calculateScanLimit(task scanTask) scanLimit { +func (c *eventBroker) calculateScanLimit(task *dispatcherStat) scanLimit { return scanLimit{ maxDMLBytes: task.getCurrentScanLimitInBytes(), } @@ -656,9 +576,11 @@ func (c *eventBroker) calculateScanLimit(task scanTask) scanLimit { func (c *eventBroker) doScan(ctx context.Context, task scanTask) { var interrupted bool defer func() { + task.grant.releaseAll() task.isTaskScanning.Store(false) - if interrupted { - c.pushTask(task, false) + pending := task.scanPending.Swap(false) + if interrupted || pending { + c.pushTask(task.dispatcherStat, false) } }() scanCtx, finishScan := task.beginScan(ctx) @@ -683,7 +605,7 @@ func (c *eventBroker) doScan(ctx context.Context, task scanTask) { return } - needScan, request := c.getScanTaskRequest(task) + needScan, request := c.getScanTaskRequest(task.dispatcherStat) if !needScan { return } @@ -694,6 +616,7 @@ func (c *eventBroker) doScan(ctx context.Context, task scanTask) { // Therefore, we need to consider the priority of each task in the future and allocate rate limits based on priority. // My current idea is to divide rate limits into 3 different levels, and decide which rate limit to use according to lastScanBytes. if !c.scanRateLimiter.AllowN(time.Now(), int(task.lastScanBytes.Load())) { + task.markScanAdmissionWaiting(time.Now()) log.Debug("scan rate limit exceeded", zap.Stringer("dispatcher", task.id), zap.Int64("lastScanBytes", task.lastScanBytes.Load()), @@ -701,7 +624,7 @@ func (c *eventBroker) doScan(ctx context.Context, task scanTask) { return } - item, ok := c.changefeedMap.Load(changefeedID) + _, ok := c.changefeedMap.Load(changefeedID) if !ok { log.Info("changefeed status is not found, skip scan", zap.Stringer("changefeed", changefeedID), @@ -710,48 +633,15 @@ func (c *eventBroker) doScan(ctx context.Context, task scanTask) { return } - status := item.(*changefeedStatus) - item, ok = status.availableMemoryQuota.Load(remoteID) - if !ok { - log.Info("available memory quota is not set, skip scan", - zap.String("changefeed", changefeedID.String()), zap.String("remote", remoteID.String())) - return - } - - available := item.(*atomic.Uint64) - if available.Load() < c.scanLimitInBytes { - task.resetScanLimit() - } - - sl := c.calculateScanLimit(task) - ok = allocQuota(available, uint64(sl.maxDMLBytes)) - if !ok { - log.Debug("changefeed available memory quota is not enough, skip scan", - zap.String("changefeed", changefeedID.String()), - zap.String("remote", remoteID.String()), - zap.Uint64("available", available.Load()), - zap.Uint64("required", uint64(sl.maxDMLBytes))) - c.sendSignalResolvedTs(task) - metrics.EventServiceSkipScanCount.WithLabelValues("changefeed_quota").Inc() - return - } - - if uint64(sl.maxDMLBytes) > task.availableMemoryQuota.Load() { - releaseQuota(available, uint64(sl.maxDMLBytes)) - log.Debug("dispatcher available memory quota is not enough, skip scan", zap.Stringer("dispatcher", task.id), zap.Uint64("available", task.availableMemoryQuota.Load()), zap.Int64("required", int64(sl.maxDMLBytes))) - c.sendSignalResolvedTs(task) - metrics.EventServiceSkipScanCount.WithLabelValues("dispatcher_quota").Inc() - return - } + sl := scanLimit{maxDMLBytes: int64(task.grant.bytes)} scanner := newEventScanner(c.eventStore, c.schemaStore, c.mounter, task.info.GetMode()) - scannedBytes, events, progress, interrupted, err := scanner.scan(scanCtx, task, request, sl) + scannedBytes, events, progress, interrupted, err := scanner.scan(scanCtx, task.dispatcherStat, request, sl) if interrupted { metrics.EventServiceInterruptScanCount.Inc() } if err != nil { - releaseQuota(available, uint64(sl.maxDMLBytes)) if task.isRemoved.Load() { return } @@ -762,11 +652,7 @@ func (c *eventBroker) doScan(ctx context.Context, task scanTask) { zap.Uint64("sentResolvedTs", task.sentResolvedTs.Load()), zap.Error(err)) return } - if scannedBytes < 0 { - releaseQuota(available, uint64(sl.maxDMLBytes)) - } else if scannedBytes < sl.maxDMLBytes { - releaseQuota(available, uint64(sl.maxDMLBytes-scannedBytes)) - } + task.grant.commit(scannedBytes) if scannedBytes > int64(c.scanLimitInBytes) { log.Info("scan bytes exceeded the limit, there must be a big transaction", zap.Stringer("dispatcher", task.id), zap.Int64("scannedBytes", scannedBytes), zap.Int64("limit", int64(c.scanLimitInBytes))) @@ -791,19 +677,19 @@ func (c *eventBroker) doScan(ctx context.Context, task scanTask) { if !ok { log.Panic("expect a DMLEvent, but got", zap.Any("event", e)) } - c.sendDML(remoteID, dmls, task) + c.sendDML(remoteID, dmls, task.dispatcherStat) case event.TypeDDLEvent: ddl, ok := e.(*event.DDLEvent) if !ok { log.Panic("expect a DDLEvent, but got", zap.Any("event", e)) } - c.sendDDL(ctx, remoteID, ddl, task) + c.sendDDL(ctx, remoteID, ddl, task.dispatcherStat) case event.TypeResolvedEvent: re, ok := e.(event.ResolvedEvent) if !ok { log.Panic("expect a ResolvedEvent, but got", zap.Any("event", e)) } - c.sendResolvedTs(task, re.ResolvedTs) + c.sendResolvedTs(task.dispatcherStat, re.ResolvedTs) default: log.Panic("unknown event type", zap.Any("event", e)) } @@ -1006,6 +892,10 @@ func (c *eventBroker) onNotify(d *dispatcherStat, resolvedTs uint64, commitTs ui d.lastReceivedResolvedTsTime.Store(time.Now()) updateMetricEventStoreOutputResolved(d.info.GetMode()) d.onLatestCommitTs(commitTs) + if d.isTaskScanning.Load() { + d.scanPending.Store(true) + return + } if c.scanReady(d) { c.pushTask(d, true) } @@ -1013,25 +903,104 @@ func (c *eventBroker) onNotify(d *dispatcherStat, resolvedTs uint64, commitTs ui } func (c *eventBroker) pushTask(d *dispatcherStat, force bool) { - if d.isRemoved.Load() { + task, ok := c.prepareScanTask(d) + if !ok { return } - // make sure only one scan task can run at the same time. - if !d.isTaskScanning.CompareAndSwap(false, true) { + if force { + c.taskChan[d.scanWorkerIndex] <- task + d.changefeedStat.scanAdmission.recordGrant(d) return } - if force { - c.taskChan[d.scanWorkerIndex] <- d - } else { - timer := time.NewTimer(time.Millisecond * 10) - select { - case c.taskChan[d.scanWorkerIndex] <- d: - case <-timer.C: - d.isTaskScanning.Store(false) + timer := time.NewTimer(time.Millisecond * 10) + defer timer.Stop() + select { + case c.taskChan[d.scanWorkerIndex] <- task: + d.changefeedStat.scanAdmission.recordGrant(d) + case <-timer.C: + c.releasePreparedScanTask(task) + } +} + +func (c *eventBroker) tryPushWaitingTask(d *dispatcherStat) { + task, ok := c.prepareScanTask(d) + if !ok { + return + } + select { + case c.taskChan[d.scanWorkerIndex] <- task: + d.changefeedStat.scanAdmission.recordGrant(d) + default: + c.releasePreparedScanTask(task) + } +} + +func (c *eventBroker) prepareScanTask(d *dispatcherStat) (scanTask, bool) { + if d.isRemoved.Load() { + return scanTask{}, false + } + + // isTaskScanning covers both queued and running tasks. + if !d.isTaskScanning.CompareAndSwap(false, true) { + d.scanPending.Store(true) + return scanTask{}, false + } + + now := time.Now() + available, ok := d.changefeedStat.scanAdmission.availableMemory(node.ID(d.info.GetServerID())) + if ok && available < c.scanLimitInBytes { + d.resetScanLimit() + } + scanLimit := c.calculateScanLimit(d) + dispatcherAvailable := d.availableMemoryQuota.Load() + if uint64(scanLimit.maxDMLBytes) > dispatcherAvailable { + d.resetScanLimit() + scanLimit = c.calculateScanLimit(d) + } + if ok && d.isScanAdmissionProtected(now) { + protectedBytes := min(uint64(maxScanLimitInBytes), available, dispatcherAvailable) + if protectedBytes > uint64(scanLimit.maxDMLBytes) { + scanLimit.maxDMLBytes = int64(protectedBytes) + } + } + + grant, reason := d.changefeedStat.scanAdmission.tryGrant(d, uint64(scanLimit.maxDMLBytes), now) + if reason != scanAdmissionGranted { + d.isTaskScanning.Store(false) + d.markScanAdmissionWaiting(now) + c.sendSignalResolvedTs(d) + switch reason { + case scanAdmissionDispatcherQuota: + log.Debug("dispatcher available memory quota is not enough, skip scan", + zap.Stringer("dispatcher", d.id), + zap.Uint64("available", d.availableMemoryQuota.Load()), + zap.Int64("required", scanLimit.maxDMLBytes)) + metrics.EventServiceSkipScanCount.WithLabelValues("dispatcher_quota").Inc() + case scanAdmissionQuotaNotInitialized: + log.Info("available memory quota is not set, skip scan", + zap.Stringer("changefeed", d.changefeedStat.changefeedID), + zap.String("remote", d.info.GetServerID())) + metrics.EventServiceSkipScanCount.WithLabelValues("changefeed_quota").Inc() + default: + log.Debug("changefeed available memory quota is not enough, skip scan", + zap.Stringer("changefeed", d.changefeedStat.changefeedID), + zap.String("remote", d.info.GetServerID()), + zap.Uint64("available", available), + zap.Int64("required", scanLimit.maxDMLBytes), + zap.Bool("protected", d.isScanAdmissionProtected(now))) + metrics.EventServiceSkipScanCount.WithLabelValues("changefeed_quota").Inc() } + return scanTask{}, false } + return scanTask{dispatcherStat: d, grant: grant}, true +} + +func (c *eventBroker) releasePreparedScanTask(task scanTask) { + task.grant.releaseAll() + task.markScanAdmissionWaiting(time.Now()) + task.isTaskScanning.Store(false) } func (c *eventBroker) getDispatcher(id common.DispatcherID) *atomic.Pointer[dispatcherStat] { @@ -1202,7 +1171,6 @@ func (c *eventBroker) removeChangefeedStatus(status *changefeedStatus) { } filter.GetSharedFilterStorage().RemoveFilter(changefeedID) - deleteScanWindowMetrics(changefeedID.String()) } func (c *eventBroker) resetDispatcher(dispatcherInfo DispatcherInfo) error { @@ -1371,9 +1339,6 @@ func (c *eventBroker) getOrSetChangefeedStatus(info DispatcherInfo) *changefeedS return actual.(*changefeedStatus) } log.Info("new changefeed status", zap.Stringer("changefeedID", changefeedID)) - if status.scanWindowController != nil { - initializeScanWindowMetrics(changefeedID.String()) - } return status } @@ -1439,34 +1404,22 @@ func (c *eventBroker) handleCongestionControl(from node.ID, m *event.CongestionC } holder := make(map[common.GID]uint64, len(availables)) - usage := make(map[common.GID]float64, len(availables)) - memoryRelease := make(map[common.GID]uint32, len(availables)) dispatcherAvailable := make(map[common.DispatcherID]uint64, len(availables)) for _, item := range availables { holder[item.Gid] = item.Available - if m.HasUsageRatio() { - usage[item.Gid] = item.UsageRatio - } - memoryRelease[item.Gid] = item.MemoryReleaseCount for dispatcherID, available := range item.DispatcherAvailable { dispatcherAvailable[dispatcherID] = available } } - now := time.Now() c.changefeedMap.Range(func(k, v interface{}) bool { changefeedID := k.(common.ChangeFeedID) changefeed := v.(*changefeedStatus) availableInMsg, ok := holder[changefeedID.ID()] if ok { - changefeed.availableMemoryQuota.Store(from, atomic.NewUint64(availableInMsg)) + changefeed.scanAdmission.updateAvailableMemory(from, availableInMsg) metrics.EventServiceAvailableMemoryQuotaGaugeVec.WithLabelValues(changefeedID.String()).Set(float64(availableInMsg)) } - if m.HasUsageRatio() { - if ratio, okUsage := usage[changefeedID.ID()]; okUsage && ok { - changefeed.updateMemoryUsage(now, ratio, memoryRelease[changefeedID.ID()]) - } - } return true }) diff --git a/pkg/eventservice/event_broker_test.go b/pkg/eventservice/event_broker_test.go index 33342f506d..8e0a1df5e3 100644 --- a/pkg/eventservice/event_broker_test.go +++ b/pkg/eventservice/event_broker_test.go @@ -24,7 +24,6 @@ import ( "github.com/pingcap/log" "github.com/pingcap/ticdc/eventpb" - "github.com/pingcap/ticdc/logservice/eventstore" "github.com/pingcap/ticdc/pkg/common" appcontext "github.com/pingcap/ticdc/pkg/common/context" "github.com/pingcap/ticdc/pkg/common/event" @@ -146,6 +145,9 @@ func TestOnNotify(t *testing.T) { require.Equal(t, disp.loadScanProgress().txnStartTs, uint64(0)) disp.setHandshaked() + status := broker.getOrSetChangefeedStatus(disInfo) + status.scanAdmission.updateAvailableMemory( + node.ID(disp.info.GetServerID()), broker.scanLimitInBytes) // Case 1: The resolvedTs is greater than the startTs, it should be updated. notifyMsgs := notifyMsg{101, 1} @@ -167,6 +169,7 @@ func TestOnNotify(t *testing.T) { notifyMsgs = notifyMsg{103, 101} broker.onNotify(disp, notifyMsgs.resolvedTs, notifyMsgs.latestCommitTs) require.Equal(t, uint64(103), disp.receivedResolvedTs.Load()) + require.True(t, disp.scanPending.Load()) after := time.After(50 * time.Millisecond) select { case <-after: @@ -176,20 +179,20 @@ func TestOnNotify(t *testing.T) { require.Fail(t, "should not trigger a new scan task") } - // Case 4: Do scan, it will update the sentResolvedTs. - status := broker.getOrSetChangefeedStatus(disInfo) - status.availableMemoryQuota.Store(node.ID(task.info.GetServerID()), atomic.NewUint64(broker.scanLimitInBytes)) - + // Case 4: Do scan. The pending notification should be coalesced into exactly + // one follow-up task after the current task finishes. broker.doScan(context.TODO(), task) - require.False(t, disp.isTaskScanning.Load()) require.Equal(t, notifyMsgs.resolvedTs, disp.sentResolvedTs.Load()) + require.True(t, disp.isTaskScanning.Load()) + followUpTask := <-broker.taskChan[disp.scanWorkerIndex] + broker.doScan(context.TODO(), followUpTask) + require.False(t, disp.isTaskScanning.Load()) log.Info("pass case 4") notifyMsgs5 := notifyMsg{104, 101} // Set the schemaStore's maxDDLCommitTs to the sentResolvedTs, so the broker will not scan the schemaStore. ss.maxDDLCommitTs = disp.sentResolvedTs.Load() broker.onNotify(disp, notifyMsgs5.resolvedTs, notifyMsgs5.latestCommitTs) - broker.doScan(context.TODO(), task) require.Equal(t, notifyMsgs5.resolvedTs, disp.sentResolvedTs.Load()) log.Info("Pass case 6") } @@ -209,7 +212,7 @@ func TestAddDispatcherUnregisterOnSchemaStoreError(t *testing.T) { require.Equal(t, uint64(1), es.unregisterCount.Load()) } -func TestDoScanReleasesChangefeedQuotaOnDispatcherQuotaFailure(t *testing.T) { +func TestPrepareScanTaskLeavesChangefeedQuotaOnDispatcherQuotaFailure(t *testing.T) { broker, _, _, _ := newEventBrokerForTest() defer broker.close() @@ -223,12 +226,14 @@ func TestDoScanReleasesChangefeedQuotaOnDispatcherQuotaFailure(t *testing.T) { disp.availableMemoryQuota.Store(minScanLimitInBytes - 1) serverID := node.ID(info.GetServerID()) - changefeedQuota := atomic.NewUint64(minScanLimitInBytes * 2) - status.availableMemoryQuota.Store(serverID, changefeedQuota) + status.scanAdmission.updateAvailableMemory(serverID, minScanLimitInBytes*2) - broker.doScan(context.Background(), disp) + _, ok := broker.prepareScanTask(disp) + require.False(t, ok) - require.Equal(t, uint64(minScanLimitInBytes*2), changefeedQuota.Load()) + available, initialized := status.scanAdmission.availableMemory(serverID) + require.True(t, initialized) + require.Equal(t, uint64(minScanLimitInBytes*2), available) } func TestDoScanReleasesChangefeedQuotaOnScanError(t *testing.T) { @@ -247,8 +252,7 @@ func TestDoScanReleasesChangefeedQuotaOnScanError(t *testing.T) { status := broker.getOrSetChangefeedStatus(info) serverID := node.ID(info.GetServerID()) - changefeedQuota := atomic.NewUint64(minScanLimitInBytes * 2) - status.availableMemoryQuota.Store(serverID, changefeedQuota) + status.scanAdmission.updateAvailableMemory(serverID, minScanLimitInBytes*2) schemaStore.getTableInfoError = errors.New("mock get table info error") require.NoError(t, eventStore.AppendEvents(info.GetID(), 102, &common.RawKVEntry{ @@ -258,9 +262,15 @@ func TestDoScanReleasesChangefeedQuotaOnScanError(t *testing.T) { Value: []byte("value"), })) - broker.doScan(context.Background(), disp) + task, ok := broker.prepareScanTask(disp) + require.True(t, ok) + status.scanAdmission.recordGrant(disp) + broker.doScan(context.Background(), task) - require.Equal(t, uint64(minScanLimitInBytes*2), changefeedQuota.Load()) + available, initialized := status.scanAdmission.availableMemory(serverID) + require.True(t, initialized) + require.Equal(t, uint64(minScanLimitInBytes*2), available) + require.Equal(t, uint64(minScanLimitInBytes*2), disp.availableMemoryQuota.Load()) } func TestTableTriggerDispatcherMetricCount(t *testing.T) { @@ -278,263 +288,67 @@ func TestTableTriggerDispatcherMetricCount(t *testing.T) { require.InDelta(t, baseline, testutil.ToFloat64(metrics.EventServiceDispatcherGauge.WithLabelValues("1")), 1e-9) } -func TestScanRangeCappedByScanWindow(t *testing.T) { - broker, _, _, _ := newEventBrokerForTest() - // Close the broker, so we can catch all message in the test. - broker.close() - - info := newMockDispatcherInfoForTest(t) - info.epoch = 1 - changefeedStatus := broker.getOrSetChangefeedStatus(info) - - disp := newDispatcherStat(info, 1, 1, nil, changefeedStatus) - disp.seq.Store(1) - - dispPtr := &atomic.Pointer[dispatcherStat]{} - dispPtr.Store(disp) - changefeedStatus.addDispatcher(disp.id, dispPtr) - - baseTime := time.Now() - baseTs := oracle.GoTimeToTS(baseTime) - disp.sentResolvedTs.Store(baseTs) - disp.receivedResolvedTs.Store(oracle.GoTimeToTS(baseTime.Add(20 * time.Second))) - disp.eventStoreCommitTs.Store(oracle.GoTimeToTS(baseTime.Add(15 * time.Second))) - changefeedStatus.refreshMinSentResolvedTs() - - needScan, dataRange := broker.getScanTaskRequest(disp) - require.True(t, needScan) - require.Equal(t, oracle.GoTimeToTS(baseTime.Add(defaultScanInterval)), dataRange.Range.CommitTsEnd) -} - -func TestGetScanTaskDataRangeEmptyAfterCappingDoesNotResetScanRange(t *testing.T) { - broker, _, _, _ := newEventBrokerForTest() - // Close the broker, so we can catch all message in the test. - broker.close() - - info := newMockDispatcherInfoForTest(t) - info.epoch = 1 - changefeedStatus := broker.getOrSetChangefeedStatus(info) - - disp := newDispatcherStat(info, 1, 1, nil, changefeedStatus) - disp.seq.Store(1) - - baseTime := time.Now() - baseTs := oracle.GoTimeToTS(baseTime) - commitStart := oracle.GoTimeToTS(baseTime.Add(20 * time.Second)) - lastStartTs := commitStart - 1 - - disp.sentResolvedTs.Store(baseTs) - disp.receivedResolvedTs.Store(oracle.GoTimeToTS(baseTime.Add(40 * time.Second))) - disp.eventStoreCommitTs.Store(commitStart) - disp.updateScanRange(commitStart, lastStartTs) - - changefeedStatus.minSentTs.Store(baseTs) - changefeedStatus.scanInterval.Store(int64(defaultScanInterval)) - - needScan, _ := broker.getScanTaskRequest(disp) - require.False(t, needScan) - require.Equal(t, commitStart, disp.loadScanProgress().txnCommitTs) - require.Equal(t, lastStartTs, disp.loadScanProgress().txnStartTs) -} - -func TestGetScanTaskRequestKeepsRowCursorInsideShrunkWindow(t *testing.T) { +func TestScanRangeIsIndependentAcrossDispatchers(t *testing.T) { broker, _, schemaStore, _ := newEventBrokerForTest() - // Close the broker, so we can catch all messages in the test. - broker.close() - - info := newMockDispatcherInfoForTest(t) - info.epoch = 1 - changefeedStatus := broker.getOrSetChangefeedStatus(info) - disp := newDispatcherStat(info, 1, 1, nil, changefeedStatus) - disp.seq.Store(1) - - baseTime := time.Now() - baseTs := oracle.GoTimeToTS(baseTime) - cursorCommitTs := oracle.GoTimeToTS(baseTime.Add(20 * time.Second)) - resolvedTs := oracle.GoTimeToTS(baseTime.Add(40 * time.Second)) - position := eventstore.ScanPosition("row-cursor") - - disp.sentResolvedTs.Store(baseTs) - disp.receivedResolvedTs.Store(resolvedTs) - disp.eventStoreCommitTs.Store(cursorCommitTs) - disp.updateScanRangeWithPosition(cursorCommitTs, cursorCommitTs-1, position) - changefeedStatus.minSentTs.Store(baseTs) - changefeedStatus.scanInterval.Store(int64(defaultScanInterval)) - schemaStore.resolvedTs = resolvedTs - - needScan, request := broker.getScanTaskRequest(disp) - require.True(t, needScan) - require.Equal(t, cursorCommitTs, request.Range.CommitTsStart) - require.Equal(t, cursorCommitTs, request.Range.CommitTsEnd) - require.Equal(t, position, request.Cursor.Position) -} - -func TestGetScanTaskDataRangeEmptyAfterCappingWithPendingDDLEventUsesLocalWindow(t *testing.T) { - broker, _, ss, _ := newEventBrokerForTest() - // Close the broker, so we can catch all message in the test. - broker.close() - - info := newMockDispatcherInfoForTest(t) - info.epoch = 1 - changefeedStatus := broker.getOrSetChangefeedStatus(info) - - disp := newDispatcherStat(info, 1, 1, nil, changefeedStatus) - disp.seq.Store(1) - - baseTime := time.Now() - baseTs := oracle.GoTimeToTS(baseTime) - commitStart := oracle.GoTimeToTS(baseTime.Add(20 * time.Second)) - ddlCommitTs := oracle.GoTimeToTS(baseTime.Add(23 * time.Second)) - resolvedTs := oracle.GoTimeToTS(baseTime.Add(40 * time.Second)) - - disp.sentResolvedTs.Store(baseTs) - disp.receivedResolvedTs.Store(resolvedTs) - disp.eventStoreCommitTs.Store(commitStart) - disp.updateScanRange(commitStart, commitStart-1) - - changefeedStatus.minSentTs.Store(baseTs) - changefeedStatus.scanInterval.Store(int64(defaultScanInterval)) - - ss.resolvedTs = resolvedTs - ss.maxDDLCommitTs = ddlCommitTs - - needScan, dataRange := broker.getScanTaskRequest(disp) - require.True(t, needScan) - require.Equal(t, commitStart, dataRange.Range.CommitTsStart) - require.Equal(t, oracle.GoTimeToTS(oracle.GetTimeFromTS(commitStart).Add(defaultScanInterval)), dataRange.Range.CommitTsEnd) -} - -func TestGetScanTaskDataRangeEmptyAfterCappingWithPendingSyncPointCrossesSyncPoint(t *testing.T) { - broker, _, ss, _ := newEventBrokerForTest() - // Close the broker, so we can catch all message in the test. broker.close() + changefeedID := common.NewChangefeedID4Test("default", "test") + status := addChangefeedStatusToBrokerForTest(t, broker, changefeedID, 0) baseTime := time.Now() baseTs := oracle.GoTimeToTS(baseTime) - commitStart := oracle.GoTimeToTS(baseTime.Add(20 * time.Second)) - nextSyncPointTs := oracle.GoTimeToTS(baseTime.Add(23 * time.Second)) resolvedTs := oracle.GoTimeToTS(baseTime.Add(40 * time.Second)) - info := newMockDispatcherInfoForTest(t) - info.epoch = 1 - info.enableSyncPoint = true - info.nextSyncPoint = nextSyncPointTs - info.syncPointInterval = 10 * time.Second - changefeedStatus := broker.getOrSetChangefeedStatus(info) - - disp := newDispatcherStat(info, 1, 1, nil, changefeedStatus) - disp.seq.Store(1) - - disp.sentResolvedTs.Store(baseTs) - disp.receivedResolvedTs.Store(resolvedTs) - disp.eventStoreCommitTs.Store(commitStart) - disp.updateScanRange(commitStart, commitStart-1) - - changefeedStatus.minSentTs.Store(baseTs) - changefeedStatus.scanInterval.Store(int64(time.Second)) - - ss.resolvedTs = resolvedTs - ss.maxDDLCommitTs = 0 - - needScan, dataRange := broker.getScanTaskRequest(disp) - require.True(t, needScan) - require.Equal(t, commitStart, dataRange.Range.CommitTsStart) - require.Equal(t, nextSyncPointTs+1, dataRange.Range.CommitTsEnd) -} - -func TestGetScanTaskDataRangeRingWaitWithThreeDispatchersCanAdvancePendingDDL(t *testing.T) { - broker, _, ss, _ := newEventBrokerForTest() - // Close the broker, so we can catch all message in the test. - broker.close() - - changefeedID := common.NewChangefeedID4Test("default", "test") - changefeedStatus := addChangefeedStatusToBrokerForTest(t, broker, changefeedID, 0) - changefeedStatus.scanInterval.Store(int64(1 * time.Second)) - - baseTime := time.Now() - ts100 := oracle.GoTimeToTS(baseTime) - ts101 := oracle.GoTimeToTS(baseTime.Add(1 * time.Second)) - ts102 := oracle.GoTimeToTS(baseTime.Add(2 * time.Second)) - ts103 := oracle.GoTimeToTS(baseTime.Add(3 * time.Second)) - ts110 := oracle.GoTimeToTS(baseTime.Add(10 * time.Second)) - newDispatcher := func(tableID int64, sentTs uint64) *dispatcherStat { - info := newMockDispatcherInfo(t, ts100, common.NewDispatcherID(), tableID, eventpb.ActionType_ACTION_TYPE_REGISTER) + info := newMockDispatcherInfo( + t, baseTs, common.NewDispatcherID(), tableID, + eventpb.ActionType_ACTION_TYPE_REGISTER) info.epoch = 1 - mustInitChangefeedStatusFilter(t, changefeedStatus, info, broker.timezone) - disp := newDispatcherStat(info, 1, 1, nil, changefeedStatus) - disp.seq.Store(1) - disp.sentResolvedTs.Store(sentTs) - disp.lastReceivedHeartbeatTime.Store(time.Now().Unix()) - - dispPtr := &atomic.Pointer[dispatcherStat]{} - dispPtr.Store(disp) - changefeedStatus.addDispatcher(disp.id, dispPtr) - return disp + mustInitChangefeedStatusFilter(t, status, info, broker.timezone) + dispatcher := newDispatcherStat(info, 1, 1, nil, status) + dispatcher.seq.Store(1) + dispatcher.sentResolvedTs.Store(sentTs) + pointer := &atomic.Pointer[dispatcherStat]{} + pointer.Store(dispatcher) + status.addDispatcher(dispatcher.id, pointer) + return dispatcher } - // D0(table trigger) and D2(other table) form the same changefeed. - // D0 lags at ts100, so global scan window base is pinned at ts100. - _ = newDispatcher(common.DDLSpanTableID, ts100) - // D1 is the blocked table waiting to cross a truncate ddl barrier at ts103. - d1 := newDispatcher(1313112, ts101) - _ = newDispatcher(1313999, ts110) - - changefeedStatus.refreshMinSentResolvedTs() - require.Equal(t, ts100, changefeedStatus.minSentTs.Load()) - - d1.receivedResolvedTs.Store(ts110) - d1.eventStoreCommitTs.Store(ts103) - d1.updateScanRange(ts101, ts101-1) - - ss.resolvedTs = ts110 - ss.maxDDLCommitTs = ts103 - - // Round 1: global cap makes range empty (end=ts101), fallback should locally move it to ts102. - needScan, dataRange := broker.getScanTaskRequest(d1) - require.True(t, needScan) - require.Equal(t, ts101, dataRange.Range.CommitTsStart) - require.Equal(t, ts102, dataRange.Range.CommitTsEnd) + _ = newDispatcher(100, baseTs) + fast := newDispatcher(101, oracle.GoTimeToTS(baseTime.Add(20*time.Second))) + fast.receivedResolvedTs.Store(resolvedTs) + fast.eventStoreCommitTs.Store(oracle.GoTimeToTS(baseTime.Add(30 * time.Second))) + schemaStore.resolvedTs = resolvedTs - // Round 2: still globally capped by ts100, but fallback should continue moving to ts103, - // which allows this dispatcher to eventually reach the pending truncate ddl barrier. - d1.updateScanRange(ts102, 0) - needScan, dataRange = broker.getScanTaskRequest(d1) + needScan, request := broker.getScanTaskRequest(fast) require.True(t, needScan) - require.Equal(t, ts102, dataRange.Range.CommitTsStart) - require.Equal(t, ts103, dataRange.Range.CommitTsEnd) -} - -func TestHandleCongestionControlV2DoesNotResetScanIntervalOnMemoryRelease(t *testing.T) { - broker, _, _, _ := newEventBrokerForTest() - defer broker.close() - - changefeedID := common.NewChangefeedID4Test("default", "test") - status := addChangefeedStatusToBrokerForTest(t, broker, changefeedID, time.Second*10) - - status.scanInterval.Store(int64(40 * time.Second)) - - control := event.NewCongestionControlWithVersion(event.CongestionControlVersion2) - control.AddAvailableMemoryWithDispatchersAndUsageAndReleaseCount(changefeedID.ID(), 0, 0.5, nil, 1) - broker.handleCongestionControl(node.ID("event-collector-1"), control) - - require.Equal(t, int64(40*time.Second), status.scanInterval.Load()) + require.Equal(t, resolvedTs, request.Range.CommitTsEnd) } -func TestHandleCongestionControlV1DoesNotAdjustScanInterval(t *testing.T) { - broker, _, _, _ := newEventBrokerForTest() - defer broker.close() - - changefeedID := common.NewChangefeedID4Test("default", "test") - status := addChangefeedStatusToBrokerForTest(t, broker, changefeedID, time.Second*10) +func TestHandleCongestionControlUpdatesScanAdmission(t *testing.T) { + for _, tc := range []struct { + name string + version int + }{ + {name: "v1", version: event.CongestionControlVersion1}, + {name: "v2", version: event.CongestionControlVersion2}, + } { + t.Run(tc.name, func(t *testing.T) { + broker, _, _, _ := newEventBrokerForTest() + defer broker.close() - status.scanInterval.Store(int64(40 * time.Second)) + changefeedID := common.NewChangefeedID4Test("default", t.Name()) + status := addChangefeedStatusToBrokerForTest(t, broker, changefeedID, 0) + control := event.NewCongestionControlWithVersion(tc.version) + control.AddAvailableMemoryWithDispatchers(changefeedID.ID(), 4096, nil) + serverID := node.ID("event-collector-1") - control := event.NewCongestionControl() - control.AddAvailableMemoryWithDispatchers(changefeedID.ID(), 0, nil) - broker.handleCongestionControl(node.ID("event-collector-1"), control) + broker.handleCongestionControl(serverID, control) - require.Equal(t, int64(40*time.Second), status.scanInterval.Load()) + available, ok := status.scanAdmission.availableMemory(serverID) + require.True(t, ok) + require.Equal(t, uint64(4096), available) + }) + } } func TestDoScanSkipWhenChangefeedStatusNotFound(t *testing.T) { @@ -549,6 +363,9 @@ func TestDoScanSkipWhenChangefeedStatusNotFound(t *testing.T) { disp := broker.getDispatcher(disInfo.GetID()).Load() require.NotNil(t, disp) disp.setHandshaked() + status := broker.getOrSetChangefeedStatus(disInfo) + status.scanAdmission.updateAvailableMemory( + node.ID(disp.info.GetServerID()), broker.scanLimitInBytes) broker.onNotify(disp, 102, 101) require.True(t, disp.isTaskScanning.Load()) @@ -588,16 +405,21 @@ func TestDoScanKeepsRowLevelProgressAfterSendingFragment(t *testing.T) { require.NotNil(t, disp) disp.setHandshaked() disp.currentScanLimitInBytes.Store(1) + disp.availableMemoryQuota.Store(broker.scanLimitInBytes) disp.receivedResolvedTs.Store(resolvedTs) disp.eventStoreCommitTs.Store(resolvedTs) status := broker.getOrSetChangefeedStatus(dispInfo) - status.availableMemoryQuota.Store(node.ID(dispInfo.GetServerID()), atomic.NewUint64(broker.scanLimitInBytes)) + status.scanAdmission.updateAvailableMemory( + node.ID(dispInfo.GetServerID()), broker.scanLimitInBytes) mockSchemaStore.AppendDDLEvent(dispInfo.GetTableSpan().TableID, ddlEvent) require.NoError(t, mockStore.AppendEvents(dispInfo.GetID(), resolvedTs, kvEvents...)) - broker.doScan(context.Background(), disp) + task, ok := broker.prepareScanTask(disp) + require.True(t, ok) + status.scanAdmission.recordGrant(disp) + broker.doScan(context.Background(), task) require.Equal(t, resolvedTs, disp.loadScanProgress().txnCommitTs) require.Equal(t, kvEvents[0].StartTs, disp.loadScanProgress().txnStartTs) diff --git a/pkg/eventservice/scan_admission.go b/pkg/eventservice/scan_admission.go new file mode 100644 index 0000000000..1901efeee8 --- /dev/null +++ b/pkg/eventservice/scan_admission.go @@ -0,0 +1,241 @@ +// Copyright 2026 PingCAP, Inc. +// +// Licensed under the Apache License, Version 2.0 (the "License"); +// you may not use this file except in compliance with the License. +// You may obtain a copy of the License at +// +// http://www.apache.org/licenses/LICENSE-2.0 +// +// Unless required by applicable law or agreed to in writing, software +// distributed under the License is distributed on an "AS IS" BASIS, +// See the License for the specific language governing permissions and +// limitations under the License. + +package eventservice + +import ( + "context" + "sync" + "time" + + "github.com/pingcap/ticdc/pkg/node" + "go.uber.org/atomic" +) + +const ( + scanAdmissionSweepInterval = time.Second + scanAdmissionWaitThreshold = scanAdmissionSweepInterval +) + +type scanAdmissionWaitReason int + +const ( + scanAdmissionGranted scanAdmissionWaitReason = iota + scanAdmissionChangefeedQuota + scanAdmissionDispatcherQuota + scanAdmissionProtectedReserve + scanAdmissionQuotaNotInitialized +) + +type nodeScanAdmission struct { + availableMemoryQuota atomic.Uint64 + protectedReserve atomic.Uint64 +} + +type scanAdmissionController struct { + nodes sync.Map // node.ID -> *nodeScanAdmission + nextGrantEpoch atomic.Uint64 +} + +func newScanAdmissionController() *scanAdmissionController { + return &scanAdmissionController{} +} + +func (c *scanAdmissionController) updateAvailableMemory(nodeID node.ID, available uint64) { + actual, _ := c.nodes.LoadOrStore(nodeID, &nodeScanAdmission{}) + actual.(*nodeScanAdmission).availableMemoryQuota.Store(available) +} + +func (c *scanAdmissionController) availableMemory(nodeID node.ID) (uint64, bool) { + value, ok := c.nodes.Load(nodeID) + if !ok { + return 0, false + } + return value.(*nodeScanAdmission).availableMemoryQuota.Load(), true +} + +func (c *scanAdmissionController) clearProtectedReserves() { + c.nodes.Range(func(_, value any) bool { + value.(*nodeScanAdmission).protectedReserve.Store(0) + return true + }) +} + +func (c *scanAdmissionController) setProtectedReserve(nodeID node.ID, reserve uint64) { + value, ok := c.nodes.Load(nodeID) + if !ok { + return + } + value.(*nodeScanAdmission).protectedReserve.Store(reserve) +} + +func (c *scanAdmissionController) tryGrant( + dispatcher *dispatcherStat, + requestedBytes uint64, + now time.Time, +) (scanAdmissionGrant, scanAdmissionWaitReason) { + value, ok := c.nodes.Load(node.ID(dispatcher.info.GetServerID())) + if !ok { + return scanAdmissionGrant{}, scanAdmissionQuotaNotInitialized + } + nodeAdmission := value.(*nodeScanAdmission) + + if !allocQuota(&dispatcher.availableMemoryQuota, requestedBytes) { + return scanAdmissionGrant{}, scanAdmissionDispatcherQuota + } + + protected := dispatcher.isScanAdmissionProtected(now) + reserve := nodeAdmission.protectedReserve.Load() + if !allocQuotaWithReserve(&nodeAdmission.availableMemoryQuota, requestedBytes, reserve, protected) { + releaseQuota(&dispatcher.availableMemoryQuota, requestedBytes) + if !protected && nodeAdmission.availableMemoryQuota.Load() >= requestedBytes { + return scanAdmissionGrant{}, scanAdmissionProtectedReserve + } + return scanAdmissionGrant{}, scanAdmissionChangefeedQuota + } + + return scanAdmissionGrant{ + bytes: requestedBytes, + changefeedAvailable: &nodeAdmission.availableMemoryQuota, + dispatcherAvailable: &dispatcher.availableMemoryQuota, + }, scanAdmissionGranted +} + +func (c *scanAdmissionController) recordGrant(dispatcher *dispatcherStat) { + dispatcher.clearScanAdmissionWait() + dispatcher.lastScanGrantEpoch.Store(c.nextGrantEpoch.Add(1)) +} + +func allocQuotaWithReserve(quota *atomic.Uint64, nBytes uint64, reserve uint64, protected bool) bool { + for { + available := quota.Load() + if available < nBytes { + return false + } + if !protected && available-nBytes < reserve { + return false + } + if quota.CompareAndSwap(available, available-nBytes) { + return true + } + } +} + +type scanAdmissionGrant struct { + bytes uint64 + changefeedAvailable *atomic.Uint64 + dispatcherAvailable *atomic.Uint64 +} + +func (g *scanAdmissionGrant) releaseAll() { + g.commit(0) +} + +func (g *scanAdmissionGrant) commit(scannedBytes int64) { + if g.bytes == 0 { + return + } + usedBytes := uint64(0) + if scannedBytes > 0 { + usedBytes = min(uint64(scannedBytes), g.bytes) + } + unusedBytes := g.bytes - usedBytes + if unusedBytes > 0 { + releaseQuota(g.changefeedAvailable, unusedBytes) + releaseQuota(g.dispatcherAvailable, unusedBytes) + } + g.bytes = 0 +} + +func (c *eventBroker) runScanAdmissionScheduler(ctx context.Context) error { + ticker := time.NewTicker(scanAdmissionSweepInterval) + defer ticker.Stop() + for { + select { + case <-ctx.Done(): + return context.Cause(ctx) + case now := <-ticker.C: + c.scheduleWaitingDispatchers(now) + } + } +} + +func (c *eventBroker) scheduleWaitingDispatchers(now time.Time) { + c.changefeedMap.Range(func(_, value any) bool { + c.scheduleChangefeedWaitingDispatchers(value.(*changefeedStatus), now) + return true + }) +} + +func (c *eventBroker) scheduleChangefeedWaitingDispatchers(status *changefeedStatus, now time.Time) { + protectedWaiters := make(map[node.ID]*dispatcherStat) + regularWaiters := make(map[node.ID]*dispatcherStat) + status.scanAdmission.clearProtectedReserves() + + status.dispatchers.Range(func(_, value any) bool { + dispatcher := value.(*atomic.Pointer[dispatcherStat]).Load() + if !eligibleScanAdmissionWaiter(dispatcher) { + return true + } + nodeID := node.ID(dispatcher.info.GetServerID()) + if dispatcher.isScanAdmissionProtected(now) { + storeEarlierWaiter(protectedWaiters, nodeID, dispatcher) + } else { + storeEarlierWaiter(regularWaiters, nodeID, dispatcher) + } + return true + }) + + for nodeID, dispatcher := range protectedWaiters { + status.scanAdmission.setProtectedReserve(nodeID, maxScanLimitInBytes) + c.tryPushWaitingTask(dispatcher) + } + + for nodeID, dispatcher := range regularWaiters { + if _, hasProtected := protectedWaiters[nodeID]; !hasProtected { + c.tryPushWaitingTask(dispatcher) + } + } +} + +func eligibleScanAdmissionWaiter(dispatcher *dispatcherStat) bool { + return dispatcher != nil && + !dispatcher.isRemoved.Load() && + dispatcher.seq.Load() > 0 && + dispatcher.scanAdmissionWaitingSince.Load() > 0 +} + +func storeEarlierWaiter( + waiters map[node.ID]*dispatcherStat, + nodeID node.ID, + dispatcher *dispatcherStat, +) { + current, ok := waiters[nodeID] + if !ok || scanAdmissionWaiterLess(dispatcher, current) { + waiters[nodeID] = dispatcher + } +} + +func scanAdmissionWaiterLess(left, right *dispatcherStat) bool { + leftEpoch := left.lastScanGrantEpoch.Load() + rightEpoch := right.lastScanGrantEpoch.Load() + if leftEpoch != rightEpoch { + return leftEpoch < rightEpoch + } + leftWaitingSince := left.scanAdmissionWaitingSince.Load() + rightWaitingSince := right.scanAdmissionWaitingSince.Load() + if leftWaitingSince != rightWaitingSince { + return leftWaitingSince < rightWaitingSince + } + return left.id.Less(right.id) +} diff --git a/pkg/eventservice/scan_admission_test.go b/pkg/eventservice/scan_admission_test.go new file mode 100644 index 0000000000..9d61774072 --- /dev/null +++ b/pkg/eventservice/scan_admission_test.go @@ -0,0 +1,223 @@ +// Copyright 2026 PingCAP, Inc. +// +// Licensed under the Apache License, Version 2.0 (the "License"); +// you may not use this file except in compliance with the License. +// You may obtain a copy of the License at +// +// http://www.apache.org/licenses/LICENSE-2.0 +// +// Unless required by applicable law or agreed to in writing, software +// distributed under the License is distributed on an "AS IS" BASIS, +// See the License for the specific language governing permissions and +// limitations under the License. + +package eventservice + +import ( + "sync" + stdatomic "sync/atomic" + "testing" + "time" + + "github.com/pingcap/ticdc/eventpb" + "github.com/pingcap/ticdc/pkg/common" + "github.com/pingcap/ticdc/pkg/node" + "github.com/stretchr/testify/require" + uberatomic "go.uber.org/atomic" +) + +func newScanAdmissionDispatcherForTest(t *testing.T, controller *scanAdmissionController) *dispatcherStat { + t.Helper() + info := newMockDispatcherInfoForTest(t) + info.epoch = 1 + status := newChangefeedStatus(info.GetChangefeedID(), info.GetSyncPointInterval()) + status.scanAdmission = controller + dispatcher := newDispatcherStat(info, 1, 1, nil, status) + dispatcher.availableMemoryQuota.Store(2 * minScanLimitInBytes) + return dispatcher +} + +func TestScanAdmissionGrantReservesAndRefundsBothQuotas(t *testing.T) { + controller := newScanAdmissionController() + dispatcher := newScanAdmissionDispatcherForTest(t, controller) + serverID := node.ID(dispatcher.info.GetServerID()) + controller.updateAvailableMemory(serverID, 2*minScanLimitInBytes) + + grant, reason := controller.tryGrant(dispatcher, minScanLimitInBytes, time.Now()) + require.Equal(t, scanAdmissionGranted, reason) + available, ok := controller.availableMemory(serverID) + require.True(t, ok) + require.Equal(t, uint64(minScanLimitInBytes), available) + require.Equal(t, uint64(minScanLimitInBytes), dispatcher.availableMemoryQuota.Load()) + + grant.releaseAll() + available, ok = controller.availableMemory(serverID) + require.True(t, ok) + require.Equal(t, uint64(2*minScanLimitInBytes), available) + require.Equal(t, uint64(2*minScanLimitInBytes), dispatcher.availableMemoryQuota.Load()) +} + +func TestScanAdmissionFailureLeavesBothQuotasUnchanged(t *testing.T) { + controller := newScanAdmissionController() + dispatcher := newScanAdmissionDispatcherForTest(t, controller) + serverID := node.ID(dispatcher.info.GetServerID()) + controller.updateAvailableMemory(serverID, minScanLimitInBytes-1) + + _, reason := controller.tryGrant(dispatcher, minScanLimitInBytes, time.Now()) + require.Equal(t, scanAdmissionChangefeedQuota, reason) + available, ok := controller.availableMemory(serverID) + require.True(t, ok) + require.Equal(t, uint64(minScanLimitInBytes-1), available) + require.Equal(t, uint64(2*minScanLimitInBytes), dispatcher.availableMemoryQuota.Load()) + + controller.updateAvailableMemory(serverID, 2*minScanLimitInBytes) + dispatcher.availableMemoryQuota.Store(minScanLimitInBytes - 1) + _, reason = controller.tryGrant(dispatcher, minScanLimitInBytes, time.Now()) + require.Equal(t, scanAdmissionDispatcherQuota, reason) + available, ok = controller.availableMemory(serverID) + require.True(t, ok) + require.Equal(t, uint64(2*minScanLimitInBytes), available) + require.Equal(t, uint64(minScanLimitInBytes-1), dispatcher.availableMemoryQuota.Load()) +} + +func TestScanAdmissionProtectedWaiterCanUseReserve(t *testing.T) { + controller := newScanAdmissionController() + dispatcher := newScanAdmissionDispatcherForTest(t, controller) + serverID := node.ID(dispatcher.info.GetServerID()) + controller.updateAvailableMemory(serverID, minScanLimitInBytes) + controller.setProtectedReserve(serverID, minScanLimitInBytes) + + _, reason := controller.tryGrant(dispatcher, minScanLimitInBytes, time.Now()) + require.Equal(t, scanAdmissionProtectedReserve, reason) + + now := time.Now() + dispatcher.scanAdmissionWaitingSince.Store(now.Add(-scanAdmissionWaitThreshold).UnixNano()) + grant, reason := controller.tryGrant(dispatcher, minScanLimitInBytes, now) + require.Equal(t, scanAdmissionGranted, reason) + grant.releaseAll() +} + +func TestScanAdmissionPendingSyncpointIsProtected(t *testing.T) { + controller := newScanAdmissionController() + dispatcher := newScanAdmissionDispatcherForTest(t, controller) + dispatcher.enableSyncPoint = true + dispatcher.nextSyncPoint.Store(100) + dispatcher.receivedResolvedTs.Store(101) + serverID := node.ID(dispatcher.info.GetServerID()) + controller.updateAvailableMemory(serverID, minScanLimitInBytes) + controller.setProtectedReserve(serverID, minScanLimitInBytes) + + grant, reason := controller.tryGrant(dispatcher, minScanLimitInBytes, time.Now()) + require.Equal(t, scanAdmissionGranted, reason) + grant.releaseAll() +} + +func TestScanAdmissionConcurrentGrantsDoNotExceedQuota(t *testing.T) { + controller := newScanAdmissionController() + dispatcher := newScanAdmissionDispatcherForTest(t, controller) + serverID := node.ID(dispatcher.info.GetServerID()) + const grantCount = uint64(8) + controller.updateAvailableMemory(serverID, grantCount*minScanLimitInBytes) + dispatcher.availableMemoryQuota.Store(grantCount * minScanLimitInBytes) + + var successful stdatomic.Uint64 + var wg sync.WaitGroup + for range 64 { + wg.Go(func() { + _, reason := controller.tryGrant(dispatcher, minScanLimitInBytes, time.Now()) + if reason == scanAdmissionGranted { + successful.Add(1) + } + }) + } + wg.Wait() + + require.Equal(t, grantCount, successful.Load()) + available, ok := controller.availableMemory(serverID) + require.True(t, ok) + require.Zero(t, available) + require.Zero(t, dispatcher.availableMemoryQuota.Load()) +} + +func TestScanAdmissionSchedulerSelectsOneOldestProtectedWaiterPerNode(t *testing.T) { + broker, _, _, _ := newEventBrokerForTest() + broker.close() + + firstInfo := newMockDispatcherInfoForTest(t) + firstInfo.epoch = 1 + secondInfo := newMockDispatcherInfoForTest(t) + secondInfo.epoch = 1 + status := newChangefeedStatus(firstInfo.GetChangefeedID(), 0) + + newWaitingDispatcher := func(info *mockDispatcherInfo, waitingSince time.Time) *dispatcherStat { + dispatcher := newDispatcherStat( + info, uint64(len(broker.taskChan)), uint64(len(broker.messageCh)), nil, status) + dispatcher.seq.Store(1) + dispatcher.availableMemoryQuota.Store(2 * maxScanLimitInBytes) + dispatcher.scanAdmissionWaitingSince.Store(waitingSince.UnixNano()) + pointer := &uberatomic.Pointer[dispatcherStat]{} + pointer.Store(dispatcher) + status.addDispatcher(dispatcher.id, pointer) + return dispatcher + } + + now := time.Now() + first := newWaitingDispatcher(firstInfo, now.Add(-2*scanAdmissionWaitThreshold)) + second := newWaitingDispatcher(secondInfo, now.Add(-scanAdmissionWaitThreshold)) + serverID := node.ID(first.info.GetServerID()) + status.scanAdmission.updateAvailableMemory(serverID, 2*maxScanLimitInBytes) + + broker.scheduleChangefeedWaitingDispatchers(status, now) + + require.True(t, first.isTaskScanning.Load()) + require.False(t, second.isTaskScanning.Load()) + task := <-broker.taskChan[first.scanWorkerIndex] + require.Same(t, first, task.dispatcherStat) + require.Equal(t, uint64(maxScanLimitInBytes), task.grant.bytes) + broker.releasePreparedScanTask(task) + + broker.scheduleChangefeedWaitingDispatchers(status, now.Add(scanAdmissionSweepInterval)) + require.False(t, first.isTaskScanning.Load()) + require.True(t, second.isTaskScanning.Load()) + task = <-broker.taskChan[second.scanWorkerIndex] + require.Same(t, second, task.dispatcherStat) + broker.releasePreparedScanTask(task) +} + +func BenchmarkScanAdmissionSweepOneMillionDispatchers(b *testing.B) { + const dispatcherCount = 1_000_000 + now := time.Now() + info := newMockDispatcherInfo(nil, 1, common.NewDispatcherID(), 1, + eventpb.ActionType_ACTION_TYPE_REGISTER) + info.epoch = 1 + status := newChangefeedStatus(info.GetChangefeedID(), 0) + serverID := node.ID(info.GetServerID()) + status.scanAdmission.updateAvailableMemory(serverID, 2*maxScanLimitInBytes) + + for i := range dispatcherCount { + dispatcher := &dispatcherStat{ + id: common.DispatcherID{Low: uint64(i + 1)}, + changefeedStat: status, + info: info, + } + dispatcher.seq.Store(1) + dispatcher.availableMemoryQuota.Store(2 * maxScanLimitInBytes) + dispatcher.scanAdmissionWaitingSince.Store( + now.Add(-scanAdmissionWaitThreshold).UnixNano()) + pointer := &uberatomic.Pointer[dispatcherStat]{} + pointer.Store(dispatcher) + status.addDispatcher(dispatcher.id, pointer) + } + + broker := &eventBroker{ + taskChan: []chan scanTask{make(chan scanTask, 1)}, + scanLimitInBytes: 256 * 1024 * 1024, + } + b.ReportAllocs() + b.ResetTimer() + for range b.N { + broker.scheduleChangefeedWaitingDispatchers(status, time.Now()) + task := <-broker.taskChan[0] + broker.releasePreparedScanTask(task) + } +} diff --git a/pkg/eventservice/scan_window.go b/pkg/eventservice/scan_window.go deleted file mode 100644 index c3ee6b7b97..0000000000 --- a/pkg/eventservice/scan_window.go +++ /dev/null @@ -1,893 +0,0 @@ -// Copyright 2026 PingCAP, Inc. -// -// Licensed under the Apache License, Version 2.0 (the "License"); -// you may not use this file except in compliance with the License. -// You may obtain a copy of the License at -// -// http://www.apache.org/licenses/LICENSE-2.0 -// -// Unless required by applicable law or agreed to in writing, software -// distributed under the License is distributed on an "AS IS" BASIS, -// See the License for the specific language governing permissions and -// limitations under the License. - -package eventservice - -import ( - "sync" - "time" - - "github.com/pingcap/log" - "github.com/pingcap/ticdc/pkg/metrics" - "github.com/tikv/client-go/v2/oracle" - "go.uber.org/atomic" - "go.uber.org/zap" -) - -const ( - // defaultScanInterval is the initial scan interval used when starting up - // or when the current interval is invalid. - defaultScanInterval = 5 * time.Second - - // minScanInterval is the minimum allowed scan interval. Even under critical - // memory pressure, the interval will never go below this value. - minScanInterval = 1 * time.Second - - // maxScanInterval is the maximum allowed scan interval. Even under very low - // memory pressure, the interval will never exceed this value. - maxScanInterval = 30 * time.Minute - - // scanIntervalAdjustCooldown is the minimum time that must pass between - // scan interval increases. This prevents oscillation by enforcing a waiting - // period before allowing another increase. Decreases are not affected by - // this cooldown and are applied immediately. - scanIntervalAdjustCooldown = 30 * time.Second - - // memoryUsageWindowDuration is the duration of the sliding window for - // collecting memory usage samples. Samples older than this duration are - // pruned from the window. - memoryUsageWindowDuration = 30 * time.Second - - // memoryUsageHighThreshold (70%) triggers a moderate reduction of the scan - // interval to 1/2 of its current value when memory usage exceeds this level. - memoryUsageHighThreshold = 0.7 - - // memoryUsageCriticalThreshold (90%) triggers an aggressive reduction of - // the scan interval once memory usage exceeds this level. - memoryUsageCriticalThreshold = 0.9 - - // memoryUsageEmergencyThreshold (98%) triggers the strongest emergency brake. - memoryUsageEmergencyThreshold = 0.98 - - // memoryUsageLowThreshold (20%) allows the scan interval to be increased - // by 25% when both max and average memory usage are below this level. - memoryUsageLowThreshold = 0.2 - - // scanWindowLowPressureFastEMAOffset widens the fast EMA threshold slightly - // for low-pressure recovery decisions. - scanWindowLowPressureFastEMAOffset = 0.03 - - // scanWindowLowPressureSlowEMAOffset widens the slow EMA threshold slightly - // for low-pressure recovery decisions. - scanWindowLowPressureSlowEMAOffset = 0.02 - - // memoryUsageVeryLowThreshold (10%) allows the scan interval to be increased - // by 50% when both max and average memory usage are below this level. This - // increase may exceed the normal sync point interval cap. - memoryUsageVeryLowThreshold = 0.1 - - // scanWindowModeratePressureThreshold is the smoothed usage threshold that - // starts accumulating pressure score for gradual interval reductions. - scanWindowModeratePressureThreshold = 0.55 - - // scanWindowHighPressureThreshold triggers a stronger but still bounded - // interval reduction when sustained high pressure is observed. - scanWindowHighPressureThreshold = 0.75 - - // scanWindowPressureAdjustCooldown is the minimum time between non-critical - // downward adjustments. It prevents the controller from overreacting before - // previous interval changes have time to take effect. - scanWindowPressureAdjustCooldown = 10 * time.Second - - // scanWindowCriticalBrakeCooldown deduplicates repeated critical brakes - // caused by the same short burst. Without this cooldown, one peak retained - // in the usage window can repeatedly trigger critical_brake on every report. - scanWindowCriticalBrakeCooldown = 10 * time.Second - - // scanWindowReleaseRecoveryCooldown is the minimum time after a downward - // adjustment before the controller is allowed to recover upward again. - scanWindowReleaseRecoveryCooldown = 15 * time.Second - - // scanWindowVeryLowRecoveryCooldown is the minimum time after a recent - // instability event before the controller can re-enter the aggressive - // very_low_recovery path. - scanWindowVeryLowRecoveryCooldown = 90 * time.Second - - // scanWindowFloorRecoveryCooldown allows the controller to escape from the - // default floor faster once the observed pressure has clearly fallen. This - // specifically mitigates feedback-lag cases where a late critical report - // pushes the interval to the floor after the real pressure has already eased. - scanWindowFloorRecoveryCooldown = 5 * time.Second - - // scanWindowEmergencyBrakePlateauInterval keeps the emergency brake - // continuous when transitioning from the small-window moderate brake path to - // the large-window strong brake path. - scanWindowEmergencyBrakePlateauInterval = 3 * defaultScanInterval - - // scanWindowEmergencyMinIntervalUnlockSamples is the minimum number of - // observed samples before emergency pressure is allowed to drive the scan - // window below the default floor toward the minimum interval. - scanWindowEmergencyMinIntervalUnlockSamples = 3 - - // scanWindowFastUsageAlpha controls the responsiveness of the short-term EMA. - scanWindowFastUsageAlpha = 0.4 - - // scanWindowSlowUsageAlpha controls the responsiveness of the long-term EMA. - scanWindowSlowUsageAlpha = 0.2 - - // scanWindowPressureTriggerScore is the score required to trigger a gradual - // downward adjustment under sustained but non-critical pressure. - scanWindowPressureTriggerScore = 3.0 - - // scanWindowPressureScoreCeiling bounds the pressure accumulator. - scanWindowPressureScoreCeiling = 8.0 - - // scanWindowPressureReliefPerRelease is the amount of accumulated pressure - // cleared by one downstream release pulse. - scanWindowPressureReliefPerRelease = 2.0 - - // scanWindowTargetBandLower and scanWindowTargetBandUpper define the desired - // operating region for observed pressure related signals. - scanWindowTargetBandLower = 0.30 - scanWindowTargetBandUpper = 0.50 - - // scanWindowStaleDispatcherHeartbeatThreshold is the duration after which a - // dispatcher is treated as stale for scan window base ts calculation if it - // hasn't sent heartbeat updates. This prevents stale dispatchers (for example, - // after frequent table truncate) from blocking scan window advancement for the - // whole changefeed. - // - // Note: This is intentionally much smaller than heartbeatTimeout, which is - // used for actual dispatcher removal. - scanWindowStaleDispatcherHeartbeatThreshold = 1 * time.Minute -) - -type memoryUsageSample struct { - ts time.Time - ratio float64 -} - -type memoryUsageWindow struct { - window time.Duration - mu sync.Mutex - samples []memoryUsageSample -} - -type memoryUsageStats struct { - avg float64 - max float64 - first float64 - last float64 - span time.Duration - cnt int -} - -type scanWindowReport struct { - usageRatio float64 - memoryReleaseCount uint32 -} - -type scanWindowDecisionReason string - -const ( - scanWindowDecisionNone scanWindowDecisionReason = "none" - scanWindowDecisionCriticalBrake scanWindowDecisionReason = "critical_brake" - scanWindowDecisionHighPressure scanWindowDecisionReason = "high_pressure" - scanWindowDecisionSustainedPressure scanWindowDecisionReason = "sustained_pressure" - scanWindowDecisionLowRecovery scanWindowDecisionReason = "low_recovery" - scanWindowDecisionVeryLowRecovery scanWindowDecisionReason = "very_low_recovery" -) - -type scanWindowDecision struct { - newInterval time.Duration - maxInterval time.Duration - reason scanWindowDecisionReason - usage memoryUsageStats - fastUsageEMA float64 - slowUsageEMA float64 - pressureScore float64 -} - -type scanWindowBandState int32 - -const ( - scanWindowBandUnknown scanWindowBandState = iota - scanWindowBandBelow - scanWindowBandIn - scanWindowBandAbove -) - -type adaptiveScanWindowController struct { - mu sync.Mutex - - usageWindow *memoryUsageWindow - - lastAdjustTime time.Time - lastDownAdjustTime time.Time - lastCriticalTime time.Time - lastInstabilityTime time.Time - - fastUsageEMA float64 - slowUsageEMA float64 - emaInitialized bool - - pressureScore float64 -} - -func newMemoryUsageWindow(window time.Duration) *memoryUsageWindow { - return &memoryUsageWindow{ - window: window, - } -} - -func newAdaptiveScanWindowController(now time.Time) *adaptiveScanWindowController { - return &adaptiveScanWindowController{ - usageWindow: newMemoryUsageWindow(memoryUsageWindowDuration), - lastAdjustTime: now, - lastDownAdjustTime: now, - } -} - -func (w *memoryUsageWindow) addSample(now time.Time, ratio float64) { - if ratio < 0 { - ratio = 0 - } - w.mu.Lock() - defer w.mu.Unlock() - - w.samples = append(w.samples, memoryUsageSample{ts: now, ratio: ratio}) - w.pruneLocked(now) -} - -func (w *memoryUsageWindow) reset() { - w.mu.Lock() - defer w.mu.Unlock() - w.samples = nil -} - -func (w *memoryUsageWindow) stats(now time.Time) memoryUsageStats { - w.mu.Lock() - defer w.mu.Unlock() - - w.pruneLocked(now) - if len(w.samples) == 0 { - return memoryUsageStats{} - } - - sum := 0.0 - firstRatio := w.samples[0].ratio - maxRatio := firstRatio - for _, sample := range w.samples { - sum += sample.ratio - if sample.ratio > maxRatio { - maxRatio = sample.ratio - } - } - - return memoryUsageStats{ - avg: sum / float64(len(w.samples)), - max: maxRatio, - first: firstRatio, - last: w.samples[len(w.samples)-1].ratio, - span: now.Sub(w.samples[0].ts), - cnt: len(w.samples), - } -} - -func (w *memoryUsageWindow) pruneLocked(now time.Time) { - cutoff := now.Add(-w.window) - idx := 0 - for idx < len(w.samples) && w.samples[idx].ts.Before(cutoff) { - idx++ - } - if idx > 0 { - w.samples = w.samples[idx:] - } -} - -func (c *changefeedStatus) updateMemoryUsage(now time.Time, usageRatio float64, memoryReleaseCount uint32) { - if c.scanWindowController == nil { - return - } - - normalizedUsageRatio := normalizeUsageRatio(usageRatio) - current := time.Duration(c.scanInterval.Load()) - decision := c.scanWindowController.OnCongestionReport(now, current, c.maxScanInterval(), scanWindowReport{ - usageRatio: normalizedUsageRatio, - memoryReleaseCount: memoryReleaseCount, - }) - c.observeScanWindowControllerMetrics(normalizedUsageRatio, memoryReleaseCount, current, decision) - if decision.newInterval == current { - return - } - - c.scanInterval.Store(int64(decision.newInterval)) - metrics.EventServiceScanWindowIntervalGaugeVec.WithLabelValues(c.changefeedID.String()).Set(decision.newInterval.Seconds()) - - log.Info("scan interval adjusted", - zap.Stringer("changefeedID", c.changefeedID), - zap.String("reason", string(decision.reason)), - zap.Duration("oldInterval", current), - zap.Duration("newInterval", decision.newInterval), - zap.Duration("maxInterval", decision.maxInterval), - zap.Float64("avgUsage", decision.usage.avg), - zap.Float64("maxUsage", decision.usage.max), - zap.Float64("firstUsage", decision.usage.first), - zap.Float64("lastUsage", decision.usage.last), - zap.Float64("fastUsageEMA", decision.fastUsageEMA), - zap.Float64("slowUsageEMA", decision.slowUsageEMA), - zap.Float64("pressureScore", decision.pressureScore), - zap.Uint32("memoryReleaseCount", memoryReleaseCount), - zap.Bool("syncPointEnabled", c.isSyncpointEnabled()), - zap.Duration("syncPointInterval", c.syncPointInterval)) -} - -func initializeScanWindowMetrics(changefeed string) { - metrics.EventServiceScanWindowBaseTsGaugeVec.WithLabelValues(changefeed).Set(0) - metrics.EventServiceScanWindowIntervalGaugeVec.WithLabelValues(changefeed).Set(defaultScanInterval.Seconds()) - metrics.EventServiceScanWindowUsageRatioGaugeVec.WithLabelValues(changefeed, "report").Set(0) - metrics.EventServiceScanWindowUsageRatioGaugeVec.WithLabelValues(changefeed, "avg").Set(0) - metrics.EventServiceScanWindowUsageRatioGaugeVec.WithLabelValues(changefeed, "max").Set(0) - metrics.EventServiceScanWindowUsageEMAGaugeVec.WithLabelValues(changefeed, "fast").Set(0) - metrics.EventServiceScanWindowUsageEMAGaugeVec.WithLabelValues(changefeed, "slow").Set(0) - metrics.EventServiceScanWindowTargetBandGaugeVec.WithLabelValues(changefeed, "report").Set(0) - metrics.EventServiceScanWindowTargetBandGaugeVec.WithLabelValues(changefeed, "fast").Set(0) - metrics.EventServiceScanWindowTargetBandGaugeVec.WithLabelValues(changefeed, "slow").Set(0) - metrics.EventServiceScanWindowPressureScoreGaugeVec.WithLabelValues(changefeed).Set(0) -} - -func deleteScanWindowMetrics(changefeed string) { - // Available memory quota is a changefeed-lifecycle metric that is cleaned - // together with scan window metrics when the changefeed status is removed. - metrics.EventServiceAvailableMemoryQuotaGaugeVec.DeleteLabelValues(changefeed) - metrics.EventServiceScanWindowBaseTsGaugeVec.DeleteLabelValues(changefeed) - metrics.EventServiceScanWindowIntervalGaugeVec.DeleteLabelValues(changefeed) - metrics.EventServiceScanWindowUsageRatioGaugeVec.DeleteLabelValues(changefeed, "report") - metrics.EventServiceScanWindowUsageRatioGaugeVec.DeleteLabelValues(changefeed, "avg") - metrics.EventServiceScanWindowUsageRatioGaugeVec.DeleteLabelValues(changefeed, "max") - metrics.EventServiceScanWindowUsageEMAGaugeVec.DeleteLabelValues(changefeed, "fast") - metrics.EventServiceScanWindowUsageEMAGaugeVec.DeleteLabelValues(changefeed, "slow") - metrics.EventServiceScanWindowTargetBandGaugeVec.DeleteLabelValues(changefeed, "report") - metrics.EventServiceScanWindowTargetBandGaugeVec.DeleteLabelValues(changefeed, "fast") - metrics.EventServiceScanWindowTargetBandGaugeVec.DeleteLabelValues(changefeed, "slow") - metrics.EventServiceScanWindowTargetBandCrossCount.DeleteLabelValues(changefeed, "report") - metrics.EventServiceScanWindowTargetBandCrossCount.DeleteLabelValues(changefeed, "fast") - metrics.EventServiceScanWindowTargetBandCrossCount.DeleteLabelValues(changefeed, "slow") - metrics.EventServiceScanWindowPressureScoreGaugeVec.DeleteLabelValues(changefeed) - metrics.EventServiceScanWindowMemoryReleaseCount.DeleteLabelValues(changefeed) - for _, reason := range []scanWindowDecisionReason{ - scanWindowDecisionNone, - scanWindowDecisionCriticalBrake, - scanWindowDecisionHighPressure, - scanWindowDecisionSustainedPressure, - scanWindowDecisionLowRecovery, - scanWindowDecisionVeryLowRecovery, - } { - metrics.EventServiceScanWindowAdjustCount.DeleteLabelValues(changefeed, string(reason)) - } -} - -func (c *changefeedStatus) observeScanWindowControllerMetrics( - usageRatio float64, - memoryReleaseCount uint32, - current time.Duration, - decision scanWindowDecision, -) { - changefeed := c.changefeedID.String() - metrics.EventServiceScanWindowUsageRatioGaugeVec.WithLabelValues(changefeed, "report").Set(usageRatio) - metrics.EventServiceScanWindowUsageRatioGaugeVec.WithLabelValues(changefeed, "avg").Set(decision.usage.avg) - metrics.EventServiceScanWindowUsageRatioGaugeVec.WithLabelValues(changefeed, "max").Set(decision.usage.max) - metrics.EventServiceScanWindowUsageEMAGaugeVec.WithLabelValues(changefeed, "fast").Set(decision.fastUsageEMA) - metrics.EventServiceScanWindowUsageEMAGaugeVec.WithLabelValues(changefeed, "slow").Set(decision.slowUsageEMA) - c.observeScanWindowTargetBandMetrics(changefeed, "report", usageRatio, &c.reportBandState) - c.observeScanWindowTargetBandMetrics(changefeed, "fast", decision.fastUsageEMA, &c.fastBandState) - c.observeScanWindowTargetBandMetrics(changefeed, "slow", decision.slowUsageEMA, &c.slowBandState) - metrics.EventServiceScanWindowPressureScoreGaugeVec.WithLabelValues(changefeed).Set(decision.pressureScore) - if memoryReleaseCount > 0 { - metrics.EventServiceScanWindowMemoryReleaseCount.WithLabelValues(changefeed).Add(float64(memoryReleaseCount)) - } - if decision.newInterval != current { - metrics.EventServiceScanWindowAdjustCount.WithLabelValues(changefeed, string(decision.reason)).Inc() - } -} - -func (c *changefeedStatus) observeScanWindowTargetBandMetrics( - changefeed string, - metricType string, - value float64, - state *atomic.Int32, -) { - currentState := classifyScanWindowBandState(value) - if currentState == scanWindowBandIn { - metrics.EventServiceScanWindowTargetBandGaugeVec.WithLabelValues(changefeed, metricType).Set(1) - } else { - metrics.EventServiceScanWindowTargetBandGaugeVec.WithLabelValues(changefeed, metricType).Set(0) - } - - previousState := scanWindowBandState(state.Swap(int32(currentState))) - if previousState != scanWindowBandUnknown && previousState != currentState { - metrics.EventServiceScanWindowTargetBandCrossCount.WithLabelValues(changefeed, metricType).Inc() - } -} - -func classifyScanWindowBandState(value float64) scanWindowBandState { - switch { - case value < scanWindowTargetBandLower: - return scanWindowBandBelow - case value > scanWindowTargetBandUpper: - return scanWindowBandAbove - default: - return scanWindowBandIn - } -} - -const ( - minIncreaseSamples = 10 // Minimum samples needed before allowing increase - minIncreaseSpanNumerator = 4 // Observation span must be at least 4/5 of window - minIncreaseSpanDenominator = 5 -) - -func (c *adaptiveScanWindowController) OnCongestionReport(now time.Time, current time.Duration, maxInterval time.Duration, report scanWindowReport) scanWindowDecision { - c.mu.Lock() - defer c.mu.Unlock() - - if current <= 0 { - current = defaultScanInterval - } - if maxInterval < minScanInterval { - maxInterval = minScanInterval - } - - c.usageWindow.addSample(now, report.usageRatio) - usage := c.usageWindow.stats(now) - c.updateUsageEMALocked(report.usageRatio) - - if decision, ok := c.tryCriticalBrakeLocked(now, current, maxInterval, usage); ok { - return decision - } - - c.updatePressureScoreLocked(usage) - if report.memoryReleaseCount > 0 { - c.relievePressureLocked(report.memoryReleaseCount) - } - - if c.shouldReduceForHighPressureLocked(now, usage) { - newInterval := scanWindowPressureInterval(current, max(scaleDuration(current, 3, 4), defaultScanInterval)) - c.noteAdjustmentLocked(now, true) - return scanWindowDecision{ - newInterval: newInterval, - maxInterval: maxInterval, - reason: scanWindowDecisionHighPressure, - usage: usage, - fastUsageEMA: c.fastUsageEMA, - slowUsageEMA: c.slowUsageEMA, - pressureScore: c.pressureScore, - } - } - - if c.shouldReduceForSustainedPressureLocked(now, usage) { - newInterval := scanWindowPressureInterval(current, max(scaleDuration(current, 9, 10), defaultScanInterval)) - c.noteAdjustmentLocked(now, true) - return scanWindowDecision{ - newInterval: newInterval, - maxInterval: maxInterval, - reason: scanWindowDecisionSustainedPressure, - usage: usage, - fastUsageEMA: c.fastUsageEMA, - slowUsageEMA: c.slowUsageEMA, - pressureScore: c.pressureScore, - } - } - - if c.shouldRecoverFromFloorLocked(now, current, usage) { - newInterval := min(scaleDuration(current, 5, 4), maxInterval) - if newInterval > current { - c.noteAdjustmentLocked(now, false) - return scanWindowDecision{ - newInterval: newInterval, - maxInterval: maxInterval, - reason: scanWindowDecisionLowRecovery, - usage: usage, - fastUsageEMA: c.fastUsageEMA, - slowUsageEMA: c.slowUsageEMA, - pressureScore: c.pressureScore, - } - } - } - - if !c.allowedToIncreaseLocked(now, usage) { - return scanWindowDecision{ - newInterval: current, - maxInterval: maxInterval, - reason: scanWindowDecisionNone, - usage: usage, - fastUsageEMA: c.fastUsageEMA, - slowUsageEMA: c.slowUsageEMA, - pressureScore: c.pressureScore, - } - } - - if c.isVeryLowPressureLocked(usage) && c.allowedToVeryLowRecoverLocked(now) { - effectiveMaxInterval := maxScanInterval - numerator, denominator := scanWindowVeryLowRecoveryScale(current) - newInterval := min(scaleDuration(current, numerator, denominator), effectiveMaxInterval) - if newInterval > current { - c.noteAdjustmentLocked(now, false) - return scanWindowDecision{ - newInterval: newInterval, - maxInterval: effectiveMaxInterval, - reason: scanWindowDecisionVeryLowRecovery, - usage: usage, - fastUsageEMA: c.fastUsageEMA, - slowUsageEMA: c.slowUsageEMA, - pressureScore: c.pressureScore, - } - } - } - - if current < maxInterval && c.isLowPressureLocked(usage) { - numerator, denominator := scanWindowLowRecoveryScale(current) - newInterval := min(scaleDuration(current, numerator, denominator), maxInterval) - if newInterval > current { - c.noteAdjustmentLocked(now, false) - return scanWindowDecision{ - newInterval: newInterval, - maxInterval: maxInterval, - reason: scanWindowDecisionLowRecovery, - usage: usage, - fastUsageEMA: c.fastUsageEMA, - slowUsageEMA: c.slowUsageEMA, - pressureScore: c.pressureScore, - } - } - } - - return scanWindowDecision{ - newInterval: current, - maxInterval: maxInterval, - reason: scanWindowDecisionNone, - usage: usage, - fastUsageEMA: c.fastUsageEMA, - slowUsageEMA: c.slowUsageEMA, - pressureScore: c.pressureScore, - } -} - -func scanWindowVeryLowRecoveryScale(current time.Duration) (numerator int64, denominator int64) { - switch { - case current >= 120*time.Second: - return 11, 10 - case current >= 60*time.Second: - return 6, 5 - default: - return 3, 2 - } -} - -func scanWindowEmergencyBrakeInterval(current time.Duration, allowMinInterval bool) time.Duration { - if current <= defaultScanInterval && allowMinInterval { - return max(current/2, minScanInterval) - } - if current <= 6*defaultScanInterval { - return scanWindowPressureInterval(current, max(current/2, defaultScanInterval)) - } - return max(current/4, scanWindowEmergencyBrakePlateauInterval) -} - -func scanWindowPressureInterval(current time.Duration, next time.Duration) time.Duration { - return min(next, current) -} - -func scanWindowLowRecoveryScale(current time.Duration) (numerator int64, denominator int64) { - switch { - case current >= 120*time.Second: - return 21, 20 - case current >= 60*time.Second: - return 11, 10 - default: - return 5, 4 - } -} - -func (c *adaptiveScanWindowController) tryCriticalBrakeLocked( - now time.Time, - current time.Duration, - maxInterval time.Duration, - usage memoryUsageStats, -) (scanWindowDecision, bool) { - if now.Sub(c.lastCriticalTime) < scanWindowCriticalBrakeCooldown { - return scanWindowDecision{}, false - } - - switch { - case usage.last > memoryUsageEmergencyThreshold: - newInterval := scanWindowEmergencyBrakeInterval(current, c.shouldAllowEmergencyMinIntervalLocked(current, usage)) - c.lastCriticalTime = now - c.noteAdjustmentLocked(now, true) - return scanWindowDecision{ - newInterval: newInterval, - maxInterval: maxInterval, - reason: scanWindowDecisionCriticalBrake, - usage: usage, - fastUsageEMA: c.fastUsageEMA, - slowUsageEMA: c.slowUsageEMA, - pressureScore: c.pressureScore, - }, true - case usage.last > memoryUsageCriticalThreshold: - newInterval := scanWindowPressureInterval(current, max(current/2, defaultScanInterval)) - c.lastCriticalTime = now - c.noteAdjustmentLocked(now, true) - return scanWindowDecision{ - newInterval: newInterval, - maxInterval: maxInterval, - reason: scanWindowDecisionCriticalBrake, - usage: usage, - fastUsageEMA: c.fastUsageEMA, - slowUsageEMA: c.slowUsageEMA, - pressureScore: c.pressureScore, - }, true - default: - return scanWindowDecision{}, false - } -} - -func (c *adaptiveScanWindowController) shouldAllowEmergencyMinIntervalLocked( - current time.Duration, - usage memoryUsageStats, -) bool { - return current <= defaultScanInterval && - usage.cnt >= scanWindowEmergencyMinIntervalUnlockSamples && - c.fastUsageEMA >= memoryUsageCriticalThreshold -} - -func (c *adaptiveScanWindowController) shouldRecoverFromFloorLocked( - now time.Time, - current time.Duration, - usage memoryUsageStats, -) bool { - if current > defaultScanInterval { - return false - } - if now.Sub(c.lastAdjustTime) < scanWindowFloorRecoveryCooldown { - return false - } - if now.Sub(c.lastDownAdjustTime) < scanWindowFloorRecoveryCooldown { - return false - } - if usage.cnt < 3 { - return false - } - - return usage.last < 0.35 && - usage.avg < scanWindowModeratePressureThreshold && - c.fastUsageEMA < 0.45 && - c.slowUsageEMA < 0.40 && - c.pressureScore < 1.5 -} - -func (c *adaptiveScanWindowController) updateUsageEMALocked(value float64) { - if !c.emaInitialized { - c.fastUsageEMA = value - c.slowUsageEMA = value - c.emaInitialized = true - return - } - c.fastUsageEMA = ema(c.fastUsageEMA, value, scanWindowFastUsageAlpha) - c.slowUsageEMA = ema(c.slowUsageEMA, value, scanWindowSlowUsageAlpha) -} - -func (c *adaptiveScanWindowController) updatePressureScoreLocked(usage memoryUsageStats) { - switch { - case c.fastUsageEMA >= scanWindowHighPressureThreshold || - c.slowUsageEMA >= scanWindowHighPressureThreshold || - usage.last >= memoryUsageHighThreshold: - c.pressureScore = min(c.pressureScore+2, scanWindowPressureScoreCeiling) - case c.fastUsageEMA >= scanWindowModeratePressureThreshold || - c.slowUsageEMA >= scanWindowModeratePressureThreshold || - usage.avg >= scanWindowModeratePressureThreshold: - c.pressureScore = min(c.pressureScore+1, scanWindowPressureScoreCeiling) - case c.fastUsageEMA < 0.30 && c.slowUsageEMA < 0.25 && usage.last < 0.30: - c.pressureScore = max(0.0, c.pressureScore-1.5) - default: - c.pressureScore = max(0.0, c.pressureScore-0.5) - } -} - -func (c *adaptiveScanWindowController) relievePressureLocked(memoryReleaseCount uint32) { - relief := min(float64(memoryReleaseCount)*scanWindowPressureReliefPerRelease, scanWindowPressureScoreCeiling) - c.pressureScore = max(0.0, c.pressureScore-relief) -} - -func (c *adaptiveScanWindowController) shouldReduceForHighPressureLocked(now time.Time, usage memoryUsageStats) bool { - if now.Sub(c.lastDownAdjustTime) < scanWindowPressureAdjustCooldown { - return false - } - - return c.fastUsageEMA >= scanWindowHighPressureThreshold || - c.slowUsageEMA >= scanWindowHighPressureThreshold || - usage.last >= memoryUsageHighThreshold -} - -func (c *adaptiveScanWindowController) shouldReduceForSustainedPressureLocked(now time.Time, usage memoryUsageStats) bool { - if now.Sub(c.lastDownAdjustTime) < scanWindowPressureAdjustCooldown { - return false - } - if c.pressureScore < scanWindowPressureTriggerScore { - return false - } - return c.fastUsageEMA >= scanWindowModeratePressureThreshold || - c.slowUsageEMA >= scanWindowModeratePressureThreshold || - usage.avg >= scanWindowModeratePressureThreshold -} - -func (c *adaptiveScanWindowController) allowedToIncreaseLocked(now time.Time, usage memoryUsageStats) bool { - minIncreaseSpan := memoryUsageWindowDuration * minIncreaseSpanNumerator / minIncreaseSpanDenominator - return now.Sub(c.lastAdjustTime) >= scanIntervalAdjustCooldown && - now.Sub(c.lastDownAdjustTime) >= scanWindowReleaseRecoveryCooldown && - usage.cnt >= minIncreaseSamples && - usage.span >= minIncreaseSpan && - c.pressureScore < 1 -} - -func (c *adaptiveScanWindowController) allowedToVeryLowRecoverLocked(now time.Time) bool { - if c.lastInstabilityTime.IsZero() { - return true - } - return now.Sub(c.lastInstabilityTime) >= scanWindowVeryLowRecoveryCooldown -} - -func (c *adaptiveScanWindowController) isVeryLowPressureLocked(usage memoryUsageStats) bool { - return usage.max < memoryUsageVeryLowThreshold && - usage.avg < memoryUsageVeryLowThreshold && - c.fastUsageEMA < memoryUsageVeryLowThreshold && - c.slowUsageEMA < memoryUsageVeryLowThreshold -} - -func (c *adaptiveScanWindowController) isLowPressureLocked(usage memoryUsageStats) bool { - return usage.max < memoryUsageLowThreshold && - usage.avg < memoryUsageLowThreshold && - c.fastUsageEMA < memoryUsageLowThreshold+scanWindowLowPressureFastEMAOffset && - c.slowUsageEMA < memoryUsageLowThreshold+scanWindowLowPressureSlowEMAOffset -} - -func (c *adaptiveScanWindowController) noteAdjustmentLocked(now time.Time, downward bool) { - c.lastAdjustTime = now - if downward { - c.lastDownAdjustTime = now - c.lastInstabilityTime = now - } -} - -func (c *adaptiveScanWindowController) setLastAdjustTimeForTest(now time.Time) { - c.mu.Lock() - defer c.mu.Unlock() - c.lastAdjustTime = now -} - -func (c *adaptiveScanWindowController) setLastDownAdjustTimeForTest(now time.Time) { - c.mu.Lock() - defer c.mu.Unlock() - c.lastDownAdjustTime = now -} - -func normalizeUsageRatio(usageRatio float64) float64 { - if usageRatio != usageRatio || usageRatio < 0 { - return 0 - } - if usageRatio > 1 { - return 1 - } - return usageRatio -} - -func ema(previous float64, value float64, alpha float64) float64 { - return previous + alpha*(value-previous) -} - -func (c *changefeedStatus) maxScanInterval() time.Duration { - if !c.isSyncpointEnabled() { - return maxScanInterval - } - - interval := c.syncPointInterval - if interval <= 0 { - return maxScanInterval - } - - if interval < maxScanInterval { - return interval - } - return maxScanInterval -} - -func (c *changefeedStatus) refreshMinSentResolvedTs() { - if c.scanWindowController == nil { - return - } - - now := time.Now() - minSentResolvedTs := ^uint64(0) - minSentResolvedTsWithStale := ^uint64(0) - hasEligible := false - hasNonStale := false - c.dispatchers.Range(func(_ any, value any) bool { - dispatcher := value.(*atomic.Pointer[dispatcherStat]).Load() - if dispatcher == nil || dispatcher.isRemoved.Load() || dispatcher.seq.Load() == 0 { - return true - } - - hasEligible = true - sentResolvedTs := dispatcher.sentResolvedTs.Load() - if sentResolvedTs < minSentResolvedTsWithStale { - minSentResolvedTsWithStale = sentResolvedTs - } - - lastHeartbeatTime := dispatcher.lastReceivedHeartbeatTime.Load() - if lastHeartbeatTime > 0 && - now.Sub(time.Unix(lastHeartbeatTime, 0)) > scanWindowStaleDispatcherHeartbeatThreshold { - log.Info("dispatcher is stale, skip it's sent resolved ts", zap.Stringer("changefeedID", c.changefeedID), zap.Stringer("dispatcherID", dispatcher.id)) - return true - } - - hasNonStale = true - if sentResolvedTs < minSentResolvedTs { - minSentResolvedTs = sentResolvedTs - } - return true - }) - - if !hasEligible { - c.storeMinSentTs(0) - return - } - if !hasNonStale { - c.storeMinSentTs(minSentResolvedTsWithStale) - return - } - c.storeMinSentTs(minSentResolvedTs) -} - -func (c *changefeedStatus) getScanMaxTs() uint64 { - if c.scanWindowController == nil { - return 0 - } - - baseTs := c.minSentTs.Load() - if baseTs == 0 { - return 0 - } - interval := time.Duration(c.scanInterval.Load()) - if interval <= 0 { - interval = defaultScanInterval - } - - return oracle.GoTimeToTS(oracle.GetTimeFromTS(baseTs).Add(interval)) -} - -func (c *changefeedStatus) storeMinSentTs(value uint64) { - prev := c.minSentTs.Load() - if prev == value { - return - } - c.minSentTs.Store(value) - if c.scanWindowController != nil { - metrics.EventServiceScanWindowBaseTsGaugeVec.WithLabelValues(c.changefeedID.String()).Set(float64(value)) - } -} - -func scaleDuration(d time.Duration, numerator int64, denominator int64) time.Duration { - if numerator <= 0 || denominator <= 0 { - return d - } - return time.Duration(int64(d) * numerator / denominator) -} diff --git a/pkg/eventservice/scan_window_test.go b/pkg/eventservice/scan_window_test.go deleted file mode 100644 index bba0ebf221..0000000000 --- a/pkg/eventservice/scan_window_test.go +++ /dev/null @@ -1,486 +0,0 @@ -// Copyright 2026 PingCAP, Inc. -// -// Licensed under the Apache License, Version 2.0 (the "License"); -// you may not use this file except in compliance with the License. -// You may obtain a copy of the License at -// -// http://www.apache.org/licenses/LICENSE-2.0 -// -// Unless required by applicable law or agreed to in writing, software -// distributed under the License is distributed on an "AS IS" BASIS, -// See the License for the specific language governing permissions and -// limitations under the License. - -package eventservice - -import ( - "testing" - "time" - - "github.com/pingcap/ticdc/pkg/common" - "github.com/pingcap/ticdc/pkg/metrics" - "github.com/prometheus/client_golang/prometheus/testutil" - "github.com/stretchr/testify/require" - "github.com/tikv/client-go/v2/oracle" - "go.uber.org/atomic" -) - -func markScanWindowReadyForIncrease(status *changefeedStatus, now time.Time) { - status.scanWindowController.setLastAdjustTimeForTest(now.Add(-scanIntervalAdjustCooldown - time.Second)) - status.scanWindowController.setLastDownAdjustTimeForTest(now.Add(-scanWindowReleaseRecoveryCooldown - time.Second)) -} - -func markScanWindowReadyForDecrease(status *changefeedStatus, now time.Time) { - status.scanWindowController.setLastDownAdjustTimeForTest(now.Add(-scanWindowPressureAdjustCooldown - time.Second)) -} - -func TestAdjustScanIntervalLowPressureSlowsRecoveryForLargeWindow(t *testing.T) { - t.Parallel() - - status := newChangefeedStatus(common.NewChangefeedID4Test("default", "test"), 10*time.Minute) - - now := time.Now() - markScanWindowReadyForIncrease(status, now) - status.scanInterval.Store(int64(80 * time.Second)) - - for i := 0; i <= int(memoryUsageWindowDuration/time.Second); i++ { - status.updateMemoryUsage(now.Add(time.Duration(i)*time.Second), 0.15, 0) - } - require.Equal(t, int64(88*time.Second), status.scanInterval.Load()) -} - -func TestAdjustScanIntervalVeryLowPressureSlowsRecoveryForVeryLargeWindow(t *testing.T) { - t.Parallel() - - status := newChangefeedStatus(common.NewChangefeedID4Test("default", "test"), 10*time.Minute) - - now := time.Now() - markScanWindowReadyForIncrease(status, now) - status.scanInterval.Store(int64(150 * time.Second)) - - for i := 0; i <= int(memoryUsageWindowDuration/time.Second); i++ { - status.updateMemoryUsage(now.Add(time.Duration(i)*time.Second), 0, 0) - } - require.Equal(t, int64(165*time.Second), status.scanInterval.Load()) -} - -func TestAdjustScanIntervalHighPressureUsesBoundedReduction(t *testing.T) { - t.Parallel() - - status := newChangefeedStatus(common.NewChangefeedID4Test("default", "test"), 1*time.Minute) - now := time.Now() - markScanWindowReadyForDecrease(status, now) - - status.scanInterval.Store(int64(40 * time.Second)) - status.updateMemoryUsage(now.Add(memoryUsageWindowDuration), 0.8, 0) - require.Equal(t, int64(30*time.Second), status.scanInterval.Load()) -} - -func TestAdjustScanIntervalDoesNotKeepReducingAfterTransientHighPressure(t *testing.T) { - t.Parallel() - - status := newChangefeedStatus(common.NewChangefeedID4Test("default", t.Name()), 1*time.Minute) - changefeed := status.changefeedID.String() - t.Cleanup(func() { - deleteScanWindowMetrics(changefeed) - }) - - now := time.Now() - markScanWindowReadyForDecrease(status, now) - status.scanInterval.Store(int64(40 * time.Second)) - - status.updateMemoryUsage(now, 0.8, 0) - require.Equal(t, int64(30*time.Second), status.scanInterval.Load()) - - for i := 1; i <= int(scanWindowPressureAdjustCooldown/time.Second)+1; i++ { - status.updateMemoryUsage(now.Add(time.Duration(i)*time.Second), 0.1, 0) - } - require.Equal(t, int64(30*time.Second), status.scanInterval.Load()) -} - -func TestAdjustScanIntervalCriticalPressure(t *testing.T) { - t.Parallel() - - status := newChangefeedStatus(common.NewChangefeedID4Test("default", "test"), 1*time.Minute) - status.scanInterval.Store(int64(40 * time.Second)) - status.updateMemoryUsage(time.Now().Add(memoryUsageWindowDuration), 1, 0) - require.Equal(t, int64(scanWindowEmergencyBrakePlateauInterval), status.scanInterval.Load()) -} - -func TestAdjustScanIntervalCriticalPressureIgnoresLowPressureHistory(t *testing.T) { - t.Parallel() - - status := newChangefeedStatus(common.NewChangefeedID4Test("default", t.Name()), 10*time.Minute) - changefeed := status.changefeedID.String() - t.Cleanup(func() { - deleteScanWindowMetrics(changefeed) - }) - - now := time.Now() - status.scanInterval.Store(int64(40 * time.Second)) - for i := 0; i < 5; i++ { - status.updateMemoryUsage(now.Add(time.Duration(i)*time.Second), 0.05, 0) - } - require.Equal(t, int64(40*time.Second), status.scanInterval.Load()) - - status.updateMemoryUsage(now.Add(5*time.Second), 0.95, 0) - require.Equal(t, int64(20*time.Second), status.scanInterval.Load()) -} - -func TestAdjustScanIntervalCriticalPressureUsesDefaultFloor(t *testing.T) { - t.Parallel() - - status := newChangefeedStatus(common.NewChangefeedID4Test("default", "test"), 10*time.Minute) - status.scanInterval.Store(int64(8 * time.Second)) - status.updateMemoryUsage(time.Now().Add(memoryUsageWindowDuration), 0.95, 0) - require.Equal(t, int64(defaultScanInterval), status.scanInterval.Load()) -} - -func TestAdjustScanIntervalHighPressureDoesNotIncreaseBelowDefaultFloor(t *testing.T) { - t.Parallel() - - status := newChangefeedStatus(common.NewChangefeedID4Test("default", "test"), 1*time.Minute) - now := time.Now() - markScanWindowReadyForDecrease(status, now) - - status.scanInterval.Store(int64(2 * time.Second)) - status.updateMemoryUsage(now.Add(memoryUsageWindowDuration), 0.8, 0) - require.Equal(t, int64(2*time.Second), status.scanInterval.Load()) -} - -func TestAdjustScanIntervalCriticalPressureDoesNotIncreaseBelowDefaultFloor(t *testing.T) { - t.Parallel() - - status := newChangefeedStatus(common.NewChangefeedID4Test("default", "test"), 10*time.Minute) - status.scanInterval.Store(int64(2 * time.Second)) - status.updateMemoryUsage(time.Now().Add(memoryUsageWindowDuration), 0.95, 0) - require.Equal(t, int64(2*time.Second), status.scanInterval.Load()) -} - -func TestAdjustScanIntervalEmergencyPressureUsesModerateBrakeForSmallWindow(t *testing.T) { - t.Parallel() - - status := newChangefeedStatus(common.NewChangefeedID4Test("default", "test"), 10*time.Minute) - status.scanInterval.Store(int64(20 * time.Second)) - status.updateMemoryUsage(time.Now().Add(memoryUsageWindowDuration), 1, 0) - require.Equal(t, int64(10*time.Second), status.scanInterval.Load()) -} - -func TestScanWindowEmergencyBrakeIntervalIsContinuousAtThirtySeconds(t *testing.T) { - t.Parallel() - - require.Equal(t, 15*time.Second, scanWindowEmergencyBrakeInterval(30*time.Second, false)) - require.Equal(t, 15*time.Second, scanWindowEmergencyBrakeInterval(31*time.Second, false)) - require.Equal(t, 15*time.Second, scanWindowEmergencyBrakeInterval(60*time.Second, false)) -} - -func TestScanWindowEmergencyBrakeIntervalUsesStrongBrakeForLargeWindow(t *testing.T) { - t.Parallel() - - require.Equal(t, 20*time.Second, scanWindowEmergencyBrakeInterval(80*time.Second, false)) -} - -func TestAdjustScanIntervalEmergencyPressureUsesDefaultFloorForVerySmallWindow(t *testing.T) { - t.Parallel() - - status := newChangefeedStatus(common.NewChangefeedID4Test("default", "test"), 10*time.Minute) - status.scanInterval.Store(int64(8 * time.Second)) - status.updateMemoryUsage(time.Now().Add(memoryUsageWindowDuration), 1, 0) - require.Equal(t, int64(defaultScanInterval), status.scanInterval.Load()) -} - -func TestAdjustScanIntervalEmergencyPressureDoesNotImmediatelyDropBelowDefaultFloor(t *testing.T) { - t.Parallel() - - status := newChangefeedStatus(common.NewChangefeedID4Test("default", "test"), 10*time.Minute) - status.scanInterval.Store(int64(defaultScanInterval)) - status.updateMemoryUsage(time.Now().Add(memoryUsageWindowDuration), 1, 0) - require.Equal(t, int64(defaultScanInterval), status.scanInterval.Load()) -} - -func TestAdjustScanIntervalEmergencyPressureDoesNotIncreaseBelowDefaultFloor(t *testing.T) { - t.Parallel() - - status := newChangefeedStatus(common.NewChangefeedID4Test("default", "test"), 10*time.Minute) - status.scanInterval.Store(int64(2 * time.Second)) - status.updateMemoryUsage(time.Now().Add(memoryUsageWindowDuration), 1, 0) - require.Equal(t, int64(2*time.Second), status.scanInterval.Load()) -} - -func TestAdjustScanIntervalEmergencyPressureCanReachMinFloorWhenSustained(t *testing.T) { - t.Parallel() - - status := newChangefeedStatus(common.NewChangefeedID4Test("default", "test"), 10*time.Minute) - status.scanInterval.Store(int64(defaultScanInterval)) - start := time.Now() - - for i := 0; i <= int(memoryUsageWindowDuration/time.Second); i++ { - status.updateMemoryUsage(start.Add(time.Duration(i)*time.Second), 1, 0) - } - - require.Equal(t, int64(minScanInterval), status.scanInterval.Load()) -} - -func TestAdjustScanIntervalRecoversFromFloorBeforeNormalIncreaseCooldown(t *testing.T) { - t.Parallel() - - status := newChangefeedStatus(common.NewChangefeedID4Test("default", "test"), 10*time.Minute) - now := time.Now() - status.scanInterval.Store(int64(defaultScanInterval)) - status.scanWindowController.setLastAdjustTimeForTest(now.Add(-scanWindowFloorRecoveryCooldown - time.Second)) - status.scanWindowController.setLastDownAdjustTimeForTest(now.Add(-scanWindowFloorRecoveryCooldown - time.Second)) - - for i, usage := range []float64{0.30, 0.25, 0.20, 0.18, 0.15} { - status.updateMemoryUsage(now.Add(time.Duration(i)*time.Second), usage, 0) - } - require.Greater(t, status.scanInterval.Load(), int64(defaultScanInterval)) -} - -func TestUpdateMemoryUsageDoesNotResetScanIntervalOnMemoryRelease(t *testing.T) { - t.Parallel() - - status := newChangefeedStatus(common.NewChangefeedID4Test("default", "test"), 1*time.Minute) - now := time.Now() - status.scanInterval.Store(int64(40 * time.Second)) - - status.updateMemoryUsage(now, 0.5, 1) - require.Equal(t, int64(40*time.Second), status.scanInterval.Load()) -} - -func TestUpdateMemoryUsageRecordsScanWindowObservationMetrics(t *testing.T) { - status := newChangefeedStatus(common.NewChangefeedID4Test("default", t.Name()), 1*time.Minute) - changefeed := status.changefeedID.String() - t.Cleanup(func() { - deleteScanWindowMetrics(changefeed) - }) - - now := time.Now() - status.scanInterval.Store(int64(40 * time.Second)) - - status.updateMemoryUsage(now, 0.6, 1) - - require.InDelta(t, 0.6, testutil.ToFloat64(metrics.EventServiceScanWindowUsageRatioGaugeVec.WithLabelValues(changefeed, "report")), 1e-9) - require.InDelta(t, 0.6, testutil.ToFloat64(metrics.EventServiceScanWindowUsageRatioGaugeVec.WithLabelValues(changefeed, "avg")), 1e-9) - require.InDelta(t, 0.6, testutil.ToFloat64(metrics.EventServiceScanWindowUsageRatioGaugeVec.WithLabelValues(changefeed, "max")), 1e-9) - require.InDelta(t, 0.6, testutil.ToFloat64(metrics.EventServiceScanWindowUsageEMAGaugeVec.WithLabelValues(changefeed, "fast")), 1e-9) - require.InDelta(t, 0.6, testutil.ToFloat64(metrics.EventServiceScanWindowUsageEMAGaugeVec.WithLabelValues(changefeed, "slow")), 1e-9) - require.InDelta(t, 0, testutil.ToFloat64(metrics.EventServiceScanWindowTargetBandGaugeVec.WithLabelValues(changefeed, "report")), 1e-9) - require.InDelta(t, 0, testutil.ToFloat64(metrics.EventServiceScanWindowTargetBandGaugeVec.WithLabelValues(changefeed, "fast")), 1e-9) - require.InDelta(t, 0, testutil.ToFloat64(metrics.EventServiceScanWindowTargetBandGaugeVec.WithLabelValues(changefeed, "slow")), 1e-9) - require.InDelta(t, 0, testutil.ToFloat64(metrics.EventServiceScanWindowPressureScoreGaugeVec.WithLabelValues(changefeed)), 1e-9) - require.InDelta(t, 1, testutil.ToFloat64(metrics.EventServiceScanWindowMemoryReleaseCount.WithLabelValues(changefeed)), 1e-9) -} - -func TestUpdateMemoryUsageRecordsScanWindowAdjustCount(t *testing.T) { - status := newChangefeedStatus(common.NewChangefeedID4Test("default", t.Name()), 1*time.Minute) - changefeed := status.changefeedID.String() - t.Cleanup(func() { - deleteScanWindowMetrics(changefeed) - }) - - now := time.Now() - markScanWindowReadyForDecrease(status, now) - status.scanInterval.Store(int64(40 * time.Second)) - - status.updateMemoryUsage(now.Add(memoryUsageWindowDuration), 0.8, 0) - - require.Equal(t, int64(30*time.Second), status.scanInterval.Load()) - require.InDelta(t, 1, testutil.ToFloat64(metrics.EventServiceScanWindowAdjustCount.WithLabelValues(changefeed, string(scanWindowDecisionHighPressure))), 1e-9) -} - -func TestUpdateMemoryUsageRecordsScanWindowTargetBandMetrics(t *testing.T) { - status := newChangefeedStatus(common.NewChangefeedID4Test("default", t.Name()), 10*time.Minute) - changefeed := status.changefeedID.String() - t.Cleanup(func() { - deleteScanWindowMetrics(changefeed) - }) - - start := time.Now() - status.updateMemoryUsage(start, 0.20, 0) - status.updateMemoryUsage(start.Add(time.Second), 0.40, 0) - status.updateMemoryUsage(start.Add(2*time.Second), 0.60, 0) - - require.InDelta(t, 0, testutil.ToFloat64(metrics.EventServiceScanWindowTargetBandGaugeVec.WithLabelValues(changefeed, "report")), 1e-9) - require.InDelta(t, 1, testutil.ToFloat64(metrics.EventServiceScanWindowTargetBandGaugeVec.WithLabelValues(changefeed, "fast")), 1e-9) - require.InDelta(t, 1, testutil.ToFloat64(metrics.EventServiceScanWindowTargetBandGaugeVec.WithLabelValues(changefeed, "slow")), 1e-9) - require.InDelta(t, 2, testutil.ToFloat64(metrics.EventServiceScanWindowTargetBandCrossCount.WithLabelValues(changefeed, "report")), 1e-9) - require.InDelta(t, 1, testutil.ToFloat64(metrics.EventServiceScanWindowTargetBandCrossCount.WithLabelValues(changefeed, "fast")), 1e-9) - require.InDelta(t, 1, testutil.ToFloat64(metrics.EventServiceScanWindowTargetBandCrossCount.WithLabelValues(changefeed, "slow")), 1e-9) -} - -func TestAdjustScanIntervalIncreaseWithJitteredSamples(t *testing.T) { - t.Parallel() - - status := newChangefeedStatus(common.NewChangefeedID4Test("default", "test"), 1*time.Minute) - - start := time.Now() - markScanWindowReadyForIncrease(status, start) - - status.scanInterval.Store(int64(40 * time.Second)) - - // Use a >1s interval to simulate heartbeat jitter, so the window span will be - // slightly less than memoryUsageWindowDuration. - step := 1100 * time.Millisecond - for i := 0; i < 28; i++ { - status.updateMemoryUsage(start.Add(time.Duration(i)*step), 0.15, 0) - } - require.Equal(t, int64(50*time.Second), status.scanInterval.Load()) -} - -func TestAdjustScanIntervalReducesOnSustainedPressure(t *testing.T) { - t.Parallel() - - status := newChangefeedStatus(common.NewChangefeedID4Test("default", "test"), 1*time.Minute) - now := time.Now() - markScanWindowReadyForDecrease(status, now) - - status.scanInterval.Store(int64(40 * time.Second)) - - status.updateMemoryUsage(now, 0.60, 0) - status.updateMemoryUsage(now.Add(1*time.Second), 0.60, 0) - status.updateMemoryUsage(now.Add(2*time.Second), 0.60, 0) - require.Equal(t, int64(36*time.Second), status.scanInterval.Load()) -} - -func TestAdjustScanIntervalSustainedPressureDoesNotIncreaseBelowDefaultFloor(t *testing.T) { - t.Parallel() - - status := newChangefeedStatus(common.NewChangefeedID4Test("default", "test"), 1*time.Minute) - now := time.Now() - markScanWindowReadyForDecrease(status, now) - - status.scanInterval.Store(int64(2 * time.Second)) - - status.updateMemoryUsage(now, 0.60, 0) - status.updateMemoryUsage(now.Add(1*time.Second), 0.60, 0) - status.updateMemoryUsage(now.Add(2*time.Second), 0.60, 0) - require.Equal(t, int64(2*time.Second), status.scanInterval.Load()) -} - -func TestAdjustScanIntervalDoesNotIncreaseBeforeCooldown(t *testing.T) { - t.Parallel() - - status := newChangefeedStatus(common.NewChangefeedID4Test("default", "test"), 1*time.Minute) - now := time.Now() - status.scanInterval.Store(int64(40 * time.Second)) - - for i := 0; i < 10; i++ { - status.updateMemoryUsage(now.Add(time.Duration(i)*time.Second), 0.05, 0) - } - require.Equal(t, int64(40*time.Second), status.scanInterval.Load()) -} - -func TestRefreshMinSentResolvedTsMinAndSkipRules(t *testing.T) { - t.Parallel() - - status := newChangefeedStatus(common.NewChangefeedID4Test("default", "test"), 1*time.Minute) - - stale := &dispatcherStat{} - stale.seq.Store(1) - stale.sentResolvedTs.Store(10) - stale.lastReceivedHeartbeatTime.Store(time.Now().Add(-scanWindowStaleDispatcherHeartbeatThreshold - time.Second).Unix()) - - removed := &dispatcherStat{} - removed.seq.Store(1) - removed.sentResolvedTs.Store(150) - removed.isRemoved.Store(true) - - uninitialized := &dispatcherStat{} - uninitialized.seq.Store(0) - uninitialized.sentResolvedTs.Store(10) - - first := &dispatcherStat{} - first.seq.Store(1) - first.sentResolvedTs.Store(200) - - second := &dispatcherStat{} - second.seq.Store(1) - second.sentResolvedTs.Store(50) - - stalePtr := &atomic.Pointer[dispatcherStat]{} - stalePtr.Store(stale) - status.addDispatcher(common.NewDispatcherID(), stalePtr) - - removedPtr := &atomic.Pointer[dispatcherStat]{} - removedPtr.Store(removed) - status.addDispatcher(common.NewDispatcherID(), removedPtr) - - uninitializedPtr := &atomic.Pointer[dispatcherStat]{} - uninitializedPtr.Store(uninitialized) - status.addDispatcher(common.NewDispatcherID(), uninitializedPtr) - - firstPtr := &atomic.Pointer[dispatcherStat]{} - firstPtr.Store(first) - status.addDispatcher(common.NewDispatcherID(), firstPtr) - - secondPtr := &atomic.Pointer[dispatcherStat]{} - secondPtr.Store(second) - status.addDispatcher(common.NewDispatcherID(), secondPtr) - - status.refreshMinSentResolvedTs() - require.Equal(t, uint64(50), status.minSentTs.Load()) - - second.isRemoved.Store(true) - status.refreshMinSentResolvedTs() - require.Equal(t, uint64(200), status.minSentTs.Load()) - - stale.isRemoved.Store(true) - first.seq.Store(0) - status.refreshMinSentResolvedTs() - require.Equal(t, uint64(0), status.minSentTs.Load()) -} - -func TestRefreshMinSentResolvedTsStaleFallback(t *testing.T) { - t.Parallel() - - status := newChangefeedStatus(common.NewChangefeedID4Test("default", "test"), 1*time.Minute) - - stale := &dispatcherStat{} - stale.seq.Store(1) - stale.sentResolvedTs.Store(123) - stale.lastReceivedHeartbeatTime.Store(time.Now().Add(-scanWindowStaleDispatcherHeartbeatThreshold - time.Second).Unix()) - - stalePtr := &atomic.Pointer[dispatcherStat]{} - stalePtr.Store(stale) - status.addDispatcher(common.NewDispatcherID(), stalePtr) - - status.refreshMinSentResolvedTs() - require.Equal(t, uint64(123), status.minSentTs.Load()) -} - -func TestGetScanMaxTsFallbackInterval(t *testing.T) { - t.Parallel() - - status := newChangefeedStatus(common.NewChangefeedID4Test("default", "test"), 1*time.Minute) - - baseTime := time.Unix(1234, 0) - baseTs := oracle.GoTimeToTS(baseTime) - status.minSentTs.Store(baseTs) - - status.scanInterval.Store(0) - require.Equal(t, oracle.GoTimeToTS(baseTime.Add(defaultScanInterval)), status.getScanMaxTs()) - - status.scanInterval.Store(int64(10 * time.Second)) - require.Equal(t, oracle.GoTimeToTS(baseTime.Add(10*time.Second)), status.getScanMaxTs()) - - status.minSentTs.Store(0) - require.Equal(t, uint64(0), status.getScanMaxTs()) -} - -func TestScanWindowDisabledSkipsAdjustmentAndCap(t *testing.T) { - t.Parallel() - - status := newChangefeedStatusWithScanWindow( - common.NewChangefeedID4Test("default", "test"), - 1*time.Minute, - false, - ) - require.Nil(t, status.scanWindowController) - - status.scanInterval.Store(int64(40 * time.Second)) - status.updateMemoryUsage(time.Now().Add(memoryUsageWindowDuration), 1, 0) - require.Equal(t, int64(40*time.Second), status.scanInterval.Load()) - - status.minSentTs.Store(oracle.GoTimeToTS(time.Unix(1234, 0))) - require.Equal(t, uint64(0), status.getScanMaxTs()) -} diff --git a/pkg/metrics/event_service.go b/pkg/metrics/event_service.go index 283eb1fdfd..3aedaf1b83 100644 --- a/pkg/metrics/event_service.go +++ b/pkg/metrics/event_service.go @@ -55,69 +55,6 @@ var ( Name: "resolved_ts_lag", Help: "resolved ts lag of eventService in seconds", }, []string{"type"}) - EventServiceScanWindowBaseTsGaugeVec = prometheus.NewGaugeVec( - prometheus.GaugeOpts{ - Namespace: "ticdc", - Subsystem: "event_service", - Name: "scan_window_base_ts", - Help: "The base ts of the scan window for each changefeed", - }, []string{"changefeed"}) - EventServiceScanWindowIntervalGaugeVec = prometheus.NewGaugeVec( - prometheus.GaugeOpts{ - Namespace: "ticdc", - Subsystem: "event_service", - Name: "scan_window_interval", - Help: "The scan window interval in seconds for each changefeed", - }, []string{"changefeed"}) - EventServiceScanWindowUsageRatioGaugeVec = prometheus.NewGaugeVec( - prometheus.GaugeOpts{ - Namespace: "ticdc", - Subsystem: "event_service", - Name: "scan_window_usage_ratio", - Help: "The usage ratio observed by the scan window controller for each changefeed", - }, []string{"changefeed", "type"}) - EventServiceScanWindowUsageEMAGaugeVec = prometheus.NewGaugeVec( - prometheus.GaugeOpts{ - Namespace: "ticdc", - Subsystem: "event_service", - Name: "scan_window_usage_ema", - Help: "The usage EMA values used by the scan window controller for each changefeed", - }, []string{"changefeed", "type"}) - EventServiceScanWindowTargetBandGaugeVec = prometheus.NewGaugeVec( - prometheus.GaugeOpts{ - Namespace: "ticdc", - Subsystem: "event_service", - Name: "scan_window_target_band", - Help: "Whether the observed scan window value is currently inside the target band for each changefeed", - }, []string{"changefeed", "type"}) - EventServiceScanWindowTargetBandCrossCount = prometheus.NewCounterVec( - prometheus.CounterOpts{ - Namespace: "ticdc", - Subsystem: "event_service", - Name: "scan_window_target_band_cross_count", - Help: "The number of target band state changes observed by the scan window controller for each changefeed", - }, []string{"changefeed", "type"}) - EventServiceScanWindowPressureScoreGaugeVec = prometheus.NewGaugeVec( - prometheus.GaugeOpts{ - Namespace: "ticdc", - Subsystem: "event_service", - Name: "scan_window_pressure_score", - Help: "The pressure score maintained by the scan window controller for each changefeed", - }, []string{"changefeed"}) - EventServiceScanWindowMemoryReleaseCount = prometheus.NewCounterVec( - prometheus.CounterOpts{ - Namespace: "ticdc", - Subsystem: "event_service", - Name: "scan_window_memory_release_count", - Help: "The number of memory release events reported to the scan window controller for each changefeed", - }, []string{"changefeed"}) - EventServiceScanWindowAdjustCount = prometheus.NewCounterVec( - prometheus.CounterOpts{ - Namespace: "ticdc", - Subsystem: "event_service", - Name: "scan_window_adjust_count", - Help: "The number of scan window adjustments made by the controller for each changefeed", - }, []string{"changefeed", "reason"}) EventServiceScanDuration = prometheus.NewHistogram( prometheus.HistogramOpts{ Namespace: "ticdc", @@ -261,15 +198,6 @@ func initEventServiceMetrics(registry *prometheus.Registry) { registry.MustRegister(EventServiceSendEventDuration) registry.MustRegister(EventServiceResolvedTsGauge) registry.MustRegister(EventServiceResolvedTsLagGauge) - registry.MustRegister(EventServiceScanWindowBaseTsGaugeVec) - registry.MustRegister(EventServiceScanWindowIntervalGaugeVec) - registry.MustRegister(EventServiceScanWindowUsageRatioGaugeVec) - registry.MustRegister(EventServiceScanWindowUsageEMAGaugeVec) - registry.MustRegister(EventServiceScanWindowTargetBandGaugeVec) - registry.MustRegister(EventServiceScanWindowTargetBandCrossCount) - registry.MustRegister(EventServiceScanWindowPressureScoreGaugeVec) - registry.MustRegister(EventServiceScanWindowMemoryReleaseCount) - registry.MustRegister(EventServiceScanWindowAdjustCount) registry.MustRegister(EventServiceScanDuration) registry.MustRegister(EventServiceScannedCount) registry.MustRegister(EventServiceDispatcherGauge) diff --git a/utils/dynstream/memory_control.go b/utils/dynstream/memory_control.go index 93af18b8ed..b4fb4cbf4b 100644 --- a/utils/dynstream/memory_control.go +++ b/utils/dynstream/memory_control.go @@ -53,7 +53,9 @@ type areaMemStat[A Area, P Path, T Event, D Dest, H Handler[A, P, T, D]] struct feedbackChan chan<- Feedback[A, P, D] pathCount atomic.Int64 + totalInputSize atomic.Int64 totalPendingSize atomic.Int64 + totalHandlingSize atomic.Int64 paused atomic.Bool lastSendFeedbackTime atomic.Value algorithm MemoryControlAlgorithm @@ -227,6 +229,10 @@ func (as *areaMemStat[A, P, T, D, H]) memoryUsageRatio() float64 { return float64(as.totalPendingSize.Load()) / float64(as.settings.Load().maxPendingSize) } +func (as *areaMemStat[A, P, T, D, H]) residentSize() int64 { + return as.totalInputSize.Load() + as.totalPendingSize.Load() + as.totalHandlingSize.Load() +} + func (as *areaMemStat[A, P, T, D, H]) updateAreaPauseState(path *pathInfo[A, P, T, D, H]) { pause, resume, memoryUsageRatio := as.algorithm.ShouldPauseArea( as.paused.Load(), @@ -340,8 +346,12 @@ func (m *memControl[A, P, T, D, H]) addPathToArea(path *pathInfo[A, P, T, D, H], // This method is called after the path is removed. func (m *memControl[A, P, T, D, H]) removePathFromArea(path *pathInfo[A, P, T, D, H]) { area := path.areaMemStat + inputSize := path.inputSize.Swap(0) + area.totalInputSize.Add(-inputSize) pendingSize := path.pendingSize.Swap(0) area.decPendingSize(path, pendingSize) + handlingSize := path.handlingSize.Swap(0) + area.totalHandlingSize.Add(-handlingSize) m.mutex.Lock() defer m.mutex.Unlock() @@ -362,12 +372,12 @@ func (m *memControl[A, P, T, D, H]) getMetrics() MemoryMetric[A, P] { areaMetric := AreaMemoryMetric[A, P]{ AreaValue: area.area, PathAvailableMemory: make(map[P]int64), - UsedMemoryValue: area.totalPendingSize.Load(), + UsedMemoryValue: area.residentSize(), MaxMemoryValue: int64(area.settings.Load().maxPendingSize), PathMaxMemoryValue: int64(area.settings.Load().pathMaxPendingSize), } area.pathMap.Range(func(k, v any) bool { - usedMemory := v.(*pathInfo[A, P, T, D, H]).pendingSize.Load() + usedMemory := v.(*pathInfo[A, P, T, D, H]).residentSize() availableMemory := max(0, areaMetric.PathMaxMemoryValue-usedMemory) areaMetric.PathAvailableMemory[k.(P)] = availableMemory return true diff --git a/utils/dynstream/memory_control_test.go b/utils/dynstream/memory_control_test.go index 69812ea849..9cedc6bf87 100644 --- a/utils/dynstream/memory_control_test.go +++ b/utils/dynstream/memory_control_test.go @@ -50,10 +50,17 @@ func TestMemControlAddRemovePath(t *testing.T) { mc.addPathToArea(path, settings, feedbackChan) require.NotNil(t, path.areaMemStat) require.Equal(t, int64(1), path.areaMemStat.pathCount.Load()) + path.addInputSize(10) + path.pendingSize.Store(20) + path.areaMemStat.totalPendingSize.Store(20) + path.addHandlingSize(30) + require.Equal(t, int64(60), path.areaMemStat.residentSize()) // Test removing path mc.removePathFromArea(path) require.Equal(t, int64(0), path.areaMemStat.pathCount.Load()) + require.Zero(t, path.residentSize()) + require.Zero(t, path.areaMemStat.residentSize()) require.Empty(t, mc.areaStatMap) } diff --git a/utils/dynstream/parallel_dynamic_stream_test.go b/utils/dynstream/parallel_dynamic_stream_test.go index cc37d87423..c1d1340b3c 100644 --- a/utils/dynstream/parallel_dynamic_stream_test.go +++ b/utils/dynstream/parallel_dynamic_stream_test.go @@ -175,6 +175,73 @@ func TestParallelDynamicStreamMemoryControl(t *testing.T) { require.Equal(t, int64(1), inc.Load()) } +func TestMemoryMetricsIncludeInputAndAwaitingBytes(t *testing.T) { + handler := &mockHandler{} + var handleWait sync.WaitGroup + handleWait.Add(1) + handleReleased := false + stream := newParallelDynamicStream("test", handler, Option{ + StreamCount: 1, + EnableMemoryControl: true, + handleWait: &handleWait, + }) + stream.Start() + defer stream.Close() + defer func() { + if !handleReleased { + handleWait.Done() + } + }() + + settings := NewAreaSettingsWithMaxPendingSize(1024, MemoryControlForEventCollector, "test") + require.NoError(t, stream.AddPath("path1", "dest1", settings)) + + startNotify := &sync.WaitGroup{} + doneNotify := &sync.WaitGroup{} + e := newMockEvent(1, "path1", 0, nil, startNotify, doneNotify) + e.value = 10 + e.await = true + stream.Push("path1", e) + + expectedBytes := int64(stream.eventExtraSize + e.value) + require.Eventually(t, func() bool { + metrics := stream.GetMetrics().MemoryControl.AreaMemoryMetrics + return len(metrics) == 1 && metrics[0].MemoryUsage() == expectedBytes + }, time.Second, 10*time.Millisecond) + + handleWait.Done() + handleReleased = true + startNotify.Wait() + doneNotify.Wait() + + require.Eventually(t, func() bool { + metrics := stream.GetMetrics().MemoryControl.AreaMemoryMetrics + if len(metrics) != 1 || metrics[0].MemoryUsage() != expectedBytes { + return false + } + stream.pathMap.RLock() + path := stream.pathMap.m["path1"] + stream.pathMap.RUnlock() + return path.pendingSize.Load() == 0 + }, time.Second, 10*time.Millisecond) + + stream.Wake("path1") + require.Eventually(t, func() bool { + metrics := stream.GetMetrics().MemoryControl.AreaMemoryMetrics + return len(metrics) == 1 && metrics[0].MemoryUsage() == 0 + }, time.Second, 10*time.Millisecond) + + syncDone := &sync.WaitGroup{} + syncEvent := newMockEvent(2, "path1", 0, nil, nil, syncDone) + syncEvent.value = 10 + stream.Push("path1", syncEvent) + syncDone.Wait() + require.Eventually(t, func() bool { + metrics := stream.GetMetrics().MemoryControl.AreaMemoryMetrics + return len(metrics) == 1 && metrics[0].MemoryUsage() == 0 + }, time.Second, 10*time.Millisecond) +} + func TestFeedBack(t *testing.T) { t.Parallel() fb1 := Feedback[int, string, any]{ diff --git a/utils/dynstream/stream.go b/utils/dynstream/stream.go index 14184ca342..9195a5f921 100644 --- a/utils/dynstream/stream.go +++ b/utils/dynstream/stream.go @@ -147,6 +147,7 @@ func (s *stream[A, P, T, D, H]) addEvent(event eventWrap[A, P, T, D, H]) { if s.closed.Load() { return } + event.pathInfo.addInputSize(int64(event.eventSize)) eventChan := s.eventChan if s.option.UseBuffer { eventChan = s.inChan @@ -158,6 +159,7 @@ func (s *stream[A, P, T, D, H]) addEvent(event eventWrap[A, P, T, D, H]) { // Slow path: with close check while waiting select { case <-s.ctx.Done(): + event.pathInfo.releaseInputSize(int64(event.eventSize)) case eventChan <- event: } } @@ -247,8 +249,10 @@ func (s *stream[A, P, T, D, H]) receiver() { // It handles the events. func (s *stream[A, P, T, D, H]) handleLoop() { handleEvent := func(e eventWrap[A, P, T, D, H]) { + defer e.pathInfo.releaseInputSize(int64(e.eventSize)) switch { case e.wake: + e.pathInfo.releaseHandlingSize() s.eventQueue.wakePath(e.pathInfo) case e.newPath: s.eventQueue.initPath(e.pathInfo) @@ -344,10 +348,13 @@ Loop: batchMetrics.bytes.Observe(float64(nBytes)) path.lastHandleEventTs.Store(uint64(s.handler.GetTimestamp(eventBuf[0]))) + path.addHandlingSize(int64(nBytes)) path.blocking.Store(s.handler.Handle(path.dest, eventBuf...)) if path.blocking.Load() { s.eventQueue.blockPath(path) + } else { + path.releaseHandlingSize() } cleanUpEventBuf() @@ -387,7 +394,9 @@ type pathInfo[A Area, P Path, T Event, D Dest, H Handler[A, P, T, D]] struct { // Fields used by the memory control. areaMemStat *areaMemStat[A, P, T, D, H] - pendingSize atomic.Int64 // The total size(bytes) of pending events in the pendingQueue of the path. + inputSize atomic.Int64 // The total size(bytes) of events waiting in stream input buffers. + pendingSize atomic.Int64 // The total size(bytes) of pending events in the pendingQueue of the path. + handlingSize atomic.Int64 // The batch owned by an asynchronous handler until Wake. lastHandleEventTs atomic.Uint64 } @@ -459,6 +468,58 @@ func (pi *pathInfo[A, P, T, D, H]) updatePendingSize(delta int64) { } } +func (pi *pathInfo[A, P, T, D, H]) residentSize() int64 { + return pi.inputSize.Load() + pi.pendingSize.Load() + pi.handlingSize.Load() +} + +func (pi *pathInfo[A, P, T, D, H]) addInputSize(size int64) { + if size <= 0 || pi.areaMemStat == nil { + return + } + pi.inputSize.Add(size) + pi.areaMemStat.totalInputSize.Add(size) +} + +func (pi *pathInfo[A, P, T, D, H]) releaseInputSize(size int64) { + if size <= 0 || pi.areaMemStat == nil { + return + } + released := subtractTrackedSize(&pi.inputSize, size) + pi.areaMemStat.totalInputSize.Add(-released) +} + +func (pi *pathInfo[A, P, T, D, H]) addHandlingSize(size int64) { + if size <= 0 || pi.areaMemStat == nil { + return + } + pi.handlingSize.Add(size) + pi.areaMemStat.totalHandlingSize.Add(size) + if pi.removed.Load() { + pi.releaseHandlingSize() + } +} + +func (pi *pathInfo[A, P, T, D, H]) releaseHandlingSize() { + if pi.areaMemStat == nil { + return + } + released := pi.handlingSize.Swap(0) + pi.areaMemStat.totalHandlingSize.Add(-released) +} + +func subtractTrackedSize(value *atomic.Int64, size int64) int64 { + for { + current := value.Load() + if current <= 0 { + return 0 + } + released := min(current, size) + if value.CompareAndSwap(current, current-released) { + return released + } + } +} + // eventWrap contains the event and the path info. // It can be a event or a wake signal. type eventWrap[A Area, P Path, T Event, D Dest, H Handler[A, P, T, D]] struct { diff --git a/utils/dynstream/stream_test.go b/utils/dynstream/stream_test.go index 01e35cc1b1..97c44ee1ab 100644 --- a/utils/dynstream/stream_test.go +++ b/utils/dynstream/stream_test.go @@ -32,6 +32,7 @@ type mockEvent struct { path string value int sleep time.Duration + await bool work mockWork @@ -76,10 +77,10 @@ func (h *mockHandler) Handle(dest any, events ...*mockEvent) (await bool) { if event.done != nil { event.done.Done() } - return false + return event.await } -func (h *mockHandler) GetSize(event *mockEvent) int { return 0 } +func (h *mockHandler) GetSize(event *mockEvent) int { return event.value } func (h *mockHandler) GetArea(path string, dest any) int { return 0 } func (h *mockHandler) GetMetricLabel(dest any) string { return "test" } func (h *mockHandler) GetTimestamp(event *mockEvent) Timestamp { return 0 }