Fix batching tests and processor improvements (#252)

open-telemetry · Oct 9, 2019 · d3b8133 · d3b8133
1 parent 14e0d60
commit d3b8133
Show file tree

Hide file tree

Showing 3 changed files with 174 additions and 167 deletions.
diff --git a/src/OpenTelemetry/Trace/Export/BatchingSpanProcessor.cs b/src/OpenTelemetry/Trace/Export/BatchingSpanProcessor.cs
@@ -38,7 +38,7 @@ public class BatchingSpanProcessor : SpanProcessor, IDisposable
         private readonly TimeSpan scheduleDelay;
         private readonly Task worker;
         private CancellationTokenSource cts;
-        private int currentQueueSize;
+        private volatile int currentQueueSize;
         private bool stopping = false;
 
         /// <summary>
@@ -113,6 +113,7 @@ public override void OnEnd(Span span)
             }
 
             Interlocked.Increment(ref this.currentQueueSize);
+
             this.exportQueue.Enqueue(span);
         }
 
@@ -130,9 +131,7 @@ public override async Task ShutdownAsync(CancellationToken cancellationToken)
                 // if there are more items, continue until cancellation token allows
                 while (this.currentQueueSize > 0 && !cancellationToken.IsCancellationRequested)
                 {
-                    Debug.WriteLine($"!!! {DateTime.UtcNow:o}  export " + this.currentQueueSize);
                     await this.ExportBatchAsync(cancellationToken).ConfigureAwait(false);
-                    Debug.WriteLine($"!!! {DateTime.UtcNow:o}  export finished" + this.currentQueueSize);
                 }
 
                 // there is no point in waiting for a worker task if cancellation happens
@@ -161,16 +160,19 @@ private async Task ExportBatchAsync(CancellationToken cancellationToken)
                     return;
                 }
 
-                var nextBatchSize = Math.Min(this.currentQueueSize, this.maxExportBatchSize);
-
-                if (nextBatchSize == 0)
+                List<Span> batch = null;
+                if (this.exportQueue.TryDequeue(out var nextSpan))
+                {
+                    Interlocked.Decrement(ref this.currentQueueSize);
+                    batch = new List<Span> { nextSpan };
+                }
+                else
                 {
+                    // nothing in queue
                     return;
                 }
 
-                var batch = new List<Span>(nextBatchSize);
-
-                while (batch.Count < nextBatchSize && this.exportQueue.TryDequeue(out var nextSpan))
+                while (batch.Count < this.maxExportBatchSize && this.exportQueue.TryDequeue(out nextSpan))
                 {
                     Interlocked.Decrement(ref this.currentQueueSize);
                     batch.Add(nextSpan);

diff --git a/test/OpenTelemetry.Tests/Impl/Testing/Export/TestExporter.cs b/test/OpenTelemetry.Tests/Impl/Testing/Export/TestExporter.cs
@@ -14,12 +14,11 @@
 // limitations under the License.
 // </copyright>
 
-using System.Collections.Concurrent;
-
 namespace OpenTelemetry.Testing.Export
 {
     using System;
     using System.Collections.Generic;
+    using System.Collections.Concurrent;
     using System.Threading;
     using System.Threading.Tasks;
     using OpenTelemetry.Trace;