From 23171df8fc2f5ca42b6f249b75bd6fe6c2f86ab2 Mon Sep 17 00:00:00 2001 From: lidezhu Date: Fri, 17 Jul 2026 09:54:45 +0800 Subject: [PATCH 1/6] logpuller: extract region request scheduler from subscription client --- .../logpuller/region_request_scheduler.go | 178 +++++++++++++++++ logservice/logpuller/requested_store.go | 83 ++++++++ logservice/logpuller/subscription_client.go | 187 +----------------- .../logpuller/subscription_client_test.go | 27 +-- 4 files changed, 286 insertions(+), 189 deletions(-) create mode 100644 logservice/logpuller/region_request_scheduler.go create mode 100644 logservice/logpuller/requested_store.go diff --git a/logservice/logpuller/region_request_scheduler.go b/logservice/logpuller/region_request_scheduler.go new file mode 100644 index 0000000000..e81d169e9f --- /dev/null +++ b/logservice/logpuller/region_request_scheduler.go @@ -0,0 +1,178 @@ +// 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/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" +) + +// 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 { + client *subscriptionClient + + // 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 requestedStore. 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(client *subscriptionClient) *regionRequestScheduler { + pullerConfig := config.GetGlobalServerConfig().Debug.Puller + workerCount := int(client.config.RegionRequestWorkerPerStore) + if workerCount <= 0 { + workerCount = 1 + } + workerWindow := (pullerConfig.PendingRegionRequestQueueSize + workerCount - 1) / workerCount + return ®ionRequestScheduler{ + client: client, + taskQueue: priorityqueue.New[*regionPriorityTask](), + workerCount: workerCount, + workerWindow: workerWindow, + maxWindowMultiplier: pullerConfig.RegionRequestMaxWindowMultiplier, + } +} + +func (s *regionRequestScheduler) submit(region regionInfo) { + s.taskQueue.Push(NewRegionPriorityTask( + region, s.client.pdClock.CurrentTS(), s.sequence.Add(1))) +} + +func (s *regionRequestScheduler) run(ctx context.Context, group *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, ok := s.attachRPCContext(ctx, task.GetRegionInfo()) + if !ok { + continue + } + + store := s.getOrCreateStore(ctx, group, region.rpcCtx.Addr) + task.updateRegion(region, s.client.pdClock.CurrentTS()) + if !store.submit(task) { + 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 *regionRequestScheduler) attachRPCContext( + ctx context.Context, + region regionInfo, +) (regionInfo, bool) { + bo := tikv.NewBackoffer(ctx, tikvRequestMaxBackoff) + rpcCtx, err := s.client.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.client.onRegionFail(newRegionErrorInfo(region, &rpcCtxUnavailableErr{verID: region.verID})) + return region, false +} + +func (s *regionRequestScheduler) getOrCreateStore( + ctx context.Context, + group *errgroup.Group, + storeAddr string, +) *requestedStore { + if value, ok := s.stores.Load(storeAddr); ok { + return value.(*requestedStore) + } + + store := newRequestedStore( + s.client, storeAddr, s.workerCount, s.workerWindow, s.maxWindowMultiplier) + // run is the only writer. Publish the store after its immutable worker list + // is complete, then start its workers. + s.stores.Store(storeAddr, store) + store.run(ctx, group) + return store +} + +func (s *regionRequestScheduler) broadcastDeregister( + subID SubscriptionID, + filterLoop bool, +) { + s.stores.Range(func(_, value any) bool { + value.(*requestedStore).broadcastDeregister(subID, filterLoop) + return true + }) +} + +func (s *regionRequestScheduler) inflightCount() int { + count := 0 + s.stores.Range(func(_, value any) bool { + count += value.(*requestedStore).inflightCount() + return true + }) + return count +} + +func (s *regionRequestScheduler) close() { + s.taskQueue.Close() +} + +func (s *regionRequestScheduler) closeStores() { + s.stores.Range(func(_, value any) bool { + value.(*requestedStore).close() + return true + }) +} diff --git a/logservice/logpuller/requested_store.go b/logservice/logpuller/requested_store.go new file mode 100644 index 0000000000..f98b59b679 --- /dev/null +++ b/logservice/logpuller/requested_store.go @@ -0,0 +1,83 @@ +// 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" +) + +// requestedStore owns the 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 requestedStore struct { + storeAddr string + workers []*regionRequestWorker + nextWorker atomic.Uint64 +} + +func newRequestedStore( + client *subscriptionClient, + storeAddr string, + workerCount int, + workerWindow int, + maxWindowMultiplier int, +) *requestedStore { + store := &requestedStore{ + storeAddr: storeAddr, + workers: make([]*regionRequestWorker, 0, workerCount), + } + for i := 0; i < workerCount; i++ { + store.workers = append(store.workers, newRegionRequestWorker( + client, store, workerWindow, maxWindowMultiplier)) + } + return store +} + +func (s *requestedStore) run(ctx context.Context, group *errgroup.Group) { + for _, worker := range s.workers { + group.Go(func() error { return worker.Run(ctx) }) + } +} + +func (s *requestedStore) submit(task *regionPriorityTask) bool { + if len(s.workers) == 0 { + return false + } + index := (s.nextWorker.Add(1) - 1) % uint64(len(s.workers)) + return s.workers[index].admission.submit(task) +} + +func (s *requestedStore) broadcastDeregister(subID SubscriptionID, filterLoop bool) { + for _, worker := range s.workers { + worker.controlQueue.push(deregisterRequest{subID: subID, filterLoop: filterLoop}) + } +} + +func (s *requestedStore) close() { + for _, worker := range s.workers { + worker.admission.close() + } +} + +func (s *requestedStore) inflightCount() int { + count := 0 + for _, worker := range s.workers { + count += worker.admission.stats().inflight + } + return count +} diff --git a/logservice/logpuller/subscription_client.go b/logservice/logpuller/subscription_client.go index 2542936b6a..a7edab9ac5 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,12 @@ 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" @@ -131,8 +126,6 @@ type subscriptionClient struct { pdClock pdutil.Clock lockResolver txnutil.LockResolver - stores sync.Map - // the credential to connect tikv credential *security.Credential @@ -142,16 +135,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 @@ -168,7 +157,6 @@ func NewSubscriptionClient( subClient := &subscriptionClient{ config: config, - stores: sync.Map{}, pd: pd, regionCache: appcontext.GetService[*tikv.RegionCache](appcontext.RegionCache), pdClock: appcontext.GetService[pdutil.Clock](appcontext.DefaultPDClock), @@ -177,7 +165,6 @@ func NewSubscriptionClient( credential: credential, rangeTaskCh: make(chan rangeTask, 1024), - regionTaskQueue: priorityqueue.New[*regionPriorityTask](), resolveLockTaskCh: make(chan resolveLockTask, 1024), resolveLockRateLimiter: newResolveLockRateLimiter(), } @@ -185,6 +172,7 @@ func NewSubscriptionClient( subClient.failureHandler = newRegionFailureHandler(subClient) subClient.eventSink = newRegionEventSink(subClient.ctx, subClient.failureHandler) subClient.spanRegistry = newSpanRegistry(subClient.pd, subClient.pdClock) + subClient.regionScheduler = newRegionRequestScheduler(subClient) subClient.initMetrics() return subClient @@ -212,12 +200,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 - }) + pendingRegionReqCount := s.regionScheduler.inflightCount() metrics.SubscriptionClientRequestedRegionCount.WithLabelValues("pending").Set(float64(pendingRegionReqCount)) s.eventSink.UpdateMetrics() @@ -303,7 +286,7 @@ func (s *subscriptionClient) Run(ctx context.Context) error { g.Go(func() error { return s.updateMetrics(ctx) }) g.Go(func() error { return s.eventSink.Run(ctx) }) 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.failureHandler.Run(ctx) }) g.Go(func() error { return s.handleResolveLockTasks(ctx) }) g.Go(func() error { return s.spanRegistry.Run(ctx) }) @@ -317,7 +300,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 +311,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) } @@ -353,155 +336,6 @@ 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. @@ -605,8 +439,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,8 +455,7 @@ 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) diff --git a/logservice/logpuller/subscription_client_test.go b/logservice/logpuller/subscription_client_test.go index c3ef992a34..4fb76df372 100644 --- a/logservice/logpuller/subscription_client_test.go +++ b/logservice/logpuller/subscription_client_test.go @@ -328,9 +328,9 @@ 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) + store := &requestedStore{storeAddr: "store-1", workers: []*regionRequestWorker{worker}} + client.regionScheduler = ®ionRequestScheduler{client: client} + client.regionScheduler.stores.Store(store.storeAddr, store) client.setTableStopped(span) @@ -421,9 +421,10 @@ func TestPushRegionEventToDSUnblocksOnClose(t *testing.T) { ds: &mockDynamicStream{}, } sink.cond = sync.NewCond(&sink.mu) - client := &subscriptionClient{ - eventSink: sink, - regionTaskQueue: priorityqueue.New[*regionPriorityTask](), + client := &subscriptionClient{eventSink: sink} + client.regionScheduler = ®ionRequestScheduler{ + client: client, + taskQueue: priorityqueue.New[*regionPriorityTask](), } client.ctx, client.cancel = context.WithCancel(context.Background()) @@ -452,15 +453,15 @@ func TestPushRegionEventToDSUnblocksOnClose(t *testing.T) { func TestBroadcastDeregisterUsesWorkerControlQueue(t *testing.T) { client := &subscriptionClient{} + scheduler := ®ionRequestScheduler{client: client} admission := newRegionAdmissionController(1, 1) worker := ®ionRequestWorker{ admission: admission, controlQueue: newControlQueue(), } - store := &requestedStore{storeAddr: "store-1"} - store.requestWorkers.s = []*regionRequestWorker{worker} - client.stores.Store(store.storeAddr, store) + store := &requestedStore{storeAddr: "store-1", workers: []*regionRequestWorker{worker}} + scheduler.stores.Store(store.storeAddr, store) dummyRegion := regionInfo{ subscribedSpan: &subscribedSpan{subID: SubscriptionID(2)}, @@ -468,7 +469,7 @@ func TestBroadcastDeregisterUsesWorkerControlQueue(t *testing.T) { } 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) @@ -480,8 +481,10 @@ func TestBroadcastDeregisterUsesWorkerControlQueue(t *testing.T) { func TestRequestedStoreDistributesRegionsAcrossWorkerBuffers(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 := &requestedStore{ + storeAddr: "store-1", + workers: []*regionRequestWorker{worker1, worker2}, + } for i := uint64(1); i <= 4; i++ { region := regionInfo{ From 520d3481b3c9a590755fab1fc559ed0df9b211e3 Mon Sep 17 00:00:00 2001 From: lidezhu Date: Fri, 17 Jul 2026 11:07:32 +0800 Subject: [PATCH 2/6] refactor --- .../logpuller/region_failure_handler.go | 49 +++++++---- .../logpuller/region_request_scheduler.go | 34 ++++++-- logservice/logpuller/region_request_worker.go | 49 ++++++----- .../logpuller/region_request_worker_test.go | 43 ++++------ logservice/logpuller/requested_store.go | 6 +- logservice/logpuller/subscription_client.go | 86 +++++++++---------- .../logpuller/subscription_client_test.go | 22 +++-- 7 files changed, 159 insertions(+), 130 deletions(-) diff --git a/logservice/logpuller/region_failure_handler.go b/logservice/logpuller/region_failure_handler.go index ee89e805df..34434bedde 100644 --- a/logservice/logpuller/region_failure_handler.go +++ b/logservice/logpuller/region_failure_handler.go @@ -19,6 +19,7 @@ import ( "time" "github.com/pingcap/log" + "github.com/pingcap/ticdc/heartbeatpb" "github.com/pingcap/ticdc/pkg/errors" "github.com/pingcap/ticdc/pkg/metrics" "github.com/tikv/client-go/v2/tikv" @@ -41,14 +42,26 @@ 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, heartbeatpb.TableSpan, *subscribedSpan, bool, bool) } -func newRegionFailureHandler(client *subscriptionClient) *regionFailureHandler { +func newRegionFailureHandler( + regionCache *tikv.RegionCache, + onTableDrained func(*subscribedSpan), + scheduleRegionRequest func(context.Context, regionInfo), + scheduleRangeRequest func(context.Context, heartbeatpb.TableSpan, *subscribedSpan, bool, bool), +) *regionFailureHandler { return ®ionFailureHandler{ - cache: newErrCache(), - client: client, + cache: newErrCache(), + regionCache: regionCache, + onTableDrained: onTableDrained, + scheduleRegionRequest: scheduleRegionRequest, + scheduleRangeRequest: scheduleRangeRequest, } } @@ -59,7 +72,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) @@ -100,28 +113,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.scheduleRangeRequest(ctx, errInfo.span, errInfo.subscribedSpan, errInfo.filterLoop, errInfo.wasInitialized) return nil } if innerErr.GetRegionNotFound() != nil { metricFeedRegionNotFoundCounter.Inc() - r.client.scheduleRangeRequest(ctx, errInfo.span, errInfo.subscribedSpan, errInfo.filterLoop, errInfo.wasInitialized) + r.scheduleRangeRequest(ctx, errInfo.span, errInfo.subscribedSpan, errInfo.filterLoop, errInfo.wasInitialized) 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,24 +153,24 @@ 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.scheduleRangeRequest(ctx, errInfo.span, errInfo.subscribedSpan, errInfo.filterLoop, errInfo.wasInitialized) 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.scheduleRangeRequest(ctx, errInfo.span, errInfo.subscribedSpan, errInfo.filterLoop, errInfo.wasInitialized) 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. diff --git a/logservice/logpuller/region_request_scheduler.go b/logservice/logpuller/region_request_scheduler.go index e81d169e9f..0d457d7080 100644 --- a/logservice/logpuller/region_request_scheduler.go +++ b/logservice/logpuller/region_request_scheduler.go @@ -34,7 +34,9 @@ import ( // resolution and retry policy remain owned by subscriptionClient and // regionFailureHandler respectively. type regionRequestScheduler struct { - client *subscriptionClient + upstream *upstreamHandle + eventSink *regionEventSink + failureHandler *regionFailureHandler // taskQueue orders all Regions before they are assigned to a TiKV store. taskQueue *priorityqueue.PriorityQueue[*regionPriorityTask] @@ -52,15 +54,22 @@ type regionRequestScheduler struct { maxWindowMultiplier int } -func newRegionRequestScheduler(client *subscriptionClient) *regionRequestScheduler { +func newRegionRequestScheduler( + clientConfig *SubscriptionClientConfig, + upstream *upstreamHandle, + eventSink *regionEventSink, + failureHandler *regionFailureHandler, +) *regionRequestScheduler { pullerConfig := config.GetGlobalServerConfig().Debug.Puller - workerCount := int(client.config.RegionRequestWorkerPerStore) + workerCount := int(clientConfig.RegionRequestWorkerPerStore) if workerCount <= 0 { workerCount = 1 } workerWindow := (pullerConfig.PendingRegionRequestQueueSize + workerCount - 1) / workerCount return ®ionRequestScheduler{ - client: client, + upstream: upstream, + eventSink: eventSink, + failureHandler: failureHandler, taskQueue: priorityqueue.New[*regionPriorityTask](), workerCount: workerCount, workerWindow: workerWindow, @@ -70,7 +79,7 @@ func newRegionRequestScheduler(client *subscriptionClient) *regionRequestSchedul func (s *regionRequestScheduler) submit(region regionInfo) { s.taskQueue.Push(NewRegionPriorityTask( - region, s.client.pdClock.CurrentTS(), s.sequence.Add(1))) + region, s.upstream.pdClock.CurrentTS(), s.sequence.Add(1))) } func (s *regionRequestScheduler) run(ctx context.Context, group *errgroup.Group) error { @@ -96,7 +105,7 @@ func (s *regionRequestScheduler) run(ctx context.Context, group *errgroup.Group) } store := s.getOrCreateStore(ctx, group, region.rpcCtx.Addr) - task.updateRegion(region, s.client.pdClock.CurrentTS()) + task.updateRegion(region, s.upstream.pdClock.CurrentTS()) if !store.submit(task) { return context.Canceled } @@ -113,7 +122,7 @@ func (s *regionRequestScheduler) attachRPCContext( region regionInfo, ) (regionInfo, bool) { bo := tikv.NewBackoffer(ctx, tikvRequestMaxBackoff) - rpcCtx, err := s.client.regionCache.GetTiKVRPCContext( + rpcCtx, err := s.upstream.regionCache.GetTiKVRPCContext( bo, region.verID, kvclientv2.ReplicaReadLeader, 0) if rpcCtx != nil { region.rpcCtx = rpcCtx @@ -125,7 +134,7 @@ func (s *regionRequestScheduler) attachRPCContext( zap.Uint64("regionID", region.verID.GetID()), zap.Error(err)) } - s.client.onRegionFail(newRegionErrorInfo(region, &rpcCtxUnavailableErr{verID: region.verID})) + s.failureHandler.Report(newRegionErrorInfo(region, &rpcCtxUnavailableErr{verID: region.verID})) return region, false } @@ -139,7 +148,14 @@ func (s *regionRequestScheduler) getOrCreateStore( } store := newRequestedStore( - s.client, storeAddr, s.workerCount, s.workerWindow, s.maxWindowMultiplier) + s.upstream, + s.eventSink, + s.failureHandler, + storeAddr, + s.workerCount, + s.workerWindow, + s.maxWindowMultiplier, + ) // run is the only writer. Publish the store after its immutable worker list // is complete, then start its workers. s.stores.Store(storeAddr, store) diff --git a/logservice/logpuller/region_request_worker.go b/logservice/logpuller/region_request_worker.go index 0e6a5507fb..458a7aeba1 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 + store *requestedStore admission *regionAdmissionController controlQueue *controlQueue @@ -98,19 +103,23 @@ type regionRequestWorker struct { } func newRegionRequestWorker( - client *subscriptionClient, + upstream *upstreamHandle, + eventSink *regionEventSink, + failureHandler *regionFailureHandler, store *requestedStore, 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, + store: store, + 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) @@ -163,13 +172,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 +198,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 } @@ -218,7 +227,7 @@ func (s *regionRequestWorker) runStream(ctx context.Context, firstReq *regionReq }() g, gctx := errgroup.WithContext(ctx) - conn, err := Connect(gctx, s.client.credential, s.store.storeAddr) + conn, err := Connect(gctx, s.upstream.credential, s.store.storeAddr) if err != nil { log.Warn("region request worker create grpc stream failed", zap.Uint64("workerID", s.workerID), @@ -318,7 +327,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 +349,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 +365,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, }) @@ -404,7 +413,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{}, @@ -447,7 +456,7 @@ func (s *regionRequestWorker) sendRegionRequest(conn *ConnAndClient, req *region if region.subscribedSpan.stopped.Load() { req.abort() - s.client.onRegionFail(newRegionErrorInfo(region, &storeStreamErr{})) + s.failureHandler.Report(newRegionErrorInfo(region, &storeStreamErr{})) return nil } @@ -504,7 +513,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..70affb2d0a 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" @@ -84,7 +83,7 @@ func TestRegionRequestWorkerIgnoresDuplicateActiveRegion(t *testing.T) { worker := ®ionRequestWorker{ admission: admission, store: &requestedStore{storeAddr: "store-1"}, - client: &subscriptionClient{}, + 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), @@ -408,7 +399,7 @@ func TestFailPendingRegionsReschedulesWorkerBuffer(t *testing.T) { 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) { @@ -417,7 +408,7 @@ func TestProcessRegionSendTaskSendFailureCleansSentRequest(t *testing.T) { admission: admission, controlQueue: newControlQueue(), store: &requestedStore{storeAddr: "store-1"}, - client: &subscriptionClient{}, + upstream: &upstreamHandle{}, tracker: newRegionTracker(), } @@ -448,7 +439,7 @@ func TestProcessRegionSendTaskDoesNotSendRemovedRequest(t *testing.T) { admission: admission, controlQueue: newControlQueue(), store: &requestedStore{storeAddr: "store-1"}, - client: &subscriptionClient{}, + upstream: &upstreamHandle{}, tracker: newRegionTracker(), } region := prepareRegionForSendTest(createTestRegionInfo(1, 1)) @@ -491,7 +482,7 @@ func TestProcessRegionSendTaskSendEOFIsRetriable(t *testing.T) { admission: admission, controlQueue: newControlQueue(), store: &requestedStore{storeAddr: "store-1"}, - client: &subscriptionClient{}, + upstream: &upstreamHandle{}, tracker: newRegionTracker(), } region := prepareRegionForSendTest(createTestRegionInfo(1, 1)) @@ -524,10 +515,9 @@ func TestProcessRegionSendTaskHandlesDeregisterFromControlQueue(t *testing.T) { admission: newRegionAdmissionController(1, 1), controlQueue: newControlQueue(), store: &requestedStore{storeAddr: "store-1"}, - client: &subscriptionClient{ - eventSink: ®ionEventSink{ds: ds}, - }, - tracker: newRegionTracker(), + 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()) diff --git a/logservice/logpuller/requested_store.go b/logservice/logpuller/requested_store.go index f98b59b679..97294a8e33 100644 --- a/logservice/logpuller/requested_store.go +++ b/logservice/logpuller/requested_store.go @@ -31,7 +31,9 @@ type requestedStore struct { } func newRequestedStore( - client *subscriptionClient, + upstream *upstreamHandle, + eventSink *regionEventSink, + failureHandler *regionFailureHandler, storeAddr string, workerCount int, workerWindow int, @@ -43,7 +45,7 @@ func newRequestedStore( } for i := 0; i < workerCount; i++ { store.workers = append(store.workers, newRegionRequestWorker( - client, store, workerWindow, maxWindowMultiplier)) + upstream, eventSink, failureHandler, store, workerWindow, maxWindowMultiplier)) } return store } diff --git a/logservice/logpuller/subscription_client.go b/logservice/logpuller/subscription_client.go index a7edab9ac5..1266af764a 100644 --- a/logservice/logpuller/subscription_client.go +++ b/logservice/logpuller/subscription_client.go @@ -30,7 +30,6 @@ import ( "github.com/pingcap/ticdc/pkg/security" "github.com/pingcap/ticdc/pkg/spanz" "github.com/pingcap/ticdc/pkg/util" - "github.com/prometheus/client_golang/prometheus" "github.com/tikv/client-go/v2/tikv" pd "github.com/tikv/pd/client" "go.uber.org/zap" @@ -90,8 +89,21 @@ type SubscriptionClientConfig struct { RegionRequestWorkerPerStore uint } -type sharedClientMetrics struct { - batchResolvedSize prometheus.Observer +// upstreamHandle contains the stable TiKV and PD dependencies shared by the +// region request pipeline. Runtime components keep their own event and error +// dependencies instead of using this as a general service container. +type upstreamHandle struct { + pd pd.Client + regionCache *tikv.RegionCache + pdClock pdutil.Clock + credential *security.Credential + clusterID uint64 +} + +// 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. @@ -115,19 +127,12 @@ 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 + ctx context.Context + cancel context.CancelFunc + config *SubscriptionClientConfig + upstream *upstreamHandle - // the credential to connect tikv - credential *security.Credential + lockResolver txnutil.LockResolver // failureHandler handles failed regions and owns reschedule/retry decisions. failureHandler *regionFailureHandler @@ -156,25 +161,33 @@ func NewSubscriptionClient( ) SubscriptionClient { subClient := &subscriptionClient{ config: config, - - 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), 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.regionScheduler = newRegionRequestScheduler(subClient) - - subClient.initMetrics() + subClient.spanRegistry = newSpanRegistry(subClient.upstream.pd, subClient.upstream.pdClock) + subClient.regionScheduler = newRegionRequestScheduler( + subClient.config, + subClient.upstream, + subClient.eventSink, + subClient.failureHandler, + ) return subClient } @@ -187,11 +200,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() @@ -269,17 +277,13 @@ 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 { + if s.upstream == nil || s.upstream.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) @@ -331,11 +335,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) -} - 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. @@ -380,7 +379,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)), diff --git a/logservice/logpuller/subscription_client_test.go b/logservice/logpuller/subscription_client_test.go index 4fb76df372..4387fa2054 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, @@ -329,7 +329,7 @@ func TestStopTaskUsesSubscribedSpanFilterLoop(t *testing.T) { require.Equal(t, regionlock.LockRangeStatusSuccess, res.Status) worker := ®ionRequestWorker{controlQueue: newControlQueue()} store := &requestedStore{storeAddr: "store-1", workers: []*regionRequestWorker{worker}} - client.regionScheduler = ®ionRequestScheduler{client: client} + client.regionScheduler = ®ionRequestScheduler{} client.regionScheduler.stores.Store(store.storeAddr, store) client.setTableStopped(span) @@ -340,12 +340,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 +364,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 +374,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,7 +415,7 @@ 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{}, @@ -423,7 +423,6 @@ func TestPushRegionEventToDSUnblocksOnClose(t *testing.T) { sink.cond = sync.NewCond(&sink.mu) client := &subscriptionClient{eventSink: sink} client.regionScheduler = ®ionRequestScheduler{ - client: client, taskQueue: priorityqueue.New[*regionPriorityTask](), } client.ctx, client.cancel = context.WithCancel(context.Background()) @@ -432,7 +431,7 @@ func TestPushRegionEventToDSUnblocksOnClose(t *testing.T) { done := make(chan struct{}) go func() { - client.pushRegionEventToDS(SubscriptionID(1), regionEvent{}) + sink.Push(SubscriptionID(1), regionEvent{}) close(done) }() @@ -452,8 +451,7 @@ func TestPushRegionEventToDSUnblocksOnClose(t *testing.T) { } func TestBroadcastDeregisterUsesWorkerControlQueue(t *testing.T) { - client := &subscriptionClient{} - scheduler := ®ionRequestScheduler{client: client} + scheduler := ®ionRequestScheduler{} admission := newRegionAdmissionController(1, 1) worker := ®ionRequestWorker{ From 980bd5da06abbbcccd28affe4c83cf08c94a6ad1 Mon Sep 17 00:00:00 2001 From: lidezhu Date: Fri, 17 Jul 2026 14:20:20 +0800 Subject: [PATCH 3/6] refactor --- .../logpuller/region_failure_handler.go | 4 +- .../logpuller/region_request_scheduler.go | 77 +++++++++---------- logservice/logpuller/region_request_worker.go | 22 +++--- .../logpuller/region_request_worker_test.go | 14 ++-- ..._store.go => store_request_worker_pool.go} | 30 ++++---- logservice/logpuller/subscription_client.go | 16 +--- .../logpuller/subscription_client_test.go | 18 ++--- pkg/metrics/log_puller.go | 2 +- server/server.go | 4 +- 9 files changed, 80 insertions(+), 107 deletions(-) rename logservice/logpuller/{requested_store.go => store_request_worker_pool.go} (64%) diff --git a/logservice/logpuller/region_failure_handler.go b/logservice/logpuller/region_failure_handler.go index 34434bedde..f90483b740 100644 --- a/logservice/logpuller/region_failure_handler.go +++ b/logservice/logpuller/region_failure_handler.go @@ -85,7 +85,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 { @@ -177,7 +176,7 @@ func (r *regionFailureHandler) handleError(ctx context.Context, errInfo regionEr 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 @@ -236,7 +235,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 index 0d457d7080..f6bb45e21d 100644 --- a/logservice/logpuller/region_request_scheduler.go +++ b/logservice/logpuller/region_request_scheduler.go @@ -29,6 +29,8 @@ import ( "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 @@ -42,9 +44,9 @@ type regionRequestScheduler struct { 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 requestedStore. Stores are created only by - // run, but are also read by metrics and deregistration goroutines. - stores sync.Map + // workerPools maps TiKV addresses to storeRequestWorkerPool. Pools are + // created only by run, but are also read by metrics and deregistration goroutines. + workerPools sync.Map // workerCount is the configured number of request workers per store. workerCount int @@ -55,16 +57,12 @@ type regionRequestScheduler struct { } func newRegionRequestScheduler( - clientConfig *SubscriptionClientConfig, upstream *upstreamHandle, eventSink *regionEventSink, failureHandler *regionFailureHandler, ) *regionRequestScheduler { pullerConfig := config.GetGlobalServerConfig().Debug.Puller - workerCount := int(clientConfig.RegionRequestWorkerPerStore) - if workerCount <= 0 { - workerCount = 1 - } + workerCount := regionRequestWorkerPerStore workerWindow := (pullerConfig.PendingRegionRequestQueueSize + workerCount - 1) / workerCount return ®ionRequestScheduler{ upstream: upstream, @@ -82,8 +80,8 @@ func (s *regionRequestScheduler) submit(region regionInfo) { region, s.upstream.pdClock.CurrentTS(), s.sequence.Add(1))) } -func (s *regionRequestScheduler) run(ctx context.Context, group *errgroup.Group) error { - defer s.closeStores() +func (s *regionRequestScheduler) run(ctx context.Context, workerGroup *errgroup.Group) error { + defer s.closeWorkerPools() for { select { case <-ctx.Done(): @@ -99,55 +97,50 @@ func (s *regionRequestScheduler) run(ctx context.Context, group *errgroup.Group) return err } - region, ok := s.attachRPCContext(ctx, task.GetRegionInfo()) - if !ok { + region, err := s.attachRPCContext(ctx, task.GetRegionInfo()) + if err != nil { + s.failureHandler.Report(newRegionErrorInfo(region, err)) continue } - store := s.getOrCreateStore(ctx, group, region.rpcCtx.Addr) + workerPool := s.getOrCreateWorkerPool(ctx, workerGroup, region.rpcCtx.Addr) task.updateRegion(region, s.upstream.pdClock.CurrentTS()) - if !store.submit(task) { + if !workerPool.submit(task) { 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 *regionRequestScheduler) attachRPCContext( ctx context.Context, region regionInfo, -) (regionInfo, bool) { +) (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, true + return region, nil } if err != nil { - log.Debug("subscription client get rpc context fail", + 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)) } - s.failureHandler.Report(newRegionErrorInfo(region, &rpcCtxUnavailableErr{verID: region.verID})) - return region, false + return region, &rpcCtxUnavailableErr{verID: region.verID} } -func (s *regionRequestScheduler) getOrCreateStore( +func (s *regionRequestScheduler) getOrCreateWorkerPool( ctx context.Context, - group *errgroup.Group, + workerGroup *errgroup.Group, storeAddr string, -) *requestedStore { - if value, ok := s.stores.Load(storeAddr); ok { - return value.(*requestedStore) +) *storeRequestWorkerPool { + if value, ok := s.workerPools.Load(storeAddr); ok { + return value.(*storeRequestWorkerPool) } - store := newRequestedStore( + workerPool := newStoreRequestWorkerPool( s.upstream, s.eventSink, s.failureHandler, @@ -156,27 +149,27 @@ func (s *regionRequestScheduler) getOrCreateStore( s.workerWindow, s.maxWindowMultiplier, ) - // run is the only writer. Publish the store after its immutable worker list - // is complete, then start its workers. - s.stores.Store(storeAddr, store) - store.run(ctx, group) - return store + // The scheduler run loop is the only writer. Publish the pool after its + // immutable worker list is complete, then start its workers. + s.workerPools.Store(storeAddr, workerPool) + workerPool.startWorkers(ctx, workerGroup) + return workerPool } func (s *regionRequestScheduler) broadcastDeregister( subID SubscriptionID, filterLoop bool, ) { - s.stores.Range(func(_, value any) bool { - value.(*requestedStore).broadcastDeregister(subID, filterLoop) + s.workerPools.Range(func(_, value any) bool { + value.(*storeRequestWorkerPool).broadcastDeregister(subID, filterLoop) return true }) } func (s *regionRequestScheduler) inflightCount() int { count := 0 - s.stores.Range(func(_, value any) bool { - count += value.(*requestedStore).inflightCount() + s.workerPools.Range(func(_, value any) bool { + count += value.(*storeRequestWorkerPool).inflightCount() return true }) return count @@ -186,9 +179,9 @@ func (s *regionRequestScheduler) close() { s.taskQueue.Close() } -func (s *regionRequestScheduler) closeStores() { - s.stores.Range(func(_, value any) bool { - value.(*requestedStore).close() +func (s *regionRequestScheduler) closeWorkerPools() { + s.workerPools.Range(func(_, value any) bool { + value.(*storeRequestWorkerPool).close() return true }) } diff --git a/logservice/logpuller/region_request_worker.go b/logservice/logpuller/region_request_worker.go index 458a7aeba1..298e2cc266 100644 --- a/logservice/logpuller/region_request_worker.go +++ b/logservice/logpuller/region_request_worker.go @@ -95,7 +95,7 @@ type regionRequestWorker struct { upstream *upstreamHandle eventSink *regionEventSink failureHandler *regionFailureHandler - store *requestedStore + storeAddr string admission *regionAdmissionController controlQueue *controlQueue @@ -106,7 +106,7 @@ func newRegionRequestWorker( upstream *upstreamHandle, eventSink *regionEventSink, failureHandler *regionFailureHandler, - store *requestedStore, + storeAddr string, currentWindow int, maxWindowMultiplier int, ) *regionRequestWorker { @@ -116,7 +116,7 @@ func newRegionRequestWorker( upstream: upstream, eventSink: eventSink, failureHandler: failureHandler, - store: store, + storeAddr: storeAddr, admission: newRegionAdmissionController(currentWindow, maxWindowMultiplier), controlQueue: newControlQueue(), tracker: newRegionTracker(), @@ -207,7 +207,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{} @@ -218,20 +218,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.upstream.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() @@ -279,7 +279,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) @@ -401,7 +401,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) } @@ -451,7 +451,7 @@ 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() { diff --git a/logservice/logpuller/region_request_worker_test.go b/logservice/logpuller/region_request_worker_test.go index 70affb2d0a..9b75667040 100644 --- a/logservice/logpuller/region_request_worker_test.go +++ b/logservice/logpuller/region_request_worker_test.go @@ -82,7 +82,7 @@ func TestRegionRequestWorkerIgnoresDuplicateActiveRegion(t *testing.T) { admission := newRegionAdmissionController(10, 1) worker := ®ionRequestWorker{ admission: admission, - store: &requestedStore{storeAddr: "store-1"}, + storeAddr: "store-1", upstream: &upstreamHandle{}, tracker: newRegionTracker(), } @@ -407,7 +407,7 @@ func TestProcessRegionSendTaskSendFailureCleansSentRequest(t *testing.T) { worker := ®ionRequestWorker{ admission: admission, controlQueue: newControlQueue(), - store: &requestedStore{storeAddr: "store-1"}, + storeAddr: "store-1", upstream: &upstreamHandle{}, tracker: newRegionTracker(), } @@ -438,7 +438,7 @@ func TestProcessRegionSendTaskDoesNotSendRemovedRequest(t *testing.T) { worker := ®ionRequestWorker{ admission: admission, controlQueue: newControlQueue(), - store: &requestedStore{storeAddr: "store-1"}, + storeAddr: "store-1", upstream: &upstreamHandle{}, tracker: newRegionTracker(), } @@ -481,7 +481,7 @@ func TestProcessRegionSendTaskSendEOFIsRetriable(t *testing.T) { worker := ®ionRequestWorker{ admission: admission, controlQueue: newControlQueue(), - store: &requestedStore{storeAddr: "store-1"}, + storeAddr: "store-1", upstream: &upstreamHandle{}, tracker: newRegionTracker(), } @@ -514,7 +514,7 @@ func TestProcessRegionSendTaskHandlesDeregisterFromControlQueue(t *testing.T) { worker := ®ionRequestWorker{ admission: newRegionAdmissionController(1, 1), controlQueue: newControlQueue(), - store: &requestedStore{storeAddr: "store-1"}, + storeAddr: "store-1", upstream: &upstreamHandle{clusterID: 42}, eventSink: ®ionEventSink{ds: ds}, tracker: newRegionTracker(), @@ -562,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/requested_store.go b/logservice/logpuller/store_request_worker_pool.go similarity index 64% rename from logservice/logpuller/requested_store.go rename to logservice/logpuller/store_request_worker_pool.go index 97294a8e33..6f84e0019a 100644 --- a/logservice/logpuller/requested_store.go +++ b/logservice/logpuller/store_request_worker_pool.go @@ -21,16 +21,16 @@ import ( "golang.org/x/sync/errgroup" ) -// requestedStore owns the request workers connected to one TiKV store. The -// worker slice is complete before the store is published and is immutable +// storeRequestWorkerPool owns the request workers connected to one TiKV store. +// The worker slice is complete before the pool is published and is immutable // afterwards, so task submission only needs an atomic round-robin counter. -type requestedStore struct { +type storeRequestWorkerPool struct { storeAddr string workers []*regionRequestWorker nextWorker atomic.Uint64 } -func newRequestedStore( +func newStoreRequestWorkerPool( upstream *upstreamHandle, eventSink *regionEventSink, failureHandler *regionFailureHandler, @@ -38,25 +38,25 @@ func newRequestedStore( workerCount int, workerWindow int, maxWindowMultiplier int, -) *requestedStore { - store := &requestedStore{ +) *storeRequestWorkerPool { + workerPool := &storeRequestWorkerPool{ storeAddr: storeAddr, workers: make([]*regionRequestWorker, 0, workerCount), } for i := 0; i < workerCount; i++ { - store.workers = append(store.workers, newRegionRequestWorker( - upstream, eventSink, failureHandler, store, workerWindow, maxWindowMultiplier)) + workerPool.workers = append(workerPool.workers, newRegionRequestWorker( + upstream, eventSink, failureHandler, storeAddr, workerWindow, maxWindowMultiplier)) } - return store + return workerPool } -func (s *requestedStore) run(ctx context.Context, group *errgroup.Group) { +func (s *storeRequestWorkerPool) startWorkers(ctx context.Context, workerGroup *errgroup.Group) { for _, worker := range s.workers { - group.Go(func() error { return worker.Run(ctx) }) + workerGroup.Go(func() error { return worker.Run(ctx) }) } } -func (s *requestedStore) submit(task *regionPriorityTask) bool { +func (s *storeRequestWorkerPool) submit(task *regionPriorityTask) bool { if len(s.workers) == 0 { return false } @@ -64,19 +64,19 @@ func (s *requestedStore) submit(task *regionPriorityTask) bool { return s.workers[index].admission.submit(task) } -func (s *requestedStore) broadcastDeregister(subID SubscriptionID, filterLoop bool) { +func (s *storeRequestWorkerPool) broadcastDeregister(subID SubscriptionID, filterLoop bool) { for _, worker := range s.workers { worker.controlQueue.push(deregisterRequest{subID: subID, filterLoop: filterLoop}) } } -func (s *requestedStore) close() { +func (s *storeRequestWorkerPool) close() { for _, worker := range s.workers { worker.admission.close() } } -func (s *requestedStore) inflightCount() int { +func (s *storeRequestWorkerPool) inflightCount() int { count := 0 for _, worker := range s.workers { count += worker.admission.stats().inflight diff --git a/logservice/logpuller/subscription_client.go b/logservice/logpuller/subscription_client.go index 1266af764a..0b629d2fa7 100644 --- a/logservice/logpuller/subscription_client.go +++ b/logservice/logpuller/subscription_client.go @@ -84,14 +84,8 @@ type rangeTask struct { 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. Runtime components keep their own event and error -// dependencies instead of using this as a general service container. +// region request pipeline. type upstreamHandle struct { pd pd.Client regionCache *tikv.RegionCache @@ -129,7 +123,6 @@ type SubscriptionClient interface { type subscriptionClient struct { ctx context.Context cancel context.CancelFunc - config *SubscriptionClientConfig upstream *upstreamHandle lockResolver txnutil.LockResolver @@ -154,13 +147,11 @@ 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, upstream: &upstreamHandle{ pd: pd, regionCache: appcontext.GetService[*tikv.RegionCache](appcontext.RegionCache), @@ -183,7 +174,6 @@ func NewSubscriptionClient( subClient.eventSink = newRegionEventSink(subClient.ctx, subClient.failureHandler) subClient.spanRegistry = newSpanRegistry(subClient.upstream.pd, subClient.upstream.pdClock) subClient.regionScheduler = newRegionRequestScheduler( - subClient.config, subClient.upstream, subClient.eventSink, subClient.failureHandler, @@ -208,9 +198,9 @@ func (s *subscriptionClient) updateMetrics(ctx context.Context) error { case <-ctx.Done(): return ctx.Err() case <-ticker.C: - pendingRegionReqCount := s.regionScheduler.inflightCount() + inflightRegionRequestCount := s.regionScheduler.inflightCount() - metrics.SubscriptionClientRequestedRegionCount.WithLabelValues("pending").Set(float64(pendingRegionReqCount)) + metrics.SubscriptionClientRequestedRegionCount.WithLabelValues("inflight").Set(float64(inflightRegionRequestCount)) s.eventSink.UpdateMetrics() s.spanRegistry.UpdateMetrics() } diff --git a/logservice/logpuller/subscription_client_test.go b/logservice/logpuller/subscription_client_test.go index 4387fa2054..1a939d50d8 100644 --- a/logservice/logpuller/subscription_client_test.go +++ b/logservice/logpuller/subscription_client_test.go @@ -328,9 +328,9 @@ 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", workers: []*regionRequestWorker{worker}} + workerPool := &storeRequestWorkerPool{storeAddr: "store-1", workers: []*regionRequestWorker{worker}} client.regionScheduler = ®ionRequestScheduler{} - client.regionScheduler.stores.Store(store.storeAddr, store) + client.regionScheduler.workerPools.Store(workerPool.storeAddr, workerPool) client.setTableStopped(span) @@ -458,8 +458,8 @@ func TestBroadcastDeregisterUsesWorkerControlQueue(t *testing.T) { admission: admission, controlQueue: newControlQueue(), } - store := &requestedStore{storeAddr: "store-1", workers: []*regionRequestWorker{worker}} - scheduler.stores.Store(store.storeAddr, store) + workerPool := &storeRequestWorkerPool{storeAddr: "store-1", workers: []*regionRequestWorker{worker}} + scheduler.workerPools.Store(workerPool.storeAddr, workerPool) dummyRegion := regionInfo{ subscribedSpan: &subscribedSpan{subID: SubscriptionID(2)}, @@ -476,10 +476,10 @@ func TestBroadcastDeregisterUsesWorkerControlQueue(t *testing.T) { require.Equal(t, 1, admission.stats().pending) } -func TestRequestedStoreDistributesRegionsAcrossWorkerBuffers(t *testing.T) { +func TestStoreRequestWorkerPoolDistributesRegionsAcrossWorkers(t *testing.T) { worker1 := ®ionRequestWorker{admission: newRegionAdmissionController(1, 1)} worker2 := ®ionRequestWorker{admission: newRegionAdmissionController(1, 1)} - store := &requestedStore{ + workerPool := &storeRequestWorkerPool{ storeAddr: "store-1", workers: []*regionRequestWorker{worker1, worker2}, } @@ -490,7 +490,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, workerPool.submit(NewRegionPriorityTask(region, 1, i))) } require.Equal(t, 2, worker1.admission.stats().pending) @@ -526,11 +526,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..af94bfe89a 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, ) From ee875c38296069f69323760478c7f30bf7ac4d21 Mon Sep 17 00:00:00 2001 From: lidezhu Date: Fri, 17 Jul 2026 14:30:10 +0800 Subject: [PATCH 4/6] refactor --- .../logpuller/region_request_scheduler.go | 44 +++++++++---------- ...worker_pool.go => region_request_store.go} | 28 ++++++------ .../logpuller/subscription_client_test.go | 14 +++--- 3 files changed, 43 insertions(+), 43 deletions(-) rename logservice/logpuller/{store_request_worker_pool.go => region_request_store.go} (67%) diff --git a/logservice/logpuller/region_request_scheduler.go b/logservice/logpuller/region_request_scheduler.go index f6bb45e21d..2de2322128 100644 --- a/logservice/logpuller/region_request_scheduler.go +++ b/logservice/logpuller/region_request_scheduler.go @@ -44,9 +44,9 @@ type regionRequestScheduler struct { taskQueue *priorityqueue.PriorityQueue[*regionPriorityTask] // sequence is the FIFO tie-breaker for Regions in the same priority class. sequence atomic.Uint64 - // workerPools maps TiKV addresses to storeRequestWorkerPool. Pools are - // created only by run, but are also read by metrics and deregistration goroutines. - workerPools sync.Map + // 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 @@ -81,7 +81,7 @@ func (s *regionRequestScheduler) submit(region regionInfo) { } func (s *regionRequestScheduler) run(ctx context.Context, workerGroup *errgroup.Group) error { - defer s.closeWorkerPools() + defer s.closeStores() for { select { case <-ctx.Done(): @@ -103,9 +103,9 @@ func (s *regionRequestScheduler) run(ctx context.Context, workerGroup *errgroup. continue } - workerPool := s.getOrCreateWorkerPool(ctx, workerGroup, region.rpcCtx.Addr) + store := s.getOrCreateStore(ctx, workerGroup, region.rpcCtx.Addr) task.updateRegion(region, s.upstream.pdClock.CurrentTS()) - if !workerPool.submit(task) { + if !store.submit(task) { return context.Canceled } } @@ -131,16 +131,16 @@ func (s *regionRequestScheduler) attachRPCContext( return region, &rpcCtxUnavailableErr{verID: region.verID} } -func (s *regionRequestScheduler) getOrCreateWorkerPool( +func (s *regionRequestScheduler) getOrCreateStore( ctx context.Context, workerGroup *errgroup.Group, storeAddr string, -) *storeRequestWorkerPool { - if value, ok := s.workerPools.Load(storeAddr); ok { - return value.(*storeRequestWorkerPool) +) *regionRequestStore { + if value, ok := s.stores.Load(storeAddr); ok { + return value.(*regionRequestStore) } - workerPool := newStoreRequestWorkerPool( + store := newRegionRequestStore( s.upstream, s.eventSink, s.failureHandler, @@ -149,27 +149,27 @@ func (s *regionRequestScheduler) getOrCreateWorkerPool( s.workerWindow, s.maxWindowMultiplier, ) - // The scheduler run loop is the only writer. Publish the pool after its + // The scheduler run loop is the only writer. Publish the store after its // immutable worker list is complete, then start its workers. - s.workerPools.Store(storeAddr, workerPool) - workerPool.startWorkers(ctx, workerGroup) - return workerPool + s.stores.Store(storeAddr, store) + store.startWorkers(ctx, workerGroup) + return store } func (s *regionRequestScheduler) broadcastDeregister( subID SubscriptionID, filterLoop bool, ) { - s.workerPools.Range(func(_, value any) bool { - value.(*storeRequestWorkerPool).broadcastDeregister(subID, filterLoop) + s.stores.Range(func(_, value any) bool { + value.(*regionRequestStore).broadcastDeregister(subID, filterLoop) return true }) } func (s *regionRequestScheduler) inflightCount() int { count := 0 - s.workerPools.Range(func(_, value any) bool { - count += value.(*storeRequestWorkerPool).inflightCount() + s.stores.Range(func(_, value any) bool { + count += value.(*regionRequestStore).inflightCount() return true }) return count @@ -179,9 +179,9 @@ func (s *regionRequestScheduler) close() { s.taskQueue.Close() } -func (s *regionRequestScheduler) closeWorkerPools() { - s.workerPools.Range(func(_, value any) bool { - value.(*storeRequestWorkerPool).close() +func (s *regionRequestScheduler) closeStores() { + s.stores.Range(func(_, value any) bool { + value.(*regionRequestStore).close() return true }) } diff --git a/logservice/logpuller/store_request_worker_pool.go b/logservice/logpuller/region_request_store.go similarity index 67% rename from logservice/logpuller/store_request_worker_pool.go rename to logservice/logpuller/region_request_store.go index 6f84e0019a..92c44d91eb 100644 --- a/logservice/logpuller/store_request_worker_pool.go +++ b/logservice/logpuller/region_request_store.go @@ -21,16 +21,16 @@ import ( "golang.org/x/sync/errgroup" ) -// storeRequestWorkerPool owns the request workers connected to one TiKV store. -// The worker slice is complete before the pool is published and is immutable -// afterwards, so task submission only needs an atomic round-robin counter. -type storeRequestWorkerPool struct { +// 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 { storeAddr string workers []*regionRequestWorker nextWorker atomic.Uint64 } -func newStoreRequestWorkerPool( +func newRegionRequestStore( upstream *upstreamHandle, eventSink *regionEventSink, failureHandler *regionFailureHandler, @@ -38,25 +38,25 @@ func newStoreRequestWorkerPool( workerCount int, workerWindow int, maxWindowMultiplier int, -) *storeRequestWorkerPool { - workerPool := &storeRequestWorkerPool{ +) *regionRequestStore { + store := ®ionRequestStore{ storeAddr: storeAddr, workers: make([]*regionRequestWorker, 0, workerCount), } for i := 0; i < workerCount; i++ { - workerPool.workers = append(workerPool.workers, newRegionRequestWorker( + store.workers = append(store.workers, newRegionRequestWorker( upstream, eventSink, failureHandler, storeAddr, workerWindow, maxWindowMultiplier)) } - return workerPool + return store } -func (s *storeRequestWorkerPool) startWorkers(ctx context.Context, workerGroup *errgroup.Group) { +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 *storeRequestWorkerPool) submit(task *regionPriorityTask) bool { +func (s *regionRequestStore) submit(task *regionPriorityTask) bool { if len(s.workers) == 0 { return false } @@ -64,19 +64,19 @@ func (s *storeRequestWorkerPool) submit(task *regionPriorityTask) bool { return s.workers[index].admission.submit(task) } -func (s *storeRequestWorkerPool) broadcastDeregister(subID SubscriptionID, filterLoop bool) { +func (s *regionRequestStore) broadcastDeregister(subID SubscriptionID, filterLoop bool) { for _, worker := range s.workers { worker.controlQueue.push(deregisterRequest{subID: subID, filterLoop: filterLoop}) } } -func (s *storeRequestWorkerPool) close() { +func (s *regionRequestStore) close() { for _, worker := range s.workers { worker.admission.close() } } -func (s *storeRequestWorkerPool) inflightCount() int { +func (s *regionRequestStore) inflightCount() int { count := 0 for _, worker := range s.workers { count += worker.admission.stats().inflight diff --git a/logservice/logpuller/subscription_client_test.go b/logservice/logpuller/subscription_client_test.go index 1a939d50d8..23d6f81fe6 100644 --- a/logservice/logpuller/subscription_client_test.go +++ b/logservice/logpuller/subscription_client_test.go @@ -328,9 +328,9 @@ 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()} - workerPool := &storeRequestWorkerPool{storeAddr: "store-1", workers: []*regionRequestWorker{worker}} + store := ®ionRequestStore{storeAddr: "store-1", workers: []*regionRequestWorker{worker}} client.regionScheduler = ®ionRequestScheduler{} - client.regionScheduler.workerPools.Store(workerPool.storeAddr, workerPool) + client.regionScheduler.stores.Store(store.storeAddr, store) client.setTableStopped(span) @@ -458,8 +458,8 @@ func TestBroadcastDeregisterUsesWorkerControlQueue(t *testing.T) { admission: admission, controlQueue: newControlQueue(), } - workerPool := &storeRequestWorkerPool{storeAddr: "store-1", workers: []*regionRequestWorker{worker}} - scheduler.workerPools.Store(workerPool.storeAddr, workerPool) + store := ®ionRequestStore{storeAddr: "store-1", workers: []*regionRequestWorker{worker}} + scheduler.stores.Store(store.storeAddr, store) dummyRegion := regionInfo{ subscribedSpan: &subscribedSpan{subID: SubscriptionID(2)}, @@ -476,10 +476,10 @@ func TestBroadcastDeregisterUsesWorkerControlQueue(t *testing.T) { require.Equal(t, 1, admission.stats().pending) } -func TestStoreRequestWorkerPoolDistributesRegionsAcrossWorkers(t *testing.T) { +func TestRegionRequestStoreDistributesRegionsAcrossWorkers(t *testing.T) { worker1 := ®ionRequestWorker{admission: newRegionAdmissionController(1, 1)} worker2 := ®ionRequestWorker{admission: newRegionAdmissionController(1, 1)} - workerPool := &storeRequestWorkerPool{ + store := ®ionRequestStore{ storeAddr: "store-1", workers: []*regionRequestWorker{worker1, worker2}, } @@ -490,7 +490,7 @@ func TestStoreRequestWorkerPoolDistributesRegionsAcrossWorkers(t *testing.T) { subscribedSpan: &subscribedSpan{subID: 1}, lockedRangeState: ®ionlock.LockedRangeState{}, } - require.True(t, workerPool.submit(NewRegionPriorityTask(region, 1, i))) + require.True(t, store.submit(NewRegionPriorityTask(region, 1, i))) } require.Equal(t, 2, worker1.admission.stats().pending) From 4bd7b95724d31c8473828b7cdfa8f7af73ad12cb Mon Sep 17 00:00:00 2001 From: lidezhu Date: Fri, 17 Jul 2026 16:24:56 +0800 Subject: [PATCH 5/6] refactor --- .../logpuller/region_request_scheduler.go | 24 ++++++++++++------- logservice/logpuller/region_request_store.go | 5 +--- logservice/logpuller/subscription_client.go | 23 +++++++----------- .../logpuller/subscription_client_test.go | 2 +- pkg/metrics/log_puller.go | 2 +- 5 files changed, 26 insertions(+), 30 deletions(-) diff --git a/logservice/logpuller/region_request_scheduler.go b/logservice/logpuller/region_request_scheduler.go index 2de2322128..b7e3eba394 100644 --- a/logservice/logpuller/region_request_scheduler.go +++ b/logservice/logpuller/region_request_scheduler.go @@ -22,6 +22,7 @@ import ( "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" @@ -31,8 +32,8 @@ import ( const regionRequestWorkerPerStore = 8 -// regionRequestScheduler routes locked Region requests through the global -// priority queue to a worker connected to the Region's TiKV store. Range +// 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 { @@ -40,12 +41,12 @@ type regionRequestScheduler struct { eventSink *regionEventSink failureHandler *regionFailureHandler - // taskQueue orders all Regions before they are assigned to a TiKV store. + // 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 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. + // by Run, but are also read by metrics and deregistration goroutines. stores sync.Map // workerCount is the configured number of request workers per store. @@ -75,12 +76,12 @@ func newRegionRequestScheduler( } } -func (s *regionRequestScheduler) submit(region regionInfo) { +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 { +func (s *regionRequestScheduler) Run(ctx context.Context, workerGroup *errgroup.Group) error { defer s.closeStores() for { select { @@ -156,7 +157,7 @@ func (s *regionRequestScheduler) getOrCreateStore( return store } -func (s *regionRequestScheduler) broadcastDeregister( +func (s *regionRequestScheduler) BroadcastDeregister( subID SubscriptionID, filterLoop bool, ) { @@ -175,7 +176,12 @@ func (s *regionRequestScheduler) inflightCount() int { return count } -func (s *regionRequestScheduler) close() { +func (s *regionRequestScheduler) UpdateMetrics() { + metrics.SubscriptionClientRequestedRegionCount.WithLabelValues("inflight"). + Set(float64(s.inflightCount())) +} + +func (s *regionRequestScheduler) Close() { s.taskQueue.Close() } diff --git a/logservice/logpuller/region_request_store.go b/logservice/logpuller/region_request_store.go index 92c44d91eb..1a8d243e0d 100644 --- a/logservice/logpuller/region_request_store.go +++ b/logservice/logpuller/region_request_store.go @@ -21,7 +21,7 @@ import ( "golang.org/x/sync/errgroup" ) -// regionRequestStore owns the Region request workers connected to one TiKV +// 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 { @@ -57,9 +57,6 @@ func (s *regionRequestStore) startWorkers(ctx context.Context, workerGroup *errg } func (s *regionRequestStore) submit(task *regionPriorityTask) bool { - if len(s.workers) == 0 { - return false - } index := (s.nextWorker.Add(1) - 1) % uint64(len(s.workers)) return s.workers[index].admission.submit(task) } diff --git a/logservice/logpuller/subscription_client.go b/logservice/logpuller/subscription_client.go index 0b629d2fa7..924025a386 100644 --- a/logservice/logpuller/subscription_client.go +++ b/logservice/logpuller/subscription_client.go @@ -94,7 +94,7 @@ type upstreamHandle struct { clusterID uint64 } -// initialize loads the cluster metadata needed by Region request workers. It +// 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) @@ -133,7 +133,7 @@ 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 assigns locked region requests to per-store workers. regionScheduler *regionRequestScheduler // rangeTaskCh is used to receive range tasks. @@ -198,9 +198,7 @@ func (s *subscriptionClient) updateMetrics(ctx context.Context) error { case <-ctx.Done(): return ctx.Err() case <-ticker.C: - inflightRegionRequestCount := s.regionScheduler.inflightCount() - - metrics.SubscriptionClientRequestedRegionCount.WithLabelValues("inflight").Set(float64(inflightRegionRequestCount)) + s.regionScheduler.UpdateMetrics() s.eventSink.UpdateMetrics() s.spanRegistry.UpdateMetrics() } @@ -268,11 +266,6 @@ func (s *subscriptionClient) Unsubscribe(subID SubscriptionID) { } func (s *subscriptionClient) Run(ctx context.Context) error { - // s.consume = consume - if s.upstream == nil || s.upstream.pd == nil { - log.Warn("subscription client should be in test mode, skip run") - return nil - } s.upstream.initialize(ctx) g, ctx := errgroup.WithContext(ctx) @@ -280,7 +273,7 @@ func (s *subscriptionClient) Run(ctx context.Context) error { g.Go(func() error { return s.updateMetrics(ctx) }) g.Go(func() error { return s.eventSink.Run(ctx) }) g.Go(func() error { return s.handleRangeTasks(ctx) }) - g.Go(func() error { return s.regionScheduler.run(ctx, g) }) + g.Go(func() error { return s.regionScheduler.Run(ctx, g) }) 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) }) @@ -294,7 +287,7 @@ func (s *subscriptionClient) Run(ctx context.Context) error { func (s *subscriptionClient) Close(ctx context.Context) error { s.cancel() s.eventSink.Close() - s.regionScheduler.close() + s.regionScheduler.Close() return nil } @@ -305,7 +298,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.regionScheduler.broadcastDeregister(rt.subID, rt.filterLoop) + s.regionScheduler.BroadcastDeregister(rt.subID, rt.filterLoop) if rt.rangeLock.Stop() { s.onTableDrained(rt) } @@ -429,7 +422,7 @@ func (s *subscriptionClient) divideSpanAndScheduleRegionRequests( } } -// scheduleRegionRequest locks the Region's range before submitting it to the +// 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() { @@ -445,7 +438,7 @@ func (s *subscriptionClient) scheduleRegionRequest(ctx context.Context, region r switch lockRangeResult.Status { case regionlock.LockRangeStatusSuccess: region.lockedRangeState = lockRangeResult.LockedRangeState - s.regionScheduler.submit(region) + s.regionScheduler.Submit(region) case regionlock.LockRangeStatusStale: for _, r := range lockRangeResult.RetryRanges { s.scheduleRangeRequest(ctx, r, region.subscribedSpan, region.filterLoop, region.wasInitialized) diff --git a/logservice/logpuller/subscription_client_test.go b/logservice/logpuller/subscription_client_test.go index 23d6f81fe6..ef1a387045 100644 --- a/logservice/logpuller/subscription_client_test.go +++ b/logservice/logpuller/subscription_client_test.go @@ -467,7 +467,7 @@ func TestBroadcastDeregisterUsesWorkerControlQueue(t *testing.T) { } require.True(t, admission.submit(NewRegionPriorityTask(dummyRegion, 1, 1))) - scheduler.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) diff --git a/pkg/metrics/log_puller.go b/pkg/metrics/log_puller.go index af94bfe89a..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 Region requests by state.", + Help: "The number of region requests by state.", }, []string{"state"}) RegionRequestFinishScanDuration = prometheus.NewHistogram( prometheus.HistogramOpts{ From 9903a1be7b910f20f23138bbdeac77691700ee40 Mon Sep 17 00:00:00 2001 From: lidezhu Date: Fri, 17 Jul 2026 16:41:06 +0800 Subject: [PATCH 6/6] refactor --- ...mock_upstream.go => mock_upstream_test.go} | 0 logservice/logpuller/priority_task.go | 8 +--- logservice/logpuller/priority_task_test.go | 18 ++++---- .../region_admission_controller_test.go | 5 +-- .../logpuller/region_event_handler_test.go | 1 - .../logpuller/region_failure_handler.go | 21 ++++++---- .../logpuller/region_request_scheduler.go | 4 +- logservice/logpuller/region_request_store.go | 4 +- logservice/logpuller/region_request_worker.go | 6 +-- .../logpuller/region_request_worker_test.go | 2 +- logservice/logpuller/region_state.go | 3 +- logservice/logpuller/region_tracker.go | 16 ++++--- logservice/logpuller/region_tracker_test.go | 4 +- logservice/logpuller/subscription_client.go | 42 +++++++++---------- .../logpuller/subscription_client_test.go | 20 +++++---- 15 files changed, 77 insertions(+), 77 deletions(-) rename logservice/logpuller/{mock_upstream.go => mock_upstream_test.go} (100%) 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 f90483b740..7464afb940 100644 --- a/logservice/logpuller/region_failure_handler.go +++ b/logservice/logpuller/region_failure_handler.go @@ -19,7 +19,6 @@ import ( "time" "github.com/pingcap/log" - "github.com/pingcap/ticdc/heartbeatpb" "github.com/pingcap/ticdc/pkg/errors" "github.com/pingcap/ticdc/pkg/metrics" "github.com/tikv/client-go/v2/tikv" @@ -47,14 +46,14 @@ type regionFailureHandler struct { onTableDrained func(*subscribedSpan) scheduleRegionRequest func(context.Context, regionInfo) - scheduleRangeRequest func(context.Context, heartbeatpb.TableSpan, *subscribedSpan, bool, bool) + scheduleRangeRequest func(context.Context, rangeTask) } func newRegionFailureHandler( regionCache *tikv.RegionCache, onTableDrained func(*subscribedSpan), scheduleRegionRequest func(context.Context, regionInfo), - scheduleRangeRequest func(context.Context, heartbeatpb.TableSpan, *subscribedSpan, bool, bool), + scheduleRangeRequest func(context.Context, rangeTask), ) *regionFailureHandler { return ®ionFailureHandler{ cache: newErrCache(), @@ -65,6 +64,14 @@ func newRegionFailureHandler( } } +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. @@ -118,12 +125,12 @@ func (r *regionFailureHandler) handleError(ctx context.Context, errInfo regionEr } if innerErr.GetEpochNotMatch() != nil { metricFeedEpochNotMatchCounter.Inc() - r.scheduleRangeRequest(ctx, errInfo.span, errInfo.subscribedSpan, errInfo.filterLoop, errInfo.wasInitialized) + r.retryRange(ctx, errInfo) return nil } if innerErr.GetRegionNotFound() != nil { metricFeedRegionNotFoundCounter.Inc() - r.scheduleRangeRequest(ctx, errInfo.span, errInfo.subscribedSpan, errInfo.filterLoop, errInfo.wasInitialized) + r.retryRange(ctx, errInfo) return nil } if innerErr.GetCongested() != nil { @@ -156,14 +163,14 @@ func (r *regionFailureHandler) handleError(ctx context.Context, errInfo regionEr return nil case *rpcCtxUnavailableErr: metricFeedRPCCtxUnavailable.Inc() - r.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.regionCache.OnSendFail(bo, errInfo.rpcCtx, true, err) - r.scheduleRangeRequest(ctx, errInfo.span, errInfo.subscribedSpan, errInfo.filterLoop, errInfo.wasInitialized) + r.retryRange(ctx, errInfo) return nil case *storeStreamErr: metricStoreSendRequestErr.Inc() diff --git a/logservice/logpuller/region_request_scheduler.go b/logservice/logpuller/region_request_scheduler.go index b7e3eba394..f75fbf7868 100644 --- a/logservice/logpuller/region_request_scheduler.go +++ b/logservice/logpuller/region_request_scheduler.go @@ -77,7 +77,7 @@ func newRegionRequestScheduler( } func (s *regionRequestScheduler) Submit(region regionInfo) { - s.taskQueue.Push(NewRegionPriorityTask( + s.taskQueue.Push(newRegionPriorityTask( region, s.upstream.pdClock.CurrentTS(), s.sequence.Add(1))) } @@ -98,7 +98,7 @@ func (s *regionRequestScheduler) Run(ctx context.Context, workerGroup *errgroup. return err } - region, err := s.attachRPCContext(ctx, task.GetRegionInfo()) + region, err := s.attachRPCContext(ctx, task.regionInfo) if err != nil { s.failureHandler.Report(newRegionErrorInfo(region, err)) continue diff --git a/logservice/logpuller/region_request_store.go b/logservice/logpuller/region_request_store.go index 1a8d243e0d..9b9a5af025 100644 --- a/logservice/logpuller/region_request_store.go +++ b/logservice/logpuller/region_request_store.go @@ -25,7 +25,6 @@ import ( // 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 { - storeAddr string workers []*regionRequestWorker nextWorker atomic.Uint64 } @@ -40,8 +39,7 @@ func newRegionRequestStore( maxWindowMultiplier int, ) *regionRequestStore { store := ®ionRequestStore{ - storeAddr: storeAddr, - workers: make([]*regionRequestWorker, 0, workerCount), + workers: make([]*regionRequestWorker, 0, workerCount), } for i := 0; i < workerCount; i++ { store.workers = append(store.workers, newRegionRequestWorker( diff --git a/logservice/logpuller/region_request_worker.go b/logservice/logpuller/region_request_worker.go index 298e2cc266..e61303a644 100644 --- a/logservice/logpuller/region_request_worker.go +++ b/logservice/logpuller/region_request_worker.go @@ -159,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() diff --git a/logservice/logpuller/region_request_worker_test.go b/logservice/logpuller/region_request_worker_test.go index 9b75667040..c6e3f8d845 100644 --- a/logservice/logpuller/region_request_worker_test.go +++ b/logservice/logpuller/region_request_worker_test.go @@ -393,7 +393,7 @@ 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{}) 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 924025a386..ea4cd33c31 100644 --- a/logservice/logpuller/subscription_client.go +++ b/logservice/logpuller/subscription_client.go @@ -80,7 +80,6 @@ type resolveLockTask struct { type rangeTask struct { span heartbeatpb.TableSpan subscribedSpan *subscribedSpan - filterLoop bool wasInitialized bool } @@ -242,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))) @@ -270,13 +269,15 @@ func (s *subscriptionClient) Run(ctx context.Context) error { 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.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") @@ -328,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) }) } } @@ -341,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 @@ -406,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) @@ -441,7 +442,11 @@ func (s *subscriptionClient) scheduleRegionRequest(ctx context.Context, region r 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 @@ -449,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 ef1a387045..474d3c6af5 100644 --- a/logservice/logpuller/subscription_client_test.go +++ b/logservice/logpuller/subscription_client_test.go @@ -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 := ®ionRequestStore{storeAddr: "store-1", workers: []*regionRequestWorker{worker}} + const storeAddr = "store-1" + worker := ®ionRequestWorker{storeAddr: storeAddr, controlQueue: newControlQueue()} + store := ®ionRequestStore{workers: []*regionRequestWorker{worker}} client.regionScheduler = ®ionRequestScheduler{} - client.regionScheduler.stores.Store(store.storeAddr, store) + client.regionScheduler.stores.Store(storeAddr, store) client.setTableStopped(span) @@ -454,18 +455,20 @@ func TestBroadcastDeregisterUsesWorkerControlQueue(t *testing.T) { scheduler := ®ionRequestScheduler{} admission := newRegionAdmissionController(1, 1) + const storeAddr = "store-1" worker := ®ionRequestWorker{ + storeAddr: storeAddr, admission: admission, controlQueue: newControlQueue(), } - store := ®ionRequestStore{storeAddr: "store-1", workers: []*regionRequestWorker{worker}} - scheduler.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))) scheduler.BroadcastDeregister(SubscriptionID(1), true) require.Equal(t, 1, worker.controlQueue.len()) @@ -480,8 +483,7 @@ func TestRegionRequestStoreDistributesRegionsAcrossWorkers(t *testing.T) { worker1 := ®ionRequestWorker{admission: newRegionAdmissionController(1, 1)} worker2 := ®ionRequestWorker{admission: newRegionAdmissionController(1, 1)} store := ®ionRequestStore{ - storeAddr: "store-1", - workers: []*regionRequestWorker{worker1, worker2}, + workers: []*regionRequestWorker{worker1, worker2}, } for i := uint64(1); i <= 4; i++ { @@ -490,7 +492,7 @@ func TestRegionRequestStoreDistributesRegionsAcrossWorkers(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)