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

* feat: wall-clock precheck and signal suppression

* fix

* fix: fix build + tests

* fix: fix mem leaks in tests

* fix: track wall precheck block state in thread filter

* fix: arm wall precheck after recording sample

* fix: include wait states in wall precheck suppression

* Fix ProfiledThread ownership in park_state_ut

* Add Java block-state bridge for wall-clock precheck

* Fix wall-clock thread filter reset

* Gate wall-clock precheck on untraced context

* fix: avoid exact suppression for unowned blocked states

* fix: address ownership correctness review

* fix: address  thread filter review

* fix: factorize code and add support for jvmti

* fix: fix wall-clock counters and misleading comment

* fix: apply review about test and unused stuff

* fix: apply review about drainSuppressedSampledRun

* fix: clean up branch based on PR review recommendations

* fix: remove TaskBlock snapshot mechanism

* fix: preserve wall precheck tail weight

* fix: cover wall precheck tail flush in test

* fix: internalize owned blocking hooks

Author: kaahos

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

@@ -379,6 +379,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;
@@ -207,6 +208,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,10 @@
   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_UNOWNED_BLOCKED_SUPPRESSED, "wc_unowned_blocked_suppressed")            \
+  X(WC_UNOWNED_BLOCKED_RECORDED, "wc_unowned_blocked_recorded")                \
+  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

@@ -122,12 +122,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; }
 
@@ -166,13 +167,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

@@ -1822,6 +1822,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);
 }
@@ -2102,7 +2103,7 @@ void FlightRecorder::recordHeapUsage(int lock_index, long value, bool live) {
   }
 }
 
-void FlightRecorder::recordEvent(int lock_index, int tid, u64 call_trace_id,
+bool FlightRecorder::recordEvent(int lock_index, int tid, u64 call_trace_id,
                                  int event_type, Event *event) {
   OptionalSharedLockGuard locker(&_rec_lock);
   if (locker.ownsLock()) {
@@ -2137,16 +2138,20 @@ void FlightRecorder::recordEvent(int lock_index, int tid, u64 call_trace_id,
         case BCI_NATIVE_SOCKET:
           rec->recordNativeSocketSample(buf, tid, call_trace_id, (NativeSocketEvent *)event);
           break;
+        default:
+          return false;
         }
         rec->flushIfNeeded(buf);
         rec->addThread(lock_index, tid);
+        return true;
       }
   } else {
     Counters::increment(SAMPLES_DROPPED_REC_LOCK);
   }
+  return false;
 }
 
-void FlightRecorder::recordEventDelegated(int lock_index, int tid,
+bool FlightRecorder::recordEventDelegated(int lock_index, int tid,
                                           u64 correlation_id, int event_type,
                                           Event *event) {
   OptionalSharedLockGuard locker(&_rec_lock);
@@ -2165,14 +2170,16 @@ void FlightRecorder::recordEventDelegated(int lock_index, int tid,
           break;
         default:
           // Delegation is only wired for CPU/wall samples in v1.
-          break;
+          return false;
       }
       rec->flushIfNeeded(buf);
       rec->addThread(lock_index, tid);
+      return true;
     }
   } else {
     Counters::increment(SAMPLES_DROPPED_REC_LOCK);
   }
+  return false;
 }
 
 void FlightRecorder::recordLog(LogLevel level, const char *message,

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

@@ -446,14 +446,14 @@ class FlightRecorder {
 
   bool active() const { return _rec != NULL; }
 
-  void recordEvent(int lock_index, int tid, u64 call_trace_id, int event_type,
+  bool 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.StackTraceRequest.
-  void recordEventDelegated(int lock_index, int tid, u64 correlation_id,
+  bool recordEventDelegated(int lock_index, int tid, u64 correlation_id,
                             int event_type, Event *event);
 
   void recordLog(LogLevel level, const char *message, size_t len);

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

@@ -11,10 +11,12 @@
 #include "jvmThread.h"
 
 VMThread* VMThread::current() {
+    assert(VM::isHotspot());
     return VMThread::cast(JVMThread::current());
 }
 
 VMThread* VMThread::fromJavaThread(JNIEnv* env, jthread thread) {
+    assert(VM::isHotspot());
     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,95 @@ 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;
+  }
+  bool first_park = current->parkEnter();
+  ThreadFilter *tf = Profiler::instance()->threadFilter();
+  if (first_park && tf->enabled()) {
+    ThreadFilter::SlotID slot_id = current->filterSlotId();
+    if (slot_id >= 0) {
+      current->setParkBlockToken(
+          tf->enterBlockedRun(slot_id, 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 park_block_token = 0;
+  if (!current->parkExit(park_block_token) || park_block_token == 0) {
+    return;
+  }
+  ThreadFilter *tf = Profiler::instance()->threadFilter();
+  if (tf->enabled()) {
+    ThreadFilter::SlotID slot_id = ThreadFilter::tokenSlotId(park_block_token);
+    if (current->filterSlotId() == slot_id) {
+      tf->exitBlockedRun(slot_id, ThreadFilter::tokenGeneration(park_block_token));
+    }
+  }
+}
+
+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;
+  }
+  return static_cast<jlong>(tf->enterBlockedRun(slot_id, decoded));
+}
+
+extern "C" DLLEXPORT void JNICALL
+Java_com_datadoghq_profiler_JavaProfiler_blockExit0(
+    JNIEnv *env, jclass unused, jlong token) {
+  u64 block_token = static_cast<u64>(token);
+  if (block_token == 0) {
+    return;
+  }
+  ProfiledThread *current = ProfiledThread::current();
+  if (current == nullptr) {
+    return;
+  }
+  ThreadFilter::SlotID slot_id = ThreadFilter::tokenSlotId(block_token);
+  if (current->filterSlotId() != slot_id) {
+    return;
+  }
+  ThreadFilter *tf = Profiler::instance()->threadFilter();
+  if (tf->enabled()) {
+    tf->exitBlockedRun(slot_id, ThreadFilter::tokenGeneration(block_token));
+  }
+}
+
 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")