movingaverage: support concurrency safe queue in AvgOverTime

pkg/movingaverage/avg_over_time.go

@@ -15,8 +15,6 @@ package movingaverage
 
 import (
 	"time"
-
-	"github.com/phf/go-queue/queue"
 )
 
 type deltaWithInterval struct {
@@ -30,7 +28,7 @@ type deltaWithInterval struct {
 // stores recent changes that happened in the last avgInterval,
 // then calculates the change rate by (sum of changes) / (sum of intervals).
 type AvgOverTime struct {
-	que         *queue.Queue      // The element is `deltaWithInterval`, sum of all elements' interval is less than `avgInterval`
+	que         *SafeQueue        // The element is `deltaWithInterval`, sum of all elements' interval is less than `avgInterval`
 	margin      deltaWithInterval // The last element from `PopFront` in `que`
 	deltaSum    float64           // Including `margin` and all elements in `que`
 	intervalSum time.Duration     // Including `margin` and all elements in `que`
@@ -40,7 +38,7 @@ type AvgOverTime struct {
 // NewAvgOverTime returns an AvgOverTime with given interval.
 func NewAvgOverTime(interval time.Duration) *AvgOverTime {
 	return &AvgOverTime{
-		que: queue.New(),
+		que: NewSafeQueue(),
 		margin: deltaWithInterval{
 			delta:    0,
 			interval: 0,

pkg/movingaverage/queue.go

@@ -0,0 +1,54 @@
+// Copyright 2020 TiKV Project Authors.
+//
+// Licensed under the Apache License, Version 2.0 (the "License");
+// you may not use this file except in compliance with the License.
+// You may obtain a copy of the License at
+//
+//     http://www.apache.org/licenses/LICENSE-2.0
+//
+// Unless required by applicable law or agreed to in writing, software
+// distributed under the License is distributed on an "AS IS" BASIS,
+// See the License for the specific language governing permissions and
+// limitations under the License.
+
+package movingaverage
+
+import (
+	"sync"
+
+	"github.com/phf/go-queue/queue"
+)
+
+// SafeQueue is a concurrency safe queue
+type SafeQueue struct {
+	mu  sync.Mutex
+	que *queue.Queue
+}
+
+// NewSafeQueue return a SafeQueue
+func NewSafeQueue() *SafeQueue {
+	sq := &SafeQueue{}
+	sq.que = queue.New()
+	return sq
+}
+
+// Init implement init
+func (sq *SafeQueue) Init() {
+	sq.mu.Lock()
+	defer sq.mu.Unlock()
+	sq.que.Init()
+}
+
+// PushBack implement PushBack
+func (sq *SafeQueue) PushBack(v interface{}) {
+	sq.mu.Lock()
+	defer sq.mu.Unlock()
+	sq.que.PushBack(v)
+}
+
+// PopFront implement PopFront
+func (sq *SafeQueue) PopFront() interface{} {
+	sq.mu.Lock()
+	defer sq.mu.Unlock()
+	return sq.que.PopFront()
+}