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..e156617e1c8 --- /dev/null +++ b/dd-java-agent/instrumentation-testing/src/main/java/datadog/trace/agent/test/scopediag/DiagnosticEventBuffer.java @@ -0,0 +1,101 @@ +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; + +/** + * 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); + + @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() { + 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 11b1f8c95f2..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; @@ -9,6 +10,11 @@ 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.concurrent.locks.ReentrantLock; +import java.util.function.Consumer; +import java.util.function.Supplier; import org.slf4j.Logger; import org.slf4j.LoggerFactory; @@ -19,10 +25,14 @@ 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(); - /** Linearizes event admission, stop/reset, and report snapshots. */ - private final Object lifecycleLock = new Object(); + /** Serializes event consumption, stop/reset, and report snapshots. */ + 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 quiescenceWaiters = new ConcurrentLinkedQueue<>(); private final Map records = new IdentityHashMap<>(); private final Map scopeRecords = new IdentityHashMap<>(); @@ -35,8 +45,19 @@ public final class ScopeDiagnostics { Collections.newSetFromMap(new IdentityHashMap()); private long seq; private long scopeSeq; - private volatile Object recordingWindow; - private StackFilter stackFilter = new StackFilter(DEFAULT_MAX_FRAMES); + + @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); private final Listener listener = new Listener(); @@ -65,37 +86,57 @@ 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.lifecycleLock.lock(); + try { + DiagnosticEventBuffer window = INSTANCE.recordingWindow; INSTANCE.recordingWindow = null; + 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(); } - /** 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; } /** 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(); } } @@ -111,21 +152,29 @@ 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()) { + 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) { 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 +236,11 @@ static EnumSet enabledChecks(TrackScopeContinuations conf } private void clear() { + DiagnosticEventBuffer window = recordingWindow; + recordingWindow = null; + if (window != null) { + window.close(); + } records.clear(); scopeRecords.clear(); deferredCleanupScopes.clear(); @@ -195,6 +249,8 @@ private void clear() { released.clear(); seq = 0; scopeSeq = 0; + incompleteContinuations = 0; + incompleteScopes = 0; } private ScopeDiagnosticsReport snapshot() { @@ -206,19 +262,95 @@ private ScopeDiagnosticsReport snapshot() { private static final StackTraceElement[] NO_STACK = new StackTraceElement[0]; - private ScopeEvent event(ScopeEvent.Type type) { - return event(type, System.nanoTime()); + private ScopeEvent event( + ScopeEvent.Type type, long nanos, String threadName, StackFilter filter, Throwable callSite) { + StackTraceElement[] stack = + callSite == null ? NO_STACK : filter.filter(callSite.getStackTrace()); + return new ScopeEvent(type, threadName, nanos, stack); } - /** 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; - 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 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; + } + StackFilter filter = INSTANCE.stackFilter; + String threadName = Thread.currentThread().getName(); + try { + if (INSTANCE.lifecycleLock.tryLock()) { + try { + if (window == INSTANCE.recordingWindow) { + 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; + } + } + } 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(); + } + } + + private void wakeQuiescenceWaiters() { + for (Thread waiter : quiescenceWaiters) { + LockSupport.unpark(waiter); + } + } + + private void drainEvents() { + drainEvents(false, 0, true); + } + + private void drainEvents(boolean timed, long deadline, boolean waitForPublication) { + DiagnosticEventBuffer window = recordingWindow; + if (window != null) { + window.drain(timed, deadline, waitForPublication); + } + } + + /** 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( @@ -228,11 +360,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 +375,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 +396,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,45 +422,50 @@ 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, + Supplier event) { try { ContinuationRecord record = - new ContinuationRecord( - seq++, traceId, spanId, spanName, source, false, event(ScopeEvent.Type.CAPTURE)); - 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 } @@ -342,37 +477,41 @@ void onActivate( long spanId, String spanName, byte source, - long nanos) { + Supplier event) { try { - recordFor(id, traceId, spanId, spanName, source) - .addResume(event(ScopeEvent.Type.ACTIVATE, nanos)); + recordFor(id, traceId, spanId, spanName, source).addResume(event.get()); } catch (Throwable ignored) { } } - void onActivateFailed(ContextContinuation id) { + 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(ScopeEvent.Type.ACTIVATE_FAILED)); + 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 release, long resolveNanos, boolean duplicateAttempt) { + void onResolve(ContextContinuation id, boolean duplicateAttempt, Supplier 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); + 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) { } } - 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 +522,8 @@ void onScopeOpen( long spanId, String spanName, byte source, - ContextContinuation continuation) { + ContextContinuation continuation, + Supplier event) { try { if (scopeRecords.containsKey(scope)) { return; // re-activation of an already-open scope, not a new open @@ -401,7 +541,10 @@ void onScopeOpen( source, continuationSeq, deferredCleanupScopes.remove(scope), - event(ScopeEvent.Type.SCOPE_OPEN))); + event.get())); + if (!scopeRecords.get(scope).deferredCleanup()) { + incompleteScopes++; + } if (owner != null) { owner.linkScope(s); } @@ -409,21 +552,24 @@ void onScopeOpen( } } - void onScopeClose(Object scope) { + void onScopeClose(Object scope, Supplier event) { try { ScopeRecord record = scopeRecords.get(scope); - if (record != null) { - record.setClose(event(ScopeEvent.Type.SCOPE_CLOSE)); + if (record != null && !record.closed()) { + record.setClose(event.get()); + if (!record.deferredCleanup()) { + incompleteScopes--; + } } } catch (Throwable ignored) { } } - void onScopeCloseWrongThread(Object scope) { + void onScopeCloseWrongThread(Object scope, Supplier event) { try { ScopeRecord record = scopeRecords.get(scope); if (record != null) { - record.addWrongThreadClose(event(ScopeEvent.Type.SCOPE_CLOSE_WRONG_THREAD)); + record.addWrongThreadClose(event.get()); } } catch (Throwable ignored) { } @@ -433,6 +579,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 3766bcb6711..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 @@ -1,13 +1,18 @@ 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 datadog.context.Context; import datadog.context.ContextContinuation; import datadog.trace.api.DDTraceId; import datadog.trace.bootstrap.instrumentation.api.NoopScope; +import datadog.trace.test.util.PollingConditions; +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,160 @@ 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); + 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(); + + 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 = ScopeDiagnostics.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()); + } + + @Test + void stopRetainsReservationsWhoseStackCaptureIsStillInFlight() throws Exception { + ScopeDiagnostics.startRecording(); + Object window = ScopeDiagnostics.recordingWindow(); + 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<>( + () -> { + ScopeDiagnostics.stop(); + return null; + }); + Thread stopper = new Thread(stopping, "diagnostic-stop"); + stopper.setDaemon(true); + stopper.start(); + try { + 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 { + reservation.publishIfAbsent(); + stopper.join(5000); + } + } + + @Test + void callbacksConsumeEventsWithoutWaitingForAReport() { + ScopeDiagnostics.startRecording(); + DiagnosticEventBuffer window = (DiagnosticEventBuffer) ScopeDiagnostics.recordingWindow(); + ScopeDiagnostics.recordCapture( + window, Context.root().capture(), DDTraceId.from(1), 2, "op", (byte) 0); + assertFalse(window.hasPending()); + } + + @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 void failedActivationBeforeResolutionAdviceIsRetained() { ScopeDiagnostics.startRecording();