server/coordinator: startup schedulers with considering the proportion of regions

server/cluster_info.go

@@ -35,10 +35,10 @@ type clusterInfo struct {
 	id              core.IDAllocator
 	kv              *core.KV
 	meta            *metapb.Cluster
-	activeRegions   int
 	opt             *scheduleOption
 	regionStats     *regionStatistics
 	labelLevelStats *labelLevelStatistics
+	prepareChecker  *prepareChecker
 }
 
 func newClusterInfo(id core.IDAllocator, opt *scheduleOption, kv *core.KV) *clusterInfo {
@@ -48,6 +48,7 @@ func newClusterInfo(id core.IDAllocator, opt *scheduleOption, kv *core.KV) *clus
 		opt:             opt,
 		kv:              kv,
 		labelLevelStats: newLabelLevelStatistics(),
+		prepareChecker:  newPrepareChecker(),
 	}
 }
 
@@ -422,7 +423,7 @@ func (c *clusterInfo) GetFollowerStores(region *core.RegionInfo) []*core.StoreIn
 func (c *clusterInfo) isPrepared() bool {
 	c.RLock()
 	defer c.RUnlock()
-	return float64(c.core.Regions.Length())*collectFactor <= float64(c.activeRegions)
+	return c.prepareChecker.check(c)
 }
 
 // handleStoreHeartbeat updates the store status.
@@ -520,7 +521,7 @@ func (c *clusterInfo) handleRegionHeartbeat(region *core.RegionInfo) error {
 	c.Lock()
 	defer c.Unlock()
 	if isNew {
-		c.activeRegions++
+		c.prepareChecker.collect(region)
 	}
 
 	if saveCache {
@@ -711,3 +712,44 @@ func (c *clusterInfo) RegionWriteStats() []*core.RegionStat {
 	// RegionStats is a thread-safe method
 	return c.core.HotCache.RegionStats(schedule.WriteFlow)
 }
+
+type prepareChecker struct {
+	reactiveRegions map[uint64]int
+	start           time.Time
+	sum             int
+	isPrepared      bool
+}
+
+func newPrepareChecker() *prepareChecker {
+	return &prepareChecker{
+		start:           time.Now(),
+		reactiveRegions: make(map[uint64]int),
+	}
+}
+
+func (checker *prepareChecker) check(c *clusterInfo) bool {
+	if checker.isPrepared || time.Since(checker.start) > collectTimeout {
+		return true
+	}
+	if float64(c.core.Regions.Length())*collectFactor > float64(checker.sum) {
+		return false
+	}
+	for _, store := range c.core.GetStores() {
+		if !store.IsUp() {
+			continue
+		}
+		storeID := store.GetId()
+		if float64(c.core.Regions.GetStoreRegionCount(storeID))*collectFactor > float64(checker.reactiveRegions[storeID]) {
+			return false
+		}
+	}
+	checker.isPrepared = true
+	return true
+}
+
+func (checker *prepareChecker) collect(region *core.RegionInfo) {
+	for _, p := range region.GetPeers() {
+		checker.reactiveRegions[p.GetStoreId()]++
+	}
+	checker.sum++
+}

server/coordinator.go

@@ -34,6 +34,7 @@ import (
 const (
 	runSchedulerCheckInterval = 3 * time.Second
 	collectFactor             = 0.8
+	collectTimeout            = 5 * time.Minute
 	historyKeepTime           = 5 * time.Minute
 	maxScheduleRetries        = 10
 

server/coordinator_test.go

@@ -444,12 +444,19 @@ func (s *testCoordinatorSuite) TestShouldRun(c *C) {
 
 	co := newCoordinator(tc.clusterInfo, hbStreams, namespace.DefaultClassifier)
 
+	tc.addLeaderStore(1, 5)
+	tc.addLeaderStore(2, 2)
+	tc.addLeaderStore(3, 0)
+	tc.addLeaderStore(4, 0)
 	tc.LoadRegion(1, 1, 2, 3)
 	tc.LoadRegion(2, 1, 2, 3)
 	tc.LoadRegion(3, 1, 2, 3)
 	tc.LoadRegion(4, 1, 2, 3)
 	tc.LoadRegion(5, 1, 2, 3)
+	tc.LoadRegion(6, 2, 1, 4)
+	tc.LoadRegion(7, 2, 1, 4)
 	c.Assert(co.shouldRun(), IsFalse)
+	c.Assert(tc.core.Regions.GetStoreRegionCount(4), Equals, 2)
 
 	tbl := []struct {
 		regionID  uint64
@@ -458,8 +465,11 @@ func (s *testCoordinatorSuite) TestShouldRun(c *C) {
 		{1, false},
 		{2, false},
 		{3, false},
-		{4, true},
-		{5, true},
+		{4, false},
+		{5, false},
+		// store4 needs collect two region
+		{6, false},
+		{7, true},
 	}
 
 	for _, t := range tbl {
@@ -471,7 +481,7 @@ func (s *testCoordinatorSuite) TestShouldRun(c *C) {
 	nr := &metapb.Region{Id: 6, Peers: []*metapb.Peer{}}
 	newRegion := core.NewRegionInfo(nr, nil)
 	tc.handleRegionHeartbeat(newRegion)
-	c.Assert(co.cluster.activeRegions, Equals, 6)
+	c.Assert(co.cluster.prepareChecker.sum, Equals, 7)
 
 }
 
@@ -629,8 +639,8 @@ func (s *testCoordinatorSuite) TestRestart(c *C) {
 	tc.addRegionStore(2, 2)
 	tc.addRegionStore(3, 3)
 	tc.addLeaderRegion(1, 1)
-	tc.activeRegions = 1
 	region := tc.GetRegion(1)
+	tc.prepareChecker.collect(region)
 
 	// Add 1 replica on store 2.
 	co := newCoordinator(tc.clusterInfo, hbStreams, namespace.DefaultClassifier)

server/core/region.go

@@ -538,10 +538,6 @@ func (r *RegionsInfo) AddRegion(region *RegionInfo) []*metapb.Region {
 
 	r.regions.Put(region)
 
-	if region.leader == nil {
-		return overlaps
-	}
-
 	// Add to leaders and followers.
 	for _, peer := range region.GetVoters() {
 		storeID := peer.GetStoreId()