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..5bf142edfa8 --- /dev/null +++ b/internal-api/src/jmh/java/datadog/trace/util/AccumulatorBenchmark.java @@ -0,0 +1,230 @@ +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.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; +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 the alternatives it actually displaces: a single {@code LongAdder} (the + * collision-free baseline it can never beat, only approach), an independent {@code LongAdder} per + * counter guarded by a per-counter lock (the "just fix it with LongAdder" natural migration target + * -- {@code longAdderGroup*}), and the {@code ConcurrentHashMap.computeIfAbsent(key, k -> new + * AtomicLong())} anti-pattern ({@code chmAtomicLongIncrement*}) that {@link Accumulator} exists to + * avoid. The CHM variant allocates its counter under the bucket's bin lock the first time its one + * constant key is seen, 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. + * + *

{@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). With this benchmark's single counter, that per-counter lock collapses to + * one lock shared by every thread -- no thread-based distribution at all -- so it loses badly on + * the write path against {@link Accumulator}'s thread-sharded stripes, especially under contention. + * This is the realistic production baseline this class was built to replace (see {@code + * TracerHealthMetrics}'s pre-migration design, one {@code LongAdder} field per counter): + * AccumulatorBenchmark.accumulatorIncrement_lowContention avgt 6 0.010 ± 0.001 us/op + * AccumulatorBenchmark.accumulatorIncrement_highContention avgt 6 0.025 ± 0.039 us/op + * AccumulatorBenchmark.longAdderGroupIncrement_lowContention avgt 6 0.012 ± 0.001 us/op + * AccumulatorBenchmark.longAdderGroupIncrement_highContention avgt 6 1.178 ± 0.134 us/op + * AccumulatorBenchmark.accumulatorAccumulateAndReset_lowContention avgt 6 0.104 ± 0.001 us/op + * AccumulatorBenchmark.accumulatorAccumulateAndReset_highContention avgt 6 13.357 ± 1.203 us/op + * AccumulatorBenchmark.longAdderGroupAccumulateAnd_lowContention avgt 6 0.024 ± 0.001 us/op + * AccumulatorBenchmark.longAdderGroupAccumulateAnd_highContention avgt 6 2.439 ± 0.337 us/op + * At high contention, {@link Accumulator} beats the realistic {@code longAdderGroup} + * baseline by nearly 50x on increment (the call that runs on every event), but is itself + * roughly 5.5x worse than {@code longAdderGroup} on drain under that same high-contention + * topology (the call that runs once per reporting cycle) -- {@code accumulateAndReset} walks every + * stripe with a full {@code getAndSet} per counter, so more stripes (sized for core count) means + * more per-drain work than {@code longAdderGroup}'s one-lock-per-counter {@code sumThenReset}. This + * is still a clean win once weighted by call-site frequency -- the increment win is ~50x on a call + * that fires on every event, the drain loss is ~5.5x on a call that fires once per reporting cycle + * (e.g. a 30s flush tick) -- but the drain-side regression is real, not "slightly worse," and worth + * knowing before assuming this trade is free in every topology. + */ +@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 Accumulator accumulator = Accumulator.of(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 reset hazard {@link Accumulator} does, but stripes by + * counter (one lock per enum constant) instead of by thread (one shared table + * across 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) + 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(Counter.HITS); + } + + @Benchmark + @Threads(Threads.MAX) + public void accumulatorIncrement_highContention() { + accumulator.inc(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 accumulatorAccumulateAndReset_lowContention(Blackhole blackhole) { + accumulator.inc(Counter.HITS); + blackhole.consume(accumulator.accumulateAndReset()); + } + + /** + * 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#accumulateAndReset} than this. + */ + @Benchmark + @Threads(Threads.MAX) + public void accumulatorAccumulateAndReset_highContention(Blackhole blackhole) { + accumulator.inc(Counter.HITS); + blackhole.consume(accumulator.accumulateAndReset()); + } + + /** + * The realistic counterpart to {@code accumulatorAccumulateAndReset_highContention}: many writer + * threads incrementing, and a single dedicated thread polling {@link + * Accumulator#accumulateAndReset} -- not every thread doing both on every op. {@code + * accumulatorMixed-write} measures increment cost while a drain is actively running; {@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.inc(Counter.HITS); + } + + @Benchmark + @Group("accumulatorMixed") + @GroupThreads(1) + public void accumulatorMixed_drain(Blackhole blackhole) { + blackhole.consume(accumulator.accumulateAndReset()); + } + + @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)); + } +} 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..bb830b4acce --- /dev/null +++ b/internal-api/src/main/java/datadog/trace/util/Accumulator.java @@ -0,0 +1,210 @@ +package datadog.trace.util; + +import datadog.environment.ThreadSupport; +import java.util.concurrent.atomic.AtomicLongArray; + +/** + * A striped, lock-free counter primitive keyed by enum ordinal: {@code LongAdder}'s write + * scalability, but as one shared, thread-sharded table instead of one independent {@code LongAdder} + * per counter -- which avoids paying {@code LongAdder}'s per-instance striping overhead {@code + * E.values().length} times over. + * + * 

{@code
+ * enum MyCounters { FOO, BAR }
+ *
+ * Accumulator counters = Accumulator.of(MyCounters.values());
+ * counters.inc(MyCounters.FOO);
+ * counters.add(MyCounters.BAR, 5L);
+ *
+ * Accumulator.Counts drained = counters.accumulateAndReset();
+ * long foo = drained.get(MyCounters.FOO);
+ * }
+ * + *

Each counter's own {@link #accumulateAndReset} slot is read-and-zeroed with a single atomic + * {@code getAndSet}, so -- like {@code Accumulator}'s previous {@code synchronized}-stripe design, + * and unlike {@code LongAdder#sumThenReset()} -- no individual increment can land in the gap + * between summing and zeroing and be silently lost. What's gone is the previous design's + * row-wide atomicity: {@link #inc}/{@link #add} for two different counters are no longer + * guaranteed to be seen together by a concurrent {@link #accumulateAndReset}. There is no {@code + * update}-style escape hatch for grouping several counters under one atomic operation -- callers + * needing that must weigh whether the guarantee was load-bearing (most call sites are logging + * unrelated aspects of the same event, not maintaining a cross-counter invariant a reader depends + * on) or bring their own coordination. + */ +public final class Accumulator> { + /** One full cache line of {@code long}s (64 bytes), used to pad each stripe row. */ + private static final int CACHE_LINE_LONGS = 8; + + private final AtomicLongArray[] data; + private final int width; + private final E[] values; + + private Accumulator(AtomicLongArray[] data, int width, E[] values) { + this.data = data; + this.width = width; + this.values = values; + } + + /** + * @param values the enum constants naming each counter, e.g. {@code MyCounters.values()} + */ + public static > Accumulator of(E[] values) { + int width = values.length; + int paddedWidth = paddedWidth(width); + int stripes = stripeCount(); + AtomicLongArray[] data = new AtomicLongArray[stripes]; + for (int i = 0; i < stripes; i++) { + data[i] = new AtomicLongArray(paddedWidth); + } + return new Accumulator<>(data, width, 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. */ + public void inc(E key) { + add(key, 1L); + } + + /** Adds {@code delta} to the counter named by {@code key} in the calling thread's stripe. */ + public void add(E key, long delta) { + stripeOf(data).getAndAdd(key.ordinal(), delta); + } + + /** + * Combines and resets every stripe, returning the sum as a typed view. Each counter is + * read-and-zeroed with one atomic {@code getAndSet} -- see the class-level note on what atomicity + * this does and doesn't provide across different counters. + * + * @return the sum, keyed by the enum's {@code ordinal()} + */ + public Counts accumulateAndReset() { + long[] acc = new long[width]; + for (AtomicLongArray stripe : data) { + for (int i = 0; i < width; i++) { + acc[i] += stripe.getAndSet(i, 0L); + } + } + return new Counts<>(acc, values); + } + + /** + * 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()} + */ + public Counts sum() { + long[] acc = new long[width]; + for (AtomicLongArray stripe : data) { + for (int i = 0; i < width; i++) { + acc[i] += stripe.get(i); + } + } + return new Counts<>(acc, values); + } + + /** + * A typed view over a drained snapshot, returned by {@link #accumulateAndReset} or {@link #sum}: + * the same enum-ordinal type checking {@link Accumulator} provides on writes, applied to the read + * side too. + */ + public static final class Counts> { + private final long[] counts; + private final E[] values; + + private Counts(long[] counts, E[] values) { + this.counts = counts; + this.values = values; + } + + /** + * 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], 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}. */ + 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[] keys() { + 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 + * 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, values); + } + } + + /** + * 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 + * counter's own atomic slot makes that safe, just not maximally scalable under a hash collision. + */ + private static AtomicLongArray stripeOf(AtomicLongArray[] 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 CAS-retry/cache-line-bounce + * cost, the same problem {@code LongAdder}'s own {@code Cell[]} table exists to avoid. 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; + } +} 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..a8f2f6f39d8 --- /dev/null +++ b/internal-api/src/test/java/datadog/trace/util/AccumulatorFootprintTest.java @@ -0,0 +1,140 @@ +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. + * + *

{@link Accumulator}'s stripe count is fixed at creation (roughly 2x {@link + * Runtime#availableProcessors()}, minimum 4) and does not grow further as more contention arrives + * within it, while every additional concurrently-written {@code LongAdder} keeps paying its own + * {@code Cell[]} growth cost independently. {@code Accumulator}'s up-front cost is the more + * predictable one: fixed at creation, independent of runtime contention, and shared (one striped + * table) across however many counters the caller's enum declares, rather than paid per counter. The + * printed numbers below vary by run/JVM -- see the assertions for the invariants that actually + * matter. + */ +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(); + Accumulator accumulator = Accumulator.of(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.min(16, 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); + Accumulator accumulator = Accumulator.of(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"); + 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 new file mode 100644 index 00000000000..f17eab823fb --- /dev/null +++ b/internal-api/src/test/java/datadog/trace/util/AccumulatorTest.java @@ -0,0 +1,241 @@ +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.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; + +class AccumulatorTest { + + enum Counters { + FOO, + BAR, + BAZ + } + + @Test + void freshAccumulatorSumsToZero() { + Accumulator counters = Accumulator.of(Counters.values()); + Accumulator.Counts drained = counters.accumulateAndReset(); + for (Counters c : Counters.values()) { + assertEquals(0L, drained.get(c)); + } + } + + @Test + void incIncrementsByOne() { + Accumulator counters = Accumulator.of(Counters.values()); + counters.inc(Counters.FOO); + counters.inc(Counters.FOO); + counters.inc(Counters.BAR); + + Accumulator.Counts drained = counters.accumulateAndReset(); + assertEquals(2L, drained.get(Counters.FOO)); + assertEquals(1L, drained.get(Counters.BAR)); + assertEquals(0L, drained.get(Counters.BAZ)); + } + + @Test + void addAppliesArbitraryDelta() { + Accumulator counters = Accumulator.of(Counters.values()); + counters.add(Counters.BAZ, 41L); + counters.add(Counters.BAZ, 1L); + + Accumulator.Counts drained = counters.accumulateAndReset(); + assertEquals(42L, drained.get(Counters.BAZ)); + } + + @Test + void accumulateAndResetsSoASecondDrainIsZero() { + Accumulator counters = Accumulator.of(Counters.values()); + counters.inc(Counters.FOO); + + Accumulator.Counts first = counters.accumulateAndReset(); + assertEquals(1L, first.get(Counters.FOO)); + + Accumulator.Counts second = counters.accumulateAndReset(); + for (Counters c : Counters.values()) { + assertEquals(0L, second.get(c)); + } + } + + @Test + void concurrentIncrementsAreNotLost() throws InterruptedException { + Accumulator counters = Accumulator.of(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++) { + counters.inc(Counters.FOO); + } + } catch (InterruptedException e) { + Thread.currentThread().interrupt(); + } finally { + done.countDown(); + } + }); + } + start.countDown(); + assertTrue(done.await(30, TimeUnit.SECONDS)); + } finally { + pool.shutdown(); + } + + Accumulator.Counts drained = counters.accumulateAndReset(); + assertEquals((long) threadCount * incrementsPerThread, drained.get(Counters.FOO)); + } + + @Test + void concurrentAccumulateAndDuringWritesNeverExceedsWritten() + throws InterruptedException, ExecutionException, TimeoutException { + Accumulator counters = Accumulator.of(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 { + Future drainer = + pool.submit( + () -> { + while (!stop.get()) { + Accumulator.Counts drained = counters.accumulateAndReset(); + synchronized (runningTotal) { + runningTotal[0] += drained.get(Counters.FOO); + } + } + }); + + for (int t = 0; t < threadCount; t++) { + pool.execute( + () -> { + for (int i = 0; i < incrementsPerThread; i++) { + counters.inc(Counters.FOO); + } + done.countDown(); + }); + } + + assertTrue(done.await(30, TimeUnit.SECONDS)); + stop.set(true); + drainer.get(30, TimeUnit.SECONDS); + + Accumulator.Counts finalDrain = counters.accumulateAndReset(); + synchronized (runningTotal) { + runningTotal[0] += finalDrain.get(Counters.FOO); + } + + assertEquals((long) threadCount * incrementsPerThread, runningTotal[0]); + } finally { + pool.shutdown(); + } + } + + @Test + void sumDoesNotResetStripes() { + Accumulator counters = Accumulator.of(Counters.values()); + counters.inc(Counters.FOO); + + Accumulator.Counts first = counters.sum(); + assertEquals(1L, first.get(Counters.FOO)); + + // sum() didn't reset anything, so a second sum() sees the same total + Accumulator.Counts second = counters.sum(); + assertEquals(1L, second.get(Counters.FOO)); + + // and a real drain afterwards still sees the value sum() didn't consume + Accumulator.Counts drained = counters.accumulateAndReset(); + assertEquals(1L, drained.get(Counters.FOO)); + } + + @Test + void sumReflectsIncrementsMadeAfterAnEarlierSum() { + Accumulator counters = Accumulator.of(Counters.values()); + counters.inc(Counters.FOO); + counters.sum(); + + counters.inc(Counters.FOO); + Accumulator.Counts second = counters.sum(); + assertEquals(2L, second.get(Counters.FOO)); + } + + @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 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.keys().length); + + long total = 0L; + for (Counters c : drained.keys()) { + total += drained.get(c); + } + assertEquals(6L, total); + } + + @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)); + } +}