Optimize bsp

context/src/strictContextEnabledTest/java/io/opentelemetry/context/StrictContextStorageTest.java

@@ -141,7 +141,8 @@ void scope_close_onWrongThread(Supplier<Scope> method, String methodName) throws
 
   @SuppressWarnings("ReturnValueIgnored")
   void decorator_close_withLeakedScope(Supplier<Scope> method, String methodName) throws Exception {
-    Thread thread = new Thread(method::get);
+    AtomicReference<Scope> scope = new AtomicReference<>();
+    Thread thread = new Thread(() -> scope.set(method.get()));
     thread.setName("t1");
     thread.start();
     thread.join();

docs/jmh.md

@@ -0,0 +1,21 @@
+
+# how to jmh
+
+[jmh] (Java Benchmark Harness) is a tool for running benchmarks and reporting results.
+
+opentelemetry-java has a lot of micro benchmarks. They live inside 
+`jmh` directories in the appropriate module.
+
+The benchmarks are run with a gradle plugin.
+
+To run an entire suite for a module, you can run the jmh gradle task.  
+As an example, here's how you can run the benchmarks for all of 
+the sdk trace module.
+
+```
+`./gradlew :sdk:trace:jmh`
+```
+
+If you just want to run a single benchmark and not the entire suite: 
+
+`./gradlew -PjmhIncludeSingleClass=BatchSpanProcessorBenchmark :sdk:trace:jmh`

sdk-extensions/autoconfigure/src/test/java/io/opentelemetry/sdk/autoconfigure/TracerProviderConfigurationTest.java

@@ -19,7 +19,6 @@
 import java.util.Collections;
 import java.util.HashMap;
 import java.util.Map;
-import java.util.concurrent.ArrayBlockingQueue;
 import java.util.concurrent.TimeUnit;
 import org.junit.jupiter.api.Test;
 import org.junit.jupiter.api.extension.ExtendWith;
@@ -83,11 +82,6 @@ void configureSpanProcessor_empty() {
                     .extracting("exporterTimeoutNanos")
                     .isEqualTo(TimeUnit.MILLISECONDS.toNanos(30000));
                 assertThat(worker).extracting("maxExportBatchSize").isEqualTo(512);
-                assertThat(worker)
-                    .extracting("queue")
-                    .isInstanceOfSatisfying(
-                        ArrayBlockingQueue.class,
-                        queue -> assertThat(queue.remainingCapacity()).isEqualTo(2048));
                 assertThat(worker).extracting("spanExporter").isEqualTo(exporter);
               });
     } finally {
@@ -119,11 +113,6 @@ void configureSpanProcessor_configured() {
                     .extracting("exporterTimeoutNanos")
                     .isEqualTo(TimeUnit.MILLISECONDS.toNanos(4));
                 assertThat(worker).extracting("maxExportBatchSize").isEqualTo(3);
-                assertThat(worker)
-                    .extracting("queue")
-                    .isInstanceOfSatisfying(
-                        ArrayBlockingQueue.class,
-                        queue -> assertThat(queue.remainingCapacity()).isEqualTo(2));
                 assertThat(worker).extracting("spanExporter").isEqualTo(exporter);
               });
     } finally {

sdk-extensions/resources/src/test/java/io/opentelemetry/sdk/extension/resources/ProcessRuntimeResourceTest.java

@@ -19,7 +19,7 @@ class ProcessRuntimeResourceTest {
   @Test
   void shouldCreateRuntimeAttributes() {
     // when
-    Attributes attributes = ProcessRuntimeResource.get().getAttributes();
+    Attributes attributes = ProcessRuntimeResource.buildResource().getAttributes();
 
     // then
     assertThat(attributes.get(ResourceAttributes.PROCESS_RUNTIME_NAME)).isNotBlank();

sdk/trace/build.gradle.kts

@@ -45,6 +45,7 @@ dependencies {
     jmh("io.grpc:grpc-api")
     jmh("io.grpc:grpc-netty-shaded")
     jmh("org.testcontainers:testcontainers") // testContainer for OTLP collector
+    implementation("org.jctools:jctools-core:3.2.0")
 }
 
 sourceSets {

sdk/trace/src/jmh/java/io/opentelemetry/sdk/trace/SpanBenchmark.java

@@ -51,7 +51,7 @@ public final void setup() {
   @Threads(value = 1)
   @Fork(1)
   @Warmup(iterations = 5, time = 1)
-  @Measurement(iterations = 10, time = 1)
+  @Measurement(iterations = 50, time = 1)
   @OutputTimeUnit(TimeUnit.MILLISECONDS)
   public void simpleSpanStartAddEventEnd_01Thread() {
     doSpanWork();
@@ -61,7 +61,7 @@ public void simpleSpanStartAddEventEnd_01Thread() {
   @Threads(value = 5)
   @Fork(1)
   @Warmup(iterations = 5, time = 1)
-  @Measurement(iterations = 10, time = 1)
+  @Measurement(iterations = 50, time = 1)
   @OutputTimeUnit(TimeUnit.MILLISECONDS)
   public void simpleSpanStartAddEventEnd_05Threads() {
     doSpanWork();
@@ -71,7 +71,7 @@ public void simpleSpanStartAddEventEnd_05Threads() {
   @Threads(value = 2)
   @Fork(1)
   @Warmup(iterations = 5, time = 1)
-  @Measurement(iterations = 10, time = 1)
+  @Measurement(iterations = 50, time = 1)
   @OutputTimeUnit(TimeUnit.MILLISECONDS)
   public void simpleSpanStartAddEventEnd_02Threads() {
     doSpanWork();
@@ -81,7 +81,7 @@ public void simpleSpanStartAddEventEnd_02Threads() {
   @Threads(value = 10)
   @Fork(1)
   @Warmup(iterations = 5, time = 1)
-  @Measurement(iterations = 10, time = 1)
+  @Measurement(iterations = 50, time = 1)
   @OutputTimeUnit(TimeUnit.MILLISECONDS)
   public void simpleSpanStartAddEventEnd_10Threads() {
     doSpanWork();

sdk/trace/src/jmh/java/io/opentelemetry/sdk/trace/SpanPipelineBenchmark.java

@@ -11,6 +11,8 @@
 import io.opentelemetry.sdk.trace.export.BatchSpanProcessor;
 import io.opentelemetry.sdk.trace.export.SimpleSpanProcessor;
 import io.opentelemetry.sdk.trace.samplers.Sampler;
+import java.net.MalformedURLException;
+import java.net.URL;
 import java.time.Duration;
 import java.util.concurrent.TimeUnit;
 import org.openjdk.jmh.annotations.Benchmark;
@@ -66,18 +68,29 @@ public void setup() {
 
       collector.start();
 
-      String address = collector.getHost() + ":" + collector.getMappedPort(EXPOSED_PORT);
+      SpanProcessor spanProcessor = makeSpanProcessor(collector);
 
       SdkTracerProvider tracerProvider =
           SdkTracerProvider.builder()
               .setSampler(Sampler.alwaysOn())
-              .addSpanProcessor(getSpanProcessor(address))
+              .addSpanProcessor(spanProcessor)
               .build();
 
       Tracer tracerSdk = tracerProvider.get("PipelineBenchmarkTracer");
       sdkSpanBuilder = (SdkSpanBuilder) tracerSdk.spanBuilder("PipelineBenchmarkSpan");
     }
 
+    private SpanProcessor makeSpanProcessor(GenericContainer<?> collector) {
+      try {
+        String host = collector.getHost();
+        Integer port = collector.getMappedPort(EXPOSED_PORT);
+        String address = new URL("http", host, port, "").toString();
+        return getSpanProcessor(address);
+      } catch (MalformedURLException e) {
+        throw new IllegalStateException("can't make a url", e);
+      }
+    }
+
     @Benchmark
     @BenchmarkMode(Mode.Throughput)
     @Warmup(iterations = 5, time = 1)

sdk/trace/src/jmh/java/io/opentelemetry/sdk/trace/export/BatchSpanProcessorDroppedSpansBenchmark.java

@@ -11,7 +11,6 @@
 import io.opentelemetry.sdk.metrics.SdkMeterProvider;
 import io.opentelemetry.sdk.metrics.data.LongPointData;
 import io.opentelemetry.sdk.metrics.data.MetricData;
-import io.opentelemetry.sdk.metrics.export.MetricProducer;
 import io.opentelemetry.sdk.trace.ReadableSpan;
 import io.opentelemetry.sdk.trace.SdkTracerProvider;
 import io.opentelemetry.sdk.trace.data.SpanData;
@@ -52,7 +51,6 @@ public CompletableResultCode shutdown() {
 
   @State(Scope.Benchmark)
   public static class BenchmarkState {
-    private final MetricProducer metricProducer = ((SdkMeterProvider) GlobalMetricsProvider.get());
     private BatchSpanProcessor processor;
     private Tracer tracer;
     private Collection<MetricData> allMetrics;
@@ -61,6 +59,7 @@ public static class BenchmarkState {
     public final void setup() {
       SpanExporter exporter = new DelayingSpanExporter();
       processor = BatchSpanProcessor.builder(exporter).build();
+      GlobalMetricsProvider.set(SdkMeterProvider.builder().build());
 
       tracer = SdkTracerProvider.builder().build().get("benchmarkTracer");
     }
@@ -72,7 +71,7 @@ public final void tearDown() {
 
     @TearDown(Level.Iteration)
     public final void recordMetrics() {
-      allMetrics = metricProducer.collectAllMetrics();
+      allMetrics = ((SdkMeterProvider) GlobalMetricsProvider.get()).collectAllMetrics();
     }
   }
 
@@ -128,14 +127,50 @@ private long getMetric(boolean dropped) {
     }
   }
 
+  @Benchmark
+  @Fork(1)
+  @Threads(1)
+  @Warmup(iterations = 5, time = 1)
+  @Measurement(iterations = 5, time = 20)
+  @BenchmarkMode(Mode.Throughput)
+  public void export_01Threads(
+      BenchmarkState benchmarkState, @SuppressWarnings("unused") ThreadState threadState) {
+    benchmarkState.processor.onEnd(
+        (ReadableSpan) benchmarkState.tracer.spanBuilder("span").startSpan());
+  }
+
+  @Benchmark
+  @Fork(1)
+  @Threads(2)
+  @Warmup(iterations = 5, time = 1)
+  @Measurement(iterations = 5, time = 20)
+  @BenchmarkMode(Mode.Throughput)
+  public void export_02Threads(
+      BenchmarkState benchmarkState, @SuppressWarnings("unused") ThreadState threadState) {
+    benchmarkState.processor.onEnd(
+        (ReadableSpan) benchmarkState.tracer.spanBuilder("span").startSpan());
+  }
+
   /** Export spans through {@link BatchSpanProcessor}. */
   @Benchmark
   @Fork(1)
   @Threads(5)
   @Warmup(iterations = 5, time = 1)
   @Measurement(iterations = 5, time = 20)
   @BenchmarkMode(Mode.Throughput)
-  public void export(
+  public void export_05Threads(
+      BenchmarkState benchmarkState, @SuppressWarnings("unused") ThreadState threadState) {
+    benchmarkState.processor.onEnd(
+        (ReadableSpan) benchmarkState.tracer.spanBuilder("span").startSpan());
+  }
+
+  @Benchmark
+  @Fork(1)
+  @Threads(10)
+  @Warmup(iterations = 5, time = 1)
+  @Measurement(iterations = 5, time = 20)
+  @BenchmarkMode(Mode.Throughput)
+  public void export_10Threads(
       BenchmarkState benchmarkState, @SuppressWarnings("unused") ThreadState threadState) {
     benchmarkState.processor.onEnd(
         (ReadableSpan) benchmarkState.tracer.spanBuilder("span").startSpan());

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

@@ -17,13 +17,14 @@
 import io.opentelemetry.sdk.trace.ReadableSpan;
 import io.opentelemetry.sdk.trace.SpanProcessor;
 import io.opentelemetry.sdk.trace.data.SpanData;
+import org.jctools.queues.MpscArrayQueue;
 import java.util.ArrayList;
-import java.util.Collections;
-import java.util.concurrent.ArrayBlockingQueue;
-import java.util.concurrent.BlockingQueue;
+import java.util.Collection;
 import java.util.concurrent.TimeUnit;
 import java.util.concurrent.atomic.AtomicBoolean;
 import java.util.concurrent.atomic.AtomicReference;
+import java.util.concurrent.locks.Condition;
+import java.util.concurrent.locks.ReentrantLock;
 import java.util.logging.Level;
 import java.util.logging.Logger;
 
@@ -72,7 +73,7 @@ public static BatchSpanProcessorBuilder builder(SpanExporter spanExporter) {
             scheduleDelayNanos,
             maxExportBatchSize,
             exporterTimeoutNanos,
-            new ArrayBlockingQueue<>(maxQueueSize));
+            new MpscArrayQueue<ReadableSpan>(maxQueueSize));
     Thread workerThread = new DaemonThreadFactory(WORKER_THREAD_NAME).newThread(worker);
     workerThread.start();
   }
@@ -123,26 +124,27 @@ private static final class Worker implements Runnable {
     private final long scheduleDelayNanos;
     private final int maxExportBatchSize;
     private final long exporterTimeoutNanos;
-
     private long nextExportTime;
-
-    private final BlockingQueue<ReadableSpan> queue;
-
+    private final MpscArrayQueue<ReadableSpan> queue;
     private final AtomicReference<CompletableResultCode> flushRequested = new AtomicReference<>();
     private volatile boolean continueWork = true;
-    private final ArrayList<SpanData> batch;
+    private final Collection<SpanData> batch;
+    private final ReentrantLock lock;
+    private final Condition needExport;
 
     private Worker(
         SpanExporter spanExporter,
         long scheduleDelayNanos,
         int maxExportBatchSize,
         long exporterTimeoutNanos,
-        BlockingQueue<ReadableSpan> queue) {
+        MpscArrayQueue<ReadableSpan> queue) {
       this.spanExporter = spanExporter;
       this.scheduleDelayNanos = scheduleDelayNanos;
       this.maxExportBatchSize = maxExportBatchSize;
       this.exporterTimeoutNanos = exporterTimeoutNanos;
       this.queue = queue;
+      this.lock = new ReentrantLock();
+      this.needExport = lock.newCondition();
       Meter meter = GlobalMetricsProvider.getMeter("io.opentelemetry.sdk.trace");
       meter
           .longValueObserverBuilder("queueSize")
@@ -176,27 +178,38 @@ private void addSpan(ReadableSpan span) {
       if (!queue.offer(span)) {
         droppedSpans.add(1);
       }
+      if (queue.size() >= maxExportBatchSize) {
+        lock.lock();
+        try {
+          needExport.signal();
+        } finally {
+          lock.unlock();
+        }
+      }
     }
 
     @Override
     public void run() {
       updateNextExportTime();
-
       while (continueWork) {
         if (flushRequested.get() != null) {
           flush();
         }
-
+        lock.lock();
         try {
-          ReadableSpan lastElement = queue.poll(100, TimeUnit.MILLISECONDS);
-          if (lastElement != null) {
-            batch.add(lastElement.toSpanData());
+          long pollWaitTime = nextExportTime - System.nanoTime();
+          if (pollWaitTime > 0) {
+            needExport.awaitNanos(pollWaitTime);
           }
         } catch (InterruptedException e) {
           Thread.currentThread().interrupt();
           return;
+        } finally {
+          lock.unlock();
+        }
+        while (!queue.isEmpty() && batch.size() < maxExportBatchSize) {
+          batch.add(queue.poll().toSpanData());
         }
-
         if (batch.size() >= maxExportBatchSize || System.nanoTime() >= nextExportTime) {
           exportCurrentBatch();
           updateNextExportTime();
@@ -205,12 +218,8 @@ public void run() {
     }
 
     private void flush() {
-      int spansToFlush = queue.size();
-      while (spansToFlush > 0) {
-        ReadableSpan span = queue.poll();
-        assert span != null;
-        batch.add(span.toSpanData());
-        spansToFlush--;
+      while (!queue.isEmpty()) {
+        batch.add(queue.poll().toSpanData());
         if (batch.size() >= maxExportBatchSize) {
           exportCurrentBatch();
         }
@@ -247,8 +256,15 @@ private CompletableResultCode shutdown() {
 
     private CompletableResultCode forceFlush() {
       CompletableResultCode flushResult = new CompletableResultCode();
-      // we set the atomic here to trigger the worker loop to do a flush on its next iteration.
-      flushRequested.compareAndSet(null, flushResult);
+      // we set the atomic here to trigger the worker loop to do a flush of the entire queue.
+      if (flushRequested.compareAndSet(null, flushResult)) {
+        lock.lock();
+        try {
+          needExport.signal();
+        } finally {
+          lock.unlock();
+        }
+      }
       CompletableResultCode possibleResult = flushRequested.get();
       // there's a race here where the flush happening in the worker loop could complete before we
       // get what's in the atomic. In that case, just return success, since we know it succeeded in
@@ -262,8 +278,7 @@ private void exportCurrentBatch() {
       }
 
       try {
-        final CompletableResultCode result =
-            spanExporter.export(Collections.unmodifiableList(batch));
+        final CompletableResultCode result = spanExporter.export(batch);
         result.join(exporterTimeoutNanos, TimeUnit.NANOSECONDS);
         if (result.isSuccess()) {
           exportedSpans.add(batch.size());