diff --git a/logservice/logpuller/mock_upstream.go b/logservice/logpuller/mock_upstream_test.go similarity index 100% rename from logservice/logpuller/mock_upstream.go rename to logservice/logpuller/mock_upstream_test.go diff --git a/logservice/logpuller/priority_task.go b/logservice/logpuller/priority_task.go index c0ec14faa1..eb241dec0d 100644 --- a/logservice/logpuller/priority_task.go +++ b/logservice/logpuller/priority_task.go @@ -38,8 +38,7 @@ type regionPriorityTask struct { priority regionTaskPriority } -// NewRegionPriorityTask creates a new priority task for region -func NewRegionPriorityTask(regionInfo regionInfo, currentTs, sequence uint64) *regionPriorityTask { +func newRegionPriorityTask(regionInfo regionInfo, currentTs, sequence uint64) *regionPriorityTask { task := ®ionPriorityTask{ sequence: sequence, heapIndex: 0, // 0 means not in heap @@ -61,11 +60,6 @@ func (pt *regionPriorityTask) updateRegion(regionInfo regionInfo, currentTs uint pt.priority = priority } -// GetRegionInfo returns the underlying regionInfo -func (pt *regionPriorityTask) GetRegionInfo() regionInfo { - return pt.regionInfo -} - func (pt *regionPriorityTask) canUseMaxWindow() bool { return pt.priority != normalRegionPriority } diff --git a/logservice/logpuller/priority_task_test.go b/logservice/logpuller/priority_task_test.go index 4f765ce88c..6966779f11 100644 --- a/logservice/logpuller/priority_task_test.go +++ b/logservice/logpuller/priority_task_test.go @@ -48,15 +48,15 @@ func TestRegionPriorityTaskQueueOrder(t *testing.T) { currentTime := time.Now() currentTs := oracle.GoTimeToTS(currentTime) - normalTask := NewRegionPriorityTask( + normalTask := newRegionPriorityTask( newPriorityTestRegion(1, oracle.GoTimeToTS(currentTime.Add(-time.Hour)), false), currentTs, 3, ) - lowLagTask := NewRegionPriorityTask( + lowLagTask := newRegionPriorityTask( newPriorityTestRegion(2, oracle.GoTimeToTS(currentTime.Add(-10*time.Minute)), false), currentTs, 2, ) - initializedTask := NewRegionPriorityTask( + initializedTask := newRegionPriorityTask( newPriorityTestRegion(3, oracle.GoTimeToTS(currentTime.Add(-time.Hour)), true), currentTs, 1, ) @@ -78,8 +78,8 @@ func TestRegionPriorityTaskFIFOWithinPriority(t *testing.T) { currentTs := oracle.GoTimeToTS(currentTime) checkpointTs := oracle.GoTimeToTS(currentTime.Add(-time.Hour)) - first := NewRegionPriorityTask(newPriorityTestRegion(1, checkpointTs, false), currentTs, 1) - second := NewRegionPriorityTask(newPriorityTestRegion(2, checkpointTs, false), currentTs, 2) + first := newRegionPriorityTask(newPriorityTestRegion(1, checkpointTs, false), currentTs, 1) + second := newRegionPriorityTask(newPriorityTestRegion(2, checkpointTs, false), currentTs, 2) require.True(t, queue.Push(second)) require.True(t, queue.Push(first)) @@ -96,17 +96,17 @@ func TestRegionPriorityTaskLowLagBoundary(t *testing.T) { currentTime := time.Now() currentTs := oracle.GoTimeToTS(currentTime) - belowThreshold := NewRegionPriorityTask(newPriorityTestRegion( + belowThreshold := newRegionPriorityTask(newPriorityTestRegion( 1, oracle.GoTimeToTS(currentTime.Add(-lowLagRegionThreshold+time.Millisecond)), false, ), currentTs, 1) - atThreshold := NewRegionPriorityTask(newPriorityTestRegion( + atThreshold := newRegionPriorityTask(newPriorityTestRegion( 2, oracle.GoTimeToTS(currentTime.Add(-lowLagRegionThreshold)), false, ), currentTs, 2) - futureCheckpoint := NewRegionPriorityTask(newPriorityTestRegion( + futureCheckpoint := newRegionPriorityTask(newPriorityTestRegion( 3, oracle.GoTimeToTS(currentTime.Add(time.Second)), false, @@ -121,7 +121,7 @@ func TestRegionPriorityTaskRefreshesPriorityBetweenStages(t *testing.T) { checkpointTime := time.Now() checkpointTs := oracle.GoTimeToTS(checkpointTime) region := newPriorityTestRegion(1, checkpointTs, false) - task := NewRegionPriorityTask(region, oracle.GoTimeToTS(checkpointTime.Add(time.Minute)), 1) + task := newRegionPriorityTask(region, oracle.GoTimeToTS(checkpointTime.Add(time.Minute)), 1) require.Equal(t, lowLagRegionPriority, task.priority) task.updateRegion(region, oracle.GoTimeToTS(checkpointTime.Add(time.Hour))) diff --git a/logservice/logpuller/region_admission_controller_test.go b/logservice/logpuller/region_admission_controller_test.go index e8627aa98a..abc2c1a6b8 100644 --- a/logservice/logpuller/region_admission_controller_test.go +++ b/logservice/logpuller/region_admission_controller_test.go @@ -38,7 +38,6 @@ func createTestRegionInfo(subID SubscriptionID, regionID uint64) regionInfo { span, nil, &subscribedSpan{subID: subID, startTs: 100, span: span}, - false, ) } @@ -55,7 +54,7 @@ func submitRegionForAdmission( currentTs uint64, ) { t.Helper() - task := NewRegionPriorityTask(region, currentTs, region.verID.GetID()) + task := newRegionPriorityTask(region, currentTs, region.verID.GetID()) require.True(t, controller.submit(task)) } @@ -193,7 +192,7 @@ func TestRegionAdmissionControllerClose(t *testing.T) { controller := newRegionAdmissionController(1, 1) controller.close() region := prepareRegionForAdmission(createTestRegionInfo(1, 1), 1) - require.False(t, controller.submit(NewRegionPriorityTask(region, 1, 1))) + require.False(t, controller.submit(newRegionPriorityTask(region, 1, 1))) _, err := controller.pop(context.Background(), nil) require.ErrorIs(t, err, context.Canceled) diff --git a/logservice/logpuller/region_event_handler_test.go b/logservice/logpuller/region_event_handler_test.go index aef2ab63d4..5daf9497f6 100644 --- a/logservice/logpuller/region_event_handler_test.go +++ b/logservice/logpuller/region_event_handler_test.go @@ -85,7 +85,6 @@ func TestHandleEventEntryEventOutOfOrder(t *testing.T) { span, &tikv.RPCContext{}, subSpan, - false, ) region.lockedRangeState = ®ionlock.LockedRangeState{} state := newRegionFeedState(region, 1, worker, nil) diff --git a/logservice/logpuller/region_failure_handler.go b/logservice/logpuller/region_failure_handler.go index ee89e805df..7464afb940 100644 --- a/logservice/logpuller/region_failure_handler.go +++ b/logservice/logpuller/region_failure_handler.go @@ -41,17 +41,37 @@ var ( // regionFailureHandler handles failed regions and owns retry and reschedule decisions. type regionFailureHandler struct { - cache *errCache - client *subscriptionClient + cache *errCache + regionCache *tikv.RegionCache + + onTableDrained func(*subscribedSpan) + scheduleRegionRequest func(context.Context, regionInfo) + scheduleRangeRequest func(context.Context, rangeTask) } -func newRegionFailureHandler(client *subscriptionClient) *regionFailureHandler { +func newRegionFailureHandler( + regionCache *tikv.RegionCache, + onTableDrained func(*subscribedSpan), + scheduleRegionRequest func(context.Context, regionInfo), + scheduleRangeRequest func(context.Context, rangeTask), +) *regionFailureHandler { return ®ionFailureHandler{ - cache: newErrCache(), - client: client, + cache: newErrCache(), + regionCache: regionCache, + onTableDrained: onTableDrained, + scheduleRegionRequest: scheduleRegionRequest, + scheduleRangeRequest: scheduleRangeRequest, } } +func (r *regionFailureHandler) retryRange(ctx context.Context, errInfo regionErrorInfo) { + r.scheduleRangeRequest(ctx, rangeTask{ + span: errInfo.span, + subscribedSpan: errInfo.subscribedSpan, + wasInitialized: errInfo.wasInitialized, + }) +} + // Report admits a region failure into the recovery pipeline. It releases the // corresponding range lock before enqueueing the failure so new range tasks are // not blocked by stale region ownership. @@ -59,7 +79,7 @@ func (r *regionFailureHandler) Report(errInfo regionErrorInfo) { if errInfo.subscribedSpan.rangeLock.UnlockRange( errInfo.span.StartKey, errInfo.span.EndKey, errInfo.verID.GetID(), errInfo.verID.GetVer(), errInfo.resolvedTs()) { - r.client.onTableDrained(errInfo.subscribedSpan) + r.onTableDrained(errInfo.subscribedSpan) return } r.cache.add(errInfo) @@ -72,7 +92,6 @@ func (r *regionFailureHandler) Run(ctx context.Context) error { for { select { case <-ctx.Done(): - log.Info("subscription client handle errors and exit") return ctx.Err() case errInfo := <-r.cache.errCh: if err := r.handleError(ctx, errInfo); err != nil { @@ -100,28 +119,28 @@ func (r *regionFailureHandler) handleError(ctx context.Context, errInfo regionEr innerErr := eerr.err if notLeader := innerErr.GetNotLeader(); notLeader != nil { metricFeedNotLeaderCounter.Inc() - r.client.regionCache.UpdateLeader(errInfo.verID, notLeader.GetLeader(), errInfo.rpcCtx.AccessIdx) - r.client.scheduleRegionRequest(ctx, errInfo.regionInfo) + r.regionCache.UpdateLeader(errInfo.verID, notLeader.GetLeader(), errInfo.rpcCtx.AccessIdx) + r.scheduleRegionRequest(ctx, errInfo.regionInfo) return nil } if innerErr.GetEpochNotMatch() != nil { metricFeedEpochNotMatchCounter.Inc() - r.client.scheduleRangeRequest(ctx, errInfo.span, errInfo.subscribedSpan, errInfo.filterLoop, errInfo.wasInitialized) + r.retryRange(ctx, errInfo) return nil } if innerErr.GetRegionNotFound() != nil { metricFeedRegionNotFoundCounter.Inc() - r.client.scheduleRangeRequest(ctx, errInfo.span, errInfo.subscribedSpan, errInfo.filterLoop, errInfo.wasInitialized) + r.retryRange(ctx, errInfo) return nil } if innerErr.GetCongested() != nil { metricKvCongestedCounter.Inc() - r.client.scheduleRegionRequest(ctx, errInfo.regionInfo) + r.scheduleRegionRequest(ctx, errInfo.regionInfo) return nil } if innerErr.GetServerIsBusy() != nil { metricKvIsBusyCounter.Inc() - r.client.scheduleRegionRequest(ctx, errInfo.regionInfo) + r.scheduleRegionRequest(ctx, errInfo.regionInfo) return nil } if duplicated := innerErr.GetDuplicateRequest(); duplicated != nil { @@ -140,31 +159,31 @@ func (r *regionFailureHandler) handleError(ctx context.Context, errInfo regionEr zap.Uint64("subscriptionID", uint64(errInfo.subscribedSpan.subID)), zap.Stringer("error", innerErr)) metricFeedUnknownErrorCounter.Inc() - r.client.scheduleRegionRequest(ctx, errInfo.regionInfo) + r.scheduleRegionRequest(ctx, errInfo.regionInfo) return nil case *rpcCtxUnavailableErr: metricFeedRPCCtxUnavailable.Inc() - r.client.scheduleRangeRequest(ctx, errInfo.span, errInfo.subscribedSpan, errInfo.filterLoop, errInfo.wasInitialized) + r.retryRange(ctx, errInfo) return nil case *getStoreErr: metricGetStoreErr.Inc() bo := tikv.NewBackoffer(ctx, tikvRequestMaxBackoff) // cannot get the store the region belongs to, so we need to reload the region. - r.client.regionCache.OnSendFail(bo, errInfo.rpcCtx, true, err) - r.client.scheduleRangeRequest(ctx, errInfo.span, errInfo.subscribedSpan, errInfo.filterLoop, errInfo.wasInitialized) + r.regionCache.OnSendFail(bo, errInfo.rpcCtx, true, err) + r.retryRange(ctx, errInfo) return nil case *storeStreamErr: metricStoreSendRequestErr.Inc() bo := tikv.NewBackoffer(ctx, tikvRequestMaxBackoff) - r.client.regionCache.OnSendFail(bo, errInfo.rpcCtx, regionScheduleReload, err) - r.client.scheduleRegionRequest(ctx, errInfo.regionInfo) + r.regionCache.OnSendFail(bo, errInfo.rpcCtx, regionScheduleReload, err) + r.scheduleRegionRequest(ctx, errInfo.regionInfo) return nil case *requestCancelledErr: // the corresponding subscription has been unsubscribed, just ignore. return nil default: // TODO(qupeng): for some errors it's better to just deregister the region from TiKVs. - log.Warn("subscription client meets an internal error, fail the changefeed", + log.Warn("region failure cannot be recovered, fail the changefeed", zap.Uint64("subscriptionID", uint64(errInfo.subscribedSpan.subID)), zap.Error(err)) return err @@ -223,7 +242,6 @@ func (e *errCache) dispatchBatch(ctx context.Context, limit int) (int, error) { for _, errInfo := range batch { select { case <-ctx.Done(): - log.Info("subscription client dispatch err cache done") return 0, ctx.Err() case e.errCh <- errInfo: } diff --git a/logservice/logpuller/region_request_scheduler.go b/logservice/logpuller/region_request_scheduler.go new file mode 100644 index 0000000000..f75fbf7868 --- /dev/null +++ b/logservice/logpuller/region_request_scheduler.go @@ -0,0 +1,193 @@ +// 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, +// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. +// See the License for the specific language governing permissions and +// limitations under the License. + +package logpuller + +import ( + "context" + "sync" + "sync/atomic" + + "github.com/pingcap/log" + "github.com/pingcap/ticdc/pkg/config" + "github.com/pingcap/ticdc/pkg/errors" + "github.com/pingcap/ticdc/pkg/metrics" + "github.com/pingcap/ticdc/utils/priorityqueue" + kvclientv2 "github.com/tikv/client-go/v2/kv" + "github.com/tikv/client-go/v2/tikv" + "go.uber.org/zap" + "golang.org/x/sync/errgroup" +) + +const regionRequestWorkerPerStore = 8 + +// regionRequestScheduler routes locked region requests through the global +// priority queue to a worker connected to the region's TiKV store. Range +// resolution and retry policy remain owned by subscriptionClient and +// regionFailureHandler respectively. +type regionRequestScheduler struct { + upstream *upstreamHandle + eventSink *regionEventSink + failureHandler *regionFailureHandler + + // taskQueue orders all regions before they are assigned to a TiKV store. + taskQueue *priorityqueue.PriorityQueue[*regionPriorityTask] + // sequence is the FIFO tie-breaker for regions in the same priority class. + sequence atomic.Uint64 + // stores maps TiKV addresses to regionRequestStore. Stores are created only + // by Run, but are also read by metrics and deregistration goroutines. + stores sync.Map + + // workerCount is the configured number of request workers per store. + workerCount int + // workerWindow is each worker's share of the configured store window. + workerWindow int + // maxWindowMultiplier is passed to each worker's admission controller. + maxWindowMultiplier int +} + +func newRegionRequestScheduler( + upstream *upstreamHandle, + eventSink *regionEventSink, + failureHandler *regionFailureHandler, +) *regionRequestScheduler { + pullerConfig := config.GetGlobalServerConfig().Debug.Puller + workerCount := regionRequestWorkerPerStore + workerWindow := (pullerConfig.PendingRegionRequestQueueSize + workerCount - 1) / workerCount + return ®ionRequestScheduler{ + upstream: upstream, + eventSink: eventSink, + failureHandler: failureHandler, + taskQueue: priorityqueue.New[*regionPriorityTask](), + workerCount: workerCount, + workerWindow: workerWindow, + maxWindowMultiplier: pullerConfig.RegionRequestMaxWindowMultiplier, + } +} + +func (s *regionRequestScheduler) Submit(region regionInfo) { + s.taskQueue.Push(newRegionPriorityTask( + region, s.upstream.pdClock.CurrentTS(), s.sequence.Add(1))) +} + +func (s *regionRequestScheduler) Run(ctx context.Context, workerGroup *errgroup.Group) error { + defer s.closeStores() + for { + select { + case <-ctx.Done(): + return ctx.Err() + default: + } + + task, err := s.taskQueue.Pop(ctx) + if err != nil { + if errors.Is(err, priorityqueue.ErrClosed) { + return nil + } + return err + } + + region, err := s.attachRPCContext(ctx, task.regionInfo) + if err != nil { + s.failureHandler.Report(newRegionErrorInfo(region, err)) + continue + } + + store := s.getOrCreateStore(ctx, workerGroup, region.rpcCtx.Addr) + task.updateRegion(region, s.upstream.pdClock.CurrentTS()) + if !store.submit(task) { + return context.Canceled + } + } +} + +func (s *regionRequestScheduler) attachRPCContext( + ctx context.Context, + region regionInfo, +) (regionInfo, error) { + bo := tikv.NewBackoffer(ctx, tikvRequestMaxBackoff) + rpcCtx, err := s.upstream.regionCache.GetTiKVRPCContext( + bo, region.verID, kvclientv2.ReplicaReadLeader, 0) + if rpcCtx != nil { + region.rpcCtx = rpcCtx + return region, nil + } + if err != nil { + log.Debug("region request scheduler failed to get RPC context", + zap.Uint64("subscriptionID", uint64(region.subscribedSpan.subID)), + zap.Uint64("regionID", region.verID.GetID()), + zap.Error(err)) + } + return region, &rpcCtxUnavailableErr{verID: region.verID} +} + +func (s *regionRequestScheduler) getOrCreateStore( + ctx context.Context, + workerGroup *errgroup.Group, + storeAddr string, +) *regionRequestStore { + if value, ok := s.stores.Load(storeAddr); ok { + return value.(*regionRequestStore) + } + + store := newRegionRequestStore( + s.upstream, + s.eventSink, + s.failureHandler, + storeAddr, + s.workerCount, + s.workerWindow, + s.maxWindowMultiplier, + ) + // The scheduler run loop is the only writer. Publish the store after its + // immutable worker list is complete, then start its workers. + s.stores.Store(storeAddr, store) + store.startWorkers(ctx, workerGroup) + return store +} + +func (s *regionRequestScheduler) BroadcastDeregister( + subID SubscriptionID, + filterLoop bool, +) { + s.stores.Range(func(_, value any) bool { + value.(*regionRequestStore).broadcastDeregister(subID, filterLoop) + return true + }) +} + +func (s *regionRequestScheduler) inflightCount() int { + count := 0 + s.stores.Range(func(_, value any) bool { + count += value.(*regionRequestStore).inflightCount() + return true + }) + return count +} + +func (s *regionRequestScheduler) UpdateMetrics() { + metrics.SubscriptionClientRequestedRegionCount.WithLabelValues("inflight"). + Set(float64(s.inflightCount())) +} + +func (s *regionRequestScheduler) Close() { + s.taskQueue.Close() +} + +func (s *regionRequestScheduler) closeStores() { + s.stores.Range(func(_, value any) bool { + value.(*regionRequestStore).close() + return true + }) +} diff --git a/logservice/logpuller/region_request_store.go b/logservice/logpuller/region_request_store.go new file mode 100644 index 0000000000..9b9a5af025 --- /dev/null +++ b/logservice/logpuller/region_request_store.go @@ -0,0 +1,80 @@ +// 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, +// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. +// See the License for the specific language governing permissions and +// limitations under the License. + +package logpuller + +import ( + "context" + "sync/atomic" + + "golang.org/x/sync/errgroup" +) + +// regionRequestStore owns the region request workers connected to one TiKV +// store. The worker slice is complete before the store is published and is +// immutable afterwards, so task submission only needs an atomic round-robin counter. +type regionRequestStore struct { + workers []*regionRequestWorker + nextWorker atomic.Uint64 +} + +func newRegionRequestStore( + upstream *upstreamHandle, + eventSink *regionEventSink, + failureHandler *regionFailureHandler, + storeAddr string, + workerCount int, + workerWindow int, + maxWindowMultiplier int, +) *regionRequestStore { + store := ®ionRequestStore{ + workers: make([]*regionRequestWorker, 0, workerCount), + } + for i := 0; i < workerCount; i++ { + store.workers = append(store.workers, newRegionRequestWorker( + upstream, eventSink, failureHandler, storeAddr, workerWindow, maxWindowMultiplier)) + } + return store +} + +func (s *regionRequestStore) startWorkers(ctx context.Context, workerGroup *errgroup.Group) { + for _, worker := range s.workers { + workerGroup.Go(func() error { return worker.Run(ctx) }) + } +} + +func (s *regionRequestStore) submit(task *regionPriorityTask) bool { + index := (s.nextWorker.Add(1) - 1) % uint64(len(s.workers)) + return s.workers[index].admission.submit(task) +} + +func (s *regionRequestStore) broadcastDeregister(subID SubscriptionID, filterLoop bool) { + for _, worker := range s.workers { + worker.controlQueue.push(deregisterRequest{subID: subID, filterLoop: filterLoop}) + } +} + +func (s *regionRequestStore) close() { + for _, worker := range s.workers { + worker.admission.close() + } +} + +func (s *regionRequestStore) inflightCount() int { + count := 0 + for _, worker := range s.workers { + count += worker.admission.stats().inflight + } + return count +} diff --git a/logservice/logpuller/region_request_worker.go b/logservice/logpuller/region_request_worker.go index 0e6a5507fb..e61303a644 100644 --- a/logservice/logpuller/region_request_worker.go +++ b/logservice/logpuller/region_request_worker.go @@ -25,6 +25,7 @@ import ( "github.com/pingcap/kvproto/pkg/kvrpcpb" "github.com/pingcap/log" cerror "github.com/pingcap/ticdc/pkg/errors" + "github.com/pingcap/ticdc/pkg/metrics" "github.com/pingcap/ticdc/pkg/util" "github.com/pingcap/ticdc/pkg/version" "github.com/pingcap/ticdc/utils/notifyqueue" @@ -38,6 +39,8 @@ const storeReconnectBackoff = time.Second // To generate a workerID in `newRegionRequestWorker`. var workerIDGen atomic.Uint64 +var metricBatchResolvedSize = metrics.BatchResolvedEventSize.WithLabelValues("event-store") + type deregisterRequest struct { subID SubscriptionID filterLoop bool @@ -89,8 +92,10 @@ func (q *controlQueue) ready() <-chan struct{} { type regionRequestWorker struct { workerID uint64 - client *subscriptionClient - store *requestedStore + upstream *upstreamHandle + eventSink *regionEventSink + failureHandler *regionFailureHandler + storeAddr string admission *regionAdmissionController controlQueue *controlQueue @@ -98,19 +103,23 @@ type regionRequestWorker struct { } func newRegionRequestWorker( - client *subscriptionClient, - store *requestedStore, + upstream *upstreamHandle, + eventSink *regionEventSink, + failureHandler *regionFailureHandler, + storeAddr string, currentWindow int, maxWindowMultiplier int, ) *regionRequestWorker { workerID := workerIDGen.Add(1) return ®ionRequestWorker{ - workerID: workerID, - client: client, - store: store, - admission: newRegionAdmissionController(currentWindow, maxWindowMultiplier), - controlQueue: newControlQueue(), - tracker: newRegionTracker(), + workerID: workerID, + upstream: upstream, + eventSink: eventSink, + failureHandler: failureHandler, + storeAddr: storeAddr, + admission: newRegionAdmissionController(currentWindow, maxWindowMultiplier), + controlQueue: newControlQueue(), + tracker: newRegionTracker(), } } @@ -137,7 +146,7 @@ func (s *regionRequestWorker) Run(ctx context.Context) error { // firstReq still owns its lease only if the stream failed before Send. s.failStreamRegions(regionErr) if firstReq.abort() { - s.client.onRegionFail(newRegionErrorInfo(firstReq.regionInfo, regionErr)) + s.failureHandler.Report(newRegionErrorInfo(firstReq.regionInfo, regionErr)) } s.failPendingRegions(regionErr) @@ -150,10 +159,8 @@ func (s *regionRequestWorker) Run(ctx context.Context) error { // failStreamRegions transfers every request sent by a failed stream to the // recovery pipeline. func (s *regionRequestWorker) failStreamRegions(err error) { - for _, states := range s.tracker.Drain() { - for _, state := range states { - s.notifyRegionError(state, err) - } + for _, state := range s.tracker.Drain() { + s.notifyRegionError(state, err) } // The failed stream no longer owns remote registrations. s.controlQueue.drain() @@ -163,13 +170,13 @@ func (s *regionRequestWorker) failStreamRegions(err error) { // to the recovery pipeline, so they can be resolved and routed again. func (s *regionRequestWorker) failPendingRegions(err error) { for _, task := range s.admission.drain() { - s.client.onRegionFail(newRegionErrorInfo(task.regionInfo, err)) + s.failureHandler.Report(newRegionErrorInfo(task.regionInfo, err)) } } func (s *regionRequestWorker) notifyRegionError(state *regionFeedState, err error) { state.markStopped(err) - s.client.eventSink.Push( + s.eventSink.Push( SubscriptionID(state.requestID), regionEvent{states: []*regionFeedState{state}}, ) @@ -189,7 +196,7 @@ func (s *regionRequestWorker) waitForRegionRequest(ctx context.Context) (*region } func (s *regionRequestWorker) checkStoreVersion(ctx context.Context) error { - err := version.CheckStoreVersion(ctx, s.client.pd) + err := version.CheckStoreVersion(ctx, s.upstream.pd) if err == nil { return nil } @@ -198,7 +205,7 @@ func (s *regionRequestWorker) checkStoreVersion(ctx context.Context) error { } log.Error("event feed check store version fails", zap.Uint64("workerID", s.workerID), - zap.String("addr", s.store.storeAddr), + zap.String("addr", s.storeAddr), zap.Error(err)) if cerror.Is(err, cerror.ErrGetAllStoresFailed) { return &getStoreErr{} @@ -209,20 +216,20 @@ func (s *regionRequestWorker) checkStoreVersion(ctx context.Context) error { func (s *regionRequestWorker) runStream(ctx context.Context, firstReq *regionReq) (err error) { log.Info("region request worker going to create grpc stream", zap.Uint64("workerID", s.workerID), - zap.String("addr", s.store.storeAddr)) + zap.String("addr", s.storeAddr)) defer func() { log.Info("region request worker exits", zap.Uint64("workerID", s.workerID), - zap.String("addr", s.store.storeAddr), + zap.String("addr", s.storeAddr), zap.Error(err)) }() g, gctx := errgroup.WithContext(ctx) - conn, err := Connect(gctx, s.client.credential, s.store.storeAddr) + conn, err := Connect(gctx, s.upstream.credential, s.storeAddr) if err != nil { log.Warn("region request worker create grpc stream failed", zap.Uint64("workerID", s.workerID), - zap.String("addr", s.store.storeAddr), + zap.String("addr", s.storeAddr), zap.Error(err)) if conn != nil && conn.Conn != nil { _ = conn.Conn.Close() @@ -270,7 +277,7 @@ func (s *regionRequestWorker) receiveAndDispatchChangeEvents(conn *ConnAndClient if err != nil { log.Info("region request worker receive from grpc stream failed", zap.Uint64("workerID", s.workerID), - zap.String("addr", s.store.storeAddr), + zap.String("addr", s.storeAddr), zap.String("code", grpcstatus.Code(err).String()), zap.Error(err)) return normalizeStreamError(err) @@ -318,7 +325,7 @@ func (s *regionRequestWorker) dispatchRegionChangeEvents(events []*cdcpb.Event) default: log.Panic("unknown event type", zap.Any("event", event)) } - s.client.eventSink.Push(subscriptionID, regionEvent) + s.eventSink.Push(subscriptionID, regionEvent) continue } @@ -340,7 +347,7 @@ func (s *regionRequestWorker) dispatchRegionChangeEvents(events []*cdcpb.Event) func (s *regionRequestWorker) dispatchResolvedTsEvent(resolvedTsEvent *cdcpb.ResolvedTs) { subscriptionID := SubscriptionID(resolvedTsEvent.RequestId) metricsResolvedTsCount.Add(float64(len(resolvedTsEvent.Regions))) - s.client.metrics.batchResolvedSize.Observe(float64(len(resolvedTsEvent.Regions))) + metricBatchResolvedSize.Observe(float64(len(resolvedTsEvent.Regions))) if resolvedTsEvent.Ts == 0 { log.Warn("region request worker receives a resolved ts event with zero value, ignore it", zap.Uint64("workerID", s.workerID), @@ -356,7 +363,7 @@ func (s *regionRequestWorker) dispatchResolvedTsEvent(resolvedTsEvent *cdcpb.Res if len(resolvedStates) == 0 { return } - s.client.eventSink.Push(subscriptionID, regionEvent{ + s.eventSink.Push(subscriptionID, regionEvent{ resolvedTs: resolvedTsEvent.Ts, states: resolvedStates, }) @@ -392,7 +399,7 @@ func (s *regionRequestWorker) sendChangeDataRequest( zap.Uint64("workerID", s.workerID), zap.Uint64("subscriptionID", req.RequestId), zap.Uint64("regionID", req.RegionId), - zap.String("addr", s.store.storeAddr), + zap.String("addr", s.storeAddr), zap.Error(err)) return normalizeStreamError(err) } @@ -404,7 +411,7 @@ func (s *regionRequestWorker) sendDeregisterRequest( req deregisterRequest, ) error { changeDataReq := &cdcpb.ChangeDataRequest{ - Header: &cdcpb.Header{ClusterId: s.client.clusterID, TicdcVersion: version.ReleaseSemver()}, + Header: &cdcpb.Header{ClusterId: s.upstream.clusterID, TicdcVersion: version.ReleaseSemver()}, RequestId: uint64(req.subID), Request: &cdcpb.ChangeDataRequest_Deregister_{ Deregister: &cdcpb.ChangeDataRequest_Deregister{}, @@ -442,12 +449,12 @@ func (s *regionRequestWorker) sendRegionRequest(conn *ConnAndClient, req *region zap.Uint64("workerID", s.workerID), zap.Uint64("subscriptionID", uint64(subID)), zap.Uint64("regionID", region.verID.GetID()), - zap.String("addr", s.store.storeAddr), + zap.String("addr", s.storeAddr), zap.Bool("bdrMode", region.filterLoop)) if region.subscribedSpan.stopped.Load() { req.abort() - s.client.onRegionFail(newRegionErrorInfo(region, &storeStreamErr{})) + s.failureHandler.Report(newRegionErrorInfo(region, &storeStreamErr{})) return nil } @@ -504,7 +511,7 @@ func (s *regionRequestWorker) processRegionSendTask( func (s *regionRequestWorker) createRegionRequest(region regionInfo) *cdcpb.ChangeDataRequest { return &cdcpb.ChangeDataRequest{ - Header: &cdcpb.Header{ClusterId: s.client.clusterID, TicdcVersion: version.ReleaseSemver()}, + Header: &cdcpb.Header{ClusterId: s.upstream.clusterID, TicdcVersion: version.ReleaseSemver()}, RegionId: region.verID.GetID(), RequestId: uint64(region.subscribedSpan.subID), RegionEpoch: region.rpcCtx.Meta.RegionEpoch, diff --git a/logservice/logpuller/region_request_worker_test.go b/logservice/logpuller/region_request_worker_test.go index 4259904e9c..c6e3f8d845 100644 --- a/logservice/logpuller/region_request_worker_test.go +++ b/logservice/logpuller/region_request_worker_test.go @@ -25,7 +25,6 @@ import ( "github.com/pingcap/ticdc/heartbeatpb" "github.com/pingcap/ticdc/logservice/logpuller/regionlock" "github.com/pingcap/ticdc/utils/dynstream" - "github.com/prometheus/client_golang/prometheus" "github.com/stretchr/testify/require" "github.com/tikv/client-go/v2/oracle" "github.com/tikv/client-go/v2/tikv" @@ -83,8 +82,8 @@ func TestRegionRequestWorkerIgnoresDuplicateActiveRegion(t *testing.T) { admission := newRegionAdmissionController(10, 1) worker := ®ionRequestWorker{ admission: admission, - store: &requestedStore{storeAddr: "store-1"}, - client: &subscriptionClient{}, + storeAddr: "store-1", + upstream: &upstreamHandle{}, tracker: newRegionTracker(), } region := prepareRegionForSendTest(createTestRegionInfo(1, 1)) @@ -174,13 +173,8 @@ func (m *mockRegionEventDynamicStream) GetMetrics() dynstream.Metrics[int, Subsc func newDispatchResolvedTsTestWorker(regionCount int) (*regionRequestWorker, *mockRegionEventDynamicStream, *cdcpb.ResolvedTs) { ds := &mockRegionEventDynamicStream{} worker := ®ionRequestWorker{ - client: &subscriptionClient{ - metrics: sharedClientMetrics{ - batchResolvedSize: prometheus.ObserverFunc(func(float64) {}), - }, - eventSink: ®ionEventSink{ds: ds}, - }, - tracker: newRegionTracker(), + eventSink: ®ionEventSink{ds: ds}, + tracker: newRegionTracker(), } regions := make([]uint64, regionCount) for i := 0; i < regionCount; i++ { @@ -207,7 +201,7 @@ func dispatchResolvedTsEventLegacyForBenchmark(s *regionRequestWorker, resolvedT return } states := resolvedStates - s.client.pushRegionEventToDS(subscriptionID, regionEvent{ + s.eventSink.Push(subscriptionID, regionEvent{ resolvedTs: resolvedTsEvent.Ts, states: states, }) @@ -227,7 +221,7 @@ func dispatchResolvedTsEventLegacyForBenchmark(s *regionRequestWorker, resolvedT func benchmarkDispatchResolvedTsEvent(b *testing.B, regionCount int, useLegacy bool) { worker, _, event := newDispatchResolvedTsTestWorker(regionCount) ds := &countingRegionEventDynamicStream{} - worker.client.eventSink.ds = ds + worker.eventSink.ds = ds b.ReportAllocs() b.ResetTimer() for i := 0; i < b.N; i++ { @@ -348,10 +342,8 @@ func TestFailStreamRegionsReleasesSentAdmission(t *testing.T) { worker := ®ionRequestWorker{ admission: admission, controlQueue: newControlQueue(), - client: &subscriptionClient{ - eventSink: ®ionEventSink{ds: ds}, - }, - tracker: newRegionTracker(), + eventSink: ®ionEventSink{ds: ds}, + tracker: newRegionTracker(), } region := prepareRegionForSendTest(createTestRegionInfo(1, 1)) req := admitRegionRequest(t, admission, region) @@ -382,9 +374,8 @@ func TestFailPendingRegionsReschedulesWorkerBuffer(t *testing.T) { require.Equal(t, regionlock.LockRangeStatusSuccess, lock2.Status) admission := newRegionAdmissionController(1, 1) - client := &subscriptionClient{} - client.failureHandler = newRegionFailureHandler(client) - worker := ®ionRequestWorker{client: client, admission: admission} + failureHandler := ®ionFailureHandler{cache: newErrCache()} + worker := ®ionRequestWorker{failureHandler: failureHandler, admission: admission} regions := []regionInfo{ { verID: tikv.NewRegionVerID(1, 1, 1), @@ -402,13 +393,13 @@ func TestFailPendingRegionsReschedulesWorkerBuffer(t *testing.T) { }, } for i, region := range regions { - require.True(t, admission.submit(NewRegionPriorityTask(region, 1, uint64(i+1)))) + require.True(t, admission.submit(newRegionPriorityTask(region, 1, uint64(i+1)))) } worker.failPendingRegions(&storeStreamErr{}) require.Zero(t, admission.stats().pending) - require.Len(t, client.failureHandler.cache.cache, 2) + require.Len(t, failureHandler.cache.cache, 2) } func TestProcessRegionSendTaskSendFailureCleansSentRequest(t *testing.T) { @@ -416,8 +407,8 @@ func TestProcessRegionSendTaskSendFailureCleansSentRequest(t *testing.T) { worker := ®ionRequestWorker{ admission: admission, controlQueue: newControlQueue(), - store: &requestedStore{storeAddr: "store-1"}, - client: &subscriptionClient{}, + storeAddr: "store-1", + upstream: &upstreamHandle{}, tracker: newRegionTracker(), } @@ -447,8 +438,8 @@ func TestProcessRegionSendTaskDoesNotSendRemovedRequest(t *testing.T) { worker := ®ionRequestWorker{ admission: admission, controlQueue: newControlQueue(), - store: &requestedStore{storeAddr: "store-1"}, - client: &subscriptionClient{}, + storeAddr: "store-1", + upstream: &upstreamHandle{}, tracker: newRegionTracker(), } region := prepareRegionForSendTest(createTestRegionInfo(1, 1)) @@ -490,8 +481,8 @@ func TestProcessRegionSendTaskSendEOFIsRetriable(t *testing.T) { worker := ®ionRequestWorker{ admission: admission, controlQueue: newControlQueue(), - store: &requestedStore{storeAddr: "store-1"}, - client: &subscriptionClient{}, + storeAddr: "store-1", + upstream: &upstreamHandle{}, tracker: newRegionTracker(), } region := prepareRegionForSendTest(createTestRegionInfo(1, 1)) @@ -523,11 +514,10 @@ func TestProcessRegionSendTaskHandlesDeregisterFromControlQueue(t *testing.T) { worker := ®ionRequestWorker{ admission: newRegionAdmissionController(1, 1), controlQueue: newControlQueue(), - store: &requestedStore{storeAddr: "store-1"}, - client: &subscriptionClient{ - eventSink: ®ionEventSink{ds: ds}, - }, - tracker: newRegionTracker(), + storeAddr: "store-1", + upstream: &upstreamHandle{clusterID: 42}, + eventSink: ®ionEventSink{ds: ds}, + tracker: newRegionTracker(), } state := ®ionFeedState{worker: worker} require.True(t, worker.tracker.Add(1, 1, state)) @@ -544,6 +534,7 @@ func TestProcessRegionSendTaskHandlesDeregisterFromControlQueue(t *testing.T) { }() req := <-sendCh + require.Equal(t, uint64(42), req.Header.ClusterId) require.Equal(t, uint64(1), req.RequestId) require.True(t, req.FilterLoop) require.NotNil(t, req.GetDeregister()) @@ -571,9 +562,7 @@ func TestReceiveAndDispatchChangeEventsEOFIsRetriable(t *testing.T) { for _, tc := range testCases { t.Run(tc.name, func(t *testing.T) { - worker := ®ionRequestWorker{ - store: &requestedStore{storeAddr: "store-1"}, - } + worker := ®ionRequestWorker{storeAddr: "store-1"} conn := &ConnAndClient{ Client: &mockEventFeedV2Client{recvErr: tc.recvErr}, Conn: &grpc.ClientConn{}, diff --git a/logservice/logpuller/region_state.go b/logservice/logpuller/region_state.go index 0dc0d55bc0..06e3bdd5d4 100644 --- a/logservice/logpuller/region_state.go +++ b/logservice/logpuller/region_state.go @@ -57,14 +57,13 @@ func newRegionInfo( span heartbeatpb.TableSpan, rpcCtx *tikv.RPCContext, subscribedSpan *subscribedSpan, - filterLoop bool, ) regionInfo { return regionInfo{ verID: verID, span: span, rpcCtx: rpcCtx, subscribedSpan: subscribedSpan, - filterLoop: filterLoop, + filterLoop: subscribedSpan.filterLoop, } } diff --git a/logservice/logpuller/region_tracker.go b/logservice/logpuller/region_tracker.go index 866ee49f73..7066779eaf 100644 --- a/logservice/logpuller/region_tracker.go +++ b/logservice/logpuller/region_tracker.go @@ -105,16 +105,22 @@ func (t *regionTracker) TakeSubscription(subscriptionID SubscriptionID) []*regio return slices.Collect(maps.Values(states)) } -// Drain removes and returns all tracked states grouped by subscription. -func (t *regionTracker) Drain() map[SubscriptionID][]*regionFeedState { +// Drain removes and returns all tracked states. +func (t *regionTracker) Drain() []*regionFeedState { t.mu.Lock() statesBySubscription := t.statesBySubscription t.statesBySubscription = make(map[SubscriptionID]regionStatesByID) t.mu.Unlock() - drainedStates := make(map[SubscriptionID][]*regionFeedState, len(statesBySubscription)) - for subID, states := range statesBySubscription { - drainedStates[subID] = slices.Collect(maps.Values(states)) + stateCount := 0 + for _, states := range statesBySubscription { + stateCount += len(states) + } + drainedStates := make([]*regionFeedState, 0, stateCount) + for _, states := range statesBySubscription { + for _, state := range states { + drainedStates = append(drainedStates, state) + } } return drainedStates } diff --git a/logservice/logpuller/region_tracker_test.go b/logservice/logpuller/region_tracker_test.go index ad89844a3a..8a08ce0d94 100644 --- a/logservice/logpuller/region_tracker_test.go +++ b/logservice/logpuller/region_tracker_test.go @@ -37,9 +37,7 @@ func TestRegionTrackerOperations(t *testing.T) { require.Empty(t, tracker.TakeSubscription(1)) drained := tracker.Drain() - require.Equal(t, map[SubscriptionID][]*regionFeedState{ - 2: {state3}, - }, drained) + require.ElementsMatch(t, []*regionFeedState{state3}, drained) require.Nil(t, tracker.Get(2, 3)) require.Empty(t, tracker.Drain()) } diff --git a/logservice/logpuller/subscription_client.go b/logservice/logpuller/subscription_client.go index 2542936b6a..ea4cd33c31 100644 --- a/logservice/logpuller/subscription_client.go +++ b/logservice/logpuller/subscription_client.go @@ -15,7 +15,6 @@ package logpuller import ( "context" - "sync" "sync/atomic" "time" @@ -26,16 +25,11 @@ import ( "github.com/pingcap/ticdc/logservice/txnutil" "github.com/pingcap/ticdc/pkg/common" appcontext "github.com/pingcap/ticdc/pkg/common/context" - "github.com/pingcap/ticdc/pkg/config" - "github.com/pingcap/ticdc/pkg/errors" "github.com/pingcap/ticdc/pkg/metrics" "github.com/pingcap/ticdc/pkg/pdutil" "github.com/pingcap/ticdc/pkg/security" "github.com/pingcap/ticdc/pkg/spanz" "github.com/pingcap/ticdc/pkg/util" - "github.com/pingcap/ticdc/utils/priorityqueue" - "github.com/prometheus/client_golang/prometheus" - kvclientv2 "github.com/tikv/client-go/v2/kv" "github.com/tikv/client-go/v2/tikv" pd "github.com/tikv/pd/client" "go.uber.org/zap" @@ -86,17 +80,23 @@ type resolveLockTask struct { type rangeTask struct { span heartbeatpb.TableSpan subscribedSpan *subscribedSpan - filterLoop bool wasInitialized bool } -type SubscriptionClientConfig struct { - // The number of region request workers to send region task for every tikv store - RegionRequestWorkerPerStore uint +// upstreamHandle contains the stable TiKV and PD dependencies shared by the +// region request pipeline. +type upstreamHandle struct { + pd pd.Client + regionCache *tikv.RegionCache + pdClock pdutil.Clock + credential *security.Credential + clusterID uint64 } -type sharedClientMetrics struct { - batchResolvedSize prometheus.Observer +// initialize loads the cluster metadata needed by region request workers. It +// must run before the scheduler starts any workers. +func (u *upstreamHandle) initialize(ctx context.Context) { + u.clusterID = u.pd.GetClusterID(ctx) } // subscriptionClient is used to subscribe events of table ranges from TiKV. @@ -120,21 +120,11 @@ type SubscriptionClient interface { } type subscriptionClient struct { - ctx context.Context - cancel context.CancelFunc - config *SubscriptionClientConfig - metrics sharedClientMetrics - clusterID uint64 - - pd pd.Client - regionCache *tikv.RegionCache - pdClock pdutil.Clock - lockResolver txnutil.LockResolver - - stores sync.Map + ctx context.Context + cancel context.CancelFunc + upstream *upstreamHandle - // the credential to connect tikv - credential *security.Credential + lockResolver txnutil.LockResolver // failureHandler handles failed regions and owns reschedule/retry decisions. failureHandler *regionFailureHandler @@ -142,16 +132,12 @@ type subscriptionClient struct { eventSink *regionEventSink // spanRegistry tracks subscribed spans and owns span-level background tasks. spanRegistry *spanRegistry + // regionScheduler assigns locked region requests to per-store workers. + regionScheduler *regionRequestScheduler // rangeTaskCh is used to receive range tasks. // The tasks will be handled in `handleRangeTask` goroutine. rangeTaskCh chan rangeTask - // regionTaskQueue is used to receive region tasks with priority. - // The region will be handled in `handleRegions` goroutine. - regionTaskQueue *priorityqueue.PriorityQueue[*regionPriorityTask] - // regionTaskSequence provides a FIFO tie-breaker for tasks in the same - // priority class. - regionTaskSequence atomic.Uint64 // resolveLockTaskCh is used to receive resolve lock tasks. // The tasks will be handled in `handleResolveLockTasks` goroutine. resolveLockTaskCh chan resolveLockTask @@ -160,33 +146,37 @@ type subscriptionClient struct { // NewSubscriptionClient creates a client. func NewSubscriptionClient( - config *SubscriptionClientConfig, pd pd.Client, lockResolver txnutil.LockResolver, credential *security.Credential, ) SubscriptionClient { subClient := &subscriptionClient{ - config: config, - - stores: sync.Map{}, - pd: pd, - regionCache: appcontext.GetService[*tikv.RegionCache](appcontext.RegionCache), - pdClock: appcontext.GetService[pdutil.Clock](appcontext.DefaultPDClock), + upstream: &upstreamHandle{ + pd: pd, + regionCache: appcontext.GetService[*tikv.RegionCache](appcontext.RegionCache), + pdClock: appcontext.GetService[pdutil.Clock](appcontext.DefaultPDClock), + credential: credential, + }, lockResolver: lockResolver, - credential: credential, - rangeTaskCh: make(chan rangeTask, 1024), - regionTaskQueue: priorityqueue.New[*regionPriorityTask](), resolveLockTaskCh: make(chan resolveLockTask, 1024), resolveLockRateLimiter: newResolveLockRateLimiter(), } subClient.ctx, subClient.cancel = context.WithCancel(context.Background()) - subClient.failureHandler = newRegionFailureHandler(subClient) + subClient.failureHandler = newRegionFailureHandler( + subClient.upstream.regionCache, + subClient.onTableDrained, + subClient.scheduleRegionRequest, + subClient.scheduleRangeRequest, + ) subClient.eventSink = newRegionEventSink(subClient.ctx, subClient.failureHandler) - subClient.spanRegistry = newSpanRegistry(subClient.pd, subClient.pdClock) - - subClient.initMetrics() + subClient.spanRegistry = newSpanRegistry(subClient.upstream.pd, subClient.upstream.pdClock) + subClient.regionScheduler = newRegionRequestScheduler( + subClient.upstream, + subClient.eventSink, + subClient.failureHandler, + ) return subClient } @@ -199,11 +189,6 @@ func (s *subscriptionClient) AllocSubscriptionID() SubscriptionID { return SubscriptionID(subscriptionIDGen.Add(1)) } -func (s *subscriptionClient) initMetrics() { - // TODO: fix metrics - s.metrics.batchResolvedSize = metrics.BatchResolvedEventSize.WithLabelValues("event-store") -} - func (s *subscriptionClient) updateMetrics(ctx context.Context) error { ticker := time.NewTicker(10 * time.Second) defer ticker.Stop() @@ -212,14 +197,7 @@ func (s *subscriptionClient) updateMetrics(ctx context.Context) error { case <-ctx.Done(): return ctx.Err() case <-ticker.C: - pendingRegionReqCount := 0 - s.stores.Range(func(_, value any) bool { - store := value.(*requestedStore) - pendingRegionReqCount += store.inflightCount() - return true - }) - - metrics.SubscriptionClientRequestedRegionCount.WithLabelValues("pending").Set(float64(pendingRegionReqCount)) + s.regionScheduler.UpdateMetrics() s.eventSink.UpdateMetrics() s.spanRegistry.UpdateMetrics() } @@ -263,7 +241,7 @@ func (s *subscriptionClient) Subscribe( select { case <-s.ctx.Done(): log.Warn("subscribes span failed, the subscription client has closed") - case s.rangeTaskCh <- rangeTask{span: span, subscribedSpan: rt, filterLoop: rt.filterLoop}: + case s.rangeTaskCh <- rangeTask{span: span, subscribedSpan: rt}: log.Info("subscribes span done", zap.Uint64("subscriptionID", uint64(subID)), zap.Int64("tableID", span.TableID), zap.Uint64("startTs", startTs), zap.String("startKey", spanz.HexKey(span.StartKey)), zap.String("endKey", spanz.HexKey(span.EndKey))) @@ -286,27 +264,20 @@ func (s *subscriptionClient) Unsubscribe(subID SubscriptionID) { zap.Bool("exists", rt != nil)) } -func (s *subscriptionClient) pushRegionEventToDS(subID SubscriptionID, event regionEvent) { - s.eventSink.Push(subID, event) -} - func (s *subscriptionClient) Run(ctx context.Context) error { - // s.consume = consume - if s.pd == nil { - log.Warn("subscription client should be in test mode, skip run") - return nil - } - s.clusterID = s.pd.GetClusterID(ctx) + s.upstream.initialize(ctx) g, ctx := errgroup.WithContext(ctx) - g.Go(func() error { return s.updateMetrics(ctx) }) - g.Go(func() error { return s.eventSink.Run(ctx) }) + // The goroutines are listed by data flow; errgroup does not guarantee their + // actual startup order. g.Go(func() error { return s.handleRangeTasks(ctx) }) - g.Go(func() error { return s.handleRegions(ctx, g) }) + g.Go(func() error { return s.regionScheduler.Run(ctx, g) }) + g.Go(func() error { return s.eventSink.Run(ctx) }) g.Go(func() error { return s.failureHandler.Run(ctx) }) - g.Go(func() error { return s.handleResolveLockTasks(ctx) }) g.Go(func() error { return s.spanRegistry.Run(ctx) }) + g.Go(func() error { return s.handleResolveLockTasks(ctx) }) + g.Go(func() error { return s.updateMetrics(ctx) }) log.Info("subscription client starts") defer log.Info("subscription client exits") @@ -317,7 +288,7 @@ func (s *subscriptionClient) Run(ctx context.Context) error { func (s *subscriptionClient) Close(ctx context.Context) error { s.cancel() s.eventSink.Close() - s.regionTaskQueue.Close() + s.regionScheduler.Close() return nil } @@ -328,7 +299,7 @@ func (s *subscriptionClient) setTableStopped(rt *subscribedSpan) { // Set stopped to true so we can stop handling region events from the table, // then notify every existing worker to deregister the subscription. if rt.stopped.CompareAndSwap(false, true) { - s.broadcastDeregister(rt.subID, rt.filterLoop) + s.regionScheduler.BroadcastDeregister(rt.subID, rt.filterLoop) if rt.rangeLock.Stop() { s.onTableDrained(rt) } @@ -348,160 +319,6 @@ func (s *subscriptionClient) onTableDrained(rt *subscribedSpan) { s.spanRegistry.Remove(rt.subID) } -// Note: don't block the caller, otherwise there may be deadlock -func (s *subscriptionClient) onRegionFail(errInfo regionErrorInfo) { - s.failureHandler.Report(errInfo) -} - -// requestedStore represents a store that has been connected. -type requestedStore struct { - storeAddr string - nextWorker atomic.Uint64 - - requestWorkers struct { - sync.RWMutex - s []*regionRequestWorker - } -} - -func (s *requestedStore) submit(task *regionPriorityTask) bool { - s.requestWorkers.RLock() - defer s.requestWorkers.RUnlock() - - workerCount := len(s.requestWorkers.s) - if workerCount == 0 { - return false - } - index := (s.nextWorker.Add(1) - 1) % uint64(workerCount) - return s.requestWorkers.s[index].admission.submit(task) -} - -func (s *requestedStore) close() { - s.requestWorkers.RLock() - defer s.requestWorkers.RUnlock() - for _, worker := range s.requestWorkers.s { - worker.admission.close() - } -} - -func (s *requestedStore) inflightCount() int { - s.requestWorkers.RLock() - defer s.requestWorkers.RUnlock() - count := 0 - for _, worker := range s.requestWorkers.s { - count += worker.admission.stats().inflight - } - return count -} - -// handleRegions receives regionInfo from regionTaskQueue and attach rpcCtx to them, -// then send them to corresponding requestedStore. -func (s *subscriptionClient) handleRegions(ctx context.Context, eg *errgroup.Group) error { - cfg := config.GetGlobalServerConfig() - storeWindow := cfg.Debug.Puller.PendingRegionRequestQueueSize - maxWindowMultiplier := cfg.Debug.Puller.RegionRequestMaxWindowMultiplier - workerCount := int(s.config.RegionRequestWorkerPerStore) - if workerCount <= 0 { - workerCount = 1 - } - workerWindow := (storeWindow + workerCount - 1) / workerCount - getStore := func(storeAddr string) *requestedStore { - var rs *requestedStore - if v, ok := s.stores.Load(storeAddr); ok { - rs = v.(*requestedStore) - return rs - } - - rs = &requestedStore{storeAddr: storeAddr} - rs.requestWorkers.s = make([]*regionRequestWorker, 0, workerCount) - - rs.requestWorkers.Lock() - for i := 0; i < workerCount; i++ { - requestWorker := newRegionRequestWorker(s, rs, workerWindow, maxWindowMultiplier) - rs.requestWorkers.s = append(rs.requestWorkers.s, requestWorker) - } - rs.requestWorkers.Unlock() - - // Publish the store only after its immutable worker list is complete. - s.stores.Store(storeAddr, rs) - for _, requestWorker := range rs.requestWorkers.s { - eg.Go(func() error { return requestWorker.Run(ctx) }) - } - return rs - } - - defer func() { - s.stores.Range(func(_, value any) bool { - rs := value.(*requestedStore) - rs.close() - return true - }) - }() - - for { - select { - case <-ctx.Done(): - return ctx.Err() - default: - } - // Use blocking Pop to wait for tasks - regionTask, err := s.regionTaskQueue.Pop(ctx) - if err != nil { - if errors.Is(err, priorityqueue.ErrClosed) { - return nil - } - return err - } - - region := regionTask.GetRegionInfo() - region, ok := s.attachRPCContextForRegion(ctx, region) - // If attachRPCContextForRegion fails, the region will be re-scheduled. - if !ok { - continue - } - - store := getStore(region.rpcCtx.Addr) - regionTask.updateRegion(region, s.pdClock.CurrentTS()) - if !store.submit(regionTask) { - return context.Canceled - } - - log.Debug("subscription client will request a region", - zap.Uint64("subscriptionID", uint64(region.subscribedSpan.subID)), - zap.Uint64("regionID", region.verID.GetID()), - zap.String("addr", store.storeAddr)) - } -} - -func (s *subscriptionClient) broadcastDeregister(subID SubscriptionID, filterLoop bool) { - s.stores.Range(func(_ any, value any) bool { - rs := value.(*requestedStore) - rs.requestWorkers.RLock() - for _, worker := range rs.requestWorkers.s { - worker.controlQueue.push(deregisterRequest{subID: subID, filterLoop: filterLoop}) - } - rs.requestWorkers.RUnlock() - return true - }) -} - -func (s *subscriptionClient) attachRPCContextForRegion(ctx context.Context, region regionInfo) (regionInfo, bool) { - bo := tikv.NewBackoffer(ctx, tikvRequestMaxBackoff) - rpcCtx, err := s.regionCache.GetTiKVRPCContext(bo, region.verID, kvclientv2.ReplicaReadLeader, 0) - if rpcCtx != nil { - region.rpcCtx = rpcCtx - return region, true - } - if err != nil { - log.Debug("subscription client get rpc context fail", - zap.Uint64("subscriptionID", uint64(region.subscribedSpan.subID)), - zap.Uint64("regionID", region.verID.GetID()), - zap.Error(err)) - } - s.onRegionFail(newRegionErrorInfo(region, &rpcCtxUnavailableErr{verID: region.verID})) - return region, false -} - func (s *subscriptionClient) handleRangeTasks(ctx context.Context) error { g, ctx := errgroup.WithContext(ctx) // Limit the concurrent number of goroutines to convert range tasks to region tasks. @@ -512,7 +329,7 @@ func (s *subscriptionClient) handleRangeTasks(ctx context.Context) error { return ctx.Err() case task := <-s.rangeTaskCh: g.Go(func() error { - return s.divideSpanAndScheduleRegionRequests(ctx, task.span, task.subscribedSpan, task.filterLoop, task.wasInitialized) + return s.divideSpanAndScheduleRegionRequests(ctx, task) }) } } @@ -525,11 +342,11 @@ func (s *subscriptionClient) handleRangeTasks(ctx context.Context) error { // 3. Schedule a region request to subscribe the region. func (s *subscriptionClient) divideSpanAndScheduleRegionRequests( ctx context.Context, - span heartbeatpb.TableSpan, - subscribedSpan *subscribedSpan, - filterLoop bool, - wasInitialized bool, + task rangeTask, ) error { + span := task.span + subscribedSpan := task.subscribedSpan + // Limit the number of regions loaded at a time to make the load more stable. limit := 1024 nextSpan := span @@ -546,7 +363,8 @@ func (s *subscriptionClient) divideSpanAndScheduleRegionRequests( zap.Any("span", common.FormatTableSpan(&nextSpan))) backoff := tikv.NewBackoffer(ctx, tikvRequestMaxBackoff) - regions, err := s.regionCache.BatchLoadRegionsWithKeyRange(backoff, nextSpan.StartKey, nextSpan.EndKey, limit) + regions, err := s.upstream.regionCache.BatchLoadRegionsWithKeyRange( + backoff, nextSpan.StartKey, nextSpan.EndKey, limit) if err != nil { log.Warn("subscription client load regions failed", zap.Uint64("subscriptionID", uint64(subscribedSpan.subID)), @@ -589,8 +407,8 @@ func (s *subscriptionClient) divideSpanAndScheduleRegionRequests( } verID := tikv.NewRegionVerID(regionMeta.Id, regionMeta.RegionEpoch.ConfVer, regionMeta.RegionEpoch.Version) - regionInfo := newRegionInfo(verID, intersectSpan, nil, subscribedSpan, filterLoop) - regionInfo.wasInitialized = wasInitialized + regionInfo := newRegionInfo(verID, intersectSpan, nil, subscribedSpan) + regionInfo.wasInitialized = task.wasInitialized // Schedule a region request to subscribe the region. s.scheduleRegionRequest(ctx, regionInfo) @@ -605,8 +423,8 @@ func (s *subscriptionClient) divideSpanAndScheduleRegionRequests( } } -// scheduleRegionRequest locks the region's range and send the region to regionTaskQueue, -// which will be handled by handleRegions. +// scheduleRegionRequest locks the region's range before submitting it to the +// request scheduler. func (s *subscriptionClient) scheduleRegionRequest(ctx context.Context, region regionInfo) { if region.lockedRangeState != nil && region.lockedRangeState.Initialized.Load() { region.wasInitialized = true @@ -621,11 +439,14 @@ func (s *subscriptionClient) scheduleRegionRequest(ctx context.Context, region r switch lockRangeResult.Status { case regionlock.LockRangeStatusSuccess: region.lockedRangeState = lockRangeResult.LockedRangeState - s.regionTaskQueue.Push(NewRegionPriorityTask( - region, s.pdClock.CurrentTS(), s.regionTaskSequence.Add(1))) + s.regionScheduler.Submit(region) case regionlock.LockRangeStatusStale: for _, r := range lockRangeResult.RetryRanges { - s.scheduleRangeRequest(ctx, r, region.subscribedSpan, region.filterLoop, region.wasInitialized) + s.scheduleRangeRequest(ctx, rangeTask{ + span: r, + subscribedSpan: region.subscribedSpan, + wasInitialized: region.wasInitialized, + }) } default: return @@ -633,17 +454,12 @@ func (s *subscriptionClient) scheduleRegionRequest(ctx context.Context, region r } func (s *subscriptionClient) scheduleRangeRequest( - ctx context.Context, span heartbeatpb.TableSpan, - subscribedSpan *subscribedSpan, - filterLoop bool, - wasInitialized bool, + ctx context.Context, + task rangeTask, ) { select { case <-ctx.Done(): - case s.rangeTaskCh <- rangeTask{ - span: span, subscribedSpan: subscribedSpan, - filterLoop: filterLoop, wasInitialized: wasInitialized, - }: + case s.rangeTaskCh <- task: } } diff --git a/logservice/logpuller/subscription_client_test.go b/logservice/logpuller/subscription_client_test.go index c3ef992a34..474d3c6af5 100644 --- a/logservice/logpuller/subscription_client_test.go +++ b/logservice/logpuller/subscription_client_test.go @@ -66,8 +66,8 @@ func TestGenerateResolveLockTask(t *testing.T) { } consumeKVEvents := func(_ []common.RawKVEntry, _ func()) bool { return false } advanceResolvedTs := func(ts uint64) {} - client.pdClock = pdutil.NewClock4Test() - client.spanRegistry = newSpanRegistry(nil, client.pdClock) + client.upstream = &upstreamHandle{pdClock: pdutil.NewClock4Test()} + client.spanRegistry = newSpanRegistry(nil, client.upstream.pdClock) span := newSubscribedSpan( client.ctx, client.resolveLockRateLimiter, @@ -327,10 +327,11 @@ func TestStopTaskUsesSubscribedSpanFilterLoop(t *testing.T) { res := span.rangeLock.LockRange(context.Background(), rawSpan.StartKey, rawSpan.EndKey, 1, 1) require.Equal(t, regionlock.LockRangeStatusSuccess, res.Status) - worker := ®ionRequestWorker{controlQueue: newControlQueue()} - store := &requestedStore{storeAddr: "store-1"} - store.requestWorkers.s = []*regionRequestWorker{worker} - client.stores.Store(store.storeAddr, store) + const storeAddr = "store-1" + worker := ®ionRequestWorker{storeAddr: storeAddr, controlQueue: newControlQueue()} + store := ®ionRequestStore{workers: []*regionRequestWorker{worker}} + client.regionScheduler = ®ionRequestScheduler{} + client.regionScheduler.stores.Store(storeAddr, store) client.setTableStopped(span) @@ -340,12 +341,12 @@ func TestStopTaskUsesSubscribedSpanFilterLoop(t *testing.T) { require.True(t, req.filterLoop) } -func TestOnRegionFailQueuesCanceledErrorCache(t *testing.T) { +func TestRegionFailureHandlerQueuesCanceledError(t *testing.T) { client := &subscriptionClient{ eventSink: ®ionEventSink{ds: &mockDynamicStream{}}, } client.spanRegistry = newSpanRegistry(nil, nil) - client.failureHandler = newRegionFailureHandler(client) + client.failureHandler = newRegionFailureHandler(nil, client.onTableDrained, nil, nil) rawSpan := heartbeatpb.TableSpan{ TableID: 1, StartKey: []byte("a"), @@ -364,7 +365,7 @@ func TestOnRegionFailQueuesCanceledErrorCache(t *testing.T) { require.Equal(t, regionlock.LockRangeStatusSuccess, res2.Status) require.False(t, span.rangeLock.Stop()) - client.onRegionFail(newRegionErrorInfo(regionInfo{ + client.failureHandler.Report(newRegionErrorInfo(regionInfo{ verID: tikv.NewRegionVerID(1, 1, 1), span: heartbeatpb.TableSpan{TableID: 1, StartKey: []byte("a"), EndKey: []byte("m")}, subscribedSpan: span, @@ -374,7 +375,7 @@ func TestOnRegionFailQueuesCanceledErrorCache(t *testing.T) { require.Len(t, client.failureHandler.cache.cache, 1) require.Len(t, span.rangeLock.IterAll(nil).UnLockedRanges, 1) - client.onRegionFail(newRegionErrorInfo(regionInfo{ + client.failureHandler.Report(newRegionErrorInfo(regionInfo{ verID: tikv.NewRegionVerID(2, 1, 1), span: heartbeatpb.TableSpan{TableID: 1, StartKey: []byte("m"), EndKey: []byte("z")}, subscribedSpan: span, @@ -415,15 +416,15 @@ func (s *mockDynamicStream) GetMetrics() dynstream.Metrics[int, SubscriptionID] return dynstream.Metrics[int, SubscriptionID]{} } -func TestPushRegionEventToDSUnblocksOnClose(t *testing.T) { +func TestRegionEventSinkPushUnblocksOnClientClose(t *testing.T) { sink := ®ionEventSink{ ctx: context.Background(), ds: &mockDynamicStream{}, } sink.cond = sync.NewCond(&sink.mu) - client := &subscriptionClient{ - eventSink: sink, - regionTaskQueue: priorityqueue.New[*regionPriorityTask](), + client := &subscriptionClient{eventSink: sink} + client.regionScheduler = ®ionRequestScheduler{ + taskQueue: priorityqueue.New[*regionPriorityTask](), } client.ctx, client.cancel = context.WithCancel(context.Background()) @@ -431,7 +432,7 @@ func TestPushRegionEventToDSUnblocksOnClose(t *testing.T) { done := make(chan struct{}) go func() { - client.pushRegionEventToDS(SubscriptionID(1), regionEvent{}) + sink.Push(SubscriptionID(1), regionEvent{}) close(done) }() @@ -451,24 +452,25 @@ func TestPushRegionEventToDSUnblocksOnClose(t *testing.T) { } func TestBroadcastDeregisterUsesWorkerControlQueue(t *testing.T) { - client := &subscriptionClient{} + scheduler := ®ionRequestScheduler{} admission := newRegionAdmissionController(1, 1) + const storeAddr = "store-1" worker := ®ionRequestWorker{ + storeAddr: storeAddr, admission: admission, controlQueue: newControlQueue(), } - store := &requestedStore{storeAddr: "store-1"} - store.requestWorkers.s = []*regionRequestWorker{worker} - client.stores.Store(store.storeAddr, store) + store := ®ionRequestStore{workers: []*regionRequestWorker{worker}} + scheduler.stores.Store(storeAddr, store) dummyRegion := regionInfo{ subscribedSpan: &subscribedSpan{subID: SubscriptionID(2)}, lockedRangeState: ®ionlock.LockedRangeState{}, } - require.True(t, admission.submit(NewRegionPriorityTask(dummyRegion, 1, 1))) + require.True(t, admission.submit(newRegionPriorityTask(dummyRegion, 1, 1))) - client.broadcastDeregister(SubscriptionID(1), true) + scheduler.BroadcastDeregister(SubscriptionID(1), true) require.Equal(t, 1, worker.controlQueue.len()) req, ok := worker.controlQueue.tryPop() require.True(t, ok) @@ -477,11 +479,12 @@ func TestBroadcastDeregisterUsesWorkerControlQueue(t *testing.T) { require.Equal(t, 1, admission.stats().pending) } -func TestRequestedStoreDistributesRegionsAcrossWorkerBuffers(t *testing.T) { +func TestRegionRequestStoreDistributesRegionsAcrossWorkers(t *testing.T) { worker1 := ®ionRequestWorker{admission: newRegionAdmissionController(1, 1)} worker2 := ®ionRequestWorker{admission: newRegionAdmissionController(1, 1)} - store := &requestedStore{storeAddr: "store-1"} - store.requestWorkers.s = []*regionRequestWorker{worker1, worker2} + store := ®ionRequestStore{ + workers: []*regionRequestWorker{worker1, worker2}, + } for i := uint64(1); i <= 4; i++ { region := regionInfo{ @@ -489,7 +492,7 @@ func TestRequestedStoreDistributesRegionsAcrossWorkerBuffers(t *testing.T) { subscribedSpan: &subscribedSpan{subID: 1}, lockedRangeState: ®ionlock.LockedRangeState{}, } - require.True(t, store.submit(NewRegionPriorityTask(region, 1, i))) + require.True(t, store.submit(newRegionPriorityTask(region, 1, i))) } require.Equal(t, 2, worker1.admission.stats().pending) @@ -525,11 +528,7 @@ func TestSubscriptionWithFailedTiKV(t *testing.T) { // bootstrap cluster with a region which leader is in invalid store. cluster.Bootstrap(11, []uint64{1, 2, 3}, []uint64{4, 5, 6}, 6) - clientConfig := &SubscriptionClientConfig{ - RegionRequestWorkerPerStore: 2, - } client := NewSubscriptionClient( - clientConfig, pdClient, nil, // we don't need it in this unittest, so we can pass nil &security.Credential{}, diff --git a/pkg/metrics/log_puller.go b/pkg/metrics/log_puller.go index e874aa4428..a74f1b03c4 100644 --- a/pkg/metrics/log_puller.go +++ b/pkg/metrics/log_puller.go @@ -78,7 +78,7 @@ var ( Namespace: "ticdc", Subsystem: "subscription_client", Name: "requested_region_count", - Help: "The number of requested regions", + Help: "The number of region requests by state.", }, []string{"state"}) RegionRequestFinishScanDuration = prometheus.NewHistogram( prometheus.HistogramOpts{ diff --git a/server/server.go b/server/server.go index b9dc5edad8..8a604d35cc 100644 --- a/server/server.go +++ b/server/server.go @@ -198,9 +198,7 @@ func (c *server) initialize(ctx context.Context) error { conf := config.GetGlobalServerConfig() schemaStore := schemastore.New(conf.DataDir, c.pdClient) subscriptionClient := logpuller.NewSubscriptionClient( - &logpuller.SubscriptionClientConfig{ - RegionRequestWorkerPerStore: 8, - }, c.pdClient, + c.pdClient, txnutil.NewLockerResolver(), c.security, )