Replace jctools NonBlockingHashMap with ConcurrentHashMap (#9700)

amarziali · web-flow · commit 2e10824bcfe8 · 2025-10-08T12:56:24.000-04:00
diff --git a/dd-java-agent/agent-bootstrap/src/main/resources/META-INF/native-image/com.datadoghq/dd-java-agent/reflect-config.json b/dd-java-agent/agent-bootstrap/src/main/resources/META-INF/native-image/com.datadoghq/dd-java-agent/reflect-config.json
@@ -181,11 +181,5 @@
     "fields": [
       {"name": "consumerIndex", "allowUnsafeAccess": true}
     ]
-  },
-  {
-    "name" : "datadog.jctools.maps.NonBlockingHashMap",
-    "fields": [
-      {"name": "_kvs", "allowUnsafeAccess": true}
-    ]
   }
 ]
diff --git a/dd-java-agent/agent-profiling/profiling-controller-jfr/src/main/java/com/datadog/profiling/controller/jfr/parser/MetadataEvent.java b/dd-java-agent/agent-profiling/profiling-controller-jfr/src/main/java/com/datadog/profiling/controller/jfr/parser/MetadataEvent.java
@@ -2,7 +2,6 @@
 
 import java.io.IOException;
 import java.nio.charset.StandardCharsets;
-import org.jctools.maps.NonBlockingHashMapLong;
 
 /**
  * JFR Chunk metadata
@@ -17,10 +16,6 @@ public final class MetadataEvent {
   public final long duration;
   public final long metadataId;
 
-  private final NonBlockingHashMapLong<String> eventTypeNameMapBacking =
-      new NonBlockingHashMapLong<>(256);
-  private final LongMapping<String> eventTypeMap;
-
   MetadataEvent(RecordingStream stream) throws IOException {
     size = (int) stream.readVarint();
     long typeId = stream.readVarint();
@@ -31,16 +26,6 @@ public final class MetadataEvent {
     duration = stream.readVarint();
     metadataId = stream.readVarint();
     readElements(stream, readStringTable(stream));
-    eventTypeMap = eventTypeNameMapBacking::get;
-  }
-
-  /**
-   * Lazily compute and return the mappings of event type ids to event type names
-   *
-   * @return mappings of event type ids to event type names
-   */
-  public LongMapping<String> getEventTypeNameMap() {
-    return eventTypeMap;
   }
 
   private String[] readStringTable(RecordingStream stream) throws IOException {
@@ -76,10 +61,6 @@ private void readElements(RecordingStream stream, String[] stringConstants) thro
         }
       }
     }
