[WIP] add pre-signal state check for wallclock

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

@@ -364,6 +364,11 @@ Error Arguments::parse(const char *args) {
         _remote_symbolication = true;
       }
 
+      CASE("wallprecheck")
+      if (value != NULL) {
+        _wall_precheck = strcmp(value, "false") != 0 && strcmp(value, "0") != 0;
+      }
+
       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(true),
         _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_SKIPPED_PRECHECK_OS, "wc_signals_skipped_precheck_os")          \
+  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")             \

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;
+  u32 _num_skipped_precheck_os;
+  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_skipped_precheck_os(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 updateNumSkippedPrecheckOs(u32 n) {
+    if (_num_skipped_precheck_os != n) {
+      _dirty = true;
+      _num_skipped_precheck_os = 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_skipped_precheck_os);
+  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,19 @@ void Recording::recordQueueTime(Buffer *buf, int tid, QueueTimeEvent *event) {
   flushIfNeeded(buf);
 }
 
+void Recording::recordTaskBlock(Buffer *buf, int tid, TaskBlockEvent *event) {
+  int start = buf->skip(1);
+  buf->putVar64(T_TASK_BLOCK);
+  buf->putVar64(event->_start);
+  buf->putVar64(event->_end - event->_start);
+  buf->putVar64(tid);
+  writeContextSnapshot(buf, event->_ctx);
+  buf->putVar64(event->_blocker);
+  buf->putVar64(event->_unblockingSpanId);
+  writeEventSizePrefix(buf, start);
+  flushIfNeeded(buf);
+}
+
 void Recording::recordAllocation(RecordingBuffer *buf, int tid,
                                  u64 call_trace_id, AllocEvent *event) {
   int start = buf->skip(1);
@@ -1789,6 +1808,18 @@ void FlightRecorder::recordQueueTime(int lock_index, int tid,
   }
 }
 
+void FlightRecorder::recordTaskBlock(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->recordTaskBlock(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,7 @@ 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 recordTaskBlock(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 +348,7 @@ 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 recordTaskBlock(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) {
         return VMThread::cast((void*)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"
@@ -150,15 +151,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 +320,48 @@ 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 spanId,
+    jlong rootSpanId, jlong blocker, jlong unblockingSpanId) {
+  int tid = ProfiledThread::currentTid();
+  if (tid < 0) {
+    return;
+  }
+  Context context = ContextApi::snapshot();
+  context.spanId = (u64)spanId;
+  context.rootSpanId = (u64)rootSpanId;
+  recordTaskBlockIfEligible(tid, (u64)startTicks, (u64)endTicks, context,
+                            (u64)blocker, (u64)unblockingSpanId);
+}
+
+extern "C" DLLEXPORT void JNICALL
+Java_com_datadoghq_profiler_JavaProfiler_parkEnter0(
+    JNIEnv *env, jclass unused, jlong spanId, jlong rootSpanId) {
+  ProfiledThread *current = ProfiledThread::current();
+  if (current == nullptr) {
+    return;
+  }
+  current->parkEnter(spanId, rootSpanId, 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();
+  recordTaskBlockIfEligible(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) {

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

@@ -153,7 +153,20 @@ void JfrMetadata::initialize(
               << field("numExitedThreads", T_INT,
                        "Number of Exited Threads Before Handling Signal")
               << field("numPermissionDenied", T_INT,
-                       "Number of Permission Denied Errors"))
+                       "Number of Permission Denied Errors")
+              << field("numSkippedPrecheckOs", T_INT,
+                       "Signals skipped after OS-thread-state precheck (SLEEPING or "
+                       "CONDVAR_WAIT: sleep, park/parkNanos, JDK 21+ Thread.sleep via wait)")
+              << field("numSkippedParkedSpanless", T_LONG,
+                       "Signals suppressed for Java-level parked threads without an active span")
+              << field("numSkippedParkedActiveSpan", T_LONG,
+                       "Signals suppressed for Java-level parked threads with an active span")
+              << field("numTaskBlockEmitted", T_LONG,
+                       "TaskBlock events emitted for active-span blocking intervals")
+              << field("numTaskBlockSkippedSpanZero", T_LONG,
+                       "TaskBlock intervals skipped because no active span was captured")
+              << field("numTaskBlockSkippedTooShort", T_LONG,
+                       "TaskBlock intervals skipped because they were shorter than the recording threshold"))
 
           << (type("datadog.ObjectSample", T_ALLOC, "Allocation sample")
                   << category("Datadog", "Profiling")
@@ -205,6 +218,17 @@ void JfrMetadata::initialize(
                   << field("localRootSpanId", T_LONG, "Local Root Span ID") ||
               contextAttributes)
 
+          << (type("datadog.TaskBlock", T_TASK_BLOCK, "Task Block")
+                  << category("Datadog")
+                  << field("startTime", T_LONG, "Start Time", F_TIME_TICKS)
+                  << field("duration", T_LONG, "Duration", F_DURATION_TICKS)
+                  << field("eventThread", T_THREAD, "Event Thread", F_CPOOL)
+                  << field("spanId", T_LONG, "Span ID")
+                  << field("localRootSpanId", T_LONG, "Local Root Span ID")
+                  << field("blocker", T_LONG, "Blocker Identity Hash")
+                  << field("unblockingSpanId", T_LONG, "Unblocking Span ID") ||
+              contextAttributes)
+
           << (type("datadog.HeapUsage", T_HEAP_USAGE, "JVM Heap Usage")
               << category("Datadog")
               << field("startTime", T_LONG, "Start Time", F_TIME_TICKS)

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

@@ -80,6 +80,7 @@ enum JfrType {
   T_DATADOG_COUNTER = 125,
   T_UNWIND_FAILURE = 126,
   T_MALLOC = 127,
+  T_TASK_BLOCK = 128,
   T_ANNOTATION = 200,
   T_LABEL = 201,
   T_CATEGORY = 202,

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

@@ -217,6 +217,10 @@ Error LivenessTracker::initialize(Arguments &args) {
   }
 
   if (_initialized) {
+    // Tracker settings are sticky across recordings. Preserve the historical
+    // table/config behavior, but allow HeapUsage recording to be enabled later
+    // (e.g. if an earlier test initialized liveness without ':L').
+    _record_heap_usage = _record_heap_usage || args._record_heap_usage;
     // if the tracker was previously initialized return the stored result for
     // consistency this hack also means that if the profiler is started with
     // different arguments for liveness tracking those will be ignored it is