Logs: Add log record pooling

src/OpenTelemetry.Exporter.InMemory/InMemoryExporterLoggingExtensions.cs

@@ -42,9 +42,7 @@ private static ExportResult ExportLogRecord(in Batch<LogRecord> batch, ICollecti
 
             foreach (var log in batch)
             {
-                log.BufferLogScopes();
-
-                exportedItems.Add(log);
+                exportedItems.Add(log.Copy());
             }
 
             return ExportResult.Success;

src/OpenTelemetry/Batch.cs

@@ -22,6 +22,7 @@
 using System.Diagnostics;
 using System.Diagnostics.CodeAnalysis;
 using OpenTelemetry.Internal;
+using OpenTelemetry.Logs;
 
 namespace OpenTelemetry
 {
@@ -90,7 +91,11 @@ public void Dispose()
                 // Drain anything left in the batch.
                 while (this.circularBuffer.RemovedCount < this.targetCount)
                 {
-                    this.circularBuffer.Read();
+                    T item = this.circularBuffer.Read();
+                    if (typeof(T) == typeof(LogRecord))
+                    {
+                        LogRecordSharedPool.Current.Return((LogRecord)(object)item);
+                    }
                 }
             }
         }
@@ -140,6 +145,32 @@ public struct Enumerator : IEnumerator<T>
                 return false;
             };
 
+            private static readonly BatchEnumeratorMoveNextFunc MoveNextCircularBufferLogRecord = (ref Enumerator enumerator) =>
+            {
+                // Note: This type check here is to give the JIT a hint it can
+                // remove all of this code when T != LogRecord
+                if (typeof(T) == typeof(LogRecord))
+                {
+                    var circularBuffer = enumerator.circularBuffer;
+
+                    var currentItem = enumerator.Current;
+                    if (currentItem != null)
+                    {
+                        LogRecordSharedPool.Current.Return((LogRecord)(object)currentItem);
+                    }
+
+                    if (circularBuffer!.RemovedCount < enumerator.targetCount)
+                    {
+                        enumerator.current = circularBuffer.Read();
+                        return true;
+                    }
+
+                    enumerator.current = null;
+                }
+
+                return false;
+            };
+
             private static readonly BatchEnumeratorMoveNextFunc MoveNextArray = (ref Enumerator enumerator) =>
             {
                 var items = enumerator.items;
@@ -179,7 +210,7 @@ internal Enumerator(CircularBuffer<T> circularBuffer, long targetCount)
                 this.circularBuffer = circularBuffer;
                 this.targetCount = targetCount;
                 this.itemIndex = 0;
-                this.moveNextFunc = MoveNextCircularBuffer;
+                this.moveNextFunc = typeof(T) == typeof(LogRecord) ? MoveNextCircularBufferLogRecord : MoveNextCircularBuffer;
             }
 
             internal Enumerator(T[] items, long targetCount)
@@ -201,6 +232,15 @@ internal Enumerator(T[] items, long targetCount)
             /// <inheritdoc/>
             public void Dispose()
             {
+                if (typeof(T) == typeof(LogRecord))
+                {
+                    var currentItem = this.current;
+                    if (currentItem != null)
+                    {
+                        LogRecordSharedPool.Current.Return((LogRecord)(object)currentItem);
+                        this.current = null;
+                    }
+                }
             }
 
             /// <inheritdoc/>

src/OpenTelemetry/BatchExportProcessor.cs

@@ -18,6 +18,7 @@
 
 using System;
 using System.Diagnostics;
+using System.Runtime.CompilerServices;
 using System.Threading;
 using OpenTelemetry.Internal;
 
@@ -95,8 +96,8 @@ public abstract class BatchExportProcessor<T> : BaseExportProcessor<T>
         /// </summary>
         internal long ProcessedCount => this.circularBuffer.RemovedCount;
 
-        /// <inheritdoc/>
-        protected override void OnExport(T data)
+        [MethodImpl(MethodImplOptions.AggressiveInlining)]
+        internal bool TryExport(T data)
         {
             if (this.circularBuffer.TryAdd(data, maxSpinCount: 50000))
             {
@@ -111,11 +112,19 @@ protected override void OnExport(T data)
                     }
                 }
 
-                return; // enqueue succeeded
+                return true; // enqueue succeeded
             }
 
             // either the queue is full or exceeded the spin limit, drop the item on the floor
             Interlocked.Increment(ref this.droppedCount);
+
+            return false;
+        }
+
+        /// <inheritdoc/>
+        protected override void OnExport(T data)
+        {
+            this.TryExport(data);
         }
 
         /// <inheritdoc/>

src/OpenTelemetry/CHANGELOG.md

