*: filter store with lots of pending peers

conf/config.toml

@@ -46,6 +46,7 @@ address = ""
 
 [schedule]
 max-snapshot-count = 3
+max-pending-peer-count = 16
 max-store-down-time = "1h"
 leader-schedule-limit = 64
 region-schedule-limit = 16

server/cache.go

@@ -339,6 +339,7 @@ func (c *clusterInfo) handleStoreHeartbeat(stats *pdpb.StoreStats) error {
 func (c *clusterInfo) updateStoreStatus(id uint64) {
 	c.Stores.SetLeaderCount(id, c.Regions.GetStoreLeaderCount(id))
 	c.Stores.SetRegionCount(id, c.Regions.GetStoreRegionCount(id))
+	c.Stores.SetPendingPeerCount(id, c.Regions.GetStorePendingPeerCount(id))
 }
 
 // handleRegionHeartbeat updates the region information.
@@ -410,6 +411,7 @@ func (c *clusterInfo) handleRegionHeartbeat(region *core.RegionInfo) error {
 		for _, p := range region.Peers {
 			c.updateStoreStatus(p.GetStoreId())
 		}
+
 	}
 
 	c.BasicCluster.UpdateWriteStatus(region)

server/cache_test.go

@@ -528,6 +528,53 @@ func (s *testClusterInfoSuite) testRegionSplitAndMerge(c *C, cache *clusterInfo)
 	}
 }
 
+func (s *testClusterInfoSuite) TestUpdateStorePendingPeerCount(c *C) {
+	cluster := newClusterInfo(core.NewMockIDAllocator())
+	stores := newTestStores(5)
+	for _, s := range stores {
+		cluster.putStore(s)
+	}
+	peers := []*metapb.Peer{
+		{
+			Id:      2,
+			StoreId: 1,
+		},
+		{
+			Id:      3,
+			StoreId: 2,
+		},
+		{
+			Id:      3,
+			StoreId: 3,
+		},
+		{
+			Id:      4,
+			StoreId: 4,
+		},
+	}
+	origin := &core.RegionInfo{
+		Region:       &metapb.Region{Id: 1, Peers: peers[:3]},
+		Leader:       peers[0],
+		PendingPeers: peers[1:3],
+	}
+	cluster.handleRegionHeartbeat(origin)
+	checkPendingPeerCount([]int{0, 1, 1, 0}, cluster, c)
+	newRegion := &core.RegionInfo{
+		Region:       &metapb.Region{Id: 1, Peers: peers[1:]},
+		Leader:       peers[1],
+		PendingPeers: peers[3:4],
+	}
+	cluster.handleRegionHeartbeat(newRegion)
+	checkPendingPeerCount([]int{0, 0, 0, 1}, cluster, c)
+}
+
+func checkPendingPeerCount(expect []int, cache *clusterInfo, c *C) {
+	for i, e := range expect {
+		s := cache.Stores.GetStore(uint64(i + 1))
+		c.Assert(s.PendingPeerCount, Equals, e)
+	}
+}
+
 var _ = Suite(&testClusterUtilSuite{})
 
 type testClusterUtilSuite struct{}

server/config.go

