Change persistent queue to not use sized channel, improve memory usag…

…e and simplify sized_channel. (#12060)

Signed-off-by: Bogdan Drutu <[email protected]>

.chloggen/change-persistent-queue.yaml

@@ -0,0 +1,25 @@
+# Use this changelog template to create an entry for release notes.
+
+# One of 'breaking', 'deprecation', 'new_component', 'enhancement', 'bug_fix'
+change_type: enhancement
+
+# The name of the component, or a single word describing the area of concern, (e.g. otlpreceiver)
+component: exporterhelper
+
+# A brief description of the change.  Surround your text with quotes ("") if it needs to start with a backtick (`).
+note: Change persistent queue to not use sized channel, improve memory usage and simplify sized_channel.
+
+# One or more tracking issues or pull requests related to the change
+issues: [12060]
+
+# (Optional) One or more lines of additional information to render under the primary note.
+# These lines will be padded with 2 spaces and then inserted directly into the document.
+# Use pipe (|) for multiline entries.
+subtext:
+
+# Optional: The change log or logs in which this entry should be included.
+# e.g. '[user]' or '[user, api]'
+# Include 'user' if the change is relevant to end users.
+# Include 'api' if there is a change to a library API.
+# Default: '[user]'
+change_logs: [api]

exporter/exporterqueue/bounded_memory_queue.go

@@ -17,7 +17,6 @@ import (
 type boundedMemoryQueue[T any] struct {
 	component.StartFunc
 	*sizedChannel[memQueueEl[T]]
-	sizer sizer[T]
 }
 
 // memoryQueueSettings defines internal parameters for boundedMemoryQueue creation.
@@ -30,18 +29,17 @@ type memoryQueueSettings[T any] struct {
 // callback for dropped items (e.g. useful to emit metrics).
 func newBoundedMemoryQueue[T any](set memoryQueueSettings[T]) Queue[T] {
 	return &boundedMemoryQueue[T]{
-		sizedChannel: newSizedChannel[memQueueEl[T]](set.capacity, nil, 0),
-		sizer:        set.sizer,
+		sizedChannel: newSizedChannel[memQueueEl[T]](set.capacity, memQueueElSizer[T]{sizer: set.sizer}),
 	}
 }
 
 // Offer is used by the producer to submit new item to the queue. Calling this method on a stopped queue will panic.
 func (q *boundedMemoryQueue[T]) Offer(ctx context.Context, req T) error {
-	return q.sizedChannel.push(memQueueEl[T]{ctx: ctx, req: req}, q.sizer.Sizeof(req), nil)
+	return q.sizedChannel.push(memQueueEl[T]{ctx: ctx, req: req})
 }
 
 func (q *boundedMemoryQueue[T]) Read(_ context.Context) (uint64, context.Context, T, bool) {
-	item, ok := q.sizedChannel.pop(func(el memQueueEl[T]) int64 { return q.sizer.Sizeof(el.req) })
+	item, ok := q.sizedChannel.pop()
 	return 0, item.ctx, item.req, ok
 }
 
@@ -60,3 +58,11 @@ type memQueueEl[T any] struct {
 	req T
 	ctx context.Context
 }
+
+type memQueueElSizer[T any] struct {
+	sizer sizer[T]
+}
+
+func (mqes memQueueElSizer[T]) Sizeof(el memQueueEl[T]) int64 {
+	return mqes.sizer.Sizeof(el.req)
+}

exporter/exporterqueue/bounded_memory_queue_test.go

@@ -16,6 +16,8 @@ import (
 	"github.com/stretchr/testify/require"
 
 	"go.opentelemetry.io/collector/component/componenttest"
+	"go.opentelemetry.io/collector/pdata/ptrace"
+	"go.opentelemetry.io/collector/pdata/testdata"
 )
 
 // In this test we run a queue with capacity 1 and a single consumer.
@@ -98,49 +100,50 @@ func TestShutdownWhileNotEmpty(t *testing.T) {
 }
 
 func Benchmark_QueueUsage_1000_requests(b *testing.B) {
-	benchmarkQueueUsage(b, &requestSizer[fakeReq]{}, 1000)
+	benchmarkQueueUsage(b, &requestSizer[ptrace.Traces]{}, 1000)
 }
 
 func Benchmark_QueueUsage_100000_requests(b *testing.B) {
-	benchmarkQueueUsage(b, &requestSizer[fakeReq]{}, 100000)
+	benchmarkQueueUsage(b, &requestSizer[ptrace.Traces]{}, 100000)
 }
 
 func Benchmark_QueueUsage_10000_items(b *testing.B) {
 	// each request has 10 items: 1000 requests = 10000 items
-	benchmarkQueueUsage(b, &itemsSizer[fakeReq]{}, 1000)
+	benchmarkQueueUsage(b, &itemsSizer{}, 1000)
 }
 
 func Benchmark_QueueUsage_1M_items(b *testing.B) {
 	// each request has 10 items: 100000 requests = 1M items
-	benchmarkQueueUsage(b, &itemsSizer[fakeReq]{}, 100000)
+	benchmarkQueueUsage(b, &itemsSizer{}, 100000)
 }
 
 func TestQueueUsage(t *testing.T) {
 	t.Run("requests_based", func(t *testing.T) {
-		queueUsage(t, &requestSizer[fakeReq]{}, 10)
+		queueUsage(t, &requestSizer[ptrace.Traces]{}, 10)
 	})
 	t.Run("items_based", func(t *testing.T) {
-		queueUsage(t, &itemsSizer[fakeReq]{}, 10)
+		queueUsage(t, &itemsSizer{}, 10)
 	})
 }
 
-func benchmarkQueueUsage(b *testing.B, sizer sizer[fakeReq], requestsCount int) {
+func benchmarkQueueUsage(b *testing.B, sizer sizer[ptrace.Traces], requestsCount int) {
 	b.ReportAllocs()
 	for i := 0; i < b.N; i++ {
 		queueUsage(b, sizer, requestsCount)
 	}
 }
 
-func queueUsage(tb testing.TB, sizer sizer[fakeReq], requestsCount int) {
-	q := newBoundedMemoryQueue[fakeReq](memoryQueueSettings[fakeReq]{sizer: sizer, capacity: int64(10 * requestsCount)})
+func queueUsage(tb testing.TB, sizer sizer[ptrace.Traces], requestsCount int) {
+	q := newBoundedMemoryQueue[ptrace.Traces](memoryQueueSettings[ptrace.Traces]{sizer: sizer, capacity: int64(10 * requestsCount)})
 	consumed := &atomic.Int64{}
 	require.NoError(tb, q.Start(context.Background(), componenttest.NewNopHost()))
-	ac := newAsyncConsumer(q, 1, func(context.Context, fakeReq) error {
+	ac := newAsyncConsumer(q, 1, func(context.Context, ptrace.Traces) error {
 		consumed.Add(1)
 		return nil
 	})
+	td := testdata.GenerateTraces(10)
 	for j := 0; j < requestsCount; j++ {
-		require.NoError(tb, q.Offer(context.Background(), fakeReq{10}))
+		require.NoError(tb, q.Offer(context.Background(), td))
 	}
 	assert.NoError(tb, q.Shutdown(context.Background()))
 	assert.NoError(tb, ac.Shutdown(context.Background()))
@@ -158,14 +161,6 @@ func TestZeroSizeNoConsumers(t *testing.T) {
 	assert.NoError(t, q.Shutdown(context.Background()))
 }
 
-type fakeReq struct {
-	itemsCount int
-}
-
-func (r fakeReq) ItemsCount() int {
-	return r.itemsCount
-}
-
 func consume[T any](q Queue[T], consumeFunc func(context.Context, T) error) bool {
 	index, ctx, req, ok := q.Read(context.Background())
 	if !ok {