*: add hot region schedule #611

nolouch · 2017-04-11T06:29:23Z

now, collect region writes status from the heartbeat. randomly select a hot region in best write hot store then do some transfer.
cc @siddontang @disksing @zhangjinpeng1987

zhangjinpeng87 · 2017-04-11T06:50:28Z

server/region_cache.go

@@ -297,3 +302,107 @@ func (c *fifoCache) len() int {

 	return c.ll.Len()
 }
+
+type queueCache struct {


Can we abstract this to a LRUCache?

Can it be implemented with https://golang.org/pkg/container/heap/#example__intHeap

zhangjinpeng87 · 2017-04-11T06:55:56Z

server/region_cache.go

+}
+
+func (q *queueCache) Push(key interface{}, value interface{}) {
+	q.Lock()


defer q.Unlock()

zhangjinpeng87 · 2017-04-11T06:56:34Z

server/region_cache.go

+	return q.head == q.tail
+}
+
+func (q *queueCache) Push(key interface{}, value interface{}) {


Do we need return bool to indicate that if push success?

zhangjinpeng87 · 2017-04-11T06:58:08Z

server/region_cache.go

+		return
+	}
+	v := kvItem{key: key, value: value}
+	q.items[q.tail] = v


do we need check if q.tail is in not out of range?

I check is it full firstly.

zhangjinpeng87 · 2017-04-11T07:06:20Z

server/region_cache.go

