Repository: alibaba/sentinel-golang (go)

Issue:
[BUG] estimatedQueueingDuration in ThrottlingChecker may overflow, which may cause infinite waiting
<!-- Here is for bug reports and feature requests ONLY! 

If you're looking for help, please check our Dingtalk group and the Gitter room.

Please try to use English to describe your issue, or at least provide a snippet of English translation.
-->

## Issue Description

Type: *bug report*

### Describe what happened
![image](https://user-images.githubusercontent.com/13688621/100205156-b4efed80-2f3f-11eb-8c28-aa45398df9ee.png)

我加了相关的日志，发现estimatedQueueingDuration的值已经大于int64最大值了，如下
`
{"level":"info","ts":"2020-11-25 13:21:03","caller":"flow/tc_throttling.go:63","func":"flow.(*ThrottlingChecker).DoCheck","msg":"ThrottlingChecker2","pid":393,"service":"","GoID":54006,"ts-ms":1606281663019,"estimatedQueueingDuration":18446744073709551461,"currentTimeNano":1606281663019051949,"interval":1000000}
{"level":"info","ts":"2020-11-25 13:42:38","caller":"flow/tc_throttling.go:63","func":"flow.(*ThrottlingChecker).DoCheck","msg":"ThrottlingChecker2","pid":393,"service":"","GoID":68220,"ts-ms":1606282958743,"estimatedQueueingDuration":18446744073709551330,"currentTimeNano":1606282958743045270,"interval":1000000}
{"level":"info","ts":"2020-11-25 14:30:37","caller":"flow/tc_throttling.go:63","func":"flow.(*ThrottlingChecker).DoCheck","msg":"ThrottlingChecker2","pid":393,"service":"","GoID":101379,"ts-ms":1606285837180,"estimatedQueueingDuration":18446744073709551374,"currentTimeNano":1606285837180360588,"interval":1000000}
{"level":"info","ts":"2020-11-25 14:31:10","caller":"flow/tc_throttling.go:55","func":"flow.(*ThrottlingChecker).DoCheck","msg":"ThrottlingChecker1","pid":393,"service":"","GoID":101772,"ts-ms":1606285870585,"estimatedQueueingDuration":18446744073709551383,"currentTimeNano":1606285870585095028,"interval":1000000}
{"level":"info","ts":"2020-11-25 14:33:52","caller":"flow/tc_throttling.go:55","func":"flow.(*ThrottlingChecker).DoCheck","msg":"ThrottlingChecker1","pid":393,"service":"","GoID":103799,"ts-ms":1606286032195,"estimatedQueueingDuration":18446744073709551149,"currentTimeNano":1606286032195488790,"interval":1000000}
{"level":"info","ts":"2020-11-25 14:42:32","caller":"flow/tc_throttling.go:63","func":"flow.(*ThrottlingChecker).DoCheck","msg":"ThrottlingChecker2","pid":393,"service":"","GoID":110865,"ts-ms":1606286552392,"estimatedQueueingDuration":18446744073709551567,"currentTimeNano":1606286552392114392,"interval":1000000}
{"level":"info","ts":"2020-11-25 15:07:57","caller":"flow/tc_throttling.go:55","func":"flow.(*ThrottlingChecker).DoCheck","msg":"ThrottlingChecker1","pid":393,"service":"","GoID":134835,"ts-ms":1606288077250,"estimatedQueueingDuration":18446744073709551463,"currentTimeNano":1606288077250908120,"interval":1000000}
{"level":"info","ts":"2020-11-25 15:12:25","caller":"flow/tc_throttling.go:63","func":"flow.(*ThrottlingChecker).DoCheck","msg":"ThrottlingChecker2","pid":393,"service":"","GoID":139236,"ts-ms":1606288345973,"estimatedQueueingDuration":18446744073709551309,"currentTimeNano":1606288345973680338,"interval":1000000}
{"level":"info","ts":"2020-11-25 15:21:37","caller":"flow/tc_throttling.go:55","func":"flow.(*ThrottlingChecker).DoCheck","msg":"ThrottlingChecker1","pid":393,"service":"","GoID":149532,"ts-ms":1606288897647,"estimatedQueueingDuration":18446744073709551198,"currentTimeNano":1606288897647921208,"interval":1000000}
{"level":"info","ts":"2020-11-25 15:35:07","caller":"flow/tc_throttling.go:63","func":"flow.(*ThrottlingChecker).DoCheck","msg":"ThrottlingChecker2","pid":393,"service":"","GoID":166445,"ts-ms":1606289707601,"estimatedQueueingDuration":18446744073709551294,"currentTimeNano":1606289707601567729,"interval":1000000}
{"level":"info","ts":"2020-11-25 15:50:44","caller":"flow/tc_throttling.go:55","func":"flow.(*ThrottlingChecker).DoCheck","msg":"ThrottlingChecker1","pid":393,"service":"","GoID":188919,"ts-ms":1606290644177,"estimatedQueueingDuration":18446744073709551286,"currentTimeNano":1606290644177611891,"interval":1000000}
`

### Describe what you expected to happen


### How to reproduce it (as minimally and precisely as possible)

1. 
2. 
3. 

### Tell us your environment


### Anything else we need to know?


Pull request description:
Fix bugs of throttling control behavior in flow control module
<!--  Thanks for submitting a pull request! Here are some tips for you:
1. Please make sure you have read and understood the contributing guidelines: https://github.com/alibaba/blob/master/CONTRIBUTING.md
2. Please make sure the PR has a corresponding issue.
-->

### Describe what this PR does / why we need it

Fix bugs of throttling control behavior in flow control module, including overflow bug and time unit bug of wait duration.

### Does this pull request fix one issue?

Fixes #331 

### Describe how you did it

- Change semantic of `waitMs` to `nanosToWait` in `TokenResult` and polish related stat slots. Here we SHOULD use nanos instead of millis, or the queueing interval may not be accurate.
- Fix the bug that unsigned `estimatedQueueingDuration` in throttling checker may overflow. Use int64 instead.

### Describe how to verify it

Run the test cases and demo.

### Special notes for reviews

NONE.

Patch:
diff --git a/core/base/result.go b/core/base/result.go
index ce5d1da..b979357 100644
--- a/core/base/result.go
+++ b/core/base/result.go
@@ -2,6 +2,7 @@ package base
 
 import (
 	"fmt"
+	"time"
 )
 
 type BlockType uint8
@@ -58,13 +59,13 @@ func (s TokenResultStatus) String() string {
 type TokenResult struct {
 	status TokenResultStatus
 
-	blockErr *BlockError
-	waitMs   uint64
+	blockErr    *BlockError
+	nanosToWait time.Duration
 }
 
 func (r *TokenResult) DeepCopyFrom(newResult *TokenResult) {
 	r.status = newResult.status
-	r.waitMs = newResult.waitMs
+	r.nanosToWait = newResult.nanosToWait
 	if r.blockErr == nil {
 		r.blockErr = &BlockError{
 			blockType:     newResult.blockErr.blockType,
@@ -84,7 +85,7 @@ func (r *TokenResult) DeepCopyFrom(newResult *TokenResult) {
 func (r *TokenResult) ResetToPass() {
 	r.status = ResultStatusPass
 	r.blockErr = nil
-	r.waitMs = 0
+	r.nanosToWait = 0
 }
 
 func (r *TokenResult) ResetToBlocked(blockType BlockType) {
@@ -97,7 +98,7 @@ func (r *TokenResult) ResetToBlocked(blockType BlockType) {
 		r.blockErr.rule = nil
 		r.blockErr.snapshotValue = nil
 	}
-	r.waitMs = 0
+	r.nanosToWait = 0
 }
 
 func (r *TokenResult) ResetToBlockedWithMessage(blockType BlockType, blockMsg string) {
@@ -110,7 +111,7 @@ func (r *TokenResult) ResetToBlockedWithMessage(blockType BlockType, blockMsg st
 		r.blockErr.rule = nil
 		r.blockErr.snapshotValue = nil
 	}
-	r.waitMs = 0
+	r.nanosToWait = 0
 }
 
 func (r *TokenResult) ResetToBlockedWithCause(blockType BlockType, blockMsg string, rule SentinelRule, snapshot interface{}) {
@@ -123,7 +124,7 @@ func (r *TokenResult) ResetToBlockedWithCause(blockType BlockType, blockMsg stri
 		r.blockErr.rule = rule
 		r.blockErr.snapshotValue = snapshot
 	}
-	r.waitMs = 0
+	r.nanosToWait = 0
 }
 
 func (r *TokenResult) IsPass() bool {
@@ -142,8 +143,8 @@ func (r *TokenResult) BlockError() *BlockError {
 	return r.blockErr
 }
 
-func (r *TokenResult) WaitMs() uint64 {
-	return r.waitMs
+func (r *TokenResult) NanosToWait() time.Duration {
+	return r.nanosToWait
 }
 
 func (r *TokenResult) String() string {
@@ -153,45 +154,45 @@ func (r *TokenResult) String() string {
 	} else {
 		blockMsg = r.blockErr.Error()
 	}
-	return fmt.Sprintf("TokenResult{status=%s, blockErr=%s, waitMs=%d}", r.status.String(), blockMsg, r.waitMs)
+	return fmt.Sprintf("TokenResult{status=%s, blockErr=%s, nanosToWait=%d}", r.status.String(), blockMsg, r.nanosToWait)
 }
 
 func NewTokenResultPass() *TokenResult {
 	return &TokenResult{
-		status:   ResultStatusPass,
-		blockErr: nil,
-		waitMs:   0,
+		status:      ResultStatusPass,
+		blockErr:    nil,
+		nanosToWait: 0,
 	}
 }
 
 func NewTokenResultBlocked(blockType BlockType) *TokenResult {
 	return &TokenResult{
-		status:   ResultStatusBlocked,
-		blockErr: NewBlockError(blockType),
-		waitMs:   0,
+		status:      ResultStatusBlocked,
+		blockErr:    NewBlockError(blockType),
+		nanosToWait: 0,
 	}
 }
 
 func NewTokenResultBlockedWithMessage(blockType BlockType, blockMsg string) *TokenResult {
 	return &TokenResult{
-		status:   ResultStatusBlocked,
-		blockErr: NewBlockErrorWithMessage(blockType, blockMsg),
-		waitMs:   0,
+		status:      ResultStatusBlocked,
+		blockErr:    NewBlockErrorWithMessage(blockType, blockMsg),
+		nanosToWait: 0,
 	}
 }
 
 func NewTokenResultBlockedWithCause(blockType BlockType, blockMsg string, rule SentinelRule, snapshot interface{}) *TokenResult {
 	return &TokenResult{
-		status:   ResultStatusBlocked,
-		blockErr: NewBlockErrorWithCause(blockType, blockMsg, rule, snapshot),
-		waitMs:   0,
+		status:      ResultStatusBlocked,
+		blockErr:    NewBlockErrorWithCause(blockType, blockMsg, rule, snapshot),
+		nanosToWait: 0,
 	}
 }
 
-func NewTokenResultShouldWait(waitMs uint64) *TokenResult {
+func NewTokenResultShouldWait(waitNs time.Duration) *TokenResult {
 	return &TokenResult{
-		status:   ResultStatusShouldWait,
-		blockErr: nil,
-		waitMs:   waitMs,
+		status:      ResultStatusShouldWait,
+		blockErr:    nil,
+		nanosToWait: waitNs,
 	}
 }
diff --git a/core/flow/slot.go b/core/flow/slot.go
index fd81495..c56aea2 100644
--- a/core/flow/slot.go
+++ b/core/flow/slot.go
@@ -49,9 +49,9 @@ func (s *Slot) Check(ctx *base.EntryContext) *base.TokenResult {
 			return r
 		}
 		if r.Status() == base.ResultStatusShouldWait {
-			if waitMs := r.WaitMs(); waitMs > 0 {
+			if nanosToWait := r.NanosToWait(); nanosToWait > 0 {
 				// Handle waiting action.
-				time.Sleep(time.Duration(waitMs) * time.Millisecond)
+				time.Sleep(nanosToWait)
 			}
 			continue
 		}
diff --git a/core/flow/tc_throttling.go b/core/flow/tc_throttling.go
index 9196fdc..5d5c890 100644
--- a/core/flow/tc_throttling.go
+++ b/core/flow/tc_throttling.go
@@ -3,34 +3,36 @@ package flow
 import (
 	"math"
 	"sync/atomic"
+	"time"
 
 	"github.com/alibaba/sentinel-golang/core/base"
-	"github.com/alibaba/sentinel-golang/core/config"
 	"github.com/alibaba/sentinel-golang/util"
 )
 
+const (
+	BlockMsgQueueing = "flow throttling check blocked, estimated queueing time exceeds max queueing time"
+
+	MillisToNanosOffset = int64(time.Millisecond / time.Nanosecond)
+)
+
 // ThrottlingChecker limits the time interval between two requests.
 type ThrottlingChecker struct {
 	owner             *TrafficShapingController
-	maxQueueingTimeNs uint64
-	statIntervalNs    uint64
-	lastPassedTime    uint64
+	maxQueueingTimeNs int64
+	statIntervalNs    int64
+	lastPassedTime    int64
 }
 
 func NewThrottlingChecker(owner *TrafficShapingController, timeoutMs uint32, statIntervalMs uint32) *ThrottlingChecker {
-	var statIntervalNs uint64
+	var statIntervalNs int64
 	if statIntervalMs == 0 {
-		defaultIntervalMs := config.MetricStatisticIntervalMs()
-		if defaultIntervalMs == 0 {
-			defaultIntervalMs = 1000
-		}
-		statIntervalNs = uint64(defaultIntervalMs) * util.UnixTimeUnitOffset
+		statIntervalNs = 1000 * MillisToNanosOffset
 	} else {
-		statIntervalNs = uint64(statIntervalMs) * util.UnixTimeUnitOffset
+		statIntervalNs = int64(statIntervalMs) * MillisToNanosOffset
 	}
 	return &ThrottlingChecker{
 		owner:             owner,
-		maxQueueingTimeNs: uint64(timeoutMs) * util.UnixTimeUnitOffset,
+		maxQueueingTimeNs: int64(timeoutMs) * MillisToNanosOffset,
 		statIntervalNs:    statIntervalNs,
 		lastPassedTime:    0,
 	}
@@ -58,35 +60,33 @@ func (c *ThrottlingChecker) DoCheck(_ base.StatNode, batchCount uint32, threshol
 		return base.NewTokenResultBlocked(base.BlockTypeFlow)
 	}
 	// Here we use nanosecond so that we could control the queueing time more accurately.
-	curNano := util.CurrentTimeNano()
+	curNano := int64(util.CurrentTimeNano())
 
 	// The interval between two requests (in nanoseconds).
-	intervalNs := uint64(math.Ceil(float64(batchCount) / threshold * float64(c.statIntervalNs)))
+	intervalNs := int64(math.Ceil(float64(batchCount) / threshold * float64(c.statIntervalNs)))
 
 	// Expected pass time of this request.
-	expectedTime := atomic.LoadUint64(&c.lastPassedTime) + intervalNs
+	expectedTime := atomic.LoadInt64(&c.lastPassedTime) + intervalNs
 	if expectedTime <= curNano {
 		// Contention may exist here, but it's okay.
-		atomic.StoreUint64(&c.lastPassedTime, curNano)
+		atomic.StoreInt64(&c.lastPassedTime, curNano)
 		return nil
 	}
 
-	estimatedQueueingDuration := atomic.LoadUint64(&c.lastPassedTime) + intervalNs - util.CurrentTimeNano()
+	estimatedQueueingDuration := atomic.LoadInt64(&c.lastPassedTime) + intervalNs - int64(util.CurrentTimeNano())
 	if estimatedQueueingDuration > c.maxQueueingTimeNs {
-		msg := "flow throttling check blocked, estimated queueing time exceeds max queueing time"
-		return base.NewTokenResultBlockedWithCause(base.BlockTypeFlow, msg, rule, nil)
+		return base.NewTokenResultBlockedWithCause(base.BlockTypeFlow, BlockMsgQueueing, rule, nil)
 	}
 
-	oldTime := atomic.AddUint64(&c.lastPassedTime, intervalNs)
-	estimatedQueueingDuration = oldTime - util.CurrentTimeNano()
+	oldTime := atomic.AddInt64(&c.lastPassedTime, intervalNs)
+	estimatedQueueingDuration = oldTime - int64(util.CurrentTimeNano())
 	if estimatedQueueingDuration > c.maxQueueingTimeNs {
 		// Subtract the interval.
-		atomic.AddUint64(&c.lastPassedTime, ^(intervalNs - 1))
-		msg := "flow throttling check blocked, estimated queueing time exceeds max queueing time"
-		return base.NewTokenResultBlockedWithCause(base.BlockTypeFlow, msg, rule, nil)
+		atomic.AddInt64(&c.lastPassedTime, -intervalNs)
+		return base.NewTokenResultBlockedWithCause(base.BlockTypeFlow, BlockMsgQueueing, rule, nil)
 	}
 	if estimatedQueueingDuration > 0 {
-		return base.NewTokenResultShouldWait(estimatedQueueingDuration / util.UnixTimeUnitOffset)
+		return base.NewTokenResultShouldWait(time.Duration(estimatedQueueingDuration))
 	} else {
 		return base.NewTokenResultShouldWait(0)
 	}
diff --git a/core/hotspot/slot.go b/core/hotspot/slot.go
index f3e53a1..6843982 100644
--- a/core/hotspot/slot.go
+++ b/core/hotspot/slot.go
@@ -72,9 +72,9 @@ func (s *Slot) Check(ctx *base.EntryContext) *base.TokenResult {
 			return r
 		}
 		if r.Status() == base.ResultStatusShouldWait {
-			if waitMs := r.WaitMs(); waitMs > 0 {
+			if nanosToWait := r.NanosToWait(); nanosToWait > 0 {
 				// Handle waiting action.
-				time.Sleep(time.Duration(waitMs) * time.Millisecond)
+				time.Sleep(nanosToWait)
 			}
 			continue
 		}
diff --git a/core/hotspot/traffic_shaping.go b/core/hotspot/traffic_shaping.go
index fd4a5f6..68aa6cc 100644
--- a/core/hotspot/traffic_shaping.go
+++ b/core/hotspot/traffic_shaping.go
@@ -5,6 +5,7 @@ import (
 	"math"
 	"runtime"
 	"sync/atomic"
+	"time"
 
 	"github.com/alibaba/sentinel-golang/core/base"
 	"github.com/alibaba/sentinel-golang/core/hotspot/cache"
@@ -281,7 +282,7 @@ func (c *throttlingTrafficShapingController) PerformChecking(arg interface{}, ba
 				awaitTime := expectedTime - currentTimeInMs
 				if awaitTime > 0 {
 					atomic.StoreInt64(lastPassTimePtr, expectedTime)
-					return base.NewTokenResultShouldWait(uint64(awaitTime))
+					return base.NewTokenResultShouldWait(time.Duration(awaitTime) * time.Millisecond)
 				}
 				return nil
 			} else {


Test patch:
diff --git a/core/flow/tc_throttling_test.go b/core/flow/tc_throttling_test.go
index 8867d54..b478b1c 100644
--- a/core/flow/tc_throttling_test.go
+++ b/core/flow/tc_throttling_test.go
@@ -61,8 +61,8 @@ func TestThrottlingChecker_DoCheckSingleThread(t *testing.T) {
 	waitCount := int(float64(timeoutMs) / (float64(intervalMs) / threshold))
 	for i := 0; i < waitCount; i++ {
 		assert.True(t, resultList[i].Status() == base.ResultStatusShouldWait)
-		wt := resultList[i].WaitMs()
-		assert.InEpsilon(t, (i+1)*1000/int(waitCount), wt, 10)
+		wt := resultList[i].NanosToWait()
+		assert.InEpsilon(t, (i+1)*(int)(time.Second/time.Nanosecond)/waitCount, wt, 10)
 	}
 	for i := waitCount; i < reqCount; i++ {
 		assert.True(t, resultList[i].IsBlocked())