fix

Author: JqRrt

internal/trace/queue_manager.go

@@ -57,6 +57,7 @@ func newBatchQueueManager(o batchQueueManagerOptions) *BatchQueueManager {
 		dropped:    0,
 		batch:      make([]interface{}, 0, o.maxExportBatchLength),
 		batchMutex: sync.Mutex{},
+		sizeMutex:  sync.RWMutex{},
 		timer:      time.NewTimer(o.batchTimeout),
 		exportFunc: o.exportFunc,
 		stopWait:   sync.WaitGroup{},
@@ -85,6 +86,7 @@ type BatchQueueManager struct {
 	batch         []interface{}
 	batchByteSize int64
 	batchMutex    sync.Mutex
+	sizeMutex     sync.RWMutex
 	timer         *time.Timer
 
 	exportFunc func(ctx context.Context, s []interface{})
@@ -137,9 +139,12 @@ func (b *BatchQueueManager) isShouldExport() bool {
 	if len(b.batch) >= b.o.maxExportBatchLength {
 		return true
 	}
+
+	b.sizeMutex.RLock()
 	if b.batchByteSize >= int64(b.o.maxExportBatchByteSize) {
 		return true
 	}
+	b.sizeMutex.RUnlock()
 
 	return false
 }
@@ -182,7 +187,9 @@ func (b *BatchQueueManager) doExport(ctx context.Context) {
 		}
 		// delete the batch
 		b.batch = b.batch[:0]
+		b.sizeMutex.Lock()
 		b.batchByteSize = 0
+		b.sizeMutex.Unlock()
 	}
 }
 
@@ -197,9 +204,9 @@ func (b *BatchQueueManager) Enqueue(ctx context.Context, sd interface{}, byteSiz
 	var isFail bool
 	select {
 	case b.queue <- sd:
-		b.batchMutex.Lock()
+		b.sizeMutex.Lock()
 		b.batchByteSize += byteSize
-		b.batchMutex.Unlock()
+		b.sizeMutex.Unlock()
 		detailMsg = fmt.Sprintf("%s enqueue, queue length: %d", b.o.queueName, len(b.queue))
 	default: // queue is full, not block, drop
 		detailMsg = fmt.Sprintf("%s queue is full, dropped item", b.o.queueName)

internal/trace/span_processor.go

@@ -38,8 +38,8 @@ const (
 )
 
 type QueueConf struct {
-	SpanQueueLength               int
-	SpanQueueMaxExportBatchLength int
+	SpanQueueLength          int
+	SpanMaxExportBatchLength int
 }
 
 var _ SpanProcessor = (*BatchSpanProcessor)(nil)
@@ -79,13 +79,13 @@ func NewBatchSpanProcessor(
 		exporter = ex
 	}
 	var spanQueueLength = DefaultMaxQueueLength
-	var spanQueueMaxExportBatchLength = DefaultMaxExportBatchLength
+	var spanMaxExportBatchLength = DefaultMaxExportBatchLength
 	if queueConf != nil {
 		if queueConf.SpanQueueLength > 0 {
 			spanQueueLength = queueConf.SpanQueueLength
 		}
-		if queueConf.SpanQueueMaxExportBatchLength > 0 { // todo: need max limit
-			spanQueueMaxExportBatchLength = queueConf.SpanQueueMaxExportBatchLength
+		if queueConf.SpanMaxExportBatchLength > 0 { // todo: need max limit
+			spanMaxExportBatchLength = queueConf.SpanMaxExportBatchLength
 		}
 	}
 
@@ -126,7 +126,7 @@ func NewBatchSpanProcessor(
 			queueName:              queueNameSpan,
 			batchTimeout:           time.Duration(DefaultScheduleDelay) * time.Millisecond,
 			maxQueueLength:         spanQueueLength,
-			maxExportBatchLength:   spanQueueMaxExportBatchLength,
+			maxExportBatchLength:   spanMaxExportBatchLength,
 			maxExportBatchByteSize: DefaultMaxExportBatchByteSize,
 			exportFunc:             newExportSpansFunc(exporter, spanRetryQM, fileQM, finishEventProcessor),
 			finishEventProcessor:   finishEventProcessor,