From a3cb799fa93fa0306fdb894841352da562bd59a7 Mon Sep 17 00:00:00 2001 From: dongmen <414110582@qq.com> Date: Sun, 26 Jul 2026 21:30:08 +0800 Subject: [PATCH] logpuller: keep bootstrap requests flow-controlled --- .../logpuller/region_failure_handler.go | 4 +- logservice/logpuller/region_req_cache.go | 30 ++++-- logservice/logpuller/region_req_cache_test.go | 96 ++++++++++++++++--- logservice/logpuller/scan_priority_test.go | 77 ++++++++++++++- logservice/logpuller/subscription_client.go | 15 ++- .../logpuller/subscription_client_test.go | 80 +++++++++------- 6 files changed, 240 insertions(+), 62 deletions(-) diff --git a/logservice/logpuller/region_failure_handler.go b/logservice/logpuller/region_failure_handler.go index 55eee0975e..ef65dc4d7d 100644 --- a/logservice/logpuller/region_failure_handler.go +++ b/logservice/logpuller/region_failure_handler.go @@ -138,12 +138,12 @@ func (r *regionFailureHandler) handleError(ctx context.Context, errInfo regionEr } if innerErr.GetCongested() != nil { metricKvCongestedCounter.Inc() - r.client.scheduleRegionRequest(ctx, errInfo.regionInfo, retryPriority) + r.client.scheduleRegionRequest(ctx, errInfo.regionInfo, TaskLowPrior) return nil } if innerErr.GetServerIsBusy() != nil { metricKvIsBusyCounter.Inc() - r.client.scheduleRegionRequest(ctx, errInfo.regionInfo, retryPriority) + r.client.scheduleRegionRequest(ctx, errInfo.regionInfo, TaskLowPrior) return nil } if duplicated := innerErr.GetDuplicateRequest(); duplicated != nil { diff --git a/logservice/logpuller/region_req_cache.go b/logservice/logpuller/region_req_cache.go index b4478cab07..61a37fda1f 100644 --- a/logservice/logpuller/region_req_cache.go +++ b/logservice/logpuller/region_req_cache.go @@ -61,8 +61,9 @@ type requestCache struct { } // pendingCount is a flow control slot counter. - // A slot is acquired when a request is successfully enqueued into pendingQueue (see add), + // A slot is acquired before publishing a request into pendingQueue (see add), // and is released when the request is finished/removed (resolve/markStopped/markDone/clear). + // add releases the slot itself if queue publication fails. // pop and markSent don't change it. If markSent overwrites an existing request for the same region, // it will release a slot for the replaced request to avoid leaking pendingCount. pendingCount atomic.Int64 @@ -92,7 +93,8 @@ func newRequestCache(maxPendingCount int) *requestCache { } // add adds a new region request to the cache -// It blocks if pendingCount >= maxPendingCount until there's space or ctx is cancelled +// Normal data requests are limited to maxPendingCount. Forced data requests can +// use one additional slot, while stop/control requests keep their existing bypass. func (c *requestCache) add(ctx context.Context, region regionInfo, force bool) (bool, error) { start := time.Now() ticker := time.NewTicker(addReqRetryInterval) @@ -100,19 +102,23 @@ func (c *requestCache) add(ctx context.Context, region regionInfo, force bool) ( addReqRetryLimit := addReqRetryLimit for { - current := c.pendingCount.Load() - if current < c.maxPendingCount || force { - // Try to add the request + limit := c.maxPendingCount + if force { + limit++ + } + if c.tryAcquireSlot(limit, region.isStopped()) { + // Try to publish the request after reserving its slot. req := newRegionReq(region) select { case <-ctx.Done(): + c.markDone() return false, ctx.Err() case c.pendingQueue <- req: - c.pendingCount.Inc() cost := time.Since(start) metrics.SubscriptionClientAddRegionRequestDuration.Observe(cost.Seconds()) return true, nil case <-ticker.C: + c.markDone() addReqRetryLimit-- if addReqRetryLimit <= 0 { return false, nil @@ -137,6 +143,18 @@ func (c *requestCache) add(ctx context.Context, region regionInfo, force bool) ( } } +func (c *requestCache) tryAcquireSlot(limit int64, bypassLimit bool) bool { + for { + current := c.pendingCount.Load() + if !bypassLimit && current >= limit { + return false + } + if c.pendingCount.CompareAndSwap(current, current+1) { + return true + } + } +} + // pop gets the next pending request. // Note: it doesn't change pendingCount. The slot acquired in add() should be released later // (e.g. resolve/markStopped/markDone). diff --git a/logservice/logpuller/region_req_cache_test.go b/logservice/logpuller/region_req_cache_test.go index 62706a8542..515d5d7fde 100644 --- a/logservice/logpuller/region_req_cache_test.go +++ b/logservice/logpuller/region_req_cache_test.go @@ -19,6 +19,7 @@ import ( "time" "github.com/pingcap/ticdc/heartbeatpb" + "github.com/pingcap/ticdc/logservice/logpuller/regionlock" "github.com/stretchr/testify/require" "github.com/tikv/client-go/v2/tikv" ) @@ -38,7 +39,9 @@ func createTestRegionInfo(subID SubscriptionID, regionID uint64) regionInfo { span: span, } - return newRegionInfo(verID, span, nil, subscribedSpan, false) + region := newRegionInfo(verID, span, nil, subscribedSpan, false) + region.lockedRangeState = ®ionlock.LockedRangeState{} + return region } func TestRequestCacheAdd_NormalCase(t *testing.T) { @@ -77,14 +80,7 @@ func TestRequestCacheAdd_ForceFlag(t *testing.T) { require.False(t, ok) require.NoError(t, err) - // With force=true, it should still fail because the channel is full - // The force flag only bypasses the pendingCount check, not the channel capacity - region3 := createTestRegionInfo(1, 3) - ok, err = cache.add(ctx, region3, true) - require.False(t, ok) - require.NoError(t, err) - - // consume the pending queue ann add with force + // Move the normal request to sentRequests so the pending queue has room. req, err := cache.pop(ctx) require.NoError(t, err) require.NotNil(t, req) @@ -93,15 +89,91 @@ func TestRequestCacheAdd_ForceFlag(t *testing.T) { cache.markSent(req) require.Equal(t, 1, cache.getPendingCount()) + // A forced data request can use one extra slot. + region3 := createTestRegionInfo(1, 3) ok, err = cache.add(ctx, region3, true) require.True(t, ok) require.NoError(t, err) - // It is 2 since region1 is unresolved require.Equal(t, 2, cache.getPendingCount()) - // resolve region1 - cache.resolve(region1.subscribedSpan.subID, region1.verID.GetID()) + // No additional forced data request can exceed the N+1 ceiling. + req, err = cache.pop(ctx) + require.NoError(t, err) + cache.markSent(req) + region4 := createTestRegionInfo(1, 4) + ok, err = cache.add(ctx, region4, true) + require.False(t, ok) + require.NoError(t, err) + require.Equal(t, 2, cache.getPendingCount()) + + // Stop/control requests keep their existing bypass and remain accounted. + stopRegion := createTestRegionInfo(2, 5) + stopRegion.lockedRangeState = nil + ok, err = cache.add(ctx, stopRegion, true) + require.True(t, ok) + require.NoError(t, err) + require.Equal(t, 3, cache.getPendingCount()) + + stopReq, err := cache.pop(ctx) + require.NoError(t, err) + cache.markSent(stopReq) + cache.markStopped(stopReq.regionInfo.subscribedSpan.subID, stopReq.regionInfo.verID.GetID()) + require.Equal(t, 2, cache.getPendingCount()) + + require.True(t, cache.resolve(region1.subscribedSpan.subID, region1.verID.GetID())) require.Equal(t, 1, cache.getPendingCount()) + ok, err = cache.add(ctx, region4, true) + require.True(t, ok) + require.NoError(t, err) + require.Equal(t, 2, cache.getPendingCount()) +} + +func TestRequestCacheAddRollsBackReservedSlot(t *testing.T) { + cache := newRequestCache(1) + cache.pendingQueue <- newRegionReq(createTestRegionInfo(1, 1)) + + ctx, cancel := context.WithCancel(context.Background()) + cancel() + ok, err := cache.add(ctx, createTestRegionInfo(1, 2), false) + require.False(t, ok) + require.ErrorIs(t, err, context.Canceled) + require.Equal(t, 0, cache.getPendingCount()) +} + +func TestRequestCacheConcurrentForcedAddsStayWithinCeiling(t *testing.T) { + const normalLimit = 10 + cache := newRequestCache(normalLimit) + ctx := context.Background() + + for i := range normalLimit { + ok, err := cache.add(ctx, createTestRegionInfo(1, uint64(i+1)), false) + require.True(t, ok) + require.NoError(t, err) + } + for range normalLimit { + req, err := cache.pop(ctx) + require.NoError(t, err) + cache.markSent(req) + } + + const addCount = 20 + results := make(chan bool, addCount) + for i := range addCount { + go func(regionID uint64) { + ok, err := cache.add(ctx, createTestRegionInfo(1, regionID), true) + require.NoError(t, err) + results <- ok + }(uint64(normalLimit + i + 1)) + } + + successes := 0 + for range addCount { + if <-results { + successes++ + } + } + require.Equal(t, 1, successes) + require.Equal(t, normalLimit+1, cache.getPendingCount()) } func TestRequestCacheAdd_ContextCancellation(t *testing.T) { diff --git a/logservice/logpuller/scan_priority_test.go b/logservice/logpuller/scan_priority_test.go index d5501efcd2..e1259bd5c0 100644 --- a/logservice/logpuller/scan_priority_test.go +++ b/logservice/logpuller/scan_priority_test.go @@ -141,11 +141,86 @@ func TestScanPriorityUsesRestoredRegionProgress(t *testing.T) { retryRegion := newRegionInfo(tikv.NewRegionVerID(1, 1, 2), rawSpan, nil, span, false) client.scheduleRegionRequest(context.Background(), retryRegion, TaskLowPrior) retryTask := popRegionPriorityTask(t, client.regionTaskQueue) - require.Equal(t, TaskHighPrior, retryTask.taskType) + require.Equal(t, TaskLowPrior, retryTask.taskType) require.Equal(t, cdcpb.ScanPriority_SCAN_PRIORITY_HIGH, retryTask.GetRegionInfo().scanPriority) require.False(t, span.priorityPolicy.everCaughtUp.Load()) } +func TestScheduleRegionRequestSeparatesRemoteAndLocalPriority(t *testing.T) { + currentTime := time.Date(2026, time.June, 27, 12, 0, 0, 0, time.UTC) + currentTs := oracle.GoTimeToTS(currentTime) + + for _, tc := range []struct { + name string + priorRemote cdcpb.ScanPriority + inheritedLocal TaskType + startTs uint64 + expectedRemote cdcpb.ScanPriority + expectedLocal TaskType + }{ + { + name: "busy retry preserves remote high", + priorRemote: cdcpb.ScanPriority_SCAN_PRIORITY_HIGH, + inheritedLocal: TaskLowPrior, + startTs: oracle.GoTimeToTS(currentTime.Add(-time.Hour)), + expectedRemote: cdcpb.ScanPriority_SCAN_PRIORITY_HIGH, + expectedLocal: TaskLowPrior, + }, + { + name: "repair is high locally and remotely", + priorRemote: cdcpb.ScanPriority_SCAN_PRIORITY_LOW, + inheritedLocal: TaskHighPrior, + startTs: oracle.GoTimeToTS(currentTime.Add(-time.Hour)), + expectedRemote: cdcpb.ScanPriority_SCAN_PRIORITY_HIGH, + expectedLocal: TaskHighPrior, + }, + { + name: "recent bootstrap is only high remotely", + priorRemote: cdcpb.ScanPriority_SCAN_PRIORITY_LOW, + inheritedLocal: TaskLowPrior, + startTs: oracle.GoTimeToTS(currentTime.Add(-time.Minute)), + expectedRemote: cdcpb.ScanPriority_SCAN_PRIORITY_HIGH, + expectedLocal: TaskLowPrior, + }, + { + name: "old bootstrap stays low", + priorRemote: cdcpb.ScanPriority_SCAN_PRIORITY_LOW, + inheritedLocal: TaskLowPrior, + startTs: oracle.GoTimeToTS(currentTime.Add(-time.Hour)), + expectedRemote: cdcpb.ScanPriority_SCAN_PRIORITY_LOW, + expectedLocal: TaskLowPrior, + }, + } { + t.Run(tc.name, func(t *testing.T) { + pdClock := pdutil.NewClock4Test() + pdClock.(*pdutil.Clock4Test).SetTS(currentTs) + client := &subscriptionClient{ + pdClock: pdClock, + regionTaskQueue: priorityqueue.New[PriorityTask](), + } + rawSpan := heartbeatpb.TableSpan{ + TableID: 1, + StartKey: []byte("a"), + EndKey: []byte("z"), + } + span := &subscribedSpan{ + subID: SubscriptionID(1), + span: rawSpan, + startTs: tc.startTs, + rangeLock: regionlock.NewRangeLock(1, rawSpan.StartKey, rawSpan.EndKey, tc.startTs), + priorityPolicy: newScanPriorityPolicy(pdClock, 30*time.Minute), + } + region := newRegionInfo(tikv.NewRegionVerID(1, 1, 1), rawSpan, nil, span, false) + region.scanPriority = tc.priorRemote + + client.scheduleRegionRequest(context.Background(), region, tc.inheritedLocal) + task := popRegionPriorityTask(t, client.regionTaskQueue) + require.Equal(t, tc.expectedLocal, task.taskType) + require.Equal(t, tc.expectedRemote, task.GetRegionInfo().scanPriority) + }) + } +} + func popRegionPriorityTask( t *testing.T, queue *priorityqueue.PriorityQueue[PriorityTask], diff --git a/logservice/logpuller/subscription_client.go b/logservice/logpuller/subscription_client.go index fc3a0725c3..4962540045 100644 --- a/logservice/logpuller/subscription_client.go +++ b/logservice/logpuller/subscription_client.go @@ -19,6 +19,7 @@ import ( "sync/atomic" "time" + "github.com/pingcap/kvproto/pkg/cdcpb" "github.com/pingcap/kvproto/pkg/metapb" "github.com/pingcap/log" "github.com/pingcap/ticdc/heartbeatpb" @@ -652,13 +653,17 @@ func (s *subscriptionClient) scheduleRegionRequest( case regionlock.LockRangeStatusSuccess: region.lockedRangeState = lockRangeResult.LockedRangeState currentTs := s.pdClock.CurrentTS() - priority := region.subscribedSpan.priorityPolicy.resolve( - inheritedPriority, + remoteBase := inheritedPriority + if region.scanPriority == cdcpb.ScanPriority_SCAN_PRIORITY_HIGH { + remoteBase = TaskHighPrior + } + remotePriority := region.subscribedSpan.priorityPolicy.resolve( + remoteBase, region.resolvedTs(), oracle.GetTimeFromTS(currentTs), ) - region.scanPriority = priority.scanPriority() - s.regionTaskQueue.Push(NewRegionPriorityTask(priority, region, currentTs)) + region.scanPriority = remotePriority.scanPriority() + s.regionTaskQueue.Push(NewRegionPriorityTask(inheritedPriority, region, currentTs)) if log.GetLevel() <= zapcore.DebugLevel { log.Debug("cdc region scan task enqueued", zap.Uint64("subscriptionID", uint64(region.subscribedSpan.subID)), @@ -667,7 +672,7 @@ func (s *subscriptionClient) scheduleRegionRequest( zap.Uint64("regionID", region.verID.GetID()), zap.Uint64("regionEpochVersion", region.verID.GetVer()), zap.Uint64("regionEpochConfVer", region.verID.GetConfVer()), - zap.String("priority", priority.String()), + zap.String("priority", inheritedPriority.String()), zap.String("scanPriority", region.scanPriority.String()), zap.String("span", common.FormatTableSpan(®ion.span))) } diff --git a/logservice/logpuller/subscription_client_test.go b/logservice/logpuller/subscription_client_test.go index 18ce5cc75f..0298c633dc 100644 --- a/logservice/logpuller/subscription_client_test.go +++ b/logservice/logpuller/subscription_client_test.go @@ -402,54 +402,62 @@ func TestOnRegionFailQueuesCanceledErrorCache(t *testing.T) { func TestRegionRetryScanPriority(t *testing.T) { for _, tc := range []struct { - name string - priority cdcpb.ScanPriority - cdcErr *cdcpb.Error - everCaughtUp bool - expected TaskType + name string + priority cdcpb.ScanPriority + cdcErr *cdcpb.Error + everCaughtUp bool + expectedLocal TaskType + expectedRemote cdcpb.ScanPriority }{ { - name: "server is busy high", - priority: cdcpb.ScanPriority_SCAN_PRIORITY_HIGH, - cdcErr: &cdcpb.Error{ServerIsBusy: &errorpb.ServerIsBusy{}}, - expected: TaskHighPrior, + name: "server is busy high", + priority: cdcpb.ScanPriority_SCAN_PRIORITY_HIGH, + cdcErr: &cdcpb.Error{ServerIsBusy: &errorpb.ServerIsBusy{}}, + expectedLocal: TaskLowPrior, + expectedRemote: cdcpb.ScanPriority_SCAN_PRIORITY_HIGH, }, { - name: "server is busy low", - priority: cdcpb.ScanPriority_SCAN_PRIORITY_LOW, - cdcErr: &cdcpb.Error{ServerIsBusy: &errorpb.ServerIsBusy{}}, - expected: TaskLowPrior, + name: "server is busy low", + priority: cdcpb.ScanPriority_SCAN_PRIORITY_LOW, + cdcErr: &cdcpb.Error{ServerIsBusy: &errorpb.ServerIsBusy{}}, + expectedLocal: TaskLowPrior, + expectedRemote: cdcpb.ScanPriority_SCAN_PRIORITY_LOW, }, { - name: "server is busy low after catch up", - priority: cdcpb.ScanPriority_SCAN_PRIORITY_LOW, - cdcErr: &cdcpb.Error{ServerIsBusy: &errorpb.ServerIsBusy{}}, - everCaughtUp: true, - expected: TaskHighPrior, + name: "server is busy low after catch up", + priority: cdcpb.ScanPriority_SCAN_PRIORITY_LOW, + cdcErr: &cdcpb.Error{ServerIsBusy: &errorpb.ServerIsBusy{}}, + everCaughtUp: true, + expectedLocal: TaskLowPrior, + expectedRemote: cdcpb.ScanPriority_SCAN_PRIORITY_HIGH, }, { - name: "congested high", - priority: cdcpb.ScanPriority_SCAN_PRIORITY_HIGH, - cdcErr: &cdcpb.Error{Congested: &cdcpb.Congested{}}, - expected: TaskHighPrior, + name: "congested high", + priority: cdcpb.ScanPriority_SCAN_PRIORITY_HIGH, + cdcErr: &cdcpb.Error{Congested: &cdcpb.Congested{}}, + expectedLocal: TaskLowPrior, + expectedRemote: cdcpb.ScanPriority_SCAN_PRIORITY_HIGH, }, { - name: "congested low", - priority: cdcpb.ScanPriority_SCAN_PRIORITY_LOW, - cdcErr: &cdcpb.Error{Congested: &cdcpb.Congested{}}, - expected: TaskLowPrior, + name: "congested low", + priority: cdcpb.ScanPriority_SCAN_PRIORITY_LOW, + cdcErr: &cdcpb.Error{Congested: &cdcpb.Congested{}}, + expectedLocal: TaskLowPrior, + expectedRemote: cdcpb.ScanPriority_SCAN_PRIORITY_LOW, }, { - name: "unknown retry high", - priority: cdcpb.ScanPriority_SCAN_PRIORITY_HIGH, - cdcErr: &cdcpb.Error{}, - expected: TaskHighPrior, + name: "unknown retry high", + priority: cdcpb.ScanPriority_SCAN_PRIORITY_HIGH, + cdcErr: &cdcpb.Error{}, + expectedLocal: TaskHighPrior, + expectedRemote: cdcpb.ScanPriority_SCAN_PRIORITY_HIGH, }, { - name: "unknown retry low", - priority: cdcpb.ScanPriority_SCAN_PRIORITY_LOW, - cdcErr: &cdcpb.Error{}, - expected: TaskLowPrior, + name: "unknown retry low", + priority: cdcpb.ScanPriority_SCAN_PRIORITY_LOW, + cdcErr: &cdcpb.Error{}, + expectedLocal: TaskLowPrior, + expectedRemote: cdcpb.ScanPriority_SCAN_PRIORITY_LOW, }, } { t.Run(tc.name, func(t *testing.T) { @@ -471,8 +479,8 @@ func TestRegionRetryScanPriority(t *testing.T) { defer cancel() task, err := client.regionTaskQueue.Pop(ctx) require.NoError(t, err) - require.Equal(t, tc.expected, task.(*regionPriorityTask).taskType) - require.Equal(t, tc.expected.scanPriority(), task.GetRegionInfo().scanPriority) + require.Equal(t, tc.expectedLocal, task.(*regionPriorityTask).taskType) + require.Equal(t, tc.expectedRemote, task.GetRegionInfo().scanPriority) }) } }