ref: switch Buffer-Processor to be lock-free but discarding

src/Sentry/Internal/BatchBuffer.cs

@@ -4,77 +4,88 @@ namespace Sentry.Internal;
 /// A slim wrapper over an <see cref="System.Array"/>,
 /// intended for buffering.
 /// </summary>
+/// <remarks>
+/// <para><see cref="Capacity"/> is thread-safe.</para>
+/// <para><see cref="TryAdd(T, out int)"/> is thread-safe.</para>
+/// <para><see cref="ToArrayAndClear()"/> is not thread-safe.</para>
+/// <para><see cref="ToArrayAndClear(int)"/> is not thread-safe.</para>
+/// </remarks>
 internal sealed class BatchBuffer<T>
 {
     private readonly T[] _array;
-    private int _count;
+    private int _additions;
 
     public BatchBuffer(int capacity)
     {
-        ThrowIfNegativeOrZero(capacity, nameof(capacity));
+        ThrowIfLessThanTwo(capacity, nameof(capacity));
 
         _array = new T[capacity];
-        _count = 0;
+        _additions = 0;
     }
 
-    internal int Count => _count;
+    //internal int Count => _count;
     internal int Capacity => _array.Length;
-    internal bool IsEmpty => _count == 0 && _array.Length != 0;
-    internal bool IsFull => _count == _array.Length;
+    internal bool IsEmpty => _additions == 0;
+    internal bool IsFull => _additions >= _array.Length;
 
-    internal bool TryAdd(T item)
+    internal bool TryAdd(T item, out int count)
     {
-        if (_count < _array.Length)
+        count = Interlocked.Increment(ref _additions);
+
+        if (count <= _array.Length)
         {
-            _array[_count] = item;
-            _count++;
+            _array[count - 1] = item;
             return true;
         }
 
         return false;
     }
 
-    internal T[] ToArray()
+    internal T[] ToArrayAndClear()
+    {
+        return ToArrayAndClear(_additions);
+    }
+
+    internal T[] ToArrayAndClear(int length)
+    {
+        var array = ToArray(length);
+        Clear(length);
+        return array;
+    }
+
+    private T[] ToArray(int length)
     {
-        if (_count == 0)
+        if (length == 0)
         {
             return Array.Empty<T>();
         }
 
-        var array = new T[_count];
-        Array.Copy(_array, array, _count);
+        var array = new T[length];
+        Array.Copy(_array, array, length);
         return array;
     }
 
-    internal void Clear()
+    private void Clear(int length)
     {
-        if (_count == 0)
+        if (length == 0)
         {
             return;
         }
 
-        var count = _count;
-        _count = 0;
-        Array.Clear(_array, 0, count);
-    }
-
-    internal T[] ToArrayAndClear()
-    {
-        var array = ToArray();
-        Clear();
-        return array;
+        _additions = 0;
+        Array.Clear(_array, 0, length);
     }
 
-    private static void ThrowIfNegativeOrZero(int capacity, string paramName)
+    private static void ThrowIfLessThanTwo(int capacity, string paramName)
     {
-        if (capacity <= 0)
+        if (capacity < 2)
         {
-            ThrowNegativeOrZero(capacity, paramName);
+            ThrowLessThanTwo(capacity, paramName);
         }
     }
 
-    private static void ThrowNegativeOrZero(int capacity, string paramName)
+    private static void ThrowLessThanTwo(int capacity, string paramName)
     {
-        throw new ArgumentOutOfRangeException(paramName, capacity, "Argument must neither be negative nor zero.");
+        throw new ArgumentOutOfRangeException(paramName, capacity, "Argument must be at least two.");
     }
 }

src/Sentry/Internal/BatchProcessor.cs

@@ -18,9 +18,12 @@ internal sealed class BatchProcessor : IDisposable
     private readonly IHub _hub;
     private readonly TimeSpan _batchInterval;
     private readonly IDiagnosticLogger? _diagnosticLogger;
+
     private readonly Timer _timer;
-    private readonly BatchBuffer<SentryLog> _logs;
-    private readonly object _lock;
+    private readonly object _timerCallbackLock;
+    private readonly BatchBuffer<SentryLog> _buffer1;
+    private readonly BatchBuffer<SentryLog> _buffer2;
+    private volatile BatchBuffer<SentryLog> _activeBuffer;
 
     private DateTime _lastFlush = DateTime.MinValue;
 
