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,11 @@
   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_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 +107,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,57 +109,118 @@ class WallClockEpochEvent {
   u32 _num_failed_samples;
   u32 _num_exited_threads;
   u32 _num_permission_denied;
+  u64 _num_suppressed_sampled_run;
+  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_task_block_emitted(0), _num_task_block_skipped_span_zero(0),
+        _num_task_block_skipped_too_short(0) {}
 
   bool hasChanged() { return _dirty; }
 
+  // Snapshot semantics: _num_samplable_threads tracks the current count of
+  // profilable threads at the time of the iteration. Dirties only on change,
+  // so steady-state thread counts collapse adjacent iterations into a single
+  // emitted epoch.
   void updateNumSamplableThreads(u32 num_samplable_threads) {
     if (_num_samplable_threads != num_samplable_threads) {
       _dirty = true;
       _num_samplable_threads = num_samplable_threads;
     }
   }
 
-  void updateNumSuccessfulSamples(u32 num_successful_samples) {
-    if (_num_successful_samples != num_successful_samples) {
+  // Accumulator semantics: the addNum*() methods accumulate per-iteration
+  // event counts into a since-last-emit total. This guarantees that when
+  // multiple iterations collapse into a single emitted epoch (because
+  // hasChanged() stays false across them — e.g. steady-state workloads where
+  // adjacent drained counter values happen to match), the per-iteration
+  // contributions are summed into the next emit rather than discarded.
+  // newEpoch() resets these fields to 0 at emit boundaries so each emitted
+  // epoch reports drained-counter activity for its own _duration_millis
+  // interval. JFR consumers can sum the field across all epochs to recover
+  // the exact cumulative count over the recording.
+  void addNumSuccessfulSamples(u32 n) {
+    if (n > 0) {
       _dirty = true;
-      _num_successful_samples = num_successful_samples;
+      _num_successful_samples += n;
     }
   }
 
-  void updateNumFailedSamples(u32 num_failed_samples) {
-    if (_num_failed_samples != num_failed_samples) {
+  void addNumFailedSamples(u32 n) {
+    if (n > 0) {
       _dirty = true;
-      _num_failed_samples = num_failed_samples;
+      _num_failed_samples += n;
     }
   }
 
-  void updateNumExitedThreads(u32 num_exited_threads) {
-    if (_num_exited_threads != num_exited_threads) {
+  void addNumExitedThreads(u32 n) {
+    if (n > 0) {
       _dirty = true;
-      _num_exited_threads = num_exited_threads;
+      _num_exited_threads += n;
     }
   }
 
-  void updateNumPermissionDenied(u32 num_permission_denied) {
-    if (_num_permission_denied != num_permission_denied) {
+  void addNumPermissionDenied(u32 n) {
+    if (n > 0) {
       _dirty = true;
-      _num_permission_denied = num_permission_denied;
+      _num_permission_denied += n;
+    }
+  }
+
+  void addNumSuppressedSampledRun(u64 n) {
+    if (n > 0) {
+      _dirty = true;
+      _num_suppressed_sampled_run += n;
+    }
+  }
+
+  void addNumTaskBlockEmitted(u64 n) {
+    if (n > 0) {
+      _dirty = true;
+      _num_task_block_emitted += n;
+    }
+  }
+
+  void addNumTaskBlockSkippedSpanZero(u64 n) {
+    if (n > 0) {
+      _dirty = true;
+      _num_task_block_skipped_span_zero += n;
+    }
+  }
+
+  void addNumTaskBlockSkippedTooShort(u64 n) {
+    if (n > 0) {
+      _dirty = true;
+      _num_task_block_skipped_too_short += n;
     }
   }
 
   void endEpoch(u64 millis) { _duration_millis = millis; }
 
   void clean() { _dirty = false; }
 
+  // Reset the accumulator fields at emit boundaries so each emitted epoch
+  // reports only the activity that occurred since the previous emit.
+  // _num_samplable_threads is a snapshot (current count), not an accumulator,
+  // and is intentionally preserved across newEpoch() so steady-state thread
+  // counts continue to collapse via the != check in updateNumSamplableThreads.
   void newEpoch(u64 start_time) {
     _dirty = false;
     _start_time = start_time;
+    _num_successful_samples = 0;
+    _num_failed_samples = 0;
+    _num_exited_threads = 0;
+    _num_permission_denied = 0;
+    _num_suppressed_sampled_run = 0;
+    _num_task_block_emitted = 0;
+    _num_task_block_skipped_span_zero = 0;
+    _num_task_block_skipped_too_short = 0;
   }
 };
 
@@ -184,4 +245,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,10 @@ 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_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 +1574,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 +1809,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 +1822,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);