Skip to content

schedule: respect limits in concurrent patrol workers #10925

New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Open
rleungx wants to merge 9 commits into
base: master
Choose a base branch
from
+147 −15
Open

schedule: respect limits in concurrent patrol workers #10925

Show file tree
Hide file tree
Changes from all commits
Commits
Show all changes
9 commits
Select commit Hold shift + click to select a range
e48894e
schedule: respect limits in concurrent patrol workers
rleungx Jun 22, 2026
16c8144
schedule: clean up patrol operator admission
rleungx Jul 6, 2026
e946a88
tests: wait for affinity group visibility
rleungx Jul 6, 2026
71fe19b
tests: avoid unused affinity wait result
rleungx Jul 6, 2026
c8040fa
tests: avoid assertions inside affinity wait loop
rleungx Jul 6, 2026
7118142
schedule: report patrol pending work
rleungx Jul 9, 2026
919f8c8
schedule: clarify patrol operator checks
rleungx Jul 9, 2026
37d9c56
schedule: simplify patrol admission locking
rleungx Jul 9, 2026
8e7ec80
schedule: account for merge candidate count
rleungx Jul 10, 2026
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
61 changes: 46 additions & 15 deletions pkg/schedule/checker/checker_controller.go
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -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
Expand Down Expand Up @@ -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 {
Expand Down Expand Up @@ -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)
Expand All @@ -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
Expand All @@ -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.
Expand All @@ -403,6 +425,15 @@ func (c *Controller) tryAddOperators(region *core.RegionInfo) {
c.RemovePendingProcessedRegion(id)
return
}
c.operatorMu.Lock()

@lhy1024 lhy1024 Jul 9, 2026

Copy link
Copy Markdown
Member

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

The new coarse-grained operatorMu does fix the stale-count race between patrol workers, but I think the merge / affinity-merge boundary case is still not covered.

CheckRegion() still only checks OperatorCount(OpMerge) < GetMergeScheduleLimit() before returning a merge pair. Since a merge candidate contains two operators, merge-schedule-limit = 3 and current OperatorCount(OpMerge) = 2 can still admit another pair and end up with 4 merge operators.

I verified this on the latest head with the following focused test, which fails because the final merge count becomes 4:

package checker

import (
	"context"
	"testing"

	"github.com/stretchr/testify/require"

	"github.com/tikv/pd/pkg/mock/mockcluster"
	"github.com/tikv/pd/pkg/mock/mockconfig"
	"github.com/tikv/pd/pkg/schedule/hbstream"
	"github.com/tikv/pd/pkg/schedule/operator"
)

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))
}

Could we either account for the candidate size when checking merge / affinity limits, or explicitly document that these limits may be exceeded by one pair?

@rleungx rleungx Jul 10, 2026

Copy link
Copy Markdown
Member Author

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

Fixed in 8e7ec80. Merge and affinity candidates now check their full candidate count against the remaining schedule-limit capacity (candidateCount <= limit - current) before admission, so a pair is rejected when only one slot remains. I added the provided merge boundary regression plus focused capacity cases. Verified with the full checker package, the existing patrol replica-limit regression, focused race tests, and make static.

@lhy1024 lhy1024 Jul 10, 2026

Copy link
Copy Markdown
Member

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

I want to clarify my earlier suggestion here. The concern about merge candidates being different from single-operator candidates is still valid, but suggesting a strict current + candidateCount <= limit check was too strict if we want to preserve the existing merge-pair semantics.

The original issue this PR is fixing is the stale-count race between concurrent patrol workers. It is not necessarily trying to change the existing single-threaded behavior where one merge schedule produces two operators. Existing behavior/tests seem to allow one merge pair to be created when merge-schedule-limit = 1, and existing TestPatrolRegionConcurrency uses mergeScheduleLimit = 15 and waits for at least 15 operators.

With the current len(ops)-based capacity check, an odd merge limit can stop below the configured limit. For example, with merge-schedule-limit = 15, the merge checker can stop at 14 because the next merge pair has 2 operators. I verified this on the latest head by running:

make gotest GOTEST_ARGS='./server/cluster -run TestPatrolRegionConcurrency -count=1 -timeout=2m'

and TestPatrolRegionConcurrency fails because the len(oc.GetOperators()) >= mergeScheduleLimit condition is never satisfied.

Could we preserve the existing merge-pair semantics while still serializing admission to avoid the concurrent stale-count overshoot? In other words, the admission lock should prevent multiple workers from racing on the same old count, but we probably should not make odd merge limits reject a whole pair before reaching the configured limit.

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
Expand Down
56 changes: 56 additions & 0 deletions pkg/schedule/checker/merge_checker_test.go
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -20,6 +20,7 @@ import (
"testing"
"time"

"github.com/stretchr/testify/require"
"github.com/stretchr/testify/suite"
"go.uber.org/goleak"

Expand Down Expand Up @@ -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
Expand Down
45 changes: 45 additions & 0 deletions server/cluster/cluster_test.go
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -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 {
Expand Down
Loading