feat(wall): support signal suppression for unfiltered threads

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

@@ -1,5 +1,5 @@
 /*
- * Copyright 2023 Datadog, Inc
+ * Copyright 2023, 2026 Datadog, Inc
  *
  * Licensed under the Apache License, Version 2.0 (the "License");
  * you may not use this file except in compliance with the License.
@@ -57,6 +57,8 @@
   X(THREAD_NAMES_COUNT, "thread_names_count")                                  \
   X(THREAD_FILTER_PAGES, "thread_filter_pages")                                \
   X(THREAD_FILTER_BYTES, "thread_filter_bytes")                                \
+  X(THREAD_REGISTRY_BOOTSTRAP_FAILURES, "thread_registry_bootstrap_failures")  \
+  X(THREAD_REGISTRY_STALE_SLOTS_RETIRED, "thread_registry_stale_slots_retired") \
   X(JMETHODID_SKIPPED, "jmethodid_skipped_count")                              \
   X(CODECACHE_NATIVE_SIZE_BYTES, "codecache_native_size_bytes")                \
   X(CODECACHE_NATIVE_COUNT, "native_codecache_count")                          \
@@ -67,6 +69,9 @@
   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_PRECHECK_REGISTRY_LOOKUPS, "wc_precheck_registry_lookups")             \
+  X(WC_PRECHECK_CANDIDATES_REJECTED, "wc_precheck_candidates_rejected")       \
+  X(WC_PRECHECK_LOOKUP_BUDGET_EXHAUSTED, "wc_precheck_lookup_budget_exhausted") \
   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")                             \

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

@@ -1,5 +1,6 @@
 /*
  * Copyright 2017 Andrei Pangin
+ * Copyright 2026, Datadog, Inc.
  *
  * Licensed under the Apache License, Version 2.0 (the "License");
  * you may not use this file except in compliance with the License.
@@ -51,6 +52,9 @@ class Engine {
   virtual void stop();
   virtual long interval() const { return 0L; }
 
+  // Whether empty-filter wall prechecks can consume ThreadFilter registry state.
+  virtual bool supportsUnfilteredWallPrecheck() const { return false; }
+
   virtual int registerThread(int tid) { return -1; }
   virtual void unregisterThread(int tid) {}
 

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

@@ -138,6 +138,30 @@ Java_com_datadoghq_profiler_JavaProfiler_getSamples(JNIEnv *env,
 
 // some duplication between add and remove, though we want to avoid having an extra branch in the hot path
 
+static ThreadFilter::SlotID ensureCurrentThreadFilterSlot(
+    ThreadFilter *thread_filter, ProfiledThread *current) {
+  int tid = current->tid();
+  if (unlikely(tid < 0)) {
+    return -1;
+  }
+
+  ThreadFilter::SlotID slot_id = current->filterSlotId();
+  if (likely(slot_id >= 0)) {
+    if (likely(thread_filter->activeSlotForId(slot_id, tid) != nullptr)) {
+      return slot_id;
+    }
+    current->setFilterSlotId(-1);
+  }
+
+  // Startup can register this TID centrally, but it cannot update another
+  // pthread's TLS. registerThread(tid) reuses that existing slot.
+  slot_id = thread_filter->registerThread(tid);
+  if (slot_id >= 0) {
+    current->setFilterSlotId(slot_id);
+  }
+  return slot_id;
+}
+
 // JavaCritical is faster JNI, but more restrictive - parameters and return value have to be
 // primitives or arrays of primitive types.
 // We direct corresponding JNI calls to JavaCritical to make sure the parameters/return value
@@ -155,24 +179,14 @@ JavaCritical_com_datadoghq_profiler_JavaProfiler_filterThreadAdd0() {
     return;
   }
   ThreadFilter *thread_filter = Profiler::instance()->threadFilter();
-  if (unlikely(!thread_filter->enabled())) {
+  if (unlikely(!thread_filter->registryActive())) {
     return;
   }
 
-  int slot_id = current->filterSlotId();
-  if (unlikely(slot_id == -1)) {
-    // Thread doesn't have a slot ID yet (e.g., main thread), so register it
-    // Happens when we are not enabled before thread start
-    slot_id = thread_filter->registerThread();
-    current->setFilterSlotId(slot_id);
-  }
-
-  if (unlikely(slot_id == -1)) {
+  int slot_id = ensureCurrentThreadFilterSlot(thread_filter, current);
+  if (unlikely(slot_id < 0)) {
     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);
 }
 
@@ -189,12 +203,13 @@ JavaCritical_com_datadoghq_profiler_JavaProfiler_filterThreadRemove0() {
     return;
   }
   ThreadFilter *thread_filter = Profiler::instance()->threadFilter();
-  if (unlikely(!thread_filter->enabled())) {
+  if (unlikely(!thread_filter->registryActive())) {
     return;
   }
 
   int slot_id = current->filterSlotId();
-  if (unlikely(slot_id == -1)) {
+  if (unlikely(slot_id == -1 ||
+               thread_filter->activeSlotForId(slot_id, tid) == nullptr)) {
     // Thread doesn't have a slot ID yet - nothing to remove
     return;
   }
@@ -351,8 +366,8 @@ Java_com_datadoghq_profiler_JavaProfiler_parkEnter0(JNIEnv *env, jclass unused)
   }
   bool first_park = current->parkEnter();
   ThreadFilter *tf = Profiler::instance()->threadFilter();
-  if (first_park && tf->enabled()) {
-    ThreadFilter::SlotID slot_id = current->filterSlotId();
+  if (first_park && tf->registryActive()) {
+    ThreadFilter::SlotID slot_id = ensureCurrentThreadFilterSlot(tf, current);
     if (slot_id >= 0) {
       current->setParkBlockToken(
           tf->enterBlockedRun(slot_id, OSThreadState::CONDVAR_WAIT));
@@ -373,9 +388,10 @@ Java_com_datadoghq_profiler_JavaProfiler_parkExit0(
     return;
   }
   ThreadFilter *tf = Profiler::instance()->threadFilter();
-  if (tf->enabled()) {
+  if (tf->registryActive()) {
     ThreadFilter::SlotID slot_id = ThreadFilter::tokenSlotId(park_block_token);
-    if (current->filterSlotId() == slot_id) {
+    if (tf->activeSlotForId(current->filterSlotId(), current->tid()) != nullptr &&
+        current->filterSlotId() == slot_id) {
       tf->exitBlockedRun(slot_id, ThreadFilter::tokenGeneration(park_block_token));
     }
   }
@@ -402,10 +418,10 @@ Java_com_datadoghq_profiler_JavaProfiler_blockEnter0(
     return 0;
   }
   ThreadFilter *tf = Profiler::instance()->threadFilter();
-  if (!tf->enabled()) {
+  if (!tf->registryActive()) {
     return 0;
   }
-  ThreadFilter::SlotID slot_id = current->filterSlotId();
+  ThreadFilter::SlotID slot_id = ensureCurrentThreadFilterSlot(tf, current);
   if (slot_id < 0) {
     return 0;
   }
@@ -423,12 +439,13 @@ Java_com_datadoghq_profiler_JavaProfiler_blockExit0(
   if (current == nullptr) {
     return;
   }
+  ThreadFilter *tf = Profiler::instance()->threadFilter();
   ThreadFilter::SlotID slot_id = ThreadFilter::tokenSlotId(block_token);
-  if (current->filterSlotId() != slot_id) {
+  if (current->filterSlotId() != slot_id ||
+      tf->activeSlotForId(slot_id, current->tid()) == nullptr) {
     return;
   }
-  ThreadFilter *tf = Profiler::instance()->threadFilter();
-  if (tf->enabled()) {
+  if (tf->registryActive()) {
     tf->exitBlockedRun(slot_id, ThreadFilter::tokenGeneration(block_token));
   }
 }

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

@@ -29,6 +29,10 @@ bool JVMThread::initialize() {
   return _jvm_thread.initialize(current_thread);
 }
 
+bool JVMThread::supportsNativeThreadIdLookup() {
+  return VM::isOpenJ9() || VMThread::hasNativeThreadId();
+}
+
 int JVMThread::nativeThreadId(JNIEnv* jni, jthread thread) {
     return VM::isOpenJ9() ? J9Support::GetOSThreadID(thread) : VMThread::nativeThreadId(jni, thread);
 }

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

@@ -23,9 +23,12 @@ class JVMThread {
     /*
      * The initialization happens in early startup, in single-threaded mode,
      * no synchronization is needed
-     */
+    */
     static bool initialize();
 
