Avoid linear queue.size() calls in span producers by storing queue size separately

Author: raipc

sdk/trace-shaded-deps/src/main/java/io/opentelemetry/sdk/trace/internal/JcTools.java

@@ -64,21 +64,23 @@ public static long capacity(Queue<?> queue) {
    * @throws IllegalArgumentException if maxExportBatchSize is negative
    */
   @SuppressWarnings("unchecked")
-  public static <T> void drain(Queue<T> queue, int limit, Consumer<T> consumer) {
+  public static <T> int drain(Queue<T> queue, int limit, Consumer<T> consumer) {
     if (queue instanceof MessagePassingQueue) {
-      ((MessagePassingQueue<T>) queue).drain(consumer::accept, limit);
+      return ((MessagePassingQueue<T>) queue).drain(consumer::accept, limit);
     } else {
-      drainNonJcQueue(queue, limit, consumer);
+      return drainNonJcQueue(queue, limit, consumer);
     }
   }
 
-  private static <T> void drainNonJcQueue(
+  private static <T> int drainNonJcQueue(
       Queue<T> queue, int maxExportBatchSize, Consumer<T> consumer) {
     int polledCount = 0;
     T item;
-    while (polledCount++ < maxExportBatchSize && (item = queue.poll()) != null) {
+    while (polledCount < maxExportBatchSize && (item = queue.poll()) != null) {
       consumer.accept(item);
+      ++polledCount;
     }
+    return polledCount;
   }
 
   private JcTools() {}

sdk/trace/src/main/java/io/opentelemetry/sdk/trace/export/BatchSpanProcessor.java

@@ -173,6 +173,7 @@ private static final class Worker implements Runnable {
     private long nextExportTime;
 
     private final Queue<ReadableSpan> queue;
+    private final AtomicInteger queueSize = new AtomicInteger();
     // When waiting on the spans queue, exporter thread sets this atomic to the number of more
     // spans it needs before doing an export. Writer threads would then wait for the queue to reach
     // spansNeeded size before notifying the exporter thread about new entries.
@@ -237,7 +238,7 @@ private void addSpan(ReadableSpan span) {
       if (!queue.offer(span)) {
         processedSpansCounter.add(1, droppedAttrs);
       } else {
-        if (queue.size() >= spansNeeded.get()) {
+        if (queueSize.incrementAndGet() >= spansNeeded.get()) {
           signal.offer(true);
         }
       }
@@ -251,8 +252,11 @@ public void run() {
         if (flushRequested.get() != null) {
           flush();
         }
-        JcTools.drain(
+        int drained = JcTools.drain(
             queue, maxExportBatchSize - batch.size(), span -> batch.add(span.toSpanData()));
+        if (drained > 0) {
+          queueSize.accumulateAndGet(-drained, Integer::sum);
+        }
 
         if (batch.size() >= maxExportBatchSize || System.nanoTime() >= nextExportTime) {
           exportCurrentBatch();