Merge branch 'main' into add-featuregate-examples

Author: parv18050212

cmd/otelcorecol/go.mod

@@ -74,7 +74,6 @@ require (
 	github.com/prometheus/client_model v0.6.2 // indirect
 	github.com/prometheus/common v0.64.0 // indirect
 	github.com/prometheus/procfs v0.16.1 // indirect
-	github.com/puzpuzpuz/xsync/v3 v3.5.1 // indirect
 	github.com/rs/cors v1.11.1 // indirect
 	github.com/shirou/gopsutil/v4 v4.25.5 // indirect
 	github.com/spf13/cobra v1.9.1 // indirect

cmd/otelcorecol/go.sum

@@ -40,7 +40,6 @@ require (
 	github.com/modern-go/concurrent v0.0.0-20180306012644-bacd9c7ef1dd // indirect
 	github.com/modern-go/reflect2 v1.0.2 // indirect
 	github.com/pmezard/go-difflib v1.0.0 // indirect
-	github.com/puzpuzpuz/xsync/v3 v3.5.1 // indirect
 	go.opentelemetry.io/auto/sdk v1.1.0 // indirect
 	go.opentelemetry.io/collector/config/configretry v1.33.0 // indirect
 	go.opentelemetry.io/collector/consumer/consumererror v0.127.0 // indirect

exporter/debugexporter/go.mod

@@ -6,56 +6,45 @@ import (
 	"context"
 	"sync"
 
-	"github.com/puzpuzpuz/xsync/v3"
-
 	"go.opentelemetry.io/collector/component"
 	"go.opentelemetry.io/collector/exporter/exporterhelper/internal/request"
 	"go.opentelemetry.io/collector/exporter/exporterhelper/internal/sender"
 )
 
+type batcherSettings[T any] struct {
+	sizerType   request.SizerType
+	sizer       request.Sizer[T]
+	partitioner Partitioner[T]
+	next        sender.SendFunc[T]
+	maxWorkers  int
+}
+
 type multiBatcher struct {
 	cfg         BatchConfig
-	workerPool  chan struct{}
+	wp          *workerPool
 	sizerType   request.SizerType
 	sizer       request.Sizer[request.Request]
 	partitioner Partitioner[request.Request]
 	consumeFunc sender.SendFunc[request.Request]
 
 	singleShard *shardBatcher
-	shards      *xsync.MapOf[string, *shardBatcher]
+	shards      sync.Map
 }
 
 var _ Batcher[request.Request] = (*multiBatcher)(nil)
 
 func newMultiBatcher(bCfg BatchConfig, bSet batcherSettings[request.Request]) *multiBatcher {
-	var workerPool chan struct{}
-	if bSet.maxWorkers != 0 {
-		workerPool = make(chan struct{}, bSet.maxWorkers)
-		for i := 0; i < bSet.maxWorkers; i++ {
-			workerPool <- struct{}{}
-		}
-	}
 	mb := &multiBatcher{
 		cfg:         bCfg,
-		workerPool:  workerPool,
+		wp:          newWorkerPool(bSet.maxWorkers),
 		sizerType:   bSet.sizerType,
 		sizer:       bSet.sizer,
 		partitioner: bSet.partitioner,
 		consumeFunc: bSet.next,
 	}
 
 	if bSet.partitioner == nil {
-		mb.singleShard = &shardBatcher{
-			cfg:         bCfg,
-			workerPool:  mb.workerPool,
-			sizerType:   bSet.sizerType,
-			sizer:       bSet.sizer,
-			consumeFunc: bSet.next,
-			stopWG:      sync.WaitGroup{},
-			shutdownCh:  make(chan struct{}, 1),
-		}
-	} else {
-		mb.shards = xsync.NewMapOf[string, *shardBatcher]()
+		mb.singleShard = newShard(mb.cfg, mb.sizerType, mb.sizer, mb.wp, mb.consumeFunc)
 	}
 	return mb
 }
@@ -66,25 +55,24 @@ func (mb *multiBatcher) getShard(ctx context.Context, req request.Request) *shar
 	}
 
 	key := mb.partitioner.GetKey(ctx, req)
-	result, _ := mb.shards.LoadOrCompute(key, func() *shardBatcher {
-		s := &shardBatcher{
-			cfg:         mb.cfg,
-			workerPool:  mb.workerPool,
-			sizerType:   mb.sizerType,
-			sizer:       mb.sizer,
-			consumeFunc: mb.consumeFunc,
-			stopWG:      sync.WaitGroup{},
-			shutdownCh:  make(chan struct{}, 1),
-		}
-		s.start(ctx, nil)
-		return s
-	})
-	return result
+	// Fast path, shard already created.
+	s, found := mb.shards.Load(key)
+	if found {
+		return s.(*shardBatcher)
+	}
+	newS := newShard(mb.cfg, mb.sizerType, mb.sizer, mb.wp, mb.consumeFunc)
+	_ = newS.Start(ctx, nil)
+	s, loaded := mb.shards.LoadOrStore(key, newS)
+	// If not loaded, there was a race condition in adding the new shard. Shutdown the newly created shard.
+	if loaded {
+		_ = newS.Shutdown(ctx)
+	}
+	return s.(*shardBatcher)
 }
 
 func (mb *multiBatcher) Start(ctx context.Context, host component.Host) error {
 	if mb.singleShard != nil {
-		mb.singleShard.start(ctx, host)
+		return mb.singleShard.Start(ctx, host)
 	}
 	return nil
 }