@@ -25,6 +25,9 @@
 * Handle possible exception when initializing the default service name.
   ([#3405](https://github.com/open-telemetry/opentelemetry-dotnet/pull/3405))
 
+* `LogRecord` instances are now reused to reduce memory pressure
+  ([#3385](https://github.com/open-telemetry/opentelemetry-dotnet/pull/3385))
+
 ## 1.3.0
 
 Released 2022-Jun-03

src/OpenTelemetry/CompositeProcessor.cs

@@ -26,7 +26,7 @@ namespace OpenTelemetry
 {
     public class CompositeProcessor<T> : BaseProcessor<T>
     {
-        private readonly DoublyLinkedListNode head;
+        internal readonly DoublyLinkedListNode Head;
         private DoublyLinkedListNode tail;
         private bool disposed;
 
@@ -40,8 +40,8 @@ public CompositeProcessor(IEnumerable<BaseProcessor<T>> processors)
                 throw new ArgumentException($"'{iter}' is null or empty", nameof(iter));
             }
 
-            this.head = new DoublyLinkedListNode(iter.Current);
-            this.tail = this.head;
+            this.Head = new DoublyLinkedListNode(iter.Current);
+            this.tail = this.Head;
 
             while (iter.MoveNext())
             {
@@ -66,7 +66,7 @@ public CompositeProcessor<T> AddProcessor(BaseProcessor<T> processor)
         /// <inheritdoc/>
         public override void OnEnd(T data)
         {
-            for (var cur = this.head; cur != null; cur = cur.Next)
+            for (var cur = this.Head; cur != null; cur = cur.Next)
             {
                 cur.Value.OnEnd(data);
             }
@@ -75,7 +75,7 @@ public override void OnEnd(T data)
         /// <inheritdoc/>
         public override void OnStart(T data)
         {
-            for (var cur = this.head; cur != null; cur = cur.Next)
+            for (var cur = this.Head; cur != null; cur = cur.Next)
             {
                 cur.Value.OnStart(data);
             }
@@ -85,7 +85,7 @@ internal override void SetParentProvider(BaseProvider parentProvider)
         {
             base.SetParentProvider(parentProvider);
 
-            for (var cur = this.head; cur != null; cur = cur.Next)
+            for (var cur = this.Head; cur != null; cur = cur.Next)
             {
                 cur.Value.SetParentProvider(parentProvider);
             }
@@ -99,7 +99,7 @@ protected override bool OnForceFlush(int timeoutMilliseconds)
                 ? null
                 : Stopwatch.StartNew();
 
-            for (var cur = this.head; cur != null; cur = cur.Next)
+            for (var cur = this.Head; cur != null; cur = cur.Next)
             {
                 if (sw == null)
                 {
@@ -125,7 +125,7 @@ protected override bool OnShutdown(int timeoutMilliseconds)
                 ? null
                 : Stopwatch.StartNew();
 
-            for (var cur = this.head; cur != null; cur = cur.Next)
+            for (var cur = this.Head; cur != null; cur = cur.Next)
             {
                 if (sw == null)
                 {
@@ -150,7 +150,7 @@ protected override void Dispose(bool disposing)
             {
                 if (disposing)
                 {
-                    for (var cur = this.head; cur != null; cur = cur.Next)
+                    for (var cur = this.Head; cur != null; cur = cur.Next)
                     {
                         try
                         {
@@ -169,7 +169,7 @@ protected override void Dispose(bool disposing)
             base.Dispose(disposing);
         }
 
-        private class DoublyLinkedListNode
+        internal sealed class DoublyLinkedListNode
         {
             public readonly BaseProcessor<T> Value;
 

src/OpenTelemetry/Logs/BatchLogRecordExportProcessor.cs

@@ -57,9 +57,14 @@ public override void OnEnd(LogRecord data)
             // happen here.
             Debug.Assert(data != null, "LogRecord was null.");
 
-            data!.BufferLogScopes();
+            data!.Buffer();
 
-            base.OnEnd(data);
+            data.AddReference();
+
+            if (!this.TryExport(data))
+            {
+                LogRecordSharedPool.Current.Return(data);
+            }
         }
     }
 }

src/OpenTelemetry/Logs/LogRecord.cs

@@ -19,6 +19,8 @@
 using System;
 using System.Collections.Generic;
 using System.Diagnostics;
+using System.Runtime.CompilerServices;
+using System.Threading;
 using Microsoft.Extensions.Logging;
 using OpenTelemetry.Internal;
 
@@ -30,15 +32,21 @@ namespace OpenTelemetry.Logs
     public sealed class LogRecord
     {
         internal LogRecordData Data;
+        internal List<KeyValuePair<string, object?>>? AttributeStorage;
+        internal List<object?>? BufferedScopes;
+        internal int PoolReferenceCount = int.MaxValue;
 
         private static readonly Action<object?, List<object?>> AddScopeToBufferedList = (object? scope, List<object?> state) =>
         {
             state.Add(scope);
         };
 
-        private List<object?>? bufferedScopes;
+        internal LogRecord()
+        {
+        }
 
         // Note: Some users are calling this with reflection. Try not to change the signature to be nice.
+        [Obsolete("Call LogRecordPool.Rent instead.")]
         internal LogRecord(
             IExternalScopeProvider? scopeProvider,
             DateTime timestamp,
@@ -191,9 +199,9 @@ public void ForEachScope<TState>(Action<LogRecordScope, TState> callback, TState
 
             var forEachScopeState = new ScopeForEachState<TState>(callback, state);
 
-            if (this.bufferedScopes != null)
+            if (this.BufferedScopes != null)
             {
-                foreach (object? scope in this.bufferedScopes)
+                foreach (object? scope in this.BufferedScopes)
                 {
                     ScopeForEachState<TState>.ForEachScope(scope, forEachScopeState);
                 }
@@ -213,22 +221,99 @@ public void ForEachScope<TState>(Action<LogRecordScope, TState> callback, TState
             return ref this.Data;
         }
 
+        [MethodImpl(MethodImplOptions.AggressiveInlining)]
+        internal void ResetReferenceCount()
+        {
+            this.PoolReferenceCount = 1;
+        }
+
+        [MethodImpl(MethodImplOptions.AggressiveInlining)]
+        internal void AddReference()
+        {
+            Interlocked.Increment(ref this.PoolReferenceCount);
+        }
+
+        [MethodImpl(MethodImplOptions.AggressiveInlining)]
+        internal int RemoveReference()
+        {
+            return Interlocked.Decrement(ref this.PoolReferenceCount);
+        }
+
+        // Note: Typically called when LogRecords are added into a batch so they
+        // can be safely processed outside of the log call chain.
+        internal void Buffer()
+        {
+            // Note: State values are buffered because some states are not safe
+            // to access outside of the log call chain. See:
+            // https://github.com/open-telemetry/opentelemetry-dotnet/issues/2905
+            this.BufferLogStateValues();
+
+            this.BufferLogScopes();
+
+            // Note: There is no buffering of "State" only "StateValues". We
+            // don't inspect "object State" at all. It is undefined what
+            // exporters will do with "State". Some might ignore it, some might
+            // attempt to access it as a list. That is potentially dangerous.
+            // TODO: Investigate what to do here. Should we obsolete State and
+            // just use the StateValues design?
+        }
+
+        internal LogRecord Copy()
+        {
+            // Note: We only buffer scopes here because state values are copied
+            // directly below.
+            this.BufferLogScopes();
+
+            return new()
+            {
+                Data = this.Data,
+                State = this.State,
+                StateValues = this.StateValues == null ? null : new List<KeyValuePair<string, object?>>(this.StateValues),
+                BufferedScopes = this.BufferedScopes == null ? null : new List<object?>(this.BufferedScopes),
+            };
+        }
+
+        /// <summary>
+        /// Buffers the state values attached to the log into a list so that
+        /// they can be safely processed after the log message lifecycle has
+        /// ended.
+        /// </summary>
+        private void BufferLogStateValues()
+        {
+            var stateValues = this.StateValues;
+            if (stateValues == null || stateValues == this.AttributeStorage)
+            {
+                return;
+            }
+
+            var attributeStorage = this.AttributeStorage ??= new List<KeyValuePair<string, object?>>(stateValues.Count);
+
+            // Note: AddRange here will copy all of the KeyValuePairs from
+            // stateValues to AttributeStorage. This "captures" the state and
+            // fixes issues where the values are generated at enumeration time
+            // like
+            // https://github.com/open-telemetry/opentelemetry-dotnet/issues/2905.
+            attributeStorage.AddRange(stateValues);
+
+            this.StateValues = attributeStorage;
+        }
+
         /// <summary>
         /// Buffers the scopes attached to the log into a list so that they can
         /// be safely processed after the log message lifecycle has ended.
         /// </summary>
-        internal void BufferLogScopes()
+        private void BufferLogScopes()
         {
-            if (this.ScopeProvider == null || this.bufferedScopes != null)
+            if (this.ScopeProvider == null)
             {
                 return;
             }
 
-            List<object?> scopes = new List<object?>();
+            List<object?> scopes = this.BufferedScopes ??= new List<object?>(LogRecordPoolHelper.DefaultMaxNumberOfScopes);
 
-            this.ScopeProvider?.ForEachScope(AddScopeToBufferedList, scopes);
+            this.ScopeProvider.ForEachScope(AddScopeToBufferedList, scopes);
 
-            this.bufferedScopes = scopes;
+            this.ScopeProvider = null;
         }
 
         private readonly struct ScopeForEachState<TState>