feat(profiler): wall-clock signal suppression for idle threads

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

@@ -386,6 +386,14 @@ Error Arguments::parse(const char *args) {
         _jvmtistacks = true;
       }
 
+      CASE("wallprecheck")
+      if (value != NULL) {
+        _wall_precheck = strcmp(value, "false") != 0 && strcmp(value, "0") != 0;
+      } else {
+        // No value means enable
+        _wall_precheck = true;
+      }
+
       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;
@@ -205,6 +206,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

@@ -65,6 +65,8 @@
   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(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,13 @@ class WallClockEpochEvent {
   u32 _num_failed_samples;
   u32 _num_exited_threads;
   u32 _num_permission_denied;
+  u64 _num_suppressed_sampled_run;
 
   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) {}
 
   bool hasChanged() { return _dirty; }
 
@@ -153,13 +154,21 @@ class WallClockEpochEvent {
     }
   }
 
+  void addNumSuppressedSampledRun(u64 n) {
+    if (n > 0) {
+      _dirty = true;
+      _num_suppressed_sampled_run += n;
+    }
+  }
+
   void endEpoch(u64 millis) { _duration_millis = millis; }
 
   void clean() { _dirty = false; }
 
   void newEpoch(u64 start_time) {
     _dirty = false;
     _start_time = start_time;
+    _num_suppressed_sampled_run = 0;
   }
 };
 

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

@@ -1762,6 +1762,7 @@ 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);
   writeEventSizePrefix(buf, start);
   flushIfNeeded(buf);
 }

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,7 +24,7 @@
 #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"
@@ -155,10 +155,13 @@ JavaCritical_com_datadoghq_profiler_JavaProfiler_filterThreadAdd0() {
     slot_id = thread_filter->registerThread();
     current->setFilterSlotId(slot_id);
   }
-  
+
   if (unlikely(slot_id == -1)) {
     return;  // Failed to register thread
   }
+  // Reset suppression state so a new thread occupying this slot does not inherit
+  // stale state from its predecessor. Must happen before add().
+  thread_filter->resetSlotRunState(slot_id);
   thread_filter->add(tid, slot_id);
 }
 
@@ -314,6 +317,87 @@ Java_com_datadoghq_profiler_JavaProfiler_recordQueueEnd0(
   Profiler::instance()->recordQueueTime(tid, &event);
 }
 
