Data race on BatchSpanProcessor

[peolivei2]

We're seeing the data race below for SDK 1.14.0:

[dmathieu]

Could you provide a reproduction example?

[peolivei2]

@dmathieu the problem here is the approach the SDK is taking for handling the close/send of the queue channel. It's not guaranteeing that it won't send to the channel after closing, but instead relying on a recover after a panic. That's a valid data race as documented here:

• https://go.dev/doc/articles/race_detector#Unsynchronized_send_and_close_operations
• runtime: close of a channel with pending sends incorrectly flagged by race detector golang/go#30372

It doesn't repro consistently because of another race condition in the SDK:

My suggestion here would be to not close the queue channel at all, then combine the selects above so you just atop adding to the queue after the shutdown.

[dmathieu]

Combining the selects wouldn't be a reliable solution either, as Go doesn't run select/case statements in the order they are specified, but in shuffle order.
So every time, we could be appending to the queue again and returning true, or detect the stopCh has been closed, and return false.

See https://medium.com/a-journey-with-go/go-ordering-in-select-statements-fd0ff80fd8d6

[pellared]

I quickly looked at the code and it uses channels improperly.

The "consumer" goroutine is closing the channel:

opentelemetry-go/sdk/trace/batch_span_processor.go

Lines 321 to 350 in fd5284f

	// drainQueue awaits the any caller that had added to bsp.stopWait
	// to finish the enqueue, then exports the final batch.
	func (bsp *batchSpanProcessor) drainQueue() {
	ctx, cancel := context.WithCancel(context.Background())
	defer cancel()
	for {
	select {
	case sd := <-bsp.queue:
	if sd == nil {
	if err := bsp.exportSpans(ctx); err != nil {
	otel.Handle(err)
	}
	return
	}
	bsp.batchMutex.Lock()
	bsp.batch = append(bsp.batch, sd)
	shouldExport := len(bsp.batch) == bsp.o.MaxExportBatchSize
	bsp.batchMutex.Unlock()
	if shouldExport {
	if err := bsp.exportSpans(ctx); err != nil {
	otel.Handle(err)
	}
	}
	default:
	close(bsp.queue)
	}
	}
	}

Moreover, the "producer" goroutine has some hacks to prevent panics if it sends to a closed channel.:

opentelemetry-go/sdk/trace/batch_span_processor.go

Lines 361 to 395 in fd5284f

	func recoverSendOnClosedChan() {
	x := recover()
	switch err := x.(type) {
	case nil:
	return
	case runtime.Error:
	if err.Error() == "send on closed channel" {
	return
	}
	}
	panic(x)
	}
	func (bsp *batchSpanProcessor) enqueueBlockOnQueueFull(ctx context.Context, sd ReadOnlySpan) bool {
	if !sd.SpanContext().IsSampled() {
	return false
	}
	// This ensures the bsp.queue<- below does not panic as the
	// processor shuts down.
	defer recoverSendOnClosedChan()
	select {
	case <-bsp.stopCh:
	return false
	default:
	}
	select {
	case bsp.queue <- sd:
	return true
	case <-ctx.Done():
	return false
	}
	}

I guess there may be more problems, but it looks clear that the synchronization is not implemented properly.

Reference: https://go101.org/article/channel-closing.html