From 54a2a2fdca03bb85754149a69c62ade92547df13 Mon Sep 17 00:00:00 2001 From: Douglas Q Hawkins Date: Fri, 23 Oct 2026 20:26:09 -0400 Subject: [PATCH 1/2] Replace the per-trace sampling state lock with compare-and-set Each PTags allocated a lock Object and took its monitor for every sampling decision, though the sampling state is already an immutable snapshot behind one volatile field. Publish decisions with a compare-and-set on that field instead: compute the next state from the one read, and retry if another decision got there first. A deferred otel check is computed the same way, returning the next state rather than installing it. canChangeDecisionMaker becomes volatile, cleared before the decision maker is locked so a racing decision either lands first or retries seeing it cleared. setOtelTraceState had no callers and is removed. Co-Authored-By: Claude Opus 5.5 --- .../core/propagation/ptags/PTagsFactory.java | 379 ++++++++++-------- 1 file changed, 205 insertions(+), 174 deletions(-) diff --git a/dd-trace-core/src/main/java/datadog/trace/core/propagation/ptags/PTagsFactory.java b/dd-trace-core/src/main/java/datadog/trace/core/propagation/ptags/PTagsFactory.java index 997fada2b3b..d80239d9afa 100644 --- a/dd-trace-core/src/main/java/datadog/trace/core/propagation/ptags/PTagsFactory.java +++ b/dd-trace-core/src/main/java/datadog/trace/core/propagation/ptags/PTagsFactory.java @@ -22,6 +22,7 @@ import java.util.Map; import java.util.Objects; import java.util.concurrent.atomic.AtomicIntegerFieldUpdater; +import java.util.concurrent.atomic.AtomicReferenceFieldUpdater; import javax.annotation.Nonnull; public class PTagsFactory implements PropagationTags.Factory { @@ -118,9 +119,15 @@ static class PTags extends PropagationTags { // tags that don't require any modifications and propagated as-is private final List tagPairs; - private final Object samplingStateLock = new Object(); + /** + * Sampling decisions publish a new immutable {@link SamplingState} with a compare-and-set, so a + * {@code PTags} needs no lock object and concurrent decisions never block each other. + */ + private static final AtomicReferenceFieldUpdater SAMPLING_STATE_UPDATER = + AtomicReferenceFieldUpdater.newUpdater(PTags.class, SamplingState.class, "samplingState"); - private boolean canChangeDecisionMaker; + /** Only ever cleared; read inside each decision's compare-and-set loop. */ + private volatile boolean canChangeDecisionMaker; private static final AtomicIntegerFieldUpdater TRACE_SOURCE_UPDATER = AtomicIntegerFieldUpdater.newUpdater(PTags.class, "traceSource"); @@ -266,48 +273,71 @@ static PTags withError(PTagsFactory factory, String error) { @Override public void updateTraceSamplingPriority(int samplingPriority, int samplingMechanism) { - synchronized (samplingStateLock) { - if (samplingPriority != PrioritySampling.UNSET && canChangeDecisionMaker - || samplingMechanism == SamplingMechanism.EXTERNAL_OVERRIDE) { - if (deferOtelCheck( - samplingPriority, samplingMechanism, getKnuthSamplingRateTagValue(), true)) { - return; - } + SamplingState current; + SamplingState next; + do { + current = samplingState; + if (!(samplingPriority != PrioritySampling.UNSET && canChangeDecisionMaker + || samplingMechanism == SamplingMechanism.EXTERNAL_OVERRIDE)) { + return; + } + TagValue knuthSamplingRate = getKnuthSamplingRateTagValue(current); + next = + deferOtelCheck(current, samplingPriority, samplingMechanism, knuthSamplingRate, true); + if (next == null) { OtelTraceState nextOtelTraceState = - reconcileOtelTraceState(getOtelTraceState(), samplingPriority, samplingMechanism); - installSamplingState(samplingPriority, samplingMechanism, nextOtelTraceState); + reconcileOtelTraceState( + otelTraceStateOf(current), samplingPriority, samplingMechanism); + next = + nextSamplingState( + current, + samplingPriority, + samplingMechanism, + nextOtelTraceState, + knuthSamplingRate, + true); } - } + } while (next != current && !SAMPLING_STATE_UPDATER.compareAndSet(this, current, next)); } @Override public boolean tryUpdateTraceSamplingPriority( int samplingPriority, int samplingMechanism, boolean allowOverride) { - synchronized (samplingStateLock) { - if (samplingPriority == PrioritySampling.UNSET) { - return false; - } - SamplingState current = samplingState; + if (samplingPriority == PrioritySampling.UNSET) { + return false; + } + SamplingState current; + SamplingState next; + do { + current = samplingState; if (!allowOverride && current.getSamplingPriority() != PrioritySampling.UNSET) { return false; } boolean updateDecisionMaker = canChangeDecisionMaker || samplingMechanism == SamplingMechanism.EXTERNAL_OVERRIDE; - TagValue knuthSamplingRate = getKnuthSamplingRateTagValue(); - if (deferOtelCheck( - samplingPriority, samplingMechanism, knuthSamplingRate, updateDecisionMaker)) { - return true; + TagValue knuthSamplingRate = getKnuthSamplingRateTagValue(current); + next = + deferOtelCheck( + current, + samplingPriority, + samplingMechanism, + knuthSamplingRate, + updateDecisionMaker); + if (next == null) { + OtelTraceState nextOtelTraceState = + reconcileOtelTraceState( + otelTraceStateOf(current), samplingPriority, samplingMechanism); + next = + nextSamplingState( + current, + samplingPriority, + samplingMechanism, + nextOtelTraceState, + knuthSamplingRate, + updateDecisionMaker); } - OtelTraceState nextOtelTraceState = - reconcileOtelTraceState(getOtelTraceState(), samplingPriority, samplingMechanism); - installSamplingState( - samplingPriority, - samplingMechanism, - nextOtelTraceState, - knuthSamplingRate, - updateDecisionMaker); - return true; - } + } while (next != current && !SAMPLING_STATE_UPDATER.compareAndSet(this, current, next)); + return true; } @Override @@ -318,105 +348,112 @@ public boolean tryUpdateProbabilitySamplingDecision( boolean rateLimiterRejected, long traceIdLowOrderBits, boolean allowOverride) { - synchronized (samplingStateLock) { - SamplingState current = samplingState; + TagValue nextKnuthSamplingRate = knuthSamplingRateTagValue(sampleRate); + SamplingState current; + SamplingState next; + do { + current = samplingState; if (!allowOverride && current.getSamplingPriority() != PrioritySampling.UNSET) { return false; } - TagValue nextKnuthSamplingRate = knuthSamplingRateTagValue(sampleRate); boolean updateDecisionMaker = canChangeDecisionMaker || samplingMechanism == SamplingMechanism.EXTERNAL_OVERRIDE; - OtelTraceState nextOtelTraceState = getOtelTraceState(); - if (nextOtelTraceState == null) { - if (!rateLimiterRejected) { - // A fresh decision always yields a new otel state, so there's no no-op check to make - TagValue nextDecisionMakerTagValue = - nextDecisionMakerTagValue(samplingPriority, samplingMechanism, updateDecisionMaker); - clearCachedHeader(W3C); - samplingState = - new ProbabilitySamplingState( - samplingPriority, - tracestate, - nextDecisionMakerTagValue, - nextKnuthSamplingRate, - traceIdLowOrderBits, - sampleRate); - return true; - } - } else if (rateLimiterRejected) { - nextOtelTraceState = nextOtelTraceState.withoutThreshold(); + OtelTraceState nextOtelTraceState = otelTraceStateOf(current); + if (nextOtelTraceState == null && !rateLimiterRejected) { + // A fresh decision always yields a new otel state, so there's no no-op check to make + TagValue nextDecisionMakerTagValue = + nextDecisionMakerTagValue( + current, samplingPriority, samplingMechanism, updateDecisionMaker); + clearCachedHeader(W3C); + next = + new ProbabilitySamplingState( + samplingPriority, + current.getTracestate(), + nextDecisionMakerTagValue, + nextKnuthSamplingRate, + traceIdLowOrderBits, + sampleRate); } else { - nextOtelTraceState = nextOtelTraceState.withoutInheritedThreshold(); + if (nextOtelTraceState != null) { + nextOtelTraceState = + rateLimiterRejected + ? nextOtelTraceState.withoutThreshold() + : nextOtelTraceState.withoutInheritedThreshold(); + } + next = + nextSamplingState( + current, + samplingPriority, + samplingMechanism, + nextOtelTraceState, + nextKnuthSamplingRate, + updateDecisionMaker); } - installSamplingState( - samplingPriority, - samplingMechanism, - nextOtelTraceState, - nextKnuthSamplingRate, - updateDecisionMaker); - return true; - } + } while (next != current && !SAMPLING_STATE_UPDATER.compareAndSet(this, current, next)); + return true; } @Override public void forceKeep(int samplingMechanism) { - synchronized (samplingStateLock) { - OtelTraceState nextOtelTraceState = getOtelTraceState(); + SamplingState current; + SamplingState next; + do { + current = samplingState; + OtelTraceState nextOtelTraceState = otelTraceStateOf(current); if (nextOtelTraceState != null) { nextOtelTraceState = nextOtelTraceState.forNonProbabilityDecision(); } - installSamplingState(PrioritySampling.USER_KEEP, samplingMechanism, nextOtelTraceState); - } - } - - private void installSamplingState( - int samplingPriority, int samplingMechanism, OtelTraceState nextOtelTraceState) { - installSamplingState( - samplingPriority, - samplingMechanism, - nextOtelTraceState, - getKnuthSamplingRateTagValue(), - true); + next = + nextSamplingState( + current, + PrioritySampling.USER_KEEP, + samplingMechanism, + nextOtelTraceState, + getKnuthSamplingRateTagValue(current), + true); + } while (next != current && !SAMPLING_STATE_UPDATER.compareAndSet(this, current, next)); } - @SuppressWarnings("StringEquality") - @SuppressFBWarnings( - value = "ES_COMPARING_STRINGS_WITH_EQ", - justification = - "Identity preserves the raw tracestate reference used by the cached sampling state.") - private void installSamplingState( + /** + * The state after a decision on {@code current}, or {@code current} itself when nothing would + * change. Has no effect beyond clearing header caches, so a decision whose compare-and-set + * loses can recompute it from the newer state. + */ + private SamplingState nextSamplingState( + SamplingState current, int samplingPriority, int samplingMechanism, OtelTraceState nextOtelTraceState, TagValue nextKnuthSamplingRateTagValue, boolean updateDecisionMaker) { - SamplingState currentState = samplingState; TagValue nextDecisionMakerTagValue = - nextDecisionMakerTagValue(samplingPriority, samplingMechanism, updateDecisionMaker); - if (currentState.getSamplingPriority() == samplingPriority - && currentState.getTracestate() == tracestate - && currentState.getOtelTraceState() == nextOtelTraceState - && Objects.equals(currentState.getDecisionMaker(), nextDecisionMakerTagValue) - && Objects.equals(currentState.getKnuthSamplingRate(), nextKnuthSamplingRateTagValue)) { - return; + nextDecisionMakerTagValue( + current, samplingPriority, samplingMechanism, updateDecisionMaker); + if (current.getSamplingPriority() == samplingPriority + && current.getOtelTraceState() == nextOtelTraceState + && Objects.equals(current.getDecisionMaker(), nextDecisionMakerTagValue) + && Objects.equals(current.getKnuthSamplingRate(), nextKnuthSamplingRateTagValue)) { + return current; } clearCachedHeader(W3C); - samplingState = - new SamplingState( - samplingPriority, - tracestate, - nextOtelTraceState, - nextDecisionMakerTagValue, - nextKnuthSamplingRateTagValue); + return new SamplingState( + samplingPriority, + current.getTracestate(), + nextOtelTraceState, + nextDecisionMakerTagValue, + nextKnuthSamplingRateTagValue); } /** - * The decision maker value after a decision, clearing the cached headers if it changes. Called - * with the sampling state lock held. + * The decision maker value after a decision on {@code current}, clearing the cached headers if + * it changes. */ private TagValue nextDecisionMakerTagValue( - int samplingPriority, int samplingMechanism, boolean updateDecisionMaker) { - TagValue nextDecisionMakerTagValue = getDecisionMakerTagValue(); + SamplingState current, + int samplingPriority, + int samplingMechanism, + boolean updateDecisionMaker) { + TagValue nextDecisionMakerTagValue = getDecisionMakerTagValue(current); if (updateDecisionMaker && samplingPriority > 0) { // TODO should try to keep the old sampling mechanism if we override the value? if (samplingMechanism == SamplingMechanism.EXTERNAL_OVERRIDE) { @@ -472,23 +509,25 @@ private static SamplingState initialSamplingState( * Records a decision against an extracted otel member that hasn't been parsed, without parsing * it. For an external override or unknown mechanism, reconciling the member is only a * consistency check, which can wait for the first read; a local decision rewrites the member - * and parses it. Called with the sampling state lock held. + * and parses it. Returns the state after the decision, {@code current} itself when nothing + * changes, or null when the decision can't be deferred. */ - private boolean deferOtelCheck( + private SamplingState deferOtelCheck( + SamplingState current, int samplingPriority, int samplingMechanism, TagValue knuthSamplingRate, boolean updateDecisionMaker) { - SamplingState current = samplingState; if (!(current instanceof InheritedSamplingState) || ((InheritedSamplingState) current).isResolved() || (samplingMechanism != SamplingMechanism.EXTERNAL_OVERRIDE && samplingMechanism != SamplingMechanism.UNKNOWN)) { - return false; + return null; } InheritedSamplingState inherited = (InheritedSamplingState) current; TagValue decisionMaker = - nextDecisionMakerTagValue(samplingPriority, samplingMechanism, updateDecisionMaker); + nextDecisionMakerTagValue( + current, samplingPriority, samplingMechanism, updateDecisionMaker); byte pendingChecks = InheritedSamplingState.withCheck(inherited.pendingChecks(), samplingPriority > 0); if (inherited.getSamplingPriority() == samplingPriority @@ -496,13 +535,15 @@ private boolean deferOtelCheck( && Objects.equals(inherited.getDecisionMaker(), decisionMaker) && Objects.equals(inherited.getKnuthSamplingRate(), knuthSamplingRate)) { // nothing changes, e.g. the root span re-applying the extracted priority - return true; + return current; } clearCachedHeader(W3C); - samplingState = - inherited.withDecision( - samplingPriority, tracestate, decisionMaker, knuthSamplingRate, pendingChecks); - return true; + return inherited.withDecision( + samplingPriority, + current.getTracestate(), + decisionMaker, + knuthSamplingRate, + pendingChecks); } private static OtelTraceState reconcileOtelTraceState( @@ -965,95 +1006,85 @@ public void updateW3CTracestateFrom(W3CTracestate source, int overridingPriority // as the trace context's own priority override reconciles its member pendingChecks = InheritedSamplingState.withCheck(pendingChecks, overridingPriority > 0); } - synchronized (samplingStateLock) { - clearCachedHeader(W3C); - int samplingPriority = samplingState.getSamplingPriority(); + SamplingState current; + SamplingState next; + do { + current = samplingState; + int samplingPriority = current.getSamplingPriority(); + byte nextPendingChecks = pendingChecks; if (samplingPriority != PrioritySampling.UNSET) { - pendingChecks = InheritedSamplingState.withCheck(pendingChecks, samplingPriority > 0); + nextPendingChecks = + InheritedSamplingState.withCheck(nextPendingChecks, samplingPriority > 0); } - this.tracestate = source.tracestate; - this.samplingState = + next = new InheritedSamplingState( samplingPriority, source.tracestate, - getDecisionMakerTagValue(), - getKnuthSamplingRateTagValue(), + getDecisionMakerTagValue(current), + getKnuthSamplingRateTagValue(current), source.rawOtelTraceState, source.otelOriginalSize, - pendingChecks); - } + nextPendingChecks); + } while (!SAMPLING_STATE_UPDATER.compareAndSet(this, current, next)); + this.tracestate = source.tracestate; + clearCachedHeader(W3C); } private void setW3CTracestate(String tracestate, OtelTraceState otelTraceState) { - synchronized (samplingStateLock) { - clearCachedHeader(W3C); - int samplingPriority = samplingState.getSamplingPriority(); - if (otelTraceState != null + SamplingState current; + SamplingState next; + do { + current = samplingState; + int samplingPriority = current.getSamplingPriority(); + OtelTraceState nextOtelTraceState = otelTraceState; + if (nextOtelTraceState != null && samplingPriority != PrioritySampling.UNSET - && !otelTraceState.isConsistentWith(samplingPriority > 0)) { - otelTraceState = otelTraceState.withoutThreshold(); + && !nextOtelTraceState.isConsistentWith(samplingPriority > 0)) { + nextOtelTraceState = nextOtelTraceState.withoutThreshold(); } - this.tracestate = tracestate; - this.samplingState = + next = new SamplingState( samplingPriority, tracestate, - otelTraceState, - getDecisionMakerTagValue(), - getKnuthSamplingRateTagValue()); - } + nextOtelTraceState, + getDecisionMakerTagValue(current), + getKnuthSamplingRateTagValue(current)); + } while (!SAMPLING_STATE_UPDATER.compareAndSet(this, current, next)); + this.tracestate = tracestate; + clearCachedHeader(W3C); } - private OtelTraceState getOtelTraceState() { + private static OtelTraceState otelTraceStateOf(SamplingState samplingState) { return (OtelTraceState) samplingState.getOtelTraceState(); } - @SuppressWarnings("StringEquality") - @SuppressFBWarnings( - value = "ES_COMPARING_STRINGS_WITH_EQ", - justification = - "Identity preserves the raw tracestate reference used by the cached sampling state.") - void setOtelTraceState(OtelTraceState otelTraceState) { - synchronized (samplingStateLock) { - SamplingState currentState = samplingState; - if (currentState.getTracestate() == tracestate - && currentState.getOtelTraceState() == otelTraceState) { - return; - } - clearCachedHeader(W3C); - samplingState = - new SamplingState( - currentState.getSamplingPriority(), - tracestate, - otelTraceState, - getDecisionMakerTagValue(currentState), - getKnuthSamplingRateTagValue(currentState)); - } - } - String getError() { return this.error; } @Override public void updateAndLockDecisionMaker(PropagationTags source) { - synchronized (samplingStateLock) { - if (source instanceof PTags) { - canChangeDecisionMaker = false; - TagValue decisionMakerTagValue = ((PTags) source).getDecisionMakerTagValue(); - if (decisionMakerTagValue != null) { - clearCachedHeader(DATADOG); - clearCachedHeader(W3C); - } - SamplingState currentState = samplingState; - samplingState = - new SamplingState( - currentState.getSamplingPriority(), - tracestate, - getOtelTraceState(), - decisionMakerTagValue, - getKnuthSamplingRateTagValue(currentState)); - } + if (!(source instanceof PTags)) { + return; + } + // cleared first, so a racing decision either lands before the swap below or sees it cleared + canChangeDecisionMaker = false; + TagValue decisionMakerTagValue = ((PTags) source).getDecisionMakerTagValue(); + SamplingState current; + SamplingState next; + do { + current = samplingState; + next = + new SamplingState( + current.getSamplingPriority(), + current.getTracestate(), + otelTraceStateOf(current), + decisionMakerTagValue, + getKnuthSamplingRateTagValue(current)); + } while (!SAMPLING_STATE_UPDATER.compareAndSet(this, current, next)); + if (decisionMakerTagValue != null) { + clearCachedHeader(DATADOG); + clearCachedHeader(W3C); } } } From f82a285384cd3e5468a9dfac309e77040984102b Mon Sep 17 00:00:00 2001 From: Douglas Q Hawkins Date: Fri, 23 Oct 2026 20:26:12 -0400 Subject: [PATCH 2/2] Test that concurrent sampling decisions don't lose updates Co-Authored-By: Claude Opus 5.5 --- .../ptags/ConcurrentSamplingDecisionTest.java | 89 +++++++++++++++++++ 1 file changed, 89 insertions(+) create mode 100644 dd-trace-core/src/test/java/datadog/trace/core/propagation/ptags/ConcurrentSamplingDecisionTest.java diff --git a/dd-trace-core/src/test/java/datadog/trace/core/propagation/ptags/ConcurrentSamplingDecisionTest.java b/dd-trace-core/src/test/java/datadog/trace/core/propagation/ptags/ConcurrentSamplingDecisionTest.java new file mode 100644 index 00000000000..ea3ebd776e3 --- /dev/null +++ b/dd-trace-core/src/test/java/datadog/trace/core/propagation/ptags/ConcurrentSamplingDecisionTest.java @@ -0,0 +1,89 @@ +package datadog.trace.core.propagation.ptags; + +import static datadog.trace.api.sampling.PrioritySampling.SAMPLER_KEEP; +import static datadog.trace.api.sampling.PrioritySampling.USER_KEEP; +import static datadog.trace.api.sampling.SamplingMechanism.AGENT_RATE; +import static datadog.trace.api.sampling.SamplingMechanism.MANUAL; +import static datadog.trace.core.propagation.PropagationTags.HeaderType.DATADOG; +import static org.junit.jupiter.api.Assertions.assertEquals; + +import datadog.trace.core.propagation.PropagationTags; +import java.util.concurrent.CountDownLatch; +import java.util.concurrent.ExecutorService; +import java.util.concurrent.Executors; +import java.util.concurrent.Future; +import org.junit.jupiter.api.AfterEach; +import org.junit.jupiter.api.Test; + +/** + * Concurrent decisions on one trace's tags must behave as if they ran one at a time, in some order. + * Each trial races two decisions whose outcome is the same in either order, so a lost update shows + * up as a different result. + */ +class ConcurrentSamplingDecisionTest { + private static final int TRIALS = 2_000; + + private final PropagationTags.Factory factory = PropagationTags.factory(); + private final ExecutorService executor = Executors.newFixedThreadPool(2); + + @AfterEach + void shutdown() { + executor.shutdownNow(); + } + + /** A manual keep overrides a sampler decision made before it and blocks one made after it. */ + @Test + void forceKeepWinsOverRacingSamplerDecision() throws Exception { + for (int trial = 0; trial < TRIALS; ++trial) { + PropagationTags ptags = factory.empty(); + + race( + () -> ptags.forceKeep(MANUAL), + () -> + ptags.tryUpdateProbabilitySamplingDecision( + SAMPLER_KEEP, AGENT_RATE, 0.5, false, 1L, false)); + + assertEquals(USER_KEEP, ptags.getSamplingPriority(), "trial " + trial); + assertEquals("-4", String.valueOf(ptags.samplingState().getDecisionMaker())); + } + } + + /** + * Locking the decision maker wins over a racing decision: one made before it is overwritten, and + * one made after it sees the decision maker locked. The racing decision's own decision maker + * ({@code -1}) differs from the locked one, so a decision that slips in after the lock shows. + */ + @Test + void lockedDecisionMakerWinsOverRacingDecision() throws Exception { + PropagationTags source = factory.fromHeaderValue(DATADOG, "_dd.p.dm=-3"); + for (int trial = 0; trial < TRIALS; ++trial) { + PropagationTags ptags = factory.empty(); + + race( + () -> ptags.updateAndLockDecisionMaker(source), + () -> ptags.updateTraceSamplingPriority(USER_KEEP, AGENT_RATE)); + + assertEquals( + "-3", String.valueOf(ptags.samplingState().getDecisionMaker()), "trial " + trial); + } + } + + private void race(Runnable first, Runnable second) throws Exception { + CountDownLatch start = new CountDownLatch(1); + Future a = executor.submit(() -> runAfter(start, first)); + Future b = executor.submit(() -> runAfter(start, second)); + start.countDown(); + a.get(); + b.get(); + } + + private static void runAfter(CountDownLatch start, Runnable action) { + try { + start.await(); + } catch (InterruptedException e) { + Thread.currentThread().interrupt(); + return; + } + action.run(); + } +}