+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());
+  ThreadFilter *tf = Profiler::instance()->threadFilter();
+  if (tf->enabled()) {
+    tf->enterBlockedRun(current->filterSlotId(), OSThreadState::CONDVAR_WAIT);
+  }
+}
+
+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 = {};
+  current->parkExit(start_ticks, park_context);
+  ThreadFilter *tf = Profiler::instance()->threadFilter();
+  if (tf->enabled()) {
+    tf->exitBlockedRun(current->filterSlotId());
+  }
+}
+
+static bool decodeJavaBlockState(jint state, OSThreadState &decoded) {
+  if (state == static_cast<jint>(OSThreadState::SLEEPING)) {
+    decoded = OSThreadState::SLEEPING;
+    return true;
+  }
+  decoded = OSThreadState::UNKNOWN;
+  return false;
+}
+
+extern "C" DLLEXPORT jlong JNICALL
+Java_com_datadoghq_profiler_JavaProfiler_blockEnter0(
+    JNIEnv *env, jclass unused, jint state) {
+  OSThreadState decoded;
+  if (!decodeJavaBlockState(state, decoded)) {
+    return 0;
+  }
+  ProfiledThread *current = ProfiledThread::current();
+  if (current == nullptr) {
+    return 0;
+  }
+  ThreadFilter *tf = Profiler::instance()->threadFilter();
+  if (!tf->enabled()) {
+    return 0;
+  }
+  ThreadFilter::SlotID slot_id = current->filterSlotId();
+  if (slot_id < 0) {
+    return 0;
+  }
+  tf->enterBlockedRun(slot_id, decoded);
+  return static_cast<jlong>(slot_id) + 1;
+}
+
+extern "C" DLLEXPORT void JNICALL
+Java_com_datadoghq_profiler_JavaProfiler_blockExit0(
+    JNIEnv *env, jclass unused, jlong token) {
+  if (token <= 0) {
+    return;
+  }
+  ProfiledThread *current = ProfiledThread::current();
+  if (current == nullptr) {
+    return;
+  }
+  ThreadFilter::SlotID slot_id = static_cast<ThreadFilter::SlotID>(token - 1);
+  if (current->filterSlotId() != slot_id) {
+    return;
+  }
+  ThreadFilter *tf = Profiler::instance()->threadFilter();
+  if (tf->enabled()) {
+    tf->exitBlockedRun(slot_id);
+  }
+}
+
 extern "C" DLLEXPORT jlong JNICALL
 Java_com_datadoghq_profiler_JavaProfiler_currentTicks0(JNIEnv *env,
                                                        jclass unused) {

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

@@ -155,7 +155,9 @@ 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("numSuppressedSampledRun", T_LONG,
+                       "Signals suppressed by the wall-clock once-per-run filter"))
 
           << (type("datadog.ObjectSample", T_ALLOC, "Allocation sample")
                   << category("Datadog", "Profiling")

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

@@ -216,12 +216,10 @@ Error LivenessTracker::initialize(Arguments &args) {
     return Error::OK;
   }
 
-  // _record_heap_usage controls per-session JFR event emission only, not the
-  // tracking table. Update it before the _initialized guard so each profiler
-  // start gets the correct setting even when the table persists across recordings.
-  _record_heap_usage = args._record_heap_usage;
-
   if (_initialized) {
+    // Once heap usage recording has been enabled it stays on for the JVM lifetime,
+    // so a later recording without ':L' does not silently drop HeapUsage events.
+    _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
@@ -270,6 +268,8 @@ Error LivenessTracker::initialize(Arguments &args) {
                                    // enough for 1G of heap
   _table = (TrackingEntry *)malloc(sizeof(TrackingEntry) * _table_cap);
 
+  _record_heap_usage = args._record_heap_usage;
+
   _gc_epoch = 0;
   _last_gc_epoch = 0;
 

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

@@ -1,6 +1,6 @@
 /*
  * Copyright The async-profiler authors
- * Copyright 2024, 2025 Datadog, Inc
+ * Copyright 2024, 2026 Datadog, Inc
  * SPDX-License-Identifier: Apache-2.0
  */
 
@@ -80,6 +80,7 @@ void Profiler::onThreadStart(jvmtiEnv *jvmti, JNIEnv *jni, jthread thread) {
   if (_thread_filter.enabled()) {
     int slot_id = _thread_filter.registerThread();
     current->setFilterSlotId(slot_id);
+    _thread_filter.resetSlotRunState(slot_id);
     _thread_filter.remove(slot_id);  // Remove from filtering initially
   }
   if (thread != NULL) {
@@ -550,7 +551,7 @@ void Profiler::recordDeferredSample(int tid, u64 call_trace_id, jint event_type,
   _locks[lock_index].unlock();
 }
 
-void Profiler::recordSample(void *ucontext, u64 counter, int tid,
+bool Profiler::recordSample(void *ucontext, u64 counter, int tid,
                             jint event_type, u64 call_trace_id, Event *event) {
   atomicIncRelaxed(_total_samples);
 
@@ -566,7 +567,7 @@ void Profiler::recordSample(void *ucontext, u64 counter, int tid,
       // collected trace
       PerfEvents::resetBuffer(tid);
     }
-    return;
+    return false;
   }
 
   bool truncated = false;
@@ -616,6 +617,7 @@ void Profiler::recordSample(void *ucontext, u64 counter, int tid,
   _jfr.recordEvent(lock_index, tid, call_trace_id, event_type, event);
 
   _locks[lock_index].unlock();
+  return true;
 }
 
 void Profiler::recordSampleDelegated(void *ucontext, u64 weight, int tid,
@@ -1203,6 +1205,7 @@ Error Profiler::start(Arguments &args, bool reset) {
   _omit_stacktraces = args._lightweight;
   _remote_symbolication = args._remote_symbolication;
   _libs->setRemoteSymbolication(_remote_symbolication);
+  _wall_precheck = args._wall_precheck;
   _event_mask =
       ((args._event != NULL && strcmp(args._event, EVENT_NOOP) != 0) ? EM_CPU
                                                                      : 0) |
@@ -1301,10 +1304,14 @@ Error Profiler::start(Arguments &args, bool reset) {
 
   // Minor optim: Register the current thread (start thread won't be called)
   if (_thread_filter.enabled()) {
+    _thread_filter.clearActive();
     ProfiledThread *current = ProfiledThread::current();
     assert(current != nullptr);
-    int slot_id = _thread_filter.registerThread();
-    current->setFilterSlotId(slot_id);
+    int slot_id = current->filterSlotId();
+    if (slot_id < 0) {
+      slot_id = _thread_filter.registerThread();
+      current->setFilterSlotId(slot_id);
+    }
     _thread_filter.remove(slot_id);  // Remove from filtering initially (matches onThreadStart behavior)
   }
 

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

@@ -1,6 +1,6 @@
 /*
  * Copyright The async-profiler authors
- * Copyright 2025, Datadog, Inc.
+ * Copyright 2025, 2026, Datadog, Inc.
  * SPDX-License-Identifier: Apache-2.0
  */
 
@@ -104,6 +104,7 @@ class alignas(alignof(SpinLock)) Profiler {
   // ASGCT paths) or _total_samples (written by every recording path).
   alignas(DEFAULT_CACHE_LINE_SIZE) volatile u64 _sample_seq;
   alignas(DEFAULT_CACHE_LINE_SIZE) u64 _failures[ASGCT_FAILURE_TYPES];
+  bool _wall_precheck = false;
 
   SpinLock _class_map_lock;
   SpinLock _locks[CONCURRENCY_LEVEL];
@@ -367,7 +368,7 @@ class alignas(alignof(SpinLock)) Profiler {
   NativeFrameResolution resolveNativeFrameForWalkVM(uintptr_t pc, int lock_index);
   int convertNativeTrace(int native_frames, const void **callchain,
                          ASGCT_CallFrame *frames, int lock_index);
-  void recordSample(void *ucontext, u64 weight, int tid, jint event_type,
+  bool recordSample(void *ucontext, u64 weight, int tid, jint event_type,
                     u64 call_trace_id, Event *event);
   // Delegated sample path: stack-walking is performed by the HotSpot JFR
   // RequestStackTrace extension (the JVM emits the stack trace into its own

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

@@ -64,7 +64,6 @@ void ProfiledThread::release() {
   }
 }
 
-
 int ProfiledThread::currentTid() {
   ProfiledThread *tls = current();
   if (tls != NULL) {