Add the micro-benchmark for thread filtering (#237)

jbachorik · zhengyu123 · commit c0d8e917a7db · 2025-07-09T13:29:16.000-04:00
* Add the micro-benchmark for thread filtering

* Do not test for obviously invalid thread id

* Relax threadfilter mem order
diff --git a/ddprof-lib/src/main/cpp/threadFilter.cpp b/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);
   }
 }
diff --git a/ddprof-lib/src/main/java/com/datadoghq/profiler/JavaProfiler.java b/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;
     }
 
diff --git a/ddprof-lib/src/test/cpp/ddprof_ut.cpp b/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));
diff --git a/ddprof-stresstest/build.gradle b/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 {
diff --git a/ddprof-stresstest/src/jmh/java/com/datadoghq/profiler/stresstest/Main.java b/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))
diff --git a/ddprof-stresstest/src/jmh/java/com/datadoghq/profiler/stresstest/WhiteboxProfiler.java b/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();
diff --git a/ddprof-stresstest/src/jmh/java/com/datadoghq/profiler/stresstest/scenarios/counters/CapturingLambdas.java b/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;
diff --git a/ddprof-stresstest/src/jmh/java/com/datadoghq/profiler/stresstest/scenarios/counters/DumpRecording.java b/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;
diff --git a/ddprof-stresstest/src/jmh/java/com/datadoghq/profiler/stresstest/scenarios/counters/GraphMutation.java b/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;
diff --git a/ddprof-stresstest/src/jmh/java/com/datadoghq/profiler/stresstest/scenarios/counters/GraphState.java b/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.*;
diff --git a/ddprof-stresstest/src/jmh/java/com/datadoghq/profiler/stresstest/scenarios/counters/NanoTime.java b/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;
diff --git a/ddprof-stresstest/src/jmh/java/com/datadoghq/profiler/stresstest/scenarios/counters/ThreadFilterBenchmark.java b/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;
diff --git a/ddprof-stresstest/src/jmh/java/com/datadoghq/profiler/stresstest/scenarios/counters/TracedParallelWork.java b/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;
diff --git a/ddprof-stresstest/src/jmh/java/com/datadoghq/profiler/stresstest/scenarios/throughput/ThreadFilterBenchmark.java b/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();
+    }
+}

Original file line number	Diff line number	Diff line change
`@@ -135,7 +135,7 @@ void ThreadFilter::add(int thread_id) {`
`135`	`135`	`thread_id = mapThreadId(thread_id);`
`136`	`136`	`assert(b == bitmap(thread_id));`
`137`	`137`	`u64 bit = 1ULL << (thread_id & 0x3f);`
`138`		`- if (!(__sync_fetch_and_or(&word(b, thread_id), bit) & bit)) {`
	`138`	`+ if (!(__atomic_fetch_or(&word(b, thread_id), bit, __ATOMIC_RELAXED) & bit)) {`
`139`	`139`	`atomicInc(_size);`
`140`	`140`	`}`
`141`	`141`	`}`
`@@ -148,7 +148,7 @@ void ThreadFilter::remove(int thread_id) {`
`148`	`148`	`}`
`149`	`149`
`150`	`150`	`u64 bit = 1ULL << (thread_id & 0x3f);`
`151`		`- if (__sync_fetch_and_and(&word(b, thread_id), ~bit) & bit) {`
	`151`	`+ if (__atomic_fetch_and(&word(b, thread_id), ~bit, __ATOMIC_RELAXED) & bit) {`
`152`	`152`	`atomicInc(_size, -1);`
`153`	`153`	`}`
`154`	`154`	`}`
Original file line number	Diff line number	Diff line change
`@@ -39,7 +39,7 @@ task runStressTests(type: Exec) {`
`39`	`39`	`}`
`40`	`40`	`group = 'Execution'`
`41`	`41`	`description = 'Run JMH stresstests'`
`42`		`- commandLine "${javaHome}/bin/java", '-jar', 'build/libs/stresstests.jar'`
	`42`	`+ commandLine "${javaHome}/bin/java", '-jar', 'build/libs/stresstests.jar', 'counters.*'`
`43`	`43`	`}`
`44`	`44`
`45`	`45`	`tasks.withType(JavaCompile).configureEach {`
Original file line number	Diff line number	Diff line change
`@@ -1,4 +1,4 @@`
`1`		`-package com.datadoghq.profiler.stresstest.scenarios;`
	`1`	`+package com.datadoghq.profiler.stresstest.scenarios.counters;`
`2`	`2`
`3`	`3`	`import com.datadoghq.profiler.stresstest.Configuration;`
`4`	`4`	`import org.openjdk.jmh.annotations.Benchmark;`