feat: Add optional Task support for BatchExportProcessor and PeriodicExportingMetricReader

This change adds the optional support for threads for BatchExportProcessor and PeriodicExportingMetricReader, in order to support .NET's WebAssembly single threaded mode.

Author: jeromelaban

src/OpenTelemetry/.publicApi/Stable/PublicAPI.Shipped.txt

@@ -31,9 +31,9 @@ OpenTelemetry.Batch<T>.Enumerator.Enumerator() -> void
 OpenTelemetry.Batch<T>.Enumerator.MoveNext() -> bool
 OpenTelemetry.Batch<T>.Enumerator.Reset() -> void
 OpenTelemetry.BatchActivityExportProcessor
-OpenTelemetry.BatchActivityExportProcessor.BatchActivityExportProcessor(OpenTelemetry.BaseExporter<System.Diagnostics.Activity!>! exporter, int maxQueueSize = 2048, int scheduledDelayMilliseconds = 5000, int exporterTimeoutMilliseconds = 30000, int maxExportBatchSize = 512) -> void
+OpenTelemetry.BatchActivityExportProcessor.BatchActivityExportProcessor(OpenTelemetry.BaseExporter<System.Diagnostics.Activity!>! exporter, int maxQueueSize, int scheduledDelayMilliseconds, int exporterTimeoutMilliseconds, int maxExportBatchSize) -> void
 OpenTelemetry.BatchExportProcessor<T>
-OpenTelemetry.BatchExportProcessor<T>.BatchExportProcessor(OpenTelemetry.BaseExporter<T!>! exporter, int maxQueueSize = 2048, int scheduledDelayMilliseconds = 5000, int exporterTimeoutMilliseconds = 30000, int maxExportBatchSize = 512) -> void
+OpenTelemetry.BatchExportProcessor<T>.BatchExportProcessor(OpenTelemetry.BaseExporter<T!>! exporter, int maxQueueSize, int scheduledDelayMilliseconds, int exporterTimeoutMilliseconds, int maxExportBatchSize) -> void
 OpenTelemetry.BatchExportProcessorOptions<T>
 OpenTelemetry.BatchExportProcessorOptions<T>.BatchExportProcessorOptions() -> void
 OpenTelemetry.BatchExportProcessorOptions<T>.ExporterTimeoutMilliseconds.get -> int
@@ -45,7 +45,7 @@ OpenTelemetry.BatchExportProcessorOptions<T>.MaxQueueSize.set -> void
 OpenTelemetry.BatchExportProcessorOptions<T>.ScheduledDelayMilliseconds.get -> int
 OpenTelemetry.BatchExportProcessorOptions<T>.ScheduledDelayMilliseconds.set -> void
 OpenTelemetry.BatchLogRecordExportProcessor
-OpenTelemetry.BatchLogRecordExportProcessor.BatchLogRecordExportProcessor(OpenTelemetry.BaseExporter<OpenTelemetry.Logs.LogRecord!>! exporter, int maxQueueSize = 2048, int scheduledDelayMilliseconds = 5000, int exporterTimeoutMilliseconds = 30000, int maxExportBatchSize = 512) -> void
+OpenTelemetry.BatchLogRecordExportProcessor.BatchLogRecordExportProcessor(OpenTelemetry.BaseExporter<OpenTelemetry.Logs.LogRecord!>! exporter, int maxQueueSize, int scheduledDelayMilliseconds, int exporterTimeoutMilliseconds, int maxExportBatchSize) -> void
 OpenTelemetry.CompositeProcessor<T>
 OpenTelemetry.CompositeProcessor<T>.AddProcessor(OpenTelemetry.BaseProcessor<T>! processor) -> OpenTelemetry.CompositeProcessor<T>!
 OpenTelemetry.CompositeProcessor<T>.CompositeProcessor(System.Collections.Generic.IEnumerable<OpenTelemetry.BaseProcessor<T>!>! processors) -> void
@@ -256,7 +256,7 @@ OpenTelemetry.Metrics.MetricType.LongSum = 26 -> OpenTelemetry.Metrics.MetricTyp
 OpenTelemetry.Metrics.MetricType.LongSumNonMonotonic = 138 -> OpenTelemetry.Metrics.MetricType
 OpenTelemetry.Metrics.MetricTypeExtensions
 OpenTelemetry.Metrics.PeriodicExportingMetricReader
