Change the memory queue implementation to not pre-allocate capacity objects.

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

Author: bogdandrutu

.chloggen/sized-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 the memory queue implementation to not pre-allocate capacity objects.
+
+# One or more tracking issues or pull requests related to the change
+issues: [12070]
+
+# (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: This change improves memory usage of the collector under low utilization and is a prerequisite for supporting different other size limitations (number of items, bytes).
+
+# 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: [user]

exporter/exporterqueue/bounded_memory_queue.go

@@ -16,7 +16,7 @@ import (
 // the producer are dropped.
 type boundedMemoryQueue[T any] struct {
 	component.StartFunc
-	*sizedChannel[memQueueEl[T]]
+	*sizedQueue[memQueueEl[T]]
 }
 
 // memoryQueueSettings defines internal parameters for boundedMemoryQueue creation.
@@ -29,17 +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, memQueueElSizer[T]{sizer: set.sizer}),
+		sizedQueue: newSizedQueue[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})
+	return q.sizedQueue.push(memQueueEl[T]{ctx: ctx, req: req})
 }
 
 func (q *boundedMemoryQueue[T]) Read(_ context.Context) (uint64, context.Context, T, bool) {
-	item, ok := q.sizedChannel.pop()
+	item, ok := q.sizedQueue.pop()
 	return 0, item.ctx, item.req, ok
 }
 
@@ -50,7 +50,7 @@ func (q *boundedMemoryQueue[T]) OnProcessingFinished(uint64, error) {
 
 // Shutdown closes the queue channel to initiate draining of the queue.
 func (q *boundedMemoryQueue[T]) Shutdown(context.Context) error {
-	q.sizedChannel.shutdown()
+	q.sizedQueue.shutdown()
 	return nil
 }
 

exporter/exporterqueue/sized_channel.go

@@ -0,0 +1,135 @@
+// Copyright The OpenTelemetry Authors
+// SPDX-License-Identifier: Apache-2.0
+
+package exporterqueue // import "go.opentelemetry.io/collector/exporter/exporterqueue"
+
+import (
+	"errors"
+	"sync"
+)
+
+var errInvalidSize = errors.New("invalid element size")
+
+type node[T any] struct {
+	data T
+	size int64
+	next *node[T]
+}
+
+type linkedQueue[T any] struct {
+	head *node[T]
+	tail *node[T]
+}
+
+func (l *linkedQueue[T]) push(data T, size int64) {
+	n := &node[T]{data: data, size: size}
+	if l.tail == nil {
+		l.head = n
+		l.tail = n
+		return
+	}
+	l.tail.next = n
+	l.tail = n
+}
+
+func (l *linkedQueue[T]) pop() (T, int64) {
+	n := l.head
+	l.head = n.next
+	if l.head == nil {
+		l.tail = nil
+	}
+	n.next = nil
+	return n.data, n.size
+}
+
+// sizedQueue is a channel wrapper for sized elements with a capacity set to a total size of all the elements.
+// The channel will accept elements until the total size of the elements reaches the capacity.
+type sizedQueue[T any] struct {
+	sizer sizer[T]
+	cap   int64
+
+	mu          sync.Mutex
+	hasElements *sync.Cond
+	items       *linkedQueue[T]
+	size        int64
+	stopped     bool
+}
+
+// newSizedQueue creates a sized elements channel. Each element is assigned a size by the provided sizer.
+// capacity is the capacity of the queue.
+func newSizedQueue[T any](capacity int64, sizer sizer[T]) *sizedQueue[T] {
+	sq := &sizedQueue[T]{
+		sizer: sizer,
+		cap:   capacity,
+		items: &linkedQueue[T]{},
+	}
+	sq.hasElements = sync.NewCond(&sq.mu)
+	return sq
+}
+
+// push puts the element into the queue with the given sized if there is enough capacity.
+// Returns an error if the queue is full.
+func (sq *sizedQueue[T]) push(el T) error {
+	elSize := sq.sizer.Sizeof(el)
+	if elSize == 0 {
+		return nil
+	}
+
+	if elSize <= 0 {
+		return errInvalidSize
+	}
+
+	sq.mu.Lock()
+	defer sq.mu.Unlock()
+
+	if sq.size+elSize > sq.cap {
+		return ErrQueueIsFull
+	}
+
+	sq.size += elSize
+	sq.items.push(el, elSize)
+	// Signal one consumer if any.
+	sq.hasElements.Signal()
+	return nil
+}
+
+// pop removes the element from the queue and returns it.
+// The call blocks until there is an item available or the queue is stopped.
+// The function returns true when an item is consumed or false if the queue is stopped and emptied.
+func (sq *sizedQueue[T]) pop() (T, bool) {
+	sq.mu.Lock()
+	defer sq.mu.Unlock()
+
+	for {
+		if sq.size > 0 {
+			el, elSize := sq.items.pop()
+			sq.size -= elSize
+			return el, true
+		}
+
+		if sq.stopped {
+			var el T
+			return el, false
+		}
+
+		sq.hasElements.Wait()
+	}
+}
+
+// shutdown closes the queue channel to initiate draining of the queue.
+func (sq *sizedQueue[T]) shutdown() {
+	sq.mu.Lock()
+	defer sq.mu.Unlock()
+	sq.stopped = true
+	sq.hasElements.Broadcast()
+}
+
+func (sq *sizedQueue[T]) Size() int {
+	sq.mu.Lock()
+	defer sq.mu.Unlock()
+	return int(sq.size)
+}
+
+func (sq *sizedQueue[T]) Capacity() int {
+	return int(sq.cap)
+}

exporter/exporterqueue/sized_queue.go

@@ -16,8 +16,8 @@ func (s sizerInt) Sizeof(el int) int64 {
 	return int64(el)
 }
 
-func TestSizedChannel(t *testing.T) {
-	q := newSizedChannel[int](7, sizerInt{})
+func TestSizedQueue(t *testing.T) {
+	q := newSizedQueue[int](7, sizerInt{})
 	require.NoError(t, q.push(1))
 	assert.Equal(t, 1, q.Size())
 	assert.Equal(t, 7, q.Capacity())
@@ -45,7 +45,38 @@ func TestSizedChannel(t *testing.T) {
 	assert.Equal(t, 0, el)
 }
 
+func TestSizedQueue_DrainAllElements(t *testing.T) {
+	q := newSizedQueue[int](7, sizerInt{})
+	require.NoError(t, q.push(1))
+	require.NoError(t, q.push(3))
+
+	el, ok := q.pop()
+	assert.Equal(t, 1, el)
+	assert.True(t, ok)
+	assert.Equal(t, 3, q.Size())
+
+	q.shutdown()
+	el, ok = q.pop()
+	assert.Equal(t, 3, el)
+	assert.True(t, ok)
+	assert.Equal(t, 0, q.Size())
+
+	el, ok = q.pop()
+	assert.False(t, ok)
+	assert.Equal(t, 0, el)
+}
+
 func TestSizedChannel_OfferInvalidSize(t *testing.T) {
-	q := newSizedChannel[int](1, sizerInt{})
-	require.ErrorIs(t, q.push(0), errInvalidSize)
+	q := newSizedQueue[int](1, sizerInt{})
+	require.ErrorIs(t, q.push(-1), errInvalidSize)
+}
+
+func TestSizedChannel_OfferZeroSize(t *testing.T) {
+	q := newSizedQueue[int](1, sizerInt{})
+	require.NoError(t, q.push(0))
+	q.shutdown()
+	// Because the size 0 is ignored, nothing to drain.
+	el, ok := q.pop()
+	assert.False(t, ok)
+	assert.Equal(t, 0, el)
 }