Add the micro-benchmark for thread filtering

ddprof-lib/src/main/cpp/threadFilter.cpp

@@ -135,7 +135,7 @@ void ThreadFilter::add(int thread_id) {
   thread_id = mapThreadId(thread_id);
   assert(b == bitmap(thread_id));
   u64 bit = 1ULL << (thread_id & 0x3f);
-  if (!(__sync_fetch_and_or(&word(b, thread_id), bit) & bit)) {
+  if (!(__atomic_fetch_or(&word(b, thread_id), bit, __ATOMIC_RELAXED) & bit)) {
     atomicInc(_size);
   }
 }
@@ -148,7 +148,7 @@ void ThreadFilter::remove(int thread_id) {
   }
 
   u64 bit = 1ULL << (thread_id & 0x3f);
-  if (__sync_fetch_and_and(&word(b, thread_id), ~bit) & bit) {
+  if (__atomic_fetch_and(&word(b, thread_id), ~bit, __ATOMIC_RELAXED) & bit) {
     atomicInc(_size, -1);
   }
 }

ddprof-lib/src/main/java/com/datadoghq/profiler/JavaProfiler.java

@@ -37,12 +37,15 @@ public final class JavaProfiler {
     static final Unsafe UNSAFE;
     static {
         Unsafe unsafe = null;
-        String version = System.getProperty("java.version");
-        try {
-            Field f = Unsafe.class.getDeclaredField("theUnsafe");
-            f.setAccessible(true);
-            unsafe = (Unsafe) f.get(null);
-        } catch (Exception ignore) { }
+        // a safety and testing valve to disable unsafe access
+        if (!Boolean.getBoolean("ddprof.disable_unsafe")) {
+            try {
+                Field f = Unsafe.class.getDeclaredField("theUnsafe");
+                f.setAccessible(true);
+                unsafe = (Unsafe) f.get(null);
+            } catch (Exception ignore) {
+            }
+        }
         UNSAFE = unsafe;
     }
 

ddprof-lib/src/test/cpp/ddprof_ut.cpp

@@ -129,7 +129,7 @@
         // increase step gradually to create different bit densities
         int step = 1;
         int size = 0;
-        for (int tid = 0; tid < maxTid - step - 1; tid += step, size++) {
+        for (int tid = 1; tid < maxTid - step - 1; tid += step, size++) {
             EXPECT_FALSE(filter.accept(tid));
             filter.add(tid);
             EXPECT_TRUE(filter.accept(tid));

ddprof-stresstest/build.gradle

@@ -39,7 +39,7 @@ task runStressTests(type: Exec) {
   }
   group = 'Execution'
   description = 'Run JMH stresstests'
-  commandLine "${javaHome}/bin/java", '-jar', 'build/libs/stresstests.jar'
+  commandLine "${javaHome}/bin/java", '-jar', 'build/libs/stresstests.jar', 'counters.*'
 }
 
 tasks.withType(JavaCompile).configureEach {

ddprof-stresstest/src/jmh/java/com/datadoghq/profiler/stresstest/Main.java

@@ -16,11 +16,18 @@ public class Main {
     public static final String SCENARIOS_PACKAGE = "com.datadoghq.profiler.stresstest.scenarios.";
 
     public static void main(String... args) throws Exception {
+        String filter = "*";
+        if (args.length == 1) {
+            filter = args[0];
+        } else if (args.length > 1) {
+            System.err.println("Usage: java -jar ddprof-stresstest.jar [scenario filter]");
+            System.exit(1);
+        }
         CommandLineOptions commandLineOptions = new CommandLineOptions(args);
         Mode mode = Mode.AverageTime;
         Options options = new OptionsBuilder()
                 .parent(new CommandLineOptions(args))
-                .include(SCENARIOS_PACKAGE + "*")
+                .include(SCENARIOS_PACKAGE + filter)
                 .addProfiler(WhiteboxProfiler.class)
                 .forks(commandLineOptions.getForkCount().orElse(1))
                 .warmupIterations(commandLineOptions.getWarmupIterations().orElse(0))

ddprof-stresstest/src/jmh/java/com/datadoghq/profiler/stresstest/WhiteboxProfiler.java

@@ -46,12 +46,16 @@ public Collection<? extends Result> afterIteration(BenchmarkParams benchmarkPara
             JavaProfiler.getInstance().stop();
             long fileSize = Files.size(jfr);
             Files.deleteIfExists(jfr);
-            List<ScalarResult> results = new ArrayList<>();
-            results.add(new ScalarResult("jfr_filesize_bytes", fileSize, "", AggregationPolicy.MAX));
-            for (Map.Entry<String, Long> counter : JavaProfiler.getInstance().getDebugCounters().entrySet()) {
-                results.add(new ScalarResult(counter.getKey(), counter.getValue(), "", AggregationPolicy.MAX));
+            if (!Boolean.parseBoolean(benchmarkParams.getParam("skipResults"))) {
+                List<ScalarResult> results = new ArrayList<>();
+                results.add(new ScalarResult("jfr_filesize_bytes", fileSize, "", AggregationPolicy.MAX));
+                for (Map.Entry<String, Long> counter : JavaProfiler.getInstance().getDebugCounters().entrySet()) {
+                    results.add(new ScalarResult(counter.getKey(), counter.getValue(), "", AggregationPolicy.MAX));
+                }
+                return results;
+            } else {
+                return Collections.emptyList();
             }
-            return results;
         } catch (IOException e) {
             e.printStackTrace();
             return Collections.emptyList();

ddprof-stresstest/src/jmh/java/com/datadoghq/profiler/stresstest/scenarios/CapturingLambdas.java → ddprof-stresstest/src/jmh/java/com/datadoghq/profiler/stresstest/scenarios/counters/CapturingLambdas.java

@@ -1,4 +1,4 @@
-package com.datadoghq.profiler.stresstest.scenarios;
+package com.datadoghq.profiler.stresstest.scenarios.counters;
 
 import com.datadoghq.profiler.stresstest.Configuration;
 import org.openjdk.jmh.annotations.Benchmark;

ddprof-stresstest/src/jmh/java/com/datadoghq/profiler/stresstest/scenarios/DumpRecording.java → ddprof-stresstest/src/jmh/java/com/datadoghq/profiler/stresstest/scenarios/counters/DumpRecording.java

@@ -1,4 +1,4 @@
-package com.datadoghq.profiler.stresstest.scenarios;
+package com.datadoghq.profiler.stresstest.scenarios.counters;
 
 import com.datadoghq.profiler.JavaProfiler;
 import com.datadoghq.profiler.stresstest.Configuration;

ddprof-stresstest/src/jmh/java/com/datadoghq/profiler/stresstest/scenarios/GraphMutation.java → ddprof-stresstest/src/jmh/java/com/datadoghq/profiler/stresstest/scenarios/counters/GraphMutation.java

@@ -1,4 +1,4 @@
-package com.datadoghq.profiler.stresstest.scenarios;
+package com.datadoghq.profiler.stresstest.scenarios.counters;
 
 import org.openjdk.jmh.annotations.Benchmark;
 import org.openjdk.jmh.annotations.Threads;

ddprof-stresstest/src/jmh/java/com/datadoghq/profiler/stresstest/scenarios/GraphState.java → ddprof-stresstest/src/jmh/java/com/datadoghq/profiler/stresstest/scenarios/counters/GraphState.java

@@ -1,4 +1,4 @@
-package com.datadoghq.profiler.stresstest.scenarios;
+package com.datadoghq.profiler.stresstest.scenarios.counters;
 
 import com.datadoghq.profiler.stresstest.Configuration;
 import org.openjdk.jmh.annotations.*;

ddprof-stresstest/src/jmh/java/com/datadoghq/profiler/stresstest/scenarios/NanoTime.java → ddprof-stresstest/src/jmh/java/com/datadoghq/profiler/stresstest/scenarios/counters/NanoTime.java

@@ -1,4 +1,4 @@
-package com.datadoghq.profiler.stresstest.scenarios;
+package com.datadoghq.profiler.stresstest.scenarios.counters;
 
 import com.datadoghq.profiler.stresstest.Configuration;
 import org.openjdk.jmh.annotations.Benchmark;

ddprof-stresstest/src/jmh/java/com/datadoghq/profiler/stresstest/scenarios/ThreadFilterBenchmark.java → ddprof-stresstest/src/jmh/java/com/datadoghq/profiler/stresstest/scenarios/counters/ThreadFilterBenchmark.java

@@ -1,9 +1,8 @@
-package com.datadoghq.profiler.stresstest.scenarios;
+package com.datadoghq.profiler.stresstest.scenarios.counters;
 
 import com.datadoghq.profiler.JavaProfiler;
 import com.datadoghq.profiler.stresstest.Configuration;
 import org.openjdk.jmh.annotations.*;
-import org.openjdk.jmh.infra.Blackhole;
 
 import java.io.FileWriter;
 import java.io.IOException;

ddprof-stresstest/src/jmh/java/com/datadoghq/profiler/stresstest/scenarios/TracedParallelWork.java → ddprof-stresstest/src/jmh/java/com/datadoghq/profiler/stresstest/scenarios/counters/TracedParallelWork.java

@@ -1,4 +1,4 @@
-package com.datadoghq.profiler.stresstest.scenarios;
+package com.datadoghq.profiler.stresstest.scenarios.counters;
 
 import com.datadoghq.profiler.ContextSetter;
 import com.datadoghq.profiler.JavaProfiler;

ddprof-stresstest/src/jmh/java/com/datadoghq/profiler/stresstest/scenarios/throughput/ThreadFilterBenchmark.java

@@ -0,0 +1,113 @@
+package com.datadoghq.profiler.stresstest.scenarios.throughput;
+
+import com.datadoghq.profiler.JavaProfiler;
+import com.datadoghq.profiler.stresstest.Configuration;
+import org.openjdk.jmh.annotations.Benchmark;
+import org.openjdk.jmh.annotations.BenchmarkMode;
+import org.openjdk.jmh.annotations.Fork;
+import org.openjdk.jmh.annotations.Level;
+import org.openjdk.jmh.annotations.Measurement;
+import org.openjdk.jmh.annotations.Mode;
+import org.openjdk.jmh.annotations.OutputTimeUnit;
+import org.openjdk.jmh.annotations.Param;
+import org.openjdk.jmh.annotations.Scope;
+import org.openjdk.jmh.annotations.Setup;
+import org.openjdk.jmh.annotations.State;
+import org.openjdk.jmh.annotations.Threads;
+import org.openjdk.jmh.annotations.Warmup;
+import org.openjdk.jmh.infra.Blackhole;
+
+import java.io.IOException;
+import java.util.concurrent.TimeUnit;
+
+@State(Scope.Benchmark)
+public class ThreadFilterBenchmark extends Configuration {
+    private JavaProfiler profiler;
+
+    @Param(BASE_COMMAND + ",filter=1")
+    public String command;
+
+    @Param("true")
+    public String skipResults;
+
+    @Param({"0", "7", "70000"})
+    public String workload;
+
+    private long workloadNum = 0;
+
+    @Setup(Level.Trial)
+    public void setup() throws IOException {
+        profiler = JavaProfiler.getInstance();
+        workloadNum = Long.parseLong(workload);
+    }
+
+    @Benchmark
+    @BenchmarkMode(Mode.Throughput)
+    @Fork(value = 3, warmups = 3)
+    @Warmup(iterations = 5)
+    @Measurement(iterations = 8)
+    @Threads(1)
+    @OutputTimeUnit(TimeUnit.MILLISECONDS)
+    public void threadFilterStress01() throws InterruptedException {
+        profiler.addThread();
+        // Simulate per-thread work
+        Blackhole.consumeCPU(workloadNum);
+        profiler.removeThread();
+    }
+
+    @Benchmark
+    @BenchmarkMode(Mode.Throughput)
+    @Fork(value = 3, warmups = 3)
+    @Warmup(iterations = 5)
+    @Measurement(iterations = 8)
+    @Threads(2)
+    @OutputTimeUnit(TimeUnit.MILLISECONDS)
+    public void threadFilterStress02() throws InterruptedException {
+        profiler.addThread();
+        // Simulate per-thread work
+        Blackhole.consumeCPU(workloadNum);
+        profiler.removeThread();
+    }
+
+    @Benchmark
+    @BenchmarkMode(Mode.Throughput)
+    @Fork(value = 3, warmups = 3)
+    @Warmup(iterations = 5)
+    @Measurement(iterations = 8)
+    @Threads(4)
+    @OutputTimeUnit(TimeUnit.MILLISECONDS)
+    public void threadFilterStress04() throws InterruptedException {
+        profiler.addThread();
+        // Simulate per-thread work
+        Blackhole.consumeCPU(workloadNum);
+        profiler.removeThread();
+    }
+
+    @Benchmark
+    @BenchmarkMode(Mode.Throughput)
+    @Fork(value = 3, warmups = 3)
+    @Warmup(iterations = 5)
+    @Measurement(iterations = 8)
+    @Threads(8)
+    @OutputTimeUnit(TimeUnit.MILLISECONDS)
+    public void threadFilterStress08() throws InterruptedException {
+        profiler.addThread();
+        // Simulate per-thread work
+        Blackhole.consumeCPU(workloadNum);
+        profiler.removeThread();
+    }
+
+    @Benchmark
+    @BenchmarkMode(Mode.Throughput)
+    @Fork(value = 3, warmups = 3)
+    @Warmup(iterations = 5)
+    @Measurement(iterations = 8)
+    @Threads(16)
+    @OutputTimeUnit(TimeUnit.MILLISECONDS)
+    public void threadFilterStress16() throws InterruptedException {
+        profiler.addThread();
+        // Simulate per-thread work
+        Blackhole.consumeCPU(workloadNum);
+        profiler.removeThread();
+    }
+}