checker: refine split scatter pending collection

Signed-off-by: lhy1024 <admin@liudos.us>

Author: lhy1024

pkg/schedule/checker/split_scatter.go

@@ -34,6 +34,9 @@ import (
 
 const (
 	splitScatterPendingLimit = 1024
+	// The actual retry cadence is also bounded by the checker dispatch loop. This
+	// is only the minimum interval to avoid retrying the same pending item too
+	// frequently when checker ticks are fast.
 	splitScatterRetryBackoff = time.Second
 	splitScatterPendingTTL   = 3 * time.Minute
 )
@@ -53,7 +56,10 @@ type splitScatterController struct {
 	regionScatterer *scatter.RegionScatterer
 
 	pendingMu syncutil.RWMutex
-	pending   map[uint64]splitScatterPendingItem
+	// pending maps a pending region ID to its latest split-scatter batch item.
+	// The item keeps its batch group so stale snapshots cannot mutate a newer
+	// pending entry for the same region.
+	pending map[uint64]splitScatterPendingItem
 }
 
 func newSplitScatterController(
@@ -90,10 +96,10 @@ func (c *splitScatterController) collectTopPendingSplitScatter(limit int) []spli
 	// to block pending updates. Stale snapshots are safe because delay/delete
 	// recheck regionID + group before mutating pending.
 	pendingSnapshot := make([]splitScatterPendingItem, 0, len(c.pending))
-	expiredRegionIDs := make([]uint64, 0)
-	for regionID, pending := range c.pending {
+	expiredSnapshot := make([]splitScatterPendingItem, 0)
+	for _, pending := range c.pending {
 		if !pending.expireAt.IsZero() && !now.Before(pending.expireAt) {
-			expiredRegionIDs = append(expiredRegionIDs, regionID)
+			expiredSnapshot = append(expiredSnapshot, pending)
 			continue
 		}
 		pendingSnapshot = append(pendingSnapshot, pending)
@@ -124,6 +130,9 @@ func (c *splitScatterController) collectTopPendingSplitScatter(limit int) []spli
 		candidates = append(candidates, pending)
 	}
 	sort.Slice(candidates, func(i, j int) bool {
+		if !candidates[i].expireAt.Equal(candidates[j].expireAt) {
+			return candidates[i].expireAt.Before(candidates[j].expireAt)
+		}
 		if candidates[i].group != candidates[j].group {
 			return candidates[i].group < candidates[j].group
 		}
@@ -133,16 +142,13 @@ func (c *splitScatterController) collectTopPendingSplitScatter(limit int) []spli
 		candidates = candidates[:limit]
 	}
 
-	if len(expiredRegionIDs) > 0 {
+	if len(expiredSnapshot) > 0 {
 		expiredCount := 0
 		c.pendingMu.Lock()
-		for _, regionID := range expiredRegionIDs {
-			pending, ok := c.pending[regionID]
-			if !ok {
-				continue
-			}
-			if !pending.expireAt.IsZero() && !now.Before(pending.expireAt) {
-				delete(c.pending, regionID)
+		for _, expired := range expiredSnapshot {
+			pending, ok := c.pending[expired.regionID]
+			if ok && pending.group == expired.group && !pending.expireAt.IsZero() && !now.Before(pending.expireAt) {
+				delete(c.pending, expired.regionID)
 				expiredCount++
 			}
 		}
@@ -175,7 +181,8 @@ func (c *splitScatterController) deletePendingSplitScatter(expected splitScatter
 	delete(c.pending, expected.regionID)
 }
 
-func (c *splitScatterController) removeExpiredPendingSplitScatterLocked(now time.Time) int {
+func (c *splitScatterController) removeExpiredPendingSplitScatterLocked() int {
+	now := time.Now()
 	expiredCount := 0
 	for regionID, pending := range c.pending {
 		if !pending.expireAt.IsZero() && !now.Before(pending.expireAt) {
@@ -207,7 +214,7 @@ func (c *splitScatterController) recordSplitScatterBatch(sourceRegionID uint64,
 	}
 	c.pendingMu.Lock()
 	defer c.pendingMu.Unlock()
-	if expiredCount := c.removeExpiredPendingSplitScatterLocked(time.Now()); expiredCount > 0 {
+	if expiredCount := c.removeExpiredPendingSplitScatterLocked(); expiredCount > 0 {
 		splitScatterPendingExpiredCounter.Add(float64(expiredCount))
 	}
 	newPendingCount := 0
@@ -220,6 +227,9 @@ func (c *splitScatterController) recordSplitScatterBatch(sourceRegionID uint64,
 		}
 	}
 	if len(c.pending)+newPendingCount > splitScatterPendingLimit {
+		// Keep each split batch atomic. Recording only part of a source/child
+		// batch would make the scatter group incomplete, so skip the whole batch
+		// when the remaining capacity cannot fit it.
 		splitScatterPendingDroppedCounter.Add(float64(newPendingCount))
 		log.Info("skip recording split scatter batch due to pending limit",
 			zap.Uint64("source-region-id", sourceRegionID),

pkg/schedule/checker/split_scatter_test.go

@@ -229,6 +229,30 @@ func TestCollectTopPendingSortsBeforeLimit(t *testing.T) {
 	re.Equal(uint64(100), pending[0].regionID)
 }
 
+func TestCollectTopPendingPrioritizesNearExpiration(t *testing.T) {
+	re := require.New(t)
+	controller, tc, _, cleanup := newTestSplitScatterController(t)
+	defer cleanup()
+
+	controller.RecordSplitScatterBatch(100, []uint64{101})
+	controller.RecordSplitScatterBatch(200, []uint64{201})
+	putSplitScatterRegion(tc, 100, "m", "n", 0)
+	putSplitScatterRegion(tc, 101, "n", "o", 120)
+	putSplitScatterRegion(tc, 200, "o", "p", 0)
+	putSplitScatterRegion(tc, 201, "p", "q", 120)
+	advanceSplitScatterSourceVersion(t, tc)
+	advanceSplitScatterRegionVersion(t, tc, 200)
+
+	expireSplitScatterPendingAt(t, controller, 100, time.Now().Add(2*time.Minute))
+	expireSplitScatterPendingAt(t, controller, 101, time.Now().Add(2*time.Minute))
+	expireSplitScatterPendingAt(t, controller, 200, time.Now().Add(time.Minute))
+	expireSplitScatterPendingAt(t, controller, 201, time.Now().Add(time.Minute))
+
+	pending := controller.collectTopPendingSplitScatter(1)
+	re.Len(pending, 1)
+	re.Equal(uint64(200), pending[0].regionID)
+}
+
 func TestCollectTopPendingDefersBatchUntilSourceVersionAdvances(t *testing.T) {
 	re := require.New(t)
 	controller, tc, _, cleanup := newTestSplitScatterController(t)

pkg/schedule/scatter/region_scatterer.go

@@ -69,7 +69,7 @@ const (
 	// AdminScatterOperatorDesc is used by external admin/API scatter requests.
 	AdminScatterOperatorDesc = "scatter-region"
 	// InternalScatterOperatorDesc is used by PD-internal split-scatter dispatch.
-	InternalScatterOperatorDesc = "internal-scatter-region"
+	InternalScatterOperatorDesc = "split-scatter-region"
 )
 
 type selectedStoreCounter interface {