+	return v.value
+}
+
+func (q *queueCache) Delete(key interface{}) {


Please add bool return value to indicate if deleted or not exist.

zhangjinpeng87 · 2017-04-11T07:06:52Z

server/region_cache.go

+}
+
+func (q *queueCache) Delete(key interface{}) {
+	q.Lock()


defer q.Unlock()

zhangjinpeng87 · 2017-04-11T07:09:26Z

server/region_cache.go

+
+func (q *queueCache) Delete(key interface{}) {
+	q.Lock()
+	id, ok := q.index[key]


zhangjinpeng87 · 2017-04-11T07:24:26Z

server/balancer.go

+func (m MetaRegionStatus) Less(i, j int) bool { return m[i].WriteBytes > m[j].WriteBytes }
+
+type RegionHotStatus struct {
+	TotalWriteBytes uint64           `json:"total_write"`


What the difference between TotalWriteBytes and MetaStatus.WriteBytes?

zhangjinpeng87 · 2017-04-11T07:30:05Z

server/balancer.go

+type balanceHotRegionScheduler struct {
+	opt         *scheduleOption
+	limit       uint64
+	scoreStatus map[uint64]RegionHotStatus


Please add comment for this map, what the key is.

zhangjinpeng87 · 2017-04-11T08:15:08Z

server/balancer.go

+func newBalanceHotRegionScheduler(opt *scheduleOption) *balanceHotRegionScheduler {
+	return &balanceHotRegionScheduler{
+		opt:         opt,
+		limit:       1,


Can we modify the limit?

use a const var

andelf · 2017-04-11T07:09:37Z

server/api/router.go

@@ -60,6 +60,10 @@ func createRouter(prefix string, svr *server.Server) *mux.Router {
 	router.HandleFunc("/api/v1/labels", labelsHandler.Get).Methods("GET")
 	router.HandleFunc("/api/v1/labels/stores", labelsHandler.GetStores).Methods("GET")

+	hotStatusHandler := newHotStatusHandler(svr, rd)


hot or hotspot?

/api/v1/hotspot/host
/api/v1/hotspot/store?

andelf · 2017-04-11T07:12:22Z

server/balancer.go

+	items := cluster.writeStatus.GetList()
+	for _, item := range items {
+		r := item.value.(RegionStateValue)
+		if r.UpdateTimes < 5 {


make this 5 a const?

andelf · 2017-04-11T07:14:35Z

server/balancer.go

+
+type RegionStateValue struct {
+	RegionID       uint64    `json:"region_id"`
+	WriteBytes     uint64    `json:"write_bytes"`


andelf · 2017-04-11T07:41:18Z

server/region_cache.go

@@ -297,3 +302,107 @@ func (c *fifoCache) len() int {

 	return c.ll.Len()
 }
+
+type queueCache struct {


Can it be implemented with https://golang.org/pkg/container/heap/#example__intHeap

zhangjinpeng87 · 2017-04-11T09:20:22Z

server/balancer.go

+}
+
+func (l *balanceHotRegionScheduler) clearScore() {
+	for sid, status := range l.scoreStatus {


I prefer clear the map directly.

zhangjinpeng87 · 2017-04-12T11:08:41Z

server/region_cache.go

+	return q.isFull()
+}
+
+func (q *hashCache) isFull() bool {


why there are two isFull?

isFullLocked

zhangjinpeng87 · 2017-04-12T11:09:29Z

server/region.go

@@ -50,6 +51,7 @@ func (r *RegionInfo) clone() *RegionInfo {
 		Leader:       proto.Clone(r.Leader).(*metapb.Peer),
 		DownPeers:    downPeers,
 		PendingPeers: pendingPeers,
+		WriteBytes:   r.WriteBytes,


WrittenBytes

zhangjinpeng87 · 2017-04-12T11:14:23Z

server/coordinator.go

+		Scheduler:   s,
+		opt:         c.opt,
+		limiter:     c.limiter,
+		interval:    minInterval,


Why there are two interval?

please add comments.

zhangjinpeng87 · 2017-04-12T11:24:35Z

server/cache.go

+	res := make(map[uint64]uint64)
+	for _, s := range c.getStores() {
+		res[s.GetId()] = s.status.GetBytesWritten()
+		res[0] += s.status.GetBytesWritten()


remove this line. If you need total written bytes, please sum all the element in the map.

siddontang · 2017-04-13T05:49:05Z

server/balancer.go

+type RegionStateValue struct {
+	RegionID       uint64    `json:"region_id"`
+	WriteBytes     uint64    `json:"write_bytes"`
+	UpdateTimes    int       `json:"update_times"`


UpdateCount

siddontang · 2017-04-13T05:49:22Z

server/balancer.go

+	UpdateTimes    int       `json:"update_times"`
+	LastUpdateTime time.Time `json:"last_update_time"`
+	StoreID        uint64    `json:"-"`
+	antiTimes      int


what is antiTimes?

siddontang · 2017-04-13T05:53:12Z

server/region_cache.go

@@ -297,3 +297,70 @@ func (c *fifoCache) len() int {

 	return c.ll.Len()
 }
+
+type hashCache struct {


please us another PR and add test for this structure

siddontang · 2017-04-13T05:54:00Z

server/region_cache.go

+	q.Lock()
+	defer q.Unlock()
+	if q.isFull() {
+		log.Info("Debug queue is full")


better to return a full error.
Btw, what's Debug mean here?

I will remove those log later. just for observe in test now.

siddontang · 2017-04-13T05:54:35Z

server/region_cache.go

+	return item
+}
+
+func (q *hashCache) GetList() []interface{} {


/GetList/List/s

siddontang · 2017-04-13T05:55:09Z

server/coordinator.go

 			if !s.AllowSchedule() {
 				continue
 			}
 			if op := s.Schedule(c.cluster); op != nil {
 				c.addOperator(op)
 			}
+			if s.Scheduler.GetName() == "balance-hot-region-scheduler" {


use a const var

disksing · 2017-04-25T07:34:02Z

LGTM.

…cap/pd into hot-region-schedule

siddontang · 2017-04-25T08:27:21Z

PTAL @andelf @zhangjinpeng1987

andelf

LGTM

CLAassistant · 2017-04-26T03:34:28Z

All committers have signed the CLA.

…cap/pd into hot-region-schedule

siddontang · 2017-04-26T09:04:39Z

server/api/router.go

@@ -60,6 +60,10 @@ func createRouter(prefix string, svr *server.Server) *mux.Router {
 	router.HandleFunc("/api/v1/labels", labelsHandler.Get).Methods("GET")
 	router.HandleFunc("/api/v1/labels/stores", labelsHandler.GetStores).Methods("GET")

+	hotStatusHandler := newHotStatusHandler(handler, rd)


Please let pd-ctl support this later.

siddontang · 2017-04-26T09:06:35Z

server/balancer.go

+		status, ok := h.scoreStatus[storeID]
+		if !ok {
+			status = &StoreHotRegions{
+				RegionsStat: make(RegionsStat, 0, 100),


use const for 100

siddontang · 2017-04-26T09:08:44Z

server/balancer.go

+	// the hottest region in the store not move
+	// randomly pick a region from 1 .. length-1
+	// TODO: consider hot degree when pick
+	rr := h.r.Int31n(int32(length-1)) + 1


seem you can use Intn(length - 1) + 1

siddontang · 2017-04-26T09:10:23Z

server/balancer.go

+
+	avgRegionCount := hotRegionTotalCount / float64(len(h.scoreStatus))
+	// Multiplied by 0.75 to avoid transfer back and forth
+	limit := uint64((float64(s.RegionsStat.Len()) - avgRegionCount) * 0.75)


define a const var for 0.75

siddontang · 2017-04-26T09:15:11Z

Rest LGTM

siddontang · 2017-04-27T09:44:20Z

server/coordinator.go

+
+	writeStatLRUMaxLen            = 1000
+	storeHotRegionsDefaultLen     = 100
+	hotRegionLimitCoeff           = 0.75


factor may be better than coeff?

nolouch added 4 commits April 7, 2017 20:53

*: add hot region schedule

e2982b7

*: adjust state outqueue state

2a2ef7f

*:calculate avg bytes write

ca4c248

*: clean up

7d8e7bf