From 06fbaf0ed969ca54e5acd4d4ef49a26318214518 Mon Sep 17 00:00:00 2001 From: Andrea Marziali Date: Mon, 26 Oct 2026 19:28:49 +0100 Subject: [PATCH 1/4] Fix carrier starvation in scope diagnostics --- .../test/scopediag/ScopeDiagnostics.java | 220 +++++++++++------- .../ScopeDiagnosticsAdmissionTest.java | 106 +++++++++ 2 files changed, 236 insertions(+), 90 deletions(-) diff --git a/dd-java-agent/instrumentation-testing/src/main/java/datadog/trace/agent/test/scopediag/ScopeDiagnostics.java b/dd-java-agent/instrumentation-testing/src/main/java/datadog/trace/agent/test/scopediag/ScopeDiagnostics.java index 11b1f8c95f2..5dd65d96fa1 100644 --- a/dd-java-agent/instrumentation-testing/src/main/java/datadog/trace/agent/test/scopediag/ScopeDiagnostics.java +++ b/dd-java-agent/instrumentation-testing/src/main/java/datadog/trace/agent/test/scopediag/ScopeDiagnostics.java @@ -9,6 +9,9 @@ import java.util.IdentityHashMap; import java.util.Map; import java.util.Set; +import java.util.concurrent.ConcurrentLinkedQueue; +import java.util.concurrent.locks.LockSupport; +import java.util.function.Consumer; import org.slf4j.Logger; import org.slf4j.LoggerFactory; @@ -21,9 +24,14 @@ public final class ScopeDiagnostics { private static final ScopeDiagnostics INSTANCE = new ScopeDiagnostics(); - /** Linearizes event admission, stop/reset, and report snapshots. */ + /** Serializes event consumption, stop/reset, and report snapshots. */ private final Object lifecycleLock = new Object(); + // Callbacks also run on virtual-thread carriers. They must never wait for a lock whose owner + // may be an unmounted virtual thread waiting for one of those same carriers. + private final ConcurrentLinkedQueue pendingEvents = new ConcurrentLinkedQueue<>(); + private final ConcurrentLinkedQueue quiescenceWaiters = new ConcurrentLinkedQueue<>(); + private final Map records = new IdentityHashMap<>(); private final Map scopeRecords = new IdentityHashMap<>(); private final Set deferredCleanupScopes = @@ -36,7 +44,7 @@ public final class ScopeDiagnostics { private long seq; private long scopeSeq; private volatile Object recordingWindow; - private StackFilter stackFilter = new StackFilter(DEFAULT_MAX_FRAMES); + private volatile StackFilter stackFilter = new StackFilter(DEFAULT_MAX_FRAMES); private final Listener listener = new Listener(); @@ -70,13 +78,16 @@ public static void startRecording(int maxFrames) { INSTANCE.stackFilter = new StackFilter(maxFrames); INSTANCE.recordingWindow = new Object(); } + INSTANCE.wakeQuiescenceWaiters(); } /** Stops recording. Recorded data remains queryable until reset. */ public static void stop() { synchronized (INSTANCE.lifecycleLock) { + INSTANCE.drainEvents(); INSTANCE.recordingWindow = null; } + INSTANCE.wakeQuiescenceWaiters(); } /** Discards all recorded data. */ @@ -85,9 +96,10 @@ public static void reset() { INSTANCE.recordingWindow = null; INSTANCE.clear(); } + INSTANCE.wakeQuiescenceWaiters(); } - /** Each probe keeps this identity until admission under the lifecycle lock. */ + /** Each probe keeps this identity so delayed callbacks cannot enter a later recording. */ static Object recordingWindow() { return INSTANCE.recordingWindow; } @@ -95,6 +107,7 @@ static Object recordingWindow() { /** Returns an immutable snapshot of the events recorded so far. */ public static ScopeDiagnosticsReport report() { synchronized (INSTANCE.lifecycleLock) { + INSTANCE.drainEvents(); return INSTANCE.snapshot(); } } @@ -111,21 +124,26 @@ public static void awaitQuiescence(long timeoutMillis) { "Invalid scope diagnostics quiescence budget: " + timeoutMillis); } long deadline = System.nanoTime() + timeoutMillis * 1_000_000L; - synchronized (INSTANCE.lifecycleLock) { - while (INSTANCE.recordingWindow != null && INSTANCE.snapshot().hasIncompleteLifecycles()) { + Thread waiter = Thread.currentThread(); + INSTANCE.quiescenceWaiters.add(waiter); + try { + while (!waiter.isInterrupted()) { + synchronized (INSTANCE.lifecycleLock) { + INSTANCE.drainEvents(); + if (INSTANCE.recordingWindow == null || !INSTANCE.snapshot().hasIncompleteLifecycles()) { + return; + } + } long remaining = deadline - System.nanoTime(); if (remaining <= 0) { return; } - try { - long millis = remaining / 1_000_000L; - int nanos = (int) (remaining % 1_000_000L); - INSTANCE.lifecycleLock.wait(millis, nanos); - } catch (InterruptedException interrupted) { - Thread.currentThread().interrupt(); - return; - } + // Registration precedes the predicate check; an enqueue racing with park retains its + // permit. + LockSupport.parkNanos(INSTANCE, remaining); } + } finally { + INSTANCE.quiescenceWaiters.remove(waiter); } } @@ -187,6 +205,7 @@ static EnumSet enabledChecks(TrackScopeContinuations conf } private void clear() { + pendingEvents.clear(); records.clear(); scopeRecords.clear(); deferredCleanupScopes.clear(); @@ -206,10 +225,6 @@ private ScopeDiagnosticsReport snapshot() { private static final StackTraceElement[] NO_STACK = new StackTraceElement[0]; - private ScopeEvent event(ScopeEvent.Type type) { - return event(type, System.nanoTime()); - } - /** Uses the supplied event time while capturing the thread and stack at the call site. */ private ScopeEvent event(ScopeEvent.Type type, long nanos) { // Avoid stack walking when call sites are disabled because it perturbs recorded timings. @@ -217,10 +232,39 @@ private ScopeEvent event(ScopeEvent.Type type, long nanos) { StackTraceElement[] stack = filter.maxFrames() <= 0 ? NO_STACK : filter.filter(new Throwable().getStackTrace()); ScopeEvent event = new ScopeEvent(type, Thread.currentThread().getName(), nanos, stack); - lifecycleLock.notifyAll(); return event; } + private static void record( + Object window, ScopeEvent.Type type, long nanos, Consumer handler) { + if (window == null || window != INSTANCE.recordingWindow) { + return; + } + // Capture the originating thread and call site before handing the event to the report reader. + ScopeEvent event = type == null ? null : INSTANCE.event(type, nanos); + INSTANCE.pendingEvents.add( + () -> { + if (window == INSTANCE.recordingWindow) { + handler.accept(event); + } + }); + INSTANCE.wakeQuiescenceWaiters(); + } + + private void wakeQuiescenceWaiters() { + for (Thread waiter : quiescenceWaiters) { + LockSupport.unpark(waiter); + } + } + + /** Only recording controls and report readers consume events, under the lifecycle lock. */ + private void drainEvents() { + Runnable event; + while ((event = pendingEvents.poll()) != null) { + event.run(); + } + } + static void recordCapture( Object window, ContextContinuation id, @@ -228,11 +272,11 @@ static void recordCapture( long spanId, String spanName, byte source) { - synchronized (INSTANCE.lifecycleLock) { - if (window != null && window == INSTANCE.recordingWindow) { - INSTANCE.listener.onCapture(id, traceId, spanId, spanName, source); - } - } + record( + window, + ScopeEvent.Type.CAPTURE, + System.nanoTime(), + event -> INSTANCE.listener.onCapture(id, traceId, spanId, spanName, source, event)); } static void recordActivate( @@ -243,19 +287,19 @@ static void recordActivate( String spanName, byte source, long nanos) { - synchronized (INSTANCE.lifecycleLock) { - if (window != null && window == INSTANCE.recordingWindow) { - INSTANCE.listener.onActivate(id, traceId, spanId, spanName, source, nanos); - } - } + record( + window, + ScopeEvent.Type.ACTIVATE, + nanos, + event -> INSTANCE.listener.onActivate(id, traceId, spanId, spanName, source, event)); } static void recordActivateFailed(Object window, ContextContinuation id) { - synchronized (INSTANCE.lifecycleLock) { - if (window != null && window == INSTANCE.recordingWindow) { - INSTANCE.listener.onActivateFailed(id); - } - } + record( + window, + ScopeEvent.Type.ACTIVATE_FAILED, + System.nanoTime(), + event -> INSTANCE.listener.onActivateFailed(id, event)); } static void recordResolve( @@ -264,24 +308,22 @@ static void recordResolve( boolean release, long resolveNanos, boolean duplicateAttempt) { - synchronized (INSTANCE.lifecycleLock) { - if (window != null && window == INSTANCE.recordingWindow) { - boolean firstResolution = INSTANCE.resolved.add(id); - boolean repeatedRelease = release && !INSTANCE.released.add(id); - if (duplicateAttempt || firstResolution || repeatedRelease) { - INSTANCE.listener.onResolve( - id, release, resolveNanos, duplicateAttempt || repeatedRelease); - } - } - } + record( + window, + release ? ScopeEvent.Type.RESOLVE_RELEASE : ScopeEvent.Type.RESOLVE_FINISH, + resolveNanos, + event -> { + boolean firstResolution = INSTANCE.resolved.add(id); + boolean repeatedRelease = release && !INSTANCE.released.add(id); + if (duplicateAttempt || firstResolution || repeatedRelease) { + INSTANCE.listener.onResolve(id, duplicateAttempt || repeatedRelease, event); + } + }); } static void recordRootWritten(Object window, DDTraceId traceId) { - synchronized (INSTANCE.lifecycleLock) { - if (window != null && window == INSTANCE.recordingWindow) { - INSTANCE.listener.onRootWritten(traceId); - } - } + long nanos = System.nanoTime(); + record(window, null, nanos, ignored -> INSTANCE.listener.onRootWritten(traceId, nanos)); } static void recordScopeOpen( @@ -292,44 +334,46 @@ static void recordScopeOpen( String spanName, byte source, ContextContinuation continuation) { - synchronized (INSTANCE.lifecycleLock) { - if (window != null && window == INSTANCE.recordingWindow) { - INSTANCE.listener.onScopeOpen(scope, traceId, spanId, spanName, source, continuation); - } - } + record( + window, + ScopeEvent.Type.SCOPE_OPEN, + System.nanoTime(), + event -> + INSTANCE.listener.onScopeOpen( + scope, traceId, spanId, spanName, source, continuation, event)); } static void recordScopeClose(Object window, Object scope) { - synchronized (INSTANCE.lifecycleLock) { - if (window != null && window == INSTANCE.recordingWindow) { - INSTANCE.listener.onScopeClose(scope); - } - } + record( + window, + ScopeEvent.Type.SCOPE_CLOSE, + System.nanoTime(), + event -> INSTANCE.listener.onScopeClose(scope, event)); } static void recordScopeCloseWrongThread(Object window, Object scope) { - synchronized (INSTANCE.lifecycleLock) { - if (window != null && window == INSTANCE.recordingWindow) { - INSTANCE.listener.onScopeCloseWrongThread(scope); - } - } + record( + window, + ScopeEvent.Type.SCOPE_CLOSE_WRONG_THREAD, + System.nanoTime(), + event -> INSTANCE.listener.onScopeCloseWrongThread(scope, event)); } static void recordDeferredScopeCleanup(Object window, Object scope) { - synchronized (INSTANCE.lifecycleLock) { - if (window != null && window == INSTANCE.recordingWindow) { - INSTANCE.listener.onDeferredScopeCleanup(scope); - } - } + record(window, null, 0, ignored -> INSTANCE.listener.onDeferredScopeCleanup(scope)); } private final class Listener { void onCapture( - ContextContinuation id, DDTraceId traceId, long spanId, String spanName, byte source) { + ContextContinuation id, + DDTraceId traceId, + long spanId, + String spanName, + byte source, + ScopeEvent event) { try { ContinuationRecord record = - new ContinuationRecord( - seq++, traceId, spanId, spanName, source, false, event(ScopeEvent.Type.CAPTURE)); + new ContinuationRecord(seq++, traceId, spanId, spanName, source, false, event); records.put(id, record); } catch (Throwable ignored) { // diagnostics must never disturb the tracer @@ -342,37 +386,32 @@ void onActivate( long spanId, String spanName, byte source, - long nanos) { + ScopeEvent event) { try { - recordFor(id, traceId, spanId, spanName, source) - .addResume(event(ScopeEvent.Type.ACTIVATE, nanos)); + recordFor(id, traceId, spanId, spanName, source).addResume(event); } catch (Throwable ignored) { } } - void onActivateFailed(ContextContinuation id) { + void onActivateFailed(ContextContinuation id, ScopeEvent event) { try { // The probe observed cancellation; capture and resolution need not be in this window. - recordFor(id, DDTraceId.ZERO, 0, null, (byte) -1) - .addFailedActivation(event(ScopeEvent.Type.ACTIVATE_FAILED)); + recordFor(id, DDTraceId.ZERO, 0, null, (byte) -1).addFailedActivation(event); } catch (Throwable ignored) { } } - void onResolve( - ContextContinuation id, boolean release, long resolveNanos, boolean duplicateAttempt) { + void onResolve(ContextContinuation id, boolean duplicateAttempt, ScopeEvent event) { try { - ScopeEvent.Type type = - release ? ScopeEvent.Type.RESOLVE_RELEASE : ScopeEvent.Type.RESOLVE_FINISH; recordFor(id, DDTraceId.ZERO, 0, null, (byte) -1) - .setTerminalOrExtra(event(type, resolveNanos), duplicateAttempt); + .setTerminalOrExtra(event, duplicateAttempt); } catch (Throwable ignored) { } } - void onRootWritten(DDTraceId traceId) { + void onRootWritten(DDTraceId traceId, long nanos) { try { - rootWrittenNanos.putIfAbsent(traceId, System.nanoTime()); + rootWrittenNanos.putIfAbsent(traceId, nanos); } catch (Throwable ignored) { } } @@ -383,7 +422,8 @@ void onScopeOpen( long spanId, String spanName, byte source, - ContextContinuation continuation) { + ContextContinuation continuation, + ScopeEvent event) { try { if (scopeRecords.containsKey(scope)) { return; // re-activation of an already-open scope, not a new open @@ -401,7 +441,7 @@ void onScopeOpen( source, continuationSeq, deferredCleanupScopes.remove(scope), - event(ScopeEvent.Type.SCOPE_OPEN))); + event)); if (owner != null) { owner.linkScope(s); } @@ -409,21 +449,21 @@ void onScopeOpen( } } - void onScopeClose(Object scope) { + void onScopeClose(Object scope, ScopeEvent event) { try { ScopeRecord record = scopeRecords.get(scope); if (record != null) { - record.setClose(event(ScopeEvent.Type.SCOPE_CLOSE)); + record.setClose(event); } } catch (Throwable ignored) { } } - void onScopeCloseWrongThread(Object scope) { + void onScopeCloseWrongThread(Object scope, ScopeEvent event) { try { ScopeRecord record = scopeRecords.get(scope); if (record != null) { - record.addWrongThreadClose(event(ScopeEvent.Type.SCOPE_CLOSE_WRONG_THREAD)); + record.addWrongThreadClose(event); } } catch (Throwable ignored) { } diff --git a/dd-java-agent/instrumentation-testing/src/test/java/datadog/trace/agent/test/scopediag/ScopeDiagnosticsAdmissionTest.java b/dd-java-agent/instrumentation-testing/src/test/java/datadog/trace/agent/test/scopediag/ScopeDiagnosticsAdmissionTest.java index 3766bcb6711..c70796c6147 100644 --- a/dd-java-agent/instrumentation-testing/src/test/java/datadog/trace/agent/test/scopediag/ScopeDiagnosticsAdmissionTest.java +++ b/dd-java-agent/instrumentation-testing/src/test/java/datadog/trace/agent/test/scopediag/ScopeDiagnosticsAdmissionTest.java @@ -1,5 +1,6 @@ package datadog.trace.agent.test.scopediag; +import static java.util.concurrent.TimeUnit.SECONDS; import static org.junit.jupiter.api.Assertions.assertEquals; import static org.junit.jupiter.api.Assertions.assertFalse; import static org.junit.jupiter.api.Assertions.assertTrue; @@ -8,6 +9,10 @@ import datadog.context.ContextContinuation; import datadog.trace.api.DDTraceId; import datadog.trace.bootstrap.instrumentation.api.NoopScope; +import datadog.trace.test.util.PollingConditions; +import java.lang.reflect.Field; +import java.util.concurrent.FutureTask; +import java.util.concurrent.locks.LockSupport; import org.junit.jupiter.api.AfterEach; import org.junit.jupiter.api.Test; import org.tabletest.junit.TableTest; @@ -18,6 +23,107 @@ void reset() { ScopeDiagnostics.reset(); } + @Test + void callbacksDoNotWaitForTheReportLock() throws Exception { + ScopeDiagnostics.startRecording(); + Object window = ScopeDiagnostics.recordingWindow(); + ContextContinuation continuation = Context.root().capture(); + Object scope = new Object(); + DDTraceId trace = DDTraceId.from(1); + FutureTask callbacks = + new FutureTask<>( + () -> { + ScopeDiagnostics.recordCapture(window, continuation, trace, 2, "op", (byte) 0); + ScopeDiagnostics.recordActivate( + window, continuation, trace, 2, "op", (byte) 0, System.nanoTime()); + ScopeDiagnostics.recordScopeOpen( + window, scope, trace, 2, "op", (byte) 0, continuation); + ScopeDiagnostics.recordDeferredScopeCleanup(window, scope); + ScopeDiagnostics.recordScopeCloseWrongThread(window, scope); + ScopeDiagnostics.recordScopeClose(window, scope); + ScopeDiagnostics.recordResolve(window, continuation, true, System.nanoTime(), false); + ScopeDiagnostics.recordRootWritten(window, trace); + ScopeDiagnostics.recordActivateFailed(window, continuation); + return null; + }); + Thread carrier = new Thread(callbacks, "diagnostic-carrier"); + carrier.setDaemon(true); + Field lock = ScopeDiagnostics.class.getDeclaredField("lifecycleLock"); + lock.setAccessible(true); + synchronized (lock.get(instance())) { + carrier.start(); + // A carrier must return even when the diagnostic monitor's owner cannot run yet. + callbacks.get(5, SECONDS); + } + ScopeDiagnostics.stop(); + + ScopeDiagnosticsReport report = ScopeDiagnostics.report(); + assertEquals(1, report.records().size()); + ContinuationRecord record = report.records().get(0); + assertEquals("diagnostic-carrier", record.capture().threadName); + assertEquals(1, record.resumes().size()); + assertEquals("diagnostic-carrier", record.resumes().get(0).threadName); + assertEquals(ContinuationStatus.RELEASED, record.status()); + assertEquals(1, record.failedActivations().size()); + assertEquals(1, report.scopeRecords().size()); + ScopeRecord scopeRecord = report.scopeRecords().get(0); + assertTrue(scopeRecord.closed()); + assertEquals("diagnostic-carrier", scopeRecord.open().threadName); + assertEquals("diagnostic-carrier", scopeRecord.close().threadName); + assertEquals(1, scopeRecord.wrongThreadCloses().size()); + assertEquals(0, report.lateCount()); + assertEquals(0, report.leakCount()); + } + + @Test + void queuedResolutionWakesQuiescenceWaiter() throws Exception { + ScopeDiagnostics.startRecording(); + Object window = ScopeDiagnostics.recordingWindow(); + ContextContinuation continuation = Context.root().capture(); + ScopeDiagnostics.recordCapture(window, continuation, DDTraceId.from(1), 2, "op", (byte) 0); + FutureTask waiting = + new FutureTask<>( + () -> { + ScopeDiagnostics.awaitQuiescence(30_000); + return null; + }); + Thread waiter = new Thread(waiting, "diagnostic-quiescence"); + waiter.setDaemon(true); + waiter.start(); + try { + Object diagnostics = instance(); + new PollingConditions(5) + .eventually(() -> assertEquals(diagnostics, LockSupport.getBlocker(waiter))); + ScopeDiagnostics.recordResolve(window, continuation, true, System.nanoTime(), false); + waiting.get(5, SECONDS); + assertFalse(ScopeDiagnostics.report().hasIncompleteLifecycles()); + } finally { + waiter.interrupt(); + waiter.join(5000); + } + } + + @Test + void resetDiscardsQueuedEvents() { + ScopeDiagnostics.startRecording(); + ScopeDiagnostics.recordCapture( + ScopeDiagnostics.recordingWindow(), + Context.root().capture(), + DDTraceId.from(1), + 2, + "old", + (byte) 0); + ScopeDiagnostics.reset(); + ScopeDiagnostics.startRecording(); + assertTrue(ScopeDiagnostics.report().records().isEmpty()); + } + + private static Object instance() throws Exception { + Field instance = ScopeDiagnostics.class.getDeclaredField("INSTANCE"); + instance.setAccessible(true); + return instance.get(null); + } + @Test void failedActivationBeforeResolutionAdviceIsRetained() { ScopeDiagnostics.startRecording(); From 0911528ce2743619bf51773117acacdaec11a920 Mon Sep 17 00:00:00 2001 From: Andrea Marziali Date: Mon, 26 Oct 2026 19:46:20 +0100 Subject: [PATCH 2/4] suggestions --- .../test/scopediag/ScopeDiagnostics.java | 35 ++++- .../ScopeDiagnosticsAdmissionTest.java | 125 ++++++++++++++++++ 2 files changed, 154 insertions(+), 6 deletions(-) diff --git a/dd-java-agent/instrumentation-testing/src/main/java/datadog/trace/agent/test/scopediag/ScopeDiagnostics.java b/dd-java-agent/instrumentation-testing/src/main/java/datadog/trace/agent/test/scopediag/ScopeDiagnostics.java index 5dd65d96fa1..f400d7fd374 100644 --- a/dd-java-agent/instrumentation-testing/src/main/java/datadog/trace/agent/test/scopediag/ScopeDiagnostics.java +++ b/dd-java-agent/instrumentation-testing/src/main/java/datadog/trace/agent/test/scopediag/ScopeDiagnostics.java @@ -86,6 +86,7 @@ public static void stop() { synchronized (INSTANCE.lifecycleLock) { INSTANCE.drainEvents(); INSTANCE.recordingWindow = null; + INSTANCE.pendingEvents.clear(); } INSTANCE.wakeQuiescenceWaiters(); } @@ -129,7 +130,7 @@ public static void awaitQuiescence(long timeoutMillis) { try { while (!waiter.isInterrupted()) { synchronized (INSTANCE.lifecycleLock) { - INSTANCE.drainEvents(); + INSTANCE.drainEvents(deadline); if (INSTANCE.recordingWindow == null || !INSTANCE.snapshot().hasIncompleteLifecycles()) { return; } @@ -242,13 +243,19 @@ private static void record( } // Capture the originating thread and call site before handing the event to the report reader. ScopeEvent event = type == null ? null : INSTANCE.event(type, nanos); - INSTANCE.pendingEvents.add( + Runnable pending = () -> { if (window == INSTANCE.recordingWindow) { handler.accept(event); } - }); - INSTANCE.wakeQuiescenceWaiters(); + }; + INSTANCE.pendingEvents.add(pending); + // Reset/stop may have cleared the queue while this callback was constructing its event. + if (window != INSTANCE.recordingWindow) { + INSTANCE.pendingEvents.remove(pending); + } else { + INSTANCE.wakeQuiescenceWaiters(); + } } private void wakeQuiescenceWaiters() { @@ -259,12 +266,28 @@ private void wakeQuiescenceWaiters() { /** Only recording controls and report readers consume events, under the lifecycle lock. */ private void drainEvents() { - Runnable event; - while ((event = pendingEvents.poll()) != null) { + drainEvents(0); + } + + /** A zero deadline drains the published prefix without a time budget. */ + private void drainEvents(long deadline) { + // Each drain needs its own boundary: a timed-out drain may leave its marker in the queue. + Runnable boundary = new DrainBoundary(); + pendingEvents.add(boundary); + while (deadline == 0 || System.nanoTime() - deadline < 0) { + Runnable event = pendingEvents.poll(); + if (event == null || event == boundary) { + return; + } event.run(); } } + private static final class DrainBoundary implements Runnable { + @Override + public void run() {} + } + static void recordCapture( Object window, ContextContinuation id, diff --git a/dd-java-agent/instrumentation-testing/src/test/java/datadog/trace/agent/test/scopediag/ScopeDiagnosticsAdmissionTest.java b/dd-java-agent/instrumentation-testing/src/test/java/datadog/trace/agent/test/scopediag/ScopeDiagnosticsAdmissionTest.java index c70796c6147..01f9bca4fad 100644 --- a/dd-java-agent/instrumentation-testing/src/test/java/datadog/trace/agent/test/scopediag/ScopeDiagnosticsAdmissionTest.java +++ b/dd-java-agent/instrumentation-testing/src/test/java/datadog/trace/agent/test/scopediag/ScopeDiagnosticsAdmissionTest.java @@ -11,7 +11,10 @@ import datadog.trace.bootstrap.instrumentation.api.NoopScope; import datadog.trace.test.util.PollingConditions; import java.lang.reflect.Field; +import java.util.concurrent.ConcurrentLinkedQueue; import java.util.concurrent.FutureTask; +import java.util.concurrent.atomic.AtomicBoolean; +import java.util.concurrent.atomic.AtomicInteger; import java.util.concurrent.locks.LockSupport; import org.junit.jupiter.api.AfterEach; import org.junit.jupiter.api.Test; @@ -118,6 +121,128 @@ void resetDiscardsQueuedEvents() { assertTrue(ScopeDiagnostics.report().records().isEmpty()); } + @TableTest({ + "Scenario | Reset", + "reset | true ", + "stop | false" + }) + void staleProducerCannotRetainEventsAfterRecordingEnds(boolean reset) throws Exception { + ScopeDiagnostics.startRecording(); + Object window = ScopeDiagnostics.recordingWindow(); + Field queueField = ScopeDiagnostics.class.getDeclaredField("pendingEvents"); + queueField.setAccessible(true); + Object diagnostics = instance(); + Object originalQueue = queueField.get(diagnostics); + AtomicBoolean firstAdd = new AtomicBoolean(true); + ConcurrentLinkedQueue delayedQueue = + new ConcurrentLinkedQueue() { + @Override + public boolean add(Runnable event) { + if (firstAdd.getAndSet(false)) { + // End recording after the callback's window check but before publication. + if (reset) { + ScopeDiagnostics.reset(); + } else { + ScopeDiagnostics.stop(); + } + } + return super.add(event); + } + }; + try { + queueField.set(diagnostics, delayedQueue); + ScopeDiagnostics.recordCapture( + window, Context.root().capture(), DDTraceId.from(1), 2, "stale", (byte) 0); + // Do not call report/reset here: either would hide a stale item retained in the static queue. + assertTrue(delayedQueue.isEmpty()); + } finally { + queueField.set(diagnostics, originalQueue); + } + } + + @TableTest({ + "Operation ", + "report ", + "stop ", + "quiescence" + }) + void drainingDoesNotChaseNewEvents(String operation) throws Exception { + ScopeDiagnostics.startRecording(); + ScopeDiagnostics.recordCapture( + ScopeDiagnostics.recordingWindow(), + Context.root().capture(), + DDTraceId.from(1), + 2, + "incomplete", + (byte) 0); + ConcurrentLinkedQueue queue = pendingEvents(); + AtomicBoolean producing = new AtomicBoolean(true); + AtomicInteger consumed = new AtomicInteger(); + Runnable event = + new Runnable() { + @Override + public void run() { + consumed.incrementAndGet(); + if (producing.get()) { + queue.add(this); + } + } + }; + queue.add(event); + FutureTask reader = + new FutureTask<>( + () -> { + switch (operation) { + case "report": + ScopeDiagnostics.report(); + break; + case "stop": + ScopeDiagnostics.stop(); + break; + case "quiescence": + ScopeDiagnostics.awaitQuiescence(20); + break; + default: + throw new AssertionError(operation); + } + return null; + }); + Thread consumer = new Thread(reader, "diagnostic-reader"); + consumer.setDaemon(true); + consumer.start(); + try { + reader.get(5, SECONDS); + assertTrue(consumed.get() > 0); + } finally { + producing.set(false); + consumer.join(5000); + } + } + + @Test + void expiredQuiescenceBudgetLeavesBacklogForTheReport() throws Exception { + ScopeDiagnostics.startRecording(); + ScopeDiagnostics.recordCapture( + ScopeDiagnostics.recordingWindow(), + Context.root().capture(), + DDTraceId.from(1), + 2, + "incomplete", + (byte) 0); + ScopeDiagnostics.awaitQuiescence(0); + assertFalse(pendingEvents().isEmpty()); + ScopeDiagnosticsReport report = ScopeDiagnostics.report(); + assertEquals(1, report.leakCount()); + assertTrue(pendingEvents().isEmpty()); + } + + @SuppressWarnings("unchecked") + private static ConcurrentLinkedQueue pendingEvents() throws Exception { + Field queue = ScopeDiagnostics.class.getDeclaredField("pendingEvents"); + queue.setAccessible(true); + return (ConcurrentLinkedQueue) queue.get(instance()); + } + private static Object instance() throws Exception { Field instance = ScopeDiagnostics.class.getDeclaredField("INSTANCE"); instance.setAccessible(true); From 94557e0be5939004ee02bf387505663b5a358349 Mon Sep 17 00:00:00 2001 From: Andrea Marziali Date: Mon, 26 Oct 2026 20:57:27 +0100 Subject: [PATCH 3/4] more suggestions --- .../test/scopediag/DiagnosticEventBuffer.java | 96 ++++++++ .../test/scopediag/ScopeDiagnostics.java | 218 ++++++++++++------ .../scopediag/DiagnosticEventBufferTest.java | 131 +++++++++++ .../ScopeDiagnosticsAdmissionTest.java | 164 ++++--------- 4 files changed, 420 insertions(+), 189 deletions(-) create mode 100644 dd-java-agent/instrumentation-testing/src/main/java/datadog/trace/agent/test/scopediag/DiagnosticEventBuffer.java create mode 100644 dd-java-agent/instrumentation-testing/src/test/java/datadog/trace/agent/test/scopediag/DiagnosticEventBufferTest.java diff --git a/dd-java-agent/instrumentation-testing/src/main/java/datadog/trace/agent/test/scopediag/DiagnosticEventBuffer.java b/dd-java-agent/instrumentation-testing/src/main/java/datadog/trace/agent/test/scopediag/DiagnosticEventBuffer.java new file mode 100644 index 00000000000..bd26dae76f6 --- /dev/null +++ b/dd-java-agent/instrumentation-testing/src/main/java/datadog/trace/agent/test/scopediag/DiagnosticEventBuffer.java @@ -0,0 +1,96 @@ +package datadog.trace.agent.test.scopediag; + +import java.util.concurrent.atomic.AtomicReference; +import java.util.concurrent.locks.LockSupport; + +/** + * Ordered callback reservations with a single consumer protected by the diagnostic lifecycle lock. + */ +final class DiagnosticEventBuffer { + private static final Runnable NOOP = () -> {}; + private static final Event CLOSED = new Event(); + + private Event consumed = new Event(); + // A successful CAS reserves FIFO order before stack capture; the volatile next/action writes + // publish the link and payload separately. Only the consumer touches consumed and draining. + private final AtomicReference tail = new AtomicReference<>(consumed); + private boolean draining; + + Event reserve() { + Event event = new Event(); + Event previous; + do { + previous = tail.get(); + if (previous == CLOSED) { + return null; + } + } while (!tail.compareAndSet(previous, event)); + previous.next = event; + return event; + } + + /** Stops admission atomically; all successful reservations remain available to the consumer. */ + void close() { + Event previous = tail.getAndSet(CLOSED); + if (previous != CLOSED) { + previous.next = CLOSED; + } + } + + boolean hasPending() { + Event last = tail.get(); + return consumed != last && consumed.next != CLOSED; + } + + boolean isNext(Event event) { + return !draining && consumed.next == event; + } + + /** Drains one finite prefix; opportunistic consumers never wait for a producer. */ + void drain(boolean timed, long deadline, boolean waitForPublication) { + if (draining) { + return; + } + draining = true; + try { + Event boundary = tail.get(); + while (consumed != boundary && (!timed || System.nanoTime() - deadline < 0)) { + Event event = consumed.next; + if (event == CLOSED) { + return; + } + Runnable action = event == null ? null : event.action; + if (action == null) { + if (!waitForPublication) { + return; + } + // Only control/report threads wait. Callback threads always use non-waiting drains and + // tryLock, so holding the lifecycle lock here cannot block a carrier callback. + LockSupport.parkNanos(this, 100_000); + continue; + } + consumed.next = null; + consumed = event; + event.action = NOOP; + action.run(); + } + } finally { + draining = false; + } + } + + static final class Event { + private volatile Event next; + private volatile Runnable action; + + void publish(Runnable action) { + this.action = action; + } + + void publishIfAbsent() { + if (action == null) { + action = NOOP; + } + } + } +} diff --git a/dd-java-agent/instrumentation-testing/src/main/java/datadog/trace/agent/test/scopediag/ScopeDiagnostics.java b/dd-java-agent/instrumentation-testing/src/main/java/datadog/trace/agent/test/scopediag/ScopeDiagnostics.java index f400d7fd374..cbf8bd6a67c 100644 --- a/dd-java-agent/instrumentation-testing/src/main/java/datadog/trace/agent/test/scopediag/ScopeDiagnostics.java +++ b/dd-java-agent/instrumentation-testing/src/main/java/datadog/trace/agent/test/scopediag/ScopeDiagnostics.java @@ -11,7 +11,9 @@ import java.util.Set; import java.util.concurrent.ConcurrentLinkedQueue; import java.util.concurrent.locks.LockSupport; +import java.util.concurrent.locks.ReentrantLock; import java.util.function.Consumer; +import java.util.function.Supplier; import org.slf4j.Logger; import org.slf4j.LoggerFactory; @@ -22,14 +24,13 @@ public final class ScopeDiagnostics { private static final int DEFAULT_MAX_FRAMES = 6; private static final long DEFAULT_QUIESCENCE_TIMEOUT_MILLIS = 250; - private static final ScopeDiagnostics INSTANCE = new ScopeDiagnostics(); + static final ScopeDiagnostics INSTANCE = new ScopeDiagnostics(); /** Serializes event consumption, stop/reset, and report snapshots. */ - private final Object lifecycleLock = new Object(); + final ReentrantLock lifecycleLock = new ReentrantLock(); // Callbacks also run on virtual-thread carriers. They must never wait for a lock whose owner // may be an unmounted virtual thread waiting for one of those same carriers. - private final ConcurrentLinkedQueue pendingEvents = new ConcurrentLinkedQueue<>(); private final ConcurrentLinkedQueue quiescenceWaiters = new ConcurrentLinkedQueue<>(); private final Map records = new IdentityHashMap<>(); @@ -43,7 +44,9 @@ public final class ScopeDiagnostics { Collections.newSetFromMap(new IdentityHashMap()); private long seq; private long scopeSeq; - private volatile Object recordingWindow; + private long incompleteContinuations; + private long incompleteScopes; + private volatile DiagnosticEventBuffer recordingWindow; private volatile StackFilter stackFilter = new StackFilter(DEFAULT_MAX_FRAMES); private final Listener listener = new Listener(); @@ -73,29 +76,40 @@ public static void startRecording() { /** Clears any prior data and starts recording, keeping up to {@code maxFrames} per stack. */ public static void startRecording(int maxFrames) { ScopeContinuationTransformer.install(); - synchronized (INSTANCE.lifecycleLock) { + INSTANCE.lifecycleLock.lock(); + try { INSTANCE.clear(); INSTANCE.stackFilter = new StackFilter(maxFrames); - INSTANCE.recordingWindow = new Object(); + INSTANCE.recordingWindow = new DiagnosticEventBuffer(); + } finally { + INSTANCE.lifecycleLock.unlock(); } INSTANCE.wakeQuiescenceWaiters(); } /** Stops recording. Recorded data remains queryable until reset. */ public static void stop() { - synchronized (INSTANCE.lifecycleLock) { - INSTANCE.drainEvents(); + INSTANCE.lifecycleLock.lock(); + try { + DiagnosticEventBuffer window = INSTANCE.recordingWindow; INSTANCE.recordingWindow = null; - INSTANCE.pendingEvents.clear(); + if (window != null) { + window.close(); + window.drain(false, 0, true); + } + } finally { + INSTANCE.lifecycleLock.unlock(); } INSTANCE.wakeQuiescenceWaiters(); } /** Discards all recorded data. */ public static void reset() { - synchronized (INSTANCE.lifecycleLock) { - INSTANCE.recordingWindow = null; + INSTANCE.lifecycleLock.lock(); + try { INSTANCE.clear(); + } finally { + INSTANCE.lifecycleLock.unlock(); } INSTANCE.wakeQuiescenceWaiters(); } @@ -107,9 +121,12 @@ static Object recordingWindow() { /** Returns an immutable snapshot of the events recorded so far. */ public static ScopeDiagnosticsReport report() { - synchronized (INSTANCE.lifecycleLock) { + INSTANCE.lifecycleLock.lock(); + try { INSTANCE.drainEvents(); return INSTANCE.snapshot(); + } finally { + INSTANCE.lifecycleLock.unlock(); } } @@ -129,11 +146,14 @@ public static void awaitQuiescence(long timeoutMillis) { INSTANCE.quiescenceWaiters.add(waiter); try { while (!waiter.isInterrupted()) { - synchronized (INSTANCE.lifecycleLock) { - INSTANCE.drainEvents(deadline); - if (INSTANCE.recordingWindow == null || !INSTANCE.snapshot().hasIncompleteLifecycles()) { + INSTANCE.lifecycleLock.lock(); + try { + INSTANCE.drainEvents(true, deadline, false); + if (INSTANCE.recordingWindow == null || !INSTANCE.hasIncompleteLifecycles()) { return; } + } finally { + INSTANCE.lifecycleLock.unlock(); } long remaining = deadline - System.nanoTime(); if (remaining <= 0) { @@ -206,7 +226,11 @@ static EnumSet enabledChecks(TrackScopeContinuations conf } private void clear() { - pendingEvents.clear(); + DiagnosticEventBuffer window = recordingWindow; + recordingWindow = null; + if (window != null) { + window.close(); + } records.clear(); scopeRecords.clear(); deferredCleanupScopes.clear(); @@ -215,6 +239,8 @@ private void clear() { released.clear(); seq = 0; scopeSeq = 0; + incompleteContinuations = 0; + incompleteScopes = 0; } private ScopeDiagnosticsReport snapshot() { @@ -226,34 +252,69 @@ private ScopeDiagnosticsReport snapshot() { private static final StackTraceElement[] NO_STACK = new StackTraceElement[0]; - /** Uses the supplied event time while capturing the thread and stack at the call site. */ - private ScopeEvent event(ScopeEvent.Type type, long nanos) { - // Avoid stack walking when call sites are disabled because it perturbs recorded timings. - StackFilter filter = stackFilter; + private ScopeEvent event( + ScopeEvent.Type type, long nanos, String threadName, StackFilter filter, Throwable callSite) { StackTraceElement[] stack = - filter.maxFrames() <= 0 ? NO_STACK : filter.filter(new Throwable().getStackTrace()); - ScopeEvent event = new ScopeEvent(type, Thread.currentThread().getName(), nanos, stack); - return event; + callSite == null ? NO_STACK : filter.filter(callSite.getStackTrace()); + return new ScopeEvent(type, threadName, nanos, stack); } private static void record( - Object window, ScopeEvent.Type type, long nanos, Consumer handler) { - if (window == null || window != INSTANCE.recordingWindow) { + Object identity, ScopeEvent.Type type, long nanos, Consumer> handler) { + DiagnosticEventBuffer window = INSTANCE.recordingWindow; + if (window == null || window != identity) { + return; + } + // Reservation is admission: closing the buffer atomically refuses all later callbacks. + // Reserve before collecting a stack so slow stack walks cannot reorder lifecycle events. + DiagnosticEventBuffer.Event pending = window.reserve(); + if (pending == null) { return; } - // Capture the originating thread and call site before handing the event to the report reader. - ScopeEvent event = type == null ? null : INSTANCE.event(type, nanos); - Runnable pending = - () -> { + StackFilter filter = INSTANCE.stackFilter; + String threadName = Thread.currentThread().getName(); + try { + if (INSTANCE.lifecycleLock.tryLock()) { + try { if (window == INSTANCE.recordingWindow) { - handler.accept(event); + window.drain(false, 0, false); + if (window.isNext(pending)) { + // The common path filters ignored callbacks before even constructing a Throwable. + pending.publish( + () -> + handler.accept( + () -> + INSTANCE.event( + type, + nanos, + threadName, + filter, + filter.maxFrames() <= 0 ? null : new Throwable()))); + window.drain(false, 0, false); + return; + } } - }; - INSTANCE.pendingEvents.add(pending); - // Reset/stop may have cleared the queue while this callback was constructing its event. - if (window != INSTANCE.recordingWindow) { - INSTANCE.pendingEvents.remove(pending); - } else { + } finally { + INSTANCE.lifecycleLock.unlock(); + } + } + // A contended callback must preserve its originating stack, but materialization and filtering + // can wait until the consumer knows this event will actually be retained. + Throwable callSite = type == null || filter.maxFrames() <= 0 ? null : new Throwable(); + pending.publish( + () -> handler.accept(() -> INSTANCE.event(type, nanos, threadName, filter, callSite))); + if (INSTANCE.lifecycleLock.tryLock()) { + try { + if (window == INSTANCE.recordingWindow) { + window.drain(false, 0, false); + } + } finally { + INSTANCE.lifecycleLock.unlock(); + } + } + } finally { + // Even a failed stack capture must release the reservation to a stopping/reporting reader. + pending.publishIfAbsent(); INSTANCE.wakeQuiescenceWaiters(); } } @@ -264,28 +325,22 @@ private void wakeQuiescenceWaiters() { } } - /** Only recording controls and report readers consume events, under the lifecycle lock. */ private void drainEvents() { - drainEvents(0); - } - - /** A zero deadline drains the published prefix without a time budget. */ - private void drainEvents(long deadline) { - // Each drain needs its own boundary: a timed-out drain may leave its marker in the queue. - Runnable boundary = new DrainBoundary(); - pendingEvents.add(boundary); - while (deadline == 0 || System.nanoTime() - deadline < 0) { - Runnable event = pendingEvents.poll(); - if (event == null || event == boundary) { - return; - } - event.run(); + drainEvents(false, 0, true); + } + + private void drainEvents(boolean timed, long deadline, boolean waitForPublication) { + DiagnosticEventBuffer window = recordingWindow; + if (window != null) { + window.drain(timed, deadline, waitForPublication); } } - private static final class DrainBoundary implements Runnable { - @Override - public void run() {} + /** Inspects live state without copying every record and call stack on every wakeup. */ + private boolean hasIncompleteLifecycles() { + return incompleteContinuations != 0 + || incompleteScopes != 0 + || (recordingWindow != null && recordingWindow.hasPending()); } static void recordCapture( @@ -393,11 +448,14 @@ void onCapture( long spanId, String spanName, byte source, - ScopeEvent event) { + Supplier event) { try { ContinuationRecord record = - new ContinuationRecord(seq++, traceId, spanId, spanName, source, false, event); - records.put(id, record); + new ContinuationRecord(seq++, traceId, spanId, spanName, source, false, event.get()); + ContinuationRecord previous = records.put(id, record); + if (previous == null || previous.orphan || previous.isResolved()) { + incompleteContinuations++; + } } catch (Throwable ignored) { // diagnostics must never disturb the tracer } @@ -409,25 +467,34 @@ void onActivate( long spanId, String spanName, byte source, - ScopeEvent event) { + Supplier event) { try { - recordFor(id, traceId, spanId, spanName, source).addResume(event); + recordFor(id, traceId, spanId, spanName, source).addResume(event.get()); } catch (Throwable ignored) { } } - void onActivateFailed(ContextContinuation id, ScopeEvent event) { + void onActivateFailed(ContextContinuation id, Supplier event) { try { // The probe observed cancellation; capture and resolution need not be in this window. - recordFor(id, DDTraceId.ZERO, 0, null, (byte) -1).addFailedActivation(event); + ContinuationRecord record = recordFor(id, DDTraceId.ZERO, 0, null, (byte) -1); + ScopeEvent retained = event.get(); + if (!record.orphan && !record.isResolved()) { + incompleteContinuations--; + } + record.addFailedActivation(retained); } catch (Throwable ignored) { } } - void onResolve(ContextContinuation id, boolean duplicateAttempt, ScopeEvent event) { + void onResolve(ContextContinuation id, boolean duplicateAttempt, Supplier event) { try { - recordFor(id, DDTraceId.ZERO, 0, null, (byte) -1) - .setTerminalOrExtra(event, duplicateAttempt); + ContinuationRecord record = recordFor(id, DDTraceId.ZERO, 0, null, (byte) -1); + ScopeEvent retained = event.get(); + if (!record.orphan && !record.isResolved()) { + incompleteContinuations--; + } + record.setTerminalOrExtra(retained, duplicateAttempt); } catch (Throwable ignored) { } } @@ -446,7 +513,7 @@ void onScopeOpen( String spanName, byte source, ContextContinuation continuation, - ScopeEvent event) { + Supplier event) { try { if (scopeRecords.containsKey(scope)) { return; // re-activation of an already-open scope, not a new open @@ -464,7 +531,10 @@ void onScopeOpen( source, continuationSeq, deferredCleanupScopes.remove(scope), - event)); + event.get())); + if (!scopeRecords.get(scope).deferredCleanup()) { + incompleteScopes++; + } if (owner != null) { owner.linkScope(s); } @@ -472,21 +542,24 @@ void onScopeOpen( } } - void onScopeClose(Object scope, ScopeEvent event) { + void onScopeClose(Object scope, Supplier event) { try { ScopeRecord record = scopeRecords.get(scope); - if (record != null) { - record.setClose(event); + if (record != null && !record.closed()) { + record.setClose(event.get()); + if (!record.deferredCleanup()) { + incompleteScopes--; + } } } catch (Throwable ignored) { } } - void onScopeCloseWrongThread(Object scope, ScopeEvent event) { + void onScopeCloseWrongThread(Object scope, Supplier event) { try { ScopeRecord record = scopeRecords.get(scope); if (record != null) { - record.addWrongThreadClose(event); + record.addWrongThreadClose(event.get()); } } catch (Throwable ignored) { } @@ -496,6 +569,9 @@ void onDeferredScopeCleanup(Object scope) { try { ScopeRecord record = scopeRecords.get(scope); if (record != null) { + if (!record.closed() && !record.deferredCleanup()) { + incompleteScopes--; + } record.markDeferredCleanup(); } else { deferredCleanupScopes.add(scope); diff --git a/dd-java-agent/instrumentation-testing/src/test/java/datadog/trace/agent/test/scopediag/DiagnosticEventBufferTest.java b/dd-java-agent/instrumentation-testing/src/test/java/datadog/trace/agent/test/scopediag/DiagnosticEventBufferTest.java new file mode 100644 index 00000000000..9b1d8f937fb --- /dev/null +++ b/dd-java-agent/instrumentation-testing/src/test/java/datadog/trace/agent/test/scopediag/DiagnosticEventBufferTest.java @@ -0,0 +1,131 @@ +package datadog.trace.agent.test.scopediag; + +import static java.util.concurrent.TimeUnit.SECONDS; +import static org.junit.jupiter.api.Assertions.assertEquals; +import static org.junit.jupiter.api.Assertions.assertFalse; +import static org.junit.jupiter.api.Assertions.assertNull; +import static org.junit.jupiter.api.Assertions.assertTrue; + +import java.util.ArrayList; +import java.util.Arrays; +import java.util.List; +import java.util.concurrent.CountDownLatch; +import java.util.concurrent.FutureTask; +import java.util.concurrent.atomic.AtomicInteger; +import org.junit.jupiter.api.Test; + +class DiagnosticEventBufferTest { + @Test + void stackCaptureCompletionCannotReorderReservations() { + DiagnosticEventBuffer buffer = new DiagnosticEventBuffer(); + List consumed = new ArrayList<>(); + DiagnosticEventBuffer.Event finish = buffer.reserve(); + DiagnosticEventBuffer.Event release = buffer.reserve(); + release.publish(() -> consumed.add("release")); + buffer.drain(false, 0, false); + assertTrue(consumed.isEmpty()); + finish.publish(() -> consumed.add("finish")); + buffer.drain(false, 0, false); + assertEquals(Arrays.asList("finish", "release"), consumed); + assertFalse(buffer.hasPending()); + } + + @Test + void closeRetainsAdmittedEventsAndRefusesLateProducers() { + DiagnosticEventBuffer buffer = new DiagnosticEventBuffer(); + AtomicInteger consumed = new AtomicInteger(); + DiagnosticEventBuffer.Event event = buffer.reserve(); + buffer.close(); + assertNull(buffer.reserve()); + event.publish(consumed::incrementAndGet); + buffer.drain(false, 0, true); + buffer.close(); + assertNull(buffer.reserve()); + assertEquals(1, consumed.get()); + assertFalse(buffer.hasPending()); + } + + @Test + void drainDoesNotChaseNewEvents() { + DiagnosticEventBuffer buffer = new DiagnosticEventBuffer(); + AtomicInteger consumed = new AtomicInteger(); + buffer + .reserve() + .publish( + () -> { + consumed.incrementAndGet(); + buffer.reserve().publish(consumed::incrementAndGet); + }); + buffer.drain(false, 0, false); + assertEquals(1, consumed.get()); + assertTrue(buffer.hasPending()); + buffer.drain(false, 0, false); + assertEquals(2, consumed.get()); + } + + @Test + void timedOutDrainPreservesBacklog() { + DiagnosticEventBuffer buffer = new DiagnosticEventBuffer(); + AtomicInteger consumed = new AtomicInteger(); + buffer.reserve().publish(consumed::incrementAndGet); + buffer.drain(true, System.nanoTime(), false); + assertEquals(0, consumed.get()); + assertTrue(buffer.hasPending()); + buffer.drain(false, 0, false); + assertEquals(1, consumed.get()); + } + + @Test + void failedProducerDoesNotStrandLaterEvents() { + DiagnosticEventBuffer buffer = new DiagnosticEventBuffer(); + AtomicInteger consumed = new AtomicInteger(); + DiagnosticEventBuffer.Event failed = buffer.reserve(); + buffer.reserve().publish(consumed::incrementAndGet); + failed.publishIfAbsent(); + buffer.drain(false, 0, true); + assertEquals(1, consumed.get()); + assertFalse(buffer.hasPending()); + } + + @Test + void concurrentProducersRacingCloseLoseNoReservations() throws Exception { + DiagnosticEventBuffer buffer = new DiagnosticEventBuffer(); + AtomicInteger admitted = new AtomicInteger(); + AtomicInteger consumed = new AtomicInteger(); + CountDownLatch started = new CountDownLatch(4); + List> producers = new ArrayList<>(); + for (int i = 0; i < 4; i++) { + FutureTask producer = + new FutureTask<>( + () -> { + for (int n = 0; n < 1000; n++) { + DiagnosticEventBuffer.Event event = buffer.reserve(); + if (event == null) { + break; + } + admitted.incrementAndGet(); + if (n == 0) { + started.countDown(); + } + event.publish(consumed::incrementAndGet); + } + return null; + }); + producers.add(producer); + Thread thread = new Thread(producer, "diagnostic-producer-" + i); + thread.setDaemon(true); + thread.start(); + } + try { + assertTrue(started.await(5, SECONDS)); + } finally { + buffer.close(); + } + for (FutureTask producer : producers) { + producer.get(5, SECONDS); + } + buffer.drain(false, 0, true); + assertEquals(admitted.get(), consumed.get()); + assertFalse(buffer.hasPending()); + } +} diff --git a/dd-java-agent/instrumentation-testing/src/test/java/datadog/trace/agent/test/scopediag/ScopeDiagnosticsAdmissionTest.java b/dd-java-agent/instrumentation-testing/src/test/java/datadog/trace/agent/test/scopediag/ScopeDiagnosticsAdmissionTest.java index 01f9bca4fad..3efc0165584 100644 --- a/dd-java-agent/instrumentation-testing/src/test/java/datadog/trace/agent/test/scopediag/ScopeDiagnosticsAdmissionTest.java +++ b/dd-java-agent/instrumentation-testing/src/test/java/datadog/trace/agent/test/scopediag/ScopeDiagnosticsAdmissionTest.java @@ -3,6 +3,7 @@ import static java.util.concurrent.TimeUnit.SECONDS; import static org.junit.jupiter.api.Assertions.assertEquals; import static org.junit.jupiter.api.Assertions.assertFalse; +import static org.junit.jupiter.api.Assertions.assertNull; import static org.junit.jupiter.api.Assertions.assertTrue; import datadog.context.Context; @@ -10,11 +11,7 @@ import datadog.trace.api.DDTraceId; import datadog.trace.bootstrap.instrumentation.api.NoopScope; import datadog.trace.test.util.PollingConditions; -import java.lang.reflect.Field; -import java.util.concurrent.ConcurrentLinkedQueue; import java.util.concurrent.FutureTask; -import java.util.concurrent.atomic.AtomicBoolean; -import java.util.concurrent.atomic.AtomicInteger; import java.util.concurrent.locks.LockSupport; import org.junit.jupiter.api.AfterEach; import org.junit.jupiter.api.Test; @@ -51,12 +48,13 @@ void callbacksDoNotWaitForTheReportLock() throws Exception { }); Thread carrier = new Thread(callbacks, "diagnostic-carrier"); carrier.setDaemon(true); - Field lock = ScopeDiagnostics.class.getDeclaredField("lifecycleLock"); - lock.setAccessible(true); - synchronized (lock.get(instance())) { + ScopeDiagnostics.INSTANCE.lifecycleLock.lock(); + try { carrier.start(); // A carrier must return even when the diagnostic monitor's owner cannot run yet. callbacks.get(5, SECONDS); + } finally { + ScopeDiagnostics.INSTANCE.lifecycleLock.unlock(); } ScopeDiagnostics.stop(); @@ -94,7 +92,7 @@ void queuedResolutionWakesQuiescenceWaiter() throws Exception { waiter.setDaemon(true); waiter.start(); try { - Object diagnostics = instance(); + Object diagnostics = ScopeDiagnostics.INSTANCE; new PollingConditions(5) .eventually(() -> assertEquals(diagnostics, LockSupport.getBlocker(waiter))); ScopeDiagnostics.recordResolve(window, continuation, true, System.nanoTime(), false); @@ -121,132 +119,62 @@ void resetDiscardsQueuedEvents() { assertTrue(ScopeDiagnostics.report().records().isEmpty()); } - @TableTest({ - "Scenario | Reset", - "reset | true ", - "stop | false" - }) - void staleProducerCannotRetainEventsAfterRecordingEnds(boolean reset) throws Exception { + @Test + void stopRetainsReservationsWhoseStackCaptureIsStillInFlight() throws Exception { ScopeDiagnostics.startRecording(); Object window = ScopeDiagnostics.recordingWindow(); - Field queueField = ScopeDiagnostics.class.getDeclaredField("pendingEvents"); - queueField.setAccessible(true); - Object diagnostics = instance(); - Object originalQueue = queueField.get(diagnostics); - AtomicBoolean firstAdd = new AtomicBoolean(true); - ConcurrentLinkedQueue delayedQueue = - new ConcurrentLinkedQueue() { - @Override - public boolean add(Runnable event) { - if (firstAdd.getAndSet(false)) { - // End recording after the callback's window check but before publication. - if (reset) { - ScopeDiagnostics.reset(); - } else { - ScopeDiagnostics.stop(); - } - } - return super.add(event); - } - }; - try { - queueField.set(diagnostics, delayedQueue); - ScopeDiagnostics.recordCapture( - window, Context.root().capture(), DDTraceId.from(1), 2, "stale", (byte) 0); - // Do not call report/reset here: either would hide a stale item retained in the static queue. - assertTrue(delayedQueue.isEmpty()); - } finally { - queueField.set(diagnostics, originalQueue); - } - } - - @TableTest({ - "Operation ", - "report ", - "stop ", - "quiescence" - }) - void drainingDoesNotChaseNewEvents(String operation) throws Exception { - ScopeDiagnostics.startRecording(); - ScopeDiagnostics.recordCapture( - ScopeDiagnostics.recordingWindow(), - Context.root().capture(), - DDTraceId.from(1), - 2, - "incomplete", - (byte) 0); - ConcurrentLinkedQueue queue = pendingEvents(); - AtomicBoolean producing = new AtomicBoolean(true); - AtomicInteger consumed = new AtomicInteger(); - Runnable event = - new Runnable() { - @Override - public void run() { - consumed.incrementAndGet(); - if (producing.get()) { - queue.add(this); - } - } - }; - queue.add(event); - FutureTask reader = + ContextContinuation continuation = Context.root().capture(); + ScopeDiagnostics.recordCapture(window, continuation, DDTraceId.from(1), 2, "op", (byte) 0); + DiagnosticEventBuffer.Event reservation = ((DiagnosticEventBuffer) window).reserve(); + ScopeDiagnostics.recordResolve(window, continuation, true, System.nanoTime(), false); + FutureTask stopping = new FutureTask<>( () -> { - switch (operation) { - case "report": - ScopeDiagnostics.report(); - break; - case "stop": - ScopeDiagnostics.stop(); - break; - case "quiescence": - ScopeDiagnostics.awaitQuiescence(20); - break; - default: - throw new AssertionError(operation); - } + ScopeDiagnostics.stop(); return null; }); - Thread consumer = new Thread(reader, "diagnostic-reader"); - consumer.setDaemon(true); - consumer.start(); + Thread stopper = new Thread(stopping, "diagnostic-stop"); + stopper.setDaemon(true); + stopper.start(); try { - reader.get(5, SECONDS); - assertTrue(consumed.get() > 0); + new PollingConditions(5).eventually(() -> assertNull(ScopeDiagnostics.recordingWindow())); + // Publish after stop closes admission; the reservation still belongs to this recording. + reservation.publish(() -> {}); + stopping.get(5, SECONDS); + assertEquals(0, ScopeDiagnostics.report().leakCount()); } finally { - producing.set(false); - consumer.join(5000); + reservation.publishIfAbsent(); + stopper.join(5000); } } @Test - void expiredQuiescenceBudgetLeavesBacklogForTheReport() throws Exception { + void callbacksConsumeEventsWithoutWaitingForAReport() { ScopeDiagnostics.startRecording(); + DiagnosticEventBuffer window = (DiagnosticEventBuffer) ScopeDiagnostics.recordingWindow(); ScopeDiagnostics.recordCapture( - ScopeDiagnostics.recordingWindow(), - Context.root().capture(), - DDTraceId.from(1), - 2, - "incomplete", - (byte) 0); - ScopeDiagnostics.awaitQuiescence(0); - assertFalse(pendingEvents().isEmpty()); - ScopeDiagnosticsReport report = ScopeDiagnostics.report(); - assertEquals(1, report.leakCount()); - assertTrue(pendingEvents().isEmpty()); - } - - @SuppressWarnings("unchecked") - private static ConcurrentLinkedQueue pendingEvents() throws Exception { - Field queue = ScopeDiagnostics.class.getDeclaredField("pendingEvents"); - queue.setAccessible(true); - return (ConcurrentLinkedQueue) queue.get(instance()); + window, Context.root().capture(), DDTraceId.from(1), 2, "op", (byte) 0); + assertFalse(window.hasPending()); } - private static Object instance() throws Exception { - Field instance = ScopeDiagnostics.class.getDeclaredField("INSTANCE"); - instance.setAccessible(true); - return instance.get(null); + @Test + void firstResolutionAndDuplicateReleaseArePreservedBehindDelayedAdmission() { + ScopeDiagnostics.startRecording(); + DiagnosticEventBuffer window = (DiagnosticEventBuffer) ScopeDiagnostics.recordingWindow(); + ContextContinuation continuation = Context.root().capture(); + ScopeDiagnostics.recordCapture(window, continuation, DDTraceId.from(1), 2, "op", (byte) 0); + DiagnosticEventBuffer.Event delayed = window.reserve(); + ScopeDiagnostics.recordResolve(window, continuation, false, 1, false); + ScopeDiagnostics.recordResolve(window, continuation, true, 2, false); + ScopeDiagnostics.recordResolve(window, continuation, true, 3, true); + delayed.publish(() -> {}); + ScopeDiagnosticsReport report = ScopeDiagnostics.report(); + ContinuationRecord record = report.records().get(0); + assertEquals(ContinuationStatus.FINISHED, record.status()); + assertEquals(1, record.terminal().nanos); + assertEquals(1, record.extraTerminals().size()); + assertEquals(3, record.extraTerminals().get(0).nanos); + assertEquals(1, report.doubleCount()); } @Test From 69bbf76b67c156f464a4b44902bcec4aed3da455 Mon Sep 17 00:00:00 2001 From: Andrea Marziali Date: Mon, 26 Oct 2026 21:16:43 +0100 Subject: [PATCH 4/4] spotbugs --- .../agent/test/scopediag/DiagnosticEventBuffer.java | 5 +++++ .../trace/agent/test/scopediag/ScopeDiagnostics.java | 10 ++++++++++ 2 files changed, 15 insertions(+) diff --git a/dd-java-agent/instrumentation-testing/src/main/java/datadog/trace/agent/test/scopediag/DiagnosticEventBuffer.java b/dd-java-agent/instrumentation-testing/src/main/java/datadog/trace/agent/test/scopediag/DiagnosticEventBuffer.java index bd26dae76f6..e156617e1c8 100644 --- a/dd-java-agent/instrumentation-testing/src/main/java/datadog/trace/agent/test/scopediag/DiagnosticEventBuffer.java +++ b/dd-java-agent/instrumentation-testing/src/main/java/datadog/trace/agent/test/scopediag/DiagnosticEventBuffer.java @@ -1,5 +1,6 @@ package datadog.trace.agent.test.scopediag; +import edu.umd.cs.findbugs.annotations.SuppressFBWarnings; import java.util.concurrent.atomic.AtomicReference; import java.util.concurrent.locks.LockSupport; @@ -14,6 +15,10 @@ final class DiagnosticEventBuffer { // A successful CAS reserves FIFO order before stack capture; the volatile next/action writes // publish the link and payload separately. Only the consumer touches consumed and draining. private final AtomicReference tail = new AtomicReference<>(consumed); + + @SuppressFBWarnings( + value = "AT_STALE_THREAD_WRITE_OF_PRIMITIVE", + justification = "Only consumer methods access this field, while holding the lifecycle lock") private boolean draining; Event reserve() { diff --git a/dd-java-agent/instrumentation-testing/src/main/java/datadog/trace/agent/test/scopediag/ScopeDiagnostics.java b/dd-java-agent/instrumentation-testing/src/main/java/datadog/trace/agent/test/scopediag/ScopeDiagnostics.java index cbf8bd6a67c..da21eaaed6c 100644 --- a/dd-java-agent/instrumentation-testing/src/main/java/datadog/trace/agent/test/scopediag/ScopeDiagnostics.java +++ b/dd-java-agent/instrumentation-testing/src/main/java/datadog/trace/agent/test/scopediag/ScopeDiagnostics.java @@ -2,6 +2,7 @@ import datadog.context.ContextContinuation; import datadog.trace.api.DDTraceId; +import edu.umd.cs.findbugs.annotations.SuppressFBWarnings; import java.util.ArrayList; import java.util.Collections; import java.util.EnumSet; @@ -44,8 +45,17 @@ public final class ScopeDiagnostics { Collections.newSetFromMap(new IdentityHashMap()); private long seq; private long scopeSeq; + + @SuppressFBWarnings( + value = "AT_NONATOMIC_64BIT_PRIMITIVE", + justification = "All reads and writes, including Listener callbacks, hold lifecycleLock") private long incompleteContinuations; + + @SuppressFBWarnings( + value = "AT_NONATOMIC_64BIT_PRIMITIVE", + justification = "All reads and writes, including Listener callbacks, hold lifecycleLock") private long incompleteScopes; + private volatile DiagnosticEventBuffer recordingWindow; private volatile StackFilter stackFilter = new StackFilter(DEFAULT_MAX_FRAMES);