[feature request] Improve Protobuff Serialization Strategy: Replace ThreadStatic Array with Segmented IBufferWriter

[paulomorgado]

Package

OpenTelemetry.Exporter.OpenTelemetryProtocol

Is your feature request related to a problem?

Improve Protobuff Serialization Strategy: Replace ThreadStatic Array with Segmented IBufferWriter<byte>

Description

Currently, the protobuf serialization strategy relies on a ThreadStatic-anchored byte array that grows as needed, involving data copying on each resize. This approach has several limitations in terms of performance, memory efficiency, and scalability.

Proposal

Replace the current buffer implementation with a segmented buffer that:

• Implements IBufferWriter<byte>
• Produces a ReadOnlySequence<byte> for downstream consumption

Benefits of Segmented Buffer Approach

✅ Zero-Copy Composition

• ReadOnlySequence<byte> allows chaining multiple memory segments without copying data.
• Avoids the overhead of reallocating and copying data as the buffer grows.

✅ Better Memory Utilization

• Segments can be pooled and reused, reducing GC pressure.
• Avoids large contiguous allocations, which are more prone to fragmentation and LOH (Large Object Heap) allocations.

✅ Thread Safety and Isolation

• ThreadStatic buffers are shared across all usages on the same thread, which can lead to subtle bugs or data leaks.
• A segmented buffer can be scoped per operation, ensuring isolation and correctness.

✅ Scalability

• Segmented buffers scale better in high-throughput or concurrent environments.
• They can be backed by ArrayPool<byte> or custom memory pools for fine-grained control.

✅ Compatibility with Pipelines and Span-based APIs

• IBufferWriter<byte> is the standard interface used by System.IO.Pipelines and other modern .NET APIs.
• Enables seamless integration with serializers, encoders, and network stacks.

Drawbacks of Current ThreadStatic Array Approach

• ❌ Requires manual resizing and copying, which is error-prone and inefficient.
• ❌ Can lead to memory leaks or excessive memory retention if not carefully managed.
• ❌ Not composable or compatible with modern .NET streaming APIs.

Suggested Implementation

Introduce a PooledSegmentedBufferWriter class that:

• Implements IBufferWriter<byte>
• Internally manages a list of pooled segments
• Exposes a ReadOnlySequence<byte> for reading

What is the expected behavior?

The protobuf serialization process should use a PooledSegmentedBufferWriter to efficiently accumulate serialized data without reallocating or copying memory as the buffer grows. The writer should:

• Dynamically allocate and reuse pooled memory segments.
• Implement IBufferWriter<byte> to support efficient, span-based writing.
• Produce a ReadOnlySequence<byte> that represents the complete serialized payload without requiring additional copying or transformation.
• Avoid reliance on ThreadStatic buffers, ensuring thread safety and reducing memory retention risks.
• Integrate seamlessly with existing serialization pipelines and support high-throughput scenarios with minimal GC overhead.

Which alternative solutions or features have you considered?

Alternative 1: ThreadStatic Byte Array with Manual Resizing (Current Implementation)

This approach uses a ThreadStatic-anchored byte array that grows as needed. While it avoids allocations in some cases by reusing the same buffer per thread, it has several drawbacks:

• Requires manual resizing and copying, which is inefficient and error-prone.
• Can lead to memory retention issues if buffers grow large and are never reset.
• Not safe across async boundaries or parallel operations.
• Difficult to integrate with modern span-based or streaming APIs.

Alternative 2: ArrayPool with Manual Buffer Management

Using ArrayPool<byte> to rent and return buffers is a common way to reduce allocations:

• Avoids repeated allocations by reusing pooled arrays.
• Still requires manual resizing and copying when the buffer grows.
• Does not support segmented composition — the buffer must be contiguous.
• Managing lifetime and return of the buffer adds complexity and risk of leaks or corruption.

Alternative 3: RecyclableMemoryStream

RecyclableMemoryStream from the Microsoft.IO package is a high-performance, pooled stream implementation that:

• Implements IBufferWriter<byte> out of the box.
• Can produce a ReadOnlySequence<byte>.
• Reduces LOH allocations and GC pressure through segment pooling.

While this is a strong candidate, it introduces an external dependency and adds some complexity in terms of configuration and diagnostics. It also abstracts away some control over segment size and memory layout, which may be important in performance-critical scenarios.

Why PooledSegmentedBufferWriter is the Best Option

• Offers fine-grained control over segment size, pooling strategy, and memory layout.
• Fully compatible with IBufferWriter<byte> and ReadOnlySequence<byte>, enabling efficient

Additional context

No response

/cc @Kielek, @martincostello

[alanwest]

We discussed this a bit during our SIG meeting today.

@CodeBlanch shared some thoughts. Summarizing...