@@ -305,7 +305,8 @@ func (c *Config) configFromFile(path string) error {
 type ScheduleConfig struct {
 	// If the snapshot count of one store is greater than this value,
 	// it will never be used as a source or target store.
-	MaxSnapshotCount uint64 `toml:"max-snapshot-count,omitempty" json:"max-snapshot-count"`
+	MaxSnapshotCount    uint64 `toml:"max-snapshot-count,omitempty" json:"max-snapshot-count"`
+	MaxPendingPeerCount uint64 `toml:"max-pending-peer-count,omitempty" json:"max-pending-peer-count"`
 	// MaxStoreDownTime is the max duration after which
 	// a store will be considered to be down if it hasn't reported heartbeats.
 	MaxStoreDownTime typeutil.Duration `toml:"max-store-down-time,omitempty" json:"max-store-down-time"`
@@ -344,6 +345,7 @@ type SchedulerConfig struct {
 const (
 	defaultMaxReplicas          = 3
 	defaultMaxSnapshotCount     = 3
+	defaultMaxPendingPeerCount  = 16
 	defaultMaxStoreDownTime     = time.Hour
 	defaultLeaderScheduleLimit  = 64
 	defaultRegionScheduleLimit  = 12
@@ -358,6 +360,7 @@ var defaultSchedulers = SchedulerConfigs{
 
 func (c *ScheduleConfig) adjust() {
 	adjustUint64(&c.MaxSnapshotCount, defaultMaxSnapshotCount)
+	adjustUint64(&c.MaxPendingPeerCount, defaultMaxPendingPeerCount)
 	adjustDuration(&c.MaxStoreDownTime, defaultMaxStoreDownTime)
 	adjustUint64(&c.LeaderScheduleLimit, defaultLeaderScheduleLimit)
 	adjustUint64(&c.RegionScheduleLimit, defaultRegionScheduleLimit)
@@ -431,6 +434,10 @@ func (o *scheduleOption) GetMaxSnapshotCount() uint64 {
 	return o.load().MaxSnapshotCount
 }
 
+func (o *scheduleOption) GetMaxPendingPeerCount() uint64 {
+	return o.load().MaxPendingPeerCount
+}
+
 func (o *scheduleOption) GetMaxStoreDownTime() time.Duration {
 	return o.load().MaxStoreDownTime.Duration
 }

server/core/region.go

@@ -245,19 +245,21 @@ func (rm *regionMap) Delete(id uint64) {
 
 // RegionsInfo for export
 type RegionsInfo struct {
-	tree      *regionTree
-	regions   *regionMap            // regionID -> regionInfo
-	leaders   map[uint64]*regionMap // storeID -> regionID -> regionInfo
-	followers map[uint64]*regionMap // storeID -> regionID -> regionInfo
+	tree         *regionTree
+	regions      *regionMap            // regionID -> regionInfo
+	leaders      map[uint64]*regionMap // storeID -> regionID -> regionInfo
+	followers    map[uint64]*regionMap // storeID -> regionID -> regionInfo
+	pendingPeers map[uint64]*regionMap // storeID -> regionID -> regionInfo
 }
 
 // NewRegionsInfo creates RegionsInfo with tree, regions, leaders and followers
 func NewRegionsInfo() *RegionsInfo {
 	return &RegionsInfo{
-		tree:      newRegionTree(),
-		regions:   newRegionMap(),
-		leaders:   make(map[uint64]*regionMap),
-		followers: make(map[uint64]*regionMap),
+		tree:         newRegionTree(),
+		regions:      newRegionMap(),
+		leaders:      make(map[uint64]*regionMap),
+		followers:    make(map[uint64]*regionMap),
+		pendingPeers: make(map[uint64]*regionMap),
 	}
 }
 
@@ -319,6 +321,16 @@ func (r *RegionsInfo) AddRegion(region *RegionInfo) {
 			store.Put(region)
 		}
 	}
+
+	for _, peer := range region.PendingPeers {
+		storeID := peer.GetStoreId()
+		store, ok := r.pendingPeers[storeID]
+		if !ok {
+			store = newRegionMap()
+			r.pendingPeers[storeID] = store
+		}
+		store.Put(region)
+	}
 }
 
 // RemoveRegion remove RegionInfo from regionTree and regionMap
@@ -332,6 +344,7 @@ func (r *RegionsInfo) RemoveRegion(region *RegionInfo) {
 		storeID := peer.GetStoreId()
 		r.leaders[storeID].Delete(region.GetId())
 		r.followers[storeID].Delete(region.GetId())
+		r.pendingPeers[storeID].Delete(region.GetId())
 	}
 }
 
@@ -372,6 +385,11 @@ func (r *RegionsInfo) GetStoreRegionCount(storeID uint64) int {
 	return r.GetStoreLeaderCount(storeID) + r.GetStoreFollowerCount(storeID)
 }
 
+// GetStorePendingPeerCount gets the total count of  a store's region that includes pending peer
+func (r *RegionsInfo) GetStorePendingPeerCount(storeID uint64) int {
+	return r.pendingPeers[storeID].Len()
+}
+
 // GetStoreLeaderCount get the total count of a store's leader RegionInfo
 func (r *RegionsInfo) GetStoreLeaderCount(storeID uint64) int {
 	return r.leaders[storeID].Len()

server/core/store.go

@@ -29,12 +29,13 @@ type StoreInfo struct {
 	*metapb.Store
 	Stats *pdpb.StoreStats
 	// Blocked means that the store is blocked from balance.
-	blocked         bool
-	LeaderCount     int
-	RegionCount     int
-	LastHeartbeatTS time.Time
-	LeaderWeight    float64
-	RegionWeight    float64
+	blocked          bool
+	LeaderCount      int
+	RegionCount      int
+	PendingPeerCount int
+	LastHeartbeatTS  time.Time
+	LeaderWeight     float64
+	RegionWeight     float64
 }
 
 // NewStoreInfo creates StoreInfo with meta data.
@@ -49,14 +50,15 @@ func NewStoreInfo(store *metapb.Store) *StoreInfo {
 // Clone creates a copy of current StoreInfo.
 func (s *StoreInfo) Clone() *StoreInfo {
 	return &StoreInfo{
-		Store:           proto.Clone(s.Store).(*metapb.Store),
-		Stats:           proto.Clone(s.Stats).(*pdpb.StoreStats),
-		blocked:         s.blocked,
-		LeaderCount:     s.LeaderCount,
-		RegionCount:     s.RegionCount,
-		LastHeartbeatTS: s.LastHeartbeatTS,
-		LeaderWeight:    s.LeaderWeight,
-		RegionWeight:    s.RegionWeight,
+		Store:            proto.Clone(s.Store).(*metapb.Store),
+		Stats:            proto.Clone(s.Stats).(*pdpb.StoreStats),
+		blocked:          s.blocked,
+		LeaderCount:      s.LeaderCount,
+		RegionCount:      s.RegionCount,
+		PendingPeerCount: s.PendingPeerCount,
+		LastHeartbeatTS:  s.LastHeartbeatTS,
+		LeaderWeight:     s.LeaderWeight,
+		RegionWeight:     s.RegionWeight,
 	}
 }
 
@@ -335,6 +337,13 @@ func (s *StoresInfo) SetRegionCount(storeID uint64, regionCount int) {
 	}
 }
 
+// SetPendingPeerCount sets the pengding count to a storeInfo
+func (s *StoresInfo) SetPendingPeerCount(storeID uint64, pendingPeerCount int) {
+	if store, ok := s.stores[storeID]; ok {
+		store.PendingPeerCount = pendingPeerCount
+	}
+}
+
 // TotalWrittenBytes return the total written bytes of all StoreInfo
 func (s *StoresInfo) TotalWrittenBytes() uint64 {
 	var totalWrittenBytes uint64

server/schedule/filters.go

@@ -130,6 +130,28 @@ func (f *healthFilter) FilterTarget(store *core.StoreInfo) bool {
 	return f.filter(store)
 }
 
+type pendingPeerCountFilter struct {
+	opt Options
+}
+
+// NewPendingPeerCountFilter creates a Filter that filters all stores that are
+// currently handling too many pengding peers.
+func NewPendingPeerCountFilter(opt Options) Filter {
+	return &pendingPeerCountFilter{opt: opt}
+}
+
+func (p *pendingPeerCountFilter) filter(store *core.StoreInfo) bool {
+	return store.PendingPeerCount > int(p.opt.GetMaxPendingPeerCount())
+}
+
+func (p *pendingPeerCountFilter) FilterSource(store *core.StoreInfo) bool {
+	return p.filter(store)
+}
+
+func (p *pendingPeerCountFilter) FilterTarget(store *core.StoreInfo) bool {
+	return p.filter(store)
+}
+
 type snapshotCountFilter struct {
 	opt Options
 }

server/schedule/opts.go

@@ -23,6 +23,7 @@ type Options interface {
 	GetRegionScheduleLimit() uint64
 
 	GetMaxSnapshotCount() uint64
+	GetMaxPendingPeerCount() uint64
 	GetMaxStoreDownTime() time.Duration
 
 	GetMaxReplicas() int

server/schedulers/balance_region.go

@@ -48,6 +48,7 @@ func newBalanceRegionScheduler(opt schedule.Options, limiter *schedule.Limiter)
 		schedule.NewHealthFilter(opt),
 		schedule.NewSnapshotCountFilter(opt),
 		schedule.NewStorageThresholdFilter(opt),
+		schedule.NewPendingPeerCountFilter(opt),
 	}
 	base := newBaseScheduler(limiter)
 	return &balanceRegionScheduler{

server/schedulers/mockcluster.go

@@ -211,6 +211,7 @@ func (mc *mockCluster) addLeaderRegionWithReadInfo(regionID uint64, leaderID uin
 const (
 	defaultMaxReplicas          = 3
 	defaultMaxSnapshotCount     = 3
+	defaultMaxPendingPeerCount  = 16
 	defaultMaxStoreDownTime     = time.Hour
 	defaultLeaderScheduleLimit  = 64
 	defaultRegionScheduleLimit  = 12
@@ -223,6 +224,7 @@ type MockSchedulerOptions struct {
 	RegionScheduleLimit   uint64
 	LeaderScheduleLimit   uint64
 	MaxSnapshotCount      uint64
+	MaxPendingPeerCount   uint64
 	MaxStoreDownTime      time.Duration
 	MaxReplicas           int
 	LocationLabels        []string
@@ -237,6 +239,7 @@ func newMockSchedulerOptions() *MockSchedulerOptions {
 	mso.MaxStoreDownTime = defaultMaxStoreDownTime
 	mso.MaxReplicas = defaultMaxReplicas
 	mso.HotRegionLowThreshold = schedule.HotRegionLowThreshold
+	mso.MaxPendingPeerCount = defaultMaxPendingPeerCount
 	return mso
 }
 
@@ -255,6 +258,11 @@ func (mso *MockSchedulerOptions) GetMaxSnapshotCount() uint64 {
 	return mso.MaxSnapshotCount
 }
 
+// GetMaxPendingPeerCount mock method
+func (mso *MockSchedulerOptions) GetMaxPendingPeerCount() uint64 {
+	return mso.MaxPendingPeerCount
+}
+
 // GetMaxStoreDownTime mock method
 func (mso *MockSchedulerOptions) GetMaxStoreDownTime() time.Duration {
 	return mso.MaxStoreDownTime