Flush in batches and pool lists for emitted records

Author: evgenyfedorov2

src/Libraries/Microsoft.AspNetCore.Diagnostics.Middleware/Buffering/IncomingRequestLogBuffer.cs

@@ -9,13 +9,19 @@
 using System.Threading;
 using Microsoft.Extensions.Diagnostics.Buffering;
 using Microsoft.Extensions.Logging.Abstractions;
+using Microsoft.Extensions.ObjectPool;
 using Microsoft.Extensions.Options;
 using Microsoft.Shared.Diagnostics;
+using Microsoft.Shared.Pools;
 
 namespace Microsoft.AspNetCore.Diagnostics.Buffering;
 
 internal sealed class IncomingRequestLogBuffer
 {
+    private const int MaxBatchSize = 256;
+    private static readonly ObjectPool<List<DeserializedLogRecord>> _recordsToEmitListPool =
+        PoolFactory.CreateListPoolWithCapacity<DeserializedLogRecord>(MaxBatchSize);
+
     private readonly IBufferedLogger _bufferedLogger;
     private readonly LogBufferingFilterRuleSelector _ruleSelector;
     private readonly IOptionsMonitor<PerRequestLogBufferingOptions> _options;
@@ -107,21 +113,38 @@ public void Flush()
         }
         SerializedLogRecord[] bufferedRecords = bufferToFlush.ToArray();
 
-
-        var recordsToEmit = new List<DeserializedLogRecord>(bufferedRecords.Length);
-        foreach (SerializedLogRecord bufferedRecord in bufferedRecords)
+        // Process records in batches
+        for (int offset = 0; offset < bufferedRecords.Length; offset += MaxBatchSize)
         {
-            recordsToEmit.Add(new DeserializedLogRecord(
-                bufferedRecord.Timestamp,
-                bufferedRecord.LogLevel,
-                bufferedRecord.EventId,
-                bufferedRecord.Exception,
-                bufferedRecord.FormattedMessage,
-                bufferedRecord.Attributes));
+            int currentBatchSize = Math.Min(MaxBatchSize, bufferedRecords.Length - offset);
+            List<DeserializedLogRecord>? recordsToEmit = null;
+            try
+            {
+                recordsToEmit = _recordsToEmitListPool.Get();
+
+                for (int i = 0; i < currentBatchSize; i++)
+                {
+                    SerializedLogRecord bufferedRecord = bufferedRecords[offset + i];
+                    recordsToEmit.Add(new DeserializedLogRecord(
+                        bufferedRecord.Timestamp,
+                        bufferedRecord.LogLevel,
+                        bufferedRecord.EventId,
+                        bufferedRecord.Exception,
+                        bufferedRecord.FormattedMessage,
+                        bufferedRecord.Attributes));
+                }
+
+                _bufferedLogger.LogRecords(recordsToEmit);
+            }
+            finally
+            {
+                if(recordsToEmit is not null)
+                {
+                    _recordsToEmitListPool.Return(recordsToEmit);
+                }
+            }
         }
 
-        _bufferedLogger.LogRecords(recordsToEmit);
-
         SerializedLogRecordFactory.Return(bufferedRecords);
     }
 

src/Libraries/Microsoft.Extensions.Telemetry/Buffering/GlobalBuffer.cs

@@ -8,13 +8,19 @@
 using System.Linq;
 using System.Threading;
 using Microsoft.Extensions.Logging.Abstractions;
+using Microsoft.Extensions.ObjectPool;
 using Microsoft.Extensions.Options;
 using Microsoft.Shared.Diagnostics;