• The resizing of the byte array is likely just an initial start up cost. That is, once the application reaches a steady state the size of the buffer is likely to remain stable.
• An unusually large export theoretically could ballon the size of the buffer leading to increased memory consumption for the rest of the lifetime of the process, though this is probably not very much of a concern in practice.
• Adding more complexity to managing the size of the buffer may introduce greater performance issues than it is seeking to solve.

[paulomorgado]

These are snapshots of the same process of a video streamming service receiving 3 RTSP feeds and stream each one to 2 WebRTC clients.

---

Process Uptime: 0 days 0:04:00.000
Kernel time: 0 days 0:03:43.000
User time: 0 days 0:15:44.000

Name	Inc %	Inc	Inc Ct
System.Private.CoreLib!Byte[] (Bytes > 1M,NoPtrs,ElemSize=1)	85,2	104 858 608	42
+ System.Private.CoreLib!Buffers.SharedArrayPool+ThreadLocalArray[] (Ptrs,ElemSize=16)	55,4	68 158 136	29
+ System.Private.CoreLib!Byte[][] (Ptrs,ElemSize=8)	22,1	27 263 192	9
+ My.StreamingProtocol.Models.Bgra32FrameData	6,0	7 340 104	3
+ My.RTSP.SharpRtspStreamService+d__22	1,7	2 097 176	1
+ [local vars] {MinDepth 2}	0,0	0	0
+ System.Memory!Buffers.ArrayMemoryPool+ArrayMemoryPoolBuffer {MinDepth 12}	0,0	0	0
+ System.Threading.Channels!Channels.ChannelReader, IDisposable, Int32, Int32>>+d__12 {MinDepth 13}	0,0	0	0
System.Private.CoreLib!Byte[] (Bytes > 100K,NoPtrs,ElemSize=1)	3,6	4 477 462	14
+ System.Private.CoreLib!Buffers.SharedArrayPool+ThreadLocalArray[] (Ptrs,ElemSize=16)	1,1	1 310 888	7
+ [local vars]	0,6	786 504	3
+ OpenTelemetry.Exporter.OpenTelemetryProtocol!OpenTelemetry.Exporter.OtlpTraceExporter	0,6	750 024	1
+ OpenTelemetry.Exporter.OpenTelemetryProtocol!OpenTelemetry.Exporter.OtlpMetricExporter	0,6	750 024	1
+ OpenTelemetry.Exporter.OpenTelemetryProtocol!OpenTelemetry.Exporter.OpenTelemetryProtocol.Implementation.Serializer.ProtobufOtlpLogSerializer+SerializationState	0,6	750 024	1
+ System.Private.CoreLib!IO.MemoryStream	0,1	129 998	1
+ OpenTelemetry.Exporter.OpenTelemetryProtocol!OpenTelemetry.Exporter.OtlpLogExporter {MinDepth 9}	0,0	0	0

---

Process Uptime: 0 days 3:30:51.000
Kernel time: 0 days 3:09:44.000
User time: 0 days 16:14:15.000

Name	Inc %	Inc	Inc Ct
System.Private.CoreLib!Byte[] (Bytes > 1M,NoPtrs,ElemSize=1)	66,1	37 151 920	14
+ System.Private.CoreLib!Buffers.SharedArrayPool+ThreadLocalArray[] (Ptrs,ElemSize=16)	26,1	14 680 184	5
+ My.StreamingProtocol.Models.Bgra32FrameData	20,5	11 534 432	4
+ [Pinned local vars]	7,5	4 194 328	1
+ System.Private.CoreLib!Byte[][] (Ptrs,ElemSize=8)	3,7	2 097 176	1
+ System.Threading.Channels!Channels.ChannelReader, IDisposable, Int32, Int32>>+d__12	3,7	2 097 176	1
+ OpenTelemetry.Exporter.OpenTelemetryProtocol!OpenTelemetry.Exporter.OpenTelemetryProtocol.Implementation.Serializer.ProtobufOtlpLogSerializer+SerializationState	2,7	1 500 024	1
+ My.RTSP.SharpRtspStreamService+d__22	1,9	1 048 600	1
+ [local vars] {MinDepth 2}	0,0	0	0
+ System.Memory!Buffers.ArrayMemoryPool+ArrayMemoryPoolBuffer {MinDepth 4}	0,0	0	0
+ OpenTelemetry.Exporter.OpenTelemetryProtocol!OpenTelemetry.Exporter.OtlpLogExporter {MinDepth 9}	0,0	0	0
System.Private.CoreLib!Byte[] (Bytes > 100K,NoPtrs,ElemSize=1)	5,7	3 203 054	9
+ [local vars]	2,8	1 573 008	6
+ OpenTelemetry.Exporter.OpenTelemetryProtocol!OpenTelemetry.Exporter.OtlpTraceExporter	1,3	750 024	1
+ OpenTelemetry.Exporter.OpenTelemetryProtocol!OpenTelemetry.Exporter.OtlpMetricExporter	1,3	750 024	1
+ System.Private.CoreLib!IO.MemoryStream	0,2	129 998	1