-    // only event types are currently collected
-    if (name != null && id != null && "jdk.jfr.Event".equals(superType)) {
-      eventTypeNameMapBacking.put(Long.parseLong(id), name);
-    }
     // now inspect all the enclosed elements
     int elemCount = (int) stream.readVarint();
     for (int i = 0; i < elemCount; i++) {
diff --git a/dd-trace-core/src/jmh/java/datadog/trace/common/NonBlockingHashMapBenchmark.java b/dd-trace-core/src/jmh/java/datadog/trace/common/NonBlockingHashMapBenchmark.java
@@ -0,0 +1,68 @@
+package datadog.trace.common;
+
+import static java.util.concurrent.TimeUnit.MICROSECONDS;
+import static java.util.concurrent.TimeUnit.SECONDS;
+
+import de.thetaphi.forbiddenapis.SuppressForbidden;
+import java.util.concurrent.ConcurrentHashMap;
+import org.jctools.maps.NonBlockingHashMap;
+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.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;
+
+/*
+JDK 1.8
+Benchmark                                            Mode  Cnt  Score   Error  Units
+NonBlockingHashMapBenchmark.benchConcurrentHashMap   avgt       1.153          us/op
+NonBlockingHashMapBenchmark.benchNonBlockingHashMap  avgt       1.457          us/op
+
+JDK 21
+Benchmark                                            Mode  Cnt  Score   Error  Units
+NonBlockingHashMapBenchmark.benchConcurrentHashMap   avgt       1.088          us/op
+NonBlockingHashMapBenchmark.benchNonBlockingHashMap  avgt       1.278          us/op
+ */
+@State(Scope.Benchmark)
+@Warmup(iterations = 1, time = 30, timeUnit = SECONDS)
+@Measurement(iterations = 1, time = 30, timeUnit = SECONDS)
+@BenchmarkMode(Mode.AverageTime)
+@OutputTimeUnit(MICROSECONDS)
+@Fork(value = 1)
+@SuppressForbidden
+public class NonBlockingHashMapBenchmark {
+  private NonBlockingHashMap nonBlockingHashMap;
+  private ConcurrentHashMap concurrentHashMap;
+
+  @Setup(Level.Iteration)
+  public void setup() {
+    nonBlockingHashMap = new NonBlockingHashMap(512);
+    concurrentHashMap = new ConcurrentHashMap(512);
+    for (int i = 0; i < 256; i++) {
+      nonBlockingHashMap.put("test" + i, "test");
+      concurrentHashMap.put("test" + i, "test");
+    }
+  }
+
+  @Benchmark
+  @Threads(Threads.MAX)
+  public void benchNonBlockingHashMap(Blackhole blackhole) {
+    nonBlockingHashMap.put("test", "test");
+    blackhole.consume(nonBlockingHashMap.remove("test"));
+  }
+
+  @Benchmark
+  @Threads(Threads.MAX)
+  public void benchConcurrentHashMap(Blackhole blackhole) {
+    concurrentHashMap.put("test", "test");
+    blackhole.consume(concurrentHashMap.remove("test"));
+  }
+}
diff --git a/dd-trace-core/src/main/java/datadog/trace/common/metrics/Aggregator.java b/dd-trace-core/src/main/java/datadog/trace/common/metrics/Aggregator.java
@@ -8,8 +8,8 @@
 import java.util.Map;
 import java.util.Queue;
 import java.util.Set;
+import java.util.concurrent.ConcurrentMap;
 import java.util.concurrent.TimeUnit;
-import org.jctools.maps.NonBlockingHashMap;
 import org.jctools.queues.MessagePassingQueue;
 import org.jctools.queues.MpscCompoundQueue;
 import org.slf4j.Logger;
@@ -24,7 +24,7 @@ final class Aggregator implements Runnable {
   private final Queue<Batch> batchPool;
   private final MpscCompoundQueue<InboxItem> inbox;
   private final LRUCache<MetricKey, AggregateMetric> aggregates;
-  private final NonBlockingHashMap<MetricKey, Batch> pending;
+  private final ConcurrentMap<MetricKey, Batch> pending;
   private final Set<MetricKey> commonKeys;
   private final MetricWriter writer;
   // the reporting interval controls how much history will be buffered
@@ -40,7 +40,7 @@ final class Aggregator implements Runnable {
       MetricWriter writer,
       Queue<Batch> batchPool,
       MpscCompoundQueue<InboxItem> inbox,
-      NonBlockingHashMap<MetricKey, Batch> pending,
+      ConcurrentMap<MetricKey, Batch> pending,
       final Set<MetricKey> commonKeys,
       int maxAggregates,
       long reportingInterval,
@@ -61,7 +61,7 @@ final class Aggregator implements Runnable {
       MetricWriter writer,
       Queue<Batch> batchPool,
       MpscCompoundQueue<InboxItem> inbox,
-      NonBlockingHashMap<MetricKey, Batch> pending,
+      ConcurrentMap<MetricKey, Batch> pending,
       final Set<MetricKey> commonKeys,
       int maxAggregates,
       long reportingInterval,
diff --git a/dd-trace-core/src/main/java/datadog/trace/common/metrics/ConflatingMetricsAggregator.java b/dd-trace-core/src/main/java/datadog/trace/common/metrics/ConflatingMetricsAggregator.java
@@ -42,10 +42,10 @@
 import java.util.Queue;
 import java.util.Set;
 import java.util.concurrent.CompletableFuture;
+import java.util.concurrent.ConcurrentHashMap;
 import java.util.concurrent.Future;
 import java.util.concurrent.TimeUnit;
 import java.util.function.Function;
-import org.jctools.maps.NonBlockingHashMap;
 import org.jctools.queues.MpscCompoundQueue;
 import org.jctools.queues.SpmcArrayQueue;
 import org.slf4j.Logger;
@@ -90,8 +90,8 @@ public final class ConflatingMetricsAggregator implements MetricsAggregator, Eve
 
   private final Set<String> ignoredResources;
   private final Queue<Batch> batchPool;
-  private final NonBlockingHashMap<MetricKey, Batch> pending;
-  private final NonBlockingHashMap<MetricKey, MetricKey> keys;
+  private final ConcurrentHashMap<MetricKey, Batch> pending;
+  private final ConcurrentHashMap<MetricKey, MetricKey> keys;
   private final Thread thread;
   private final MpscCompoundQueue<InboxItem> inbox;
   private final Sink sink;
@@ -178,8 +178,8 @@ public ConflatingMetricsAggregator(
     this.ignoredResources = ignoredResources;
     this.inbox = new MpscCompoundQueue<>(queueSize);
     this.batchPool = new SpmcArrayQueue<>(maxAggregates);
-    this.pending = new NonBlockingHashMap<>(maxAggregates * 4 / 3);
-    this.keys = new NonBlockingHashMap<>();
+    this.pending = new ConcurrentHashMap<>(maxAggregates * 4 / 3);
+    this.keys = new ConcurrentHashMap<>();
     this.features = features;
     this.healthMetrics = healthMetric;
     this.sink = sink;

Original file line number	Diff line number	Diff line change
`@@ -181,11 +181,5 @@`
`181`	`181`	`"fields": [`
`182`	`182`	`{"name": "consumerIndex", "allowUnsafeAccess": true}`
`183`	`183`	`]`
`184`		`- },`
`185`		`- {`
`186`		`- "name" : "datadog.jctools.maps.NonBlockingHashMap",`
`187`		`- "fields": [`
`188`		`- {"name": "_kvs", "allowUnsafeAccess": true}`
`189`		`- ]`
`190`	`184`	`}`
`191`	`185`	`]`