-OpenTelemetry.Metrics.PeriodicExportingMetricReader.PeriodicExportingMetricReader(OpenTelemetry.BaseExporter<OpenTelemetry.Metrics.Metric!>! exporter, int exportIntervalMilliseconds = 60000, int exportTimeoutMilliseconds = 30000) -> void
+OpenTelemetry.Metrics.PeriodicExportingMetricReader.PeriodicExportingMetricReader(OpenTelemetry.BaseExporter<OpenTelemetry.Metrics.Metric!>! exporter, int exportIntervalMilliseconds, int exportTimeoutMilliseconds) -> void
 OpenTelemetry.Metrics.PeriodicExportingMetricReaderOptions
 OpenTelemetry.Metrics.PeriodicExportingMetricReaderOptions.ExportIntervalMilliseconds.get -> int?
 OpenTelemetry.Metrics.PeriodicExportingMetricReaderOptions.ExportIntervalMilliseconds.set -> void

src/OpenTelemetry/.publicApi/Stable/PublicAPI.Unshipped.txt

@@ -0,0 +1,4 @@
+OpenTelemetry.BatchActivityExportProcessor.BatchActivityExportProcessor(OpenTelemetry.BaseExporter<System.Diagnostics.Activity!>! exporter, bool useThreads = true, int maxQueueSize = 2048, int scheduledDelayMilliseconds = 5000, int exporterTimeoutMilliseconds = 30000, int maxExportBatchSize = 512) -> void
+OpenTelemetry.BatchExportProcessor<T>.BatchExportProcessor(OpenTelemetry.BaseExporter<T!>! exporter, bool useThreads = true, int maxQueueSize = 2048, int scheduledDelayMilliseconds = 5000, int exporterTimeoutMilliseconds = 30000, int maxExportBatchSize = 512) -> void
+OpenTelemetry.BatchLogRecordExportProcessor.BatchLogRecordExportProcessor(OpenTelemetry.BaseExporter<OpenTelemetry.Logs.LogRecord!>! exporter, bool useThreads = true, int maxQueueSize = 2048, int scheduledDelayMilliseconds = 5000, int exporterTimeoutMilliseconds = 30000, int maxExportBatchSize = 512) -> void
+OpenTelemetry.Metrics.PeriodicExportingMetricReader.PeriodicExportingMetricReader(OpenTelemetry.BaseExporter<OpenTelemetry.Metrics.Metric!>! exporter, bool useThreads = true, int exportIntervalMilliseconds = 60000, int exportTimeoutMilliseconds = 30000) -> void

src/OpenTelemetry/BatchExportProcessor.cs

@@ -20,16 +20,13 @@ public abstract class BatchExportProcessor<T> : BaseExportProcessor<T>
     internal const int DefaultMaxExportBatchSize = 512;
 
     internal readonly int MaxExportBatchSize;
+
     internal readonly int ScheduledDelayMilliseconds;
     internal readonly int ExporterTimeoutMilliseconds;
 
     private readonly CircularBuffer<T> circularBuffer;
-    private readonly Thread exporterThread;
-    private readonly AutoResetEvent exportTrigger = new(false);
-    private readonly ManualResetEvent dataExportedNotification = new(false);
-    private readonly ManualResetEvent shutdownTrigger = new(false);
-    private long shutdownDrainTarget = long.MaxValue;
-    private long droppedCount;
+    private readonly BatchExportWorker<T> worker;
+    private readonly bool useThreads;
     private bool disposed;
 
     /// <summary>
