[WIP] Add events for critical path analysis

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

@@ -109,12 +109,13 @@ class WallClockEpochEvent {
   u32 _num_failed_samples;
   u32 _num_exited_threads;
   u32 _num_permission_denied;
+  u32 _num_skipped_sleeping;
 
   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_sleeping(0) {}
 
   bool hasChanged() { return _dirty; }
 
@@ -153,6 +154,10 @@ class WallClockEpochEvent {
     }
   }
 
+  void updateNumSkippedSleeping(u32 n) {
+    if (_num_skipped_sleeping != n) { _dirty = true; _num_skipped_sleeping = n; }
+  }
+
   void endEpoch(u64 millis) { _duration_millis = millis; }
 
   void clean() { _dirty = false; }
@@ -166,22 +171,37 @@ class WallClockEpochEvent {
 class TraceRootEvent {
 public:
   u64 _local_root_span_id;
+  u64 _parent_span_id;
+  u64 _start_ticks;
   u32 _label;
   u32 _operation;
 
-  TraceRootEvent(u64 local_root_span_id, u32 label, u32 operation)
-      : _local_root_span_id(local_root_span_id), _label(label),
-        _operation(operation){};
+  TraceRootEvent(u64 local_root_span_id, u64 parent_span_id, u64 start_ticks,
+                 u32 label, u32 operation)
+      : _local_root_span_id(local_root_span_id),
+        _parent_span_id(parent_span_id), _start_ticks(start_ticks),
+        _label(label), _operation(operation){};
 };
 
+typedef struct TaskBlockEvent {
+  u64 _start_ticks;
+  u64 _end_ticks;
+  u64 _span_id;
+  u64 _root_span_id;
+  uintptr_t _blocker;
+  u64 _unblocking_span_id;
+} TaskBlockEvent;
+
 typedef struct QueueTimeEvent {
   u64 _start;
   u64 _end;
   u32 _task;
   u32 _scheduler;
   u32 _origin;
   u32 _queueType;
-  u32 _queueLength;
+    u32 _queueLength;
+    u64 _submitting_span_id;
+    u64 _consuming_span_id;  // 0: use current Context; else: override JFR spanId
 } QueueTimeEvent;
 
 #endif // _EVENT_H

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

@@ -1489,6 +1489,24 @@ void Recording::writeCurrentContext(Buffer *buf) {
   }
 }
 
