Demonstrate Tier-1 tracer lifecycle failures with formal methods

.gitignore

@@ -76,6 +76,8 @@ replay_pid*
 **/src/*/generated/*
 !dd-java-agent/benchmark/releases/*.jar
 **/errors/*.log
+**/jcstress-results-*.bin.gz
+/dd-trace-core/results/
 
 # Magic for local JMC built
 /vendor/jmc-libs

dd-trace-core/build.gradle

@@ -69,6 +69,30 @@ excludedClassesCoverage += [
 
 addTestSuite('traceAgentTest')
 
+sourceSets {
+  jcstress {
+    java.srcDir 'src/jcstress/java'
+    compileClasspath += sourceSets.main.output
+    runtimeClasspath += output + compileClasspath
+  }
+}
+
+configurations {
+  jcstressImplementation.extendsFrom implementation
+  jcstressRuntimeClasspath {
+    resolutionStrategy.force 'com.datadoghq:dd-instrument-java:0.0.5'
+  }
+}
+
+tasks.register('jcstress', JavaExec) {
+  group = 'verification'
+  description = 'Runs Tier-1 JCStress concurrency probes.'
+  dependsOn tasks.named('jcstressClasses')
+  classpath = sourceSets.jcstress.runtimeClasspath
+  mainClass = 'org.openjdk.jcstress.Main'
+  args '-t', 'datadog.trace.core.*', '-f', '1', '-iters', '5', '-time', '200'
+}
+
 tasks.named('forkedTest', Test) {
   // Needed for FootprintForkedTest on Java 17
   jvmArgs += ['-Djol.magicFieldOffset=true']
@@ -118,6 +142,9 @@ dependencies {
   testImplementation group: 'org.msgpack', name: 'jackson-dataformat-msgpack', version: '0.8.20'
   testImplementation group: 'org.openjdk.jol', name: 'jol-core', version: '0.17'
   testImplementation group: 'commons-codec', name: 'commons-codec', version: '1.3'
+
+  jcstressImplementation group: 'org.openjdk.jcstress', name: 'jcstress-core', version: '0.16'
+  jcstressAnnotationProcessor group: 'org.openjdk.jcstress', name: 'jcstress-core', version: '0.16'
   testImplementation group: 'com.amazonaws', name: 'aws-lambda-java-events', version:'3.11.0'
   testImplementation group: 'com.google.protobuf', name: 'protobuf-java', version: '3.14.0'
   testImplementation libs.jnr.unixsocket

dd-trace-core/src/jcstress/java/datadog/trace/core/PendingTraceBufferConcurrentFlushStressTest.java

@@ -0,0 +1,83 @@
+package datadog.trace.core;
+
+import static org.openjdk.jcstress.annotations.Expect.ACCEPTABLE;
+import static org.openjdk.jcstress.annotations.Expect.FORBIDDEN;
+
+import datadog.trace.api.Config;
+import datadog.trace.api.time.SystemTimeSource;
+import datadog.trace.core.monitor.HealthMetrics;
+import java.util.concurrent.atomic.AtomicBoolean;
+import java.util.concurrent.atomic.AtomicInteger;
+import org.openjdk.jcstress.annotations.Actor;
+import org.openjdk.jcstress.annotations.Arbiter;
+import org.openjdk.jcstress.annotations.JCStressTest;
+import org.openjdk.jcstress.annotations.Outcome;
+import org.openjdk.jcstress.annotations.State;
+import org.openjdk.jcstress.infra.results.I_Result;
+
+@JCStressTest
+@Outcome(id = "1", expect = ACCEPTABLE, desc = "The caller's flush wrote its accepted element.")
+@Outcome(id = "0", expect = FORBIDDEN, desc = "A concurrent marker released flush too early.")
+@State
+public class PendingTraceBufferConcurrentFlushStressTest {
+  private final PendingTraceBuffer.DelayingPendingTraceBuffer buffer =
+      new PendingTraceBuffer.DelayingPendingTraceBuffer(
+          16, SystemTimeSource.INSTANCE, Config.get(), null, HealthMetrics.NO_OP);
+  private final RecordingElement element = new RecordingElement();
+
+  public PendingTraceBufferConcurrentFlushStressTest() {
+    buffer.getWorker().start();
+  }
+
+  @Actor
+  public void competingFlush() {
+    buffer.flush();
+  }
+
+  @Actor
+  public void enqueueThenFlush(I_Result result) {
+    buffer.enqueue(element);
+    buffer.flush();
+    result.r1 = element.writes.get();
+  }
+
+  @Arbiter
+  public void close() {
+    buffer.close();
+  }
+
+  private static final class RecordingElement implements PendingTraceBuffer.Element {
+    private final AtomicBoolean enqueued = new AtomicBoolean();
+    private final AtomicInteger writes = new AtomicInteger();
+
+    @Override
+    public long oldestFinishedTime() {
+      return Long.MAX_VALUE;
+    }
+
+    @Override
+    public boolean lastReferencedNanosAgo(long nanos) {
+      return false;
+    }
+
+    @Override
+    public void write() {
+      writes.incrementAndGet();
+    }
+
+    @Override
+    public DDSpan getRootSpan() {
+      return null;
+    }
+
+    @Override
+    public boolean setEnqueued(boolean newValue) {
+      return enqueued.compareAndSet(!newValue, newValue);
+    }
+
+    @Override
+    public boolean writeOnBufferFull() {
+      return true;
+    }
+  }
+}

dd-trace-core/src/jcstress/java/datadog/trace/core/scopemanager/ScopeContinuationMultipleResumeStressTest.java

@@ -0,0 +1,66 @@
+package datadog.trace.core.scopemanager;
+
+import static org.openjdk.jcstress.annotations.Expect.ACCEPTABLE;
+import static org.openjdk.jcstress.annotations.Expect.FORBIDDEN;
+
+import datadog.context.Context;
+import datadog.context.ContextContinuation;
+import datadog.context.ContextScope;
+import datadog.trace.bootstrap.instrumentation.api.AgentTraceCollector;
+import java.util.concurrent.atomic.AtomicInteger;
+import org.openjdk.jcstress.annotations.Actor;
+import org.openjdk.jcstress.annotations.Arbiter;
+import org.openjdk.jcstress.annotations.JCStressTest;
+import org.openjdk.jcstress.annotations.Outcome;
+import org.openjdk.jcstress.annotations.State;
+import org.openjdk.jcstress.infra.results.I_Result;
+
+@JCStressTest
+@Outcome(id = "1", expect = ACCEPTABLE, desc = "All resumed scopes converge on one finalization.")
+@Outcome(expect = FORBIDDEN, desc = "A resumed scope leaked or finalized the continuation twice.")
+@State
+public class ScopeContinuationMultipleResumeStressTest {
+  private final RecordingCollector collector = new RecordingCollector();
+  private final ScopeContinuation continuation =
+      new ScopeContinuation(
+              new ContinuableScopeManager(0, false), Context.root(), (byte) 0, collector)
+          .register()
+          .hold();
+
+  @Actor
+  public void firstResume() {
+    resumeAndClose();
+  }
+
+  @Actor
+  public void secondResume() {
+    resumeAndClose();
+  }
+
+  @Actor
+  public void release() {
+    continuation.release();
+  }
+
+  @Arbiter
+  public void observe(I_Result result) {
+    result.r1 = collector.removals.get();
+  }
+
+  private void resumeAndClose() {
+    ContextScope scope = continuation.resume();
+    scope.close();
+  }
+
+  private static final class RecordingCollector implements AgentTraceCollector {
+    private final AtomicInteger removals = new AtomicInteger();
+
+    @Override
+    public void registerContinuation(ContextContinuation continuation) {}
+
+    @Override
+    public void removeContinuation(ContextContinuation continuation) {
+      removals.incrementAndGet();
+    }
+  }
+}

dd-trace-core/src/jcstress/java/datadog/trace/core/scopemanager/ScopeContinuationReleaseResumeStressTest.java

@@ -0,0 +1,57 @@
+package datadog.trace.core.scopemanager;
+
+import static org.openjdk.jcstress.annotations.Expect.ACCEPTABLE;
+import static org.openjdk.jcstress.annotations.Expect.FORBIDDEN;
+
+import datadog.context.Context;
+import datadog.context.ContextContinuation;
+import datadog.context.ContextScope;
+import datadog.trace.bootstrap.instrumentation.api.AgentTraceCollector;
+import java.util.concurrent.atomic.AtomicInteger;
+import org.openjdk.jcstress.annotations.Actor;
+import org.openjdk.jcstress.annotations.Arbiter;
+import org.openjdk.jcstress.annotations.JCStressTest;
+import org.openjdk.jcstress.annotations.Outcome;
+import org.openjdk.jcstress.annotations.State;
+import org.openjdk.jcstress.infra.results.I_Result;
+
+@JCStressTest
+@Outcome(id = "1", expect = ACCEPTABLE, desc = "The continuation is finalized exactly once.")
+@Outcome(expect = FORBIDDEN, desc = "The continuation leaked or was finalized more than once.")
+@State
+public class ScopeContinuationReleaseResumeStressTest {
+  private final RecordingCollector collector = new RecordingCollector();
+  private final ScopeContinuation continuation =
+      new ScopeContinuation(
+              new ContinuableScopeManager(0, false), Context.root(), (byte) 0, collector)
+          .register()
+          .hold();
+
+  @Actor
+  public void release() {
+    continuation.release();
+  }
+
+  @Actor
+  public void resumeAndClose() {
+    ContextScope scope = continuation.resume();
+    scope.close();
+  }
+
+  @Arbiter
+  public void observe(I_Result result) {
+    result.r1 = collector.removals.get();
+  }
+
+  private static final class RecordingCollector implements AgentTraceCollector {
+    private final AtomicInteger removals = new AtomicInteger();
+
+    @Override
+    public void registerContinuation(ContextContinuation continuation) {}
+
+    @Override
+    public void removeContinuation(ContextContinuation continuation) {
+      removals.incrementAndGet();
+    }
+  }
+}