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

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

@@ -64,21 +64,23 @@
    * @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,7 @@ public void run() {
         if (flushRequested.get() != null) {
           flush();
         }
-        JcTools.drain(
-            queue, maxExportBatchSize - batch.size(), span -> batch.add(span.toSpanData()));
+        drain(maxExportBatchSize - batch.size());
 
         if (batch.size() >= maxExportBatchSize || System.nanoTime() >= nextExportTime) {
           exportCurrentBatch();
@@ -274,13 +274,17 @@ public void run() {
       }
     }
 
+    private int drain(int limit) {
+      int drained = JcTools.drain(queue, limit, span -> batch.add(span.toSpanData()));
+      queueSize.addAndGet(-drained);
+      return drained;
+    }
+
     private void flush() {
-      int spansToFlush = queue.size();
+      int spansToFlush = queueSize.get();
       while (spansToFlush > 0) {
-        ReadableSpan span = queue.poll();
-        assert span != null;
-        batch.add(span.toSpanData());
-        spansToFlush--;
+        int drained = drain(maxExportBatchSize - batch.size());
+        spansToFlush -= drained;
         if (batch.size() >= maxExportBatchSize) {
           exportCurrentBatch();
         }