+void Recording::writeQueueTimeContext(Buffer *buf, QueueTimeEvent *event) {
+  u64 spanId = 0;
+  u64 rootSpanId = 0;
+  bool hasContext = ContextApi::get(spanId, rootSpanId);
+  if (event->_consuming_span_id != 0) {
+    spanId = event->_consuming_span_id;
+  }
+  buf->putVar64(spanId);
+  buf->putVar64(rootSpanId);
+  buf->putVar64(event->_submitting_span_id);
+
+  size_t numAttrs = Profiler::instance()->numContextAttributes();
+  ProfiledThread* thrd = hasContext ? ProfiledThread::currentSignalSafe() : nullptr;
+  for (size_t i = 0; i < numAttrs; i++) {
+    buf->putVar32(thrd != nullptr ? thrd->getOtelTagEncoding(i) : 0);
+  }
+}
+
 void Recording::writeEventSizePrefix(Buffer *buf, int start) {
   int size = buf->offset() - start;
   assert(size < MAX_JFR_EVENT_SIZE);
@@ -1543,13 +1561,63 @@ void Recording::recordTraceRoot(Buffer *buf, int tid, TraceRootEvent *event) {
   flushIfNeeded(buf);
   int start = buf->skip(1);
   buf->putVar64(T_ENDPOINT);
-  buf->putVar64(TSC::ticks());
-  buf->put8(0);
+  buf->putVar64(event->_start_ticks);
+  buf->putVar64(TSC::ticks() - event->_start_ticks);
   buf->putVar32(tid);
   buf->put8(0);
   buf->putVar32(event->_label);
   buf->putVar32(event->_operation);
   buf->putVar64(event->_local_root_span_id);
+  buf->putVar64(event->_parent_span_id);
+  writeEventSizePrefix(buf, start);
+  flushIfNeeded(buf);
+}
+
+void Recording::recordTaskBlock(Buffer *buf, int tid, TaskBlockEvent *event) {
+  flushIfNeeded(buf);
+  int start = buf->skip(1);
+  buf->putVar64(T_TASK_BLOCK);
+  buf->putVar64(event->_start_ticks);
+  buf->putVar64(event->_end_ticks - event->_start_ticks);
+  buf->putVar32(tid);
+  buf->put8(0);
+  buf->putVar64(event->_span_id);
+  buf->putVar64(event->_root_span_id);
+  buf->putVar64((u64)event->_blocker);
+  buf->putVar64(event->_unblocking_span_id);
+  writeEventSizePrefix(buf, start);
+  flushIfNeeded(buf);
+}
+
+void Recording::recordSpanNode(Buffer *buf, int tid, u64 spanId, u64 parentSpanId, u64 rootSpanId,
+                                u64 startNanos, u64 durationNanos, u32 encodedOperation, u32 encodedResource) {
+  // Convert epoch nanoseconds to JFR ticks so that standard JFR tooling (JMC, Mission
+  // Control) can correlate SpanNode events with other events on the recording timeline.
+  // _start_time is in microseconds; multiply by 1000 to get the recording epoch in nanos.
+  u64 start_epoch_nanos = _start_time * 1000ULL;
+  // Use signed arithmetic: a span that started in a previous JFR chunk has
+  // startNanos < start_epoch_nanos. Unsigned subtraction would wrap around and
+  // produce a huge positive u64, making (u64)(negative_double) undefined behaviour.
+  // With signed delta the result is a negative tick offset, placing the event just
+  // before the chunk boundary, which is the correct behaviour for pre-chunk spans.
+  long long delta_nanos = (long long)startNanos - (long long)start_epoch_nanos;
+  long long delta_ticks = (long long)((double)delta_nanos * TSC::frequency() / NANOTIME_FREQ);
+  u64 startTicks = (u64)((long long)_start_ticks + delta_ticks);
+  u64 durationTicks = (u64)((double)durationNanos * TSC::frequency() / NANOTIME_FREQ);
+
+  flushIfNeeded(buf);
+  int start = buf->skip(1);
+  buf->putVar64(T_SPAN_NODE);
+  buf->putVar64(startTicks);       // startTime (F_TIME_TICKS)
+  buf->putVar64(durationTicks);    // duration (F_DURATION_TICKS)
+  buf->putVar32(tid);              // eventThread (F_CPOOL)
+  buf->putVar64(spanId);
+  buf->putVar64(parentSpanId);
+  buf->putVar64(rootSpanId);
+  buf->putVar64(startNanos);       // startNanos — epoch ns, used by backend extractor
+  buf->putVar64(durationNanos);    // durationNanos — ns
+  buf->putVar32(encodedOperation);
+  buf->putVar32(encodedResource);
   writeEventSizePrefix(buf, start);
   flushIfNeeded(buf);
 }
@@ -1565,7 +1633,7 @@ void Recording::recordQueueTime(Buffer *buf, int tid, QueueTimeEvent *event) {
   buf->putVar64(event->_scheduler);
   buf->putVar64(event->_queueType);
   buf->putVar64(event->_queueLength);
-  writeCurrentContext(buf);
+  writeQueueTimeContext(buf, event);
   writeEventSizePrefix(buf, start);
   flushIfNeeded(buf);
 }
@@ -1773,6 +1841,7 @@ void FlightRecorder::recordTraceRoot(int lock_index, int tid,
     if (rec != nullptr) {
       Buffer *buf = rec->buffer(lock_index);
       rec->recordTraceRoot(buf, tid, event);
+      rec->addThread(lock_index, tid);
     }
   }
 }
@@ -1785,6 +1854,37 @@ void FlightRecorder::recordQueueTime(int lock_index, int tid,
     if (rec != nullptr) {
       Buffer *buf = rec->buffer(lock_index);
       rec->recordQueueTime(buf, tid, event);
+      rec->addThread(lock_index, 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);
+      rec->addThread(lock_index, tid);
+    }
+  }
+}
+
+void FlightRecorder::recordSpanNode(int lock_index, int tid, u64 spanId, u64 parentSpanId, u64 rootSpanId,
+                                    u64 startNanos, u64 durationNanos, u32 encodedOperation, u32 encodedResource) {
+  OptionalSharedLockGuard locker(&_rec_lock);
+  if (locker.ownsLock()) {
+    Recording* rec = _rec;
+    if (rec != nullptr) {
+      Buffer *buf = rec->buffer(lock_index);
+      rec->recordSpanNode(buf, tid, spanId, parentSpanId, rootSpanId, startNanos, durationNanos, encodedOperation, encodedResource);
+      // Register the emitting thread in the JFR thread CPOOL so that JMC can resolve
+      // the eventThread reference. Without this, threads that emit SpanNode events but
+      // have no CPU/wall profiling samples in the current chunk are absent from the
+      // CPOOL, causing IMCThread to be null in the backend and threadId=0 ("unknown span").
+      rec->addThread(lock_index, tid);
     }
   }
 }

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