+using Microsoft.Shared.Pools;
 
 namespace Microsoft.Extensions.Diagnostics.Buffering;
 
 internal sealed class GlobalBuffer : IDisposable
 {
+    private const int MaxBatchSize = 256;
+    private static readonly ObjectPool<List<DeserializedLogRecord>> _recordsToEmitListPool =
+        PoolFactory.CreateListPoolWithCapacity<DeserializedLogRecord>(MaxBatchSize);
+
     private readonly IOptionsMonitor<GlobalLogBufferingOptions> _options;
     private readonly IBufferedLogger _bufferedLogger;
     private readonly TimeProvider _timeProvider;
@@ -127,20 +133,38 @@ public void Flush()
 
         SerializedLogRecord[] bufferedRecords = bufferToFlush.ToArray();
 
-        var recordsToEmit = new List<DeserializedLogRecord>(bufferedRecords.Length);
-        foreach (SerializedLogRecord bufferedRecord in bufferedRecords)
+        // Process records in batches
+        for (int offset = 0; offset < bufferedRecords.Length; offset += MaxBatchSize)
         {
-            recordsToEmit.Add(new DeserializedLogRecord(
-                bufferedRecord.Timestamp,
-                bufferedRecord.LogLevel,
-                bufferedRecord.EventId,
-                bufferedRecord.Exception,
-                bufferedRecord.FormattedMessage,
-                bufferedRecord.Attributes));
+            int currentBatchSize = Math.Min(MaxBatchSize, bufferedRecords.Length - offset);
+            List<DeserializedLogRecord>? recordsToEmit = null;
+            try
+            {
+                recordsToEmit = _recordsToEmitListPool.Get();
+
+                for (int i = 0; i < currentBatchSize; i++)
+                {
+                    SerializedLogRecord bufferedRecord = bufferedRecords[offset + i];
+                    recordsToEmit.Add(new DeserializedLogRecord(
+                        bufferedRecord.Timestamp,
+                        bufferedRecord.LogLevel,
+                        bufferedRecord.EventId,
+                        bufferedRecord.Exception,
+                        bufferedRecord.FormattedMessage,
+                        bufferedRecord.Attributes));
+                }
+
+                _bufferedLogger.LogRecords(recordsToEmit);
+            }
+            finally
+            {
+                if(recordsToEmit is not null)
+                {
+                    _recordsToEmitListPool.Return(recordsToEmit);
+                }
+            }
         }
 
-        _bufferedLogger.LogRecords(recordsToEmit);
-
         SerializedLogRecordFactory.Return(bufferedRecords);
     }
 

src/Shared/Pools/PoolFactory.cs

@@ -118,6 +118,25 @@ public static ObjectPool<List<T>> CreateListPool<T>(int maxCapacity = DefaultCap
         return MakePool(PooledListPolicy<T>.Instance, maxCapacity);
     }
 
+    /// <summary>
+    /// Creates an object pool of <see cref="List{T}"/> instances, each with provided <see cref="listCapacity"/>.
+    /// </summary>
+    /// <typeparam name="T">The type of object held by the lists.</typeparam>
+    /// <param name="listCapacity">The capacity of each created <see cref="List{T}"/> instance.</param>
+    /// <param name="maxCapacity">
+    /// The maximum number of items to keep in the pool.
+    /// This defaults to 1024.
+    /// This value is a recommendation, the pool may keep more objects than this.
+    /// </param>
+    /// <returns>The pool.</returns>
+    public static ObjectPool<List<T>> CreateListPoolWithCapacity<T>(int listCapacity, int maxCapacity = DefaultCapacity)
+    {
+        _ = Throw.IfLessThan(maxCapacity, 1);
+        _ = Throw.IfLessThan(listCapacity, 0);
+
+        return MakePool(PooledListWithCapacityPolicy<T>.Instance(listCapacity), maxCapacity);
+    }
+
     /// <summary>
     /// Creates an object pool of <see cref="Dictionary{TKey, TValue}"/> instances.
     /// </summary>

src/Shared/Pools/PooledListWithCapacityPolicy.cs

@@ -0,0 +1,30 @@
+// Licensed to the .NET Foundation under one or more agreements.
+// The .NET Foundation licenses this file to you under the MIT license.
+
+using System.Collections.Generic;
+using Microsoft.Extensions.ObjectPool;
+
+namespace Microsoft.Shared.Pools;
+
+/// <summary>
+/// An object pool policy for lists with capacity.
+/// </summary>
+[System.Diagnostics.CodeAnalysis.ExcludeFromCodeCoverage]
+internal sealed class PooledListWithCapacityPolicy<T> : PooledObjectPolicy<List<T>>
+{
+    private readonly int _listCapacity;
+    public static PooledListWithCapacityPolicy<T> Instance (int listCapacity) => new(listCapacity);
+
+    private PooledListWithCapacityPolicy(int listCapacity)
+    {
+        _listCapacity = listCapacity;
+    }
+
+    public override List<T> Create() => new(_listCapacity);
+
+    public override bool Return(List<T> obj)
+    {
+        obj.Clear();
+        return true;
+    }
+}