diff --git a/pkg/schedule/checker/checker_controller.go b/pkg/schedule/checker/checker_controller.go index 7da5ae7bb49..dd8453c805d 100644 --- a/pkg/schedule/checker/checker_controller.go +++ b/pkg/schedule/checker/checker_controller.go @@ -83,7 +83,8 @@ type Controller struct { // duration is the duration of the last patrol round. // It's exported, so it should be protected by a mutex. - duration atomic.Value // Store as time.Duration + duration atomic.Value // Store as time.Duration + operatorMu sync.Mutex // interval is the config interval of patrol regions. // It's used to update the ticker, so we need to @@ -281,20 +282,29 @@ func (c *Controller) checkPriorityRegions() { removes = append(removes, id) continue } - ops := c.CheckRegion(region) - // it should skip if region needs to merge - if len(ops) == 0 || ops[0].HasRelatedMergeRegion() { - continue - } - if !c.opController.ExceedStoreLimit(ops...) { - c.opController.AddWaitingOperator(ops...) - } + c.addPriorityOperator(region) } for _, v := range removes { c.RemovePriorityRegions(v) } } +func (c *Controller) addPriorityOperator(region *core.RegionInfo) { + c.operatorMu.Lock() + defer c.operatorMu.Unlock() + if c.cluster.IsSchedulingHalted() { + return + } + ops := c.CheckRegion(region) + // It should skip if region needs to merge. + if len(ops) == 0 || ops[0].HasRelatedMergeRegion() { + return + } + if !c.opController.ExceedStoreLimit(ops...) { + c.opController.AddWaitingOperator(ops...) + } +} + // CheckRegion will check the region and add a new operator if needed. // The function is exposed for test purpose. func (c *Controller) CheckRegion(region *core.RegionInfo) []*operator.Operator { @@ -368,9 +378,13 @@ func (c *Controller) CheckRegion(region *core.RegionInfo) []*operator.Operator { } if ops := measureChecker(c.metrics.checkRegionHistograms[affinityChecker], func() []*operator.Operator { - if opController.OperatorCount(operator.OpAffinity) < c.conf.GetAffinityScheduleLimit() { - // It makes sure that two affinity merge operators can be added successfully altogether. - return c.affinityChecker.Check(region) + current := opController.OperatorCount(operator.OpAffinity) + limit := c.conf.GetAffinityScheduleLimit() + if current < limit { + ops := c.affinityChecker.Check(region) + if hasOperatorCapacity(current, limit, uint64(len(ops))) { + return ops + } } if c.affinityChecker.hasAffinityGroups() { operator.IncOperatorLimitCounter(c.affinityChecker.GetType(), operator.OpAffinity) @@ -381,9 +395,13 @@ func (c *Controller) CheckRegion(region *core.RegionInfo) []*operator.Operator { } if ops := measureChecker(c.metrics.checkRegionHistograms[mergeChecker], func() []*operator.Operator { - if opController.OperatorCount(operator.OpMerge) < c.conf.GetMergeScheduleLimit() { - // It makes sure that two operators can be added successfully altogether. - return c.mergeChecker.Check(region) + current := opController.OperatorCount(operator.OpMerge) + limit := c.conf.GetMergeScheduleLimit() + if current < limit { + ops := c.mergeChecker.Check(region) + if hasOperatorCapacity(current, limit, uint64(len(ops))) { + return ops + } } operator.IncOperatorLimitCounter(c.mergeChecker.GetType(), operator.OpMerge) return nil @@ -393,6 +411,10 @@ func (c *Controller) CheckRegion(region *core.RegionInfo) []*operator.Operator { return nil } +func hasOperatorCapacity(current, limit, candidateCount uint64) bool { + return current < limit && candidateCount <= limit-current +} + func (c *Controller) tryAddOperators(region *core.RegionInfo) { if region == nil { // the region could be recent split, continue to wait. @@ -403,6 +425,15 @@ func (c *Controller) tryAddOperators(region *core.RegionInfo) { c.RemovePendingProcessedRegion(id) return } + c.operatorMu.Lock() + defer c.operatorMu.Unlock() + if c.cluster.IsSchedulingHalted() { + return + } + if c.opController.GetOperator(id) != nil { + c.RemovePendingProcessedRegion(id) + return + } ops := c.CheckRegion(region) if len(ops) == 0 { return diff --git a/pkg/schedule/checker/merge_checker_test.go b/pkg/schedule/checker/merge_checker_test.go index 80a72238ee4..a1ff1386c95 100644 --- a/pkg/schedule/checker/merge_checker_test.go +++ b/pkg/schedule/checker/merge_checker_test.go @@ -20,6 +20,7 @@ import ( "testing" "time" + "github.com/stretchr/testify/require" "github.com/stretchr/testify/suite" "go.uber.org/goleak" @@ -56,6 +57,61 @@ func TestMergeCheckerTestSuite(t *testing.T) { suite.Run(t, new(mergeCheckerTestSuite)) } +func TestTryAddOperatorsRejectsMergePairPastLimit(t *testing.T) { + re := require.New(t) + ctx, cancel := context.WithCancel(context.Background()) + defer cancel() + + opt := mockconfig.NewTestOptions() + tc := mockcluster.NewCluster(ctx, opt) + tc.SetMergeScheduleLimit(3) + tc.SetSplitMergeInterval(0) + for storeID := uint64(1); storeID <= 3; storeID++ { + tc.AddRegionStore(storeID, 0) + } + tc.AddLeaderRegionWithRange(1, "", "a", 1, 2, 3) + tc.AddLeaderRegionWithRange(2, "a", "b", 1, 2, 3) + tc.AddLeaderRegionWithRange(3, "b", "c", 1, 2, 3) + tc.AddLeaderRegionWithRange(4, "c", "d", 1, 2, 3) + + stream := hbstream.NewTestHeartbeatStreams(ctx, tc, false) + defer stream.Close() + oc := operator.NewController(ctx, tc.GetBasicCluster(), tc.GetSharedConfig(), stream) + controller := NewController(ctx, tc, tc.GetCheckerConfig(), oc) + + controller.tryAddOperators(tc.GetRegion(1)) + re.Equal(uint64(2), oc.OperatorCount(operator.OpMerge)) + + controller.tryAddOperators(tc.GetRegion(3)) + re.LessOrEqual(oc.OperatorCount(operator.OpMerge), uint64(3)) +} + +func TestHasOperatorCapacity(t *testing.T) { + testCases := []struct { + name string + current uint64 + limit uint64 + candidateCount uint64 + expected bool + }{ + {name: "no capacity", current: 0, limit: 0, candidateCount: 1, expected: false}, + {name: "single candidate fits", current: 2, limit: 3, candidateCount: 1, expected: true}, + {name: "pair fits", current: 2, limit: 4, candidateCount: 2, expected: true}, + {name: "pair exceeds remaining capacity", current: 2, limit: 3, candidateCount: 2, expected: false}, + {name: "current exceeds limit", current: 4, limit: 3, candidateCount: 1, expected: false}, + } + + for _, testCase := range testCases { + t.Run(testCase.name, func(t *testing.T) { + require.Equal(t, testCase.expected, hasOperatorCapacity( + testCase.current, + testCase.limit, + testCase.candidateCount, + )) + }) + } +} + func (suite *mergeCheckerTestSuite) SetupTest() { cfg := mockconfig.NewTestOptions() gcInterval = 100 * time.Millisecond diff --git a/server/cluster/cluster_test.go b/server/cluster/cluster_test.go index b8080d7a45f..f492118cee9 100644 --- a/server/cluster/cluster_test.go +++ b/server/cluster/cluster_test.go @@ -3195,6 +3195,51 @@ func TestPatrolRegionConcurrency(t *testing.T) { re.NoError(failpoint.Disable("github.com/tikv/pd/pkg/schedule/checker/breakPatrol")) } +func TestPatrolRegionConcurrencyRespectReplicaLimit(t *testing.T) { + re := require.New(t) + + const ( + regionNum = 128 + replicaScheduleLimit = 1 + ) + + tc, co, cleanup := prepare(func(cfg *sc.ScheduleConfig) { + cfg.PatrolRegionWorkerCount = 8 + cfg.ReplicaScheduleLimit = replicaScheduleLimit + }, nil, nil, re) + oc := co.GetOperatorController() + checker := co.GetCheckerController() + done := make(chan struct{}) + defer func() { + cleanup() + select { + case <-done: + case <-time.After(time.Second): + re.Fail("patrol regions did not stop") + } + }() + + for i := range 3 { + re.NoError(tc.addRegionStore(uint64(i+1), regionNum)) + } + for i := range regionNum { + re.NoError(tc.addLeaderRegion(uint64(i+1), 1)) + } + + go func() { + checker.PatrolRegions() + close(done) + }() + + testutil.Eventually(re, func() bool { + count := oc.OperatorCount(operator.OpReplica) + return count > uint64(replicaScheduleLimit) || + (count == uint64(replicaScheduleLimit) && len(checker.GetPendingProcessedRegions()) > 0) + }) + re.LessOrEqual(oc.OperatorCount(operator.OpReplica), uint64(replicaScheduleLimit)) + re.Len(oc.GetOperators(), replicaScheduleLimit) +} + func checkOperatorDuplicate(re *require.Assertions, ops []*operator.Operator) { regionMap := make(map[uint64]struct{}) for _, op := range ops {