From c3def0011e5b285c73b2e4461b5be9f9fabc7ffd Mon Sep 17 00:00:00 2001 From: Douglas Q Hawkins Date: Mon, 31 Aug 2026 12:16:28 -0400 Subject: [PATCH 01/19] Add Accumulator: a striped long counter primitive as an alternative to LongAdder Enum-keyed long[]-per-stripe storage with cache-line padding, threadId&mask stripe selection, and combine+reset performed atomically under each stripe's own lock -- closing the non-atomic sumThenReset() loss window LongAdder has. Includes a JMH benchmark against LongAdder and the ConcurrentHashMap.computeIfAbsent(AtomicLong::new) anti-pattern. APMLP-1779 Co-Authored-By: Claude Sonnet 5 --- .../trace/util/AccumulatorBenchmark.java | 106 +++++++++ .../java/datadog/trace/util/Accumulator.java | 205 ++++++++++++++++++ .../datadog/trace/util/AccumulatorTest.java | 170 +++++++++++++++ 3 files changed, 481 insertions(+) create mode 100644 internal-api/src/jmh/java/datadog/trace/util/AccumulatorBenchmark.java create mode 100644 internal-api/src/main/java/datadog/trace/util/Accumulator.java create mode 100644 internal-api/src/test/java/datadog/trace/util/AccumulatorTest.java diff --git a/internal-api/src/jmh/java/datadog/trace/util/AccumulatorBenchmark.java b/internal-api/src/jmh/java/datadog/trace/util/AccumulatorBenchmark.java new file mode 100644 index 00000000000..d7c494ad9df --- /dev/null +++ b/internal-api/src/jmh/java/datadog/trace/util/AccumulatorBenchmark.java @@ -0,0 +1,106 @@ +package datadog.trace.util; + +import static java.util.concurrent.TimeUnit.MICROSECONDS; + +import java.util.concurrent.ConcurrentHashMap; +import java.util.concurrent.atomic.AtomicLong; +import java.util.concurrent.atomic.LongAdder; +import org.openjdk.jmh.annotations.Benchmark; +import org.openjdk.jmh.annotations.BenchmarkMode; +import org.openjdk.jmh.annotations.Fork; +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.State; +import org.openjdk.jmh.annotations.Threads; +import org.openjdk.jmh.annotations.Warmup; +import org.openjdk.jmh.infra.Blackhole; + +/** + * {@link Accumulator} vs {@link LongAdder} vs the {@code ConcurrentHashMap.computeIfAbsent(key, k + * -> new AtomicLong())} anti-pattern, at one thread (no contention) and at {@link Threads#MAX} + * (heavy contention). The CHM variant allocates its counter under the bucket's bin lock on first + * sight of a key -- exactly the pathology {@link Accumulator} exists to avoid -- so its comparison + * here is against that allocation-under-lock step, not against a pre-warmed map. + */ +@State(Scope.Benchmark) +@Warmup(iterations = 1, time = 10) +@Measurement(iterations = 3, time = 10) +@BenchmarkMode(Mode.AverageTime) +@OutputTimeUnit(MICROSECONDS) +@Fork(2) +public class AccumulatorBenchmark { + + enum Counter { + HITS + } + + private final LongAdder adder = new LongAdder(); + private final long[][] accumulator = Accumulator.create(Counter.values()); + private final ConcurrentHashMap chm = new ConcurrentHashMap<>(); + + @Benchmark + @Threads(1) + public void longAdderIncrement_lowContention() { + adder.increment(); + } + + @Benchmark + @Threads(Threads.MAX) + public void longAdderIncrement_highContention() { + adder.increment(); + } + + @Benchmark + @Threads(1) + public void accumulatorIncrement_lowContention() { + Accumulator.inc(accumulator, Counter.HITS); + } + + @Benchmark + @Threads(Threads.MAX) + public void accumulatorIncrement_highContention() { + Accumulator.inc(accumulator, Counter.HITS); + } + + @Benchmark + @Threads(1) + public void chmAtomicLongIncrement_lowContention() { + chm.computeIfAbsent("hits", k -> new AtomicLong()).incrementAndGet(); + } + + @Benchmark + @Threads(Threads.MAX) + public void chmAtomicLongIncrement_highContention() { + chm.computeIfAbsent("hits", k -> new AtomicLong()).incrementAndGet(); + } + + @Benchmark + @Threads(1) + public void longAdderSumThenReset_lowContention(Blackhole blackhole) { + adder.increment(); + blackhole.consume(adder.sumThenReset()); + } + + @Benchmark + @Threads(Threads.MAX) + public void longAdderSumThenReset_highContention(Blackhole blackhole) { + adder.increment(); + blackhole.consume(adder.sumThenReset()); + } + + @Benchmark + @Threads(1) + public void accumulatorAccumulateAnd_lowContention(Blackhole blackhole) { + Accumulator.inc(accumulator, Counter.HITS); + blackhole.consume(Accumulator.accumulateAnd(accumulator)); + } + + @Benchmark + @Threads(Threads.MAX) + public void accumulatorAccumulateAnd_highContention(Blackhole blackhole) { + Accumulator.inc(accumulator, Counter.HITS); + blackhole.consume(Accumulator.accumulateAnd(accumulator)); + } +} diff --git a/internal-api/src/main/java/datadog/trace/util/Accumulator.java b/internal-api/src/main/java/datadog/trace/util/Accumulator.java new file mode 100644 index 00000000000..29f6d155e17 --- /dev/null +++ b/internal-api/src/main/java/datadog/trace/util/Accumulator.java @@ -0,0 +1,205 @@ +package datadog.trace.util; + +import datadog.trace.api.function.Strategy; +import datadog.trace.api.function.StrategyConsumer; +import java.util.Arrays; +import java.util.function.Consumer; + +/** + * A striped accumulator primitive: {@code LongAdder}'s write scalability, without {@code + * LongAdder}'s reset hazard. + * + *

{@code LongAdder} is reached for reflexively as "a cheap atomic counter," but it solves a + * narrower problem (genuine many-thread write contention) and its {@code sumThenReset()} is + * documented as not atomic against concurrent updates: it walks its cells summing-then- + * zeroing one at a time, so an increment landing on a cell after it's summed but before it's zeroed + * is silently and permanently lost. {@link #accumulateAnd} closes that window by combining and + * resetting each stripe under the same lock that guards its writers, so no increment can land in + * the gap. + * + *

Each stripe's state is a bare {@code long[]}, not a named-field struct. An {@code enum} + * assigns a name to each position via its ordinal, so name and position are the same declaration + * and cannot drift apart. This also makes {@link #combine} and {@link #reset} generic, branchless, + * fixed-trip-count array loops -- exactly the shape C2's superword optimizer reliably + * auto-vectorizes -- so they are implemented once here instead of once per caller. + * + * 

{@code
+ * enum MyCounters { FOO, BAR }
+ *
+ * long[][] data = Accumulator.create(MyCounters.values());
+ * Accumulator.inc(data, MyCounters.FOO);
+ * Accumulator.update(data, stripe -> {
+ *   Accumulator.inc(stripe, MyCounters.FOO);
+ *   Accumulator.inc(stripe, MyCounters.BAR);
+ * });
+ *
+ * long[] drained = Accumulator.accumulateAnd(data); // combine + reset, atomically per stripe
+ * long foo = drained[MyCounters.FOO.ordinal()];
+ * }
+ * + *

Non-additive counters (max, "ever seen" bitmask, first-occurrence timestamp) are out of scope: + * the per-stripe operation this class provides is {@code +=} via {@link #inc}/{@link #add}, + * combined with {@code +=} in {@link #combine}. A stripeable operator only needs to be associative + * and commutative, not literally addition, but no such escape hatch is wired up here -- add one (a + * caller-supplied {@code LongBinaryOperator} strategy) only when a real candidate needs it. + * + *

This class is a pure namespace over caller-owned {@code long[][]} state, in the same style as + * {@link Hashtable} and {@link FlatHashtable} -- it allocates no container object and is not itself + * a strategy consumer's receiver. + * + *

Not built here (deliberately): a struct-{@code T}-per-stripe fallback, for a subsystem + * whose per-stripe state doesn't fit named {@code long} slots, with mutate/combine/extract as + * {@code @Strategy}-annotated seams -- reach for it only if a real candidate can't be expressed as + * an enum-keyed {@code long[]}. Likewise a raw/embedded tier (caller owns the stripe array + * directly, no owning container) -- a reserve tool for a future {@code dd-trace-core} + * hottest-per-span-path candidate, not needed by the current reporting-cadence migration targets. + * Neither is stubbed out; build it when a real caller needs it (see APMLP-1779). + */ +public final class Accumulator { + private Accumulator() {} + + /** One full cache line of {@code long}s (64 bytes), used to pad each stripe's row. */ + private static final int CACHE_LINE_LONGS = 8; + + /** + * Creates the backing storage for an accumulator over {@code values}: one {@code long[]} row per + * stripe, sized to {@code values.length} plus at least one trailing cache line of padding so + * adjacent stripe rows don't false-share. + * + *

Stripe count is fixed at a power of two derived from {@link Runtime#availableProcessors()}; + * it is not a per-call knob (see {@link #stripeCount()}). + * + * @param values the enum constants naming each counter, e.g. {@code MyCounters.values()} + * @return a new {@code long[stripeCount][paddedWidth]} array, zero-initialized + */ + public static > long[][] create(E[] values) { + int paddedWidth = paddedWidth(values.length); + int stripes = stripeCount(); + long[][] data = new long[stripes][]; + for (int i = 0; i < stripes; i++) { + data[i] = new long[paddedWidth]; + } + return data; + } + + /** + * Increments the counter named by {@code key} in the calling thread's stripe by one. + * + *

Convenience for the common case: selects the calling thread's stripe, takes its lock, and + * increments. To perform several increments under a single held lock, use {@link #update}. + */ + public static > void inc(long[][] data, E key) { + add(data, key, 1L); + } + + /** + * Adds {@code delta} to the counter named by {@code key} in the calling thread's stripe. + * + * @see #inc(long[][], Enum) + */ + public static > void add(long[][] data, E key, long delta) { + add(stripeOf(data), key, delta); + } + + /** + * Increments the counter named by {@code key} in {@code stripe} by one, under {@code stripe}'s + * own lock. + * + *

Intended for use inside an {@link #update} lambda, where {@code stripe} is already the + * calling thread's selected row: {@code synchronized} is reentrant, so calling this here does not + * deadlock or take a second lock. + */ + public static > void inc(long[] stripe, E key) { + add(stripe, key, 1L); + } + + /** + * Adds {@code delta} to the counter named by {@code key} in {@code stripe}, under {@code + * stripe}'s own lock. + * + * @see #inc(long[], Enum) + */ + public static > void add(long[] stripe, E key, long delta) { + synchronized (stripe) { + stripe[key.ordinal()] += delta; + } + } + + /** + * Runs {@code mutator} against the calling thread's stripe under a single held lock -- the escape + * hatch for performing several related updates atomically with respect to a concurrent {@link + * #accumulateAnd}. + * + * @param mutator a strategy over the selected stripe; keep it small and non-capturing so it + * inlines into the lock's critical section + */ + @StrategyConsumer + public static void update(long[][] data, @Strategy Consumer mutator) { + long[] stripe = stripeOf(data); + synchronized (stripe) { + mutator.accept(stripe); + } + } + + /** + * Combines and resets every stripe, returning the sum. Each stripe is locked for exactly as long + * as it takes to fold its values into the result and zero it -- the same lock held by {@link + * #inc}/{@link #add}/{@link #update} -- so no writer can land an increment in the gap between + * summing and zeroing the way {@code LongAdder#sumThenReset()} allows. + * + * @return a new array the same length as one stripe's row, indexed by the enum's {@code + * ordinal()} for the positions actually in use (trailing padding positions are always zero) + */ + public static long[] accumulateAnd(long[][] data) { + long[] acc = new long[data[0].length]; + for (long[] stripe : data) { + synchronized (stripe) { + combine(acc, stripe); + reset(stripe); + } + } + return acc; + } + + /** + * {@code acc[i] += stripe[i]} for every index -- a fixed-trip-count loop C2 can auto-vectorize. + */ + private static void combine(long[] acc, long[] stripe) { + for (int i = 0; i < acc.length; i++) { + acc[i] += stripe[i]; + } + } + + /** Zeroes every position of {@code stripe}, via the JVM-intrinsic {@link Arrays#fill}. */ + private static void reset(long[] stripe) { + Arrays.fill(stripe, 0L); + } + + /** + * The calling thread's stripe: cheap masking, no allocation, no map lookup. + * + *

Multiple threads can map to the same stripe (this is masking, not a bijection); each + * stripe's own lock makes that safe, just not maximally scalable under a hash collision. + */ + private static long[] stripeOf(long[][] data) { + int mask = data.length - 1; + int idx = (int) (Thread.currentThread().getId() & mask); + return data[idx]; + } + + /** + * A fixed, power-of-two stripe count sized to {@link Runtime#availableProcessors()} (rounded down + * to the nearest power of two, minimum one). Not exposed as a per-call override: a mandatory + * sizing knob on every caller fails the "print test" of self-explanatory API design. + */ + private static int stripeCount() { + int cpus = Runtime.getRuntime().availableProcessors(); + return Integer.highestOneBit(Math.max(1, cpus)); + } + + /** Rounds {@code width} up to a whole number of cache lines, plus one full trailing line. */ + private static int paddedWidth(int width) { + int wholeLines = ((width + CACHE_LINE_LONGS - 1) / CACHE_LINE_LONGS) * CACHE_LINE_LONGS; + return wholeLines + CACHE_LINE_LONGS; + } +} diff --git a/internal-api/src/test/java/datadog/trace/util/AccumulatorTest.java b/internal-api/src/test/java/datadog/trace/util/AccumulatorTest.java new file mode 100644 index 00000000000..b842de011e2 --- /dev/null +++ b/internal-api/src/test/java/datadog/trace/util/AccumulatorTest.java @@ -0,0 +1,170 @@ +package datadog.trace.util; + +import static org.junit.jupiter.api.Assertions.assertEquals; +import static org.junit.jupiter.api.Assertions.assertTrue; + +import java.util.concurrent.CountDownLatch; +import java.util.concurrent.ExecutorService; +import java.util.concurrent.Executors; +import java.util.concurrent.TimeUnit; +import java.util.concurrent.atomic.AtomicBoolean; +import org.junit.jupiter.api.Test; + +class AccumulatorTest { + + enum Counters { + FOO, + BAR, + BAZ + } + + @Test + void freshAccumulatorSumsToZero() { + long[][] data = Accumulator.create(Counters.values()); + long[] drained = Accumulator.accumulateAnd(data); + for (Counters c : Counters.values()) { + assertEquals(0L, drained[c.ordinal()]); + } + } + + @Test + void incIncrementsByOne() { + long[][] data = Accumulator.create(Counters.values()); + Accumulator.inc(data, Counters.FOO); + Accumulator.inc(data, Counters.FOO); + Accumulator.inc(data, Counters.BAR); + + long[] drained = Accumulator.accumulateAnd(data); + assertEquals(2L, drained[Counters.FOO.ordinal()]); + assertEquals(1L, drained[Counters.BAR.ordinal()]); + assertEquals(0L, drained[Counters.BAZ.ordinal()]); + } + + @Test + void addAppliesArbitraryDelta() { + long[][] data = Accumulator.create(Counters.values()); + Accumulator.add(data, Counters.BAZ, 41L); + Accumulator.add(data, Counters.BAZ, 1L); + + long[] drained = Accumulator.accumulateAnd(data); + assertEquals(42L, drained[Counters.BAZ.ordinal()]); + } + + @Test + void updateAppliesSeveralOpsUnderOneLock() { + long[][] data = Accumulator.create(Counters.values()); + Accumulator.update( + data, + stripe -> { + Accumulator.inc(stripe, Counters.FOO); + Accumulator.inc(stripe, Counters.FOO); + Accumulator.add(stripe, Counters.BAR, 5L); + }); + + long[] drained = Accumulator.accumulateAnd(data); + assertEquals(2L, drained[Counters.FOO.ordinal()]); + assertEquals(5L, drained[Counters.BAR.ordinal()]); + } + + @Test + void accumulateAndResetsSoASecondDrainIsZero() { + long[][] data = Accumulator.create(Counters.values()); + Accumulator.inc(data, Counters.FOO); + + long[] first = Accumulator.accumulateAnd(data); + assertEquals(1L, first[Counters.FOO.ordinal()]); + + long[] second = Accumulator.accumulateAnd(data); + for (Counters c : Counters.values()) { + assertEquals(0L, second[c.ordinal()]); + } + } + + @Test + void drainedRowsAreAllTheSameLength() { + long[][] data = Accumulator.create(Counters.values()); + long[] drained = Accumulator.accumulateAnd(data); + assertEquals(data[0].length, drained.length); + assertTrue(drained.length >= Counters.values().length); + } + + @Test + void concurrentIncrementsAreNotLost() throws InterruptedException { + long[][] data = Accumulator.create(Counters.values()); + int threadCount = 16; + int incrementsPerThread = 10_000; + + ExecutorService pool = Executors.newFixedThreadPool(threadCount); + CountDownLatch start = new CountDownLatch(1); + CountDownLatch done = new CountDownLatch(threadCount); + try { + for (int t = 0; t < threadCount; t++) { + pool.execute( + () -> { + try { + start.await(); + for (int i = 0; i < incrementsPerThread; i++) { + Accumulator.inc(data, Counters.FOO); + } + } catch (InterruptedException e) { + Thread.currentThread().interrupt(); + } finally { + done.countDown(); + } + }); + } + start.countDown(); + assertTrue(done.await(30, TimeUnit.SECONDS)); + } finally { + pool.shutdown(); + } + + long[] drained = Accumulator.accumulateAnd(data); + assertEquals((long) threadCount * incrementsPerThread, drained[Counters.FOO.ordinal()]); + } + + @Test + void concurrentAccumulateAndDuringWritesNeverExceedsWritten() throws InterruptedException { + long[][] data = Accumulator.create(Counters.values()); + int threadCount = 8; + int incrementsPerThread = 5_000; + + ExecutorService pool = Executors.newFixedThreadPool(threadCount + 1); + CountDownLatch done = new CountDownLatch(threadCount); + AtomicBoolean stop = new AtomicBoolean(false); + long[] runningTotal = {0L}; + + try { + pool.execute( + () -> { + while (!stop.get()) { + long[] drained = Accumulator.accumulateAnd(data); + synchronized (runningTotal) { + runningTotal[0] += drained[Counters.FOO.ordinal()]; + } + } + }); + + for (int t = 0; t < threadCount; t++) { + pool.execute( + () -> { + for (int i = 0; i < incrementsPerThread; i++) { + Accumulator.inc(data, Counters.FOO); + } + done.countDown(); + }); + } + + assertTrue(done.await(30, TimeUnit.SECONDS)); + stop.set(true); + long[] finalDrain = Accumulator.accumulateAnd(data); + synchronized (runningTotal) { + runningTotal[0] += finalDrain[Counters.FOO.ordinal()]; + } + + assertEquals((long) threadCount * incrementsPerThread, runningTotal[0]); + } finally { + pool.shutdown(); + } + } +} From 48e01bb6c3e508126dabb33baf298285ea4753f6 Mon Sep 17 00:00:00 2001 From: Douglas Q Hawkins Date: Mon, 31 Aug 2026 13:20:43 -0400 Subject: [PATCH 02/19] Oversize Accumulator's default stripe count to reduce contention collisions Sizing stripes to exactly availableProcessors() left collisions likely under real contention (birthday-paradox: n(n-1)/(2m) expected colliding pairs), and a collision costs a blocking synchronized wait rather than LongAdder's cheap CAS retry. Doubling the stripe count (floor 4) cuts accumulatorIncrement_highContention from ~0.097 to ~0.040 us/op at the cost of a pricier but far rarer accumulateAnd drain -- the right trade since inc/add run on every call while accumulateAnd runs on a reporting cadence. Benchmark javadoc updated with the re-measured numbers. Co-Authored-By: Claude Sonnet 5 --- .../trace/util/AccumulatorBenchmark.java | 32 +++++++++++++++++++ .../java/datadog/trace/util/Accumulator.java | 19 ++++++++--- 2 files changed, 46 insertions(+), 5 deletions(-) diff --git a/internal-api/src/jmh/java/datadog/trace/util/AccumulatorBenchmark.java b/internal-api/src/jmh/java/datadog/trace/util/AccumulatorBenchmark.java index d7c494ad9df..3a85b47afbd 100644 --- a/internal-api/src/jmh/java/datadog/trace/util/AccumulatorBenchmark.java +++ b/internal-api/src/jmh/java/datadog/trace/util/AccumulatorBenchmark.java @@ -23,6 +23,38 @@ * (heavy contention). The CHM variant allocates its counter under the bucket's bin lock on first * sight of a key -- exactly the pathology {@link Accumulator} exists to avoid -- so its comparison * here is against that allocation-under-lock step, not against a pre-warmed map. + * + *

