fix(taskblock): harden JVM producer initialization

Author: kaahos

build-logic/conventions/src/main/kotlin/com/datadoghq/native/config/ConfigurationPresets.kt

@@ -1,3 +1,18 @@
+/*
+ * Copyright 2026, Datadog, Inc.
+ *
+ * Licensed under the Apache License, Version 2.0 (the "License");
+ * you may not use this file except in compliance with the License.
+ * You may obtain a copy of the License at
+ *
+ *     http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
 
 package com.datadoghq.native.config
 
@@ -149,7 +164,7 @@ object ConfigurationPresets {
                 config.compilerArgs.set(
                     listOf("-O0", "-g", "-DDEBUG") + commonLinuxCompilerArgs(version)
                 )
-                config.linkerArgs.set(commonLinuxLinkerArgs())
+                config.linkerArgs.set(commonLinuxLinkerArgs() + listOf("-Wl,-z,nodelete"))
             }
             Platform.MACOS -> {
                 config.compilerArgs.set(

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

@@ -78,7 +78,20 @@ Java_com_datadoghq_profiler_JavaProfiler_init0(
   }
 
   // JavaVM* has already been stored when the native library was loaded so we can pass nullptr here
-  return VM::initProfilerBridge(nullptr, true, delegateMonitorWaitEvents);
+  ProfilerBridgeInitResult result =
+      VM::initProfilerBridge(nullptr, true, delegateMonitorWaitEvents);
+  if (result == ProfilerBridgeInitResult::MONITOR_EVENTS_DELEGATION_CONFLICT) {
+    throwNew(env, "java/lang/IllegalStateException",
+             "Monitor-event ownership conflicts with the profiler's "
+             "process-wide initialization");
+    return JNI_FALSE;
+  }
+  if (result != ProfilerBridgeInitResult::SUCCESS) {
+    throwNew(env, "java/lang/IllegalStateException",
+             "Failed to initialize the profiler bridge");
+    return JNI_FALSE;
+  }
+  return JNI_TRUE;
 }
 
 extern "C" DLLEXPORT void JNICALL
@@ -144,32 +157,6 @@ Java_com_datadoghq_profiler_JavaProfiler_getSamples(JNIEnv *env,
   return (jlong)Profiler::instance()->total_samples();
 }
 
-// some duplication between add and remove, though we want to avoid having an extra branch in the hot path
-
-static ThreadFilter::SlotID ensureCurrentThreadFilterSlot(
-    ThreadFilter *thread_filter, ProfiledThread *current) {
-  int tid = current->tid();
-  if (unlikely(tid < 0)) {
-    return -1;
-  }
-
-  ThreadFilter::SlotID slot_id = current->filterSlotId();
-  if (likely(slot_id >= 0)) {
-    if (likely(thread_filter->activeSlotForId(slot_id, tid) != nullptr)) {
-      return slot_id;
-    }
-    current->setFilterSlotId(-1);
-  }
-
-  // Startup can register this TID centrally, but it cannot update another
-  // pthread's TLS. registerThread(tid) reuses that existing slot.
-  slot_id = thread_filter->registerThread(tid);
-  if (slot_id >= 0) {
-    current->setFilterSlotId(slot_id);
-  }
-  return slot_id;
-}
-
 // JavaCritical is faster JNI, but more restrictive - parameters and return value have to be
 // primitives or arrays of primitive types.
 // We direct corresponding JNI calls to JavaCritical to make sure the parameters/return value
@@ -191,7 +178,7 @@ JavaCritical_com_datadoghq_profiler_JavaProfiler_filterThreadAdd0() {
     return;
   }
 
-  int slot_id = ensureCurrentThreadFilterSlot(thread_filter, current);
+  int slot_id = thread_filter->ensureCurrentThreadSlot(current);
   if (unlikely(slot_id < 0)) {
     return;  // Failed to register thread
   }
@@ -385,7 +372,7 @@ Java_com_datadoghq_profiler_JavaProfiler_parkEnter0(
   ThreadFilter *tf = profiler->threadFilter();
   if (context.spanId == 0 && tf->registryActive() &&
       (profiler->taskBlockEnabled() || tf->enabled())) {
-    ThreadFilter::SlotID slot_id = ensureCurrentThreadFilterSlot(tf, current);
+    ThreadFilter::SlotID slot_id = tf->ensureCurrentThreadSlot(current);
     if (slot_id >= 0) {
       current->setParkBlockToken(tf->enterBlockedRun(
           slot_id, OSThreadState::CONDVAR_WAIT, BlockRunOwner::JAVA));
@@ -412,32 +399,10 @@ Java_com_datadoghq_profiler_JavaProfiler_parkExit0(
     return;
   }
   Profiler *profiler = Profiler::instance();
-  bool recording_enabled = profiler->taskBlockEnabled();
-  bool activity = profiler->tryEnterTaskBlockActivity();
-  if (!activity) profiler->waitForTaskBlockRotation();
-
-  ThreadFilter *tf = profiler->threadFilter();
-  ThreadFilter::SlotID slot_id = ThreadFilter::tokenSlotId(park_block_token);
-  ThreadFilter::SlotID current_slot = current->filterSlotId();
-  if (current_slot < 0) current_slot = tf->slotIdByTid(current->tid());
-  BlockRunSnapshot snapshot{};
-  bool exited = current_slot == slot_id &&
-      tf->snapshotAndExitBlockedRun(
-          slot_id, ThreadFilter::tokenGeneration(park_block_token), &snapshot);
-
-  if (!activity) {
-    Counters::increment(TASK_BLOCK_DROPPED_ROTATION);
-    return;
-  }
-  if (recording_enabled && exited && snapshot.context_eligible) {
-    recordTaskBlockIfEligible(
-        current->tid(), thread, 1, start_ticks, TSC::ticks(), context,
-        static_cast<u64>(blocker), static_cast<u64>(unblockingSpanId),
-        snapshot.active_state, true);
-  } else if (recording_enabled && exited && !snapshot.context_eligible) {
-    Counters::increment(TASK_BLOCK_SKIPPED_TRACE_CONTEXT);
-  }
-  profiler->leaveTaskBlockActivity();
+  finishTaskBlockAtExit(
+      current, profiler->threadFilter(), thread, 1, park_block_token,
+      start_ticks, context, static_cast<u64>(blocker),
+      static_cast<u64>(unblockingSpanId));
 }
 
 static bool decodeJavaBlockState(jint state, OSThreadState &decoded) {
@@ -469,7 +434,7 @@ Java_com_datadoghq_profiler_JavaProfiler_blockEnter0(
   if (!profiler->taskBlockEnabled() && !tf->enabled()) {
     return 0;
   }
-  ThreadFilter::SlotID slot_id = ensureCurrentThreadFilterSlot(tf, current);
+  ThreadFilter::SlotID slot_id = tf->ensureCurrentThreadSlot(current);
   if (slot_id < 0) return 0;
   return static_cast<jlong>(tf->enterBlockedRun(slot_id, decoded));
 }
@@ -510,7 +475,7 @@ Java_com_datadoghq_profiler_JavaProfiler_beginTaskBlock0(
   }
   ThreadFilter *tf = profiler->threadFilter();
   if (!tf->unfilteredWallTrackingActive()) return 0;
-  ThreadFilter::SlotID slot_id = ensureCurrentThreadFilterSlot(tf, current);
+  ThreadFilter::SlotID slot_id = tf->ensureCurrentThreadSlot(current);
   if (slot_id < 0) return 0;
 
   Context context = ContextApi::snapshot();

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

@@ -16,6 +16,7 @@
 #include "jvmThread.h"
 #include "libraries.h"
 #include "log.h"
+#include "mutex.h"
 #include "os.h"
 #include "profiler.h"
 #include "safeAccess.h"
@@ -55,6 +56,12 @@ bool VM::_monitor_events_delegated = false;
 bool VM::_native_monitor_events_available = false;
 bool VM::_is_adaptive_gc_boundary_flag_set = false;
 
+// Serializes the one-time bridge installation and ownership negotiation.
+// Callback readers need no synchronization because ownership is assigned
+// before callbacks can be enabled and is never changed afterward.
+static Mutex profiler_bridge_init_lock;
+static bool profiler_bridge_initialized = false;
+
 jvmtiExtensionFunction VM::_request_stack_trace = nullptr;
 jvmtiExtensionFunction VM::_init_request_stack_trace = nullptr;
 
@@ -86,7 +93,7 @@ static void monitorBlockEnter(jvmtiEnv *jvmti, JNIEnv *jni, jthread thread,
       !JVMSupport::isPlatformThread(jni, thread)) {
     return;
   }
-  ProfiledThread *current = ProfiledThread::current();
+  ProfiledThread *current = ProfiledThread::initCurrentThreadSignalSafe();
   if (current == nullptr) return;
   Context context = ContextApi::snapshot();
   if (context.spanId != 0) {
@@ -101,10 +108,15 @@ static void monitorBlockEnter(jvmtiEnv *jvmti, JNIEnv *jni, jthread thread,
     bool current_owner = false;
     if (token != 0) {
       ThreadFilter::SlotID slot_id = ThreadFilter::tokenSlotId(token);
-      BlockRunSnapshot snapshot = tf->snapshotBlockedRun(slot_id);
-      current_owner = current->filterSlotId() == slot_id && snapshot.active &&
-          snapshot.owner == BlockRunOwner::JVMTI &&
-          snapshot.generation == ThreadFilter::tokenGeneration(token);
+      ThreadFilter::Slot *slot = current->filterSlotId() == slot_id
+          ? tf->activeSlotForId(slot_id, current->tid())
+          : nullptr;
+      if (slot != nullptr) {
+        BlockRunSnapshot snapshot = slot->snapshotBlockRun();
+        current_owner = snapshot.active &&
+            snapshot.owner == BlockRunOwner::JVMTI &&
+            snapshot.generation == ThreadFilter::tokenGeneration(token);
+      }
     }
     if (current_owner) {
       return;
@@ -117,12 +129,8 @@ static void monitorBlockEnter(jvmtiEnv *jvmti, JNIEnv *jni, jthread thread,
   }
 
   ThreadFilter *tf = profiler->threadFilter();
-  ThreadFilter::SlotID slot_id = current->filterSlotId();
-  if (slot_id < 0) {
-    slot_id = tf->slotIdByTid(current->tid());
-    if (slot_id >= 0) current->setFilterSlotId(slot_id);
-  }
-  if (!tf->allThreads() || slot_id < 0) {
+  ThreadFilter::SlotID slot_id = tf->ensureCurrentThreadSlot(current);
+  if (!tf->unfilteredWallTrackingActive() || slot_id < 0) {
     current->clearMonitorBlock();
     return;
   }
@@ -154,31 +162,8 @@ static void monitorBlockExit(JNIEnv *jni, jthread thread, OSThreadState state) {
   }
 
   Profiler *profiler = Profiler::instance();
-  bool recording_enabled = profiler->taskBlockEnabled();
-  bool activity = profiler->tryEnterTaskBlockActivity();
-  if (!activity) profiler->waitForTaskBlockRotation();
-
-  ThreadFilter *tf = profiler->threadFilter();
-  ThreadFilter::SlotID slot_id = ThreadFilter::tokenSlotId(token);
-  ThreadFilter::SlotID current_slot = current->filterSlotId();
-  if (current_slot < 0) current_slot = tf->slotIdByTid(current->tid());
-  BlockRunSnapshot snapshot{};
-  bool exited = current_slot == slot_id &&
-      tf->snapshotAndExitBlockedRun(
-          slot_id, ThreadFilter::tokenGeneration(token), &snapshot);
-
-  if (!activity) {
-    Counters::increment(TASK_BLOCK_DROPPED_ROTATION);
-    return;
-  }
-  if (recording_enabled && exited && snapshot.context_eligible) {
-    recordTaskBlockIfEligible(current->tid(), thread, 0, start_ticks,
-                              TSC::ticks(), context, blocker, 0,
-                              snapshot.active_state, true);
-  } else if (recording_enabled && exited && !snapshot.context_eligible) {
-    Counters::increment(TASK_BLOCK_SKIPPED_TRACE_CONTEXT);
-  }
-  profiler->leaveTaskBlockActivity();
+  finishTaskBlockAtExit(current, profiler->threadFilter(), thread, 0, token,
+                        start_ticks, context, blocker, 0);
 }
 
 static void JNICALL MonitorContendedEnter(jvmtiEnv *jvmti, JNIEnv *jni,
@@ -576,16 +561,25 @@ bool VM::initializeRequestStackTrace() {
   return false;
 }
 
-bool VM::initProfilerBridge(JavaVM *vm, bool attach,
-                            bool delegateMonitorEvents) {
+ProfilerBridgeInitResult VM::initProfilerBridge(JavaVM *vm, bool attach,
+                                                bool delegateMonitorEvents) {
+  MutexLocker init_locker(profiler_bridge_init_lock);
+  if (profiler_bridge_initialized) {
+    bool requested_delegation =
+        delegateMonitorEvents && _native_monitor_events_available;
+    return requested_delegation == _monitor_events_delegated
+        ? ProfilerBridgeInitResult::SUCCESS
+        : ProfilerBridgeInitResult::MONITOR_EVENTS_DELEGATION_CONFLICT;
+  }
+
   TEST_LOG("VM::initProfilerBridge");
   if (!initShared(vm)) {
-    return false;
+    return ProfilerBridgeInitResult::FAILURE;
   }
 
   CodeCache *lib = openJvmLibrary();
   if (lib == nullptr) {
-    return false;
+    return ProfilerBridgeInitResult::FAILURE;
   }
 
   if (!attach && hotspot_version() == 8 && OS::isLinux()) {
@@ -708,7 +702,8 @@ bool VM::initProfilerBridge(JavaVM *vm, bool attach,
 
   OS::installSignalHandler(WAKEUP_SIGNAL, NULL, wakeupHandler);
 
-  return true;
+  profiler_bridge_initialized = true;
+  return ProfilerBridgeInitResult::SUCCESS;
 }
 
 bool VM::setNativeMonitorEventsEnabled(bool enabled) {
@@ -859,7 +854,8 @@ Agent_OnLoad(JavaVM* vm, char* options, void* reserved) {
         return ARGUMENTS_ERROR;
     }
 
-    if (!VM::initProfilerBridge(vm, false)) {
+    if (VM::initProfilerBridge(vm, false) !=
+        ProfilerBridgeInitResult::SUCCESS) {
         Log::error("JVM does not support Tool Interface");
         return COMMAND_ERROR;
     }

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

@@ -132,6 +132,15 @@ class JavaVersionAccess {
    static int get_hotspot_version(char* prop_value);
 };
 
+// The profiler bridge is process-wide and initialized exactly once. Later Java
+// API initialization may reuse it only with the same effective Object.wait
+// ownership.
+enum class ProfilerBridgeInitResult {
+  SUCCESS,
+  FAILURE,
+  MONITOR_EVENTS_DELEGATION_CONFLICT,
+};
+
 class VM {
   friend class VMTestAccessor;
 
@@ -179,8 +188,8 @@ class VM {
   static JVM_GetManagement _getManagement;
 
   static bool initLibrary(JavaVM *vm);
-  static bool initProfilerBridge(JavaVM *vm, bool attach,
-                                 bool delegateMonitorEvents = false);
+  static ProfilerBridgeInitResult initProfilerBridge(
+      JavaVM *vm, bool attach, bool delegateMonitorEvents = false);
 
   static jvmtiEnv *jvmti() { return _jvmti; }
 

ddprof-lib/src/main/java/com/datadoghq/profiler/JavaProfiler.java

@@ -137,8 +137,9 @@ public static synchronized JavaProfiler getInstance(String libLocation, String s
     /**
      * Get a {@linkplain JavaProfiler} instance with explicit monitor-event ownership.
      *
-     * <p>The first successful initialization fixes this process-wide setting because the native
-     * profiler is a singleton. When delegation is enabled, Java instrumentation owns
+     * <p>The first successful native bridge initialization fixes this process-wide setting because
+     * the native profiler is a singleton. This may occur during {@code -agentpath} startup before
+     * this method is called. When delegation is enabled, Java instrumentation owns
      * {@code Object.wait} TaskBlock intervals and native JVMTI wait callbacks are suppressed;
      * native JVMTI callbacks continue to own synchronized monitor contention.
      *
@@ -147,6 +148,8 @@ public static synchronized JavaProfiler getInstance(String libLocation, String s
      * @param delegateMonitorWaitEvents whether Java instrumentation owns {@code Object.wait} intervals
      * @return the process-wide profiler instance
      * @throws IOException if the native library cannot be loaded
+     * @throws IllegalStateException if monitor ownership conflicts with an earlier native bridge
+     *     initialization
      */
     public static synchronized JavaProfiler getInstance(String libLocation, String scratchDir,
             boolean delegateMonitorWaitEvents) throws IOException {