feat: emit TaskBlock events for JVM blocking states

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
 

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

@@ -70,7 +70,8 @@ class JniString {
 };
 
 extern "C" DLLEXPORT jboolean JNICALL
-Java_com_datadoghq_profiler_JavaProfiler_init0(JNIEnv *env, jclass unused) {
+Java_com_datadoghq_profiler_JavaProfiler_init0(
+    JNIEnv *env, jclass unused, jboolean delegateMonitorWaitEvents) {
   Error error = Profiler::instance()->init();
   if (error) {
     throwNew(env, "java/lang/IllegalStateException", error.message());
@@ -79,13 +80,22 @@ Java_com_datadoghq_profiler_JavaProfiler_init0(JNIEnv *env, jclass unused) {
 
 
   // JavaVM* has already been stored when the native library was loaded so we can pass nullptr here
-  if (VM::initProfilerBridge(nullptr, true)) {
-    // Attach ProfiledThread
-    ProfiledThread::initCurrentThreadSignalSafe();
-    return JNI_TRUE;
-  } else {
+  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;
+  }
+  // Attach ProfiledThread
+  ProfiledThread::initCurrentThreadSignalSafe();
+  return JNI_TRUE;
 }
 
 extern "C" DLLEXPORT void JNICALL
@@ -110,6 +120,12 @@ Java_com_datadoghq_profiler_JavaProfiler_getTid0(JNIEnv *env, jclass unused) {
   return OS::threadId();
 }
 
+extern "C" DLLEXPORT jboolean JNICALL
+Java_com_datadoghq_profiler_JavaProfiler_monitorWaitEventsDelegated0(
+    JNIEnv *env, jclass unused) {
+  return VM::monitorWaitEventsDelegated();
+}
+
 extern "C" DLLEXPORT jstring JNICALL
 Java_com_datadoghq_profiler_JavaProfiler_execute0(JNIEnv *env, jobject unused,
                                                   jstring command) {
@@ -389,51 +405,68 @@ Java_com_datadoghq_profiler_JavaProfiler_recordQueueEnd0(
 }
 
 extern "C" DLLEXPORT jboolean JNICALL
-Java_com_datadoghq_profiler_JavaProfiler_parkEnter0(JNIEnv *env, jclass unused) {
+Java_com_datadoghq_profiler_JavaProfiler_parkEnter0(
+    JNIEnv *env, jclass unused, jthread thread, jboolean isVirtual) {
+  // Virtuality is resolved once on the Java side; re-deriving it here would cost a
+  // GetVersion() plus an IsVirtualThread() JNI round-trip on every park.
+  if (isVirtual != JNI_FALSE) {
+    return JNI_FALSE;
+  }
   ProfiledThread *current = ProfiledThread::initCurrentThreadSignalSafe();
   if (current == nullptr) {
     return JNI_FALSE;
   }
+  Context context = ContextApi::snapshot();
+  if (!current->parkEnter(TSC::ticks(), context)) {
+    return JNI_FALSE;
+  }
 
-  bool first_park = current->parkEnter();
-  ThreadFilter *tf = Profiler::instance()->threadFilter();
-  if (first_park && tf->registryActive()) {
+  Profiler *profiler = Profiler::instance();
+  ThreadFilter *tf = profiler->threadFilter();
+  if (context.spanId == 0 && tf->registryActive() &&
+      (profiler->taskBlockEnabled() || tf->enabled())) {
     ThreadFilter::SlotID slot_id = tf->ensureCurrentThreadSlot(current);
     if (slot_id >= 0) {
-      current->setParkBlockToken(
-          tf->enterBlockedRun(slot_id, OSThreadState::CONDVAR_WAIT));
+      current->setParkBlockToken(tf->enterBlockedRun(
+          slot_id, OSThreadState::CONDVAR_WAIT, BlockRunOwner::JAVA));
     }
   }
-  return first_park ? JNI_TRUE : JNI_FALSE;
+  return JNI_TRUE;
 }
 
 extern "C" DLLEXPORT void JNICALL
 Java_com_datadoghq_profiler_JavaProfiler_parkExit0(
-    JNIEnv *env, jclass unused, jlong blocker, jlong unblockingSpanId) {
+    JNIEnv *env, jclass unused, jthread thread, jboolean isVirtual,
+    jlong blocker, jlong unblockingSpanId) {
+  if (isVirtual != JNI_FALSE) {
+    return;
+  }
   ProfiledThread *current = ProfiledThread::initCurrentThreadSignalSafe();
   if (current == nullptr) {
     return;
   }
-
+  u64 start_ticks = 0;
   u64 park_block_token = 0;
-  if (!current->parkExit(park_block_token) || park_block_token == 0) {
+  Context context{};
+  if (!current->parkExit(start_ticks, context, park_block_token) ||
+      park_block_token == 0) {
     return;
   }
-  ThreadFilter *tf = Profiler::instance()->threadFilter();
-  if (tf->registryActive()) {
-    ThreadFilter::SlotID slot_id = ThreadFilter::tokenSlotId(park_block_token);
-    if (tf->activeSlotForId(current->filterSlotId(), current->tid()) != nullptr &&
-        current->filterSlotId() == slot_id) {
-      tf->exitBlockedRun(slot_id, ThreadFilter::tokenGeneration(park_block_token));
-    }
-  }
+  finishTaskBlockAtExit(current, Profiler::instance()->threadFilter(), thread,
+                        1, park_block_token, start_ticks, context,
+                        static_cast<u64>(blocker),
+                        static_cast<u64>(unblockingSpanId));
 }
 
 static bool decodeJavaBlockState(jint state, OSThreadState &decoded) {
   if (state == static_cast<jint>(OSThreadState::SLEEPING)) {
     decoded = OSThreadState::SLEEPING;
     return true;
   }
+  if (state == static_cast<jint>(OSThreadState::OBJECT_WAIT)) {
+    decoded = OSThreadState::OBJECT_WAIT;
+    return true;
+  }
   decoded = OSThreadState::UNKNOWN;
   return false;
 }
@@ -454,13 +487,14 @@ static bool isCurrentJniThread(JNIEnv* env, jthread thread) {
 
 extern "C" DLLEXPORT jlong JNICALL
 Java_com_datadoghq_profiler_JavaProfiler_blockEnter0(
-    JNIEnv *env, jclass unused, jint state) {
-  ProfiledThread *current = ProfiledThread::initCurrentThreadSignalSafe();
-  if (current == nullptr) {
+    JNIEnv *env, jclass unused, jthread thread, jboolean isVirtual,
+    jint state) {
+  OSThreadState decoded;
+  if (!decodeJavaBlockState(state, decoded) || isVirtual != JNI_FALSE) {
     return 0;
   }
-  OSThreadState decoded;
-  if (!decodeJavaBlockState(state, decoded)) {
+  ProfiledThread *current = ProfiledThread::initCurrentThreadSignalSafe();
+  if (current == nullptr) {
     return 0;
   }
   u64 span_id = 0, root_span_id = 0;
@@ -480,9 +514,10 @@ Java_com_datadoghq_profiler_JavaProfiler_blockEnter0(
 
 extern "C" DLLEXPORT void JNICALL
 Java_com_datadoghq_profiler_JavaProfiler_blockExit0(
-    JNIEnv *env, jclass unused, jlong token) {
+    JNIEnv *env, jclass unused, jthread thread, jboolean isVirtual,
+    jlong token) {
   u64 block_token = static_cast<u64>(token);
-  if (block_token == 0) {
+  if (block_token == 0 || isVirtual != JNI_FALSE) {
     return;
   }
 

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

@@ -18,6 +18,8 @@
 
 #include <jni.h>
 
+#include <atomic>
+
 using JniFunction = void (JNICALL*)();
 using IsVirtualThreadFunction = jboolean (JNICALL*)(JNIEnv*, jobject);
 
@@ -44,11 +46,21 @@ bool JVMSupport::isPlatformThread(JNIEnv* jni, jthread thread) {
 
     const JniFunction* functions =
         reinterpret_cast<const JniFunction*>(jni->functions);
+    if (functions == nullptr) return false;
     IsVirtualThreadFunction is_virtual_thread =
         reinterpret_cast<IsVirtualThreadFunction>(
             functions[IS_VIRTUAL_THREAD_INDEX]);
-    return is_virtual_thread != nullptr &&
-        is_virtual_thread(jni, thread) == JNI_FALSE;
+    if (is_virtual_thread == nullptr) {
+        static std::atomic<bool> warning_emitted{false};
+        bool expected = false;
+        if (warning_emitted.compare_exchange_strong(expected, true,
+                                                     std::memory_order_relaxed)) {
+            LOG_WARN("JNI version 19 or later does not expose IsVirtualThread; "
+                     "JVM producer callbacks will be ignored");
+        }
+        return false;
+    }
+    return is_virtual_thread(jni, thread) == JNI_FALSE;
 }
 
 bool JVMSupport::initialize() {

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

@@ -1541,6 +1541,29 @@ Error Profiler::init() {
   return Error::OK;
 }
 
+void Profiler::setTaskBlockEnabled(bool enabled) {
+  if (enabled) {
+    // Keep callback admission closed until native setup has either completed
+    // or rolled back, so partial event enablement cannot create paired state.
+    bool monitor_events_enabled =
+        VM::nativeMonitorEventsAvailable() &&
+        VM::setNativeMonitorEventsEnabled(true);
+    _task_block_monitor_events_enabled.store(monitor_events_enabled,
+                                             std::memory_order_release);
+    _task_block_enabled.store(true, std::memory_order_release);
+    return;
+  }
+
+  _task_block_enabled.store(false, std::memory_order_release);
+  // Clear the admission flag first so no consumer can observe enabled events, then
+  // always attempt teardown. A previous enable whose setup AND rollback both failed
+  // left the flag false while JVMTI events stayed on; retrying unconditionally is the
+  // only way that leak is ever reclaimed. setNativeMonitorEventsEnabled(false) is
+  // documented as a no-op when the capability was never enabled.
+  _task_block_monitor_events_enabled.exchange(false, std::memory_order_acq_rel);
+  VM::setNativeMonitorEventsEnabled(false);
+}
+
 Error Profiler::start(Arguments &args, bool reset) {
   MutexLocker ml(_state_lock);
   Error error = checkState();
@@ -1876,9 +1899,8 @@ Error Profiler::start(Arguments &args, bool reset) {
     // Paired with drainInflight() on the stop side.
     _cpu_engine->enableEvents(true);
 
-    _task_block_enabled.store(
-        (activated & EM_WALL) && args._wall_precheck && track_unfiltered_wall,
-        std::memory_order_release);
+    setTaskBlockEnabled(
+        (activated & EM_WALL) && args._wall_precheck && track_unfiltered_wall);
     _state.store(RUNNING, std::memory_order_release);
     _start_time = time(NULL);
     __atomic_add_fetch(&_epoch, 1, __ATOMIC_RELAXED);
@@ -1903,7 +1925,7 @@ Error Profiler::stop() {
   if (state() != RUNNING) {
     return Error("Profiler is not active");
   }
-  _task_block_enabled.store(false, std::memory_order_release);
+  setTaskBlockEnabled(false);
 
   // Order matters: disable engines first so the _enabled check inside signal
   // handlers will fail for any new signal delivered from now on. drain() then
@@ -2090,7 +2112,9 @@ Error Profiler::dump(const char *path, const int length) {
     // rotateDictsAndRun rotates the dictionaries, takes lockAll() around the
     // dump (fences ASGCT/JNI writers to CallTraceStorage), then clearStandby()s
     // the rotated buffers.  StringDictionary's RefCountGuard protocol handles
-    // its own writer/reader coordination.
+    // its own writer/reader coordination; #527's classMapSharedGuard readers
+    // (deferred vtable receiver resolution) are coordinated through
+    // _class_map_lock.
     if (beginTaskBlockRotation()) {
       rotateDictsAndRun([&]{
         err = _jfr.dump(path, length);

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

@@ -133,6 +133,7 @@ class alignas(alignof(SpinLock)) Profiler {
   alignas(DEFAULT_CACHE_LINE_SIZE) u64 _failures[ASGCT_FAILURE_TYPES];
   bool _wall_precheck = false;
   std::atomic<bool> _task_block_enabled{false};
+  std::atomic<bool> _task_block_monitor_events_enabled{false};
   std::atomic<bool> _task_block_rotation{false};
   std::atomic<u64> _task_block_inflight{0};
 
@@ -185,6 +186,7 @@ class alignas(alignof(SpinLock)) Profiler {
 
   void lockAll();
   void unlockAll();
+  void setTaskBlockEnabled(bool enabled);
   bool beginTaskBlockRotation();
   void endTaskBlockRotation();
 
@@ -494,6 +496,9 @@ class alignas(alignof(SpinLock)) Profiler {
   bool taskBlockEnabled() const {
     return _task_block_enabled.load(std::memory_order_acquire);
   }
+  bool nativeMonitorTaskBlockEnabled() const {
+    return _task_block_monitor_events_enabled.load(std::memory_order_acquire);
+  }
   void writeLog(LogLevel level, const char *message);
   void writeLog(LogLevel level, const char *message, size_t len);
   void writeDatadogProfilerSetting(int tid, int length, const char *name,

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

@@ -42,7 +42,8 @@ bool finishTaskBlockAtExit(ProfiledThread* current,
                            ThreadFilter* thread_filter, jthread thread,
                            int start_depth, u64 block_token, u64 start_ticks,
                            const Context& context, u64 blocker,
-                           u64 unblocking_span_id) {
+                           u64 unblocking_span_id, u64 end_ticks) {
+  if (end_ticks == 0) end_ticks = TSC::ticks();
   Profiler* profiler = Profiler::instance();
   bool recording_enabled = profiler->taskBlockEnabled();
   TaskBlockActivity activity;
@@ -70,6 +71,6 @@ bool finishTaskBlockAtExit(ProfiledThread* current,
   }
 
   return recordTaskBlockIfEligible(
-      current->tid(), thread, start_depth, start_ticks, TSC::ticks(), context,
+      current->tid(), thread, start_depth, start_ticks, end_ticks, context,
       blocker, unblocking_span_id, snapshot.active_state, true);
 }

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

@@ -24,11 +24,14 @@ bool recordTaskBlockAtExit(ProfiledThread* current, ThreadFilter* thread_filter,
 // Cleanup is deliberately performed even when admission is rejected so an
 // application thread never waits for rotation and suppression cannot be left
 // armed.
+// 'end_ticks' lets a caller that already had to sample the clock (e.g. to decide
+// whether the interval is worth resolving a blocker identity for) share the exact
+// same end timestamp with the eligibility check; 0 means "sample it here".
 bool finishTaskBlockAtExit(ProfiledThread* current,
                            ThreadFilter* thread_filter, jthread thread,
                            int start_depth, u64 block_token, u64 start_ticks,
                            const Context& context, u64 blocker,
-                           u64 unblocking_span_id);
+                           u64 unblocking_span_id, u64 end_ticks = 0);
 
 class TaskBlockActivity {
  private:

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

@@ -40,6 +40,9 @@ enum class BlockRunOwner : int {
 
 struct BlockRunSnapshot {
     OSThreadState active_state{OSThreadState::UNKNOWN};
+    BlockRunOwner owner{BlockRunOwner::NONE};
+    u64 generation{0};
+    bool active{false};
     bool context_eligible{false};
 };
 
@@ -285,6 +288,10 @@ class ThreadFilter {
         inline BlockRunSnapshot snapshotBlockRun() const {
             BlockRunSnapshot snapshot;
             snapshot.active_state = activeBlockState();
+            snapshot.owner = activeBlockOwner();
+            snapshot.generation = blockGeneration();
+            snapshot.active = snapshot.owner != BlockRunOwner::NONE &&
+                snapshot.active_state != OSThreadState::UNKNOWN;
             snapshot.context_eligible = activeBlockRemainedOutsideContextWindow();
             return snapshot;
         }