Bound in-flight virtual threads in VirtualThreadChurnAntagonist

Unthrottled submission (256 vthreads/~1ms) outpaced the carrier pool's
drain rate, so pending VirtualThread/Continuation objects piled up
unboundedly and OOM'd in ~45s under a 1g heap. A semaphore now caps
in-flight vthreads, throttling submission to actual execution rate.

Co-Authored-By: Claude Sonnet 5 <noreply@anthropic.com>

Author: jbachorik

ddprof-stresstest/src/chaos/java/com/datadoghq/profiler/chaos/VirtualThreadChurnAntagonist.java

@@ -11,6 +11,7 @@
 
 import java.lang.reflect.Method;
 import java.time.Duration;
+import java.util.concurrent.Semaphore;
 import java.util.concurrent.atomic.AtomicLong;
 
 /**
@@ -19,13 +20,20 @@
  *
  * <p>Reflectively detects {@code Thread.ofVirtual()} (Java 21+); gracefully
  * no-ops on older runtimes.
+ *
+ * <p>In-flight virtual threads are bounded by a semaphore: the default
+ * carrier pool has far fewer platform threads than a batch submits, so an
+ * unthrottled submission rate outpaces execution and the backlog of pending
+ * {@code VirtualThread}/{@code Continuation} objects grows without bound
+ * instead of churning.
  */
 public final class VirtualThreadChurnAntagonist implements Antagonist {
 
     private static final Method OF_VIRTUAL = resolveOfVirtual();
     private static final Method BUILDER_START = resolveBuilderStart();
 
     private final int batchSize;
+    private final Semaphore inFlight;
 
     private volatile boolean running;
     private Thread driver;
@@ -37,6 +45,7 @@ public VirtualThreadChurnAntagonist() {
 
     public VirtualThreadChurnAntagonist(int batchSize) {
         this.batchSize = batchSize;
+        this.inFlight = new Semaphore(batchSize * 4);
     }
 
     @Override
@@ -70,10 +79,26 @@ private void loop() {
         while (running) {
             for (int i = 0; i < batchSize && running; i++) {
                 final long seed = System.nanoTime() ^ i;
+                try {
+                    // Blocks once the backlog is full, throttling submission to the
+                    // carrier pool's actual drain rate instead of hoarding pending
+                    // VirtualThread/Continuation objects.
+                    inFlight.acquire();
+                } catch (InterruptedException e) {
+                    Thread.currentThread().interrupt();
+                    return;
+                }
                 try {
                     Object builder = OF_VIRTUAL.invoke(null);
-                    BUILDER_START.invoke(builder, (Runnable) () -> sink.addAndGet(burn(seed)));
+                    BUILDER_START.invoke(builder, (Runnable) () -> {
+                        try {
+                            sink.addAndGet(burn(seed));
+                        } finally {
+                            inFlight.release();
+                        }
+                    });
                 } catch (Throwable t) {
+                    inFlight.release();
                     return;
                 }
             }