@@ -271,6 +271,7 @@ class Recording {
 
   void writeContextSnapshot(Buffer *buf, Context &context);
   void writeCurrentContext(Buffer *buf);
+  void writeQueueTimeContext(Buffer *buf, QueueTimeEvent *event);
 
   void recordExecutionSample(Buffer *buf, int tid, u64 call_trace_id,
                              ExecutionEvent *event);
@@ -279,6 +280,9 @@ 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 recordSpanNode(Buffer *buf, int tid, u64 spanId, u64 parentSpanId, u64 rootSpanId,
+                      u64 startNanos, u64 durationNanos, u32 encodedOperation, u32 encodedResource);
   void recordAllocation(RecordingBuffer *buf, int tid, u64 call_trace_id,
                         AllocEvent *event);
   void recordMallocSample(Buffer *buf, int tid, u64 call_trace_id,
@@ -347,6 +351,9 @@ 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);
+  void recordSpanNode(int lock_index, int tid, u64 spanId, u64 parentSpanId, u64 rootSpanId,
+                      u64 startNanos, u64 durationNanos, u32 encodedOperation, u32 encodedResource);
 
   bool active() const { return _rec != NULL; }
 

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

@@ -10,6 +10,8 @@
 #include <stdarg.h>
 #include "hotspot/vmStructs.inline.h"
 #include "vmEntry.h"
+#include "context.h"
+#include "thread.h"
 #include "jniHelper.h"
 #include "jvmHeap.h"
 #include "jvmThread.h"
@@ -760,6 +762,40 @@ int VMThread::osThreadId() {
     return -1;
 }
 
+u64 VMThread::monitorOwnerSpanId(const void* object) {
+    if (_monitor_owner_offset == 0) return 0;
+
+    // Read the mark word from the object header.
+    uintptr_t mark = (uintptr_t)SafeAccess::load((void**)object, nullptr);
+
+    // At MonitorContendedEnter time the monitor must be inflated (MONITOR_BIT set
+    // in bits [1:0]). Guard defensively for any narrow races.
+    if ((mark & 0x3) != MONITOR_BIT) return 0;
+
+    // Extract ObjectMonitor* (pointer stored in bits 63:2, 4-byte aligned).
+    const char* monitor = (const char*)(mark & ~(uintptr_t)0x3);
+
+    // Read ObjectMonitor::_owner (JavaThread*).
+    const char* owner_thread = (const char*)SafeAccess::load(
+        (void**)(monitor + _monitor_owner_offset), nullptr);
+    if (owner_thread == nullptr) return 0;
+
+    // JavaThread* -> OS thread ID via existing VMThread infrastructure.
+    int owner_tid = VMThread::cast(owner_thread)->osThreadId();
+    if (owner_tid <= 0) return 0;
+
+    // OS thread ID -> ProfiledThread* -> span ID.
+    ProfiledThread* owner = ProfiledThread::findByTid(owner_tid);
+    if (owner == nullptr || !owner->isContextInitialized()) return 0;
+
+    OtelThreadContextRecord* record = owner->getOtelContextRecord();
+    if (record == nullptr || !record->valid) return 0;
+
+    u64 span_id = 0;
+    for (int i = 0; i < 8; i++) { span_id = (span_id << 8) | record->span_id[i]; }
+    return span_id;
+}
+
 JNIEnv* VMThread::jni() {
     if (_env_offset < 0) {
         return VM::jni();  // fallback for non-HotSpot JVM

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

@@ -232,6 +232,9 @@ typedef void* address;
         field(_osthread_id_offset, offset, MATCH_SYMBOLS("_thread_id"))                                             \
         field_with_version(_osthread_state_offset, offset, 10, MAX_VERSION, MATCH_SYMBOLS("_state"))                \
     type_end()                                                                                                      \
+    type_begin(VMObjectMonitor, MATCH_SYMBOLS("ObjectMonitor"))                                                     \
+        field(_monitor_owner_offset, offset, MATCH_SYMBOLS("_owner"))                                               \
+    type_end()                                                                                                      \
     type_begin(VMThreadShadow, MATCH_SYMBOLS("ThreadShadow"))                                                         \
         field(_thread_exception_offset, offset, MATCH_SYMBOLS("_exception_file"))                                   \
     type_end()                                                                                                      \
@@ -780,6 +783,12 @@ DECLARE(VMThread)
     static OSThreadState getOSThreadState();
 
     int osThreadId();
+
+    // Returns the span ID of the JavaThread currently owning the given monitor
+    // object, reading ObjectMonitor::_owner directly without a safepoint.
+    // Returns 0 if the owner cannot be determined.
+    static u64 monitorOwnerSpanId(const void* object);
+
     JNIEnv* jni();
 
     OSThreadState osThreadState();