feat(wall): add configurable wall thread scope

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

@@ -410,6 +410,17 @@ Error Arguments::parse(const char *args) {
           }
       }
 
+      CASE("wallscope")
+      if (value == NULL || value[0] == 0) {
+        msg = "wallscope must be 'context' or 'all'";
+      } else if (strcmp(value, "context") == 0) {
+        _wall_scope = WALL_SCOPE_CONTEXT;
+      } else if (strcmp(value, "all") == 0) {
+        _wall_scope = WALL_SCOPE_ALL;
+      } else {
+        msg = "wallscope must be 'context' or 'all'";
+      }
+
       CASE("nativemem")
       _nativemem = value == NULL ? 0 : parseUnits(value, BYTES);
       if (_nativemem < 0) {

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

@@ -87,6 +87,11 @@ enum WallclockSampler {
     JVMTI
 };
 
+enum WallScope {
+    WALL_SCOPE_CONTEXT,
+    WALL_SCOPE_ALL
+};
+
 enum Clock {
     CLK_DEFAULT,
     CLK_TSC,
@@ -161,6 +166,10 @@ class Arguments {
   }
 
 public:
+  bool wallScopeAllThreads() const {
+    return _wall_scope == WALL_SCOPE_ALL;
+  }
+
   Action _action;
   Ring _ring;
   const char *_event;
@@ -171,6 +180,7 @@ class Arguments {
   bool _wall_precheck;
   int _wall_threads_per_tick;
   WallclockSampler _wallclock_sampler;
+  WallScope _wall_scope;
   long _memory;
   bool _record_allocations;
   bool _record_liveness;
@@ -212,6 +222,7 @@ class Arguments {
         _wall_precheck(false),
         _wall_threads_per_tick(DEFAULT_WALL_THREADS_PER_TICK),
         _wallclock_sampler(ASGCT),
+        _wall_scope(WALL_SCOPE_CONTEXT),
         _memory(-1),
         _record_allocations(false),
         _record_liveness(false),

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

@@ -57,6 +57,7 @@
   X(THREAD_NAMES_COUNT, "thread_names_count")                                  \
   X(THREAD_FILTER_PAGES, "thread_filter_pages")                                \
   X(THREAD_FILTER_BYTES, "thread_filter_bytes")                                \
+  X(THREAD_FILTER_CAPACITY_EXHAUSTED, "thread_filter_capacity_exhausted")      \
   X(JMETHODID_SKIPPED, "jmethodid_skipped_count")                              \
   X(CODECACHE_NATIVE_SIZE_BYTES, "codecache_native_size_bytes")                \
   X(CODECACHE_NATIVE_COUNT, "native_codecache_count")                          \

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

@@ -144,24 +144,21 @@ 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();
+    slot_id = thread_filter->registerThread(tid);
     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);
 }
 
@@ -174,7 +171,7 @@ JavaCritical_com_datadoghq_profiler_JavaProfiler_filterThreadRemove0() {
     return;
   }
   ThreadFilter *thread_filter = Profiler::instance()->threadFilter();
-  if (unlikely(!thread_filter->enabled())) {
+  if (unlikely(!thread_filter->registryActive())) {
     return;
   }
 

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

@@ -78,11 +78,9 @@ void Profiler::onThreadStart(jvmtiEnv *jvmti, JNIEnv *jni, jthread thread) {
   ProfiledThread *current = ProfiledThread::current();
   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);
@@ -101,9 +99,11 @@ void Profiler::onThreadEnd(jvmtiEnv *jvmti, JNIEnv *jni, jthread thread) {
     int slot_id = current->filterSlotId();
     tid = current->tid();
 
-    if (_thread_filter.enabled()) {
+    if (slot_id >= 0) {
       _thread_filter.unregisterThread(slot_id);
       current->setFilterSlotId(-1);
+    } else {
+      _thread_filter.unregisterThreadByTid(tid);
     }
 
     updateThreadName(jvmti, jni, thread, false);
@@ -127,6 +127,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);
 }
@@ -1051,6 +1052,45 @@ void Profiler::updateJavaThreadNames() {
   jvmti->Deallocate((unsigned char *)thread_objects);
 }
 
+void Profiler::registerThreadsForAllThreadScope(const std::vector<int>& tids) {
+  // All-thread wall scope is the only mode that needs a pre-populated registry;
+  // context scope registers threads lazily via the ThreadStart callback.
+  if (!_thread_filter.allThreads()) {
+    return;
+  }
+  for (int tid : tids) {
+    if (tid >= 0) {
+      _thread_filter.registerThread(tid);
+    }
+  }
+}
+
+void Profiler::registerExistingJavaThreads() {
+  // The all-thread scope guard lives in registerThreadsForAllThreadScope();
+  // enumerating threads here in context scope is a one-time, harmless startup
+  // cost, and keeping the decision in one tested place avoids a second guard
+  // that no test can reach without a live JVMTI environment.
+  jvmtiEnv *jvmti = VM::jvmti();
+  JNIEnv *jni = VM::jni();
+  jint thread_count;
+  jthread *thread_objects;
+  if (jvmti->GetAllThreads(&thread_count, &thread_objects) != JVMTI_ERROR_NONE) {
+    return;
+  }
+
+  std::vector<int> tids;
+  tids.reserve(thread_count);
+  for (int i = 0; i < thread_count; ++i) {
+    jthread thread = thread_objects[i];
+    if (thread != nullptr) {
+      tids.push_back(JVMThread::nativeThreadId(jni, thread));
+      jni->DeleteLocalRef(thread);
+    }
+  }
+  jvmti->Deallocate(reinterpret_cast<unsigned char *>(thread_objects));
+  registerThreadsForAllThreadScope(tids);
+}
+
 void Profiler::updateNativeThreadNames(bool defer_initializing) {
   ThreadList *thread_list = OS::listThreads();
   constexpr size_t buffer_size = 64;
@@ -1354,19 +1394,22 @@ Error Profiler::start(Arguments &args, bool reset) {
   }
 
   // TODO: Current way of setting filter is weird with the recent changes
-  _thread_filter.init(args._filter ? args._filter : "0");
+  const bool all_threads = args.wallScopeAllThreads();
+  const char *filter = args._wall_scope == WALL_SCOPE_CONTEXT
+                           ? "0"
+                           : (args._filter ? args._filter : "0");
+  _thread_filter.init(filter, all_threads);
 
   // Minor optim: Register the current thread (start thread won't be called)
-  if (_thread_filter.enabled()) {
+  if (_thread_filter.registryActive()) {
     _thread_filter.clearActive();
     ProfiledThread *current = ProfiledThread::current();
     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);
@@ -1496,6 +1539,10 @@ Error Profiler::start(Arguments &args, bool reset) {
   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().

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,6 +234,13 @@ class alignas(alignof(SpinLock)) Profiler {
   // dump-time pass (which passes false), records the final name instead.
   void updateNativeThreadNames(bool defer_initializing = false);
 
+  // Registers the supplied native thread ids in the thread filter, but only
+  // when all-thread wall scope is active. In context scope the registry is
+  // populated lazily through the ThreadStart callback, so this bootstrap must
+  // do nothing. Split out from registerExistingJavaThreads() so the scope guard
+  // can be exercised without a live JVMTI environment.
+  void registerThreadsForAllThreadScope(const std::vector<int>& tids);
+
 
   inline void incFailure(int type) {
     if (type < ASGCT_FAILURE_TYPES) {