As you can see, it's not just initial cost.

And to get to 750,024 bytes or 1,573,008 there's a lot of LOH allocations.

I think it's overkill to use Pipes.

But something like this would be good enough and would allow to implement the total memory limits:

    internal sealed class PooledSegmentedBuffer : IBufferWriter<byte>, IDisposable
    {
        private const int DefaultBufferSize = 4096;

        private readonly List<byte[]> _buffers = new();
        private readonly ArrayPool<byte> _pool;
        private readonly int _minimalIncrease;
        private int _currentIndex;
        private int _currentOffset;
        private bool _disposed;

        public PooledSegmentedBuffer(int minimalIncrease = DefaultBufferSize, ArrayPool<byte>? pool = null)
        {
            _minimalIncrease = minimalIncrease;
            _pool = pool ?? ArrayPool<byte>.Shared;
        }

        public int Length { get; private set; }

        public void Advance(int count)
        {
            if (count < 0)
            {
                throw new ArgumentOutOfRangeException(nameof(count));
            }

            if (_buffers.Count == 0)
            {
                if (count > 0)
                {
                    throw new ArgumentOutOfRangeException(nameof(count));
                }
                return;
            }

            if (_currentOffset + count > _buffers[_currentIndex].Length)
            {
                throw new ArgumentOutOfRangeException(nameof(count));
            }

            _currentOffset += count;
            Length += count;
        }

        public Memory<byte> GetMemory(int sizeHint = 0)
        {
            EnsureCapacity(sizeHint);
            return _buffers[_currentIndex].AsMemory(_currentOffset);
        }

        public Span<byte> GetSpan(int sizeHint = 0)
        {
            EnsureCapacity(sizeHint);
            return _buffers[_currentIndex].AsSpan(_currentOffset);
        }

        public void Write(ReadOnlySpan<byte> source)
        {
            if (source.IsEmpty)
                return;

            var destination = GetSpan(source.Length);
            source.CopyTo(destination);
            Advance(source.Length);
        }

        public void Write(T[] source)
        {
            if (source == null || source.Length == 0)
                return;

            Write(source.AsSpan());
        }

        public ReadOnlySequence<byte> GetReadOnlySequence()
        {
            SequenceSegment? first = null;
            SequenceSegment? last = null;

            for (var i = 0; i < _buffers.Count; i++)
            {
                var buffer = _buffers[i];
                var length = (i == _currentIndex) ? _currentOffset : buffer.Length;

                if (length == 0)
                {
                    continue;
                }

                var segment = new SequenceSegment(buffer.AsMemory(0, length));

                if (first == null)
                {
                    first = segment;
                }

                if (last != null)
                {
                    last.SetNext(segment);
                }

                last = segment;
            }

            if (first == null || last == null)
            {
                return ReadOnlySequence<byte>.Empty;
            }

            return new ReadOnlySequence<byte>(first, 0, last, last.Memory.Length);
        }

        public void Clear()
        {
            foreach (var buffer in _buffers)
            {
                _pool.Return(buffer);
            }

            _buffers.Clear();
            Length = 0;
            _currentIndex = 0;
            _currentOffset = 0;
        }

        public void Dispose()
        {
            if (_disposed)
            {
                return;
            }

            Clear();
            _disposed = true;
        }

        private void EnsureCapacity(int sizeHint)
        {
            if (_buffers.Count == 0 || _currentOffset + sizeHint > _buffers[_currentIndex].Length)
            {
                var newSize = Math.Max(sizeHint, _minimalIncrease);
                AddNewBuffer(newSize);
            }
        }

        private void AddNewBuffer(int size)
        {
            var buffer = _pool.Rent(size);
            _buffers.Add(buffer);
            _currentIndex = _buffers.Count - 1;
            _currentOffset = 0;
        }

        private class SequenceSegment : ReadOnlySequenceSegment<byte>
        {
            public SequenceSegment(ReadOnlyMemory<byte> memory)
            {
                Memory = memory;
            }

            public void SetNext(SequenceSegment next)
            {
                Next = next;
                next.RunningIndex = RunningIndex + Memory.Length;
            }
        }
    }

The memory is rented only while in use and returned.

I don't see how using this would add any complexity to the current code. Quite the contrary.

There's a choice to be made here though, what is the chunk size to be used? The most common in the BCL (4096), which would allow these pooled arrays to be shared across several components? 32k or 64k?