add early thread-state check and TaskBlock emission for wall-clock

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

@@ -364,6 +364,14 @@ Error Arguments::parse(const char *args) {
         _remote_symbolication = true;
       }
 
+      CASE("wallprecheck")
+      if (value != NULL) {
+        _wall_precheck = strcmp(value, "false") != 0 && strcmp(value, "0") != 0;
+      } else {
+        // No value means disable
+        _wall_precheck = false;
+      }
+
       CASE("wallsampler")
       if (value != NULL) {
           switch (value[0]) {

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

@@ -168,6 +168,7 @@ class Arguments {
   long _cpu;
   long _wall;
   bool _wall_collapsing;
+  bool _wall_precheck;
   int _wall_threads_per_tick;
   WallclockSampler _wallclock_sampler;
   long _memory;
@@ -204,6 +205,7 @@ class Arguments {
         _cpu(-1),
         _wall(-1),
         _wall_collapsing(false),
+        _wall_precheck(false),
         _wall_threads_per_tick(DEFAULT_WALL_THREADS_PER_TICK),
         _wallclock_sampler(ASGCT),
         _memory(-1),

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

@@ -60,6 +60,14 @@
   X(AGCT_NATIVE_NO_JAVA_CONTEXT, "agct_native_no_java_context")                \
   X(AGCT_BLOCKED_IN_VM, "agct_blocked_in_vm")                                  \
   X(SKIPPED_WALLCLOCK_UNWINDS, "skipped_wallclock_unwinds")                    \
+  X(WC_SIGNAL_SUPPRESSED_SAMPLED_RUN, "wc_signals_suppressed_sampled_run")     \
+  X(WC_SIGNAL_SKIPPED_PARKED, "wc_signals_skipped_parked")                     \
+  X(WC_SIGNAL_SKIPPED_PARKED_SPANLESS, "wc_signals_skipped_parked_spanless")   \
+  X(WC_SIGNAL_SKIPPED_PARKED_ACTIVE_SPAN, "wc_signals_skipped_parked_active_span") \
+  X(WC_SIGNAL_QUEUE_FULL, "wc_signals_queue_full")                             \
+  X(TASK_BLOCK_EMITTED, "task_block_emitted")                                  \
+  X(TASK_BLOCK_SKIPPED_SPAN_ZERO, "task_block_skipped_span_zero")              \
+  X(TASK_BLOCK_SKIPPED_TOO_SHORT, "task_block_skipped_too_short")              \
   X(UNWINDING_TIME_ASYNC, "unwinding_ticks_async")                             \
   X(UNWINDING_TIME_JVMTI, "unwinding_ticks_jvmti")                             \
   X(CALLTRACE_STORAGE_DROPPED, "calltrace_storage_dropped_traces")             \
@@ -102,7 +110,14 @@
   X(CTIMER_SIGNAL_OWN, "ctimer_signal_own")                                     \
   X(CTIMER_SIGNAL_FOREIGN, "ctimer_signal_foreign")                             \
   X(WALLCLOCK_SIGNAL_OWN, "wallclock_signal_own")                               \
-  X(WALLCLOCK_SIGNAL_FOREIGN, "wallclock_signal_foreign")
+  X(WALLCLOCK_SIGNAL_FOREIGN, "wallclock_signal_foreign")                       \
+  /* JVMTI monitor-event callbacks (synchronized contention + Object.wait).    \
+   * Driven by the can_generate_monitor_events capability. Zero when the agent \
+   * has signalled delegate_monitor_events on JDK 21+ (Java-side modules own   \
+   * the populations); non-zero on JDK < 21 or when no delegate is registered. \
+   * Used by MonitorEventDelegationTest to verify the handshake. */            \
+  X(JVMTI_MONITOR_CONTENDED_ENTER, "jvmti_monitor_contended_enter")             \
+  X(JVMTI_OBJECT_WAIT_ENTER, "jvmti_object_wait_enter")
 #define X_ENUM(a, b) a,
 typedef enum CounterId : int {
   DD_COUNTER_TABLE(X_ENUM) DD_NUM_COUNTERS

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

@@ -109,12 +109,21 @@ class WallClockEpochEvent {
   u32 _num_failed_samples;
   u32 _num_exited_threads;
   u32 _num_permission_denied;
+  u64 _num_suppressed_sampled_run;
+  u64 _num_skipped_parked_spanless;
+  u64 _num_skipped_parked_active_span;
+  u64 _num_task_block_emitted;
+  u64 _num_task_block_skipped_span_zero;
+  u64 _num_task_block_skipped_too_short;
 
   WallClockEpochEvent(u64 start_time)
       : _dirty(false), _start_time(start_time), _duration_millis(0),
         _num_samplable_threads(0), _num_successful_samples(0),
         _num_failed_samples(0), _num_exited_threads(0),
-        _num_permission_denied(0) {}
+        _num_permission_denied(0), _num_suppressed_sampled_run(0),
+        _num_skipped_parked_spanless(0), _num_skipped_parked_active_span(0),
+        _num_task_block_emitted(0), _num_task_block_skipped_span_zero(0),
+        _num_task_block_skipped_too_short(0) {}
 
   bool hasChanged() { return _dirty; }
 
@@ -153,6 +162,48 @@ class WallClockEpochEvent {
     }
   }
 
+  void updateNumSuppressedSampledRun(u64 n) {
+    if (_num_suppressed_sampled_run != n) {
+      _dirty = true;
+      _num_suppressed_sampled_run = n;
+    }
+  }
+
+  void updateNumSkippedParkedSpanless(u64 n) {
+    if (_num_skipped_parked_spanless != n) {
+      _dirty = true;
+      _num_skipped_parked_spanless = n;
+    }
+  }
+
+  void updateNumSkippedParkedActiveSpan(u64 n) {
+    if (_num_skipped_parked_active_span != n) {
+      _dirty = true;
+      _num_skipped_parked_active_span = n;
+    }
+  }
+
+  void updateNumTaskBlockEmitted(u64 n) {
+    if (_num_task_block_emitted != n) {
+      _dirty = true;
+      _num_task_block_emitted = n;
+    }
+  }
+
+  void updateNumTaskBlockSkippedSpanZero(u64 n) {
+    if (_num_task_block_skipped_span_zero != n) {
+      _dirty = true;
+      _num_task_block_skipped_span_zero = n;
+    }
+  }
+
+  void updateNumTaskBlockSkippedTooShort(u64 n) {
+    if (_num_task_block_skipped_too_short != n) {
+      _dirty = true;
+      _num_task_block_skipped_too_short = n;
+    }
+  }
+
   void endEpoch(u64 millis) { _duration_millis = millis; }
 
   void clean() { _dirty = false; }
@@ -184,4 +235,13 @@ typedef struct QueueTimeEvent {
   u32 _queueLength;
 } QueueTimeEvent;
 
+typedef struct TaskBlockEvent {
+  u64 _start;
+  u64 _end;
+  u64 _blocker;
+  u64 _unblockingSpanId;
+  /** Span IDs and tag encodings for JFR (park exit uses snapshot from park enter). */
+  Context _ctx;
+} TaskBlockEvent;
+
 #endif // _EVENT_H

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

@@ -1535,6 +1535,12 @@ void Recording::recordWallClockEpoch(Buffer *buf, WallClockEpochEvent *event) {
   buf->putVar64(event->_num_failed_samples);
   buf->putVar64(event->_num_exited_threads);
   buf->putVar64(event->_num_permission_denied);
+  buf->putVar64(event->_num_suppressed_sampled_run);
+  buf->putVar64(event->_num_skipped_parked_spanless);
+  buf->putVar64(event->_num_skipped_parked_active_span);
+  buf->putVar64(event->_num_task_block_emitted);
+  buf->putVar64(event->_num_task_block_skipped_span_zero);
+  buf->putVar64(event->_num_task_block_skipped_too_short);
   writeEventSizePrefix(buf, start);
   flushIfNeeded(buf);
 }
@@ -1570,6 +1576,39 @@ void Recording::recordQueueTime(Buffer *buf, int tid, QueueTimeEvent *event) {
   flushIfNeeded(buf);
 }
 
+void Recording::recordTaskBlockLive(Buffer *buf, int tid, TaskBlockEvent *event) {
+  // Live capture: TaskBlock-specific fields first, then the standard context block via
+  // writeCurrentContext — same shape as recordQueueTime.
+  int start = buf->skip(1);
+  buf->putVar64(T_TASK_BLOCK);
+  buf->putVar64(event->_start);
+  buf->putVar64(event->_end - event->_start);
+  buf->putVar64(tid);
+  buf->putVar64(event->_blocker);
+  buf->putVar64(event->_unblockingSpanId);
+  writeCurrentContext(buf);
+  writeEventSizePrefix(buf, start);
+  flushIfNeeded(buf);
+}
+
+void Recording::recordTaskBlockDeferred(Buffer *buf, int tid,
+                                        TaskBlockEvent *event) {
+  // Deferred capture: context was snapshotted earlier (parkEnter / ObjectWaitEnter from
+  // JVMTI MonitorWait) and is emitted from event->_ctx via writeContextSnapshot — same
+  // on-disk layout as the live path, distinct only in the source of the spanId/rootSpanId/tag
+  // triple.
+  int start = buf->skip(1);
+  buf->putVar64(T_TASK_BLOCK);
+  buf->putVar64(event->_start);
+  buf->putVar64(event->_end - event->_start);
+  buf->putVar64(tid);
+  buf->putVar64(event->_blocker);
+  buf->putVar64(event->_unblockingSpanId);
+  writeContextSnapshot(buf, event->_ctx);
+  writeEventSizePrefix(buf, start);
+  flushIfNeeded(buf);
+}
+
 void Recording::recordAllocation(RecordingBuffer *buf, int tid,
                                  u64 call_trace_id, AllocEvent *event) {
   int start = buf->skip(1);
@@ -1772,6 +1811,7 @@ void FlightRecorder::recordTraceRoot(int lock_index, int tid,
     Recording* rec = _rec;
     if (rec != nullptr) {
       Buffer *buf = rec->buffer(lock_index);
+      rec->addThread(lock_index, tid);
       rec->recordTraceRoot(buf, tid, event);
     }
   }
@@ -1784,11 +1824,39 @@ void FlightRecorder::recordQueueTime(int lock_index, int tid,
     Recording* rec = _rec;
     if (rec != nullptr) {
       Buffer *buf = rec->buffer(lock_index);
+      rec->addThread(lock_index, tid);
+      rec->addThread(lock_index, event->_origin);
       rec->recordQueueTime(buf, tid, event);
     }
   }
 }
 
+void FlightRecorder::recordTaskBlockLive(int lock_index, int tid,
+                                         TaskBlockEvent *event) {
+  OptionalSharedLockGuard locker(&_rec_lock);
+  if (locker.ownsLock()) {
+    Recording* rec = _rec;
+    if (rec != nullptr) {
+      Buffer *buf = rec->buffer(lock_index);
+      rec->addThread(lock_index, tid);
+      rec->recordTaskBlockLive(buf, tid, event);
+    }
+  }
+}
+
+void FlightRecorder::recordTaskBlockDeferred(int lock_index, int tid,
+                                             TaskBlockEvent *event) {
+  OptionalSharedLockGuard locker(&_rec_lock);
+  if (locker.ownsLock()) {
+    Recording* rec = _rec;
+    if (rec != nullptr) {
+      Buffer *buf = rec->buffer(lock_index);
+      rec->addThread(lock_index, tid);
+      rec->recordTaskBlockDeferred(buf, tid, event);
+    }
+  }
+}
+
 void FlightRecorder::recordDatadogSetting(int lock_index, int length,
                                           const char *name, const char *value,
                                           const char *unit) {

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

@@ -279,6 +279,8 @@ class Recording {
   void recordWallClockEpoch(Buffer *buf, WallClockEpochEvent *event);
   void recordTraceRoot(Buffer *buf, int tid, TraceRootEvent *event);
   void recordQueueTime(Buffer *buf, int tid, QueueTimeEvent *event);
+  void recordTaskBlockLive(Buffer *buf, int tid, TaskBlockEvent *event);
+  void recordTaskBlockDeferred(Buffer *buf, int tid, TaskBlockEvent *event);
   void recordAllocation(RecordingBuffer *buf, int tid, u64 call_trace_id,
                         AllocEvent *event);
   void recordMallocSample(Buffer *buf, int tid, u64 call_trace_id,
@@ -347,6 +349,8 @@ class FlightRecorder {
   void wallClockEpoch(int lock_index, WallClockEpochEvent *event);
   void recordTraceRoot(int lock_index, int tid, TraceRootEvent *event);
   void recordQueueTime(int lock_index, int tid, QueueTimeEvent *event);
+  void recordTaskBlockLive(int lock_index, int tid, TaskBlockEvent *event);
+  void recordTaskBlockDeferred(int lock_index, int tid, TaskBlockEvent *event);
 
   bool active() const { return _rec != NULL; }
 

ddprof-lib/src/main/cpp/hotspot/vmStructs.inline.h

@@ -9,12 +9,21 @@
 
 #include "hotspot/vmStructs.h"
 #include "jvmThread.h"
+#include "vmEntry.h"
 
 VMThread* VMThread::current() {
+    // JVMThread::current() is the native thread self pointer. On OpenJ9/Zing it
+    // is not a HotSpot JavaThread*; only HotSpot may reinterpret it as VMThread*.
+    if (!VM::isHotspot() || JVMThread::current() == nullptr) {
+        return nullptr;
+    }
     return VMThread::cast(JVMThread::current());
 }
 
 VMThread* VMThread::fromJavaThread(JNIEnv* env, jthread thread) {
+    if (!VM::isHotspot()) {
+        return nullptr;
+    }
     assert(_eetop != nullptr);
     if (_eetop != nullptr) {
         jlong eetop = env->GetLongField(thread, _eetop);

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

@@ -24,12 +24,13 @@
 #include "counters.h"
 #include "common.h"
 #include "engine.h"
-#include "hotspot/vmStructs.h"
+#include "hotspot/vmStructs.inline.h"
 #include "incbin.h"
 #include "jvmThread.h"
 #include "os.h"
 #include "otel_process_ctx.h"
 #include "profiler.h"
+#include "taskBlockRecorder.h"
 #include "thread.h"
 #include "tsc.h"
 #include "vmEntry.h"
@@ -67,9 +68,14 @@ class JniString {
 };
 
 extern "C" DLLEXPORT jboolean JNICALL
-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
-  return VM::initProfilerBridge(nullptr, true);
+Java_com_datadoghq_profiler_JavaProfiler_init0(JNIEnv *env, jclass unused,
+                                               jboolean delegate_monitor_events) {
+  // JavaVM* has already been stored when the native library was loaded so we can pass nullptr here.
+  // delegate_monitor_events lets dd-trace-java signal that its Java-side instrumentation modules
+  // (object-wait, synchronized-contention) will cover the populations the JVMTI MonitorWait /
+  // MonitorContended* callbacks observe, so the native callbacks can be skipped on JDK 21+ to
+  // avoid the can_generate_monitor_events per-monitor instrumentation cost.
+  return VM::initProfilerBridge(nullptr, true, delegate_monitor_events == JNI_TRUE);
 }
 
 extern "C" DLLEXPORT void JNICALL
@@ -150,15 +156,21 @@ JavaCritical_com_datadoghq_profiler_JavaProfiler_filterThreadAdd0() {
 
   int slot_id = current->filterSlotId();
   if (unlikely(slot_id == -1)) {
-    // Thread doesn't have a slot ID yet (e.g., main thread), so register it
-    // Happens when we are not enabled before thread start
+    // Thread doesn't have a slot ID yet (e.g., main thread or profiler started
+    // after thread creation). Register now.
     slot_id = thread_filter->registerThread();
     current->setFilterSlotId(slot_id);
   }
-  
+
   if (unlikely(slot_id == -1)) {
     return;  // Failed to register thread
   }
+  // Refresh HotSpot VMThread* for wall thread-filter precheck (vmStructs OS state).
+  // HotSpot only: VMThread::current() asserts VM::isHotspot(). OpenJ9/Zing: leave null.
+  thread_filter->setVMThread(slot_id, VM::isHotspot() ? VMThread::current() : nullptr);
+  // Refresh ProfiledThread* so wall-clock mitigations can observe per-thread parked state.
+  // Publish pointer fields before publishing tid via add() to preserve visibility ordering.
+  thread_filter->setProfiledThread(slot_id, current);
   thread_filter->add(tid, slot_id);
 }
 
@@ -313,6 +325,47 @@ Java_com_datadoghq_profiler_JavaProfiler_recordQueueEnd0(
   Profiler::instance()->recordQueueTime(tid, &event);
 }
 
+extern "C" DLLEXPORT void JNICALL
+Java_com_datadoghq_profiler_JavaProfiler_recordTaskBlock0(
+    JNIEnv *env, jclass unused, jlong startTicks, jlong endTicks,
+    jlong blocker, jlong unblockingSpanId) {
+  int tid = ProfiledThread::currentTid();
+  if (tid < 0) {
+    return;
+  }
+  // Context (span ids + tags) is read natively from the OTEP TLS sidecar at JFR write time
+  // via writeCurrentContext, mirroring recordQueueTime. No span ids cross JNI.
+  recordTaskBlockLiveIfEligible(tid, (u64)startTicks, (u64)endTicks,
+                                (u64)blocker, (u64)unblockingSpanId);
+}
+
+extern "C" DLLEXPORT void JNICALL
+Java_com_datadoghq_profiler_JavaProfiler_parkEnter0(JNIEnv *env, jclass unused) {
+  ProfiledThread *current = ProfiledThread::current();
+  if (current == nullptr) {
+    return;
+  }
+  current->parkEnter(TSC::ticks());
+}
+
+extern "C" DLLEXPORT void JNICALL
+Java_com_datadoghq_profiler_JavaProfiler_parkExit0(
+    JNIEnv *env, jclass unused, jlong blocker, jlong unblockingSpanId) {
+  ProfiledThread *current = ProfiledThread::current();
+  if (current == nullptr) {
+    return;
+  }
+  u64 start_ticks = 0;
+  Context park_context = {};
+  if (!current->parkExit(start_ticks, park_context)) {
+    return;
+  }
+  u64 end_ticks = TSC::ticks();
+  recordTaskBlockDeferredIfEligible(current->tid(), start_ticks, end_ticks,
+                                    park_context, (u64)blocker,
+                                    (u64)unblockingSpanId);
+}
+
 extern "C" DLLEXPORT jlong JNICALL
 Java_com_datadoghq_profiler_JavaProfiler_currentTicks0(JNIEnv *env,
                                                        jclass unused) {