test: verify JVM producers in all-thread scope

Author: kaahos

ddprof-test/src/test/java/com/datadoghq/profiler/wallclock/JvmtiBasedMonitorTaskBlockTest.java

@@ -25,6 +25,6 @@ protected void withTestAssumptions() {
 
   @Override
   protected String getProfilerCommand() {
-    return "wall=1ms,wallprecheck=true,jvmtistacks=true";
+    return "wall=1ms,filter=,wallprecheck=true,jvmtistacks=true";
   }
 }

ddprof-test/src/test/java/com/datadoghq/profiler/wallclock/JvmtiBasedParkTaskBlockTest.java

@@ -25,6 +25,6 @@ protected void withTestAssumptions() {
 
   @Override
   protected String getProfilerCommand() {
-    return "wall=1ms,wallprecheck=true,jvmtistacks=true";
+    return "wall=1ms,filter=,wallprecheck=true,jvmtistacks=true";
   }
 }

ddprof-test/src/test/java/com/datadoghq/profiler/wallclock/MonitorTaskBlockTest.java

@@ -142,7 +142,7 @@ public void staleWaitStateIsRecoveredAfterProfilerRestart() throws Exception {
     Path recording = Files.createTempFile("MonitorTaskBlockTest-restart-", ".jfr");
     boolean restarted = false;
     try {
-      profiler.execute("start,wall=1ms,wallscope=all,wallprecheck=true,jfr,file="
+      profiler.execute("start,wall=1ms,filter=,wallprecheck=true,jfr,file="
           + recording.toAbsolutePath());
       restarted = true;
       synchronized (contentionMonitor) {
@@ -218,7 +218,7 @@ public void virtualMonitorCallbacksDoNotEmitCarrierTaskBlocks() throws Exception
 
   @Override
   protected String getProfilerCommand() {
-    return "wall=1ms,wallprecheck=true";
+    return "wall=1ms,filter=,wallprecheck=true";
   }
 
   protected void assertTaskBlockStackReference(IItemCollection events) {

ddprof-test/src/test/java/com/datadoghq/profiler/wallclock/ParkTaskBlockTest.java

@@ -8,13 +8,17 @@
 import com.datadoghq.profiler.AbstractProfilerTest;
 import com.datadoghq.profiler.ProfilerOwnedBlockHooks;
 import java.lang.reflect.Method;
+import java.util.concurrent.CountDownLatch;
 import java.util.concurrent.TimeUnit;
+import java.util.concurrent.atomic.AtomicBoolean;
+import java.util.concurrent.atomic.AtomicReference;
 import java.util.concurrent.locks.LockSupport;
 import org.junit.jupiter.api.Test;
 import org.junit.jupiter.api.Assumptions;
 import org.openjdk.jmc.common.item.IItemCollection;
 
 import static org.junit.jupiter.api.Assertions.assertFalse;
+import static org.junit.jupiter.api.Assertions.assertTrue;
 
 /** Verifies TaskBlock production from Java-owned platform-thread park hooks. */
 public class ParkTaskBlockTest extends AbstractProfilerTest {
@@ -94,9 +98,17 @@ public void virtualParkDoesNotMutateCarrierProducerState() throws Exception {
     TaskBlockAssertions.assertContains(events, 0, 0, BLOCKER, UNBLOCKING_SPAN_ID);
   }
 
+  @Test
+  public void platformParkSuppressesSignalsAndClearsOwnership() throws Exception {
+    long baseline = profiler.getDebugCounters()
+        .getOrDefault("wc_signals_suppressed_owned_block", 0L);
+    long afterFirstPark = runSuppressedPark(baseline);
+    runSuppressedPark(afterFirstPark);
+  }
+
   @Override
   protected String getProfilerCommand() {
-    return "wall=1ms,wallprecheck=true";
+    return "wall=1ms,filter=,wallprecheck=true";
   }
 
   protected void assertTaskBlockStackReference(IItemCollection events) {
@@ -112,4 +124,46 @@ private static void parkForMillis(long millis) {
       LockSupport.parkNanos(remaining);
     }
   }
+
+  private long runSuppressedPark(long baseline) throws Exception {
+    CountDownLatch armed = new CountDownLatch(1);
+    AtomicBoolean release = new AtomicBoolean();
+    AtomicReference<Throwable> error = new AtomicReference<>();
+    Thread worker = new Thread(() -> {
+      try {
+        ProfilerOwnedBlockHooks.parkEnter(profiler);
+        armed.countDown();
+        while (!release.get()) {
+          Thread.yield();
+        }
+      } catch (Throwable t) {
+        error.set(t);
+      } finally {
+        ProfilerOwnedBlockHooks.parkExit(profiler, BLOCKER, UNBLOCKING_SPAN_ID);
+      }
+    }, "taskblock-park-suppression");
+
+    worker.start();
+    assertTrue(armed.await(5, TimeUnit.SECONDS));
+    try {
+      waitForCounterAbove("wc_signals_suppressed_owned_block", baseline, 5_000L);
+    } finally {
+      release.set(true);
+    }
+    worker.join(5_000L);
+    assertFalse(worker.isAlive());
+    if (error.get() != null) throw new AssertionError(error.get());
+    return profiler.getDebugCounters()
+        .getOrDefault("wc_signals_suppressed_owned_block", 0L);
+  }
+
+  private void waitForCounterAbove(String name, long baseline, long timeoutMillis)
+      throws Exception {
+    long deadline = System.nanoTime() + TimeUnit.MILLISECONDS.toNanos(timeoutMillis);
+    while (System.nanoTime() < deadline) {
+      if (profiler.getDebugCounters().getOrDefault(name, 0L) > baseline) return;
+      Thread.sleep(10L);
+    }
+    throw new AssertionError("Counter did not increase: " + name);
+  }
 }

ddprof-test/src/test/java/com/datadoghq/profiler/wallclock/WallclockMitigationsCombinedTest.java

@@ -43,6 +43,7 @@ public void contextScopedThreadsRemainSampled() throws Exception {
         Thread sleeping =
                 new Thread(
                         () -> {
+                            registerCurrentThreadForWallClockProfiling();
                             ready.countDown();
                             long token = ProfilerOwnedBlockHooks.blockEnter(
                                     profiler, OSTHREAD_STATE_SLEEPING);
@@ -58,6 +59,7 @@ public void contextScopedThreadsRemainSampled() throws Exception {
         Thread parkedBusy =
                 new Thread(
                         () -> {
+                            registerCurrentThreadForWallClockProfiling();
                             long spanId = 0x1111L;
                             long rootSpanId = 0x2222L;
                             profiler.setContext(rootSpanId, spanId, 0, 0);
@@ -76,6 +78,7 @@ public void contextScopedThreadsRemainSampled() throws Exception {
         Thread runnable =
                 new Thread(
                         () -> {
+                            registerCurrentThreadForWallClockProfiling();
                             ready.countDown();
                             while (!stop.get()) {
                                 // keep runnable
@@ -121,7 +124,7 @@ public void contextScopedThreadsRemainSampled() throws Exception {
 
     @Override
     protected String getProfilerCommand() {
-        return "wall=1ms,wallprecheck=true";
+        return "wall=1ms,filter=0,wallprecheck=true";
     }
 
     private Map<String, Long> samplesByThreadName() {