Add mutex to fix some data race (#227)

Author: mrproliu

.github/workflows/plugin-tests.yaml

@@ -80,7 +80,7 @@ jobs:
           - go-redisv9
           - go-restfulv3
           - gorm
-          - kratosv2
+#          - kratosv2 temporary disable because it's not stable
           - microv4
           - mongo
           - mysql

CHANGES.md

@@ -9,6 +9,7 @@ Release Notes.
 * Support Windows plugin test.
 * Support Kafka reporter.
 * Add recover to goroutine to prevent unexpected panics.
+* Add mutex to fix some data race. 
 
 #### Plugins
 

plugins/core/context.go

@@ -17,7 +17,10 @@
 
 package core
 
-import "reflect"
+import (
+	"reflect"
+	"sync"
+)
 
 var (
 	GetGLS            = func() interface{} { return nil }
@@ -29,16 +32,21 @@ var (
 )
 
 type ContextSnapshoter interface {
-	TakeSnapShot(val interface{}) interface{}
+	TakeSnapShot() interface{}
 }
 
 type TracingContext struct {
 	activeSpan TracingSpan
 	Runtime    *RuntimeContext
 	ID         *IDContext
+
+	activeSpanLock sync.RWMutex
 }
 
-func (t *TracingContext) TakeSnapShot(val interface{}) interface{} {
+func (t *TracingContext) TakeSnapShot() interface{} {
+	if t == nil {
+		return nil
+	}
 	snapshot := newSnapshotSpan(t.ActiveSpan())
 	return &TracingContext{
 		activeSpan: snapshot,
@@ -48,13 +56,17 @@ func (t *TracingContext) TakeSnapShot(val interface{}) interface{} {
 }
 
 func (t *TracingContext) ActiveSpan() TracingSpan {
+	t.activeSpanLock.RLock()
+	defer t.activeSpanLock.RUnlock()
 	if t.activeSpan == nil || reflect.ValueOf(t.activeSpan).IsZero() {
 		return nil
 	}
 	return t.activeSpan
 }
 
 func (t *TracingContext) SaveActiveSpan(s TracingSpan) {
+	t.activeSpanLock.Lock()
+	defer t.activeSpanLock.Unlock()
 	t.activeSpan = s
 }
 

plugins/core/metrics.go

@@ -75,7 +75,7 @@ func (t *Tracer) reachNotInitMetrics() {
 func (t *Tracer) sendMetrics() {
 	meters := make([]reporter.ReportedMeter, 0)
 	// call collect hook
-	for _, hook := range t.meterCollectListeners {
+	for _, hook := range t.allMeterCollectListeners() {
 		hook()
 	}
 	t.meterMap.Range(func(key, value interface{}) bool {
@@ -90,6 +90,14 @@ func (t *Tracer) sendMetrics() {
 	t.Reporter.SendMetrics(meters)
 }
 
+func (t *Tracer) allMeterCollectListeners() []func() {
+	t.meterCollectListenersLock.RLock()
+	defer t.meterCollectListenersLock.RUnlock()
+	listeners := make([]func(), 0, len(t.meterCollectListeners))
+	listeners = append(listeners, t.meterCollectListeners...)
+	return listeners
+}
+
 func (t *Tracer) NewCounter(name string, opt interface{}) interface{} {
 	counter := newCounter(name, nil, 0)
 	if o, ok := opt.(meterOpts); ok && o != nil {
@@ -118,6 +126,8 @@ func (t *Tracer) NewHistogram(name string, minValue float64, steps []float64, op
 }
 
 func (t *Tracer) AddCollectHook(f func()) {
+	t.meterCollectListenersLock.Lock()
+	defer t.meterCollectListenersLock.Unlock()
 	t.meterCollectListeners = append(t.meterCollectListeners, f)
 }
 
@@ -347,7 +357,7 @@ func (h *histogramBucket) Bucket() float64 {
 }
 
 func (h *histogramBucket) Count() int64 {
-	return *h.value
+	return atomic.LoadInt64(h.value)
 }
 
 func (h *histogramBucket) IsNegativeInfinity() bool {

plugins/core/operator/invocation.go

@@ -27,6 +27,10 @@ type realInvocation struct {
 	returnValues []interface{}
 
 	context interface{}
+
+	// self obs
+	interTimeCost    int64
+	beforeInterStart int64
 }
 
 func (i *realInvocation) CallerInstance() interface{} {

plugins/core/sampler.go

@@ -107,10 +107,13 @@ type DynamicSampler struct {
 	currentRate float64
 	defaultRate float64
 	sampler     Sampler
+	locker      sync.RWMutex
 }
 
 // IsSampled implements IsSampled() of Sampler.
 func (s *DynamicSampler) IsSampled(operation string) bool {
+	s.locker.RLock()
+	defer s.locker.RUnlock()
 	return s.sampler.IsSampled(operation)
 }
 
@@ -136,11 +139,15 @@ func (s *DynamicSampler) Notify(eventType reporter.AgentConfigEventType, newValu
 	} else {
 		sampler = NewRandomSampler(samplingRate)
 	}
+	s.locker.Lock()
+	defer s.locker.Unlock()
 	s.sampler = sampler
 	s.currentRate = samplingRate
 }
 
 func (s *DynamicSampler) Value() string {
+	s.locker.RLock()
+	defer s.locker.RUnlock()
 	return fmt.Sprintf("%f", s.currentRate)
 }
 

plugins/core/span_tracing.go

@@ -255,12 +255,12 @@ func (rs *RootSegmentSpan) AsyncFinish() {
 }
 
 func (rs *RootSegmentSpan) end0() {
-	go func() {
-		defer func() {
-			_ = recover()
-		}()
-		rs.doneCh <- atomic.SwapInt32(rs.SegmentContext.refNum, -1)
+	defer func() {
+		_ = recover()
 	}()
+	if rs != nil && rs.doneCh != nil && rs.SegmentContext.refNum != nil {
+		rs.doneCh <- atomic.SwapInt32(rs.SegmentContext.refNum, -1)
+	}
 }
 
 func (rs *RootSegmentSpan) createRootSegmentContext(ctx *TracingContext, _ SegmentSpan) (err error) {

plugins/core/test_base.go

@@ -57,7 +57,7 @@ func SetAsNewGoroutine() {
 		return
 	}
 	if e := gls.(ContextSnapshoter); e != nil {
-		SetGLS(e.TakeSnapShot(GetGLS()))
+		SetGLS(e.TakeSnapShot())
 	}
 }
 

plugins/core/tracer.go

@@ -49,11 +49,12 @@ type Tracer struct {
 	// for plugin tools
 	tools *TracerTools
 	// for all metrics
-	meterMap              *sync.Map
-	meterCollectListeners []func()
-	ignoreSuffix          []string
-	traceIgnorePath       []string
-	mu                    sync.Mutex
+	meterMap                  *sync.Map
+	meterCollectListeners     []func()
+	meterCollectListenersLock sync.RWMutex
+	ignoreSuffix              []string
+	traceIgnorePath           []string
+	mu                        sync.Mutex
 }
 
 func (t *Tracer) Init(entity *reporter.Entity, rep reporter.Reporter, samp Sampler, logger operator.LogOperator,

plugins/core/tracing.go

@@ -172,6 +172,8 @@ func (t *Tracer) ContinueContext(snapshot interface{}) {
 			ctx = NewTracingContext()
 			SetGLS(ctx)
 		}
+		ctx.activeSpanLock.Lock()
+		defer ctx.activeSpanLock.Unlock()
 		ctx.activeSpan = snap.activeSpan
 		ctx.Runtime = snap.runtime
 	}