@@ -31,51 +34,75 @@ public BatchProcessor(IHub hub, int batchCount, TimeSpan batchInterval, IDiagnos
         _diagnosticLogger = diagnosticLogger;
 
         _timer = new Timer(OnIntervalElapsed, this, Timeout.Infinite, Timeout.Infinite);
+        _timerCallbackLock = new object();
 
-        _logs = new BatchBuffer<SentryLog>(batchCount);
-        _lock = new object();
+        _buffer1 = new BatchBuffer<SentryLog>(batchCount);
+        _buffer2 = new BatchBuffer<SentryLog>(batchCount);
+        _activeBuffer = _buffer1;
     }
 
     internal void Enqueue(SentryLog log)
     {
-        lock (_lock)
+        var activeBuffer = _activeBuffer;
+
+        if (!TryEnqueue(activeBuffer, log))
         {
-            EnqueueCore(log);
+            activeBuffer = activeBuffer == _buffer1 ? _buffer2 : _buffer1;
+            if (!TryEnqueue(activeBuffer, log))
+            {
+                _diagnosticLogger?.LogInfo("Log Buffer full ... dropping log");
+            }
         }
     }
 
-    private void EnqueueCore(SentryLog log)
+    private bool TryEnqueue(BatchBuffer<SentryLog> buffer, SentryLog log)
     {
-        var isFirstLog = _logs.IsEmpty;
-        var added = _logs.TryAdd(log);
-        Debug.Assert(added, $"Since we currently have no lock-free scenario, it's unexpected to exceed the {nameof(BatchBuffer<SentryLog>)}'s capacity.");
-
-        if (isFirstLog && !_logs.IsFull)
-        {
-            EnableTimer();
-        }
-        else if (_logs.IsFull)
+        if (buffer.TryAdd(log, out var count))
         {
-            DisableTimer();
-            Flush();
+            if (count == 1) // is first of buffer
+            {
+                EnableTimer();
+            }
+
+            if (count == buffer.Capacity) // is buffer full
+            {
+                // swap active buffer atomically
+                var currentActiveBuffer = _activeBuffer;
+                var newActiveBuffer = ReferenceEquals(currentActiveBuffer, _buffer1) ? _buffer2 : _buffer1;
+                if (Interlocked.CompareExchange(ref _activeBuffer, newActiveBuffer, currentActiveBuffer) == currentActiveBuffer)
+                {
+                    DisableTimer();
+                    Flush(buffer, count);
+                }
+            }
+
+            return true;
         }
+
+        return false;
     }
 
-    private void Flush()
+    private void Flush(BatchBuffer<SentryLog> buffer, int count)
     {
         _lastFlush = DateTime.UtcNow;
 
-        var logs = _logs.ToArrayAndClear();
+        var logs = buffer.ToArrayAndClear(count);
         _ = _hub.CaptureEnvelope(Envelope.FromLog(new StructuredLog(logs)));
     }
 
     internal void OnIntervalElapsed(object? state)
     {
-        lock (_lock)
+        lock (_timerCallbackLock)
         {
-            if (!_logs.IsEmpty && DateTime.UtcNow > _lastFlush)
+            var currentActiveBuffer = _activeBuffer;
+
+            if (!currentActiveBuffer.IsEmpty && DateTime.UtcNow > _lastFlush)
             {
-                Flush();
+                var newActiveBuffer = ReferenceEquals(currentActiveBuffer, _buffer1) ? _buffer2 : _buffer1;
+                if (Interlocked.CompareExchange(ref _activeBuffer, newActiveBuffer, currentActiveBuffer) == currentActiveBuffer)
+                {
+                    Flush(currentActiveBuffer, -1);
+                }
             }
         }
     }

test/Sentry.Tests/Internals/BatchProcessorTests.cs

@@ -102,6 +102,32 @@ public void Enqueue_BothTimeoutAndSizeReached_CaptureEnvelopes()
         AssertEnvelopes(["one"], ["two", "three"]);
     }
 
+    [Fact]
+    public async Task Enqueue_Concurrency_CaptureEnvelopes()
+    {
+        using var processor = new BatchProcessor(_hub, 100, Timeout.InfiniteTimeSpan, _diagnosticLogger);
+        using var sync = new ManualResetEvent(false);
+
+        var tasks = new Task[5];
+        for (var i = 0; i < tasks.Length; i++)
+        {
+            tasks[i] = Task.Factory.StartNew(static state =>
+            {
+                var (sync, taskIndex, processor) = ((ManualResetEvent, int, BatchProcessor))state!;
+                sync.WaitOne(5_000);
+                for (var i = 0; i < 100; i++)
+                {
+                    processor.Enqueue(CreateLog($"{taskIndex}-{i}"));
+                }
+            }, (sync, i, processor));
+        }
+
+        sync.Set();
+        await Task.WhenAll(tasks);
+
+        AssertCaptureEnvelope(5 * 100 / 100);
+    }
+
     private static SentryLog CreateLog(string message)
     {
         return new SentryLog(DateTimeOffset.MinValue, SentryId.Empty, SentryLogLevel.Trace, message);