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.
@@ -60,6 +60,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_CAPACITY_EXHAUSTED, "thread_registry_capacity_exhausted")  \
+  X(THREAD_REGISTRY_INDEX_FAILURES, "thread_registry_index_failures")          \
   X(JMETHODID_SKIPPED, "jmethodid_skipped_count")                              \
   X(CODECACHE_NATIVE_SIZE_BYTES, "codecache_native_size_bytes")                \
   X(CODECACHE_NATIVE_COUNT, "native_codecache_count")                          \
@@ -70,6 +72,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/frames.h

@@ -1,9 +1,39 @@
+/*
+ * 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.
+ * You may obtain a copy of the License at
+ *
+ *     http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
 #ifndef _FRAMES_H
 #define _FRAMES_H
 
 #include <jni.h>
+#include <jvmti.h>
 #include "vmEntry.h"
 
+inline void copyJvmtiFrames(ASGCT_CallFrame *frames,
+                            const jvmtiFrameInfo *jvmti_frames,
+                            jint num_frames) {
+  // The source and destination commonly refer to the two views of the same
+  // CallTraceBuffer union. Read both source fields before either write.
+  for (jint i = 0; i < num_frames; ++i) {
+    jmethodID method = jvmti_frames[i].method;
+    jlocation location = jvmti_frames[i].location;
+    frames[i].method_id = method;
+    frames[i].bci = static_cast<jint>(location);
+    LP64_ONLY(frames[i].padding = 0;)
+  }
+}
+
 inline int makeFrame(ASGCT_CallFrame *frames, jint type, jmethodID id) {
   frames[0].bci = type;
   frames[0].method_id = id;

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.h

@@ -23,7 +23,7 @@ class JVMThread {
     /*
      * The initialization happens in early startup, in single-threaded mode,
      * no synchronization is needed
-     */
+    */
     static bool initialize();
 
     static inline bool isInitialized() {

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

@@ -85,11 +85,12 @@ 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();
+  // Java lifecycle callbacks own registry allocation. The wall timer only
+  // looks up these entries and must never allocate slots for arbitrary OS
+  // threads returned by OS::listThreads().
+  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);
@@ -109,9 +110,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);
@@ -136,6 +139,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();
@@ -565,14 +569,7 @@ u64 Profiler::recordJVMTISample(u64 counter, int tid, jthread thread, jint event
     if (VM::jvmti()->GetStackTrace(thread, 0, _max_stack_depth, jvmti_frames, &num_frames) == JVMTI_ERROR_NONE && num_frames > 0) {
       // Convert to AsyncGetCallTrace format.
       // Note: jvmti_frames and frames may overlap.
-      for (int i = 0; i < num_frames; i++) {
-        jint bci = jvmti_frames[i].location;
-        jmethodID mid = jvmti_frames[i].method;
-        frames[i].method_id = mid;
-        frames[i].bci = bci;
-        // see https://github.com/async-profiler/async-profiler/pull/1090
-        LP64_ONLY(frames[i].padding = 0;)
-      }
+      copyJvmtiFrames(frames, jvmti_frames, num_frames);
       // On JDK 21+, GetStackTrace on a virtual thread returns only the VT's
       // logical stack; it stops at the continuation boundary and never includes
       // carrier-thread frames.  Without a synthetic root the trace appears
@@ -1463,24 +1460,34 @@ 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)
+  _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);
+
+  // Unfiltered init resets registrations before publishing registry admission.
+  // Context-filtered recordings retain identities but must clear membership.
   if (_thread_filter.enabled()) {
     _thread_filter.clearActive();
+  }
+
+  // Preserve the context-filter fast path. Unfiltered Java threads are
+  // registered by ThreadStart callbacks after the engines are activated.
+  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()) {
@@ -1531,6 +1538,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;
@@ -1580,6 +1588,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();
@@ -1603,6 +1612,12 @@ 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);
 
@@ -1627,6 +1642,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();
@@ -1678,6 +1694,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

@@ -250,7 +250,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]);