From 1118f62831a4384bdf0283ef01a626c5bdc48af5 Mon Sep 17 00:00:00 2001 From: Paul Fournillon Date: Sun, 19 Jul 2026 12:53:47 +0200 Subject: [PATCH 1/2] feat(wall): track all threads for unfiltered wall-clock precheck --- ddprof-lib/src/main/cpp/counters.h | 7 +- ddprof-lib/src/main/cpp/engine.h | 4 + ddprof-lib/src/main/cpp/javaApi.cpp | 65 +-- ddprof-lib/src/main/cpp/jvmThread.cpp | 4 + ddprof-lib/src/main/cpp/jvmThread.h | 5 +- ddprof-lib/src/main/cpp/profiler.cpp | 87 +++- ddprof-lib/src/main/cpp/profiler.h | 2 +- ddprof-lib/src/main/cpp/threadFilter.cpp | 350 +++++++++++++-- ddprof-lib/src/main/cpp/threadFilter.h | 137 +++++- ddprof-lib/src/main/cpp/wallClock.cpp | 114 +++-- ddprof-lib/src/main/cpp/wallClock.h | 58 ++- .../src/main/cpp/wallClockCandidateSelector.h | 64 +++ ddprof-lib/src/test/cpp/threadFilter_ut.cpp | 402 +++++++++++++++++- .../cpp/wallClockCandidateSelector_ut.cpp | 173 ++++++++ .../src/test/cpp/wallprecheck_args_ut.cpp | 29 ++ .../WallClockPrecheckBenchmarkHooks.java | 21 + .../WallClockPrecheckOverheadBenchmark.java | 145 +++++++ .../profiler/AbstractProfilerTest.java | 19 +- .../J9WallClockPrecheckCapabilityTest.java | 36 ++ .../JvmtiBasedUnfilteredWallPrecheckTest.java | 44 ++ .../UnfilteredWallPrecheckRestartTest.java | 127 ++++++ .../wallclock/UnfilteredWallPrecheckTest.java | 239 +++++++++++ 22 files changed, 2019 insertions(+), 113 deletions(-) create mode 100644 ddprof-lib/src/main/cpp/wallClockCandidateSelector.h create mode 100644 ddprof-lib/src/test/cpp/wallClockCandidateSelector_ut.cpp create mode 100644 ddprof-stresstest/src/jmh/java/com/datadoghq/profiler/WallClockPrecheckBenchmarkHooks.java create mode 100644 ddprof-stresstest/src/jmh/java/com/datadoghq/profiler/stresstest/scenarios/throughput/WallClockPrecheckOverheadBenchmark.java create mode 100644 ddprof-test/src/test/java/com/datadoghq/profiler/wallclock/J9WallClockPrecheckCapabilityTest.java create mode 100644 ddprof-test/src/test/java/com/datadoghq/profiler/wallclock/JvmtiBasedUnfilteredWallPrecheckTest.java create mode 100644 ddprof-test/src/test/java/com/datadoghq/profiler/wallclock/UnfilteredWallPrecheckRestartTest.java create mode 100644 ddprof-test/src/test/java/com/datadoghq/profiler/wallclock/UnfilteredWallPrecheckTest.java diff --git a/ddprof-lib/src/main/cpp/counters.h b/ddprof-lib/src/main/cpp/counters.h index 34b2e908dd..5d3286768d 100644 --- a/ddprof-lib/src/main/cpp/counters.h +++ b/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") \ diff --git a/ddprof-lib/src/main/cpp/engine.h b/ddprof-lib/src/main/cpp/engine.h index c9e75e2986..9d0de68108 100644 --- a/ddprof-lib/src/main/cpp/engine.h +++ b/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) {} diff --git a/ddprof-lib/src/main/cpp/javaApi.cpp b/ddprof-lib/src/main/cpp/javaApi.cpp index 83f6045c1a..a7be9f44b7 100644 --- a/ddprof-lib/src/main/cpp/javaApi.cpp +++ b/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)); } } diff --git a/ddprof-lib/src/main/cpp/jvmThread.cpp b/ddprof-lib/src/main/cpp/jvmThread.cpp index d249d943f7..f4f24a3978 100644 --- a/ddprof-lib/src/main/cpp/jvmThread.cpp +++ b/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); } diff --git a/ddprof-lib/src/main/cpp/jvmThread.h b/ddprof-lib/src/main/cpp/jvmThread.h index 2f5bd69104..8c7988d503 100644 --- a/ddprof-lib/src/main/cpp/jvmThread.h +++ b/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(); } diff --git a/ddprof-lib/src/main/cpp/profiler.cpp b/ddprof-lib/src/main/cpp/profiler.cpp index ca376873e6..fc25018abe 100644 --- a/ddprof-lib/src/main/cpp/profiler.cpp +++ b/ddprof-lib/src/main/cpp/profiler.cpp @@ -82,11 +82,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); @@ -106,9 +104,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); @@ -132,6 +132,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); } @@ -1056,6 +1057,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(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; @@ -1383,24 +1416,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()) { @@ -1451,6 +1493,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; @@ -1500,6 +1543,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(); @@ -1523,9 +1567,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(). @@ -1547,6 +1601,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(); @@ -1598,6 +1653,8 @@ Error Profiler::stop() { if (_event_mask & EM_CPU) _cpu_engine->stop(); + _thread_filter.deactivateRecording(); + switchLibraryTrap(false); switchThreadEvents(JVMTI_DISABLE); Libraries::instance()->refresh(); diff --git a/ddprof-lib/src/main/cpp/profiler.h b/ddprof-lib/src/main/cpp/profiler.h index c4d7e6e24e..f532e297a8 100644 --- a/ddprof-lib/src/main/cpp/profiler.h +++ b/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]); diff --git a/ddprof-lib/src/main/cpp/threadFilter.cpp b/ddprof-lib/src/main/cpp/threadFilter.cpp index 531ce75a1a..24c25825a5 100644 --- a/ddprof-lib/src/main/cpp/threadFilter.cpp +++ b/ddprof-lib/src/main/cpp/threadFilter.cpp @@ -32,12 +32,16 @@ ThreadFilter::ShardHead ThreadFilter::_free_heads[ThreadFilter::kShardCount] {}; -ThreadFilter::ThreadFilter() : _enabled(false) { +ThreadFilter::ThreadFilter() + : _enabled(false), _registry_active(false), _track_unfiltered_wall(false) { // Initialize chunk pointers to null (lazy allocation) for (int i = 0; i < kMaxChunks; ++i) { _chunks[i].store(nullptr, std::memory_order_relaxed); } _free_list = std::make_unique(kFreeListSize); + for (auto& entry : _tid_index) { + entry.store(0, std::memory_order_relaxed); + } // Initialize the first chunk initializeChunk(0); @@ -51,6 +55,9 @@ ThreadFilter::ThreadFilter() : _enabled(false) { ThreadFilter::~ThreadFilter() { // Make the filter inert for any concurrent readers _enabled.store(false, std::memory_order_release); + _registry_active.store(false, std::memory_order_release); + _track_unfiltered_wall.store(false, std::memory_order_release); + _recording_epoch.store(0, std::memory_order_release); // Reset free-list heads and nodes first for (int s = 0; s < kShardCount; ++s) { _free_heads[s].head.store(-1, std::memory_order_relaxed); @@ -59,6 +66,9 @@ ThreadFilter::~ThreadFilter() { _free_list[i].value.store(-1, std::memory_order_relaxed); _free_list[i].next.store(-1, std::memory_order_relaxed); } + for (auto& entry : _tid_index) { + entry.store(0, std::memory_order_relaxed); + } // Publish 0 chunks to stop range scans (collect) _num_chunks.store(0, std::memory_order_release); // Detach and delete chunks @@ -78,7 +88,9 @@ void ThreadFilter::initializeChunk(int chunk_idx) { // Allocate and initialize new chunk completely before swapping ChunkStorage* new_chunk = new ChunkStorage(); for (auto& slot : new_chunk->slots) { - slot.value.store(-1, std::memory_order_relaxed); + slot.tid.store(-1, std::memory_order_relaxed); + slot.recording_epoch.store(0, std::memory_order_relaxed); + slot.context_window_state.store(0, std::memory_order_relaxed); slot.active_block_state.store(OSThreadState::UNKNOWN, std::memory_order_relaxed); } @@ -92,15 +104,42 @@ void ThreadFilter::initializeChunk(int chunk_idx) { } } -ThreadFilter::SlotID ThreadFilter::registerThread() { - // If disabled, block new registrations - if (!_enabled.load(std::memory_order_acquire)) { +ThreadFilter::SlotID ThreadFilter::registerThread(int tid) { + if (!_registry_active.load(std::memory_order_acquire)) { return -1; } + std::lock_guard lock(_registry_lock); + + if (tid >= 0) { + SlotID existing = lookupSlotIdByTid(tid); + if (existing >= 0) { + RecordingEpoch epoch = recordingEpoch(); + if (epoch != 0) { + refreshSlotForRecording(slotForId(existing), epoch); + } + return existing; + } + } + + RecordingEpoch epoch = recordingEpoch(); // First, try to get a slot from the free list (lock-free stack) SlotID reused_slot = popFromFreeList(); if (reused_slot >= 0) { + Slot* slot = slotForId(reused_slot); + slot->lifecycle_generation.fetch_add(1, std::memory_order_acq_rel); + slot->recording_epoch.store(0, std::memory_order_relaxed); + slot->context_window_state.store(0, std::memory_order_relaxed); + slot->clearActiveBlockRun(OSThreadState::UNKNOWN); + slot->tid.store(tid, std::memory_order_release); + if (tid >= 0 && !indexSlot(reused_slot, tid)) { + slot->tid.store(-1, std::memory_order_release); + pushToFreeList(reused_slot); + return -1; + } + if (epoch != 0) { + slot->recording_epoch.store(epoch, std::memory_order_release); + } return reused_slot; } @@ -131,9 +170,124 @@ ThreadFilter::SlotID ThreadFilter::registerThread() { // Initialize the chunk if needed initializeChunk(chunk_idx); + Slot* slot = slotForId(index); + slot->lifecycle_generation.fetch_add(1, std::memory_order_acq_rel); + slot->recording_epoch.store(0, std::memory_order_relaxed); + slot->context_window_state.store(0, std::memory_order_relaxed); + slot->clearActiveBlockRun(OSThreadState::UNKNOWN); + slot->tid.store(tid, std::memory_order_release); + if (tid >= 0 && !indexSlot(index, tid)) { + slot->tid.store(-1, std::memory_order_release); + pushToFreeList(index); + return -1; + } + if (epoch != 0) { + slot->recording_epoch.store(epoch, std::memory_order_release); + } + return index; } +void ThreadFilter::refreshSlotForRecording(Slot* slot, RecordingEpoch epoch) { + if (slot == nullptr || epoch == 0 || slot->recordingEpoch() == epoch) { + return; + } + + // Make the retained identity ineligible before resetting its recording-local + // payload, then publish the new epoch only after the reset is complete. + slot->recording_epoch.store(0, std::memory_order_release); + slot->context_window_state.store(0, std::memory_order_relaxed); + slot->clearActiveBlockRun(OSThreadState::UNKNOWN); + slot->recording_epoch.store(epoch, std::memory_order_release); +} + +bool ThreadFilter::indexSlot(SlotID slot_id, int tid) { + unsigned start = hashTid(tid) & kTidIndexMask; + for (int probe = 0; probe < kTidIndexSize; ++probe) { + int index = (start + probe) & kTidIndexMask; + int value = _tid_index[index].load(std::memory_order_acquire); + if (value <= 0) { + _tid_index[index].store(slot_id + 1, std::memory_order_release); + return true; + } + if (value > 0) { + Slot* slot = slotForId(value - 1); + if (slot != nullptr && slot->nativeTid() == tid) { + return value - 1 == slot_id; + } + } + } + return false; +} + +void ThreadFilter::unindexSlot(SlotID slot_id, int tid) { + if (tid < 0) return; + unsigned start = hashTid(tid) & kTidIndexMask; + for (int probe = 0; probe < kTidIndexSize; ++probe) { + int index = (start + probe) & kTidIndexMask; + int value = _tid_index[index].load(std::memory_order_acquire); + if (value == 0) return; + if (value == slot_id + 1) { + int next = (index + 1) & kTidIndexMask; + int replacement = + _tid_index[next].load(std::memory_order_acquire) == 0 ? 0 : -1; + _tid_index[index].store(replacement, std::memory_order_release); + if (replacement == 0) { + int previous = (index - 1) & kTidIndexMask; + while (_tid_index[previous].load(std::memory_order_acquire) == -1) { + _tid_index[previous].store(0, std::memory_order_release); + previous = (previous - 1) & kTidIndexMask; + } + } + return; + } + } +} + +ThreadFilter::SlotID ThreadFilter::lookupSlotIdByTid(int tid) const { + if (tid < 0) return -1; + unsigned start = hashTid(tid) & kTidIndexMask; + for (int probe = 0; probe < kTidIndexSize; ++probe) { + int index = (start + probe) & kTidIndexMask; + int value = _tid_index[index].load(std::memory_order_acquire); + if (value == 0) return -1; + if (value > 0) { + Slot* slot = slotForId(value - 1); + if (slot != nullptr && slot->nativeTid() == tid) { + return value - 1; + } + } + } + return -1; +} + +ThreadFilter::Slot* ThreadFilter::lookupByTid(int tid) const { + SlotID slot_id = lookupSlotIdByTid(tid); + return slot_id < 0 ? nullptr : slotForId(slot_id); +} + +ThreadFilter::Slot* ThreadFilter::lookupByTid(int tid, + RecordingEpoch epoch) const { + if (epoch == 0 || recordingEpoch() != epoch) { + return nullptr; + } + Slot* slot = lookupByTid(tid); + return slot != nullptr && slot->recordingEpoch() == epoch ? slot : nullptr; +} + +ThreadFilter::Slot* ThreadFilter::activeSlotForId(SlotID slot_id, + int tid) const { + Slot* slot = slotForId(slot_id); + if (slot == nullptr || slot->nativeTid() != tid) { + return nullptr; + } + RecordingEpoch epoch = recordingEpoch(); + if (epoch != 0 && slot->recordingEpoch() != epoch) { + return nullptr; + } + return slot; +} + void ThreadFilter::initFreeList() { // Initialize the free list storage for (int i = 0; i < kFreeListSize; ++i) { @@ -160,7 +314,7 @@ bool ThreadFilter::accept(SlotID slot_id) const { // This is not a fast path like the add operation. ChunkStorage* chunk = _chunks[chunk_idx].load(std::memory_order_acquire); if (likely(chunk != nullptr)) { - return chunk->slots[slot_idx].value.load(std::memory_order_relaxed) != -1; + return chunk->slots[slot_idx].inContextWindow(); } return false; } @@ -176,7 +330,18 @@ void ThreadFilter::add(int tid, SlotID slot_id) { // Fast path: assume valid slot_id from registerThread() ChunkStorage* chunk = _chunks[chunk_idx].load(std::memory_order_acquire); if (likely(chunk != nullptr)) { - chunk->slots[slot_idx].value.store(tid, std::memory_order_release); + Slot& slot = chunk->slots[slot_idx]; + if (slot.nativeTid() == -1) { + std::lock_guard lock(_registry_lock); + if (slot.nativeTid() == -1) { + slot.tid.store(tid, std::memory_order_release); + if (!indexSlot(slot_id, tid)) { + slot.tid.store(-1, std::memory_order_release); + return; + } + } + } + slot.enterContextWindow(); } } @@ -198,16 +363,59 @@ void ThreadFilter::remove(SlotID slot_id) { return; } - chunk->slots[slot_idx].value.store(-1, std::memory_order_release); + chunk->slots[slot_idx].exitContextWindow(); +} + +void ThreadFilter::unregisterThread(SlotID slot_id, int expected_tid) { + std::lock_guard lock(_registry_lock); + unregisterThreadLocked(slot_id, expected_tid); } -void ThreadFilter::unregisterThread(SlotID slot_id) { +void ThreadFilter::unregisterThreadLocked(SlotID slot_id, int expected_tid) { if (slot_id < 0) return; - remove(slot_id); - resetSlotRunState(slot_id); + Slot* slot = slotForId(slot_id); + if (slot == nullptr) return; + int tid = slot->nativeTid(); + if (expected_tid >= 0 && tid != expected_tid) return; + unindexSlot(slot_id, tid); + slot->recording_epoch.store(0, std::memory_order_release); + slot->tid.store(-1, std::memory_order_release); + slot->context_window_state.store(0, std::memory_order_release); + slot->clearActiveBlockRun(OSThreadState::UNKNOWN); pushToFreeList(slot_id); } +void ThreadFilter::unregisterThreadByTid(int tid) { + std::lock_guard lock(_registry_lock); + SlotID slot_id = lookupSlotIdByTid(tid); + if (slot_id >= 0) { + unregisterThreadLocked(slot_id); + } +} + +int ThreadFilter::retireInactiveRegistrations() { + RecordingEpoch epoch = recordingEpoch(); + if (epoch == 0 || !unfilteredWallTrackingActive()) { + return 0; + } + + std::lock_guard lock(_registry_lock); + int retired = 0; + int num_chunks = _num_chunks.load(std::memory_order_acquire); + for (int chunk_idx = 0; chunk_idx < num_chunks; ++chunk_idx) { + ChunkStorage* chunk = _chunks[chunk_idx].load(std::memory_order_acquire); + if (chunk == nullptr) continue; + for (int slot_idx = 0; slot_idx < kChunkSize; ++slot_idx) { + Slot& slot = chunk->slots[slot_idx]; + if (slot.nativeTid() != -1 && slot.recordingEpoch() != epoch) { + unregisterThreadLocked((chunk_idx << kChunkShift) + slot_idx); + retired++; + } + } + } + return retired; +} + bool ThreadFilter::pushToFreeList(SlotID slot_id) { // Lock-free sharded Treiber stack push const int shard = shardOfSlot(slot_id); @@ -274,8 +482,8 @@ void ThreadFilter::collect(std::vector& tids) const { } for (const auto& slot : chunk->slots) { - int slot_tid = slot.value.load(std::memory_order_relaxed); - if (slot_tid != -1) { + int slot_tid = slot.nativeTid(); + if (slot_tid != -1 && slot.inContextWindow()) { tids.push_back(slot_tid); } } @@ -299,9 +507,10 @@ void ThreadFilter::collect(std::vector& entries) const { } for (auto& slot : chunk->slots) { - int slot_tid = slot.value.load(std::memory_order_acquire); - if (slot_tid != -1) { - entries.push_back({slot_tid, &slot}); + int slot_tid = slot.nativeTid(); + if (slot_tid != -1 && slot.inContextWindow()) { + entries.push_back({slot_tid, &slot, slot.lifecycleGeneration(), + slot.recordingEpoch()}); } } } @@ -316,7 +525,7 @@ void ThreadFilter::clearActive() { } for (auto& slot : chunk->slots) { - slot.value.store(-1, std::memory_order_release); + slot.exitContextWindow(); slot.clearActiveBlockRun(OSThreadState::UNKNOWN); } } @@ -339,7 +548,8 @@ u64 ThreadFilter::enterBlockedRun(SlotID slot_id, OSThreadState state, Slot* s = slotForId(slot_id); if (s != nullptr) { u32 generation = 0; - if (!s->trySetActiveBlockRun(state, owner, &generation)) { + if (!s->trySetActiveBlockRun(state, owner, &generation, + unfilteredWallTrackingActive())) { return 0; } return encodeBlockRunToken(slot_id, generation); @@ -363,14 +573,106 @@ bool ThreadFilter::exitBlockedRun(SlotID slot_id, u32 generation) { return true; } -void ThreadFilter::init(const char* filter) { - // Simple logic: any filter value (including "0") enables filtering - // Only explicitly registered threads via addThread() will be sampled - // Previously we had a syntax where we could manually force some thread IDs. - // This is no longer supported. - _enabled.store(filter != nullptr && strlen(filter) > 0, std::memory_order_release); +bool ThreadFilter::shouldSuppressOwnedBlock(const ThreadEntry& entry) const { + Slot* slot = entry.slot; + if (slot == nullptr || slot->nativeTid() != entry.tid || + slot->lifecycleGeneration() != entry.lifecycle_generation) { + return false; + } + + const bool unfiltered_tracking = unfilteredWallTrackingActive(); + RecordingEpoch epoch = 0; + if (unfiltered_tracking) { + epoch = recordingEpoch(); + if (epoch == 0 || entry.recording_epoch != epoch || + slot->recordingEpoch() != epoch) { + return false; + } + } + +#ifdef UNIT_TEST + if (_suppression_snapshot_hook != nullptr) { + _suppression_snapshot_hook(_suppression_snapshot_hook_arg); + } +#endif + + u32 block_generation = slot->blockGeneration(); + BlockRunOwner owner = slot->activeBlockOwner(); + OSThreadState state = slot->activeBlockState(); + bool context_eligible = + !unfiltered_tracking || slot->activeBlockRemainedOutsideContextWindow(); + bool sampled = slot->sampledThisRun(); + OSThreadState last_sampled_state = + sampled ? slot->lastSampledState() : OSThreadState::UNKNOWN; + bool suppressible_state = state == OSThreadState::SLEEPING || + state == OSThreadState::CONDVAR_WAIT || + state == OSThreadState::OBJECT_WAIT || + state == OSThreadState::MONITOR_WAIT; + if (owner == BlockRunOwner::NONE || !context_eligible || + !suppressible_state || !sampled || state != last_sampled_state) { + return false; + } + + // The payload is spread across independent atomics. Accept it only if the + // slot still represents the lifecycle and block run captured by the timer. + if (slot->activeBlockOwner() != owner || + slot->blockGeneration() != block_generation || + slot->nativeTid() != entry.tid || + slot->lifecycleGeneration() != entry.lifecycle_generation) { + return false; + } + if (unfiltered_tracking && + (recordingEpoch() != epoch || slot->recordingEpoch() != epoch || + !slot->activeBlockRemainedOutsideContextWindow())) { + return false; + } + return true; +} + +void ThreadFilter::init(const char* filter, bool track_unfiltered_wall) { + // Preserve the legacy filter contract: every non-empty value, including + // "0", enables context filtering. Empty filter disables filtering; the + // extra flag only retains metadata for unfiltered wall prechecks. + bool context_filter = filter != nullptr && strlen(filter) > 0; + bool unfiltered_tracking = track_unfiltered_wall && !context_filter; + RecordingEpoch epoch = 0; + if (unfiltered_tracking) { + epoch = _next_recording_epoch.fetch_add(1, std::memory_order_acq_rel) + 1; + if (epoch == 0) { + // Zero is reserved for inactive state. This can occur only after + // 2^64 recording starts; skip it rather than publishing ambiguity. + epoch = _next_recording_epoch.fetch_add(1, std::memory_order_acq_rel) + 1; + } + } + _recording_epoch.store(epoch, std::memory_order_release); + _track_unfiltered_wall.store(unfiltered_tracking, + std::memory_order_release); + _registry_active.store(unfiltered_tracking || context_filter, + std::memory_order_release); + _enabled.store(context_filter, std::memory_order_release); } bool ThreadFilter::enabled() const { return _enabled.load(std::memory_order_acquire); } + +bool ThreadFilter::registryActive() const { + return _registry_active.load(std::memory_order_acquire); +} + +bool ThreadFilter::unfilteredWallTrackingActive() const { + return _track_unfiltered_wall.load(std::memory_order_acquire); +} + +ThreadFilter::RecordingEpoch ThreadFilter::recordingEpoch() const { + return _recording_epoch.load(std::memory_order_acquire); +} + +void ThreadFilter::deactivateRecording() { + // Close producer admission before invalidating the recording epoch. Existing + // slots remain allocated so surviving threads can refresh them on restart. + _registry_active.store(false, std::memory_order_release); + _enabled.store(false, std::memory_order_release); + _track_unfiltered_wall.store(false, std::memory_order_release); + _recording_epoch.store(0, std::memory_order_release); +} diff --git a/ddprof-lib/src/main/cpp/threadFilter.h b/ddprof-lib/src/main/cpp/threadFilter.h index 541249e4c1..2a73f37666 100644 --- a/ddprof-lib/src/main/cpp/threadFilter.h +++ b/ddprof-lib/src/main/cpp/threadFilter.h @@ -21,6 +21,7 @@ #include #include #include +#include #include "arch.h" #include "threadState.h" @@ -37,6 +38,7 @@ enum class BlockRunOwner : int { class ThreadFilter { public: using SlotID = int; + using RecordingEpoch = u64; // Optimized limits for reasonable memory usage static constexpr int kChunkSize = 256; @@ -45,8 +47,10 @@ class ThreadFilter { static constexpr int kMaxThreads = 2048; static constexpr int kMaxChunks = (kMaxThreads + kChunkSize - 1) / kChunkSize; // = 8 chunks // High-performance free list using Treiber stack, 64 shards - static constexpr int kFreeListSize = 1024; // power-of-two for fast modulo + static constexpr int kFreeListSize = kMaxThreads; static constexpr int kShardCount = 64; // power-of-two for fast modulo + static constexpr int kTidIndexSize = 8192; // 4x maximum live slots + static constexpr int kTidIndexMask = kTidIndexSize - 1; // One cache line per slot to avoid false sharing. Slot instances are never freed // (ChunkStorage is process-lifetime), so a captured Slot* is always dereferenceable. @@ -56,8 +60,21 @@ class ThreadFilter { std::atomic unowned_blocked_pending_weight{0}; std::atomic unowned_blocked_decision_count{0}; std::atomic unowned_blocked_call_trace_id{0}; + // Packed as (epoch << 1) | in_context_window so a transition and its + // epoch change are observed atomically by block admission and exit. + std::atomic context_window_state{0}; + std::atomic lifecycle_generation{0}; + // Per-recording publication flag. A retained TID mapping is eligible for + // unfiltered suppression only when this value matches the registry's + // active recording epoch. The payload is reset before the epoch is + // release-published. + std::atomic recording_epoch{0}; + std::atomic active_block_context_epoch{0}; std::atomic unowned_blocked_state{OSThreadState::UNKNOWN}; - std::atomic value{-1}; + // Native identity and context-window membership are independent so an + // unfiltered wall recording can retain lifecycle metadata without + // changing ordinary thread selection. + std::atomic tid{-1}; std::atomic active_block_owner{static_cast(BlockRunOwner::NONE)}; std::atomic block_generation{0}; // Wall-clock once-per-run suppression state. The signal handler records the @@ -71,6 +88,10 @@ class ThreadFilter { std::atomic active_block_state{OSThreadState::UNKNOWN}; std::atomic sampled_this_run{false}; char padding[2 * DEFAULT_CACHE_LINE_SIZE + - sizeof(std::atomic) + - sizeof(std::atomic) + - sizeof(std::atomic) + - sizeof(std::atomic) - sizeof(std::atomic) - sizeof(std::atomic) - sizeof(std::atomic) @@ -82,6 +103,44 @@ class ThreadFilter { - sizeof(std::atomic) - sizeof(std::atomic)]; + inline int nativeTid() const { + return tid.load(std::memory_order_acquire); + } + inline u64 lifecycleGeneration() const { + return lifecycle_generation.load(std::memory_order_acquire); + } + inline RecordingEpoch recordingEpoch() const { + return recording_epoch.load(std::memory_order_acquire); + } + inline bool inContextWindow() const { + return (context_window_state.load(std::memory_order_acquire) & 1) != 0; + } + inline u64 contextWindowEpoch() const { + return context_window_state.load(std::memory_order_acquire) >> 1; + } + inline bool enterContextWindow() { + u64 current = context_window_state.load(std::memory_order_acquire); + while ((current & 1) == 0) { + if (context_window_state.compare_exchange_weak( + current, current + 3, std::memory_order_acq_rel, + std::memory_order_acquire)) { + return true; + } + } + return false; + } + inline bool exitContextWindow() { + u64 current = context_window_state.load(std::memory_order_acquire); + while ((current & 1) != 0) { + if (context_window_state.compare_exchange_weak( + current, current + 1, std::memory_order_acq_rel, + std::memory_order_acquire)) { + return true; + } + } + return false; + } + inline bool sampledThisRun() const { return sampled_this_run.load(std::memory_order_acquire); } @@ -147,14 +206,26 @@ class ThreadFilter { return true; } inline bool trySetActiveBlockRun(OSThreadState state, BlockRunOwner owner, - u32* generation_out) { + u32* generation_out, + bool outside_context_required) { + u64 context_state = context_window_state.load(std::memory_order_acquire); + if (outside_context_required && (context_state & 1) != 0) { + return false; + } int expected_owner = static_cast(BlockRunOwner::NONE); if (!active_block_owner.compare_exchange_strong( expected_owner, static_cast(owner), std::memory_order_acq_rel, std::memory_order_acquire)) { return false; } + if (outside_context_required && + context_window_state.load(std::memory_order_acquire) != context_state) { + active_block_owner.store(static_cast(BlockRunOwner::NONE), + std::memory_order_release); + return false; + } u32 generation = block_generation.fetch_add(1, std::memory_order_acq_rel) + 1; + active_block_context_epoch.store(context_state >> 1, std::memory_order_relaxed); resetUnownedBlockedSampling(); last_sampled_state.store(OSThreadState::UNKNOWN, std::memory_order_relaxed); sampled_this_run.store(false, std::memory_order_relaxed); @@ -167,19 +238,30 @@ class ThreadFilter { resetSampledRun(state); active_block_owner.store(static_cast(BlockRunOwner::NONE), std::memory_order_release); } + inline bool activeBlockRemainedOutsideContextWindow() const { + u64 context_state = context_window_state.load(std::memory_order_acquire); + return (context_state & 1) == 0 && + active_block_context_epoch.load(std::memory_order_acquire) == + (context_state >> 1); + } }; static_assert(sizeof(Slot) == 2 * DEFAULT_CACHE_LINE_SIZE, "Slot must be exactly two cache lines"); static_assert(std::atomic::is_always_lock_free, "Slot OSThreadState fields must be lock-free for signal-handler safety"); static_assert(std::atomic::is_always_lock_free, "Slot::sampled_this_run must be lock-free for signal-handler safety"); + static_assert(std::atomic::is_always_lock_free, + "Slot::recording_epoch must be lock-free for signal-handler safety"); ThreadFilter(); ~ThreadFilter(); - void init(const char* filter); + void init(const char* filter, bool track_unfiltered_wall = false); void initFreeList(); bool enabled() const; + bool registryActive() const; + bool unfilteredWallTrackingActive() const; + RecordingEpoch recordingEpoch() const; // Hot path methods - slot_id MUST be from registerThread(), undefined behavior otherwise bool accept(SlotID slot_id) const; void add(int tid, SlotID slot_id); @@ -197,6 +279,18 @@ class ThreadFilter { // another owner. void exitBlockedRun(SlotID slot_id); bool exitBlockedRun(SlotID slot_id, u32 generation); + // Reads the complete timer-side suppression payload and rejects it if slot + // identity or block lifecycle changes before final validation. + bool shouldSuppressOwnedBlock(const ThreadEntry& entry) const; + +#ifdef UNIT_TEST + using SuppressionSnapshotHook = void (*)(void*); + void setSuppressionSnapshotHookForTest(SuppressionSnapshotHook hook, + void* arg) { + _suppression_snapshot_hook = hook; + _suppression_snapshot_hook_arg = arg; + } +#endif static inline u64 encodeBlockRunToken(SlotID slot_id, u32 generation) { return (static_cast(generation) << 32) | static_cast(slot_id + 1); @@ -218,8 +312,14 @@ class ThreadFilter { return chunk != nullptr ? &chunk->slots[slot_idx] : nullptr; } - SlotID registerThread(); - void unregisterThread(SlotID slot_id); + SlotID registerThread(int tid = -1); + void unregisterThread(SlotID slot_id, int expected_tid = -1); + void unregisterThreadByTid(int tid); + Slot* lookupByTid(int tid) const; + Slot* lookupByTid(int tid, RecordingEpoch epoch) const; + Slot* activeSlotForId(SlotID slot_id, int tid) const; + int retireInactiveRegistrations(); + void deactivateRecording(); private: @@ -235,6 +335,10 @@ class ThreadFilter { }; std::atomic _enabled{false}; + std::atomic _registry_active{false}; + std::atomic _track_unfiltered_wall{false}; + std::atomic _recording_epoch{0}; + std::atomic _next_recording_epoch{0}; // Lazily allocated storage for chunks std::atomic _chunks[kMaxChunks]; @@ -243,6 +347,17 @@ class ThreadFilter { // Lock-free slot allocation std::atomic _next_index{0}; std::unique_ptr _free_list; + // Entries contain slot_id + 1. Zero terminates a lookup probe; -1 is a + // tombstone left by unregister. The slot's published TID is the key. + std::array, kTidIndexSize> _tid_index; + // Registration and teardown never run in a signal handler. Serializing + // writers prevents duplicate TID mappings while lookups remain lock-free. + std::mutex _registry_lock; + +#ifdef UNIT_TEST + SuppressionSnapshotHook _suppression_snapshot_hook = nullptr; + void* _suppression_snapshot_hook_arg = nullptr; +#endif // Cache line aligned to prevent false sharing between shards struct alignas(DEFAULT_CACHE_LINE_SIZE) ShardHead { std::atomic head{-1}; }; @@ -254,12 +369,22 @@ class ThreadFilter { void initializeChunk(int chunk_idx); bool pushToFreeList(SlotID slot_id); SlotID popFromFreeList(); + bool indexSlot(SlotID slot_id, int tid); + void unindexSlot(SlotID slot_id, int tid); + void refreshSlotForRecording(Slot* slot, RecordingEpoch epoch); + void unregisterThreadLocked(SlotID slot_id, int expected_tid = -1); + SlotID lookupSlotIdByTid(int tid) const; + static inline unsigned hashTid(int tid) { + return static_cast(tid) * 2654435761u; + } }; // Snapshot entry produced by ThreadFilter::collect for the wall-clock timer. struct ThreadEntry { int tid; ThreadFilter::Slot* slot; + u64 lifecycle_generation; + ThreadFilter::RecordingEpoch recording_epoch; }; #endif // _THREADFILTER_H diff --git a/ddprof-lib/src/main/cpp/wallClock.cpp b/ddprof-lib/src/main/cpp/wallClock.cpp index eac9f3fd37..562c5181da 100644 --- a/ddprof-lib/src/main/cpp/wallClock.cpp +++ b/ddprof-lib/src/main/cpp/wallClock.cpp @@ -76,19 +76,13 @@ static inline void incrementSuppressedSampledRun() { WallClockCounters::incrementSuppressedSampledRun(); } -static inline bool suppressAlreadySampledBlock(ThreadFilter::Slot* slot) { - if (slot == nullptr) { +static inline bool suppressAlreadySampledBlock(const ThreadEntry& entry) { + ThreadFilter* thread_filter = Profiler::instance()->threadFilter(); + if (!thread_filter->shouldSuppressOwnedBlock(entry)) { return false; } - OSThreadState block_state = slot->activeBlockState(); - if (slot->activeBlockOwner() != BlockRunOwner::NONE && - isPrecheckSuppressionState(block_state) && - slot->sampledThisRun() && - block_state == slot->lastSampledState()) { - incrementSuppressedSampledRun(); - return true; - } - return false; + incrementSuppressedSampledRun(); + return true; } static inline WallPrecheckResult prepareWallPrecheck(ProfiledThread* current, @@ -98,9 +92,22 @@ static inline WallPrecheckResult prepareWallPrecheck(ProfiledThread* current, return result; } + ThreadFilter* registry = Profiler::instance()->threadFilter(); ThreadFilter::Slot* slot = - Profiler::instance()->threadFilter()->slotForId(current->filterSlotId()); + registry->activeSlotForId(current->filterSlotId(), current->tid()); if (slot == nullptr) { + ThreadFilter::RecordingEpoch epoch = registry->recordingEpoch(); + slot = epoch != 0 ? registry->lookupByTid(current->tid(), epoch) + : registry->lookupByTid(current->tid()); + } + if (slot == nullptr) { + return result; + } + + // In an unfiltered recording, context threads keep their normal MethodSample + // stream. Only owned blocks that remain outside the context window may replace + // repeated signals. + if (registry->unfilteredWallTrackingActive() && slot->inContextWindow()) { return result; } @@ -108,7 +115,9 @@ static inline WallPrecheckResult prepareWallPrecheck(ProfiledThread* current, BlockRunOwner active_block_owner = slot->activeBlockOwner(); bool has_owned_block = active_block_owner != BlockRunOwner::NONE && - isPrecheckSuppressionState(active_block_state); + isPrecheckSuppressionState(active_block_state) && + (!registry->unfilteredWallTrackingActive() || + slot->activeBlockRemainedOutsideContextWindow()); if (has_owned_block) { if (slot->sampledThisRun() && active_block_state == slot->lastSampledState()) { @@ -311,7 +320,6 @@ Error BaseWallClock::start(Arguments &args) { _reservoir_size = args._wall_threads_per_tick ? args._wall_threads_per_tick : DEFAULT_WALL_THREADS_PER_TICK; - initialize(args); _running = true; @@ -349,11 +357,15 @@ void WallClockASGCT::initialize(Arguments& args) { } void WallClockASGCT::timerLoop() { - // todo: re-allocating the vector every time is not efficient + ThreadFilter* thread_filter = Profiler::instance()->threadFilter(); + const bool lazy_backfill = + _precheck && thread_filter->unfilteredWallTrackingActive(); + const ThreadFilter::RecordingEpoch recording_epoch = + lazy_backfill ? thread_filter->recordingEpoch() : 0; auto collectThreads = [&](std::vector& entries) { // Get thread IDs from the filter if it's enabled // Otherwise list all threads in the system - if (Profiler::instance()->threadFilter()->enabled()) { + if (thread_filter->enabled()) { Profiler::instance()->threadFilter()->collect(entries); } else { const int refresher_tid = Libraries::instance()->refresherTid(); @@ -365,21 +377,37 @@ void WallClockASGCT::timerLoop() { // enough; we also want to avoid the kill() round-trip and any // pending-signal accumulation). if (tid != OS::threadId() && tid != refresher_tid) { - entries.push_back({tid, nullptr}); // no-filter: precheck fast path is skipped (null guards) + entries.push_back({tid, nullptr, 0, 0}); } } delete thread_list; } + if (_precheck && !lazy_backfill) { + entries.erase(std::remove_if(entries.begin(), entries.end(), + suppressAlreadySampledBlock), + entries.end()); + } }; - auto sampleThreads = [&](ThreadEntry entry, int& num_failures, int& threads_already_exited, - int& permission_denied) { + auto sampleThreads = [&](ThreadEntry entry, int& num_failures, + int& threads_already_exited, int& permission_denied, + int& registry_lookups, bool lookup_registry_slot) { + if (lookup_registry_slot && entry.slot == nullptr) { + registry_lookups++; + ThreadFilter::Slot* slot = + thread_filter->lookupByTid(entry.tid, recording_epoch); + if (slot != nullptr) { + entry.slot = slot; + entry.lifecycle_generation = slot->lifecycleGeneration(); + entry.recording_epoch = slot->recordingEpoch(); + } + } // Timer-thread fast path (wallprecheck=true): skip the kernel IPI entirely // only when an explicit lifecycle hook still owns an already-sampled blocked // run. Raw OS thread state is intentionally not used here because the timer // thread cannot prove run boundaries for the target thread. - if (_precheck && suppressAlreadySampledBlock(entry.slot)) { - return false; + if (_precheck && suppressAlreadySampledBlock(entry)) { + return WallClockCandidateOutcome::PRECHECK_REJECTED; } if (!OS::sendSignalWithCookie(entry.tid, SIGVTALRM, SignalCookie::wallclock())) { num_failures++; @@ -395,15 +423,16 @@ void WallClockASGCT::timerLoop() { Log::debug("unexpected error %s", strerror(errno)); } } - return false; + return WallClockCandidateOutcome::SIGNAL_FAILED; } - return true; + return WallClockCandidateOutcome::SIGNAL_SENT; }; auto doNothing = []() { }; - timerLoopCommon(collectThreads, sampleThreads, doNothing, _reservoir_size, _interval); + timerLoopCommon(collectThreads, sampleThreads, doNothing, + _reservoir_size, _interval, lazy_backfill); } // WallClockJvmti: mirrors WallClockASGCT's dispatch, but the signal handler @@ -497,9 +526,14 @@ void WallClockJvmti::initialize(Arguments &args) { } void WallClockJvmti::timerLoop() { + ThreadFilter* thread_filter = Profiler::instance()->threadFilter(); + const bool lazy_backfill = + _precheck && thread_filter->unfilteredWallTrackingActive(); + const ThreadFilter::RecordingEpoch recording_epoch = + lazy_backfill ? thread_filter->recordingEpoch() : 0; auto collectThreads = [&](std::vector &entries) { const int refresher_tid = Libraries::instance()->refresherTid(); - if (Profiler::instance()->threadFilter()->enabled()) { + if (thread_filter->enabled()) { Profiler::instance()->threadFilter()->collect(entries); } else { ThreadList *thread_list = OS::listThreads(); @@ -508,17 +542,33 @@ void WallClockJvmti::timerLoop() { // Exclude the wallclock timer thread itself and the Libraries // refresher (profiler-internal). if (tid != OS::threadId() && tid != refresher_tid) { - entries.push_back({tid, nullptr}); + entries.push_back({tid, nullptr, 0, 0}); } } delete thread_list; } + if (_precheck && !lazy_backfill) { + entries.erase(std::remove_if(entries.begin(), entries.end(), + suppressAlreadySampledBlock), + entries.end()); + } }; auto sampleThreads = [&](ThreadEntry entry, int &num_failures, - int &threads_already_exited, int &permission_denied) { - if (_precheck && suppressAlreadySampledBlock(entry.slot)) { - return false; + int &threads_already_exited, int &permission_denied, + int ®istry_lookups, bool lookup_registry_slot) { + if (lookup_registry_slot && entry.slot == nullptr) { + registry_lookups++; + ThreadFilter::Slot* slot = + thread_filter->lookupByTid(entry.tid, recording_epoch); + if (slot != nullptr) { + entry.slot = slot; + entry.lifecycle_generation = slot->lifecycleGeneration(); + entry.recording_epoch = slot->recordingEpoch(); + } + } + if (_precheck && suppressAlreadySampledBlock(entry)) { + return WallClockCandidateOutcome::PRECHECK_REJECTED; } if (!OS::sendSignalWithCookie(entry.tid, SIGVTALRM, SignalCookie::wallclock())) { num_failures++; @@ -534,13 +584,13 @@ void WallClockJvmti::timerLoop() { Log::debug("unexpected error %s", strerror(errno)); } } - return false; + return WallClockCandidateOutcome::SIGNAL_FAILED; } - return true; + return WallClockCandidateOutcome::SIGNAL_SENT; }; auto doNothing = []() {}; timerLoopCommon(collectThreads, sampleThreads, doNothing, - _reservoir_size, _interval); + _reservoir_size, _interval, lazy_backfill); } diff --git a/ddprof-lib/src/main/cpp/wallClock.h b/ddprof-lib/src/main/cpp/wallClock.h index 14e3f88aa3..8bd9ef8c66 100644 --- a/ddprof-lib/src/main/cpp/wallClock.h +++ b/ddprof-lib/src/main/cpp/wallClock.h @@ -16,6 +16,7 @@ #include "threadFilter.h" #include "threadState.h" #include "tsc.h" +#include "wallClockCandidateSelector.h" #include "wallClockCounters.h" class BaseWallClock : public Engine { @@ -23,6 +24,9 @@ class BaseWallClock : public Engine { static std::atomic _enabled; std::atomic _running; protected: + // Backfill rejected candidates without letting a population of already- + // suppressed blockers restore O(N) registry lookups on every wall tick. + static constexpr size_t PRECHECK_VISIT_BUDGET_MULTIPLIER = 4; long _interval; // Maximum number of threads sampled in one iteration. This limit serves as a // throttle when generating profiling signals. Otherwise applications with too @@ -30,7 +34,6 @@ class BaseWallClock : public Engine { // limit low enough helps to avoid contention on a spin lock inside // Profiler::recordSample(). int _reservoir_size; - pthread_t _thread; virtual void timerLoop() = 0; virtual void initialize(Arguments& args) {}; @@ -44,7 +47,9 @@ class BaseWallClock : public Engine { static bool inSyscall(void* ucontext); template - void timerLoopCommon(CollectThreadsFunc collectThreads, SampleThreadsFunc sampleThreads, CleanThreadFunc cleanThreads, int reservoirSize, u64 interval) { + void timerLoopCommon(CollectThreadsFunc collectThreads, SampleThreadsFunc sampleThreads, + CleanThreadFunc cleanThreads, int reservoirSize, u64 interval, + bool lazyBackfill = false) { if (!_enabled.load(std::memory_order_acquire)) { return; } @@ -55,6 +60,13 @@ class BaseWallClock : public Engine { std::random_device rd; std::mt19937 generator(rd()); std::normal_distribution distribution(interval, stddev); + std::mt19937 candidate_generator; + if (lazyBackfill) { + std::random_device candidate_rd; + std::seed_seq candidate_seed{candidate_rd(), candidate_rd(), + candidate_rd(), candidate_rd()}; + candidate_generator.seed(candidate_seed); + } std::vector threads; threads.reserve(reservoirSize); @@ -83,12 +95,44 @@ class BaseWallClock : public Engine { int num_failures = 0; int threads_already_exited = 0; int permission_denied = 0; + int registry_lookups = 0; u32 num_successful_samples = 0; - std::vector sample = reservoir.sample(threads); - for (ThreadType thread : sample) { - if (sampleThreads(thread, num_failures, threads_already_exited, permission_denied)) { - num_successful_samples++; + if (lazyBackfill) { + WallClockCandidateStats stats = selectWallClockCandidates( + threads, + static_cast(reservoirSize), + static_cast(reservoirSize) * + PRECHECK_VISIT_BUDGET_MULTIPLIER, + candidate_generator, + [&](ThreadType thread) { + WallClockCandidateOutcome outcome = sampleThreads( + thread, num_failures, threads_already_exited, + permission_denied, registry_lookups, true); + if (outcome == WallClockCandidateOutcome::SIGNAL_SENT) { + num_successful_samples++; + } + return outcome; + }); + if (stats.precheck_rejected > 0) { + Counters::increment(WC_PRECHECK_CANDIDATES_REJECTED, + stats.precheck_rejected); + } + if (stats.visit_limit_reached) { + Counters::increment(WC_PRECHECK_LOOKUP_BUDGET_EXHAUSTED); } + } else { + std::vector sample = reservoir.sample(threads); + for (ThreadType thread : sample) { + WallClockCandidateOutcome outcome = sampleThreads( + thread, num_failures, threads_already_exited, + permission_denied, registry_lookups, false); + if (outcome == WallClockCandidateOutcome::SIGNAL_SENT) { + num_successful_samples++; + } + } + } + if (registry_lookups > 0) { + Counters::increment(WC_PRECHECK_REGISTRY_LOOKUPS, registry_lookups); } epoch.updateNumSamplableThreads(threads.size()); @@ -159,6 +203,7 @@ class WallClockASGCT : public BaseWallClock { const char* name() override { return "WallClock (ASGCT)"; } + bool supportsUnfilteredWallPrecheck() const override { return true; } }; // Wall-clock engine that uses BaseWallClock's pthread reservoir sampling loop @@ -180,6 +225,7 @@ class WallClockJvmti : public BaseWallClock { const char* name() override { return "WallClock (JVMTI)"; } + bool supportsUnfilteredWallPrecheck() const override { return true; } }; #endif // _WALLCLOCK_H diff --git a/ddprof-lib/src/main/cpp/wallClockCandidateSelector.h b/ddprof-lib/src/main/cpp/wallClockCandidateSelector.h new file mode 100644 index 0000000000..8a8358e79b --- /dev/null +++ b/ddprof-lib/src/main/cpp/wallClockCandidateSelector.h @@ -0,0 +1,64 @@ +/* + * Copyright 2026 Datadog, Inc + * SPDX-License-Identifier: Apache-2.0 + */ + +#ifndef WALL_CLOCK_CANDIDATE_SELECTOR_H +#define WALL_CLOCK_CANDIDATE_SELECTOR_H + +#include +#include +#include +#include +#include + +enum class WallClockCandidateOutcome { + SIGNAL_SENT, + SIGNAL_FAILED, + PRECHECK_REJECTED, +}; + +struct WallClockCandidateStats { + size_t visited = 0; + size_t slots_consumed = 0; + size_t precheck_rejected = 0; + bool visit_limit_reached = false; +}; + +// Visits a uniformly randomized prefix without replacement. A precheck rejection +// is the only outcome that does not consume target capacity. This runs only on +// the wall-clock timer thread. +template +WallClockCandidateStats selectWallClockCandidates(std::vector& candidates, + size_t target_size, + size_t visit_limit, + URBG& generator, + Visitor&& visitor) { + WallClockCandidateStats stats; + if (target_size == 0 || candidates.empty() || visit_limit == 0) { + return stats; + } + + size_t max_visits = std::min(candidates.size(), visit_limit); + for (size_t i = 0; i < max_visits && stats.slots_consumed < target_size; ++i) { + std::uniform_int_distribution next(i, candidates.size() - 1); + size_t selected = next(generator); + if (selected != i) { + std::swap(candidates[i], candidates[selected]); + } + + stats.visited++; + WallClockCandidateOutcome outcome = visitor(candidates[i]); + if (outcome == WallClockCandidateOutcome::PRECHECK_REJECTED) { + stats.precheck_rejected++; + } else { + stats.slots_consumed++; + } + } + stats.visit_limit_reached = + stats.slots_consumed < target_size && + stats.visited == max_visits && max_visits < candidates.size(); + return stats; +} + +#endif // WALL_CLOCK_CANDIDATE_SELECTOR_H diff --git a/ddprof-lib/src/test/cpp/threadFilter_ut.cpp b/ddprof-lib/src/test/cpp/threadFilter_ut.cpp index 4608e11697..cb3755ff72 100644 --- a/ddprof-lib/src/test/cpp/threadFilter_ut.cpp +++ b/ddprof-lib/src/test/cpp/threadFilter_ut.cpp @@ -1,5 +1,5 @@ /* - * Copyright 2025 Datadog, Inc + * Copyright 2025, 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. @@ -185,7 +185,9 @@ TEST_F(ThreadFilterTest, RecoveryAfterMaxCapacity) { int slot_id = filter->registerThread(); EXPECT_GE(slot_id, 0) << "Failed to register slot " << i << " after freeing"; new_slot_ids.push_back(slot_id); - filter->add(i + 3000, slot_id); + // Keep replacement identities disjoint from the still-live 3024..4047 + // range. The registry intentionally rejects two slots for one native TID. + filter->add(i + 5000, slot_id); } // Verify we can still register up to capacity @@ -552,3 +554,399 @@ TEST_F(ThreadFilterTest, TokenRoundTripPreservesHighGenerationBit) { EXPECT_EQ(slot_id, ThreadFilter::tokenSlotId(static_cast(java_token))); EXPECT_EQ(generation, ThreadFilter::tokenGeneration(static_cast(java_token))); } + +class ThreadRegistryTest : public ::testing::Test { +protected: + void SetUp() override { + registry.init("", true); + } + + ThreadFilter registry; +}; + +TEST_F(ThreadRegistryTest, UnfilteredTrackingSeparatesRegistrationFromContextWindow) { + + EXPECT_TRUE(registry.registryActive()); + EXPECT_TRUE(registry.unfilteredWallTrackingActive()); + EXPECT_FALSE(registry.enabled()); + + int slot_id = registry.registerThread(1234); + ASSERT_GE(slot_id, 0); + ThreadFilter::Slot* slot = registry.slotForId(slot_id); + ASSERT_NE(nullptr, slot); + EXPECT_EQ(1234, slot->nativeTid()); + EXPECT_FALSE(slot->inContextWindow()); + EXPECT_EQ(slot, registry.lookupByTid(1234)); + + std::vector context; + registry.collect(context); + EXPECT_TRUE(context.empty()); + + registry.add(1234, slot_id); + registry.collect(context); + ASSERT_EQ(1u, context.size()); + EXPECT_EQ(1234, context[0].tid); + + registry.remove(slot_id); + EXPECT_FALSE(slot->inContextWindow()); + EXPECT_EQ(slot, registry.lookupByTid(1234)); + registry.collect(context); + EXPECT_TRUE(context.empty()); +} + +TEST_F(ThreadRegistryTest, RegisteringKnownTidReturnsExistingSlotWithoutMutation) { + constexpr int tid = 4321; + int slot_id = registry.registerThread(tid); + ASSERT_GE(slot_id, 0); + ThreadFilter::Slot* slot = registry.slotForId(slot_id); + ASSERT_NE(nullptr, slot); + + u64 token = registry.enterBlockedRun(slot_id, OSThreadState::SLEEPING); + ASSERT_NE(0ULL, token); + slot->markSampledThisRun(OSThreadState::SLEEPING); + + u64 lifecycle_generation = slot->lifecycleGeneration(); + EXPECT_EQ(slot_id, registry.registerThread(tid)); + EXPECT_EQ(slot, registry.lookupByTid(tid)); + EXPECT_EQ(lifecycle_generation, slot->lifecycleGeneration()); + EXPECT_EQ(OSThreadState::SLEEPING, slot->activeBlockState()); + EXPECT_TRUE(slot->sampledThisRun()); + EXPECT_TRUE(registry.exitBlockedRun( + slot_id, ThreadFilter::tokenGeneration(token))); +} + +TEST_F(ThreadRegistryTest, ConcurrentSameTidRegistrationConvergesOnOneSlot) { + constexpr int thread_count = 32; + constexpr int tid = 8765; + std::atomic ready{0}; + std::atomic start{false}; + std::vector slots(thread_count, -1); + std::vector threads; + threads.reserve(thread_count); + + for (int i = 0; i < thread_count; ++i) { + threads.emplace_back([&, i] { + ready.fetch_add(1, std::memory_order_release); + while (!start.load(std::memory_order_acquire)) { + std::this_thread::yield(); + } + slots[i] = registry.registerThread(tid); + }); + } + + while (ready.load(std::memory_order_acquire) != thread_count) { + std::this_thread::yield(); + } + start.store(true, std::memory_order_release); + for (std::thread& thread : threads) { + thread.join(); + } + + ASSERT_GE(slots[0], 0); + for (int slot_id : slots) { + EXPECT_EQ(slots[0], slot_id); + } + ThreadFilter::Slot* slot = registry.slotForId(slots[0]); + ASSERT_NE(nullptr, slot); + EXPECT_EQ(tid, slot->nativeTid()); + EXPECT_EQ(slot, registry.lookupByTid(tid)); +} + +TEST_F(ThreadRegistryTest, ContextWindowTransitionsAreIdempotent) { + int slot_id = registry.registerThread(5678); + ASSERT_GE(slot_id, 0); + ThreadFilter::Slot* slot = registry.slotForId(slot_id); + ASSERT_NE(nullptr, slot); + + u64 initial_epoch = slot->contextWindowEpoch(); + registry.add(5678, slot_id); + EXPECT_TRUE(slot->inContextWindow()); + EXPECT_EQ(initial_epoch + 1, slot->contextWindowEpoch()); + + registry.add(5678, slot_id); + EXPECT_EQ(initial_epoch + 1, slot->contextWindowEpoch()); + + registry.remove(slot_id); + EXPECT_FALSE(slot->inContextWindow()); + EXPECT_EQ(initial_epoch + 2, slot->contextWindowEpoch()); + + registry.remove(slot_id); + EXPECT_EQ(initial_epoch + 2, slot->contextWindowEpoch()); +} + +TEST_F(ThreadRegistryTest, SlotReuseChangesLifecycleGenerationAndTidMapping) { + int slot_id = registry.registerThread(1111); + ASSERT_GE(slot_id, 0); + ThreadFilter::Slot* slot = registry.slotForId(slot_id); + ASSERT_NE(nullptr, slot); + u64 first_generation = slot->lifecycleGeneration(); + + registry.unregisterThread(slot_id); + EXPECT_EQ(nullptr, registry.lookupByTid(1111)); + + int reused_id = registry.registerThread(2222); + ASSERT_EQ(slot_id, reused_id); + EXPECT_GT(slot->lifecycleGeneration(), first_generation); + EXPECT_EQ(nullptr, registry.lookupByTid(1111)); + EXPECT_EQ(slot, registry.lookupByTid(2222)); +} + +TEST_F(ThreadRegistryTest, ContextTransitionInvalidatesOwnedRunSuppression) { + int slot_id = registry.registerThread(3333); + ASSERT_GE(slot_id, 0); + ThreadFilter::Slot* slot = registry.slotForId(slot_id); + ASSERT_NE(nullptr, slot); + + u64 token = registry.enterBlockedRun(slot_id, OSThreadState::SLEEPING); + ASSERT_NE(0u, token); + slot->markSampledThisRun(OSThreadState::SLEEPING); + EXPECT_TRUE(slot->activeBlockRemainedOutsideContextWindow()); + + registry.add(3333, slot_id); + registry.remove(slot_id); + EXPECT_FALSE(slot->activeBlockRemainedOutsideContextWindow()); + + ThreadEntry entry{3333, slot, slot->lifecycleGeneration(), + slot->recordingEpoch()}; + EXPECT_FALSE(registry.shouldSuppressOwnedBlock(entry)); +} + +TEST_F(ThreadRegistryTest, UnfilteredSuppressionValidatesIdentityAndLifecycle) { + int slot_id = registry.registerThread(4444); + ASSERT_GE(slot_id, 0); + ThreadFilter::Slot* slot = registry.slotForId(slot_id); + ASSERT_NE(nullptr, slot); + + u64 token = registry.enterBlockedRun(slot_id, OSThreadState::SLEEPING); + ASSERT_NE(0u, token); + slot->markSampledThisRun(OSThreadState::SLEEPING); + ThreadEntry entry{4444, slot, slot->lifecycleGeneration(), + slot->recordingEpoch()}; + EXPECT_TRUE(registry.shouldSuppressOwnedBlock(entry)); + + ThreadEntry wrong_tid{4445, slot, entry.lifecycle_generation, + entry.recording_epoch}; + EXPECT_FALSE(registry.shouldSuppressOwnedBlock(wrong_tid)); + ThreadEntry stale_generation{4444, slot, entry.lifecycle_generation + 1, + entry.recording_epoch}; + EXPECT_FALSE(registry.shouldSuppressOwnedBlock(stale_generation)); + + EXPECT_TRUE(registry.exitBlockedRun( + slot_id, ThreadFilter::tokenGeneration(token))); + EXPECT_FALSE(registry.shouldSuppressOwnedBlock(entry)); +} + +TEST_F(ThreadRegistryTest, ContextFilteredSuppressionPreservesHistoricalEligibility) { + registry.init("0"); + int slot_id = registry.registerThread(5555); + ASSERT_GE(slot_id, 0); + registry.add(5555, slot_id); + ThreadFilter::Slot* slot = registry.slotForId(slot_id); + ASSERT_NE(nullptr, slot); + + u64 token = registry.enterBlockedRun(slot_id, OSThreadState::SLEEPING); + ASSERT_NE(0u, token); + slot->markSampledThisRun(OSThreadState::SLEEPING); + ThreadEntry entry{5555, slot, slot->lifecycleGeneration(), + slot->recordingEpoch()}; + EXPECT_TRUE(registry.shouldSuppressOwnedBlock(entry)); +} + +TEST_F(ThreadRegistryTest, ConcurrentTidReuseInvalidatesSuppressionSnapshot) { + constexpr int tid = 5601; + int slot_id = registry.registerThread(tid); + ASSERT_GE(slot_id, 0); + ThreadFilter::Slot* slot = registry.slotForId(slot_id); + ASSERT_NE(nullptr, slot); + ASSERT_NE(0u, registry.enterBlockedRun(slot_id, OSThreadState::SLEEPING)); + slot->markSampledThisRun(OSThreadState::SLEEPING); + ThreadEntry stale{tid, slot, slot->lifecycleGeneration(), + slot->recordingEpoch()}; + + struct SnapshotPause { + std::atomic reached{false}; + std::atomic resume{false}; + } pause; + registry.setSuppressionSnapshotHookForTest( + [](void* raw) { + SnapshotPause* pause = static_cast(raw); + pause->reached.store(true, std::memory_order_release); + while (!pause->resume.load(std::memory_order_acquire)) { + std::this_thread::yield(); + } + }, + &pause); + + std::atomic suppressed{true}; + std::thread reader([&] { + suppressed.store(registry.shouldSuppressOwnedBlock(stale), + std::memory_order_release); + }); + auto deadline = std::chrono::steady_clock::now() + std::chrono::seconds(5); + while (!pause.reached.load(std::memory_order_acquire) && + std::chrono::steady_clock::now() < deadline) { + std::this_thread::yield(); + } + if (!pause.reached.load(std::memory_order_acquire)) { + pause.resume.store(true, std::memory_order_release); + reader.join(); + registry.setSuppressionSnapshotHookForTest(nullptr, nullptr); + GTEST_FAIL() << "Suppression reader did not reach the snapshot barrier"; + } + + registry.unregisterThread(slot_id, tid); + int reused_id = registry.registerThread(tid); + ThreadFilter::Slot* reused = registry.slotForId(reused_id); + u64 new_token = registry.enterBlockedRun(reused_id, OSThreadState::SLEEPING); + if (reused != nullptr && new_token != 0) { + reused->markSampledThisRun(OSThreadState::SLEEPING); + } + + pause.resume.store(true, std::memory_order_release); + reader.join(); + registry.setSuppressionSnapshotHookForTest(nullptr, nullptr); + + ASSERT_EQ(slot_id, reused_id); + ASSERT_NE(nullptr, reused); + ASSERT_NE(0u, new_token); + EXPECT_FALSE(suppressed.load(std::memory_order_acquire)); +} + +TEST_F(ThreadRegistryTest, TidIndexRemainsReusableAcrossLongThreadChurn) { + for (int tid = 1; tid <= ThreadFilter::kTidIndexSize * 3; ++tid) { + int slot_id = registry.registerThread(tid); + ASSERT_GE(slot_id, 0) << "tid=" << tid; + ThreadFilter::Slot* slot = registry.slotForId(slot_id); + ASSERT_EQ(slot, registry.lookupByTid(tid)); + registry.unregisterThread(slot_id); + ASSERT_EQ(nullptr, registry.lookupByTid(tid)); + } +} + +TEST_F(ThreadRegistryTest, ConfigurationSeparatesFilterAndUnfilteredTracking) { + registry.init("0", false); + EXPECT_TRUE(registry.enabled()); + EXPECT_TRUE(registry.registryActive()); + EXPECT_FALSE(registry.unfilteredWallTrackingActive()); + + registry.init("", false); + EXPECT_FALSE(registry.enabled()); + EXPECT_FALSE(registry.registryActive()); + EXPECT_FALSE(registry.unfilteredWallTrackingActive()); + + registry.init("", true); + EXPECT_FALSE(registry.enabled()); + EXPECT_TRUE(registry.registryActive()); + EXPECT_TRUE(registry.unfilteredWallTrackingActive()); +} + +TEST_F(ThreadRegistryTest, RecordingEpochMakesRetainedSlotInactiveUntilRefresh) { + constexpr int tid = 6101; + int slot_id = registry.registerThread(tid); + ASSERT_GE(slot_id, 0); + ThreadFilter::Slot* slot = registry.slotForId(slot_id); + ASSERT_NE(nullptr, slot); + ThreadFilter::RecordingEpoch first_epoch = registry.recordingEpoch(); + ASSERT_NE(0u, first_epoch); + EXPECT_EQ(slot, registry.lookupByTid(tid, first_epoch)); + + u64 token = registry.enterBlockedRun(slot_id, OSThreadState::SLEEPING); + ASSERT_NE(0u, token); + slot->markSampledThisRun(OSThreadState::SLEEPING); + ThreadEntry stale{tid, slot, slot->lifecycleGeneration(), + slot->recordingEpoch()}; + ASSERT_TRUE(registry.shouldSuppressOwnedBlock(stale)); + + registry.init("", true); + ThreadFilter::RecordingEpoch second_epoch = registry.recordingEpoch(); + ASSERT_NE(first_epoch, second_epoch); + EXPECT_EQ(nullptr, registry.lookupByTid(tid, first_epoch)); + EXPECT_EQ(nullptr, registry.lookupByTid(tid, second_epoch)); + EXPECT_FALSE(registry.shouldSuppressOwnedBlock(stale)); + + EXPECT_EQ(slot_id, registry.registerThread(tid)); + EXPECT_EQ(slot, registry.lookupByTid(tid, second_epoch)); + EXPECT_FALSE(slot->sampledThisRun()); + EXPECT_EQ(OSThreadState::UNKNOWN, slot->activeBlockState()); + EXPECT_EQ(BlockRunOwner::NONE, slot->activeBlockOwner()); +} + +TEST_F(ThreadRegistryTest, RetiresOnlySlotsNotRefreshedIntoCurrentEpoch) { + int retained_id = registry.registerThread(6103); + int stale_id = registry.registerThread(6104); + ASSERT_GE(retained_id, 0); + ASSERT_GE(stale_id, 0); + + registry.init("", true); + ThreadFilter::RecordingEpoch current_epoch = registry.recordingEpoch(); + EXPECT_EQ(retained_id, registry.registerThread(6103)); + + EXPECT_EQ(1, registry.retireInactiveRegistrations()); + EXPECT_NE(nullptr, registry.lookupByTid(6103, current_epoch)); + EXPECT_EQ(nullptr, registry.lookupByTid(6104)); + EXPECT_EQ(-1, registry.slotForId(stale_id)->nativeTid()); +} + +TEST_F(ThreadRegistryTest, ConcurrentRefreshAndRetirementKeepCurrentIdentity) { + constexpr int tid = 6110; + int slot_id = registry.registerThread(tid); + ASSERT_GE(slot_id, 0); + + for (int iteration = 0; iteration < 100; ++iteration) { + registry.init("", true); + ThreadFilter::RecordingEpoch epoch = registry.recordingEpoch(); + std::atomic start{false}; + int refreshed_id = -1; + + std::thread refresh([&] { + while (!start.load(std::memory_order_acquire)) { + std::this_thread::yield(); + } + refreshed_id = registry.registerThread(tid); + }); + std::thread retire([&] { + while (!start.load(std::memory_order_acquire)) { + std::this_thread::yield(); + } + registry.retireInactiveRegistrations(); + }); + + start.store(true, std::memory_order_release); + refresh.join(); + retire.join(); + + ASSERT_GE(refreshed_id, 0); + ThreadFilter::Slot* slot = registry.lookupByTid(tid, epoch); + ASSERT_NE(nullptr, slot); + EXPECT_EQ(tid, slot->nativeTid()); + slot_id = refreshed_id; + } + EXPECT_EQ(slot_id, registry.registerThread(tid)); +} + +TEST_F(ThreadRegistryTest, ExpectedTidProtectsReusedSlotDuringTeardown) { + int slot_id = registry.registerThread(6105); + ASSERT_GE(slot_id, 0); + registry.unregisterThread(slot_id, 9999); + EXPECT_NE(nullptr, registry.lookupByTid(6105)); + + registry.unregisterThread(slot_id, 6105); + EXPECT_EQ(nullptr, registry.lookupByTid(6105)); +} + +TEST_F(ThreadRegistryTest, DeactivationMakesSlotsIneligibleWithoutClearingStorage) { + constexpr int tid = 6106; + int slot_id = registry.registerThread(tid); + ASSERT_GE(slot_id, 0); + ThreadFilter::Slot* slot = registry.slotForId(slot_id); + ASSERT_NE(nullptr, slot); + ThreadFilter::RecordingEpoch epoch = registry.recordingEpoch(); + + registry.deactivateRecording(); + EXPECT_FALSE(registry.registryActive()); + EXPECT_FALSE(registry.unfilteredWallTrackingActive()); + EXPECT_EQ(0u, registry.recordingEpoch()); + EXPECT_EQ(nullptr, registry.lookupByTid(tid, epoch)); + EXPECT_EQ(slot, registry.lookupByTid(tid)); + EXPECT_EQ(-1, registry.registerThread(7777)); +} diff --git a/ddprof-lib/src/test/cpp/wallClockCandidateSelector_ut.cpp b/ddprof-lib/src/test/cpp/wallClockCandidateSelector_ut.cpp new file mode 100644 index 0000000000..77071af433 --- /dev/null +++ b/ddprof-lib/src/test/cpp/wallClockCandidateSelector_ut.cpp @@ -0,0 +1,173 @@ +/* + * Copyright 2026 Datadog, Inc + * SPDX-License-Identifier: Apache-2.0 + */ + +#include + +#include "wallClockCandidateSelector.h" + +#include +#include +#include +#include + +static std::vector makeCandidates(size_t count) { + std::vector candidates(count); + std::iota(candidates.begin(), candidates.end(), 0); + return candidates; +} + +TEST(WallClockCandidateSelectorTest, VisitsOnlyTargetSizeWithoutRejections) { + std::vector candidates = makeCandidates(1000); + std::mt19937 generator(1234); + std::set selected; + + WallClockCandidateStats stats = selectWallClockCandidates( + candidates, 10, 40, generator, [&](int tid) { + selected.insert(tid); + return WallClockCandidateOutcome::SIGNAL_SENT; + }); + + EXPECT_EQ(10u, stats.visited); + EXPECT_EQ(10u, stats.slots_consumed); + EXPECT_EQ(0u, stats.precheck_rejected); + EXPECT_EQ(10u, selected.size()); +} + +TEST(WallClockCandidateSelectorTest, PrecheckRejectedCandidatesAreBackfilled) { + std::vector candidates = makeCandidates(100); + std::mt19937 generator(42); + std::set selected; + + WallClockCandidateStats stats = selectWallClockCandidates( + candidates, 8, 100, generator, [&](int tid) { + if ((tid & 1) == 0) { + return WallClockCandidateOutcome::PRECHECK_REJECTED; + } + selected.insert(tid); + return WallClockCandidateOutcome::SIGNAL_SENT; + }); + + EXPECT_EQ(8u, stats.slots_consumed); + EXPECT_EQ(stats.slots_consumed + stats.precheck_rejected, stats.visited); + EXPECT_EQ(8u, selected.size()); + for (int tid : selected) { + EXPECT_EQ(1, tid & 1); + } +} + +TEST(WallClockCandidateSelectorTest, AllPrecheckRejectedCandidatesRespectVisitLimit) { + std::vector candidates = makeCandidates(257); + std::mt19937 generator(7); + std::set visited; + + WallClockCandidateStats stats = selectWallClockCandidates( + candidates, 10, 40, generator, [&](int tid) { + visited.insert(tid); + return WallClockCandidateOutcome::PRECHECK_REJECTED; + }); + + EXPECT_EQ(40u, stats.visited); + EXPECT_EQ(0u, stats.slots_consumed); + EXPECT_EQ(40u, stats.precheck_rejected); + EXPECT_EQ(40u, visited.size()); + EXPECT_TRUE(stats.visit_limit_reached); +} + +TEST(WallClockCandidateSelectorTest, SignalFailureConsumesCapacityWithoutBackfill) { + std::vector candidates{1, 2, 3, 4, 5}; + std::mt19937 generator(17); + int callbacks = 0; + + WallClockCandidateStats stats = selectWallClockCandidates( + candidates, 3, 12, generator, [&](int) { + callbacks++; + return WallClockCandidateOutcome::SIGNAL_FAILED; + }); + + EXPECT_EQ(3, callbacks); + EXPECT_EQ(3u, stats.visited); + EXPECT_EQ(3u, stats.slots_consumed); + EXPECT_EQ(0u, stats.precheck_rejected); +} + +TEST(WallClockCandidateSelectorTest, EmptyBoundsDoNoWork) { + std::vector candidates{1, 2, 3}; + std::vector empty; + std::mt19937 generator(1); + int callbacks = 0; + auto visitor = [&](int) { + callbacks++; + return WallClockCandidateOutcome::SIGNAL_SENT; + }; + + WallClockCandidateStats zero_target = + selectWallClockCandidates(candidates, 0, 3, generator, visitor); + WallClockCandidateStats empty_input = + selectWallClockCandidates(empty, 3, 3, generator, visitor); + WallClockCandidateStats zero_visits = + selectWallClockCandidates(candidates, 3, 0, generator, visitor); + + EXPECT_EQ(0, callbacks); + EXPECT_EQ(0u, zero_target.visited); + EXPECT_EQ(0u, empty_input.visited); + EXPECT_EQ(0u, zero_visits.visited); +} + +TEST(WallClockCandidateSelectorTest, FixedSeedProducesDeterministicTraversal) { + std::vector first = makeCandidates(50); + std::vector second = first; + std::mt19937 first_generator(2026); + std::mt19937 second_generator(2026); + std::vector first_result; + std::vector second_result; + + selectWallClockCandidates(first, 12, 24, first_generator, [&](int tid) { + first_result.push_back(tid); + return WallClockCandidateOutcome::SIGNAL_SENT; + }); + selectWallClockCandidates(second, 12, 24, second_generator, [&](int tid) { + second_result.push_back(tid); + return WallClockCandidateOutcome::SIGNAL_SENT; + }); + + EXPECT_EQ(first_result, second_result); +} + +TEST(WallClockCandidateSelectorTest, RandomizedPrefixRemainsFairAcrossCandidates) { + constexpr int candidate_count = 20; + constexpr int sample_size = 4; + constexpr int rounds = 10000; + std::vector candidates(candidate_count); + std::vector selections(candidate_count, 0); + std::mt19937 generator(2026); + + for (int round = 0; round < rounds; ++round) { + std::iota(candidates.begin(), candidates.end(), 0); + WallClockCandidateStats stats = selectWallClockCandidates( + candidates, sample_size, sample_size, generator, [&](int candidate) { + selections[candidate]++; + return WallClockCandidateOutcome::SIGNAL_SENT; + }); + ASSERT_EQ(sample_size, stats.slots_consumed); + } + + constexpr int expected = rounds * sample_size / candidate_count; + for (int count : selections) { + EXPECT_NEAR(expected, count, expected / 10); + } +} + +TEST(WallClockCandidateSelectorTest, ExhaustingInputDoesNotReportVisitLimit) { + std::vector candidates{1, 2, 3}; + std::mt19937 generator(8); + + WallClockCandidateStats stats = selectWallClockCandidates( + candidates, 5, 20, generator, + [](int) { return WallClockCandidateOutcome::PRECHECK_REJECTED; }); + + EXPECT_EQ(3u, stats.visited); + EXPECT_EQ(3u, stats.precheck_rejected); + EXPECT_FALSE(stats.visit_limit_reached); +} diff --git a/ddprof-lib/src/test/cpp/wallprecheck_args_ut.cpp b/ddprof-lib/src/test/cpp/wallprecheck_args_ut.cpp index 5579e354fe..c8b66591b3 100644 --- a/ddprof-lib/src/test/cpp/wallprecheck_args_ut.cpp +++ b/ddprof-lib/src/test/cpp/wallprecheck_args_ut.cpp @@ -5,6 +5,21 @@ #include #include "arguments.h" +#include "engine.h" +#include "j9/j9WallClock.h" +#include "wallClock.h" + +TEST(WallPrecheckCapabilityTest, OnlySupportingWallEnginesAdvertiseUnfilteredTracking) { + Engine engine; + J9WallClock j9; + WallClockASGCT asgct; + WallClockJvmti jvmti; + + EXPECT_FALSE(engine.supportsUnfilteredWallPrecheck()); + EXPECT_FALSE(j9.supportsUnfilteredWallPrecheck()); + EXPECT_TRUE(asgct.supportsUnfilteredWallPrecheck()); + EXPECT_TRUE(jvmti.supportsUnfilteredWallPrecheck()); +} TEST(WallPrecheckArgsTest, DefaultsToDisabled) { Arguments args; @@ -56,3 +71,17 @@ TEST(WallPrecheckArgsTest, EnabledWithinLongerArgString) { EXPECT_TRUE(args._wall_precheck); } +TEST(WallPrecheckArgsTest, OmittedFilterRemainsNull) { + Arguments args; + + EXPECT_EQ(nullptr, args._filter); +} + +TEST(WallPrecheckArgsTest, ExplicitEmptyFilterIsPreserved) { + Arguments args; + Error error = args.parse("filter="); + + EXPECT_FALSE(error); + ASSERT_NE(nullptr, args._filter); + EXPECT_STREQ("", args._filter); +} diff --git a/ddprof-stresstest/src/jmh/java/com/datadoghq/profiler/WallClockPrecheckBenchmarkHooks.java b/ddprof-stresstest/src/jmh/java/com/datadoghq/profiler/WallClockPrecheckBenchmarkHooks.java new file mode 100644 index 0000000000..ac96ec5b85 --- /dev/null +++ b/ddprof-stresstest/src/jmh/java/com/datadoghq/profiler/WallClockPrecheckBenchmarkHooks.java @@ -0,0 +1,21 @@ +/* + * Copyright 2026, Datadog, Inc. + * SPDX-License-Identifier: Apache-2.0 + */ + +package com.datadoghq.profiler; + +/** Exposes package-scoped owned-block hooks to the wall-clock overhead benchmark. */ +public final class WallClockPrecheckBenchmarkHooks { + private WallClockPrecheckBenchmarkHooks() {} + + /** Marks the current benchmark worker as entering an owned sleeping interval. */ + public static long enterSleeping(JavaProfiler profiler) { + return profiler.blockEnter(7); + } + + /** Closes an interval returned by {@link #enterSleeping(JavaProfiler)}. */ + public static void exit(JavaProfiler profiler, long token) { + profiler.blockExit(token); + } +} diff --git a/ddprof-stresstest/src/jmh/java/com/datadoghq/profiler/stresstest/scenarios/throughput/WallClockPrecheckOverheadBenchmark.java b/ddprof-stresstest/src/jmh/java/com/datadoghq/profiler/stresstest/scenarios/throughput/WallClockPrecheckOverheadBenchmark.java new file mode 100644 index 0000000000..1af41a979f --- /dev/null +++ b/ddprof-stresstest/src/jmh/java/com/datadoghq/profiler/stresstest/scenarios/throughput/WallClockPrecheckOverheadBenchmark.java @@ -0,0 +1,145 @@ +/* + * 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. + */ +package com.datadoghq.profiler.stresstest.scenarios.throughput; + +import com.datadoghq.profiler.JavaProfiler; +import com.datadoghq.profiler.WallClockPrecheckBenchmarkHooks; +import java.nio.file.Files; +import java.nio.file.Path; +import java.util.ArrayList; +import java.util.List; +import java.util.concurrent.CountDownLatch; +import java.util.concurrent.TimeUnit; +import java.util.concurrent.locks.LockSupport; +import org.openjdk.jmh.annotations.Benchmark; +import org.openjdk.jmh.annotations.BenchmarkMode; +import org.openjdk.jmh.annotations.Fork; +import org.openjdk.jmh.annotations.Level; +import org.openjdk.jmh.annotations.Measurement; +import org.openjdk.jmh.annotations.Mode; +import org.openjdk.jmh.annotations.OutputTimeUnit; +import org.openjdk.jmh.annotations.Param; +import org.openjdk.jmh.annotations.Scope; +import org.openjdk.jmh.annotations.Setup; +import org.openjdk.jmh.annotations.State; +import org.openjdk.jmh.annotations.TearDown; +import org.openjdk.jmh.annotations.Warmup; + +/** + * Measures steady-state wall-clock timer overhead as an owned-block thread population grows. + * + *

