Fix data race of Token result Pool

api/api.go

@@ -87,13 +87,6 @@ func Entry(resource string, opts ...EntryOption) (*base.SentinelEntry, *base.Blo
 }
 
 func entry(resource string, options *EntryOptions) (*base.SentinelEntry, *base.BlockError) {
-	var r *base.TokenResult
-	defer func() {
-		if r != nil {
-			base.RefurbishTokenResult(r)
-		}
-	}()
-
 	rw := base.NewResourceWrapper(resource, options.resourceType, options.entryType)
 	sc := options.slotChain
 
@@ -109,16 +102,20 @@ func entry(resource string, options *EntryOptions) (*base.SentinelEntry, *base.B
 		Args:         options.args,
 		Attachments:  options.attachments,
 	}
+	ctx.Data = make(map[interface{}]interface{})
 
 	e := base.NewSentinelEntry(ctx, rw, sc)
-	r = sc.Entry(ctx)
+	r := sc.Entry(ctx)
 	if r == nil {
 		// This indicates internal error in some slots, so just pass
 		return e, nil
 	}
 	if r.Status() == base.ResultStatusBlocked {
+		// r will be put to Pool in calling Exit()
+		// must finish the lifecycle of r.
+		blockErr := base.NewBlockErrorFromDeepCopy(r.BlockError())
 		e.Exit()
-		return nil, r.BlockError()
+		return nil, blockErr
 	}
 
 	return e, nil

api/tracer_test.go

@@ -40,7 +40,6 @@ func TestTraceErrorToCtx(t *testing.T) {
 		Resource: testRes,
 		StatNode: testStatNode,
 		Input:    nil,
-		Output:   nil,
 	}
 	tests[0].args.err = errors.New("biz error")
 
@@ -79,7 +78,6 @@ func TestTraceErrorToEntry(t *testing.T) {
 		Resource: testRes,
 		StatNode: testStatNode,
 		Input:    nil,
-		Output:   nil,
 	}
 	tests[0].args.entry = base.NewSentinelEntry(ctx, testRes, nil)
 	tests[0].args.err = errors.New("biz error")

core/base/block_error.go

@@ -25,6 +25,15 @@ func (e *BlockError) TriggeredValue() interface{} {
 	return e.snapshotValue
 }
 
+func NewBlockErrorFromDeepCopy(from *BlockError) *BlockError {
+	return &BlockError{
+		blockType:     from.blockType,
+		blockMsg:      from.blockMsg,
+		rule:          from.rule,
+		snapshotValue: from.snapshotValue,
+	}
+}
+
 func NewBlockError(blockType BlockType, blockMsg string) *BlockError {
 	return &BlockError{blockType: blockType, blockMsg: blockMsg}
 }

core/base/context.go

@@ -7,51 +7,44 @@ type EntryContext struct {
 	Resource *ResourceWrapper
 	StatNode StatNode
 
-	Input  *SentinelInput
-	Output *SentinelOutput
+	Input *SentinelInput
+	// the result of rule slots check
+	RuleCheckResult *TokenResult
+	// reserve for storing some intermediate data from the Entry execution process
+	Data map[interface{}]interface{}
 }
 
 func (ctx *EntryContext) StartTime() uint64 {
 	return ctx.startTime
 }
 
 func (ctx *EntryContext) IsBlocked() bool {
-	if ctx.Output.LastResult == nil {
+	if ctx.RuleCheckResult == nil {
 		return false
 	}
-	return ctx.Output.LastResult.IsBlocked()
+	return ctx.RuleCheckResult.IsBlocked()
 }
 
 func NewEmptyEntryContext() *EntryContext {
-	ctx := &EntryContext{
-		Input:  nil,
-		Output: newEmptyOutput(),
-	}
-	return ctx
+	return &EntryContext{}
 }
 
+// The input data of sentinel
 type SentinelInput struct {
 	AcquireCount uint32
 	Flag         int32
 	Args         []interface{}
-
 	// store some values in this context when calling context in slot.
 	Attachments map[interface{}]interface{}
 }
 
 func newEmptyInput() *SentinelInput {
-	return &SentinelInput{}
-}
-
-type SentinelOutput struct {
-	LastResult *TokenResult
-
-	// store output data.
-	Attachments map[interface{}]interface{}
-}
-
-func newEmptyOutput() *SentinelOutput {
-	return &SentinelOutput{}
+	return &SentinelInput{
+		AcquireCount: 1,
+		Flag:         0,
+		Args:         make([]interface{}, 0, 0),
+		Attachments:  make(map[interface{}]interface{}),
+	}
 }
 
 // Reset init EntryContext,
@@ -61,5 +54,10 @@ func (ctx *EntryContext) Reset() {
 	ctx.Resource = nil
 	ctx.StatNode = nil
 	ctx.Input = nil
-	ctx.Output = newEmptyOutput()
+	if ctx.RuleCheckResult == nil {
+		ctx.RuleCheckResult = NewTokenResultPass()
+	} else {
+		ctx.RuleCheckResult.ResetToPass()
+	}
+	ctx.Data = nil
 }

core/base/context_test.go

@@ -9,6 +9,6 @@ import (
 func TestEntryContext_IsBlocked(t *testing.T) {
 	ctx := NewEmptyEntryContext()
 	assert.False(t, ctx.IsBlocked(), "empty context with no result should indicate pass")
-	ctx.Output = &SentinelOutput{LastResult: NewTokenResultBlocked(BlockTypeUnknown, "")}
+	ctx.RuleCheckResult = NewTokenResultBlocked(BlockTypeUnknown, "Unknown")
 	assert.True(t, ctx.IsBlocked(), "context with blocked request should indicate blocked")
 }

core/base/result.go

@@ -2,10 +2,9 @@ package base
 
 import (
 	"fmt"
-	"sync"
 )
 
-type BlockType int32
+type BlockType uint8
 
 const (
 	BlockTypeUnknown BlockType = iota
@@ -14,12 +13,6 @@ const (
 	BlockTypeSystemFlow
 )
 
-var resultPool = sync.Pool{
-	New: func() interface{} {
-		return NewTokenResultEmpty()
-	},
-}
-
 func (t BlockType) String() string {
 	switch t {
 	case BlockTypeUnknown:
@@ -35,21 +28,90 @@ func (t BlockType) String() string {
 	}
 }
 
-type TokenResultStatus int32
+type TokenResultStatus uint8
 
 const (
 	ResultStatusPass TokenResultStatus = iota
 	ResultStatusBlocked
 	ResultStatusShouldWait
 )
 
+func (s TokenResultStatus) String() string {
+	switch s {
+	case ResultStatusPass:
+		return "ResultStatusPass"
+	case ResultStatusBlocked:
+		return "ResultStatusBlocked"
+	case ResultStatusShouldWait:
+		return "ResultStatusShouldWait"
+	default:
+		return "Undefined"
+	}
+}
+
 type TokenResult struct {
 	status TokenResultStatus
 
 	blockErr *BlockError
 	waitMs   uint64
 }
 
+func (r *TokenResult) DeepCopyFrom(newResult *TokenResult) {
+	r.status = newResult.status
+	r.waitMs = newResult.waitMs
+	if r.blockErr == nil {
+		r.blockErr = &BlockError{
+			blockType:     newResult.blockErr.blockType,
+			blockMsg:      newResult.blockErr.blockMsg,
+			rule:          newResult.blockErr.rule,
+			snapshotValue: newResult.blockErr.snapshotValue,
+		}
+	} else {
+		r.blockErr.blockType = newResult.blockErr.blockType
+		r.blockErr.blockMsg = newResult.blockErr.blockMsg
+		r.blockErr.rule = newResult.blockErr.rule
+		r.blockErr.snapshotValue = newResult.blockErr.snapshotValue
+	}
+}
+
+func (r *TokenResult) ResetToPass() {
+	r.status = ResultStatusPass
+	r.blockErr = nil
+	r.waitMs = 0
+}
+
+func (r *TokenResult) ResetToBlockedFrom(blockType BlockType, blockMsg string) {
+	r.status = ResultStatusBlocked
+	if r.blockErr == nil {
+		r.blockErr = &BlockError{
+			blockType: blockType,
+			blockMsg:  blockMsg,
+		}
+	} else {
+		r.blockErr.blockType = blockType
+		r.blockErr.blockMsg = blockMsg
+	}
+	r.waitMs = 0
+}
+
+func (r *TokenResult) ResetToBlockedWithCauseFrom(blockType BlockType, blockMsg string, rule SentinelRule, snapshot interface{}) {
+	r.status = ResultStatusBlocked
+	if r.blockErr == nil {
+		r.blockErr = &BlockError{
+			blockType:     blockType,
+			blockMsg:      blockMsg,
+			rule:          rule,
+			snapshotValue: snapshot,
+		}
+	} else {
+		r.blockErr.blockType = blockType
+		r.blockErr.blockMsg = blockMsg
+		r.blockErr.rule = rule
+		r.blockErr.snapshotValue = snapshot
+	}
+	r.waitMs = 0
+}
+
 func (r *TokenResult) IsPass() bool {
 	return r.status == ResultStatusPass
 }
@@ -77,48 +139,45 @@ func (r *TokenResult) String() string {
 	} else {
 		blockMsg = r.blockErr.Error()
 	}
-	return fmt.Sprintf("TokenResult{status=%d, blockErr=%s, waitMs=%d}", r.status, blockMsg, r.waitMs)
+	return fmt.Sprintf("TokenResult{status=%+v, blockErr=%s, waitMs=%d}", r.status, blockMsg, r.waitMs)
 }
 
-func NewTokenResultEmpty() *TokenResult {
+func NewTokenResultPass() *TokenResult {
 	return &TokenResult{
 		status:   ResultStatusPass,
 		blockErr: nil,
 		waitMs:   0,
 	}
 }
 
-func NewTokenResultPass() *TokenResult {
-	return resultPool.Get().(*TokenResult)
-}
-
 func NewTokenResultBlocked(blockType BlockType, blockMsg string) *TokenResult {
-	result := resultPool.Get().(*TokenResult)
-	result.status = ResultStatusBlocked
-	result.blockErr = NewBlockError(blockType, blockMsg)
-	return result
+	return &TokenResult{
+		status: ResultStatusBlocked,
+		blockErr: &BlockError{
+			blockType: blockType,
+			blockMsg:  blockMsg,
+		},
+		waitMs: 0,
+	}
 }
 
 func NewTokenResultBlockedWithCause(blockType BlockType, blockMsg string, rule SentinelRule, snapshot interface{}) *TokenResult {
-	result := resultPool.Get().(*TokenResult)
-	result.status = ResultStatusBlocked
-	result.blockErr = NewBlockErrorWithCause(blockType, blockMsg, rule, snapshot)
-	return result
+	return &TokenResult{
+		status: ResultStatusBlocked,
+		blockErr: &BlockError{
+			blockType:     blockType,
+			blockMsg:      blockMsg,
+			rule:          rule,
+			snapshotValue: snapshot,
+		},
+		waitMs: 0,
+	}
 }
 
 func NewTokenResultShouldWait(waitMs uint64) *TokenResult {
-	result := resultPool.Get().(*TokenResult)
-	result.status = ResultStatusShouldWait
-	result.waitMs = waitMs
-	return result
-}
-
-func RefurbishTokenResult(result *TokenResult) {
-	if result != nil {
-		result.status = ResultStatusPass
-		result.blockErr = nil
-		result.waitMs = 0
-
-		resultPool.Put(result)
+	return &TokenResult{
+		status:   ResultStatusShouldWait,
+		blockErr: nil,
+		waitMs:   waitMs,
 	}
 }