+    // Whether nativeThreadId can resolve a Java thread other than the caller.
+    static bool supportsNativeThreadIdLookup();
+
     static inline bool isInitialized() {
         return _tid != nullptr && _jvm_thread.isKeyValid();
     }

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

@@ -83,11 +83,9 @@ void Profiler::onThreadStart(jvmtiEnv *jvmti, JNIEnv *jni, jthread thread) {
 
   current->setJavaThread(true);
   int tid = current->tid();
-  if (_thread_filter.enabled()) {
-    int slot_id = _thread_filter.registerThread();
+  if (_thread_filter.registryActive()) {
+    int slot_id = _thread_filter.registerThread(tid);
     current->setFilterSlotId(slot_id);
-    _thread_filter.resetSlotRunState(slot_id);
-    _thread_filter.remove(slot_id);  // Remove from filtering initially
   }
   if (thread != NULL) {
     updateThreadName(jvmti, jni, thread, true);
@@ -107,9 +105,11 @@ void Profiler::onThreadEnd(jvmtiEnv *jvmti, JNIEnv *jni, jthread thread) {
     int slot_id = current->filterSlotId();
     tid = current->tid();
 
-    if (_thread_filter.enabled()) {
-      _thread_filter.unregisterThread(slot_id);
+    if (slot_id >= 0) {
+      _thread_filter.unregisterThread(slot_id, tid);
       current->setFilterSlotId(-1);
+    } else {
+      _thread_filter.unregisterThreadByTid(tid);
     }
 
     updateThreadName(jvmti, jni, thread, false);
@@ -134,6 +134,7 @@ void Profiler::onThreadEnd(jvmtiEnv *jvmti, JNIEnv *jni, jthread thread) {
   }
 
   updateThreadName(jvmti, jni, thread, false);
+  _thread_filter.unregisterThreadByTid(tid);
   _cpu_engine->unregisterThread(tid);
   _wall_engine->unregisterThread(tid);
   LivenessTracker::instance()->releaseThreadLocalState();
@@ -1075,6 +1076,38 @@ void Profiler::updateJavaThreadNames() {
   jvmti->Deallocate((unsigned char *)thread_objects);
 }
 
+void Profiler::registerExistingJavaThreads() {
+  if (!_thread_filter.unfilteredWallTrackingActive() ||
+      !JVMThread::supportsNativeThreadIdLookup()) {
+    return;
+  }
+
+  jvmtiEnv *jvmti = VM::jvmti();
+  JNIEnv *jni = VM::jni();
+  jint thread_count;
+  jthread *thread_objects;
+  if (jvmti->GetAllThreads(&thread_count, &thread_objects) != JVMTI_ERROR_NONE) {
+    Counters::increment(THREAD_REGISTRY_BOOTSTRAP_FAILURES);
+    return;
+  }
+
+  for (int i = 0; i < thread_count; ++i) {
+    jthread thread = thread_objects[i];
+    if (thread != nullptr) {
+      int tid = JVMThread::nativeThreadId(jni, thread);
+      if (tid >= 0) {
+        _thread_filter.registerThread(tid);
+      }
+      jni->DeleteLocalRef(thread);
+    }
+  }
+  jvmti->Deallocate(reinterpret_cast<unsigned char *>(thread_objects));
+  int retired = _thread_filter.retireInactiveRegistrations();
+  if (retired > 0) {
+    Counters::increment(THREAD_REGISTRY_STALE_SLOTS_RETIRED, retired);
+  }
+}
+
 void Profiler::updateNativeThreadNames(bool defer_initializing) {
   ThreadList *thread_list = OS::listThreads();
   constexpr size_t buffer_size = 64;
@@ -1402,24 +1435,33 @@ Error Profiler::start(Arguments &args, bool reset) {
     _safe_mode |= GC_TRACES | LAST_JAVA_PC;
   }
 
-  // TODO: Current way of setting filter is weird with the recent changes
-  _thread_filter.init(args._filter ? args._filter : "0");
-
-  // Minor optim: Register the current thread (start thread won't be called)
-  if (_thread_filter.enabled()) {
+  _cpu_engine = selectCpuEngine(args);
+  _wall_engine = selectWallEngine(args);
+
+  const char *filter = args._filter != nullptr ? args._filter : "0";
+  const bool track_unfiltered_wall =
+      (_event_mask & EM_WALL) != 0 && args._wall_precheck &&
+      args._filter != nullptr && args._filter[0] == '\0' &&
+      _wall_engine->supportsUnfilteredWallPrecheck();
+  _thread_filter.init(filter, track_unfiltered_wall);
+
+  // Reset per-recording state before a wall timer can inspect the registry.
+  if (_thread_filter.registryActive()) {
     _thread_filter.clearActive();
+  }
+
+  // Preserve the context-filter fast path. Unfiltered tracking bootstraps the
+  // current thread only after the wall engine has started successfully.
+  if (_thread_filter.enabled()) {
     ProfiledThread *current = ProfiledThread::initCurrentThreadSignalSafe();
     assert(current != nullptr);
     int slot_id = current->filterSlotId();
     if (slot_id < 0) {
-      slot_id = _thread_filter.registerThread();
+      slot_id = _thread_filter.registerThread(current->tid());
       current->setFilterSlotId(slot_id);
     }
-    _thread_filter.remove(slot_id);  // Remove from filtering initially (matches onThreadStart behavior)
   }
 
-  _cpu_engine = selectCpuEngine(args);
-  _wall_engine = selectWallEngine(args);
   _cstack = args._cstack;
   if (_cstack == CSTACK_DEFAULT) {
     if (VMStructs::hasStackStructs() && OS::isLinux()) {
@@ -1470,6 +1512,7 @@ Error Profiler::start(Arguments &args, bool reset) {
   _num_context_attributes = args._context_attributes.size();
   error = _jfr.start(args, reset);
   if (error) {
+    _thread_filter.deactivateRecording();
     switchLibraryTrap(false);
     _libs->stopRefresher();
     return error;
@@ -1519,6 +1562,7 @@ Error Profiler::start(Arguments &args, bool reset) {
       Log::warn("%s", error.message());
       if (_event_mask == EM_NATIVEMEM) {
         // nativemem is the only requested mode: propagate the real error
+        _thread_filter.deactivateRecording();
         disableEngines();
         switchLibraryTrap(false);
         _libs->stopRefresher();
@@ -1542,9 +1586,19 @@ Error Profiler::start(Arguments &args, bool reset) {
     }
   }
 
+  // A recoverable wall-engine failure must not leave registry work enabled for
+  // unrelated engines that did start successfully.
+  if (track_unfiltered_wall && (activated & EM_WALL) == 0) {
+    _thread_filter.init(filter, false);
+  }
+
   if (activated) {
     switchThreadEvents(JVMTI_ENABLE);
 
+    // ThreadStart events cover only threads created after the callbacks are
+    // enabled. Bootstrap registry identity for Java threads that already exist.
+    registerExistingJavaThreads();
+
     // Initialize this thread
     // Note: passing all nullptrs results in not able to resolve the thread name here.
     //      However, the thread name will be updated later in updateJavaThreadNames().
@@ -1566,6 +1620,7 @@ Error Profiler::start(Arguments &args, bool reset) {
     return Error::OK;
   }
   // no engine was activated; perform cleanup
+  _thread_filter.deactivateRecording();
   disableEngines();
   switchLibraryTrap(false);
   _libs->stopRefresher();
@@ -1617,6 +1672,8 @@ Error Profiler::stop() {
   if (_event_mask & EM_CPU)
     _cpu_engine->stop();
 
+  _thread_filter.deactivateRecording();
+
   switchLibraryTrap(false);
   switchThreadEvents(JVMTI_DISABLE);
   Libraries::instance()->refresh();

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

@@ -152,6 +152,7 @@ class alignas(alignof(SpinLock)) Profiler {
   void updateThreadName(jvmtiEnv *jvmti, JNIEnv *jni, jthread thread,
                         bool self = false);
   void updateJavaThreadNames();
+  void registerExistingJavaThreads();
   void mangle(const char *name, char *buf, size_t size);
 
   Engine *selectCpuEngine(Arguments &args);
@@ -233,7 +234,6 @@ class alignas(alignof(SpinLock)) Profiler {
   // dump-time pass (which passes false), records the final name instead.
   void updateNativeThreadNames(bool defer_initializing = false);
 
-
   inline void incFailure(int type) {
     if (type < ASGCT_FAILURE_TYPES) {
       atomicIncRelaxed(_failures[type]);