The implementation separately records registry lookup work in the {@code + * wc_precheck_registry_lookups} debug counter and bounds candidate visits to four times {@code + * walltpt} per tick. This benchmark does not report that counter; it compares {@code + * precheck=false} and {@code precheck=true} at each population to detect timer-loop throughput + * regressions independently of one-time startup registration. + */ +@BenchmarkMode(Mode.Throughput) +@OutputTimeUnit(TimeUnit.MILLISECONDS) +@Fork(value = 1, warmups = 0, jvmArgsAppend = "-Xss256k") +@Warmup(iterations = 3, time = 2) +@Measurement(iterations = 5, time = 3) +@State(Scope.Benchmark) +public class WallClockPrecheckOverheadBenchmark { + @Param({"false", "true"}) + public boolean precheck; + + @Param({"100", "500", "1000"}) + public int threadCount; + + private final List workers = new ArrayList<>(); + private volatile boolean running; + private JavaProfiler profiler; + private Path recording; + + /** Creates the requested thread population before starting the profiler. */ + @Setup(Level.Trial) + public void setup() throws Exception { + running = true; + CountDownLatch ready = new CountDownLatch(threadCount); + CountDownLatch profilerStarted = new CountDownLatch(1); + CountDownLatch armed = new CountDownLatch(threadCount); + for (int i = 0; i < threadCount; ++i) { + Thread worker = + new Thread( + () -> { + ready.countDown(); + long token = 0; + boolean armedReported = false; + try { + profilerStarted.await(); + token = WallClockPrecheckBenchmarkHooks.enterSleeping(profiler); + armed.countDown(); + armedReported = true; + while (running) { + LockSupport.parkNanos(TimeUnit.SECONDS.toNanos(1)); + } + } catch (InterruptedException interrupted) { + Thread.currentThread().interrupt(); + } finally { + if (!armedReported) { + armed.countDown(); + } + WallClockPrecheckBenchmarkHooks.exit(profiler, token); + } + }, + "wall-precheck-benchmark-" + i); + worker.setDaemon(true); + worker.start(); + workers.add(worker); + } + if (!ready.await(30, TimeUnit.SECONDS)) { + throw new IllegalStateException("Timed out creating benchmark workers"); + } + + profiler = JavaProfiler.getInstance(); + recording = Files.createTempFile("wall-precheck-overhead-", ".jfr"); + profiler.execute( + "start,wall=1ms,walltpt=16,filter=,wallprecheck=" + + precheck + + ",jfr,file=" + + recording.toAbsolutePath()); + profilerStarted.countDown(); + if (!armed.await(30, TimeUnit.SECONDS)) { + throw new IllegalStateException("Timed out arming benchmark workers"); + } + } + + /** Stops profiling and releases every background worker. */ + @TearDown(Level.Trial) + public void tearDown() throws Exception { + running = false; + for (Thread worker : workers) { + LockSupport.unpark(worker); + } + for (Thread worker : workers) { + worker.join(TimeUnit.SECONDS.toMillis(5)); + } + workers.clear(); + if (profiler != null) { + profiler.stop(); + } + if (recording != null) { + Files.deleteIfExists(recording); + } + } + + /** Provides stable foreground work whose throughput captures timer-loop interference. */ + @Benchmark + public void foregroundWork() { + org.openjdk.jmh.infra.Blackhole.consumeCPU(1_000); + } +} diff --git a/ddprof-test/src/test/java/com/datadoghq/profiler/AbstractProfilerTest.java b/ddprof-test/src/test/java/com/datadoghq/profiler/AbstractProfilerTest.java index de75c2f068..ed84e9f497 100644 --- a/ddprof-test/src/test/java/com/datadoghq/profiler/AbstractProfilerTest.java +++ b/ddprof-test/src/test/java/com/datadoghq/profiler/AbstractProfilerTest.java @@ -1,3 +1,8 @@ +/* + * Copyright 2026, Datadog, Inc. + * SPDX-License-Identifier: Apache-2.0 + */ + package com.datadoghq.profiler; import java.nio.file.Files; @@ -233,6 +238,7 @@ public void setupProfiler(TestInfo testInfo) throws Exception { jfrDump = Files.createTempFile(rootDir, testInfo.getTestMethod().map(m -> m.getDeclaringClass().getSimpleName() + "_" + m.getName()).orElse("unknown") + (testConfig.isEmpty() ? "" : "-" + testConfig.replace('/', '_')), ".jfr"); profiler = JavaProfiler.getInstance(); + beforeProfilerStart(); String command = "start," + getAmendedProfilerCommand() + ",jfr,file=" + jfrDump.toAbsolutePath(); cpuInterval = command.contains("cpu") ? parseInterval(command, "cpu") : (command.contains("interval") ? parseInterval(command, "interval") : Duration.ZERO); wallInterval = parseInterval(command, "wall"); @@ -268,6 +274,17 @@ public void cleanup() throws Exception { protected void before() throws Exception { } + /** + * Runs after the profiler instance is available but before the recording starts. + * + *

Tests may override this hook when their setup must predate profiler thread-event + * registration. + * + * @throws Exception if setup fails + */ + protected void beforeProfilerStart() throws Exception { + } + protected void after() throws Exception { } @@ -504,4 +521,4 @@ public long getRecordedCounterValue(String counterName) { } return -1; } -} \ No newline at end of file +} diff --git a/ddprof-test/src/test/java/com/datadoghq/profiler/wallclock/J9WallClockPrecheckCapabilityTest.java b/ddprof-test/src/test/java/com/datadoghq/profiler/wallclock/J9WallClockPrecheckCapabilityTest.java new file mode 100644 index 0000000000..f91eeef578 --- /dev/null +++ b/ddprof-test/src/test/java/com/datadoghq/profiler/wallclock/J9WallClockPrecheckCapabilityTest.java @@ -0,0 +1,36 @@ +/* + * Copyright 2026, Datadog, Inc. + * SPDX-License-Identifier: Apache-2.0 + */ + +package com.datadoghq.profiler.wallclock; + +import static org.junit.jupiter.api.Assertions.assertEquals; + +import com.datadoghq.profiler.AbstractProfilerTest; +import com.datadoghq.profiler.Platform; +import com.datadoghq.profiler.ProfilerOwnedBlockHooks; +import org.junit.jupiter.api.Test; + +/** Verifies that unsupported J9 wall sampling does not activate unfiltered precheck tracking. */ +public class J9WallClockPrecheckCapabilityTest extends AbstractProfilerTest { + private static final int OSTHREAD_STATE_SLEEPING = 7; + + /** Ensures owned-block hooks stay inactive when the selected wall engine cannot consume them. */ + @Test + public void unsupportedEngineDoesNotActivateRegistry() { + long token = ProfilerOwnedBlockHooks.blockEnter(profiler, OSTHREAD_STATE_SLEEPING); + + assertEquals(0L, token, "J9WallClock must not activate unfiltered precheck tracking"); + } + + @Override + protected boolean isPlatformSupported() { + return Platform.isJ9(); + } + + @Override + protected String getProfilerCommand() { + return "wall=1ms,wallsampler=jvmti,filter=,wallprecheck=true"; + } +} diff --git a/ddprof-test/src/test/java/com/datadoghq/profiler/wallclock/JvmtiBasedUnfilteredWallPrecheckTest.java b/ddprof-test/src/test/java/com/datadoghq/profiler/wallclock/JvmtiBasedUnfilteredWallPrecheckTest.java new file mode 100644 index 0000000000..1c5491e3a9 --- /dev/null +++ b/ddprof-test/src/test/java/com/datadoghq/profiler/wallclock/JvmtiBasedUnfilteredWallPrecheckTest.java @@ -0,0 +1,44 @@ +/* + * Copyright 2026, Datadog, Inc. + * SPDX-License-Identifier: Apache-2.0 + */ + +package com.datadoghq.profiler.wallclock; + +import static org.junit.jupiter.api.Assertions.assertTrue; + +import java.util.Map; +import org.junit.jupiter.api.Assumptions; + +/** Runs unfiltered owned-block precheck coverage through delegated JVMTI stack collection. */ +public class JvmtiBasedUnfilteredWallPrecheckTest extends UnfilteredWallPrecheckTest { + private boolean jvmtiDelegationAvailable; + private long requestedBefore; + + @Override + protected void before() { + Map counters = profiler.getDebugCounters(); + Assumptions.assumeTrue( + counters.getOrDefault("jvmti_stacks_init_ok", 0L) > 0, + "HotSpot RequestStackTrace JVMTI extension is not available"); + jvmtiDelegationAvailable = true; + requestedBefore = counters.getOrDefault("jvmti_stacks_requested", 0L); + } + + @Override + protected void after() { + if (!jvmtiDelegationAvailable) { + return; + } + long requestedAfter = + profiler.getDebugCounters().getOrDefault("jvmti_stacks_requested", 0L); + assertTrue( + requestedAfter > requestedBefore, + "Expected wallclock jvmtistacks path to request delegated stack traces"); + } + + @Override + protected String getProfilerCommand() { + return super.getProfilerCommand() + ",jvmtistacks=true"; + } +} diff --git a/ddprof-test/src/test/java/com/datadoghq/profiler/wallclock/UnfilteredWallPrecheckRestartTest.java b/ddprof-test/src/test/java/com/datadoghq/profiler/wallclock/UnfilteredWallPrecheckRestartTest.java new file mode 100644 index 0000000000..afc78ad543 --- /dev/null +++ b/ddprof-test/src/test/java/com/datadoghq/profiler/wallclock/UnfilteredWallPrecheckRestartTest.java @@ -0,0 +1,127 @@ +/* + * Copyright 2026, Datadog, Inc. + * SPDX-License-Identifier: Apache-2.0 + */ + +package com.datadoghq.profiler.wallclock; + +import static org.junit.jupiter.api.Assertions.assertEquals; +import static org.junit.jupiter.api.Assertions.assertNotEquals; + +import com.datadoghq.profiler.AbstractProfilerTest; +import com.datadoghq.profiler.Platform; +import com.datadoghq.profiler.ProfilerOwnedBlockHooks; +import java.nio.file.Files; +import java.nio.file.Path; +import java.util.concurrent.ExecutorService; +import java.util.concurrent.Executors; +import java.util.concurrent.Future; +import org.junit.jupiter.api.Test; + +/** Verifies that unfiltered wall registry activation does not leak across recordings. */ +public class UnfilteredWallPrecheckRestartTest extends AbstractProfilerTest { + private static final int OSTHREAD_STATE_SLEEPING = 7; + + /** Exercises enabled, disabled, CPU-only, and re-enabled tracking in one process. */ + @Test + public void recordingRestartsReconfigureUnfilteredTracking() throws Exception { + assertOwnedBlockArmed(); + stopProfiler(); + + runRecording("wall=1ms,filter=,wallprecheck=false", false); + runRecording("cpu=1ms,filter=,wallprecheck=true", false); + runRecording("wall=1ms,filter=,wallprecheck=true", true); + } + + /** Verifies epoch refresh and lazy registration for workers that survive a stopped gap. */ + @Test + public void workerLifecyclesRemainSafeAcrossStoppedGap() throws Exception { + ExecutorService survivingWorker = Executors.newSingleThreadExecutor(); + ExecutorService stoppedGapWorker = null; + Path recording = null; + boolean restarted = false; + try { + long oldToken = enterBlock(survivingWorker); + assertNotEquals(0L, oldToken, "Expected the initial worker run to be armed"); + + stopProfiler(); + stoppedGapWorker = Executors.newSingleThreadExecutor(); + // Force creation while JVMTI lifecycle callbacks are disabled. + assertEquals(0L, enterBlock(stoppedGapWorker)); + + recording = Files.createTempFile("unfiltered-wall-worker-restart-", ".jfr"); + profiler.execute( + "start," + getProfilerCommand() + ",jfr,file=" + recording.toAbsolutePath()); + restarted = true; + + long newToken = enterBlock(survivingWorker); + assertNotEquals(0L, newToken, "Expected the surviving worker to refresh its slot"); + exitBlock(survivingWorker, oldToken); + assertEquals( + 0L, + enterBlock(survivingWorker), + "A token from the previous recording cleared the current worker run"); + exitBlock(survivingWorker, newToken); + + long stoppedGapToken = enterBlock(stoppedGapWorker); + assertNotEquals( + 0L, stoppedGapToken, "Expected the stopped-gap worker to register lazily after restart"); + exitBlock(stoppedGapWorker, stoppedGapToken); + } finally { + if (restarted) { + profiler.stop(); + } + survivingWorker.shutdownNow(); + if (stoppedGapWorker != null) { + stoppedGapWorker.shutdownNow(); + } + if (recording != null) { + Files.deleteIfExists(recording); + } + } + } + + @Override + protected String getProfilerCommand() { + return "wall=1ms,filter=,wallprecheck=true"; + } + + @Override + protected boolean isPlatformSupported() { + return !Platform.isJ9(); + } + + private void assertOwnedBlockArmed() { + long token = ProfilerOwnedBlockHooks.blockEnter(profiler, OSTHREAD_STATE_SLEEPING); + assertNotEquals(0L, token, "Expected unfiltered wall precheck to arm the owned block"); + ProfilerOwnedBlockHooks.blockExit(profiler, token); + } + + private void runRecording(String command, boolean expectArmed) throws Exception { + Path recording = Files.createTempFile("unfiltered-wall-restart-", ".jfr"); + profiler.execute("start," + command + ",jfr,file=" + recording.toAbsolutePath()); + try { + long token = ProfilerOwnedBlockHooks.blockEnter(profiler, OSTHREAD_STATE_SLEEPING); + if (expectArmed) { + assertNotEquals(0L, token, "Expected unfiltered wall tracking after restart"); + ProfilerOwnedBlockHooks.blockExit(profiler, token); + } else { + assertEquals(0L, token, "Registry tracking leaked into " + command); + } + } finally { + profiler.stop(); + Files.deleteIfExists(recording); + } + } + + private long enterBlock(ExecutorService worker) throws Exception { + Future result = + worker.submit( + () -> ProfilerOwnedBlockHooks.blockEnter(profiler, OSTHREAD_STATE_SLEEPING)); + return result.get(); + } + + private void exitBlock(ExecutorService worker, long token) throws Exception { + worker.submit(() -> ProfilerOwnedBlockHooks.blockExit(profiler, token)).get(); + } +} diff --git a/ddprof-test/src/test/java/com/datadoghq/profiler/wallclock/UnfilteredWallPrecheckTest.java b/ddprof-test/src/test/java/com/datadoghq/profiler/wallclock/UnfilteredWallPrecheckTest.java new file mode 100644 index 0000000000..d98f619f10 --- /dev/null +++ b/ddprof-test/src/test/java/com/datadoghq/profiler/wallclock/UnfilteredWallPrecheckTest.java @@ -0,0 +1,239 @@ +/* + * Copyright 2026, Datadog, Inc. + * SPDX-License-Identifier: Apache-2.0 + */ + +package com.datadoghq.profiler.wallclock; + +import static org.junit.jupiter.api.Assertions.assertSame; +import static org.junit.jupiter.api.Assertions.assertTrue; + +import com.datadoghq.profiler.AbstractProfilerTest; +import com.datadoghq.profiler.Platform; +import com.datadoghq.profiler.ProfilerOwnedBlockHooks; +import java.util.concurrent.ExecutorService; +import java.util.concurrent.Executors; +import java.util.concurrent.Future; +import java.util.concurrent.FutureTask; +import java.util.concurrent.TimeUnit; +import org.junit.jupiter.api.AfterEach; +import org.junit.jupiter.api.Assumptions; +import org.junitpioneer.jupiter.RetryingTest; +import org.openjdk.jmc.common.item.IItem; +import org.openjdk.jmc.common.item.IItemCollection; +import org.openjdk.jmc.common.item.IItemIterable; +import org.openjdk.jmc.common.item.IMemberAccessor; +import org.openjdk.jmc.flightrecorder.jdk.JdkAttributes; + +/** Verifies owned-block prechecks when legacy {@code filter=} samples every thread. */ +public class UnfilteredWallPrecheckTest extends AbstractProfilerTest { + private static final int OSTHREAD_STATE_SLEEPING = 7; + private static final long SLEEP_MILLIS = 300; + private static final String PRE_EXISTING_THREAD_NAME = "unfiltered-precheck-existing"; + private static final String SUPPRESSED_RUN_COUNTER = "wc_signals_suppressed_sampled_run"; + + private ExecutorService preExistingWorker; + private Thread preExistingThread; + + /** + * Verifies that an untraced thread's owned sleeping run is sampled once and then suppressed. + * + * @throws Exception if the worker cannot complete + */ + @RetryingTest(3) + public void sleepingThreadOutsideContextWindowIsOwnedBlockSuppressed() throws Exception { + long suppressedBefore = suppressedSignals(); + assertTrue( + runPreExistingSleepingWorker(false) != 0, + "Expected native blockEnter to arm SLEEPING state"); + + stopProfiler(); + assertSuppressedSamples(PRE_EXISTING_THREAD_NAME); + assertOwnedBlockSuppressionObserved(suppressedBefore); + } + + /** + * Verifies that entering the context window prevents owned-block suppression in an + * unfiltered recording. + * + * @throws Exception if the worker cannot complete + */ + @RetryingTest(3) + public void sleepingThreadInsideContextWindowIsNotOverSuppressed() throws Exception { + assertTrue( + runPreExistingSleepingWorker(true) != 0, + "Expected native blockEnter to arm SLEEPING state"); + + stopProfiler(); + + long sampleCount = samplesForThread(PRE_EXISTING_THREAD_NAME); + assertTrue( + sampleCount >= 10, + "Expected normal MethodSample volume inside the context window, got: " + sampleCount); + } + + /** + * Verifies that a pre-existing thread can lazily bind its slot through the park hook. + * + * @throws Exception if the worker cannot complete + */ + @RetryingTest(3) + public void parkedPreExistingThreadOutsideContextWindowIsOwnedBlockSuppressed() + throws Exception { + long suppressedBefore = suppressedSignals(); + runPreExistingParkedWorker(); + + stopProfiler(); + assertSuppressedSamples(PRE_EXISTING_THREAD_NAME); + assertOwnedBlockSuppressionObserved(suppressedBefore); + } + + /** + * Retains coverage for threads whose filter slot is installed by a post-start ThreadStart event. + * + * @throws Exception if the worker cannot complete + */ + @RetryingTest(3) + public void postStartSleepingThreadStillUsesThreadStartSlot() throws Exception { + String threadName = "unfiltered-precheck-post-start"; + assertTrue( + runPostStartSleepingWorker(threadName) != 0, + "Expected ThreadStart registration to arm SLEEPING state"); + + stopProfiler(); + assertSuppressedSamples(threadName); + } + + @Override + protected void beforeProfilerStart() throws Exception { + preExistingWorker = + Executors.newSingleThreadExecutor( + task -> { + Thread worker = new Thread(task, PRE_EXISTING_THREAD_NAME); + worker.setDaemon(true); + return worker; + }); + preExistingThread = preExistingWorker.submit(Thread::currentThread).get(); + } + + /** Stops the worker that was deliberately created before profiler startup. */ + @AfterEach + public void stopPreExistingWorker() throws InterruptedException { + if (preExistingWorker == null) { + return; + } + preExistingWorker.shutdownNow(); + assertTrue( + preExistingWorker.awaitTermination(5, TimeUnit.SECONDS), + "Pre-existing wall-clock worker did not terminate"); + } + + @Override + protected boolean isPlatformSupported() { + return !Platform.isJ9(); + } + + @Override + protected void withTestAssumptions() { + Assumptions.assumeTrue( + Platform.isJavaVersionAtLeast(11), + "Sleeping-state precheck assertions are stable on JDK 11+"); + } + + @Override + protected String getProfilerCommand() { + return "wall=1ms,filter=,wallprecheck=true"; + } + + private long runPreExistingSleepingWorker(boolean enterContextWindowDuringBlock) + throws Exception { + Future sleep = + preExistingWorker.submit( + () -> { + assertSame(preExistingThread, Thread.currentThread()); + return runSleepingBlock(enterContextWindowDuringBlock); + }); + return sleep.get(); + } + + private long runPostStartSleepingWorker(String threadName) throws Exception { + FutureTask sleep = new FutureTask<>(() -> runSleepingBlock(false)); + Thread worker = new Thread(sleep, threadName); + worker.start(); + return sleep.get(); + } + + private long runSleepingBlock(boolean enterContextWindowDuringBlock) throws Exception { + long token = ProfilerOwnedBlockHooks.blockEnter(profiler, OSTHREAD_STATE_SLEEPING); + if (enterContextWindowDuringBlock) { + profiler.addThread(); + } + try { + Thread.sleep(SLEEP_MILLIS); + return token; + } finally { + ProfilerOwnedBlockHooks.blockExit(profiler, token); + if (enterContextWindowDuringBlock) { + profiler.removeThread(); + } + } + } + + private void runPreExistingParkedWorker() throws Exception { + preExistingWorker + .submit( + () -> { + assertSame(preExistingThread, Thread.currentThread()); + ProfilerOwnedBlockHooks.parkEnter(profiler); + try { + long deadline = System.nanoTime() + TimeUnit.MILLISECONDS.toNanos(SLEEP_MILLIS); + while (System.nanoTime() < deadline) { + // Keep the OS thread runnable so suppression must come from the owned park marker. + } + } finally { + ProfilerOwnedBlockHooks.parkExit( + profiler, System.identityHashCode(preExistingThread), 0L); + } + return null; + }) + .get(); + } + + private void assertOwnedBlockSuppressionObserved(long suppressedBefore) { + if (suppressedBefore >= 0) { + assertTrue( + suppressedSignals() > suppressedBefore, + "Expected owned-block once-per-run suppression counter to increase"); + } + } + + private long suppressedSignals() { + return profiler.getDebugCounters().getOrDefault(SUPPRESSED_RUN_COUNTER, -1L); + } + + private void assertSuppressedSamples(String threadName) { + long sampleCount = samplesForThread(threadName); + assertTrue(sampleCount > 0, "Expected the owned block run to be sampled once"); + assertTrue( + sampleCount < 10, + "Expected nearly no samples from owned block thread, got: " + sampleCount); + } + + private long samplesForThread(String threadName) { + long count = 0; + IItemCollection events = verifyEvents("datadog.MethodSample", false); + for (IItemIterable batch : events) { + IMemberAccessor threadNameAccessor = + JdkAttributes.EVENT_THREAD_NAME.getAccessor(batch.getType()); + if (threadNameAccessor == null) { + continue; + } + for (IItem item : batch) { + if (threadName.equals(threadNameAccessor.getMember(item))) { + count++; + } + } + } + return count; + } +} From ec76f5829d243f516dee70833f3dfbe7a2b9eaa3 Mon Sep 17 00:00:00 2001 From: Paul Fournillon Date: Tue, 21 Jul 2026 09:56:51 +0200 Subject: [PATCH 2/2] fix: apply review suggestions --- ddprof-lib/src/main/cpp/wallClockCandidateSelector.h | 2 +- ddprof-lib/src/test/cpp/wallClockCandidateSelector_ut.cpp | 2 +- 2 files changed, 2 insertions(+), 2 deletions(-) diff --git a/ddprof-lib/src/main/cpp/wallClockCandidateSelector.h b/ddprof-lib/src/main/cpp/wallClockCandidateSelector.h index 8a8358e79b..189c5da866 100644 --- a/ddprof-lib/src/main/cpp/wallClockCandidateSelector.h +++ b/ddprof-lib/src/main/cpp/wallClockCandidateSelector.h @@ -1,5 +1,5 @@ /* - * Copyright 2026 Datadog, Inc + * Copyright 2026, Datadog, Inc. * SPDX-License-Identifier: Apache-2.0 */ diff --git a/ddprof-lib/src/test/cpp/wallClockCandidateSelector_ut.cpp b/ddprof-lib/src/test/cpp/wallClockCandidateSelector_ut.cpp index 77071af433..b55073fc5e 100644 --- a/ddprof-lib/src/test/cpp/wallClockCandidateSelector_ut.cpp +++ b/ddprof-lib/src/test/cpp/wallClockCandidateSelector_ut.cpp @@ -1,5 +1,5 @@ /* - * Copyright 2026 Datadog, Inc + * Copyright 2026, Datadog, Inc. * SPDX-License-Identifier: Apache-2.0 */