statistics: make hotPeerCache concurrency safe

server/statistics/hot_peer.go

@@ -15,6 +15,7 @@ package statistics
 
 import (
 	"math"
+	"sync"
 	"time"
 
 	"github.com/tikv/pd/pkg/movingaverage"
@@ -27,43 +28,59 @@ const (
 )
 
 type dimStat struct {
-	typ         int
-	Rolling     *movingaverage.TimeMedian  // it's used to statistic hot degree and average speed.
-	LastAverage *movingaverage.AvgOverTime // it's used to obtain the average speed in last second as instantaneous speed.
+	typ int
+	mu  struct {
+		sync.RWMutex
+		Rolling     *movingaverage.TimeMedian  // it's used to statistic hot degree and average speed.
+		LastAverage *movingaverage.AvgOverTime // it's used to obtain the average speed in last second as instantaneous speed.
+	}
 }
 
 func newDimStat(typ int) *dimStat {
 	reportInterval := RegionHeartBeatReportInterval * time.Second
-	return &dimStat{
-		typ:         typ,
-		Rolling:     movingaverage.NewTimeMedian(DefaultAotSize, rollingWindowsSize, reportInterval),
-		LastAverage: movingaverage.NewAvgOverTime(reportInterval),
+	ds := &dimStat{
+		typ: typ,
 	}
+	ds.mu.Rolling = movingaverage.NewTimeMedian(DefaultAotSize, rollingWindowsSize, reportInterval)
+	ds.mu.LastAverage = movingaverage.NewAvgOverTime(reportInterval)
+	return ds
 }
 
 func (d *dimStat) Add(delta float64, interval time.Duration) {
-	d.LastAverage.Add(delta, interval)
-	d.Rolling.Add(delta, interval)
+	d.mu.Lock()
+	defer d.mu.Unlock()
+	d.mu.LastAverage.Add(delta, interval)
+	d.mu.Rolling.Add(delta, interval)
 }
 
 func (d *dimStat) isLastAverageHot(thresholds [dimLen]float64) bool {
-	return d.LastAverage.Get() >= thresholds[d.typ]
+	d.mu.RLock()
+	defer d.mu.RUnlock()
+	return d.mu.LastAverage.Get() >= thresholds[d.typ]
 }
 
 func (d *dimStat) isHot(thresholds [dimLen]float64) bool {
-	return d.Rolling.Get() >= thresholds[d.typ]
+	d.mu.RLock()
+	defer d.mu.RUnlock()
+	return d.mu.Rolling.Get() >= thresholds[d.typ]
 }
 
 func (d *dimStat) isFull() bool {
-	return d.LastAverage.IsFull()
+	d.mu.RLock()
+	defer d.mu.RUnlock()
+	return d.mu.LastAverage.IsFull()
 }
 
 func (d *dimStat) clearLastAverage() {
-	d.LastAverage.Clear()
+	d.mu.Lock()
+	defer d.mu.Unlock()
+	d.mu.LastAverage.Clear()
 }
 
 func (d *dimStat) Get() float64 {
-	return d.Rolling.Get()
+	d.mu.RLock()
+	defer d.mu.RUnlock()
+	return d.mu.Rolling.Get()
 }
 
 // HotPeerStat records each hot peer's statistics

server/statistics/hot_peer_cache_test.go

@@ -15,6 +15,7 @@ package statistics
 
 import (
 	"math/rand"
+	"sync"
 	"testing"
 	"time"
 
@@ -218,6 +219,32 @@ func newPeers(n int, pid genID, sid genID) []*metapb.Peer {
 	return peers
 }
 
+func (t *testHotPeerCache) TestUpdateHotPeerStatInConcurrency(c *C) {
+	cache := NewHotStoresStats(ReadFlow)
+	// new peer, interval is less than report interval
+	newItem := &HotPeerStat{needDelete: false, thresholds: [2]float64{0.0, 0.0}}
+	newItem = cache.updateHotPeerStat(newItem, nil, 60, 60, 30*time.Second)
+	c.Check(newItem, NotNil)
+	c.Check(newItem.HotDegree, Equals, 0)
+	c.Check(newItem.AntiCount, Equals, 0)
+	// sum of interval is less than report interval
+	oldItem := newItem
+	newItem = cache.updateHotPeerStat(newItem, oldItem, 60, 60, 10*time.Second)
+	c.Check(newItem.HotDegree, Equals, 0)
+	c.Check(newItem.AntiCount, Equals, 0)
+	wg := &sync.WaitGroup{}
+	wg.Add(100)
+	oldItem = newItem
+	newItem = cache.updateHotPeerStat(newItem, oldItem, 60, 60, 50*time.Second)
+	for i := 0; i < 100; i++ {
+		go func(newItem, oldItem *HotPeerStat, wg *sync.WaitGroup) {
+			defer wg.Done()
+			cache.updateHotPeerStat(newItem, oldItem, 60, 60, 50*time.Second)
+		}(oldItem, newItem, wg)
+	}
+	wg.Wait()
+}
+
 func (t *testHotPeerCache) TestUpdateHotPeerStat(c *C) {
 	cache := NewHotStoresStats(ReadFlow)
 

server/statistics/region_collection_test.go

@@ -14,8 +14,6 @@
 package statistics
 
 import (
-	"testing"
-
 	. "github.com/pingcap/check"
 	"github.com/pingcap/kvproto/pkg/metapb"
 	"github.com/pingcap/kvproto/pkg/pdpb"
@@ -25,10 +23,6 @@ import (
 	"github.com/tikv/pd/server/schedule/placement"
 )
 
-func TestStatistics(t *testing.T) {
-	TestingT(t)
-}
-
 var _ = Suite(&testRegionStatisticsSuite{})
 
 type testRegionStatisticsSuite struct {