@@ -42,6 +39,26 @@ public abstract class BatchExportProcessor<T> : BaseExportProcessor<T>
     /// <param name="maxExportBatchSize">The maximum batch size of every export. It must be smaller or equal to maxQueueSize. The default value is 512.</param>
     protected BatchExportProcessor(
         BaseExporter<T> exporter,
+        int maxQueueSize,
+        int scheduledDelayMilliseconds,
+        int exporterTimeoutMilliseconds,
+        int maxExportBatchSize)
+        : this(exporter, true, maxQueueSize, scheduledDelayMilliseconds, exporterTimeoutMilliseconds, maxExportBatchSize)
+    {
+    }
+
+    /// <summary>
+    /// Initializes a new instance of the <see cref="BatchExportProcessor{T}"/> class.
+    /// </summary>
+    /// <param name="exporter">Exporter instance.</param>
+    /// <param name="useThreads">Enables the use of <see cref="Thread" /> when true, <see cref="Task"/> when false.</param>
+    /// <param name="maxQueueSize">The maximum queue size. After the size is reached data are dropped. The default value is 2048.</param>
+    /// <param name="scheduledDelayMilliseconds">The delay interval in milliseconds between two consecutive exports. The default value is 5000.</param>
+    /// <param name="exporterTimeoutMilliseconds">How long the export can run before it is cancelled. The default value is 30000.</param>
+    /// <param name="maxExportBatchSize">The maximum batch size of every export. It must be smaller or equal to maxQueueSize. The default value is 512.</param>
+    protected BatchExportProcessor(
+        BaseExporter<T> exporter,
+        bool useThreads = true,
         int maxQueueSize = DefaultMaxQueueSize,
         int scheduledDelayMilliseconds = DefaultScheduledDelayMilliseconds,
         int exporterTimeoutMilliseconds = DefaultExporterTimeoutMilliseconds,
@@ -57,20 +74,16 @@ protected BatchExportProcessor(
         this.ScheduledDelayMilliseconds = scheduledDelayMilliseconds;
         this.ExporterTimeoutMilliseconds = exporterTimeoutMilliseconds;
         this.MaxExportBatchSize = maxExportBatchSize;
-        this.exporterThread = new Thread(this.ExporterProc)
-        {
-            IsBackground = true,
-#pragma warning disable CA1062 // Validate arguments of public methods - needed for netstandard2.1
-            Name = $"OpenTelemetry-{nameof(BatchExportProcessor<T>)}-{exporter.GetType().Name}",
-#pragma warning restore CA1062 // Validate arguments of public methods - needed for netstandard2.1
-        };
-        this.exporterThread.Start();
+        this.useThreads = useThreads;
+
+        this.worker = this.CreateWorker();
+        this.worker.Start();
     }
 
     /// <summary>
     /// Gets the number of telemetry objects dropped by the processor.
     /// </summary>
-    internal long DroppedCount => Volatile.Read(ref this.droppedCount);
+    internal long DroppedCount => this.worker.DroppedCount;
 
     /// <summary>
     /// Gets the number of telemetry objects received by the processor.
@@ -89,20 +102,14 @@ internal bool TryExport(T data)
         {
             if (this.circularBuffer.Count >= this.MaxExportBatchSize)
             {
-                try
-                {
-                    this.exportTrigger.Set();
-                }
-                catch (ObjectDisposedException)
-                {
-                }
+                this.worker.TriggerExport();
             }
 
             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);
+        this.worker.IncrementDroppedCount();
 
         return false;
     }