@@ -96,16 +84,15 @@ func (mb *multiBatcher) Consume(ctx context.Context, req request.Request, done D
 
 func (mb *multiBatcher) Shutdown(ctx context.Context) error {
 	if mb.singleShard != nil {
-		mb.singleShard.shutdown(ctx)
-		return nil
+		return mb.singleShard.Shutdown(ctx)
 	}
 
 	var wg sync.WaitGroup
-	wg.Add(mb.shards.Size())
-	mb.shards.Range(func(_ string, shard *shardBatcher) bool {
+	mb.shards.Range(func(_ any, shard any) bool {
+		wg.Add(1)
 		go func() {
-			shard.shutdown(ctx)
-			wg.Done()
+			defer wg.Done()
+			_ = shard.(*shardBatcher).Shutdown(ctx)
 		}()
 		return true
 	})

exporter/debugexporter/go.sum

@@ -21,18 +21,10 @@ type batch struct {
 	done multiDone
 }
 
-type batcherSettings[T any] struct {
-	sizerType   request.SizerType
-	sizer       request.Sizer[T]
-	partitioner Partitioner[T]
-	next        sender.SendFunc[T]
-	maxWorkers  int
-}
-
 // shardBatcher continuously batch incoming requests and flushes asynchronously if minimum size limit is met or on timeout.
 type shardBatcher struct {
 	cfg            BatchConfig
-	workerPool     chan struct{}
+	wp             *workerPool
 	sizerType      request.SizerType
 	sizer          request.Sizer[request.Request]
 	consumeFunc    sender.SendFunc[request.Request]
@@ -43,6 +35,17 @@ type shardBatcher struct {
 	shutdownCh     chan struct{}
 }
 
+func newShard(cfg BatchConfig, sizerType request.SizerType, sizer request.Sizer[request.Request], wp *workerPool, next sender.SendFunc[request.Request]) *shardBatcher {
+	return &shardBatcher{
+		cfg:         cfg,
+		wp:          wp,
+		sizerType:   sizerType,
+		sizer:       sizer,
+		consumeFunc: next,
+		shutdownCh:  make(chan struct{}, 1),
+	}
+}
+
 func (qb *shardBatcher) resetTimer() {
 	if qb.cfg.FlushTimeout > 0 {
 		qb.timer.Reset(qb.cfg.FlushTimeout)
@@ -88,7 +91,7 @@ func (qb *shardBatcher) Consume(ctx context.Context, req request.Request, done D
 	}
 
 	reqList, mergeSplitErr := qb.currentBatch.req.MergeSplit(ctx, int(qb.cfg.MaxSize), qb.sizerType, req)
-	// If failed to merge signal all Done callbacks from current batch as well as the current request and reset the current batch.
+	// If failed to merge signal all Done callbacks from the current batch as well as the current request and reset the current batch.
 	if mergeSplitErr != nil || len(reqList) == 0 {
 		done.OnDone(mergeSplitErr)
 		qb.currentBatchMu.Unlock()
@@ -102,7 +105,7 @@ func (qb *shardBatcher) Consume(ctx context.Context, req request.Request, done D
 
 	// We have at least one result in the reqList, if more results here is what that means:
 	// - First result will contain items from the current batch + some results from the current request.
-	// - All other results except first will contain items only from current request.
+	// - All other results except first will contain items only from the current request.
 	// - Last result may not have enough data to be flushed.
 
 	// Logic on how to deal with the current batch:
@@ -145,8 +148,12 @@ func (qb *shardBatcher) Consume(ctx context.Context, req request.Request, done D
 	}
 }
 
-// startTimeBasedFlushingGoroutine starts a goroutine that flushes on timeout.
-func (qb *shardBatcher) startTimeBasedFlushingGoroutine() {
+// Start starts the goroutine that reads from the queue and flushes asynchronously.
+func (qb *shardBatcher) Start(context.Context, component.Host) error {
+	if qb.cfg.FlushTimeout <= 0 {
+		return nil
+	}
+	qb.timer = time.NewTimer(qb.cfg.FlushTimeout)
 	qb.stopWG.Add(1)
 	go func() {
 		defer qb.stopWG.Done()
@@ -159,14 +166,16 @@ func (qb *shardBatcher) startTimeBasedFlushingGoroutine() {
 			}
 		}
 	}()
+	return nil
 }
 
-// Start starts the goroutine that reads from the queue and flushes asynchronously.
-func (qb *shardBatcher) start(_ context.Context, _ component.Host) {
-	if qb.cfg.FlushTimeout > 0 {
-		qb.timer = time.NewTimer(qb.cfg.FlushTimeout)
-		qb.startTimeBasedFlushingGoroutine()
-	}
+// Shutdown ensures that queue and all Batcher are stopped.
+func (qb *shardBatcher) Shutdown(context.Context) error {
+	close(qb.shutdownCh)
+	// Make sure execute one last flush if necessary.
+	qb.flushCurrentBatchIfNecessary()
+	qb.stopWG.Wait()
+	return nil
 }
 
 // flushCurrentBatchIfNecessary sends out the current request batch if it is not nil
@@ -188,24 +197,28 @@ func (qb *shardBatcher) flushCurrentBatchIfNecessary() {
 // flush starts a goroutine that calls consumeFunc. It blocks until a worker is available if necessary.
 func (qb *shardBatcher) flush(ctx context.Context, req request.Request, done Done) {
 	qb.stopWG.Add(1)
-	if qb.workerPool != nil {
-		<-qb.workerPool
-	}
-	go func() {
+	qb.wp.execute(func() {
 		defer qb.stopWG.Done()
 		done.OnDone(qb.consumeFunc(ctx, req))
-		if qb.workerPool != nil {
-			qb.workerPool <- struct{}{}
-		}
-	}()
+	})
 }
 
-// Shutdown ensures that queue and all Batcher are stopped.
-func (qb *shardBatcher) shutdown(_ context.Context) {
-	close(qb.shutdownCh)
-	// Make sure execute one last flush if necessary.
-	qb.flushCurrentBatchIfNecessary()
-	qb.stopWG.Wait()
+type workerPool struct {
+	workers chan struct{}
+}
+
+func newWorkerPool(maxWorkers int) *workerPool {
+	workers := make(chan struct{}, maxWorkers)
+	for i := 0; i < maxWorkers; i++ {
+		workers <- struct{}{}
+	}
+	return &workerPool{workers: workers}
+}
+
+func (wp *workerPool) execute(f func()) {
+	<-wp.workers
+	go f()
+	wp.workers <- struct{}{}
 }
 
 type multiDone []Done