Merge

Author: zhengyu123

build-logic/conventions/src/main/kotlin/com/datadoghq/profiler/ProfilerTestPlugin.kt

@@ -15,6 +15,7 @@ import org.gradle.api.provider.Property
 import org.gradle.api.tasks.Exec
 import org.gradle.api.tasks.SourceSetContainer
 import org.gradle.api.tasks.testing.Test
+import java.io.File
 import java.time.Duration
 import javax.inject.Inject
 
@@ -76,6 +77,61 @@ import javax.inject.Inject
  * ```
  */
 class ProfilerTestPlugin : Plugin<Project> {
+
+    /**
+     * Major version of the *test* JVM, read from its `release` file (`JAVA_VERSION="..."`) rather
+     * than by executing the launcher.
+     *
+     * Executing `$JAVA_TEST_HOME/bin/java -version` (PlatformUtils.testJvmMajorVersion()) is
+     * unreliable here: in the musl split-JDK matrix it has been observed to report the build JDK
+     * (21) even when the test JVM is JDK 8, which put a JDK-21-only `--add-exports` onto a JDK-8
+     * launcher and aborted it. Reading the `release` file is a pure file read of the same
+     * JAVA_TEST_HOME the executable is resolved from — deterministic, no subprocess, no exec-format
+     * or PATH hazards. Returns 0 when it cannot be determined (missing/old `release`), so callers
+     * fail safe: they omit the flag, the profiler degrades to thread-scoped storage, and the
+     * carrier-scoping tests skip — never an abort.
+     */
+    private fun testJvmMajorVersionFromRelease(): Int = try {
+        val release = File(PlatformUtils.testJavaHome(), "release")
+        val version = release.takeIf { it.isFile }
+            ?.readLines()
+            ?.firstOrNull { it.startsWith("JAVA_VERSION=") }
+            ?.substringAfter('=')?.trim()?.trim('"')
+        // "1.8.0_452" -> 8 ; "21.0.5" -> 21
+        val parts = version?.split('.').orEmpty()
+        val majorToken = when {
+            parts.isEmpty() -> ""
+            parts[0] == "1" && parts.size > 1 -> parts[1]
+            else -> parts[0]
+        }
+        majorToken.takeWhile { it.isDigit() }.toIntOrNull() ?: 0
+    } catch (e: Exception) {
+        0
+    }
+
+    /**
+     * JVM args required to enable carrier-scoped OTEL context storage
+     * (`OtelContextStorage.Mode.CARRIER`), or an empty list when the test JVM does not support it.
+     *
+     * Carrier scoping resolves `jdk.internal.misc.CarrierThreadLocal`, which lives in a
+     * non-exported package, so it needs `--add-exports java.base/jdk.internal.misc=ALL-UNNAMED`.
+     * That type only exists on JDK 21+, and the flag *aborts* a Java 8 JVM ("Unrecognized option"),
+     * so it is gated on the version of the actual test JVM.
+     *
+     * MUST be evaluated at task execution time (inside doFirst), not configuration time: the test
+     * JVM is selected via JAVA_TEST_HOME, which the CI only makes resolvable at execution time (see
+     * the `executable` assignments below).
+     */
+    private fun carrierExportJvmArgs(project: Project): List<String> {
+        val major = testJvmMajorVersionFromRelease()
+        val enabled = major >= 21
+        project.logger.info(
+            "ddprof: carrier --add-exports gate — testJavaHome={}, detected major={}, flag {}",
+            PlatformUtils.testJavaHome(), major, if (enabled) "ADDED" else "omitted"
+        )
+        return if (enabled) listOf("--add-exports=java.base/jdk.internal.misc=ALL-UNNAMED") else emptyList()
+    }
+
     override fun apply(project: Project) {
         val extension = project.extensions.create(
             "profilerTest",
@@ -238,6 +294,8 @@ class ProfilerTestPlugin : Plugin<Project> {
             testTask.doFirst {
                 val allArgs = mutableListOf<String>()
                 allArgs.addAll(testConfig.standardJvmArgs)
+                // Version-gated at execution time, when the real test JVM is resolvable.
+                allArgs.addAll(carrierExportJvmArgs(project))
 
                 if (extension.nativeLibDir.isPresent) {
                     allArgs.add("-Djava.library.path=${extension.nativeLibDir.get().asFile.absolutePath}")
@@ -255,12 +313,21 @@ class ProfilerTestPlugin : Plugin<Project> {
 
             // Sanitizer conditions
             when (testConfig.configName) {
-                "asan" -> testTask.onlyIf {
-                    PlatformUtils.locateLibasan() != null &&
-                    // Skip J9+ASAN: OpenJ9 has known GC stack-scanning and defineClass
-                    // race bugs exposed by ASAN timing
-                    // https://github.com/eclipse-openj9/openj9/issues/23514
-                    !PlatformUtils.isTestJvmJ9()
+                "asan" -> {
+                    testTask.onlyIf {
+                        PlatformUtils.locateLibasan() != null &&
+                        // Skip J9+ASAN: OpenJ9 has known GC stack-scanning and defineClass
+                        // race bugs exposed by ASAN timing
+                        // https://github.com/eclipse-openj9/openj9/issues/23514
+                        !PlatformUtils.isTestJvmJ9()
+                    }
+                    // The ASAN test JVM is pinned to -Xmx512m (see ConfigurationPresets.kt)
+                    // to keep the heap below ASan's shadow-memory region. Without forking,
+                    // Gradle reuses one JVM for the whole suite, and per-test allocations
+                    // (JFR recordings, JMC object models) accumulate until it OOMs late in
+                    // the run even though every individual test passes. Restart periodically
+                    // to bound cumulative heap growth.
+                    testTask.setForkEvery(25)
                 }
                 // TSan + JVM integration tests are incompatible: the profiler's signal
                 // handlers (SIGPROF at 1ms) are TSan-instrumented; when a signal fires
@@ -302,6 +369,8 @@ class ProfilerTestPlugin : Plugin<Project> {
 
                 // JVM args
                 allArgs.addAll(testConfig.standardJvmArgs)
+                // Version-gated at execution time, when the real test JVM (JAVA_TEST_HOME) is resolvable.
+                allArgs.addAll(carrierExportJvmArgs(project))
                 if (extension.nativeLibDir.isPresent) {
                     allArgs.add("-Djava.library.path=${extension.nativeLibDir.get().asFile.absolutePath}")
                 }
@@ -661,7 +730,13 @@ abstract class ProfilerTestExtension @Inject constructor(
     abstract val applicationMainClass: Property<String>
 
     init {
-        // Standard JVM arguments for profiler testing
+        // Standard JVM arguments for profiler testing.
+        // NOTE: JDK-version-gated flags (e.g. the carrier-scoping --add-exports) must NOT be
+        // added here. This convention is computed at configuration time, where JAVA_TEST_HOME
+        // is not yet resolvable and PlatformUtils.testJavaHome() falls back to the *build* JDK
+        // (JAVA_HOME) — which misdetects in the musl split-JDK CI (build JDK 21, test JDK 8) and
+        // would emit a JDK-21 flag onto a JDK-8 test JVM. Version-gated flags are added at
+        // execution time in the task doFirst blocks instead (see ProfilerTestPlugin).
         standardJvmArgs.convention(listOf(
             "-Djdk.attach.allowAttachSelf",      // Allow profiler to attach to self
             "-Djol.tryWithSudo=true",            // JOL memory layout analysis

ddprof-lib/src/main/cpp/ctimer.h

@@ -25,9 +25,15 @@
 #include <signal.h>
 
 class CTimer : public Engine {
+private:
+  // cppcheck-suppress unusedPrivateFunction
+  static void signalHandler(int signo, siginfo_t *siginfo, void *ucontext);
+
 protected:
-  // This is accessed from signal handlers, so must be async-signal-safe
+  // Accessed from signal handlers (including CTimerJvmti subclass), so must
+  // be async-signal-safe. Mutated via enableEvents().
   static bool _enabled;
+
   static long _interval;
   static CStack _cstack;
   static int _signal;
@@ -38,10 +44,6 @@ class CTimer : public Engine {
   int registerThread(int tid);
   void unregisterThread(int tid);
 
-private:
-  // cppcheck-suppress unusedPrivateFunction
-  static void signalHandler(int signo, siginfo_t *siginfo, void *ucontext);
-
 public:
   const char *units() { return "ns"; }
 

ddprof-lib/src/main/cpp/ctimer_linux.cpp

@@ -20,6 +20,7 @@
 #include "counters.h"
 #include "guards.h"
 #include "ctimer.h"
+#include "signalInflight.h"
 #include "debugSupport.h"
 #include "jvmThread.h"
 #include "libraries.h"
@@ -180,6 +181,8 @@ void CTimer::stop() {
   for (int i = 0; i < _max_timers; i++) {
     unregisterThread(i);
   }
+  // Note: SignalInflight::drain() is called by Profiler::stop() after
+  // disableEngines(); see signalInflight.h.
 }
 
 Error CTimerJvmti::check(Arguments &args) {
@@ -210,6 +213,8 @@ void CTimerJvmti::signalHandler(int signo, siginfo_t *siginfo, void *ucontext) {
   }
   Counters::increment(CTIMER_SIGNAL_OWN);
 
+  InflightGuard inflight;
+
   CriticalSection cs;
   if (!cs.entered()) {
     return;
@@ -261,6 +266,8 @@ void CTimer::signalHandler(int signo, siginfo_t *siginfo, void *ucontext) {
   }
   Counters::increment(CTIMER_SIGNAL_OWN);
 
+  InflightGuard inflight;
+
   // Atomically try to enter critical section - prevents all reentrancy races
   CriticalSection cs;
   if (!cs.entered()) {
@@ -270,8 +277,9 @@ void CTimer::signalHandler(int signo, siginfo_t *siginfo, void *ucontext) {
   int saved_errno = errno;
   // we want to ensure memory order because of the possibility the instance gets
   // cleared
-  if (!__atomic_load_n(&_enabled, __ATOMIC_ACQUIRE))
+  if (!__atomic_load_n(&_enabled, __ATOMIC_ACQUIRE)) {
     return;
+  }
   int tid = 0;
   ProfiledThread *current = ProfiledThread::currentSignalSafe();
   assert(current == nullptr || !current->isDeepCrashHandler());

ddprof-lib/src/main/cpp/itimer.cpp

@@ -20,13 +20,14 @@
 #include "jvmThread.h"
 #include "os.h"
 #include "profiler.h"
+#include "signalInflight.h"
 #include "stackWalker.h"
 #include "threadLocalData.h"
 #include "threadState.inline.h"
 #include "guards.h"
 #include <sys/time.h>
 
-volatile bool ITimer::_enabled = false;
+bool ITimer::_enabled = false;
 long ITimer::_interval;
 CStack ITimer::_cstack;
 
@@ -38,7 +39,8 @@ void ITimer::signalHandler(int signo, siginfo_t *siginfo, void *ucontext) {
   // is therefore vulnerable to the foreign-SIGPROF deadlock scenario this
   // feature addresses. Use CTimer (the default) when signal-origin
   // validation is required.
-  if (!_enabled)
+  InflightGuard inflight;
+  if (!__atomic_load_n(&_enabled, __ATOMIC_ACQUIRE))
     return;
 
   // Atomically try to enter critical section - prevents all reentrancy races
@@ -99,11 +101,12 @@ void ITimer::stop() {
   setitimer(ITIMER_PROF, &tv, NULL);
 }
 
-volatile bool ITimerJvmti::_enabled = false;
+bool ITimerJvmti::_enabled = false;
 long ITimerJvmti::_interval = 0;
 
 void ITimerJvmti::signalHandler(int signo, siginfo_t *siginfo, void *ucontext) {
   SIGNAL_HANDLER_GUARD();
+  InflightGuard inflight;
   CriticalSection cs;
   if (!cs.entered()) {
     return;

ddprof-lib/src/main/cpp/itimer.h

@@ -22,7 +22,7 @@
 
 class ITimer : public Engine {
 private:
-  static volatile bool _enabled;
+  static bool _enabled;
   static long _interval;
   static CStack _cstack;
 
@@ -39,7 +39,9 @@ class ITimer : public Engine {
   Error start(Arguments &args);
   void stop();
 
-  inline void enableEvents(bool enabled) { _enabled = enabled; }
+  inline void enableEvents(bool enabled) {
+    __atomic_store_n(&_enabled, enabled, __ATOMIC_RELEASE);
+  }
 };
 
 // CPU-time engine identical to ITimer in its timer mechanism (process-wide
@@ -51,7 +53,7 @@ class ITimer : public Engine {
 // stack walking rather than relying on the signal-frame PC.
 class ITimerJvmti : public Engine {
 private:
-  static volatile bool _enabled;
+  static bool _enabled;
   static long _interval;
 
   static void signalHandler(int signo, siginfo_t *siginfo, void *ucontext);

ddprof-lib/src/main/cpp/perfEvents.h

@@ -30,7 +30,7 @@ class StackContext;
 
 class PerfEvents : public Engine {
 private:
-  static volatile bool _enabled;
+  static bool _enabled;
   static int _max_events;
   static PerfEvent *_events;
   static PerfEventType *_event_type;
@@ -62,7 +62,9 @@ class PerfEvents : public Engine {
 
   static const char *getEventName(int event_id);
 
-  inline void enableEvents(bool enabled) { _enabled = enabled; }
+  inline void enableEvents(bool enabled) {
+    __atomic_store_n(&_enabled, enabled, __ATOMIC_RELEASE);
+  }
 };
 
 #else

ddprof-lib/src/main/cpp/perfEvents_linux.cpp

@@ -29,6 +29,7 @@
 #include "os.h"
 #include "perfEvents.h"
 #include "profiler.h"
+#include "signalInflight.h"
 #include "spinLock.h"
 #include "stackFrame.h"
 #include "stackWalker.h"
@@ -559,7 +560,7 @@ class PerfEvent : public SpinLock {
   friend class PerfEvents;
 };
 
-volatile bool PerfEvents::_enabled = false;
+bool PerfEvents::_enabled = false;
 int PerfEvents::_max_events = -1;
 PerfEvent *PerfEvents::_events = NULL;
 PerfEventType *PerfEvents::_event_type = NULL;
@@ -734,6 +735,7 @@ void PerfEvents::signalHandler(int signo, siginfo_t *siginfo, void *ucontext) {
     // Looks like an external signal; don't treat as a profiling event
     return;
   }
+  InflightGuard inflight;
   // Atomically try to enter critical section - prevents all reentrancy races
   CriticalSection cs;
   if (!cs.entered()) {
@@ -744,7 +746,7 @@ void PerfEvents::signalHandler(int signo, siginfo_t *siginfo, void *ucontext) {
     current->noteCPUSample(Profiler::instance()->recordingEpoch());
   }
   int tid = current != NULL ? current->tid() : OS::threadId();
-  if (_enabled) {
+  if (__atomic_load_n(&_enabled, __ATOMIC_ACQUIRE)) {
     Shims::instance().setSighandlerTid(tid);
 
     u64 counter = readCounter(siginfo, ucontext);