@@ -116,113 +123,29 @@ protected override void OnExport(T data)
     /// <inheritdoc/>
     protected override bool OnForceFlush(int timeoutMilliseconds)
     {
-        var tail = this.circularBuffer.RemovedCount;
-        var head = this.circularBuffer.AddedCount;
-
-        if (head == tail)
-        {
-            return true; // nothing to flush
-        }
-
-        try
-        {
-            this.exportTrigger.Set();
-        }
-        catch (ObjectDisposedException)
-        {
-            return false;
-        }
-
-        if (timeoutMilliseconds == 0)
-        {
-            return false;
-        }
-
-        var triggers = new WaitHandle[] { this.dataExportedNotification, this.shutdownTrigger };
-
-        var sw = timeoutMilliseconds == Timeout.Infinite
-            ? null
-            : Stopwatch.StartNew();
-
-        // There is a chance that the export thread finished processing all the data from the queue,
-        // and signaled before we enter wait here, use polling to prevent being blocked indefinitely.
-        const int pollingMilliseconds = 1000;
-
-        while (true)
-        {
-            if (sw == null)
-            {
-                try
-                {
-                    WaitHandle.WaitAny(triggers, pollingMilliseconds);
-                }
-                catch (ObjectDisposedException)
-                {
-                    return false;
-                }
-            }
-            else
-            {
-                var timeout = timeoutMilliseconds - sw.ElapsedMilliseconds;
-
-                if (timeout <= 0)
-                {
-                    return this.circularBuffer.RemovedCount >= head;
-                }
-
-                try
-                {
-                    WaitHandle.WaitAny(triggers, Math.Min((int)timeout, pollingMilliseconds));
-                }
-                catch (ObjectDisposedException)
-                {
-                    return false;
-                }
-            }
-
-            if (this.circularBuffer.RemovedCount >= head)
-            {
-                return true;
-            }
-
-            if (Volatile.Read(ref this.shutdownDrainTarget) != long.MaxValue)
-            {
-                return false;
-            }
-        }
+        return this.worker.WaitForExport(timeoutMilliseconds);
     }
 
     /// <inheritdoc/>
     protected override bool OnShutdown(int timeoutMilliseconds)
     {
-        Volatile.Write(ref this.shutdownDrainTarget, this.circularBuffer.AddedCount);
-
-        try
-        {
-            this.shutdownTrigger.Set();
-        }
-        catch (ObjectDisposedException)
-        {
-            return false;
-        }
+        var result = this.worker.Shutdown(timeoutMilliseconds);
 
         OpenTelemetrySdkEventSource.Log.DroppedExportProcessorItems(this.GetType().Name, this.exporter.GetType().Name, this.DroppedCount);
 
         if (timeoutMilliseconds == Timeout.Infinite)
         {
-            this.exporterThread.Join();
-            return this.exporter.Shutdown();
+            return this.exporter.Shutdown() && result;
         }
 
         if (timeoutMilliseconds == 0)
         {
-            return this.exporter.Shutdown(0);
+            return this.exporter.Shutdown(0) && result;
         }
 
         var sw = Stopwatch.StartNew();
-        this.exporterThread.Join(timeoutMilliseconds);
         var timeout = timeoutMilliseconds - sw.ElapsedMilliseconds;
-        return this.exporter.Shutdown((int)Math.Max(timeout, 0));
+        return this.exporter.Shutdown((int)Math.Max(timeout, 0)) && result;
     }
 
     /// <inheritdoc/>
@@ -232,9 +155,7 @@ protected override void Dispose(bool disposing)
         {
             if (disposing)
             {
-                this.exportTrigger.Dispose();
-                this.dataExportedNotification.Dispose();
-                this.shutdownTrigger.Dispose();
+                this.worker?.Dispose();
             }
 
             this.disposed = true;
@@ -243,49 +164,27 @@ protected override void Dispose(bool disposing)
         base.Dispose(disposing);
     }
 
-    private void ExporterProc()
+    private BatchExportWorker<T> CreateWorker()
     {
-        var triggers = new WaitHandle[] { this.exportTrigger, this.shutdownTrigger };
-
-        while (true)
+#if NET
+        // Use task-based worker for browser platform where threading may be limited
+        if (OperatingSystem.IsBrowser() || !this.useThreads)
         {
-            // only wait when the queue doesn't have enough items, otherwise keep busy and send data continuously
-            if (this.circularBuffer.Count < this.MaxExportBatchSize)
-            {
-                try
-                {
-                    WaitHandle.WaitAny(triggers, this.ScheduledDelayMilliseconds);
-                }
-                catch (ObjectDisposedException)
-                {
-                    // the exporter is somehow disposed before the worker thread could finish its job
-                    return;
-                }
-            }
-
-            if (this.circularBuffer.Count > 0)
-            {
-                using (var batch = new Batch<T>(this.circularBuffer, this.MaxExportBatchSize))
-                {
-                    this.exporter.Export(batch);
-                }
-
-                try
-                {
-                    this.dataExportedNotification.Set();
-                    this.dataExportedNotification.Reset();
-                }
-                catch (ObjectDisposedException)
-                {
-                    // the exporter is somehow disposed before the worker thread could finish its job
-                    return;
-                }
-            }
-
-            if (this.circularBuffer.RemovedCount >= Volatile.Read(ref this.shutdownDrainTarget))
-            {
-                return;
-            }
+            return new BatchExportTaskWorker<T>(
+                this.circularBuffer,
+                this.exporter,
+                this.MaxExportBatchSize,
+                this.ScheduledDelayMilliseconds,
+                this.ExporterTimeoutMilliseconds);
         }
+#endif
+
+        // Use thread-based worker for all other platforms
+        return new BatchExportThreadWorker<T>(
+            this.circularBuffer,
+            this.exporter,
+            this.MaxExportBatchSize,
+            this.ScheduledDelayMilliseconds,
+            this.ExporterTimeoutMilliseconds);
     }
 }