Implement option to use JVMTI/JFR stack-walker for CPU- and wall-clock-profiler

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

@@ -364,6 +364,24 @@ Error Arguments::parse(const char *args) {
         _remote_symbolication = true;
       }
 
+      CASE("jvmtistacks")
+      if (value != NULL) {
+        switch (value[0]) {
+        case 'n': // no
+        case 'f': // false
+        case '0': // 0
+          _jvmtistacks = false;
+          break;
+        case 'y': // yes
+        case 't': // true
+        case '1': // 1
+        default:
+          _jvmtistacks = true;
+        }
+      } else {
+        _jvmtistacks = true;
+      }
+
       CASE("wallsampler")
       if (value != NULL) {
           switch (value[0]) {

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

@@ -190,6 +190,7 @@ class Arguments {
   bool _lightweight;
   bool _enable_method_cleanup;
   bool _remote_symbolication;  // Enable remote symbolication for native frames
+  bool _jvmtistacks;           // Delegate CPU/wall stack walks to HotSpot JFR RequestStackTrace extension
 
   Arguments(bool persistent = false)
       : _buf(NULL),
@@ -224,7 +225,8 @@ class Arguments {
         _context_attributes({}),
         _lightweight(false),
         _enable_method_cleanup(true),
-        _remote_symbolication(false) {}
+        _remote_symbolication(false),
+        _jvmtistacks(false) {}
 
   ~Arguments();
 

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

@@ -98,7 +98,12 @@
   X(WALKVM_CONT_ENTRY_NULL,     "walkvm_cont_entry_null")                      \
   X(NATIVE_LIBS_DROPPED, "native_libs_dropped")                                \
   X(SIGACTION_PATCHED_LIBS, "sigaction_patched_libs")                          \
-  X(SIGACTION_INTERCEPTED, "sigaction_intercepted")
+  X(SIGACTION_INTERCEPTED, "sigaction_intercepted")                            \
+  X(JVMTI_STACKS_INIT_OK, "jvmti_stacks_init_ok")                              \
+  X(JVMTI_STACKS_INIT_FAILED, "jvmti_stacks_init_failed")                      \
+  X(JVMTI_STACKS_REQUESTED, "jvmti_stacks_requested")                          \
+  X(JVMTI_STACKS_FAILED_WRONG_PHASE, "jvmti_stacks_failed_wrong_phase")        \
+  X(JVMTI_STACKS_FAILED_OTHER, "jvmti_stacks_failed_other")
 #define X_ENUM(a, b) a,
 typedef enum CounterId : int {
   DD_COUNTER_TABLE(X_ENUM) DD_NUM_COUNTERS

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

@@ -25,7 +25,7 @@
 #include <signal.h>
 
 class CTimer : public Engine {
-private:
+protected:
   // This is accessed from signal handlers, so must be async-signal-safe
   static bool _enabled;
   static long _interval;
@@ -38,6 +38,7 @@ class CTimer : public Engine {
   int registerThread(int tid);
   void unregisterThread(int tid);
 
+private:
   // cppcheck-suppress unusedPrivateFunction
   static void signalHandler(int signo, siginfo_t *siginfo, void *ucontext);
 
@@ -60,6 +61,24 @@ class CTimer : public Engine {
   static int getSignal() { return _signal; }
 };
 
+// A CPU-time engine that reuses CTimer's per-thread timer_create / SIGPROF
+// dispatch, but instead of walking the stack in the signal handler delegates
+// the walk to HotSpot's JFR RequestStackTrace JVMTI extension. The sampled
+// event is emitted on our side with only a correlation ID; the JVM writes
+// the stack trace (and its own JFR stack-trace id) into the concurrent JFR
+// recording as jdk.AsyncStackTrace. See VM::canRequestStackTrace().
+class CTimerJvmti : public CTimer {
+private:
+  // cppcheck-suppress unusedPrivateFunction
+  static void signalHandler(int signo, siginfo_t *siginfo, void *ucontext);
+
+public:
+  const char *name() { return "CTimerJvmti"; }
+
+  Error check(Arguments &args);
+  Error start(Arguments &args);
+};
+
 #else
 
 class CTimer : public Engine {
@@ -75,6 +94,19 @@ class CTimer : public Engine {
   static bool supported() { return false; }
 };
 
+class CTimerJvmti : public Engine {
+public:
+  const char *name() { return "CTimerJvmti"; }
+
+  Error check(Arguments &args) {
+    return Error("CTimerJvmti is not supported on this platform");
+  }
+
+  Error start(Arguments &args) {
+    return Error("CTimerJvmti is not supported on this platform");
+  }
+};
+
 #endif // __linux__
 
 #endif // _CTIMER_H

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

@@ -142,6 +142,87 @@ void CTimer::stop() {
   }
 }
 
+Error CTimerJvmti::check(Arguments &args) {
+  if (!VM::canRequestStackTrace()) {
+    return Error("HotSpot RequestStackTrace JVMTI extension not available");
+  }
+  return CTimer::check(args);
+}
+
+Error CTimerJvmti::start(Arguments &args) {
+  if (!VM::canRequestStackTrace()) {
+    return Error("HotSpot RequestStackTrace JVMTI extension not available");
+  }
+  if (args._interval < 0) {
+    return Error("interval must be positive");
+  }
+
+  _interval = args.cpuSamplerInterval();
+  _cstack = args._cstack;
+  _signal = SIGPROF;
+
+  int max_timers = OS::getMaxThreadId();
+  if (max_timers != _max_timers) {
+    free(_timers);
+    _timers = (int *)calloc(max_timers, sizeof(int));
+    _max_timers = max_timers;
+  }
+
+  OS::installSignalHandler(_signal, CTimerJvmti::signalHandler);
+
+  Error result = Error::OK;
+  ThreadList *thread_list = OS::listThreads();
+  while (thread_list->hasNext()) {
+    int tid = thread_list->next();
+    int err = registerThread(tid);
+    if (err != 0) {
+      result = Error("Failed to register thread");
+    }
+  }
+  delete thread_list;
+
+  return Error::OK;
+}
+
+void CTimerJvmti::signalHandler(int signo, siginfo_t *siginfo, void *ucontext) {
+  CriticalSection cs;
+  if (!cs.entered()) {
+    return;
+  }
+  int saved_errno = errno;
+  if (!__atomic_load_n(&_enabled, __ATOMIC_ACQUIRE)) {
+    errno = saved_errno;
+    return;
+  }
+  int tid = 0;
+  ProfiledThread *current = ProfiledThread::currentSignalSafe();
+  assert(current == nullptr || !current->isDeepCrashHandler());
+  if (current != nullptr && JVMThread::isInitialized() && JVMThread::current() == nullptr
+      && current->inInitWindow()) {
+    current->tickInitWindow();
+    errno = saved_errno;
+    return;
+  }
+  if (current != NULL) {
+    current->noteCPUSample(Profiler::instance()->recordingEpoch());
+    tid = current->tid();
+  } else {
+    tid = OS::threadId();
+  }
+  Shims::instance().setSighandlerTid(tid);
+
+  ExecutionEvent event;
+  event._execution_mode = getThreadExecutionMode();
+  // Opted into JVMTI delegation; drop the sample if the JVM rejects the
+  // request (WRONG_PHASE if JFR is not recording, NOT_AVAILABLE if
+  // jdk.AsyncStackTrace is disabled). recordSampleDelegated() bumps the
+  // failure counters; there is no fallback to ASGCT in this engine.
+  (void)Profiler::instance()->recordSampleDelegated(ucontext, _interval, tid,
+                                                    BCI_CPU, &event);
+  Shims::instance().setSighandlerTid(-1);
+  errno = saved_errno;
+}
+
 void CTimer::signalHandler(int signo, siginfo_t *siginfo, void *ucontext) {
   // Atomically try to enter critical section - prevents all reentrancy races
   CriticalSection cs;

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

@@ -1486,6 +1486,7 @@ void Recording::writeEventSizePrefix(Buffer *buf, int start) {
 }
 
 void Recording::recordExecutionSample(Buffer *buf, int tid, u64 call_trace_id,
+                                      u64 correlation_id,
                                       ExecutionEvent *event) {
   int start = buf->skip(1);
   buf->putVar64(T_EXECUTION_SAMPLE);
@@ -1495,12 +1496,14 @@ void Recording::recordExecutionSample(Buffer *buf, int tid, u64 call_trace_id,
   buf->put8(static_cast<int>(event->_thread_state));
   buf->put8(static_cast<int>(event->_execution_mode));
   buf->putVar64(event->_weight);
+  buf->putVar64(correlation_id);
   writeCurrentContext(buf);
   writeEventSizePrefix(buf, start);
   flushIfNeeded(buf);
 }
 
 void Recording::recordMethodSample(Buffer *buf, int tid, u64 call_trace_id,
+                                   u64 correlation_id,
                                    ExecutionEvent *event) {
   int start = buf->skip(1);
   buf->putVar64(T_METHOD_SAMPLE);
@@ -1510,6 +1513,7 @@ void Recording::recordMethodSample(Buffer *buf, int tid, u64 call_trace_id,
   buf->put8(static_cast<int>(event->_thread_state));
   buf->put8(static_cast<int>(event->_execution_mode));
   buf->putVar64(event->_weight);
+  buf->putVar64(correlation_id);
   writeCurrentContext(buf);
   writeEventSizePrefix(buf, start);
   flushIfNeeded(buf);
@@ -1797,11 +1801,11 @@ void FlightRecorder::recordEvent(int lock_index, int tid, u64 call_trace_id,
       RecordingBuffer *buf = rec->buffer(lock_index);
       switch (event_type) {
       case BCI_CPU:
-          rec->recordExecutionSample(buf, tid, call_trace_id,
+          rec->recordExecutionSample(buf, tid, call_trace_id, 0,
                                      (ExecutionEvent *)event);
           break;
         case BCI_WALL:
-          rec->recordMethodSample(buf, tid, call_trace_id,
+          rec->recordMethodSample(buf, tid, call_trace_id, 0,
                                   (ExecutionEvent *)event);
           break;
         case BCI_ALLOC:
@@ -1824,6 +1828,33 @@ void FlightRecorder::recordEvent(int lock_index, int tid, u64 call_trace_id,
   }
 }
 
+void FlightRecorder::recordEventDelegated(int lock_index, int tid,
+                                          u64 correlation_id, int event_type,
+                                          Event *event) {
+  OptionalSharedLockGuard locker(&_rec_lock);
+  if (locker.ownsLock()) {
+    Recording* rec = _rec;
+    if (rec != nullptr) {
+      RecordingBuffer *buf = rec->buffer(lock_index);
+      switch (event_type) {
+        case BCI_CPU:
+          rec->recordExecutionSample(buf, tid, 0, correlation_id,
+                                     (ExecutionEvent *)event);
+          break;
+        case BCI_WALL:
+          rec->recordMethodSample(buf, tid, 0, correlation_id,
+                                  (ExecutionEvent *)event);
+          break;
+        default:
+          // Delegation is only wired for CPU/wall samples in v1.
+          return;
+      }
+      rec->flushIfNeeded(buf);
+      rec->addThread(lock_index, tid);
+    }
+  }
+}
+
 void FlightRecorder::recordLog(LogLevel level, const char *message,
                                size_t len) {
   OptionalSharedLockGuard locker(&_rec_lock);

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

@@ -273,9 +273,9 @@ class Recording {
   void writeCurrentContext(Buffer *buf);
 
   void recordExecutionSample(Buffer *buf, int tid, u64 call_trace_id,
-                             ExecutionEvent *event);
+                             u64 correlation_id, ExecutionEvent *event);
   void recordMethodSample(Buffer *buf, int tid, u64 call_trace_id,
-                          ExecutionEvent *event);
+                          u64 correlation_id, ExecutionEvent *event);
   void recordWallClockEpoch(Buffer *buf, WallClockEpochEvent *event);
   void recordTraceRoot(Buffer *buf, int tid, TraceRootEvent *event);
   void recordQueueTime(Buffer *buf, int tid, QueueTimeEvent *event);
@@ -351,6 +351,13 @@ class FlightRecorder {
   void recordEvent(int lock_index, int tid, u64 call_trace_id, int event_type,
                    Event *event);
 
+  // Emit a BCI_CPU / BCI_WALL sample with no stack-trace attached to our
+  // recording. `correlation_id` is the same jlong passed to the HotSpot
+  // RequestStackTrace extension so downstream tooling can join our event with
+  // the JVM-emitted jdk.AsyncStackTrace.
+  void recordEventDelegated(int lock_index, int tid, u64 correlation_id,
+                            int event_type, Event *event);
+
   void recordLog(LogLevel level, const char *message, size_t len);
 
   void recordDatadogSetting(int lock_index, int length, const char *name,

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

@@ -300,7 +300,7 @@ typedef void* address;
         field(_flag_type_offset, offset, MATCH_SYMBOLS("_type", "type"))                                            \
     type_end()                                                                                                      \
     type_begin(VMOop, MATCH_SYMBOLS("oopDesc"))                                                                     \
-        field(_oop_klass_offset, offset, MATCH_SYMBOLS("_metadata._klass"))                                         \
+        field(_oop_klass_offset, offset, MATCH_SYMBOLS("_metadata._klass", "_compressed_klass"))                    \
     type_end()                                                                                                      \
     type_begin(VMUniverse, MATCH_SYMBOLS("Universe", "CompressedKlassPointers"))                                    \
         field(_narrow_klass_base_addr, address, MATCH_SYMBOLS("_narrow_klass._base", "_base"))                      \