Contention result to note: at low contention, {@code accumulatorIncrement} is + * essentially free and on par with {@code longAdderIncrement}. At {@code Threads.MAX} (10 threads + * on the measurement machine), oversizing {@link Accumulator}'s stripe count from 8 (one per core) + * to 16 (roughly 2x cores, see {@code stripeCount()}) cut {@code + * accumulatorIncrement_highContention} from ~0.097 us/op to ~0.040 us/op -- fewer threads collide + * on a stripe, so fewer of them pay {@code synchronized}'s blocking wait instead of a cheap + * fast-path lock. It is still roughly 4-5x slower than {@code longAdderIncrement} (a collision-free + * CAS retry beats even an uncontended monitor enter/exit), and {@code accumulateAnd} under + * concurrent writers got correspondingly more expensive (~7.5us to ~15.5us) since draining now + * walks twice as many stripes while writers are actively landing on them. Read {@code + * accumulatorIncrement_highContention} not as "Accumulator beats LongAdder under contention" (it + * doesn't, on this shape) but as the honest cost of the drain-under-lock design that buys atomic + * combine+reset; a caller trading that safety for raw increment throughput should measure their own + * contention level before choosing between them. + * Apple M1 Max, 10 CPUs - JDK 1.8.0_382 (Zulu) - macOS/arm64 - stripeCount() = 16 + * Benchmark Mode Cnt Score Error Units + * AccumulatorBenchmark.longAdderIncrement_lowContention avgt 6 0.007 ± 0.001 us/op + * AccumulatorBenchmark.longAdderIncrement_highContention avgt 6 0.009 ± 0.001 us/op + * AccumulatorBenchmark.accumulatorIncrement_lowContention avgt 6 0.010 ± 0.001 us/op + * AccumulatorBenchmark.accumulatorIncrement_highContention avgt 6 0.040 ± 0.002 us/op + * AccumulatorBenchmark.chmAtomicLongIncrement_lowContention avgt 6 0.010 ± 0.001 us/op + * AccumulatorBenchmark.chmAtomicLongIncrement_highContention avgt 6 0.417 ± 0.543 us/op + * AccumulatorBenchmark.longAdderSumThenReset_lowContention avgt 6 0.012 ± 0.001 us/op + * AccumulatorBenchmark.longAdderSumThenReset_highContention avgt 6 2.433 ± 0.203 us/op + * AccumulatorBenchmark.accumulatorAccumulateAnd_lowContention avgt 6 0.162 ± 0.009 us/op + * AccumulatorBenchmark.accumulatorAccumulateAnd_highContention avgt 6 15.515 ± 4.094 us/op + * + * + *

(This run had some background noise from another session on the measurement machine; the + * {@code lowContention} rows and the {@code highContention} directional deltas are reliable, but + * treat the exact {@code highContention} magnitudes as approximate.) */ @State(Scope.Benchmark) @Warmup(iterations = 1, time = 10) diff --git a/internal-api/src/main/java/datadog/trace/util/Accumulator.java b/internal-api/src/main/java/datadog/trace/util/Accumulator.java index 29f6d155e17..3c1685c2865 100644 --- a/internal-api/src/main/java/datadog/trace/util/Accumulator.java +++ b/internal-api/src/main/java/datadog/trace/util/Accumulator.java @@ -66,8 +66,9 @@ private Accumulator() {} * stripe, sized to {@code values.length} plus at least one trailing cache line of padding so * adjacent stripe rows don't false-share. * - *

Stripe count is fixed at a power of two derived from {@link Runtime#availableProcessors()}; - * it is not a per-call knob (see {@link #stripeCount()}). + *

Stripe count is fixed at a power of two oversized to roughly 2x {@link + * Runtime#availableProcessors()} (minimum 4); it is not a per-call knob (see {@link + * #stripeCount()}). * * @param values the enum constants naming each counter, e.g. {@code MyCounters.values()} * @return a new {@code long[stripeCount][paddedWidth]} array, zero-initialized @@ -188,13 +189,21 @@ private static long[] stripeOf(long[][] data) { } /** - * A fixed, power-of-two stripe count sized to {@link Runtime#availableProcessors()} (rounded down - * to the nearest power of two, minimum one). Not exposed as a per-call override: a mandatory + * A fixed, power-of-two stripe count deliberately oversized to roughly 2x {@link + * Runtime#availableProcessors()} (minimum 4). Not exposed as a per-call override: a mandatory * sizing knob on every caller fails the "print test" of self-explanatory API design. + * + *

Sizing to exactly the core count leaves stripe collisions likely under real contention + * (birthday-paradox math: with {@code n} contending threads and {@code m} stripes, expected + * colliding pairs are {@code n(n-1)/(2m)}) -- and a collision costs a blocking {@code + * synchronized} wait, not a cheap CAS retry. Doubling the stripe count roughly quarters that + * collision count for a one-time, per-accumulator memory cost, at the price of a slightly more + * expensive (but far rarer) {@link #accumulateAnd} drain -- the right trade given {@link #inc}/ + * {@link #add} run on every call while {@link #accumulateAnd} runs on a reporting cadence. */ private static int stripeCount() { int cpus = Runtime.getRuntime().availableProcessors(); - return Integer.highestOneBit(Math.max(1, cpus)); + return Math.max(4, 2 * Integer.highestOneBit(Math.max(1, cpus))); } /** Rounds {@code width} up to a whole number of cache lines, plus one full trailing line. */ From 3629c97ae96029e5988aeebed9d7ce01f41a5202 Mon Sep 17 00:00:00 2001 From: Douglas Q Hawkins Date: Mon, 31 Aug 2026 13:25:58 -0400 Subject: [PATCH 03/19] Add JOL footprint test: Accumulator vs one LongAdder per counter Accumulator's realistic alternative isn't a single LongAdder but one per counter (there's no multi-counter LongAdder). Fresh instances make LongAdder look ~15x lighter, but that's an artifact of never having grown a Cell[] table under contention. Forcing real concurrent writes shows the opposite: 4 LongAdders under contention (17,560 bytes) end up over 7x heavier than Accumulator's fixed footprint (2,384 bytes), which is paid once at creation and doesn't grow with more contention or more counters, while each contended LongAdder keeps paying independently. Co-Authored-By: Claude Sonnet 5 --- .../trace/util/AccumulatorFootprintTest.java | 147 ++++++++++++++++++ 1 file changed, 147 insertions(+) create mode 100644 internal-api/src/test/java/datadog/trace/util/AccumulatorFootprintTest.java diff --git a/internal-api/src/test/java/datadog/trace/util/AccumulatorFootprintTest.java b/internal-api/src/test/java/datadog/trace/util/AccumulatorFootprintTest.java new file mode 100644 index 00000000000..2371419b2d8 --- /dev/null +++ b/internal-api/src/test/java/datadog/trace/util/AccumulatorFootprintTest.java @@ -0,0 +1,147 @@ +package datadog.trace.util; + +import static org.junit.jupiter.api.Assertions.assertTrue; +import static org.junit.jupiter.api.Assumptions.assumeFalse; + +import datadog.environment.JavaVirtualMachine; +import java.util.concurrent.CountDownLatch; +import java.util.concurrent.ExecutorService; +import java.util.concurrent.Executors; +import java.util.concurrent.TimeUnit; +import java.util.concurrent.atomic.LongAdder; +import org.junit.jupiter.api.BeforeAll; +import org.junit.jupiter.api.Test; +import org.openjdk.jol.info.GraphLayout; + +/** + * Retained-footprint comparison (JOL) for {@link Accumulator} vs the alternative it actually + * displaces: one {@code LongAdder} per counter (there is no multi-counter {@code LongAdder} -- a + * caller wanting N additive counters allocates N of them, one field each, as {@code OtlpTelemetry} + * and {@code PayloadDispatcherImpl} do today). + * + *

A freshly constructed {@code LongAdder} is nearly free -- it holds no {@code Cell[]} table + * until contention forces one -- so comparing fresh instances understates its real cost and + * flatters {@code LongAdder}. {@link Accumulator} pays its full striped array up front, at + * creation, sized for {@link Runtime#availableProcessors()} regardless of whether contention ever + * materializes. The realistic comparison is therefore not fresh-vs-fresh but contended-vs-fresh: + * what each actually costs once the counters they represent are hit by real concurrent writers, as + * they are on the telemetry paths this class targets. + * + *

Measured on a 10-CPU machine (JDK 1.8.0_382 Zulu), 4 counters, {@code + * Accumulator.stripeCount()} = 16: + * + * 

{@code
+ * fresh:      4 LongAdders =    160 bytes, Accumulator = 2384 bytes
+ * contended:  4 LongAdders =  17560 bytes, Accumulator = 2384 bytes
+ * }
+ * + * Finding: fresh, {@code LongAdder} looks ~15x lighter -- but that's an artifact of never having + * been written to concurrently. Once real contention forces each {@code LongAdder}'s {@code Cell[]} + * table to grow (each {@code Cell} is {@code @Contended}-padded against false sharing, the same + * problem {@link Accumulator}'s own padding solves), the four {@code LongAdder}s alone end up over + * 7x heavier than {@code Accumulator}'s entire fixed footprint -- and {@code Accumulator} does not + * grow further as more contention arrives within its existing stripe count, while every additional + * concurrently-written {@code LongAdder} keeps paying this cost independently. {@code + * Accumulator}'s up-front cost is the more predictable one: fixed at creation, independent of + * runtime contention, and shared (one striped array) across however many counters the caller's enum + * declares, rather than paid per counter. + */ +class AccumulatorFootprintTest { + + enum Counters { + REQUESTS, + ERRORS, + RETRIES, + BYTES_SENT + } + + @BeforeAll + static void assumeNotJ9Jvm() { + // JOL's GraphLayout relies on HotSpot-specific Unsafe internals and throws + // IllegalStateException on J9-based JVMs (IBM/Semeru) -- same guard as + // StringIndexFootprintTest / ScopeAndContinuationLayoutTest. + assumeFalse(JavaVirtualMachine.isJ9()); + } + + static long bytes(Object root) { + return GraphLayout.parseInstance(root).totalSize(); + } + + static LongAdder[] freshAdders() { + LongAdder[] adders = new LongAdder[Counters.values().length]; + for (int i = 0; i < adders.length; i++) { + adders[i] = new LongAdder(); + } + return adders; + } + + @Test + void freshFootprint() { + LongAdder[] adders = freshAdders(); + long[][] accumulator = Accumulator.create(Counters.values()); + + long adderBytes = bytes((Object) adders); + long accumulatorBytes = bytes(accumulator); + + System.out.printf( + "fresh: %d LongAdders = %6d bytes, Accumulator = %6d bytes%n", + adders.length, adderBytes, accumulatorBytes); + } + + /** + * Drives real multi-threaded contention against a fresh set of {@code LongAdder}s to force their + * {@code Cell[]} tables to grow, then compares against {@link Accumulator}'s fixed footprint -- + * the realistic comparison, since production callers write to these counters concurrently rather + * than leaving them untouched. + * + *

Cell-table growth is driven by JVM-internal CAS-collision detection, not something this test + * controls directly, so the exact grown size can vary by run/JVM; the one invariant asserted is + * monotonic growth (a contended footprint can only be at least the fresh one). + */ + @Test + void contendedFootprint() throws InterruptedException { + LongAdder[] adders = freshAdders(); + long freshAdderBytes = bytes((Object) adders); + + int threads = Math.max(4, Runtime.getRuntime().availableProcessors()); + ExecutorService pool = Executors.newFixedThreadPool(threads); + CountDownLatch start = new CountDownLatch(1); + CountDownLatch done = new CountDownLatch(threads); + try { + for (int t = 0; t < threads; t++) { + pool.execute( + () -> { + try { + start.await(); + long deadline = System.nanoTime() + TimeUnit.SECONDS.toNanos(2); + while (System.nanoTime() < deadline) { + for (LongAdder adder : adders) { + adder.increment(); + } + } + } catch (InterruptedException e) { + Thread.currentThread().interrupt(); + } finally { + done.countDown(); + } + }); + } + start.countDown(); + assertTrue(done.await(30, TimeUnit.SECONDS)); + } finally { + pool.shutdown(); + } + + long contendedAdderBytes = bytes((Object) adders); + long[][] accumulator = Accumulator.create(Counters.values()); + long accumulatorBytes = bytes(accumulator); + + System.out.printf( + "contended: %d LongAdders = %6d bytes, Accumulator = %6d bytes%n", + adders.length, contendedAdderBytes, accumulatorBytes); + + assertTrue( + contendedAdderBytes >= freshAdderBytes, + "contended LongAdder footprint should never shrink below the fresh footprint"); + } +} From fc2f55cd8131c064a8697036515f430bcb18ce9f Mon Sep 17 00:00:00 2001 From: Douglas Q Hawkins Date: Mon, 31 Aug 2026 14:50:51 -0400 Subject: [PATCH 04/19] Benchmark Accumulator against a per-counter-locked LongAdder alternative Tests the hypothesis that a LongAdder-based helper which actually closes the same sumThenReset() reset hazard (one LongAdder per counter, a per-counter lock guarding both increment and drain) would cost about the same as Accumulator. It doesn't -- it's a clean trade-off inversion, not a wash: Accumulator's thread-sharded stripes win ~10x on the increment path, while the per-counter design wins ~24x on drain, but only because this benchmark has a single counter (its drain cost scales with counter count; Accumulator's is fixed at stripe count). Documented as a data point, not adopted -- both designs close the hazard, and the difference is negligible next to real request/span work either way. Co-Authored-By: Claude Sonnet 5 --- .../trace/util/AccumulatorBenchmark.java | 79 +++++++++++++++++++ 1 file changed, 79 insertions(+) diff --git a/internal-api/src/jmh/java/datadog/trace/util/AccumulatorBenchmark.java b/internal-api/src/jmh/java/datadog/trace/util/AccumulatorBenchmark.java index 3a85b47afbd..d603a78d123 100644 --- a/internal-api/src/jmh/java/datadog/trace/util/AccumulatorBenchmark.java +++ b/internal-api/src/jmh/java/datadog/trace/util/AccumulatorBenchmark.java @@ -55,6 +55,30 @@ *

(This run had some background noise from another session on the measurement machine; the * {@code lowContention} rows and the {@code highContention} directional deltas are reliable, but * treat the exact {@code highContention} magnitudes as approximate.) + * + *

{@code longAdderGroup*}: is a "just fix it with LongAdder" helper actually cheaper? + * {@code groupInc}/{@code groupAccumulateAnd} are the natural correct fix using {@code LongAdder} + * as the payload: one {@code LongAdder} per counter, with a per-counter lock guarding both + * the increment and the drain (locking only the drain does nothing -- {@code sumThenReset()}'s + * internal race is against the {@code LongAdder}'s own CAS-based {@code add()}, not against any + * lock a caller takes). This closes the same reset hazard as {@link Accumulator}, but stripes by + * counter instead of by thread. + * AccumulatorBenchmark.accumulatorIncrement_highContention avgt 6 0.029 ± 0.051 us/op + * AccumulatorBenchmark.accumulatorAccumulateAnd_highContention avgt 6 13.431 ± 5.876 us/op + * AccumulatorBenchmark.longAdderGroupIncrement_highContention avgt 6 0.294 ± 0.088 us/op + * AccumulatorBenchmark.longAdderGroupAccumulateAnd_highContention avgt 6 0.549 ± 0.291 us/op + * Not "similar cost" -- a clean trade-off inversion. With this benchmark's single counter, + * {@code longAdderGroup}'s per-counter lock collapses to one lock for every thread (no thread-based + * distribution at all), so it loses badly on the write path: ~10x worse than {@code Accumulator}'s + * thread-sharded stripes. But its drain only has that one lock to acquire, so it wins big there: + * ~24x better than {@code Accumulator}, which always walks all 16 stripes on every drain regardless + * of counter count. That asymmetry is the whole story: {@code longAdderGroup}'s drain cost scales + * with number of counters (more counters -> more locks to drain), while {@code + * Accumulator}'s drain cost is fixed at stripe count, independent of counter count. Which design + * actually wins for a given caller depends on that caller's counter cardinality and whether its + * write traffic concentrates on a few hot counters (favors thread-sharding) or spreads across many + * (favors counter-sharding) -- not measured here, and worth checking against the real migration + * targets before treating either number as the general answer. */ @State(Scope.Benchmark) @Warmup(iterations = 1, time = 10) @@ -71,6 +95,35 @@ enum Counter { private final LongAdder adder = new LongAdder(); private final long[][] accumulator = Accumulator.create(Counter.values()); private final ConcurrentHashMap chm = new ConcurrentHashMap<>(); + private final LongAdder[] longAdderGroup = {new LongAdder()}; + + /** + * The natural "just use LongAdder" fix for the reset hazard: one {@code LongAdder} per counter, + * with a per-counter lock guarding both the increment and the drain -- external locking around + * only the drain does nothing, since {@code sumThenReset()}'s internal race is against the {@code + * LongAdder}'s own CAS-based {@code add()}, not against any lock a caller takes. This is the fair + * comparison point: it closes the same hazard {@link Accumulator} does, but stripes by + * counter (one lock per enum constant) instead of by thread (one lock per + * stripe, shared by all counters) -- so N threads hammering the *same* counter contend on one + * lock regardless of core count, with no thread-bucket distribution at all. + */ + private static void groupInc(LongAdder[] group, int ordinal) { + LongAdder counter = group[ordinal]; + synchronized (counter) { + counter.add(1L); + } + } + + private static long[] groupAccumulateAnd(LongAdder[] group) { + long[] acc = new long[group.length]; + for (int i = 0; i < group.length; i++) { + LongAdder counter = group[i]; + synchronized (counter) { + acc[i] = counter.sumThenReset(); + } + } + return acc; + } @Benchmark @Threads(1) @@ -135,4 +188,30 @@ public void accumulatorAccumulateAnd_highContention(Blackhole blackhole) { Accumulator.inc(accumulator, Counter.HITS); blackhole.consume(Accumulator.accumulateAnd(accumulator)); } + + @Benchmark + @Threads(1) + public void longAdderGroupIncrement_lowContention() { + groupInc(longAdderGroup, Counter.HITS.ordinal()); + } + + @Benchmark + @Threads(Threads.MAX) + public void longAdderGroupIncrement_highContention() { + groupInc(longAdderGroup, Counter.HITS.ordinal()); + } + + @Benchmark + @Threads(1) + public void longAdderGroupAccumulateAnd_lowContention(Blackhole blackhole) { + groupInc(longAdderGroup, Counter.HITS.ordinal()); + blackhole.consume(groupAccumulateAnd(longAdderGroup)); + } + + @Benchmark + @Threads(Threads.MAX) + public void longAdderGroupAccumulateAnd_highContention(Blackhole blackhole) { + groupInc(longAdderGroup, Counter.HITS.ordinal()); + blackhole.consume(groupAccumulateAnd(longAdderGroup)); + } } From 89d980c0f829c6df51d22131aa35b760c529fdb4 Mon Sep 17 00:00:00 2001 From: Douglas Q Hawkins Date: Mon, 31 Aug 2026 17:01:14 -0400 Subject: [PATCH 05/19] Address review comments on Accumulator Rename accumulateAnd to accumulateAndReset, use ThreadSupport.threadId() instead of the deprecated Thread.getId(), add @GuardedBy annotations on the stripe-locked helpers, add @ParametersAreNonnullByDefault, and trim the javadoc (drop the Hashtable/FlatHashtable mention, the not-yet-built non-additive-counter escape hatch, and the C2-specific vectorization detail; shorten the LongAdder comparison). Co-Authored-By: Claude Sonnet 5 --- .../trace/util/AccumulatorBenchmark.java | 6 +-- .../java/datadog/trace/util/Accumulator.java | 53 ++++++++----------- .../datadog/trace/util/AccumulatorTest.java | 20 +++---- 3 files changed, 34 insertions(+), 45 deletions(-) diff --git a/internal-api/src/jmh/java/datadog/trace/util/AccumulatorBenchmark.java b/internal-api/src/jmh/java/datadog/trace/util/AccumulatorBenchmark.java index d603a78d123..85fca0618b4 100644 --- a/internal-api/src/jmh/java/datadog/trace/util/AccumulatorBenchmark.java +++ b/internal-api/src/jmh/java/datadog/trace/util/AccumulatorBenchmark.java @@ -31,7 +31,7 @@ * accumulatorIncrement_highContention} from ~0.097 us/op to ~0.040 us/op -- fewer threads collide * on a stripe, so fewer of them pay {@code synchronized}'s blocking wait instead of a cheap * fast-path lock. It is still roughly 4-5x slower than {@code longAdderIncrement} (a collision-free - * CAS retry beats even an uncontended monitor enter/exit), and {@code accumulateAnd} under + * CAS retry beats even an uncontended monitor enter/exit), and {@code accumulateAndReset} under * concurrent writers got correspondingly more expensive (~7.5us to ~15.5us) since draining now * walks twice as many stripes while writers are actively landing on them. Read {@code * accumulatorIncrement_highContention} not as "Accumulator beats LongAdder under contention" (it @@ -179,14 +179,14 @@ public void longAdderSumThenReset_highContention(Blackhole blackhole) { @Threads(1) public void accumulatorAccumulateAnd_lowContention(Blackhole blackhole) { Accumulator.inc(accumulator, Counter.HITS); - blackhole.consume(Accumulator.accumulateAnd(accumulator)); + blackhole.consume(Accumulator.accumulateAndReset(accumulator)); } @Benchmark @Threads(Threads.MAX) public void accumulatorAccumulateAnd_highContention(Blackhole blackhole) { Accumulator.inc(accumulator, Counter.HITS); - blackhole.consume(Accumulator.accumulateAnd(accumulator)); + blackhole.consume(Accumulator.accumulateAndReset(accumulator)); } @Benchmark diff --git a/internal-api/src/main/java/datadog/trace/util/Accumulator.java b/internal-api/src/main/java/datadog/trace/util/Accumulator.java index 3c1685c2865..9b0eb4a6448 100644 --- a/internal-api/src/main/java/datadog/trace/util/Accumulator.java +++ b/internal-api/src/main/java/datadog/trace/util/Accumulator.java @@ -1,27 +1,27 @@ package datadog.trace.util; +import datadog.environment.ThreadSupport; import datadog.trace.api.function.Strategy; import datadog.trace.api.function.StrategyConsumer; import java.util.Arrays; import java.util.function.Consumer; +import javax.annotation.ParametersAreNonnullByDefault; +import javax.annotation.concurrent.GuardedBy; /** * A striped accumulator primitive: {@code LongAdder}'s write scalability, without {@code * LongAdder}'s reset hazard. * - *

{@code LongAdder} is reached for reflexively as "a cheap atomic counter," but it solves a - * narrower problem (genuine many-thread write contention) and its {@code sumThenReset()} is - * documented as not atomic against concurrent updates: it walks its cells summing-then- - * zeroing one at a time, so an increment landing on a cell after it's summed but before it's zeroed - * is silently and permanently lost. {@link #accumulateAnd} closes that window by combining and - * resetting each stripe under the same lock that guards its writers, so no increment can land in - * the gap. + *

{@code LongAdder#sumThenReset()} is documented as not atomic against concurrent + * updates: an increment landing on a cell after it's summed but before it's zeroed is silently and + * permanently lost. {@link #accumulateAndReset} closes that window by combining and resetting each + * stripe under the same lock that guards its writers. * *

Each stripe's state is a bare {@code long[]}, not a named-field struct. An {@code enum} * assigns a name to each position via its ordinal, so name and position are the same declaration * and cannot drift apart. This also makes {@link #combine} and {@link #reset} generic, branchless, - * fixed-trip-count array loops -- exactly the shape C2's superword optimizer reliably - * auto-vectorizes -- so they are implemented once here instead of once per caller. + * fixed-trip-count array loops -- the shape designed to take advantage of SIMD / vector operations + * on modern hardware -- so they are implemented once here instead of once per caller. * * 

{@code
  * enum MyCounters { FOO, BAR }
@@ -33,28 +33,14 @@
  *   Accumulator.inc(stripe, MyCounters.BAR);
  * });
  *
- * long[] drained = Accumulator.accumulateAnd(data); // combine + reset, atomically per stripe
+ * long[] drained = Accumulator.accumulateAndReset(data); // combine + reset, atomically per stripe
  * long foo = drained[MyCounters.FOO.ordinal()];
  * }
* - *

Non-additive counters (max, "ever seen" bitmask, first-occurrence timestamp) are out of scope: - * the per-stripe operation this class provides is {@code +=} via {@link #inc}/{@link #add}, - * combined with {@code +=} in {@link #combine}. A stripeable operator only needs to be associative - * and commutative, not literally addition, but no such escape hatch is wired up here -- add one (a - * caller-supplied {@code LongBinaryOperator} strategy) only when a real candidate needs it. - * - *

This class is a pure namespace over caller-owned {@code long[][]} state, in the same style as - * {@link Hashtable} and {@link FlatHashtable} -- it allocates no container object and is not itself - * a strategy consumer's receiver. - * - *

Not built here (deliberately): a struct-{@code T}-per-stripe fallback, for a subsystem - * whose per-stripe state doesn't fit named {@code long} slots, with mutate/combine/extract as - * {@code @Strategy}-annotated seams -- reach for it only if a real candidate can't be expressed as - * an enum-keyed {@code long[]}. Likewise a raw/embedded tier (caller owns the stripe array - * directly, no owning container) -- a reserve tool for a future {@code dd-trace-core} - * hottest-per-span-path candidate, not needed by the current reporting-cadence migration targets. - * Neither is stubbed out; build it when a real caller needs it (see APMLP-1779). + *

This class is a pure namespace over caller-owned {@code long[][]} state -- it allocates no + * container object and is not itself a strategy consumer's receiver. */ +@ParametersAreNonnullByDefault public final class Accumulator { private Accumulator() {} @@ -129,7 +115,7 @@ public static > void add(long[] stripe, E key, long delta) { /** * Runs {@code mutator} against the calling thread's stripe under a single held lock -- the escape * hatch for performing several related updates atomically with respect to a concurrent {@link - * #accumulateAnd}. + * #accumulateAndReset}. * * @param mutator a strategy over the selected stripe; keep it small and non-capturing so it * inlines into the lock's critical section @@ -151,7 +137,7 @@ public static void update(long[][] data, @Strategy Consumer mutator) { * @return a new array the same length as one stripe's row, indexed by the enum's {@code * ordinal()} for the positions actually in use (trailing padding positions are always zero) */ - public static long[] accumulateAnd(long[][] data) { + public static long[] accumulateAndReset(long[][] data) { long[] acc = new long[data[0].length]; for (long[] stripe : data) { synchronized (stripe) { @@ -165,6 +151,7 @@ public static long[] accumulateAnd(long[][] data) { /** * {@code acc[i] += stripe[i]} for every index -- a fixed-trip-count loop C2 can auto-vectorize. */ + @GuardedBy("stripe") private static void combine(long[] acc, long[] stripe) { for (int i = 0; i < acc.length; i++) { acc[i] += stripe[i]; @@ -172,6 +159,7 @@ private static void combine(long[] acc, long[] stripe) { } /** Zeroes every position of {@code stripe}, via the JVM-intrinsic {@link Arrays#fill}. */ + @GuardedBy("stripe") private static void reset(long[] stripe) { Arrays.fill(stripe, 0L); } @@ -184,7 +172,7 @@ private static void reset(long[] stripe) { */ private static long[] stripeOf(long[][] data) { int mask = data.length - 1; - int idx = (int) (Thread.currentThread().getId() & mask); + int idx = (int) (ThreadSupport.threadId() & mask); return data[idx]; } @@ -198,8 +186,9 @@ private static long[] stripeOf(long[][] data) { * colliding pairs are {@code n(n-1)/(2m)}) -- and a collision costs a blocking {@code * synchronized} wait, not a cheap CAS retry. Doubling the stripe count roughly quarters that * collision count for a one-time, per-accumulator memory cost, at the price of a slightly more - * expensive (but far rarer) {@link #accumulateAnd} drain -- the right trade given {@link #inc}/ - * {@link #add} run on every call while {@link #accumulateAnd} runs on a reporting cadence. + * expensive (but far rarer) {@link #accumulateAndReset} drain -- the right trade given {@link + * #inc}/ {@link #add} run on every call while {@link #accumulateAndReset} runs on a reporting + * cadence. */ private static int stripeCount() { int cpus = Runtime.getRuntime().availableProcessors(); diff --git a/internal-api/src/test/java/datadog/trace/util/AccumulatorTest.java b/internal-api/src/test/java/datadog/trace/util/AccumulatorTest.java index b842de011e2..2ddff79024f 100644 --- a/internal-api/src/test/java/datadog/trace/util/AccumulatorTest.java +++ b/internal-api/src/test/java/datadog/trace/util/AccumulatorTest.java @@ -21,7 +21,7 @@ enum Counters { @Test void freshAccumulatorSumsToZero() { long[][] data = Accumulator.create(Counters.values()); - long[] drained = Accumulator.accumulateAnd(data); + long[] drained = Accumulator.accumulateAndReset(data); for (Counters c : Counters.values()) { assertEquals(0L, drained[c.ordinal()]); } @@ -34,7 +34,7 @@ void incIncrementsByOne() { Accumulator.inc(data, Counters.FOO); Accumulator.inc(data, Counters.BAR); - long[] drained = Accumulator.accumulateAnd(data); + long[] drained = Accumulator.accumulateAndReset(data); assertEquals(2L, drained[Counters.FOO.ordinal()]); assertEquals(1L, drained[Counters.BAR.ordinal()]); assertEquals(0L, drained[Counters.BAZ.ordinal()]); @@ -46,7 +46,7 @@ void addAppliesArbitraryDelta() { Accumulator.add(data, Counters.BAZ, 41L); Accumulator.add(data, Counters.BAZ, 1L); - long[] drained = Accumulator.accumulateAnd(data); + long[] drained = Accumulator.accumulateAndReset(data); assertEquals(42L, drained[Counters.BAZ.ordinal()]); } @@ -61,7 +61,7 @@ void updateAppliesSeveralOpsUnderOneLock() { Accumulator.add(stripe, Counters.BAR, 5L); }); - long[] drained = Accumulator.accumulateAnd(data); + long[] drained = Accumulator.accumulateAndReset(data); assertEquals(2L, drained[Counters.FOO.ordinal()]); assertEquals(5L, drained[Counters.BAR.ordinal()]); } @@ -71,10 +71,10 @@ void accumulateAndResetsSoASecondDrainIsZero() { long[][] data = Accumulator.create(Counters.values()); Accumulator.inc(data, Counters.FOO); - long[] first = Accumulator.accumulateAnd(data); + long[] first = Accumulator.accumulateAndReset(data); assertEquals(1L, first[Counters.FOO.ordinal()]); - long[] second = Accumulator.accumulateAnd(data); + long[] second = Accumulator.accumulateAndReset(data); for (Counters c : Counters.values()) { assertEquals(0L, second[c.ordinal()]); } @@ -83,7 +83,7 @@ void accumulateAndResetsSoASecondDrainIsZero() { @Test void drainedRowsAreAllTheSameLength() { long[][] data = Accumulator.create(Counters.values()); - long[] drained = Accumulator.accumulateAnd(data); + long[] drained = Accumulator.accumulateAndReset(data); assertEquals(data[0].length, drained.length); assertTrue(drained.length >= Counters.values().length); } @@ -119,7 +119,7 @@ void concurrentIncrementsAreNotLost() throws InterruptedException { pool.shutdown(); } - long[] drained = Accumulator.accumulateAnd(data); + long[] drained = Accumulator.accumulateAndReset(data); assertEquals((long) threadCount * incrementsPerThread, drained[Counters.FOO.ordinal()]); } @@ -138,7 +138,7 @@ void concurrentAccumulateAndDuringWritesNeverExceedsWritten() throws Interrupted pool.execute( () -> { while (!stop.get()) { - long[] drained = Accumulator.accumulateAnd(data); + long[] drained = Accumulator.accumulateAndReset(data); synchronized (runningTotal) { runningTotal[0] += drained[Counters.FOO.ordinal()]; } @@ -157,7 +157,7 @@ void concurrentAccumulateAndDuringWritesNeverExceedsWritten() throws Interrupted assertTrue(done.await(30, TimeUnit.SECONDS)); stop.set(true); - long[] finalDrain = Accumulator.accumulateAnd(data); + long[] finalDrain = Accumulator.accumulateAndReset(data); synchronized (runningTotal) { runningTotal[0] += finalDrain[Counters.FOO.ordinal()]; } From 7107b5b5c3f25d540f5cd26b27208ca43d4c78d9 Mon Sep 17 00:00:00 2001 From: Douglas Q Hawkins Date: Tue, 1 Sep 2026 13:42:44 -0400 Subject: [PATCH 06/19] Split Accumulator into a typed wrapper and a nested EmbeddingSupport The original static, allocation-free Accumulator API let a caller index its long[][] with a different enum than the one it was created for -- compiles, but silently reads/writes the wrong slot. Move that raw API into a nested EmbeddingSupport namespace, and add a top-level Accumulator that owns its storage and binds inc/add/update/ accumulateAndReset to one enum at construction, mirroring StringIndex's own EmbeddingSupport split in this package. Also fix the stripe-count javadoc: with n contending threads and m stripes, doubling m halves the expected number of colliding pairs (n(n-1)/(2m)), not quarters it. Co-Authored-By: Claude Sonnet 5 --- .../java/datadog/trace/util/Accumulator.java | 368 +++++++++++------- 1 file changed, 223 insertions(+), 145 deletions(-) diff --git a/internal-api/src/main/java/datadog/trace/util/Accumulator.java b/internal-api/src/main/java/datadog/trace/util/Accumulator.java index 9b0eb4a6448..34e8b6e84ee 100644 --- a/internal-api/src/main/java/datadog/trace/util/Accumulator.java +++ b/internal-api/src/main/java/datadog/trace/util/Accumulator.java @@ -9,195 +9,273 @@ import javax.annotation.concurrent.GuardedBy; /** - * A striped accumulator primitive: {@code LongAdder}'s write scalability, without {@code - * LongAdder}'s reset hazard. - * - *

{@code LongAdder#sumThenReset()} is documented as not atomic against concurrent - * updates: an increment landing on a cell after it's summed but before it's zeroed is silently and - * permanently lost. {@link #accumulateAndReset} closes that window by combining and resetting each - * stripe under the same lock that guards its writers. - * - *

Each stripe's state is a bare {@code long[]}, not a named-field struct. An {@code enum} - * assigns a name to each position via its ordinal, so name and position are the same declaration - * and cannot drift apart. This also makes {@link #combine} and {@link #reset} generic, branchless, - * fixed-trip-count array loops -- the shape designed to take advantage of SIMD / vector operations - * on modern hardware -- so they are implemented once here instead of once per caller. + * A typed, instance-owning wrapper over {@link EmbeddingSupport}: ties an enum's type to its + * backing {@code long[][]} at construction, so {@link #inc}/{@link #add} can't be called with a key + * from a different enum than the one this accumulator was {@link #of created} for. Costs one + * field-load indirection per call versus calling {@link EmbeddingSupport} directly -- the same + * trade {@code StringIndex} makes over its own nested {@code EmbeddingSupport}. * * 

{@code
  * enum MyCounters { FOO, BAR }
  *
- * long[][] data = Accumulator.create(MyCounters.values());
- * Accumulator.inc(data, MyCounters.FOO);
- * Accumulator.update(data, stripe -> {
- *   Accumulator.inc(stripe, MyCounters.FOO);
- *   Accumulator.inc(stripe, MyCounters.BAR);
+ * Accumulator counters = Accumulator.of(MyCounters.values());
+ * counters.inc(MyCounters.FOO);
+ * counters.update(stripe -> {
+ *   Accumulator.EmbeddingSupport.inc(stripe, MyCounters.FOO);
+ *   Accumulator.EmbeddingSupport.inc(stripe, MyCounters.BAR);
  * });
  *
- * long[] drained = Accumulator.accumulateAndReset(data); // combine + reset, atomically per stripe
+ * long[] drained = counters.accumulateAndReset(); // combine + reset, atomically per stripe
  * long foo = drained[MyCounters.FOO.ordinal()];
  * }
* - *

This class is a pure namespace over caller-owned {@code long[][]} state -- it allocates no - * container object and is not itself a strategy consumer's receiver. + * @see EmbeddingSupport */ -@ParametersAreNonnullByDefault -public final class Accumulator { - private Accumulator() {} +public final class Accumulator> { + private final long[][] data; - /** One full cache line of {@code long}s (64 bytes), used to pad each stripe's row. */ - private static final int CACHE_LINE_LONGS = 8; + private Accumulator(long[][] data) { + this.data = data; + } /** - * Creates the backing storage for an accumulator over {@code values}: one {@code long[]} row per - * stripe, sized to {@code values.length} plus at least one trailing cache line of padding so - * adjacent stripe rows don't false-share. - * - *

Stripe count is fixed at a power of two oversized to roughly 2x {@link - * Runtime#availableProcessors()} (minimum 4); it is not a per-call knob (see {@link - * #stripeCount()}). - * * @param values the enum constants naming each counter, e.g. {@code MyCounters.values()} - * @return a new {@code long[stripeCount][paddedWidth]} array, zero-initialized */ - public static > long[][] create(E[] values) { - int paddedWidth = paddedWidth(values.length); - int stripes = stripeCount(); - long[][] data = new long[stripes][]; - for (int i = 0; i < stripes; i++) { - data[i] = new long[paddedWidth]; - } - return data; + public static > Accumulator of(E[] values) { + return new Accumulator<>(EmbeddingSupport.create(values)); } - /** - * Increments the counter named by {@code key} in the calling thread's stripe by one. - * - *

Convenience for the common case: selects the calling thread's stripe, takes its lock, and - * increments. To perform several increments under a single held lock, use {@link #update}. - */ - public static > void inc(long[][] data, E key) { - add(data, key, 1L); + /** Increments the counter named by {@code key} in the calling thread's stripe by one. */ + public void inc(E key) { + EmbeddingSupport.inc(data, key); } - /** - * Adds {@code delta} to the counter named by {@code key} in the calling thread's stripe. - * - * @see #inc(long[][], Enum) - */ - public static > void add(long[][] data, E key, long delta) { - add(stripeOf(data), key, delta); + /** Adds {@code delta} to the counter named by {@code key} in the calling thread's stripe. */ + public void add(E key, long delta) { + EmbeddingSupport.add(data, key, delta); } /** - * Increments the counter named by {@code key} in {@code stripe} by one, under {@code stripe}'s - * own lock. + * Runs {@code mutator} against the calling thread's stripe under a single held lock -- the escape + * hatch for performing several related updates atomically with respect to a concurrent {@link + * #accumulateAndReset}. * - *

Intended for use inside an {@link #update} lambda, where {@code stripe} is already the - * calling thread's selected row: {@code synchronized} is reentrant, so calling this here does not - * deadlock or take a second lock. + * @param mutator a strategy over the selected stripe; keep it small and non-capturing so it + * inlines into the lock's critical section */ - public static > void inc(long[] stripe, E key) { - add(stripe, key, 1L); + @StrategyConsumer + public void update(@Strategy Consumer mutator) { + EmbeddingSupport.update(data, mutator); } /** - * Adds {@code delta} to the counter named by {@code key} in {@code stripe}, under {@code - * stripe}'s own lock. + * Combines and resets every stripe, returning the sum. * - * @see #inc(long[], Enum) + * @return a new array indexed by the enum's {@code ordinal()} + * @see EmbeddingSupport#accumulateAndReset */ - public static > void add(long[] stripe, E key, long delta) { - synchronized (stripe) { - stripe[key.ordinal()] += delta; - } + public long[] accumulateAndReset() { + return EmbeddingSupport.accumulateAndReset(data); } /** - * Runs {@code mutator} against the calling thread's stripe under a single held lock -- the escape - * hatch for performing several related updates atomically with respect to a concurrent {@link - * #accumulateAndReset}. + * The static, raw-array tier of the striped accumulator primitive: {@code LongAdder}'s write + * scalability, without {@code LongAdder}'s reset hazard. * - * @param mutator a strategy over the selected stripe; keep it small and non-capturing so it - * inlines into the lock's critical section + *

{@code LongAdder#sumThenReset()} is documented as not atomic against concurrent + * updates: an increment landing on a cell after it's summed but before it's zeroed is silently + * and permanently lost. {@link #accumulateAndReset} closes that window by combining and resetting + * each stripe under the same lock that guards its writers. + * + *

Each stripe's state is a bare {@code long[]}, not a named-field struct. An {@code enum} + * assigns a name to each position via its ordinal, so name and position are the same declaration + * and cannot drift apart. This also makes {@link #combine} and {@link #reset} generic, + * branchless, fixed-trip-count array loops -- the shape designed to take advantage of SIMD / + * vector operations on modern hardware -- so they are implemented once here instead of once per + * caller. + * + *

This is a pure namespace over caller-owned {@code long[][]} state -- it allocates no + * container object and is not itself a strategy consumer's receiver. That means {@code create}'s + * type parameter is not bound to the one later {@code inc}/{@code add} calls infer: nothing stops + * a caller from indexing the same {@code long[][]} with a different enum than the one it was + * {@link #create}d for, which silently reads/writes the wrong slot rather than failing to + * compile. Prefer the owning {@link Accumulator} instance, which closes that hole for one + * field-load indirection per call; reach for this class directly only when that indirection is + * worth removing. + * + * 

{@code
+   * enum MyCounters { FOO, BAR }
+   *
+   * long[][] data = Accumulator.EmbeddingSupport.create(MyCounters.values());
+   * Accumulator.EmbeddingSupport.inc(data, MyCounters.FOO);
+   * Accumulator.EmbeddingSupport.update(data, stripe -> {
+   *   Accumulator.EmbeddingSupport.inc(stripe, MyCounters.FOO);
+   *   Accumulator.EmbeddingSupport.inc(stripe, MyCounters.BAR);
+   * });
+   *
+   * long[] drained = Accumulator.EmbeddingSupport.accumulateAndReset(data); // per stripe
+   * long foo = drained[MyCounters.FOO.ordinal()];
+   * }
*/ - @StrategyConsumer - public static void update(long[][] data, @Strategy Consumer mutator) { - long[] stripe = stripeOf(data); - synchronized (stripe) { - mutator.accept(stripe); + @ParametersAreNonnullByDefault + public static final class EmbeddingSupport { + private EmbeddingSupport() {} + + /** One full cache line of {@code long}s (64 bytes), used to pad each stripe's row. */ + private static final int CACHE_LINE_LONGS = 8; + + /** + * Creates the backing storage for an accumulator over {@code values}: one {@code long[]} row + * per stripe, sized to {@code values.length} plus at least one trailing cache line of padding + * so adjacent stripe rows don't false-share. + * + *

Stripe count is fixed at a power of two oversized to roughly 2x {@link + * Runtime#availableProcessors()} (minimum 4); it is not a per-call knob (see {@link + * #stripeCount()}). + * + * @param values the enum constants naming each counter, e.g. {@code MyCounters.values()} + * @return a new {@code long[stripeCount][paddedWidth]} array, zero-initialized + */ + public static > long[][] create(E[] values) { + int paddedWidth = paddedWidth(values.length); + int stripes = stripeCount(); + long[][] data = new long[stripes][]; + for (int i = 0; i < stripes; i++) { + data[i] = new long[paddedWidth]; + } + return data; } - } - /** - * Combines and resets every stripe, returning the sum. Each stripe is locked for exactly as long - * as it takes to fold its values into the result and zero it -- the same lock held by {@link - * #inc}/{@link #add}/{@link #update} -- so no writer can land an increment in the gap between - * summing and zeroing the way {@code LongAdder#sumThenReset()} allows. - * - * @return a new array the same length as one stripe's row, indexed by the enum's {@code - * ordinal()} for the positions actually in use (trailing padding positions are always zero) - */ - public static long[] accumulateAndReset(long[][] data) { - long[] acc = new long[data[0].length]; - for (long[] stripe : data) { + /** + * Increments the counter named by {@code key} in the calling thread's stripe by one. + * + *

Convenience for the common case: selects the calling thread's stripe, takes its lock, and + * increments. To perform several increments under a single held lock, use {@link #update}. + */ + public static > void inc(long[][] data, E key) { + add(data, key, 1L); + } + + /** + * Adds {@code delta} to the counter named by {@code key} in the calling thread's stripe. + * + * @see #inc(long[][], Enum) + */ + public static > void add(long[][] data, E key, long delta) { + add(stripeOf(data), key, delta); + } + + /** + * Increments the counter named by {@code key} in {@code stripe} by one, under {@code stripe}'s + * own lock. + * + *

Intended for use inside an {@link #update} lambda, where {@code stripe} is already the + * calling thread's selected row: {@code synchronized} is reentrant, so calling this here does + * not deadlock or take a second lock. + */ + public static > void inc(long[] stripe, E key) { + add(stripe, key, 1L); + } + + /** + * Adds {@code delta} to the counter named by {@code key} in {@code stripe}, under {@code + * stripe}'s own lock. + * + * @see #inc(long[], Enum) + */ + public static > void add(long[] stripe, E key, long delta) { synchronized (stripe) { - combine(acc, stripe); - reset(stripe); + stripe[key.ordinal()] += delta; } } - return acc; - } - /** - * {@code acc[i] += stripe[i]} for every index -- a fixed-trip-count loop C2 can auto-vectorize. - */ - @GuardedBy("stripe") - private static void combine(long[] acc, long[] stripe) { - for (int i = 0; i < acc.length; i++) { - acc[i] += stripe[i]; + /** + * Runs {@code mutator} against the calling thread's stripe under a single held lock -- the + * escape hatch for performing several related updates atomically with respect to a concurrent + * {@link #accumulateAndReset}. + * + * @param mutator a strategy over the selected stripe; keep it small and non-capturing so it + * inlines into the lock's critical section + */ + @StrategyConsumer + public static void update(long[][] data, @Strategy Consumer mutator) { + long[] stripe = stripeOf(data); + synchronized (stripe) { + mutator.accept(stripe); + } } - } - /** Zeroes every position of {@code stripe}, via the JVM-intrinsic {@link Arrays#fill}. */ - @GuardedBy("stripe") - private static void reset(long[] stripe) { - Arrays.fill(stripe, 0L); - } + /** + * Combines and resets every stripe, returning the sum. Each stripe is locked for exactly as + * long as it takes to fold its values into the result and zero it -- the same lock held by + * {@link #inc}/{@link #add}/{@link #update} -- so no writer can land an increment in the gap + * between summing and zeroing the way {@code LongAdder#sumThenReset()} allows. + * + * @return a new array the same length as one stripe's row, indexed by the enum's {@code + * ordinal()} for the positions actually in use (trailing padding positions are always zero) + */ + public static long[] accumulateAndReset(long[][] data) { + long[] acc = new long[data[0].length]; + for (long[] stripe : data) { + synchronized (stripe) { + combine(acc, stripe); + reset(stripe); + } + } + return acc; + } - /** - * The calling thread's stripe: cheap masking, no allocation, no map lookup. - * - *

Multiple threads can map to the same stripe (this is masking, not a bijection); each - * stripe's own lock makes that safe, just not maximally scalable under a hash collision. - */ - private static long[] stripeOf(long[][] data) { - int mask = data.length - 1; - int idx = (int) (ThreadSupport.threadId() & mask); - return data[idx]; - } + /** + * {@code acc[i] += stripe[i]} for every index -- a fixed-trip-count loop C2 can auto-vectorize. + */ + @GuardedBy("stripe") + private static void combine(long[] acc, long[] stripe) { + for (int i = 0; i < acc.length; i++) { + acc[i] += stripe[i]; + } + } - /** - * A fixed, power-of-two stripe count deliberately oversized to roughly 2x {@link - * Runtime#availableProcessors()} (minimum 4). Not exposed as a per-call override: a mandatory - * sizing knob on every caller fails the "print test" of self-explanatory API design. - * - *

Sizing to exactly the core count leaves stripe collisions likely under real contention - * (birthday-paradox math: with {@code n} contending threads and {@code m} stripes, expected - * colliding pairs are {@code n(n-1)/(2m)}) -- and a collision costs a blocking {@code - * synchronized} wait, not a cheap CAS retry. Doubling the stripe count roughly quarters that - * collision count for a one-time, per-accumulator memory cost, at the price of a slightly more - * expensive (but far rarer) {@link #accumulateAndReset} drain -- the right trade given {@link - * #inc}/ {@link #add} run on every call while {@link #accumulateAndReset} runs on a reporting - * cadence. - */ - private static int stripeCount() { - int cpus = Runtime.getRuntime().availableProcessors(); - return Math.max(4, 2 * Integer.highestOneBit(Math.max(1, cpus))); - } + /** Zeroes every position of {@code stripe}, via the JVM-intrinsic {@link Arrays#fill}. */ + @GuardedBy("stripe") + private static void reset(long[] stripe) { + Arrays.fill(stripe, 0L); + } + + /** + * The calling thread's stripe: cheap masking, no allocation, no map lookup. + * + *

Multiple threads can map to the same stripe (this is masking, not a bijection); each + * stripe's own lock makes that safe, just not maximally scalable under a hash collision. + */ + private static long[] stripeOf(long[][] data) { + int mask = data.length - 1; + int idx = (int) (ThreadSupport.threadId() & mask); + return data[idx]; + } + + /** + * A fixed, power-of-two stripe count deliberately oversized to roughly 2x {@link + * Runtime#availableProcessors()} (minimum 4). Not exposed as a per-call override: a mandatory + * sizing knob on every caller fails the "print test" of self-explanatory API design. + * + *

Sizing to exactly the core count leaves stripe collisions likely under real contention + * (birthday-paradox math: with {@code n} contending threads and {@code m} stripes, expected + * colliding pairs are {@code n(n-1)/(2m)}) -- and a collision costs a blocking {@code + * synchronized} wait, not a cheap CAS retry. Doubling the stripe count roughly halves that + * collision count for a one-time, per-accumulator memory cost, at the price of a slightly more + * expensive (but far rarer) {@link #accumulateAndReset} drain -- the right trade given {@link + * #inc}/ {@link #add} run on every call while {@link #accumulateAndReset} runs on a reporting + * cadence. + */ + private static int stripeCount() { + int cpus = Runtime.getRuntime().availableProcessors(); + return Math.max(4, 2 * Integer.highestOneBit(Math.max(1, cpus))); + } - /** Rounds {@code width} up to a whole number of cache lines, plus one full trailing line. */ - private static int paddedWidth(int width) { - int wholeLines = ((width + CACHE_LINE_LONGS - 1) / CACHE_LINE_LONGS) * CACHE_LINE_LONGS; - return wholeLines + CACHE_LINE_LONGS; + /** Rounds {@code width} up to a whole number of cache lines, plus one full trailing line. */ + private static int paddedWidth(int width) { + int wholeLines = ((width + CACHE_LINE_LONGS - 1) / CACHE_LINE_LONGS) * CACHE_LINE_LONGS; + return wholeLines + CACHE_LINE_LONGS; + } } } From 175154ed5e4d30bfc4b8fc3ccb3874047350b721 Mon Sep 17 00:00:00 2001 From: Douglas Q Hawkins Date: Tue, 1 Sep 2026 13:42:55 -0400 Subject: [PATCH 07/19] Update Accumulator tests/benchmark for the EmbeddingSupport split - Update call sites to Accumulator.EmbeddingSupport.*, and add a test covering the new typed Accumulator wrapper. - Fix a race in concurrentAccumulateAndDuringWritesNeverExceedsWritten: join the background drainer via Future.get() before the final drain and assertion, instead of racing it. - Assert accumulatorBytes < contendedAdderBytes in contendedFootprint -- the actual claim the test exists to back up, not just that the LongAdder side didn't shrink. - Correct the CHM benchmark's javadoc: the benchmark-scoped map means only the first warmup invocation allocates under the bin lock; every sampled op hits an already-warmed computeIfAbsent lookup. - Add a @Group-based accumulatorMixed-write/accumulatorMixed-drain pair modeling "many writers, one rare drainer," alongside the existing @Threads(MAX) benchmark kept as a documented worst-case upper bound. Co-Authored-By: Claude Sonnet 5 --- .../trace/util/AccumulatorBenchmark.java | 59 ++++++++-- .../trace/util/AccumulatorFootprintTest.java | 9 +- .../datadog/trace/util/AccumulatorTest.java | 102 +++++++++++------- 3 files changed, 118 insertions(+), 52 deletions(-) diff --git a/internal-api/src/jmh/java/datadog/trace/util/AccumulatorBenchmark.java b/internal-api/src/jmh/java/datadog/trace/util/AccumulatorBenchmark.java index 85fca0618b4..5b6299a7eae 100644 --- a/internal-api/src/jmh/java/datadog/trace/util/AccumulatorBenchmark.java +++ b/internal-api/src/jmh/java/datadog/trace/util/AccumulatorBenchmark.java @@ -8,6 +8,8 @@ import org.openjdk.jmh.annotations.Benchmark; import org.openjdk.jmh.annotations.BenchmarkMode; import org.openjdk.jmh.annotations.Fork; +import org.openjdk.jmh.annotations.Group; +import org.openjdk.jmh.annotations.GroupThreads; import org.openjdk.jmh.annotations.Measurement; import org.openjdk.jmh.annotations.Mode; import org.openjdk.jmh.annotations.OutputTimeUnit; @@ -20,9 +22,15 @@ /** * {@link Accumulator} vs {@link LongAdder} vs the {@code ConcurrentHashMap.computeIfAbsent(key, k * -> new AtomicLong())} anti-pattern, at one thread (no contention) and at {@link Threads#MAX} - * (heavy contention). The CHM variant allocates its counter under the bucket's bin lock on first - * sight of a key -- exactly the pathology {@link Accumulator} exists to avoid -- so its comparison - * here is against that allocation-under-lock step, not against a pre-warmed map. + * (heavy contention). The CHM variant allocates its counter under the bucket's bin lock the first + * time its one constant key is seen -- exactly the pathology {@link Accumulator} exists to avoid -- + * but since the map is a {@code @State(Scope.Benchmark)} field shared across the whole run, that + * allocation happens exactly once; every sampled op after it hits the warmed, already-present fast + * path. So this measures steady-state {@code computeIfAbsent} lookup overhead on an + * already-populated map, not the one-time allocation-under-lock cost -- still a useful number (a + * fixed, small key set that's allocated once and hit for the life of the process, as {@code + * WafMetricCollector}-style CHM counters are, spends nearly all its time in this same warmed path), + * just not the pathology the name of this benchmark might suggest. * *

Contention result to note: at low contention, {@code accumulatorIncrement} is * essentially free and on par with {@code longAdderIncrement}. At {@code Threads.MAX} (10 threads @@ -93,7 +101,7 @@ enum Counter { } private final LongAdder adder = new LongAdder(); - private final long[][] accumulator = Accumulator.create(Counter.values()); + private final long[][] accumulator = Accumulator.EmbeddingSupport.create(Counter.values()); private final ConcurrentHashMap chm = new ConcurrentHashMap<>(); private final LongAdder[] longAdderGroup = {new LongAdder()}; @@ -140,13 +148,13 @@ public void longAdderIncrement_highContention() { @Benchmark @Threads(1) public void accumulatorIncrement_lowContention() { - Accumulator.inc(accumulator, Counter.HITS); + Accumulator.EmbeddingSupport.inc(accumulator, Counter.HITS); } @Benchmark @Threads(Threads.MAX) public void accumulatorIncrement_highContention() { - Accumulator.inc(accumulator, Counter.HITS); + Accumulator.EmbeddingSupport.inc(accumulator, Counter.HITS); } @Benchmark @@ -178,15 +186,46 @@ public void longAdderSumThenReset_highContention(Blackhole blackhole) { @Benchmark @Threads(1) public void accumulatorAccumulateAnd_lowContention(Blackhole blackhole) { - Accumulator.inc(accumulator, Counter.HITS); - blackhole.consume(Accumulator.accumulateAndReset(accumulator)); + Accumulator.EmbeddingSupport.inc(accumulator, Counter.HITS); + blackhole.consume(Accumulator.EmbeddingSupport.accumulateAndReset(accumulator)); } + /** + * A deliberately pessimistic topology: every thread both writes and drains on every op, so {@code + * Threads.MAX} threads are all draining concurrently. Real callers don't do this -- see {@code + * accumulatorMixed-write}/{@code accumulatorMixed-drain} below for the "many writers, one rare + * drainer" shape this class actually targets. Kept as the worst-case upper bound: no production + * topology should be more contended on {@link Accumulator.EmbeddingSupport#accumulateAndReset} + * than this. + */ @Benchmark @Threads(Threads.MAX) public void accumulatorAccumulateAnd_highContention(Blackhole blackhole) { - Accumulator.inc(accumulator, Counter.HITS); - blackhole.consume(Accumulator.accumulateAndReset(accumulator)); + Accumulator.EmbeddingSupport.inc(accumulator, Counter.HITS); + blackhole.consume(Accumulator.EmbeddingSupport.accumulateAndReset(accumulator)); + } + + /** + * The realistic counterpart to {@code accumulatorAccumulateAnd_highContention}: many writer + * threads incrementing, and a single dedicated thread polling {@link + * Accumulator.EmbeddingSupport#accumulateAndReset} -- not every thread doing both on every op. + * {@code accumulatorMixed-write} measures increment cost while a drain is actively contending for + * stripe locks; {@code accumulatorMixed-drain} measures the drain's own cost under that same live + * write pressure. The 4:1 writer:drainer ratio is illustrative of "many writers, rare drain," not + * tuned to a specific core count. + */ + @Benchmark + @Group("accumulatorMixed") + @GroupThreads(4) + public void accumulatorMixed_write() { + Accumulator.EmbeddingSupport.inc(accumulator, Counter.HITS); + } + + @Benchmark + @Group("accumulatorMixed") + @GroupThreads(1) + public void accumulatorMixed_drain(Blackhole blackhole) { + blackhole.consume(Accumulator.EmbeddingSupport.accumulateAndReset(accumulator)); } @Benchmark diff --git a/internal-api/src/test/java/datadog/trace/util/AccumulatorFootprintTest.java b/internal-api/src/test/java/datadog/trace/util/AccumulatorFootprintTest.java index 2371419b2d8..8b47237512f 100644 --- a/internal-api/src/test/java/datadog/trace/util/AccumulatorFootprintTest.java +++ b/internal-api/src/test/java/datadog/trace/util/AccumulatorFootprintTest.java @@ -28,7 +28,7 @@ * they are on the telemetry paths this class targets. * *

Measured on a 10-CPU machine (JDK 1.8.0_382 Zulu), 4 counters, {@code - * Accumulator.stripeCount()} = 16: + * Accumulator.EmbeddingSupport.stripeCount()} = 16: * * 

{@code
  * fresh:      4 LongAdders =    160 bytes, Accumulator = 2384 bytes
@@ -78,7 +78,7 @@ static LongAdder[] freshAdders() {
   @Test
   void freshFootprint() {
     LongAdder[] adders = freshAdders();
-    long[][] accumulator = Accumulator.create(Counters.values());
+    long[][] accumulator = Accumulator.EmbeddingSupport.create(Counters.values());
 
     long adderBytes = bytes((Object) adders);
     long accumulatorBytes = bytes(accumulator);
@@ -133,7 +133,7 @@ void contendedFootprint() throws InterruptedException {
     }
 
     long contendedAdderBytes = bytes((Object) adders);
-    long[][] accumulator = Accumulator.create(Counters.values());
+    long[][] accumulator = Accumulator.EmbeddingSupport.create(Counters.values());
     long accumulatorBytes = bytes(accumulator);
 
     System.out.printf(
@@ -143,5 +143,8 @@ void contendedFootprint() throws InterruptedException {
     assertTrue(
         contendedAdderBytes >= freshAdderBytes,
         "contended LongAdder footprint should never shrink below the fresh footprint");
+    assertTrue(
+        accumulatorBytes < contendedAdderBytes,
+        "Accumulator's fixed footprint should be smaller than N contended LongAdders");
   }
 }
diff --git a/internal-api/src/test/java/datadog/trace/util/AccumulatorTest.java b/internal-api/src/test/java/datadog/trace/util/AccumulatorTest.java
index 2ddff79024f..ef7c2d5706e 100644
--- a/internal-api/src/test/java/datadog/trace/util/AccumulatorTest.java
+++ b/internal-api/src/test/java/datadog/trace/util/AccumulatorTest.java
@@ -4,9 +4,12 @@
 import static org.junit.jupiter.api.Assertions.assertTrue;
 
 import java.util.concurrent.CountDownLatch;
+import java.util.concurrent.ExecutionException;
 import java.util.concurrent.ExecutorService;
 import java.util.concurrent.Executors;
+import java.util.concurrent.Future;
 import java.util.concurrent.TimeUnit;
+import java.util.concurrent.TimeoutException;
 import java.util.concurrent.atomic.AtomicBoolean;
 import org.junit.jupiter.api.Test;
 
@@ -20,8 +23,8 @@ enum Counters {
 
   @Test
   void freshAccumulatorSumsToZero() {
-    long[][] data = Accumulator.create(Counters.values());
-    long[] drained = Accumulator.accumulateAndReset(data);
+    long[][] data = Accumulator.EmbeddingSupport.create(Counters.values());
+    long[] drained = Accumulator.EmbeddingSupport.accumulateAndReset(data);
     for (Counters c : Counters.values()) {
       assertEquals(0L, drained[c.ordinal()]);
     }
@@ -29,12 +32,12 @@ void freshAccumulatorSumsToZero() {
 
   @Test
   void incIncrementsByOne() {
-    long[][] data = Accumulator.create(Counters.values());
-    Accumulator.inc(data, Counters.FOO);
-    Accumulator.inc(data, Counters.FOO);
-    Accumulator.inc(data, Counters.BAR);
+    long[][] data = Accumulator.EmbeddingSupport.create(Counters.values());
+    Accumulator.EmbeddingSupport.inc(data, Counters.FOO);
+    Accumulator.EmbeddingSupport.inc(data, Counters.FOO);
+    Accumulator.EmbeddingSupport.inc(data, Counters.BAR);
 
-    long[] drained = Accumulator.accumulateAndReset(data);
+    long[] drained = Accumulator.EmbeddingSupport.accumulateAndReset(data);
     assertEquals(2L, drained[Counters.FOO.ordinal()]);
     assertEquals(1L, drained[Counters.BAR.ordinal()]);
     assertEquals(0L, drained[Counters.BAZ.ordinal()]);
@@ -42,39 +45,39 @@ void incIncrementsByOne() {
 
   @Test
   void addAppliesArbitraryDelta() {
-    long[][] data = Accumulator.create(Counters.values());
-    Accumulator.add(data, Counters.BAZ, 41L);
-    Accumulator.add(data, Counters.BAZ, 1L);
+    long[][] data = Accumulator.EmbeddingSupport.create(Counters.values());
+    Accumulator.EmbeddingSupport.add(data, Counters.BAZ, 41L);
+    Accumulator.EmbeddingSupport.add(data, Counters.BAZ, 1L);
 
-    long[] drained = Accumulator.accumulateAndReset(data);
+    long[] drained = Accumulator.EmbeddingSupport.accumulateAndReset(data);
     assertEquals(42L, drained[Counters.BAZ.ordinal()]);
   }
 
   @Test
   void updateAppliesSeveralOpsUnderOneLock() {
-    long[][] data = Accumulator.create(Counters.values());
-    Accumulator.update(
+    long[][] data = Accumulator.EmbeddingSupport.create(Counters.values());
+    Accumulator.EmbeddingSupport.update(
         data,
         stripe -> {
-          Accumulator.inc(stripe, Counters.FOO);
-          Accumulator.inc(stripe, Counters.FOO);
-          Accumulator.add(stripe, Counters.BAR, 5L);
+          Accumulator.EmbeddingSupport.inc(stripe, Counters.FOO);
+          Accumulator.EmbeddingSupport.inc(stripe, Counters.FOO);
+          Accumulator.EmbeddingSupport.add(stripe, Counters.BAR, 5L);
         });
 
-    long[] drained = Accumulator.accumulateAndReset(data);
+    long[] drained = Accumulator.EmbeddingSupport.accumulateAndReset(data);
     assertEquals(2L, drained[Counters.FOO.ordinal()]);
     assertEquals(5L, drained[Counters.BAR.ordinal()]);
   }
 
   @Test
   void accumulateAndResetsSoASecondDrainIsZero() {
-    long[][] data = Accumulator.create(Counters.values());
-    Accumulator.inc(data, Counters.FOO);
+    long[][] data = Accumulator.EmbeddingSupport.create(Counters.values());
+    Accumulator.EmbeddingSupport.inc(data, Counters.FOO);
 
-    long[] first = Accumulator.accumulateAndReset(data);
+    long[] first = Accumulator.EmbeddingSupport.accumulateAndReset(data);
     assertEquals(1L, first[Counters.FOO.ordinal()]);
 
-    long[] second = Accumulator.accumulateAndReset(data);
+    long[] second = Accumulator.EmbeddingSupport.accumulateAndReset(data);
     for (Counters c : Counters.values()) {
       assertEquals(0L, second[c.ordinal()]);
     }
@@ -82,15 +85,15 @@ void accumulateAndResetsSoASecondDrainIsZero() {
 
   @Test
   void drainedRowsAreAllTheSameLength() {
-    long[][] data = Accumulator.create(Counters.values());
-    long[] drained = Accumulator.accumulateAndReset(data);
+    long[][] data = Accumulator.EmbeddingSupport.create(Counters.values());
+    long[] drained = Accumulator.EmbeddingSupport.accumulateAndReset(data);
     assertEquals(data[0].length, drained.length);
     assertTrue(drained.length >= Counters.values().length);
   }
 
   @Test
   void concurrentIncrementsAreNotLost() throws InterruptedException {
-    long[][] data = Accumulator.create(Counters.values());
+    long[][] data = Accumulator.EmbeddingSupport.create(Counters.values());
     int threadCount = 16;
     int incrementsPerThread = 10_000;
 
@@ -104,7 +107,7 @@ void concurrentIncrementsAreNotLost() throws InterruptedException {
               try {
                 start.await();
                 for (int i = 0; i < incrementsPerThread; i++) {
-                  Accumulator.inc(data, Counters.FOO);
+                  Accumulator.EmbeddingSupport.inc(data, Counters.FOO);
                 }
               } catch (InterruptedException e) {
                 Thread.currentThread().interrupt();
@@ -119,13 +122,14 @@ void concurrentIncrementsAreNotLost() throws InterruptedException {
       pool.shutdown();
     }
 
-    long[] drained = Accumulator.accumulateAndReset(data);
+    long[] drained = Accumulator.EmbeddingSupport.accumulateAndReset(data);
     assertEquals((long) threadCount * incrementsPerThread, drained[Counters.FOO.ordinal()]);
   }
 
   @Test
-  void concurrentAccumulateAndDuringWritesNeverExceedsWritten() throws InterruptedException {
-    long[][] data = Accumulator.create(Counters.values());
+  void concurrentAccumulateAndDuringWritesNeverExceedsWritten()
+      throws InterruptedException, ExecutionException, TimeoutException {
+    long[][] data = Accumulator.EmbeddingSupport.create(Counters.values());
     int threadCount = 8;
     int incrementsPerThread = 5_000;
 
@@ -135,21 +139,22 @@ void concurrentAccumulateAndDuringWritesNeverExceedsWritten() throws Interrupted
     long[] runningTotal = {0L};
 
     try {
-      pool.execute(
-          () -> {
-            while (!stop.get()) {
-              long[] drained = Accumulator.accumulateAndReset(data);
-              synchronized (runningTotal) {
-                runningTotal[0] += drained[Counters.FOO.ordinal()];
-              }
-            }
-          });
+      Future drainer =
+          pool.submit(
+              () -> {
+                while (!stop.get()) {
+                  long[] drained = Accumulator.EmbeddingSupport.accumulateAndReset(data);
+                  synchronized (runningTotal) {
+                    runningTotal[0] += drained[Counters.FOO.ordinal()];
+                  }
+                }
+              });
 
       for (int t = 0; t < threadCount; t++) {
         pool.execute(
             () -> {
               for (int i = 0; i < incrementsPerThread; i++) {
-                Accumulator.inc(data, Counters.FOO);
+                Accumulator.EmbeddingSupport.inc(data, Counters.FOO);
               }
               done.countDown();
             });
@@ -157,7 +162,9 @@ void concurrentAccumulateAndDuringWritesNeverExceedsWritten() throws Interrupted
 
       assertTrue(done.await(30, TimeUnit.SECONDS));
       stop.set(true);
-      long[] finalDrain = Accumulator.accumulateAndReset(data);
+      drainer.get(30, TimeUnit.SECONDS);
+
+      long[] finalDrain = Accumulator.EmbeddingSupport.accumulateAndReset(data);
       synchronized (runningTotal) {
         runningTotal[0] += finalDrain[Counters.FOO.ordinal()];
       }
@@ -167,4 +174,21 @@ void concurrentAccumulateAndDuringWritesNeverExceedsWritten() throws Interrupted
       pool.shutdown();
     }
   }
+
+  @Test
+  void typedWrapperDelegatesToEmbeddingSupport() {
+    Accumulator counters = Accumulator.of(Counters.values());
+    counters.inc(Counters.FOO);
+    counters.inc(Counters.FOO);
+    counters.add(Counters.BAR, 5L);
+    counters.update(
+        stripe -> {
+          Accumulator.EmbeddingSupport.inc(stripe, Counters.BAZ);
+        });
+
+    long[] drained = counters.accumulateAndReset();
+    assertEquals(2L, drained[Counters.FOO.ordinal()]);
+    assertEquals(5L, drained[Counters.BAR.ordinal()]);
+    assertEquals(1L, drained[Counters.BAZ.ordinal()]);
+  }
 }

From c9f7fc2f643986cbfb901306adcc714babf1ba3f Mon Sep 17 00:00:00 2001
From: Douglas Q Hawkins 
Date: Tue, 1 Sep 2026 14:27:25 -0400
Subject: [PATCH 08/19] Wrap Accumulator's update/accumulateAndReset in typed
 Stripe/Counts views

Restores enum-ordinal type checking inside update()'s critical section
and on accumulateAndReset()'s drained result, closing the last gap left
by the EmbeddingSupport split. Stripe is constructed fresh under the
held lock and is expected to be scalar-replaced by escape analysis for
well-behaved (small, non-capturing, non-escaping) mutators; Counts
wraps the already-drained array and is a real but infrequent
per-drain allocation.

Co-Authored-By: Claude Sonnet 5 
---
 .../java/datadog/trace/util/Accumulator.java  | 85 ++++++++++++++++---
 .../datadog/trace/util/AccumulatorTest.java   | 13 ++-
 2 files changed, 76 insertions(+), 22 deletions(-)

diff --git a/internal-api/src/main/java/datadog/trace/util/Accumulator.java b/internal-api/src/main/java/datadog/trace/util/Accumulator.java
index 34e8b6e84ee..f6c181a2dcf 100644
--- a/internal-api/src/main/java/datadog/trace/util/Accumulator.java
+++ b/internal-api/src/main/java/datadog/trace/util/Accumulator.java
@@ -21,12 +21,12 @@
  * Accumulator counters = Accumulator.of(MyCounters.values());
  * counters.inc(MyCounters.FOO);
  * counters.update(stripe -> {
- *   Accumulator.EmbeddingSupport.inc(stripe, MyCounters.FOO);
- *   Accumulator.EmbeddingSupport.inc(stripe, MyCounters.BAR);
+ *   stripe.inc(MyCounters.FOO);
+ *   stripe.inc(MyCounters.BAR);
  * });
  *
- * long[] drained = counters.accumulateAndReset(); // combine + reset, atomically per stripe
- * long foo = drained[MyCounters.FOO.ordinal()];
+ * Accumulator.Counts drained = counters.accumulateAndReset(); // atomically per stripe
+ * long foo = drained.get(MyCounters.FOO);
  * }
* * @see EmbeddingSupport @@ -56,26 +56,83 @@ public void add(E key, long delta) { } /** - * Runs {@code mutator} against the calling thread's stripe under a single held lock -- the escape - * hatch for performing several related updates atomically with respect to a concurrent {@link - * #accumulateAndReset}. + * Runs {@code mutator} against a typed view of the calling thread's stripe under a single held + * lock -- the escape hatch for performing several related updates atomically with respect to a + * concurrent {@link #accumulateAndReset}. * * @param mutator a strategy over the selected stripe; keep it small and non-capturing so it - * inlines into the lock's critical section + * inlines into the lock's critical section, and don't let the {@link Stripe} escape it (store + * it, return it, hand it to another thread) -- see {@link Stripe} */ @StrategyConsumer - public void update(@Strategy Consumer mutator) { - EmbeddingSupport.update(data, mutator); + public void update(@Strategy Consumer> mutator) { + long[] stripe = EmbeddingSupport.stripeOf(data); + synchronized (stripe) { + mutator.accept(new Stripe<>(stripe)); + } } /** - * Combines and resets every stripe, returning the sum. + * A typed view over one stripe, handed to an {@link #update} strategy: the same enum-ordinal type + * checking {@link Accumulator} provides at the top level, applied inside the critical section + * too. * - * @return a new array indexed by the enum's {@code ordinal()} + *

Constructed fresh under the held lock on every {@link #update} call. A well-behaved {@link + * Strategy} mutator -- small, non-capturing, and never storing or returning this object -- lets + * escape analysis prove it doesn't escape the inlined call and scalar-replace it, so no + * allocation survives to run time. Break those rules (capture it in a field, return it, hand it + * to another thread) and it degrades to a real, per-call allocation instead of a compile-time + * fiction with no correctness difference either way -- just a cost one. + */ + public static final class Stripe> { + private final long[] stripe; + + private Stripe(long[] stripe) { + this.stripe = stripe; + } + + /** Increments the counter named by {@code key} in this stripe by one. */ + public void inc(E key) { + EmbeddingSupport.inc(stripe, key); + } + + /** Adds {@code delta} to the counter named by {@code key} in this stripe. */ + public void add(E key, long delta) { + EmbeddingSupport.add(stripe, key, delta); + } + } + + /** + * Combines and resets every stripe, returning the sum as a typed view. + * + * @return the sum, keyed by the enum's {@code ordinal()} * @see EmbeddingSupport#accumulateAndReset */ - public long[] accumulateAndReset() { - return EmbeddingSupport.accumulateAndReset(data); + public Counts accumulateAndReset() { + return new Counts<>(EmbeddingSupport.accumulateAndReset(data)); + } + + /** + * A typed view over a drained {@code long[]}, returned by {@link #accumulateAndReset}: the same + * enum-ordinal type checking {@link Accumulator} provides on writes, applied to the read side + * too. + * + *

Unlike {@link Stripe}, this is expected to escape -- the caller holds and reads it after the + * call returns -- so it's a real, per-drain allocation, not a scalar-replacement candidate. + * That's fine: {@link #accumulateAndReset} runs on a reporting cadence, not per {@link + * #inc}/{@link #add} call. + */ + public static final class Counts> { + private final long[] counts; + + private Counts(long[] counts) { + this.counts = counts; + } + + /** The counter named by {@code key}. */ + public long get(E key) { + return counts[key.ordinal()]; + } } /** diff --git a/internal-api/src/test/java/datadog/trace/util/AccumulatorTest.java b/internal-api/src/test/java/datadog/trace/util/AccumulatorTest.java index ef7c2d5706e..a93665cb238 100644 --- a/internal-api/src/test/java/datadog/trace/util/AccumulatorTest.java +++ b/internal-api/src/test/java/datadog/trace/util/AccumulatorTest.java @@ -181,14 +181,11 @@ void typedWrapperDelegatesToEmbeddingSupport() { counters.inc(Counters.FOO); counters.inc(Counters.FOO); counters.add(Counters.BAR, 5L); - counters.update( - stripe -> { - Accumulator.EmbeddingSupport.inc(stripe, Counters.BAZ); - }); + counters.update(stripe -> stripe.inc(Counters.BAZ)); - long[] drained = counters.accumulateAndReset(); - assertEquals(2L, drained[Counters.FOO.ordinal()]); - assertEquals(5L, drained[Counters.BAR.ordinal()]); - assertEquals(1L, drained[Counters.BAZ.ordinal()]); + Accumulator.Counts drained = counters.accumulateAndReset(); + assertEquals(2L, drained.get(Counters.FOO)); + assertEquals(5L, drained.get(Counters.BAR)); + assertEquals(1L, drained.get(Counters.BAZ)); } } From a8e74481cd6f300f9df9316b90d0337e4566f51f Mon Sep 17 00:00:00 2001 From: Douglas Q Hawkins Date: Tue, 1 Sep 2026 14:36:33 -0400 Subject: [PATCH 09/19] Add benchmarks for Accumulator's typed API alongside raw EmbeddingSupport Pairs typedIncrement/typedUpdate/typedAccumulateAndReset against their EmbeddingSupport equivalents so the wrapper's cost is directly visible: inc/update should track the raw calls closely (Stripe is designed to scalar-replace), while accumulateAndReset is expected to run measurably slower by roughly one small allocation per drain (Counts escapes by design). Co-Authored-By: Claude Sonnet 5 --- .../trace/util/AccumulatorBenchmark.java | 60 +++++++++++++++++++ 1 file changed, 60 insertions(+) diff --git a/internal-api/src/jmh/java/datadog/trace/util/AccumulatorBenchmark.java b/internal-api/src/jmh/java/datadog/trace/util/AccumulatorBenchmark.java index 5b6299a7eae..90a9c04e58a 100644 --- a/internal-api/src/jmh/java/datadog/trace/util/AccumulatorBenchmark.java +++ b/internal-api/src/jmh/java/datadog/trace/util/AccumulatorBenchmark.java @@ -102,6 +102,7 @@ enum Counter { private final LongAdder adder = new LongAdder(); private final long[][] accumulator = Accumulator.EmbeddingSupport.create(Counter.values()); + private final Accumulator typedAccumulator = Accumulator.of(Counter.values()); private final ConcurrentHashMap chm = new ConcurrentHashMap<>(); private final LongAdder[] longAdderGroup = {new LongAdder()}; @@ -157,6 +158,24 @@ public void accumulatorIncrement_highContention() { Accumulator.EmbeddingSupport.inc(accumulator, Counter.HITS); } + /** + * The typed {@link Accumulator} wrapper's {@link Accumulator#inc}, paired against {@code + * accumulatorIncrement*} above: same underlying {@link Accumulator.EmbeddingSupport#inc} call, + * one extra field-load indirection through the instance. Should track the raw numbers closely -- + * a divergence here would mean the indirection isn't being inlined away. + */ + @Benchmark + @Threads(1) + public void typedIncrement_lowContention() { + typedAccumulator.inc(Counter.HITS); + } + + @Benchmark + @Threads(Threads.MAX) + public void typedIncrement_highContention() { + typedAccumulator.inc(Counter.HITS); + } + @Benchmark @Threads(1) public void chmAtomicLongIncrement_lowContention() { @@ -228,6 +247,47 @@ public void accumulatorMixed_drain(Blackhole blackhole) { blackhole.consume(Accumulator.EmbeddingSupport.accumulateAndReset(accumulator)); } + /** + * {@link Accumulator#update}, paired against the raw {@link Accumulator.EmbeddingSupport#update} + * lock/dispatch it wraps: the mutator here constructs a {@link Accumulator.Stripe} under the held + * lock and immediately lets it go, which is exactly the "small, non-capturing, non-escaping" + * shape documented as a scalar-replacement candidate. If escape analysis is doing its job, this + * tracks the raw call closely; if it regresses (e.g. after a JIT/JDK change, or a mutator shape + * that stops inlining), this is the number that would move. + */ + @Benchmark + @Threads(1) + public void typedUpdate_lowContention() { + typedAccumulator.update(stripe -> stripe.inc(Counter.HITS)); + } + + @Benchmark + @Threads(Threads.MAX) + public void typedUpdate_highContention() { + typedAccumulator.update(stripe -> stripe.inc(Counter.HITS)); + } + + /** + * {@link Accumulator#accumulateAndReset}, paired against the raw {@link + * Accumulator.EmbeddingSupport#accumulateAndReset} it wraps. Unlike {@link Accumulator.Stripe}, + * {@link Accumulator.Counts} is documented to escape (the caller holds and reads it after + * return), so this is expected to run measurably slower than the raw call by roughly one small + * object allocation per drain -- not a scalar-replacement candidate, and not meant to look free. + */ + @Benchmark + @Threads(1) + public void typedAccumulateAndReset_lowContention(Blackhole blackhole) { + typedAccumulator.inc(Counter.HITS); + blackhole.consume(typedAccumulator.accumulateAndReset()); + } + + @Benchmark + @Threads(Threads.MAX) + public void typedAccumulateAndReset_highContention(Blackhole blackhole) { + typedAccumulator.inc(Counter.HITS); + blackhole.consume(typedAccumulator.accumulateAndReset()); + } + @Benchmark @Threads(1) public void longAdderGroupIncrement_lowContention() { From bf67e0b7cbfcdc295b631e4fc5e83375ed9dd0a5 Mon Sep 17 00:00:00 2001 From: Douglas Q Hawkins Date: Tue, 1 Sep 2026 19:17:46 -0400 Subject: [PATCH 10/19] Record typed-vs-raw Accumulator benchmark results in AccumulatorBenchmark javadoc Confirms the wrapper's cost is not measurable: typedIncrement/typedUpdate track EmbeddingSupport within noise (the Stripe scalar-replaces as designed), and typedAccumulateAndReset tracks the raw drain within noise at low contention (the Counts allocation doesn't show up at this granularity). Co-Authored-By: Claude Sonnet 5 --- .../trace/util/AccumulatorBenchmark.java | 27 +++++++++++++++++++ 1 file changed, 27 insertions(+) diff --git a/internal-api/src/jmh/java/datadog/trace/util/AccumulatorBenchmark.java b/internal-api/src/jmh/java/datadog/trace/util/AccumulatorBenchmark.java index 90a9c04e58a..aec7b17333d 100644 --- a/internal-api/src/jmh/java/datadog/trace/util/AccumulatorBenchmark.java +++ b/internal-api/src/jmh/java/datadog/trace/util/AccumulatorBenchmark.java @@ -87,6 +87,33 @@ * write traffic concentrates on a few hot counters (favors thread-sharding) or spreads across many * (favors counter-sharding) -- not measured here, and worth checking against the real migration * targets before treating either number as the general answer. + * + *

{@code typed*}: what does the {@link Accumulator}/{@link Accumulator.Stripe}/{@link + * Accumulator.Counts} wrapping actually cost over calling {@link Accumulator.EmbeddingSupport} + * directly? {@code typedIncrement}/{@code typedUpdate} pair against {@code + * accumulatorIncrement} (the same underlying call), and {@code typedAccumulateAndReset} pairs + * against {@code accumulatorAccumulateAnd}. + * AccumulatorBenchmark.accumulatorIncrement_lowContention avgt 6 0.010 ± 0.001 us/op + * AccumulatorBenchmark.typedIncrement_lowContention avgt 6 0.010 ± 0.001 us/op + * AccumulatorBenchmark.accumulatorIncrement_highContention avgt 6 0.033 ± 0.008 us/op + * AccumulatorBenchmark.typedIncrement_highContention avgt 6 0.025 ± 0.015 us/op + * AccumulatorBenchmark.typedUpdate_lowContention avgt 6 0.010 ± 0.001 us/op + * AccumulatorBenchmark.typedUpdate_highContention avgt 6 0.037 ± 0.017 us/op + * AccumulatorBenchmark.accumulatorAccumulateAnd_lowContention avgt 6 0.161 ± 0.003 us/op + * AccumulatorBenchmark.typedAccumulateAndReset_lowContention avgt 6 0.164 ± 0.005 us/op + * AccumulatorBenchmark.accumulatorAccumulateAnd_highContention avgt 6 17.399 ± 3.091 us/op + * AccumulatorBenchmark.typedAccumulateAndReset_highContention avgt 6 12.666 ± 1.272 us/op + * {@code typedIncrement}/{@code typedUpdate} track the raw calls within noise at both + * contention levels -- the field-load indirection through the {@link Accumulator} instance and the + * fresh {@link Accumulator.Stripe} constructed under {@code update}'s held lock both disappear, + * consistent with a small, non-capturing mutator letting escape analysis scalar-replace the {@code + * Stripe}. {@code typedAccumulateAndReset} also tracks the raw drain within noise at low contention + * (0.164 vs 0.161 us/op) -- the one-{@link Accumulator.Counts}-object-per-drain allocation it's + * documented to pay doesn't show up at this granularity. The high-contention gap in the other + * direction (12.666 vs 17.399) is not a real typed-vs-raw effect -- wrapping an already-drained + * array can only add cost, never remove it -- it's the same run-to-run lock-contention noise this + * exact measurement already shows above (13.431, 15.515, 17.399 us/op across three otherwise + * identical runs). Net: the wrapper's cost was not measurable in this run. */ @State(Scope.Benchmark) @Warmup(iterations = 1, time = 10) From 3484ed7ff67e6d5a4302e55b87b1943b09346422 Mon Sep 17 00:00:00 2001 From: Douglas Q Hawkins Date: Tue, 1 Sep 2026 21:29:56 -0400 Subject: [PATCH 11/19] Rename accumulatorAccumulateAnd* benchmarks, cap footprint test thread count The benchmark calls EmbeddingSupport.accumulateAndReset, not accumulateAnd -- rename to match. Also cap contendedFootprint's thread count at 16: uncapped availableProcessors() on a high-core build agent spins up one thread per core, all busy-spinning for two seconds, which can dominate the host during a parallel test run for no added signal over a fixed small contention level. Co-Authored-By: Claude Sonnet 5 --- .../trace/util/AccumulatorBenchmark.java | 18 +++++++++--------- .../trace/util/AccumulatorFootprintTest.java | 2 +- 2 files changed, 10 insertions(+), 10 deletions(-) diff --git a/internal-api/src/jmh/java/datadog/trace/util/AccumulatorBenchmark.java b/internal-api/src/jmh/java/datadog/trace/util/AccumulatorBenchmark.java index aec7b17333d..abfd778c17a 100644 --- a/internal-api/src/jmh/java/datadog/trace/util/AccumulatorBenchmark.java +++ b/internal-api/src/jmh/java/datadog/trace/util/AccumulatorBenchmark.java @@ -56,8 +56,8 @@ * AccumulatorBenchmark.chmAtomicLongIncrement_highContention avgt 6 0.417 ± 0.543 us/op * AccumulatorBenchmark.longAdderSumThenReset_lowContention avgt 6 0.012 ± 0.001 us/op * AccumulatorBenchmark.longAdderSumThenReset_highContention avgt 6 2.433 ± 0.203 us/op - * AccumulatorBenchmark.accumulatorAccumulateAnd_lowContention avgt 6 0.162 ± 0.009 us/op - * AccumulatorBenchmark.accumulatorAccumulateAnd_highContention avgt 6 15.515 ± 4.094 us/op + * AccumulatorBenchmark.accumulatorAccumulateAndReset_lowContention avgt 6 0.162 ± 0.009 us/op + * AccumulatorBenchmark.accumulatorAccumulateAndReset_highContention avgt 6 15.515 ± 4.094 us/op * * *

(This run had some background noise from another session on the measurement machine; the @@ -72,7 +72,7 @@ * lock a caller takes). This closes the same reset hazard as {@link Accumulator}, but stripes by * counter instead of by thread. * AccumulatorBenchmark.accumulatorIncrement_highContention avgt 6 0.029 ± 0.051 us/op - * AccumulatorBenchmark.accumulatorAccumulateAnd_highContention avgt 6 13.431 ± 5.876 us/op + * AccumulatorBenchmark.accumulatorAccumulateAndReset_highContention avgt 6 13.431 ± 5.876 us/op * AccumulatorBenchmark.longAdderGroupIncrement_highContention avgt 6 0.294 ± 0.088 us/op * AccumulatorBenchmark.longAdderGroupAccumulateAnd_highContention avgt 6 0.549 ± 0.291 us/op * Not "similar cost" -- a clean trade-off inversion. With this benchmark's single counter, @@ -92,16 +92,16 @@ * Accumulator.Counts} wrapping actually cost over calling {@link Accumulator.EmbeddingSupport} * directly? {@code typedIncrement}/{@code typedUpdate} pair against {@code * accumulatorIncrement} (the same underlying call), and {@code typedAccumulateAndReset} pairs - * against {@code accumulatorAccumulateAnd}. + * against {@code accumulatorAccumulateAndReset}. * AccumulatorBenchmark.accumulatorIncrement_lowContention avgt 6 0.010 ± 0.001 us/op * AccumulatorBenchmark.typedIncrement_lowContention avgt 6 0.010 ± 0.001 us/op * AccumulatorBenchmark.accumulatorIncrement_highContention avgt 6 0.033 ± 0.008 us/op * AccumulatorBenchmark.typedIncrement_highContention avgt 6 0.025 ± 0.015 us/op * AccumulatorBenchmark.typedUpdate_lowContention avgt 6 0.010 ± 0.001 us/op * AccumulatorBenchmark.typedUpdate_highContention avgt 6 0.037 ± 0.017 us/op - * AccumulatorBenchmark.accumulatorAccumulateAnd_lowContention avgt 6 0.161 ± 0.003 us/op + * AccumulatorBenchmark.accumulatorAccumulateAndReset_lowContention avgt 6 0.161 ± 0.003 us/op * AccumulatorBenchmark.typedAccumulateAndReset_lowContention avgt 6 0.164 ± 0.005 us/op - * AccumulatorBenchmark.accumulatorAccumulateAnd_highContention avgt 6 17.399 ± 3.091 us/op + * AccumulatorBenchmark.accumulatorAccumulateAndReset_highContention avgt 6 17.399 ± 3.091 us/op * AccumulatorBenchmark.typedAccumulateAndReset_highContention avgt 6 12.666 ± 1.272 us/op * {@code typedIncrement}/{@code typedUpdate} track the raw calls within noise at both * contention levels -- the field-load indirection through the {@link Accumulator} instance and the @@ -231,7 +231,7 @@ public void longAdderSumThenReset_highContention(Blackhole blackhole) { @Benchmark @Threads(1) - public void accumulatorAccumulateAnd_lowContention(Blackhole blackhole) { + public void accumulatorAccumulateAndReset_lowContention(Blackhole blackhole) { Accumulator.EmbeddingSupport.inc(accumulator, Counter.HITS); blackhole.consume(Accumulator.EmbeddingSupport.accumulateAndReset(accumulator)); } @@ -246,13 +246,13 @@ public void accumulatorAccumulateAnd_lowContention(Blackhole blackhole) { */ @Benchmark @Threads(Threads.MAX) - public void accumulatorAccumulateAnd_highContention(Blackhole blackhole) { + public void accumulatorAccumulateAndReset_highContention(Blackhole blackhole) { Accumulator.EmbeddingSupport.inc(accumulator, Counter.HITS); blackhole.consume(Accumulator.EmbeddingSupport.accumulateAndReset(accumulator)); } /** - * The realistic counterpart to {@code accumulatorAccumulateAnd_highContention}: many writer + * The realistic counterpart to {@code accumulatorAccumulateAndReset_highContention}: many writer * threads incrementing, and a single dedicated thread polling {@link * Accumulator.EmbeddingSupport#accumulateAndReset} -- not every thread doing both on every op. * {@code accumulatorMixed-write} measures increment cost while a drain is actively contending for diff --git a/internal-api/src/test/java/datadog/trace/util/AccumulatorFootprintTest.java b/internal-api/src/test/java/datadog/trace/util/AccumulatorFootprintTest.java index 8b47237512f..868f2ab6d0d 100644 --- a/internal-api/src/test/java/datadog/trace/util/AccumulatorFootprintTest.java +++ b/internal-api/src/test/java/datadog/trace/util/AccumulatorFootprintTest.java @@ -103,7 +103,7 @@ void contendedFootprint() throws InterruptedException { LongAdder[] adders = freshAdders(); long freshAdderBytes = bytes((Object) adders); - int threads = Math.max(4, Runtime.getRuntime().availableProcessors()); + int threads = Math.min(16, Math.max(4, Runtime.getRuntime().availableProcessors())); ExecutorService pool = Executors.newFixedThreadPool(threads); CountDownLatch start = new CountDownLatch(1); CountDownLatch done = new CountDownLatch(threads); From 28de3032f32d17f290c946fef4721990c85e3250 Mon Sep 17 00:00:00 2001 From: Douglas Q Hawkins Date: Wed, 2 Sep 2026 08:46:41 -0400 Subject: [PATCH 12/19] Add a non-destructive Accumulator.sum() for live diagnostic reads accumulateAndReset() resets on every drain, so it can't back a live snapshot (e.g. summary()) without racing whatever else drains on a reporting cadence -- whichever call empties the stripes first starves the other's delta. sum() combines every stripe without resetting it, giving a live, non-destructive read that can't interfere with a concurrent accumulateAndReset(). --- .../java/datadog/trace/util/Accumulator.java | 37 ++++++++++++++-- .../datadog/trace/util/AccumulatorTest.java | 44 +++++++++++++++++++ 2 files changed, 78 insertions(+), 3 deletions(-) diff --git a/internal-api/src/main/java/datadog/trace/util/Accumulator.java b/internal-api/src/main/java/datadog/trace/util/Accumulator.java index f6c181a2dcf..15398802779 100644 --- a/internal-api/src/main/java/datadog/trace/util/Accumulator.java +++ b/internal-api/src/main/java/datadog/trace/util/Accumulator.java @@ -113,9 +113,21 @@ public Counts accumulateAndReset() { } /** - * A typed view over a drained {@code long[]}, returned by {@link #accumulateAndReset}: the same - * enum-ordinal type checking {@link Accumulator} provides on writes, applied to the read side - * too. + * Combines every stripe without resetting it, returning the sum as a typed view -- a live, + * non-destructive snapshot for a diagnostic read (e.g. {@code summary()}) that must not perturb + * the delta a concurrent {@link #accumulateAndReset} on a reporting cadence is about to report. + * + * @return the sum, keyed by the enum's {@code ordinal()} + * @see EmbeddingSupport#sum(long[][]) + */ + public Counts sum() { + return new Counts<>(EmbeddingSupport.sum(data)); + } + + /** + * A typed view over a drained {@code long[]}, returned by {@link #accumulateAndReset} or {@link + * #sum}: the same enum-ordinal type checking {@link Accumulator} provides on writes, applied to + * the read side too. * *

Unlike {@link Stripe}, this is expected to escape -- the caller holds and reads it after the * call returns -- so it's a real, per-drain allocation, not a scalar-replacement candidate. @@ -282,6 +294,25 @@ public static long[] accumulateAndReset(long[][] data) { return acc; } + /** + * Combines every stripe without resetting it, returning the sum -- a live, non-destructive + * snapshot for a diagnostic read that must not perturb the delta a concurrent {@link + * #accumulateAndReset} on a reporting cadence is about to report. + * + * @return a new array the same length as one stripe's row, indexed by the enum's {@code + * ordinal()} for the positions actually in use (trailing padding positions are always zero) + * @see #accumulateAndReset(long[][]) + */ + public static long[] sum(long[][] data) { + long[] acc = new long[data[0].length]; + for (long[] stripe : data) { + synchronized (stripe) { + combine(acc, stripe); + } + } + return acc; + } + /** * {@code acc[i] += stripe[i]} for every index -- a fixed-trip-count loop C2 can auto-vectorize. */ diff --git a/internal-api/src/test/java/datadog/trace/util/AccumulatorTest.java b/internal-api/src/test/java/datadog/trace/util/AccumulatorTest.java index a93665cb238..12ffbba2bee 100644 --- a/internal-api/src/test/java/datadog/trace/util/AccumulatorTest.java +++ b/internal-api/src/test/java/datadog/trace/util/AccumulatorTest.java @@ -175,6 +175,50 @@ void concurrentAccumulateAndDuringWritesNeverExceedsWritten() } } + @Test + void sumDoesNotResetStripes() { + long[][] data = Accumulator.EmbeddingSupport.create(Counters.values()); + Accumulator.EmbeddingSupport.inc(data, Counters.FOO); + + long[] first = Accumulator.EmbeddingSupport.sum(data); + assertEquals(1L, first[Counters.FOO.ordinal()]); + + // sum() didn't reset anything, so a second sum() sees the same total + long[] second = Accumulator.EmbeddingSupport.sum(data); + assertEquals(1L, second[Counters.FOO.ordinal()]); + + // and a real drain afterwards still sees the value sum() didn't consume + long[] drained = Accumulator.EmbeddingSupport.accumulateAndReset(data); + assertEquals(1L, drained[Counters.FOO.ordinal()]); + } + + @Test + void sumReflectsIncrementsMadeAfterAnEarlierSum() { + long[][] data = Accumulator.EmbeddingSupport.create(Counters.values()); + Accumulator.EmbeddingSupport.inc(data, Counters.FOO); + Accumulator.EmbeddingSupport.sum(data); + + Accumulator.EmbeddingSupport.inc(data, Counters.FOO); + long[] second = Accumulator.EmbeddingSupport.sum(data); + assertEquals(2L, second[Counters.FOO.ordinal()]); + } + + @Test + void typedWrapperSumDoesNotReset() { + Accumulator counters = Accumulator.of(Counters.values()); + counters.inc(Counters.FOO); + counters.add(Counters.BAR, 5L); + + Accumulator.Counts sum = counters.sum(); + assertEquals(1L, sum.get(Counters.FOO)); + assertEquals(5L, sum.get(Counters.BAR)); + + // still there for the real drain + Accumulator.Counts drained = counters.accumulateAndReset(); + assertEquals(1L, drained.get(Counters.FOO)); + assertEquals(5L, drained.get(Counters.BAR)); + } + @Test void typedWrapperDelegatesToEmbeddingSupport() { Accumulator counters = Accumulator.of(Counters.values()); From 3c5b7ef1a39175723c45933d6a33d7e0d56a5ebb Mon Sep 17 00:00:00 2001 From: Douglas Q Hawkins Date: Wed, 2 Sep 2026 08:49:52 -0400 Subject: [PATCH 13/19] Add Accumulator.Counts.plus() to combine a stored total with a live sum Lets a caller keep a running Counts updated by periodic drains and still answer "what's the total right now" by combining it with a fresh, non-destructive sum() -- without needing to reset or drain anything just to read a live number. --- .../java/datadog/trace/util/Accumulator.java | 13 ++++++++++++ .../datadog/trace/util/AccumulatorTest.java | 21 +++++++++++++++++++ 2 files changed, 34 insertions(+) diff --git a/internal-api/src/main/java/datadog/trace/util/Accumulator.java b/internal-api/src/main/java/datadog/trace/util/Accumulator.java index 15398802779..911c73102f0 100644 --- a/internal-api/src/main/java/datadog/trace/util/Accumulator.java +++ b/internal-api/src/main/java/datadog/trace/util/Accumulator.java @@ -145,6 +145,19 @@ private Counts(long[] counts) { public long get(E key) { return counts[key.ordinal()]; } + + /** + * Adds {@code other} to this, key by key, returning a new {@link Counts} rather than mutating + * either input -- combines a stored running total with a fresh, non-destructive {@link #sum} to + * answer "what's the live total right now" without ever resetting anything. + */ + public Counts plus(Counts other) { + long[] combined = counts.clone(); + for (int i = 0; i < combined.length; i++) { + combined[i] += other.counts[i]; + } + return new Counts<>(combined); + } } /** diff --git a/internal-api/src/test/java/datadog/trace/util/AccumulatorTest.java b/internal-api/src/test/java/datadog/trace/util/AccumulatorTest.java index 12ffbba2bee..f7ae69f43e8 100644 --- a/internal-api/src/test/java/datadog/trace/util/AccumulatorTest.java +++ b/internal-api/src/test/java/datadog/trace/util/AccumulatorTest.java @@ -219,6 +219,27 @@ void typedWrapperSumDoesNotReset() { assertEquals(5L, drained.get(Counters.BAR)); } + @Test + void plusCombinesAStoredRunningTotalWithALiveSumWithoutMutatingEither() { + Accumulator counters = Accumulator.of(Counters.values()); + counters.inc(Counters.FOO); + counters.add(Counters.BAR, 5L); + + // drain once, e.g. as if a reporting cycle already ran and stored this total + Accumulator.Counts storedTotal = counters.accumulateAndReset(); + + // more activity happens after that drain, before the next one + counters.inc(Counters.FOO); + + Accumulator.Counts live = storedTotal.plus(counters.sum()); + assertEquals(2L, live.get(Counters.FOO)); + assertEquals(5L, live.get(Counters.BAR)); + + // neither input was mutated by combining them + assertEquals(1L, storedTotal.get(Counters.FOO)); + assertEquals(1L, counters.sum().get(Counters.FOO)); + } + @Test void typedWrapperDelegatesToEmbeddingSupport() { Accumulator counters = Accumulator.of(Counters.values()); From 93e2c2d4a95e4bf92868a683a0794ee21eadb025 Mon Sep 17 00:00:00 2001 From: Douglas Q Hawkins Date: Wed, 2 Sep 2026 10:20:48 -0400 Subject: [PATCH 14/19] Add Accumulator.Counts.zero() to seed a running total without a scratch Accumulator Replaces the awkward "construct an Accumulator just to call sum() on it for zeros" pattern a consumer would otherwise need to seed a stored running total before any real drain has happened. Co-Authored-By: Claude Sonnet 5 --- .../main/java/datadog/trace/util/Accumulator.java | 11 +++++++++++ .../java/datadog/trace/util/AccumulatorTest.java | 13 +++++++++++++ 2 files changed, 24 insertions(+) diff --git a/internal-api/src/main/java/datadog/trace/util/Accumulator.java b/internal-api/src/main/java/datadog/trace/util/Accumulator.java index 911c73102f0..fbae5a5ecb5 100644 --- a/internal-api/src/main/java/datadog/trace/util/Accumulator.java +++ b/internal-api/src/main/java/datadog/trace/util/Accumulator.java @@ -141,6 +141,17 @@ private Counts(long[] counts) { this.counts = counts; } + /** + * An all-zero {@link Counts}, sized for {@code values} -- for seeding a running total before + * any real drain has happened, without needing a scratch {@link Accumulator} just to call + * {@link Accumulator#sum()} on it. + * + * @param values the enum constants naming each counter, e.g. {@code MyCounters.values()} + */ + public static > Counts zero(E[] values) { + return new Counts<>(new long[values.length]); + } + /** The counter named by {@code key}. */ public long get(E key) { return counts[key.ordinal()]; diff --git a/internal-api/src/test/java/datadog/trace/util/AccumulatorTest.java b/internal-api/src/test/java/datadog/trace/util/AccumulatorTest.java index f7ae69f43e8..0c343ddb03f 100644 --- a/internal-api/src/test/java/datadog/trace/util/AccumulatorTest.java +++ b/internal-api/src/test/java/datadog/trace/util/AccumulatorTest.java @@ -219,6 +219,19 @@ void typedWrapperSumDoesNotReset() { assertEquals(5L, drained.get(Counters.BAR)); } + @Test + void zeroSeedsAnAllZeroCountsWithoutAScratchAccumulator() { + Accumulator.Counts zero = Accumulator.Counts.zero(Counters.values()); + assertEquals(0L, zero.get(Counters.FOO)); + assertEquals(0L, zero.get(Counters.BAR)); + + Accumulator counters = Accumulator.of(Counters.values()); + counters.inc(Counters.FOO); + + Accumulator.Counts live = zero.plus(counters.sum()); + assertEquals(1L, live.get(Counters.FOO)); + } + @Test void plusCombinesAStoredRunningTotalWithALiveSumWithoutMutatingEither() { Accumulator counters = Accumulator.of(Counters.values()); From 34f1830e0ad0bfd76bcd4998805d124ac52346c9 Mon Sep 17 00:00:00 2001 From: Douglas Q Hawkins Date: Wed, 2 Sep 2026 10:31:33 -0400 Subject: [PATCH 15/19] Add a contextual Accumulator.update(context, BiConsumer) overload Lets a caller pass a value the mutator needs as an explicit parameter instead of capturing it, for cases where that capture is the only thing stopping the lambda from being non-capturing (and thus allocation-free per Java's own lambda-caching behavior). Boxes the context if it's a primitive at the call site -- a real trade against the capture it replaces, not a free win. Co-Authored-By: Claude Sonnet 5 --- .../java/datadog/trace/util/Accumulator.java | 22 +++++++++++++++++++ .../datadog/trace/util/AccumulatorTest.java | 16 ++++++++++++++ 2 files changed, 38 insertions(+) diff --git a/internal-api/src/main/java/datadog/trace/util/Accumulator.java b/internal-api/src/main/java/datadog/trace/util/Accumulator.java index fbae5a5ecb5..f0ba8755974 100644 --- a/internal-api/src/main/java/datadog/trace/util/Accumulator.java +++ b/internal-api/src/main/java/datadog/trace/util/Accumulator.java @@ -4,6 +4,7 @@ import datadog.trace.api.function.Strategy; import datadog.trace.api.function.StrategyConsumer; import java.util.Arrays; +import java.util.function.BiConsumer; import java.util.function.Consumer; import javax.annotation.ParametersAreNonnullByDefault; import javax.annotation.concurrent.GuardedBy; @@ -72,6 +73,27 @@ public void update(@Strategy Consumer> mutator) { } } + /** + * Like {@link #update(Consumer)}, but passes {@code context} to {@code mutator} as an explicit + * parameter instead of letting the mutator capture it -- for a caller that would otherwise need + * to close over a local (e.g. a count) just to get it into the critical section. Note {@code + * context} is boxed if it's a primitive at the call site; that's a real allocation trade against + * the capturing lambda it replaces, not a free win -- prefer this only when {@code context} would + * otherwise be the only thing forcing a capture. + * + * @param context a value the mutator needs, passed in rather than captured + * @param mutator a strategy over {@code context} and the selected stripe; keep it small and + * non-capturing so it inlines into the lock's critical section, and don't let the {@link + * Stripe} escape it (store it, return it, hand it to another thread) -- see {@link Stripe} + */ + @StrategyConsumer + public void update(C context, @Strategy BiConsumer> mutator) { + long[] stripe = EmbeddingSupport.stripeOf(data); + synchronized (stripe) { + mutator.accept(context, new Stripe<>(stripe)); + } + } + /** * A typed view over one stripe, handed to an {@link #update} strategy: the same enum-ordinal type * checking {@link Accumulator} provides at the top level, applied inside the critical section diff --git a/internal-api/src/test/java/datadog/trace/util/AccumulatorTest.java b/internal-api/src/test/java/datadog/trace/util/AccumulatorTest.java index 0c343ddb03f..f2b2ff20471 100644 --- a/internal-api/src/test/java/datadog/trace/util/AccumulatorTest.java +++ b/internal-api/src/test/java/datadog/trace/util/AccumulatorTest.java @@ -266,4 +266,20 @@ void typedWrapperDelegatesToEmbeddingSupport() { assertEquals(5L, drained.get(Counters.BAR)); assertEquals(1L, drained.get(Counters.BAZ)); } + + @Test + void contextualUpdatePassesContextInsteadOfCapturingIt() { + Accumulator counters = Accumulator.of(Counters.values()); + + counters.update( + 5L, + (delta, stripe) -> { + stripe.inc(Counters.FOO); + stripe.add(Counters.BAR, delta); + }); + + Accumulator.Counts drained = counters.accumulateAndReset(); + assertEquals(1L, drained.get(Counters.FOO)); + assertEquals(5L, drained.get(Counters.BAR)); + } } From 8df6143d9c53dc6b6f16868097cc0f1e25313d23 Mon Sep 17 00:00:00 2001 From: Douglas Q Hawkins Date: Wed, 2 Sep 2026 11:15:19 -0400 Subject: [PATCH 16/19] Let Counts expose its own keys and add Class-based factories Accumulator/Counts now remember the enum's values() array from construction, so Counts.values() lets a caller iterate its own keys without separately threading E.values() alongside it (motivated by reducing StatsDCountReporter's call-site ceremony). Also add Accumulator.of(Class)/Counts.zero(Class) overloads alongside the existing array-based ones, for callers that'd rather pass MyEnum.class. Co-Authored-By: Claude Sonnet 5 --- .../java/datadog/trace/util/Accumulator.java | 42 +++++++++++++++---- .../datadog/trace/util/AccumulatorTest.java | 26 ++++++++++++ 2 files changed, 61 insertions(+), 7 deletions(-) diff --git a/internal-api/src/main/java/datadog/trace/util/Accumulator.java b/internal-api/src/main/java/datadog/trace/util/Accumulator.java index f0ba8755974..11c63b44a65 100644 --- a/internal-api/src/main/java/datadog/trace/util/Accumulator.java +++ b/internal-api/src/main/java/datadog/trace/util/Accumulator.java @@ -34,16 +34,25 @@ */ public final class Accumulator> { private final long[][] data; + private final E[] values; - private Accumulator(long[][] data) { + private Accumulator(long[][] data, E[] values) { this.data = data; + this.values = values; } /** * @param values the enum constants naming each counter, e.g. {@code MyCounters.values()} */ public static > Accumulator of(E[] values) { - return new Accumulator<>(EmbeddingSupport.create(values)); + return new Accumulator<>(EmbeddingSupport.create(values), values); + } + + /** + * @param enumType the enum naming each counter, e.g. {@code MyCounters.class} + */ + public static > Accumulator of(Class enumType) { + return of(enumType.getEnumConstants()); } /** Increments the counter named by {@code key} in the calling thread's stripe by one. */ @@ -131,7 +140,7 @@ public void add(E key, long delta) { * @see EmbeddingSupport#accumulateAndReset */ public Counts accumulateAndReset() { - return new Counts<>(EmbeddingSupport.accumulateAndReset(data)); + return new Counts<>(EmbeddingSupport.accumulateAndReset(data), values); } /** @@ -143,7 +152,7 @@ public Counts accumulateAndReset() { * @see EmbeddingSupport#sum(long[][]) */ public Counts sum() { - return new Counts<>(EmbeddingSupport.sum(data)); + return new Counts<>(EmbeddingSupport.sum(data), values); } /** @@ -158,9 +167,11 @@ public Counts sum() { */ public static final class Counts> { private final long[] counts; + private final E[] values; - private Counts(long[] counts) { + private Counts(long[] counts, E[] values) { this.counts = counts; + this.values = values; } /** @@ -171,7 +182,15 @@ private Counts(long[] counts) { * @param values the enum constants naming each counter, e.g. {@code MyCounters.values()} */ public static > Counts zero(E[] values) { - return new Counts<>(new long[values.length]); + return new Counts<>(new long[values.length], values); + } + + /** + * @param enumType the enum naming each counter, e.g. {@code MyCounters.class} + * @see #zero(Enum[]) + */ + public static > Counts zero(Class enumType) { + return zero(enumType.getEnumConstants()); } /** The counter named by {@code key}. */ @@ -179,6 +198,15 @@ public long get(E key) { return counts[key.ordinal()]; } + /** + * The enum constants this {@link Counts} is keyed by, in declaration order -- for a caller that + * wants to iterate every counter (e.g. reporting each one) without separately having to pass + * {@code E.values()} alongside this object. + */ + public E[] values() { + return values; + } + /** * Adds {@code other} to this, key by key, returning a new {@link Counts} rather than mutating * either input -- combines a stored running total with a fresh, non-destructive {@link #sum} to @@ -189,7 +217,7 @@ public Counts plus(Counts other) { for (int i = 0; i < combined.length; i++) { combined[i] += other.counts[i]; } - return new Counts<>(combined); + return new Counts<>(combined, values); } } diff --git a/internal-api/src/test/java/datadog/trace/util/AccumulatorTest.java b/internal-api/src/test/java/datadog/trace/util/AccumulatorTest.java index f2b2ff20471..c99c4683076 100644 --- a/internal-api/src/test/java/datadog/trace/util/AccumulatorTest.java +++ b/internal-api/src/test/java/datadog/trace/util/AccumulatorTest.java @@ -232,6 +232,32 @@ void zeroSeedsAnAllZeroCountsWithoutAScratchAccumulator() { assertEquals(1L, live.get(Counters.FOO)); } + @Test + void ofAndZeroAcceptAnEnumClassInsteadOfAValuesArray() { + Accumulator counters = Accumulator.of(Counters.class); + counters.inc(Counters.FOO); + + Accumulator.Counts zero = Accumulator.Counts.zero(Counters.class); + Accumulator.Counts live = zero.plus(counters.sum()); + assertEquals(1L, live.get(Counters.FOO)); + } + + @Test + void countsExposesItsOwnKeysWithoutASeparateValuesArray() { + Accumulator counters = Accumulator.of(Counters.values()); + counters.inc(Counters.FOO); + counters.add(Counters.BAR, 5L); + + Accumulator.Counts drained = counters.accumulateAndReset(); + assertEquals(Counters.values().length, drained.values().length); + + long total = 0L; + for (Counters c : drained.values()) { + total += drained.get(c); + } + assertEquals(6L, total); + } + @Test void plusCombinesAStoredRunningTotalWithALiveSumWithoutMutatingEither() { Accumulator counters = Accumulator.of(Counters.values()); From 9bc7abccf616a9fde3310a5d1236c78b6abe769a Mon Sep 17 00:00:00 2001 From: Douglas Q Hawkins Date: Wed, 2 Sep 2026 11:21:30 -0400 Subject: [PATCH 17/19] Rename Counts.values() to Counts.keys() values() read as reusing Enum.values()'s exact word for a different concept -- Counts is a generic primitive, not metrics-specific, so keys() (matching the existing get(E key) naming) is clearer without tying the name to any one consumer. Co-Authored-By: Claude Sonnet 5 --- .../src/main/java/datadog/trace/util/Accumulator.java | 2 +- .../src/test/java/datadog/trace/util/AccumulatorTest.java | 4 ++-- 2 files changed, 3 insertions(+), 3 deletions(-) diff --git a/internal-api/src/main/java/datadog/trace/util/Accumulator.java b/internal-api/src/main/java/datadog/trace/util/Accumulator.java index 11c63b44a65..b045506dfa8 100644 --- a/internal-api/src/main/java/datadog/trace/util/Accumulator.java +++ b/internal-api/src/main/java/datadog/trace/util/Accumulator.java @@ -203,7 +203,7 @@ public long get(E key) { * wants to iterate every counter (e.g. reporting each one) without separately having to pass * {@code E.values()} alongside this object. */ - public E[] values() { + public E[] keys() { return values; } diff --git a/internal-api/src/test/java/datadog/trace/util/AccumulatorTest.java b/internal-api/src/test/java/datadog/trace/util/AccumulatorTest.java index c99c4683076..fa10668e4d3 100644 --- a/internal-api/src/test/java/datadog/trace/util/AccumulatorTest.java +++ b/internal-api/src/test/java/datadog/trace/util/AccumulatorTest.java @@ -249,10 +249,10 @@ void countsExposesItsOwnKeysWithoutASeparateValuesArray() { counters.add(Counters.BAR, 5L); Accumulator.Counts drained = counters.accumulateAndReset(); - assertEquals(Counters.values().length, drained.values().length); + assertEquals(Counters.values().length, drained.keys().length); long total = 0L; - for (Counters c : drained.values()) { + for (Counters c : drained.keys()) { total += drained.get(c); } assertEquals(6L, total); From 7e80a21574ea640f58d6921f7ebd034f68467b92 Mon Sep 17 00:00:00 2001 From: Douglas Q Hawkins Date: Wed, 2 Sep 2026 11:57:46 -0400 Subject: [PATCH 18/19] Add Accumulator.update(long, ObjLongConsumer) to avoid boxing a primitive context --- .../java/datadog/trace/util/Accumulator.java | 31 +++++++++++++++- .../datadog/trace/util/AccumulatorTest.java | 36 ++++++++++++++++++- 2 files changed, 65 insertions(+), 2 deletions(-) diff --git a/internal-api/src/main/java/datadog/trace/util/Accumulator.java b/internal-api/src/main/java/datadog/trace/util/Accumulator.java index b045506dfa8..2e247b2e981 100644 --- a/internal-api/src/main/java/datadog/trace/util/Accumulator.java +++ b/internal-api/src/main/java/datadog/trace/util/Accumulator.java @@ -6,6 +6,7 @@ import java.util.Arrays; import java.util.function.BiConsumer; import java.util.function.Consumer; +import java.util.function.ObjLongConsumer; import javax.annotation.ParametersAreNonnullByDefault; import javax.annotation.concurrent.GuardedBy; @@ -88,7 +89,8 @@ public void update(@Strategy Consumer> mutator) { * to close over a local (e.g. a count) just to get it into the critical section. Note {@code * context} is boxed if it's a primitive at the call site; that's a real allocation trade against * the capturing lambda it replaces, not a free win -- prefer this only when {@code context} would - * otherwise be the only thing forcing a capture. + * otherwise be the only thing forcing a capture. For an {@code int} or {@code long} context, use + * {@link #update(long, ObjLongConsumer)} instead to avoid that boxing entirely. * * @param context a value the mutator needs, passed in rather than captured * @param mutator a strategy over {@code context} and the selected stripe; keep it small and @@ -103,6 +105,33 @@ public void update(C context, @Strategy BiConsumer> mutator) { } } + /** + * Like {@link #update(Object, BiConsumer)}, but for a {@code long} context -- reuses the JDK's + * {@link ObjLongConsumer} instead of the generic {@link BiConsumer}, so {@code context} is passed + * as a primitive {@code long} rather than boxed into a {@link Long}. Covers an {@code int} + * context too: it widens to {@code long} for free at the call site, no boxing either way. (A + * dedicated {@code int} overload isn't offered alongside this one -- an {@code int} argument + * would be ambiguous between the two, since it's an exact match for one and a free widening + * conversion to the other, and unrelated functional-interface types block the usual most-specific + * tiebreak.) + * + *

Note the parameter order this forces: {@link ObjLongConsumer#accept} takes {@code (T, + * long)}, so the mutator sees the stripe first and the context second -- the opposite order from + * {@link #update(Object, BiConsumer)}. + * + * @param context a primitive value the mutator needs, passed in rather than captured or boxed + * @param mutator a strategy over the selected stripe and {@code context}; keep it small and + * non-capturing so it inlines into the lock's critical section, and don't let the {@link + * Stripe} escape it (store it, return it, hand it to another thread) -- see {@link Stripe} + */ + @StrategyConsumer + public void update(long context, @Strategy ObjLongConsumer> mutator) { + long[] stripe = EmbeddingSupport.stripeOf(data); + synchronized (stripe) { + mutator.accept(new Stripe<>(stripe), context); + } + } + /** * A typed view over one stripe, handed to an {@link #update} strategy: the same enum-ordinal type * checking {@link Accumulator} provides at the top level, applied inside the critical section diff --git a/internal-api/src/test/java/datadog/trace/util/AccumulatorTest.java b/internal-api/src/test/java/datadog/trace/util/AccumulatorTest.java index fa10668e4d3..5caa32549ed 100644 --- a/internal-api/src/test/java/datadog/trace/util/AccumulatorTest.java +++ b/internal-api/src/test/java/datadog/trace/util/AccumulatorTest.java @@ -296,10 +296,44 @@ void typedWrapperDelegatesToEmbeddingSupport() { @Test void contextualUpdatePassesContextInsteadOfCapturingIt() { Accumulator counters = Accumulator.of(Counters.values()); + String context = "abcde"; + + counters.update( + context, + (ctx, stripe) -> { + stripe.inc(Counters.FOO); + stripe.add(Counters.BAR, ctx.length()); + }); + + Accumulator.Counts drained = counters.accumulateAndReset(); + assertEquals(1L, drained.get(Counters.FOO)); + assertEquals(5L, drained.get(Counters.BAR)); + } + + @Test + void intContextWidensIntoTheLongOverloadWithoutBoxing() { + Accumulator counters = Accumulator.of(Counters.values()); + int delta = 5; + + counters.update( + delta, + (stripe, d) -> { + stripe.inc(Counters.FOO); + stripe.add(Counters.BAR, d); + }); + + Accumulator.Counts drained = counters.accumulateAndReset(); + assertEquals(1L, drained.get(Counters.FOO)); + assertEquals(5L, drained.get(Counters.BAR)); + } + + @Test + void longContextualUpdateAvoidsBoxing() { + Accumulator counters = Accumulator.of(Counters.values()); counters.update( 5L, - (delta, stripe) -> { + (stripe, delta) -> { stripe.inc(Counters.FOO); stripe.add(Counters.BAR, delta); }); From 69251a58c352f0dafdbf6dd8ff2bb0131f6666e0 Mon Sep 17 00:00:00 2001 From: Douglas Q Hawkins Date: Wed, 2 Sep 2026 14:15:04 -0400 Subject: [PATCH 19/19] Only touch the real counter width in combine/reset, not the padded stripe row accumulateAndReset()/sum() drained the full paddedWidth-length stripe row, so every drain wrote through Arrays.fill into the trailing cache-line buffer that paddedWidth() adds to keep adjacent stripes from false-sharing. Thread the real width through combine()/reset() so the padding stays untouched after create()'s zero-init, matching the array size Counts.zero() already returns. --- .../trace/util/AccumulatorBenchmark.java | 9 ++-- .../java/datadog/trace/util/Accumulator.java | 54 +++++++++++-------- .../datadog/trace/util/AccumulatorTest.java | 46 +++++++++------- 3 files changed, 66 insertions(+), 43 deletions(-) diff --git a/internal-api/src/jmh/java/datadog/trace/util/AccumulatorBenchmark.java b/internal-api/src/jmh/java/datadog/trace/util/AccumulatorBenchmark.java index abfd778c17a..3b7c7eeb05f 100644 --- a/internal-api/src/jmh/java/datadog/trace/util/AccumulatorBenchmark.java +++ b/internal-api/src/jmh/java/datadog/trace/util/AccumulatorBenchmark.java @@ -233,7 +233,8 @@ public void longAdderSumThenReset_highContention(Blackhole blackhole) { @Threads(1) public void accumulatorAccumulateAndReset_lowContention(Blackhole blackhole) { Accumulator.EmbeddingSupport.inc(accumulator, Counter.HITS); - blackhole.consume(Accumulator.EmbeddingSupport.accumulateAndReset(accumulator)); + blackhole.consume( + Accumulator.EmbeddingSupport.accumulateAndReset(accumulator, Counter.values().length)); } /** @@ -248,7 +249,8 @@ public void accumulatorAccumulateAndReset_lowContention(Blackhole blackhole) { @Threads(Threads.MAX) public void accumulatorAccumulateAndReset_highContention(Blackhole blackhole) { Accumulator.EmbeddingSupport.inc(accumulator, Counter.HITS); - blackhole.consume(Accumulator.EmbeddingSupport.accumulateAndReset(accumulator)); + blackhole.consume( + Accumulator.EmbeddingSupport.accumulateAndReset(accumulator, Counter.values().length)); } /** @@ -271,7 +273,8 @@ public void accumulatorMixed_write() { @Group("accumulatorMixed") @GroupThreads(1) public void accumulatorMixed_drain(Blackhole blackhole) { - blackhole.consume(Accumulator.EmbeddingSupport.accumulateAndReset(accumulator)); + blackhole.consume( + Accumulator.EmbeddingSupport.accumulateAndReset(accumulator, Counter.values().length)); } /** diff --git a/internal-api/src/main/java/datadog/trace/util/Accumulator.java b/internal-api/src/main/java/datadog/trace/util/Accumulator.java index 2e247b2e981..746c985a5b6 100644 --- a/internal-api/src/main/java/datadog/trace/util/Accumulator.java +++ b/internal-api/src/main/java/datadog/trace/util/Accumulator.java @@ -169,7 +169,7 @@ public void add(E key, long delta) { * @see EmbeddingSupport#accumulateAndReset */ public Counts accumulateAndReset() { - return new Counts<>(EmbeddingSupport.accumulateAndReset(data), values); + return new Counts<>(EmbeddingSupport.accumulateAndReset(data, values.length), values); } /** @@ -178,10 +178,10 @@ public Counts accumulateAndReset() { * the delta a concurrent {@link #accumulateAndReset} on a reporting cadence is about to report. * * @return the sum, keyed by the enum's {@code ordinal()} - * @see EmbeddingSupport#sum(long[][]) + * @see EmbeddingSupport#sum(long[][], int) */ public Counts sum() { - return new Counts<>(EmbeddingSupport.sum(data), values); + return new Counts<>(EmbeddingSupport.sum(data, values.length), values); } /** @@ -285,7 +285,8 @@ public Counts plus(Counts other) { * Accumulator.EmbeddingSupport.inc(stripe, MyCounters.BAR); * }); * - * long[] drained = Accumulator.EmbeddingSupport.accumulateAndReset(data); // per stripe + * long[] drained = + * Accumulator.EmbeddingSupport.accumulateAndReset(data, MyCounters.values().length); * long foo = drained[MyCounters.FOO.ordinal()]; * } */ @@ -383,15 +384,20 @@ public static void update(long[][] data, @Strategy Consumer mutator) { * {@link #inc}/{@link #add}/{@link #update} -- so no writer can land an increment in the gap * between summing and zeroing the way {@code LongAdder#sumThenReset()} allows. * - * @return a new array the same length as one stripe's row, indexed by the enum's {@code - * ordinal()} for the positions actually in use (trailing padding positions are always zero) + *

Only the first {@code width} positions of each stripe are read or written -- the trailing + * cache line {@link #paddedWidth} reserves past that point is never touched again after {@link + * #create} zero-initializes it, so it stays a genuinely dead buffer between adjacent stripe + * rows instead of being read-and-rewritten (dirtying that cache line) on every drain. + * + * @param width the number of counters actually in use, e.g. {@code values.length} + * @return a new array of length {@code width}, indexed by the enum's {@code ordinal()} */ - public static long[] accumulateAndReset(long[][] data) { - long[] acc = new long[data[0].length]; + public static long[] accumulateAndReset(long[][] data, int width) { + long[] acc = new long[width]; for (long[] stripe : data) { synchronized (stripe) { - combine(acc, stripe); - reset(stripe); + combine(acc, stripe, width); + reset(stripe, width); } } return acc; @@ -402,34 +408,38 @@ public static long[] accumulateAndReset(long[][] data) { * snapshot for a diagnostic read that must not perturb the delta a concurrent {@link * #accumulateAndReset} on a reporting cadence is about to report. * - * @return a new array the same length as one stripe's row, indexed by the enum's {@code - * ordinal()} for the positions actually in use (trailing padding positions are always zero) - * @see #accumulateAndReset(long[][]) + * @param width the number of counters actually in use, e.g. {@code values.length} + * @return a new array of length {@code width}, indexed by the enum's {@code ordinal()} + * @see #accumulateAndReset(long[][], int) */ - public static long[] sum(long[][] data) { - long[] acc = new long[data[0].length]; + public static long[] sum(long[][] data, int width) { + long[] acc = new long[width]; for (long[] stripe : data) { synchronized (stripe) { - combine(acc, stripe); + combine(acc, stripe, width); } } return acc; } /** - * {@code acc[i] += stripe[i]} for every index -- a fixed-trip-count loop C2 can auto-vectorize. + * {@code acc[i] += stripe[i]} for {@code i} in {@code [0, width)} -- a fixed-trip-count loop C2 + * can auto-vectorize. */ @GuardedBy("stripe") - private static void combine(long[] acc, long[] stripe) { - for (int i = 0; i < acc.length; i++) { + private static void combine(long[] acc, long[] stripe, int width) { + for (int i = 0; i < width; i++) { acc[i] += stripe[i]; } } - /** Zeroes every position of {@code stripe}, via the JVM-intrinsic {@link Arrays#fill}. */ + /** + * Zeroes {@code stripe}'s first {@code width} positions, via the JVM-intrinsic {@link + * Arrays#fill}. Deliberately stops at {@code width}, leaving the trailing padding untouched. + */ @GuardedBy("stripe") - private static void reset(long[] stripe) { - Arrays.fill(stripe, 0L); + private static void reset(long[] stripe, int width) { + Arrays.fill(stripe, 0, width, 0L); } /** diff --git a/internal-api/src/test/java/datadog/trace/util/AccumulatorTest.java b/internal-api/src/test/java/datadog/trace/util/AccumulatorTest.java index 5caa32549ed..a767b803476 100644 --- a/internal-api/src/test/java/datadog/trace/util/AccumulatorTest.java +++ b/internal-api/src/test/java/datadog/trace/util/AccumulatorTest.java @@ -24,7 +24,8 @@ enum Counters { @Test void freshAccumulatorSumsToZero() { long[][] data = Accumulator.EmbeddingSupport.create(Counters.values()); - long[] drained = Accumulator.EmbeddingSupport.accumulateAndReset(data); + long[] drained = + Accumulator.EmbeddingSupport.accumulateAndReset(data, Counters.values().length); for (Counters c : Counters.values()) { assertEquals(0L, drained[c.ordinal()]); } @@ -37,7 +38,8 @@ void incIncrementsByOne() { Accumulator.EmbeddingSupport.inc(data, Counters.FOO); Accumulator.EmbeddingSupport.inc(data, Counters.BAR); - long[] drained = Accumulator.EmbeddingSupport.accumulateAndReset(data); + long[] drained = + Accumulator.EmbeddingSupport.accumulateAndReset(data, Counters.values().length); assertEquals(2L, drained[Counters.FOO.ordinal()]); assertEquals(1L, drained[Counters.BAR.ordinal()]); assertEquals(0L, drained[Counters.BAZ.ordinal()]); @@ -49,7 +51,8 @@ void addAppliesArbitraryDelta() { Accumulator.EmbeddingSupport.add(data, Counters.BAZ, 41L); Accumulator.EmbeddingSupport.add(data, Counters.BAZ, 1L); - long[] drained = Accumulator.EmbeddingSupport.accumulateAndReset(data); + long[] drained = + Accumulator.EmbeddingSupport.accumulateAndReset(data, Counters.values().length); assertEquals(42L, drained[Counters.BAZ.ordinal()]); } @@ -64,7 +67,8 @@ void updateAppliesSeveralOpsUnderOneLock() { Accumulator.EmbeddingSupport.add(stripe, Counters.BAR, 5L); }); - long[] drained = Accumulator.EmbeddingSupport.accumulateAndReset(data); + long[] drained = + Accumulator.EmbeddingSupport.accumulateAndReset(data, Counters.values().length); assertEquals(2L, drained[Counters.FOO.ordinal()]); assertEquals(5L, drained[Counters.BAR.ordinal()]); } @@ -74,21 +78,22 @@ void accumulateAndResetsSoASecondDrainIsZero() { long[][] data = Accumulator.EmbeddingSupport.create(Counters.values()); Accumulator.EmbeddingSupport.inc(data, Counters.FOO); - long[] first = Accumulator.EmbeddingSupport.accumulateAndReset(data); + long[] first = Accumulator.EmbeddingSupport.accumulateAndReset(data, Counters.values().length); assertEquals(1L, first[Counters.FOO.ordinal()]); - long[] second = Accumulator.EmbeddingSupport.accumulateAndReset(data); + long[] second = Accumulator.EmbeddingSupport.accumulateAndReset(data, Counters.values().length); for (Counters c : Counters.values()) { assertEquals(0L, second[c.ordinal()]); } } @Test - void drainedRowsAreAllTheSameLength() { + void drainedArrayIsExactlyWidthLong() { long[][] data = Accumulator.EmbeddingSupport.create(Counters.values()); - long[] drained = Accumulator.EmbeddingSupport.accumulateAndReset(data); - assertEquals(data[0].length, drained.length); - assertTrue(drained.length >= Counters.values().length); + long[] drained = + Accumulator.EmbeddingSupport.accumulateAndReset(data, Counters.values().length); + assertEquals(Counters.values().length, drained.length); + assertTrue(drained.length < data[0].length, "drained array should exclude stripe padding"); } @Test @@ -122,7 +127,8 @@ void concurrentIncrementsAreNotLost() throws InterruptedException { pool.shutdown(); } - long[] drained = Accumulator.EmbeddingSupport.accumulateAndReset(data); + long[] drained = + Accumulator.EmbeddingSupport.accumulateAndReset(data, Counters.values().length); assertEquals((long) threadCount * incrementsPerThread, drained[Counters.FOO.ordinal()]); } @@ -143,7 +149,9 @@ void concurrentAccumulateAndDuringWritesNeverExceedsWritten() pool.submit( () -> { while (!stop.get()) { - long[] drained = Accumulator.EmbeddingSupport.accumulateAndReset(data); + long[] drained = + Accumulator.EmbeddingSupport.accumulateAndReset( + data, Counters.values().length); synchronized (runningTotal) { runningTotal[0] += drained[Counters.FOO.ordinal()]; } @@ -164,7 +172,8 @@ void concurrentAccumulateAndDuringWritesNeverExceedsWritten() stop.set(true); drainer.get(30, TimeUnit.SECONDS); - long[] finalDrain = Accumulator.EmbeddingSupport.accumulateAndReset(data); + long[] finalDrain = + Accumulator.EmbeddingSupport.accumulateAndReset(data, Counters.values().length); synchronized (runningTotal) { runningTotal[0] += finalDrain[Counters.FOO.ordinal()]; } @@ -180,15 +189,16 @@ void sumDoesNotResetStripes() { long[][] data = Accumulator.EmbeddingSupport.create(Counters.values()); Accumulator.EmbeddingSupport.inc(data, Counters.FOO); - long[] first = Accumulator.EmbeddingSupport.sum(data); + long[] first = Accumulator.EmbeddingSupport.sum(data, Counters.values().length); assertEquals(1L, first[Counters.FOO.ordinal()]); // sum() didn't reset anything, so a second sum() sees the same total - long[] second = Accumulator.EmbeddingSupport.sum(data); + long[] second = Accumulator.EmbeddingSupport.sum(data, Counters.values().length); assertEquals(1L, second[Counters.FOO.ordinal()]); // and a real drain afterwards still sees the value sum() didn't consume - long[] drained = Accumulator.EmbeddingSupport.accumulateAndReset(data); + long[] drained = + Accumulator.EmbeddingSupport.accumulateAndReset(data, Counters.values().length); assertEquals(1L, drained[Counters.FOO.ordinal()]); } @@ -196,10 +206,10 @@ void sumDoesNotResetStripes() { void sumReflectsIncrementsMadeAfterAnEarlierSum() { long[][] data = Accumulator.EmbeddingSupport.create(Counters.values()); Accumulator.EmbeddingSupport.inc(data, Counters.FOO); - Accumulator.EmbeddingSupport.sum(data); + Accumulator.EmbeddingSupport.sum(data, Counters.values().length); Accumulator.EmbeddingSupport.inc(data, Counters.FOO); - long[] second = Accumulator.EmbeddingSupport.sum(data); + long[] second = Accumulator.EmbeddingSupport.sum(data, Counters.values().length); assertEquals(2L, second[Counters.FOO.ordinal()]); }