UTF8 caching for v0.4

dd-trace-api/src/main/java/datadog/trace/api/config/GeneralConfig.java

@@ -105,6 +105,8 @@ public final class GeneralConfig {
   public static final String JDK_SOCKET_ENABLED = "jdk.socket.enabled";
 
   public static final String OPTIMIZED_MAP_ENABLED = "optimized.map.enabled";
+  public static final String TAG_NAME_UTF8_CACHE_SIZE = "tag.name.utf8.cache.size";
+  public static final String TAG_VALUE_UTF8_CACHE_SIZE = "tag.value.utf8.cache.size";
   public static final String STACK_TRACE_LENGTH_LIMIT = "stack.trace.length.limit";
 
   public static final String SSI_INJECTION_ENABLED = "injection.enabled";

dd-trace-core/src/jmh/java/datadog/trace/common/writer/ddagent/Utf8Benchmark.java

@@ -0,0 +1,147 @@
+package datadog.trace.common.writer.ddagent;
+
+import java.nio.charset.StandardCharsets;
+import java.util.concurrent.ThreadLocalRandom;
+import org.openjdk.jmh.annotations.Benchmark;
+import org.openjdk.jmh.annotations.BenchmarkMode;
+import org.openjdk.jmh.annotations.Mode;
+import org.openjdk.jmh.infra.Blackhole;
+
+/**
+ * This benchmark isn't really intended to used to measure throughput, but rather to be used with
+ * "-prof gc" to check bytes / op.
+ *
+ * <p>Since {@link String#getBytes(java.nio.charset.Charset)} is intrinsified the caches typically
+ * perform worse throughput wise, the benefit of the caches is to reduce allocation. Intention of
+ * this benchmark is to create data that roughly resembles what might be seen in a trace payload.
+ * Tag names are quite static, tag values are mostly low cardinality, but some tag values have
+ * infinite cardinality.
+ */
+@BenchmarkMode(Mode.Throughput)
+public class Utf8Benchmark {
+  static final int NUM_LOOKUPS = 10_000;
+
+  static final String[] TAGS = {
+    "_dd.asm.keep",
+    "ci.provider",
+    "language",
+    "db.statement",
+    "ci.job.url",
+    "ci.pipeline.url",
+    "db.pool",
+    "http.forwarder",
+    "db.warehouse",
+    "custom"
+  };
+
+  static int pos = 0;
+  static int standardVal = 0;
+
+  static final String nextTag() {
+    if (pos == TAGS.length - 1) {
+      pos = 0;
+    } else {
+      pos += 1;
+    }
+    return TAGS[pos];
+  }
+
+  static final String nextValue(String tag) {
+    if (tag.equals("custom")) {
+      return nextCustomValue(tag);
+    } else {
+      return nextStandardValue(tag);
+    }
+  }
+
+  /*
+   * Produces a high cardinality value - > thousands of distinct values per tag - many 1-time values
+   */
+  static final String nextCustomValue(String tag) {
+    return tag + ThreadLocalRandom.current().nextInt();
+  }
+
+  /*
+   * Produces a moderate cardinality value - tens of distinct values per tag
+   */
+  static final String nextStandardValue(String tag) {
+    return tag + ThreadLocalRandom.current().nextInt(20);
+  }
+
+  @Benchmark
+  public static final String tagUtf8_baseline() {
+    return nextTag();
+  }
+
+  @Benchmark
+  public static final byte[] tagUtf8_nocache() {
+    String tag = nextTag();
+    return tag.getBytes(StandardCharsets.UTF_8);
+  }
+
+  static final SimpleUtf8Cache TAG_CACHE = new SimpleUtf8Cache(128);
+
+  @Benchmark
+  public static final byte[] tagUtf8_w_cache() {
+    String tag = nextTag();
+
+    byte[] cache = TAG_CACHE.getUtf8(tag);
+    if (cache != null) return cache;
+
+    return tag.getBytes(StandardCharsets.UTF_8);
+  }
+
+  @Benchmark
+  public static final void valueUtf8_baseline(Blackhole bh) {
+    for (int i = 0; i < NUM_LOOKUPS; ++i) {
+      String tag = nextTag();
+      String value = nextValue(tag);
+
+      bh.consume(tag);
+      bh.consume(value);
+    }
+  }
+
+  static final GenerationalUtf8Cache VALUE_CACHE = new GenerationalUtf8Cache(64, 128);
+
+  @Benchmark
+  public static final void valueUtf8_cache_generational(Blackhole bh) {
+    GenerationalUtf8Cache valueCache = VALUE_CACHE;
+    valueCache.recalibrate();
+
+    for (int i = 0; i < NUM_LOOKUPS; ++i) {
+      String tag = nextTag();
+      String value = nextValue(tag);
+
+      byte[] lookup = valueCache.getUtf8(value);
+      bh.consume(lookup);
+    }
+  }
+
+  static final SimpleUtf8Cache SIMPLE_VALUE_CACHE = new SimpleUtf8Cache(128);
+
+  @Benchmark
+  public static final void valueUtf8_cache_simple(Blackhole bh) {
+    SimpleUtf8Cache valueCache = SIMPLE_VALUE_CACHE;
+    valueCache.recalibrate();
+
+    for (int i = 0; i < NUM_LOOKUPS; ++i) {
+      String tag = nextTag();
+      String value = nextValue(tag);
+
+      byte[] lookup = valueCache.getUtf8(value);
+      bh.consume(lookup);
+    }
+  }
+
+  @Benchmark
+  public static final void valueUtf8_nocache(Blackhole bh) {
+    for (int i = 0; i < NUM_LOOKUPS; ++i) {
+      String tag = nextTag();
+      String value = nextValue(tag);
+
+      bh.consume(tag);
+      bh.consume(value.getBytes(StandardCharsets.UTF_8));
+    }
+  }
+}

dd-trace-core/src/main/java/datadog/trace/common/writer/ddagent/Caching.java

@@ -0,0 +1,81 @@
+package datadog.trace.common.writer.ddagent;
+
+import java.util.Arrays;
+
+/** Some common static functions used by simple & generational caches */
+final class Caching {
+  private Caching() {}
+
+  /**
+   * Provides the cache size that holds the requestedCapacity
+   *
+   * @param requestedCapacity > 0
+   * @return size >= requestedCapacity
+   */
+  static final int cacheSizeFor(int requestedCapacity) {
+    int pow;
+    for (pow = 1; pow < requestedCapacity; pow *= 2) ;
+    return pow;
+  }
+
+  /** Provides an "adjusted" (e.g. non-zero) hash for the given String */
+  static final int adjHash(String value) {
+    int hash = value.hashCode();
+    return (hash == 0) ? 0xDA7AD06 : hash;
+  }
+
+  /** Resets markers to zero */
+  static final void reset(int[] marks) {
+    Arrays.fill(marks, 0);
+  }
+
+  /**
+   * Changes the mark status of the corresponding slot in the marking array. If there was previously
+   * a matching mark, resets the slot to zero and returns true If there was previously a mismatching
+   * mark, updates the slot and returns false
+   *
+   * <p>A return value of true indicates that the requested value has likely been seen previously
+   * and cache entry should be created.
+   */
+  static final boolean mark(int[] marks, int newAdjHash) {
+    int index = bucketIndex(marks, newAdjHash);
+
+    // This is the 4th iteration of the marking strategy
+    // First version - used a mark entry, but that would prematurely
+    // burn a slot in the cache
+    // Second version - used a mark boolean, that worked well, but
+    // was a overly permissive in allowing the next request to the same slot
+    // to immediately create a CacheEntry
+    // Third version - used a mark hash that to match exactly,
+    // that could lead to access order fights over the cache slot
+    // So this version is a hybrid of 2nd & 3rd, using a bloom filter
+    // that effectively degenerates to a boolean
+
+    // This approach provides a nice balance when there's an A-B-A access pattern
+    // The first A will mark the slot
+    // Then B will mark the slot with A | B
+    // Then either A or B can claim and reset the slot
+
+    int priorMarkHash = marks[index];
+    boolean match = ((priorMarkHash & newAdjHash) == newAdjHash);
+    if (match) {
+      marks[index] = 0;
+    } else {
+      marks[index] = priorMarkHash | newAdjHash;
+    }
+    return match;
+  }
+
+  /** Provides the corresponding index into the marking array */
+  static final int bucketIndex(int[] marks, int adjHash) {
+    return adjHash & (marks.length - 1);
+  }
+
+  /**
+   * Provides the corresponding index into an entry array Assumes that array size was determined by
+   * using {@Caching#cacheSizeFor}
+   */
+  static final <E> int bucketIndex(E[] entries, int adjHash) {
+    return adjHash & (entries.length - 1);
+  }
+}