From 4351e79aa72c9c46b37757820ed7be82dcbd49b6 Mon Sep 17 00:00:00 2001 From: Paul Fournillon Date: Wed, 15 Jul 2026 11:44:48 +0200 Subject: [PATCH 1/9] feat(taskblock): add JVM blocking producers # Conflicts: # ddprof-lib/src/main/cpp/javaApi.cpp # ddprof-lib/src/main/java/com/datadoghq/profiler/JavaProfiler.java --- ddprof-lib/src/main/cpp/javaApi.cpp | 89 +++++-- ddprof-lib/src/main/cpp/profiler.cpp | 7 + ddprof-lib/src/main/cpp/profiler.h | 1 + ddprof-lib/src/main/cpp/threadLocalData.h | 96 ++++++- ddprof-lib/src/main/cpp/vmEntry.cpp | 189 +++++++++++++- ddprof-lib/src/main/cpp/vmEntry.h | 12 +- .../com/datadoghq/profiler/JavaProfiler.java | 51 +++- ddprof-lib/src/test/cpp/park_state_ut.cpp | 73 ++++++ .../profiler/JavaProfilerApiSurfaceTest.java | 10 + .../JvmtiBasedMonitorTaskBlockTest.java | 30 +++ .../JvmtiBasedParkTaskBlockTest.java | 30 +++ .../wallclock/MonitorTaskBlockTest.java | 240 ++++++++++++++++++ .../profiler/wallclock/ParkTaskBlockTest.java | 115 +++++++++ .../wallclock/TaskBlockAssertions.java | 11 + .../WallclockMitigationsCombinedTest.java | 5 +- 15 files changed, 911 insertions(+), 48 deletions(-) create mode 100644 ddprof-test/src/test/java/com/datadoghq/profiler/wallclock/JvmtiBasedMonitorTaskBlockTest.java create mode 100644 ddprof-test/src/test/java/com/datadoghq/profiler/wallclock/JvmtiBasedParkTaskBlockTest.java create mode 100644 ddprof-test/src/test/java/com/datadoghq/profiler/wallclock/MonitorTaskBlockTest.java create mode 100644 ddprof-test/src/test/java/com/datadoghq/profiler/wallclock/ParkTaskBlockTest.java diff --git a/ddprof-lib/src/main/cpp/javaApi.cpp b/ddprof-lib/src/main/cpp/javaApi.cpp index b2f1028fa..c0b93a27c 100644 --- a/ddprof-lib/src/main/cpp/javaApi.cpp +++ b/ddprof-lib/src/main/cpp/javaApi.cpp @@ -69,7 +69,8 @@ class JniString { }; extern "C" DLLEXPORT jboolean JNICALL -Java_com_datadoghq_profiler_JavaProfiler_init0(JNIEnv *env, jclass unused) { +Java_com_datadoghq_profiler_JavaProfiler_init0( + JNIEnv *env, jclass unused, jboolean delegateMonitorWaitEvents) { Error error = Profiler::instance()->init(); if (error) { throwNew(env, "java/lang/IllegalStateException", error.message()); @@ -77,7 +78,7 @@ Java_com_datadoghq_profiler_JavaProfiler_init0(JNIEnv *env, jclass unused) { } // JavaVM* has already been stored when the native library was loaded so we can pass nullptr here - return VM::initProfilerBridge(nullptr, true); + return VM::initProfilerBridge(nullptr, true, delegateMonitorWaitEvents); } extern "C" DLLEXPORT void JNICALL @@ -94,6 +95,12 @@ Java_com_datadoghq_profiler_JavaProfiler_getTid0(JNIEnv *env, jclass unused) { return OS::threadId(); } +extern "C" DLLEXPORT jboolean JNICALL +Java_com_datadoghq_profiler_JavaProfiler_monitorEventsDelegated0( + JNIEnv *env, jclass unused) { + return VM::monitorEventsDelegated(); +} + extern "C" DLLEXPORT jstring JNICALL Java_com_datadoghq_profiler_JavaProfiler_execute0(JNIEnv *env, jobject unused, jstring command) { @@ -360,43 +367,78 @@ Java_com_datadoghq_profiler_JavaProfiler_recordQueueEnd0( } extern "C" DLLEXPORT jboolean JNICALL -Java_com_datadoghq_profiler_JavaProfiler_parkEnter0(JNIEnv *env, jclass unused) { +Java_com_datadoghq_profiler_JavaProfiler_parkEnter0( + JNIEnv *env, jclass unused, jthread thread) { + if (!JVMSupport::isPlatformThread(env, thread)) { + return JNI_FALSE; + } ProfiledThread *current = ProfiledThread::initCurrentThreadSignalSafe(); if (current == nullptr) { return JNI_FALSE; } - bool first_park = current->parkEnter(); - ThreadFilter *tf = Profiler::instance()->threadFilter(); - if (first_park && tf->registryActive()) { + Context context = ContextApi::snapshot(); + if (!current->parkEnter(TSC::ticks(), context)) { + return JNI_FALSE; + } + + Profiler *profiler = Profiler::instance(); + ThreadFilter *tf = profiler->threadFilter(); + if (context.spanId == 0 && tf->registryActive() && + (profiler->taskBlockEnabled() || tf->enabled())) { ThreadFilter::SlotID slot_id = ensureCurrentThreadFilterSlot(tf, current); if (slot_id >= 0) { - current->setParkBlockToken( - tf->enterBlockedRun(slot_id, OSThreadState::CONDVAR_WAIT)); + current->setParkBlockToken(tf->enterBlockedRun( + slot_id, OSThreadState::CONDVAR_WAIT, BlockRunOwner::JAVA)); } } - return first_park ? JNI_TRUE : JNI_FALSE; + return JNI_TRUE; } extern "C" DLLEXPORT void JNICALL Java_com_datadoghq_profiler_JavaProfiler_parkExit0( - JNIEnv *env, jclass unused, jlong blocker, jlong unblockingSpanId) { + JNIEnv *env, jclass unused, jthread thread, jlong blocker, + jlong unblockingSpanId) { + if (!JVMSupport::isPlatformThread(env, thread)) { + return; + } ProfiledThread *current = ProfiledThread::initCurrentThreadSignalSafe(); if (current == nullptr) { return; } - + u64 start_ticks = 0; u64 park_block_token = 0; - if (!current->parkExit(park_block_token) || park_block_token == 0) { + Context context{}; + if (!current->parkExit(start_ticks, context, park_block_token) || + park_block_token == 0) { return; } - ThreadFilter *tf = Profiler::instance()->threadFilter(); - if (tf->registryActive()) { - ThreadFilter::SlotID slot_id = ThreadFilter::tokenSlotId(park_block_token); - if (tf->activeSlotForId(current->filterSlotId(), current->tid()) != nullptr && - current->filterSlotId() == slot_id) { - tf->exitBlockedRun(slot_id, ThreadFilter::tokenGeneration(park_block_token)); - } + Profiler *profiler = Profiler::instance(); + bool recording_enabled = profiler->taskBlockEnabled(); + bool activity = profiler->tryEnterTaskBlockActivity(); + if (!activity) profiler->waitForTaskBlockRotation(); + + ThreadFilter *tf = profiler->threadFilter(); + ThreadFilter::SlotID slot_id = ThreadFilter::tokenSlotId(park_block_token); + ThreadFilter::SlotID current_slot = current->filterSlotId(); + if (current_slot < 0) current_slot = tf->slotIdByTid(current->tid()); + BlockRunSnapshot snapshot{}; + bool exited = current_slot == slot_id && + tf->snapshotAndExitBlockedRun( + slot_id, ThreadFilter::tokenGeneration(park_block_token), &snapshot); + + if (!activity) { + Counters::increment(TASK_BLOCK_DROPPED_ROTATION); + return; } + if (recording_enabled && exited && snapshot.context_eligible) { + recordTaskBlockIfEligible( + current->tid(), thread, 1, start_ticks, TSC::ticks(), context, + static_cast(blocker), static_cast(unblockingSpanId), + snapshot.active_state, true); + } else if (recording_enabled && exited && !snapshot.context_eligible) { + Counters::increment(TASK_BLOCK_SKIPPED_TRACE_CONTEXT); + } + profiler->leaveTaskBlockActivity(); } static bool decodeJavaBlockState(jint state, OSThreadState &decoded) { @@ -410,9 +452,10 @@ static bool decodeJavaBlockState(jint state, OSThreadState &decoded) { extern "C" DLLEXPORT jlong JNICALL Java_com_datadoghq_profiler_JavaProfiler_blockEnter0( - JNIEnv *env, jclass unused, jint state) { + JNIEnv *env, jclass unused, jthread thread, jint state) { OSThreadState decoded; - if (!decodeJavaBlockState(state, decoded)) { + if (!decodeJavaBlockState(state, decoded) || + !JVMSupport::isPlatformThread(env, thread)) { return 0; } ProfiledThread *current = ProfiledThread::initCurrentThreadSignalSafe(); @@ -434,9 +477,9 @@ Java_com_datadoghq_profiler_JavaProfiler_blockEnter0( extern "C" DLLEXPORT void JNICALL Java_com_datadoghq_profiler_JavaProfiler_blockExit0( - JNIEnv *env, jclass unused, jlong token) { + JNIEnv *env, jclass unused, jthread thread, jlong token) { u64 block_token = static_cast(token); - if (block_token == 0) { + if (block_token == 0 || !JVMSupport::isPlatformThread(env, thread)) { return; } ProfiledThread *current = ProfiledThread::initCurrentThreadSignalSafe(); diff --git a/ddprof-lib/src/main/cpp/profiler.cpp b/ddprof-lib/src/main/cpp/profiler.cpp index 7147e298c..986a29693 100644 --- a/ddprof-lib/src/main/cpp/profiler.cpp +++ b/ddprof-lib/src/main/cpp/profiler.cpp @@ -1746,6 +1746,9 @@ Error Profiler::start(Arguments &args, bool reset) { _task_block_enabled.store( (activated & EM_WALL) && args._wall_precheck && track_unfiltered_wall, std::memory_order_release); + _task_block_monitor_events_enabled = + taskBlockEnabled() && VM::nativeMonitorEventsAvailable() && + VM::setNativeMonitorEventsEnabled(true); _state.store(RUNNING, std::memory_order_release); _start_time = time(NULL); __atomic_add_fetch(&_epoch, 1, __ATOMIC_RELAXED); @@ -1771,6 +1774,10 @@ Error Profiler::stop() { return Error("Profiler is not active"); } _task_block_enabled.store(false, std::memory_order_release); + if (_task_block_monitor_events_enabled) { + VM::setNativeMonitorEventsEnabled(false); + _task_block_monitor_events_enabled = false; + } // Order matters: disable engines first so the _enabled check inside signal // handlers will fail for any new signal delivered from now on. drain() then diff --git a/ddprof-lib/src/main/cpp/profiler.h b/ddprof-lib/src/main/cpp/profiler.h index 310fbe804..8561bb878 100644 --- a/ddprof-lib/src/main/cpp/profiler.h +++ b/ddprof-lib/src/main/cpp/profiler.h @@ -133,6 +133,7 @@ class alignas(alignof(SpinLock)) Profiler { alignas(DEFAULT_CACHE_LINE_SIZE) u64 _failures[ASGCT_FAILURE_TYPES]; bool _wall_precheck = false; std::atomic _task_block_enabled{false}; + bool _task_block_monitor_events_enabled = false; std::atomic _task_block_rotation{false}; std::atomic _task_block_inflight{0}; diff --git a/ddprof-lib/src/main/cpp/threadLocalData.h b/ddprof-lib/src/main/cpp/threadLocalData.h index fa347ec9a..80f360d5f 100644 --- a/ddprof-lib/src/main/cpp/threadLocalData.h +++ b/ddprof-lib/src/main/cpp/threadLocalData.h @@ -45,7 +45,8 @@ class ProfiledThread : public ThreadLocalData { TYPE_MASK = TYPE_JAVA_THREAD | TYPE_NOT_JAVA_THREAD }; - static constexpr u32 FLAG_PARKED = 0x4u; // next free bit after TYPE_MASK (0x1|0x2) + static constexpr u32 FLAG_PARKED = 0x4u; + static constexpr u32 FLAG_MONITOR_BLOCKED = 0x8u; // We are allowing several levels of nesting because we can be // eg. in a crash handler when wallclock signal kicks in, @@ -77,10 +78,17 @@ class ProfiledThread : public ThreadLocalData { u64 _call_trace_id; u32 _recording_epoch; u32 _misc_flags; + u64 _park_start_ticks; u64 _park_block_token; + Context _park_context; u64 _task_block_start_ticks; u64 _task_block_token; Context _task_block_context; + u64 _monitor_start_ticks; + Context _monitor_context; + u64 _monitor_blocker; + u64 _monitor_block_token; + OSThreadState _monitor_block_state; int _filter_slot_id; // Slot ID for thread filtering uint8_t _init_window; // Countdown for JVM thread init race window (PROF-13072) uint8_t _signal_depth; // Nested signal-handler depth (see SignalHandlerScope) @@ -103,8 +111,11 @@ class ProfiledThread : public ThreadLocalData { ProfiledThread(int tid) : ThreadLocalData(), _jmp_buf(nullptr), _pc(0), _sp(0), _span_id(0), _crash_depth(0), _tid(tid), _cpu_epoch(0), _wall_epoch(0), _call_trace_id(0), _recording_epoch(0), _misc_flags(0), - _park_block_token(0), _task_block_start_ticks(0), - _task_block_token(0), _task_block_context{}, _filter_slot_id(-1), + _park_start_ticks(0), _park_block_token(0), _park_context{}, + _task_block_start_ticks(0), _task_block_token(0), _task_block_context{}, + _monitor_start_ticks(0), _monitor_context{}, _monitor_blocker(0), + _monitor_block_token(0), _monitor_block_state(OSThreadState::UNKNOWN), + _filter_slot_id(-1), _init_window(0), _signal_depth(0), _otel_ctx_initialized(false), @@ -347,11 +358,24 @@ class ProfiledThread : public ThreadLocalData { _otel_local_root_span_id = 0; } - inline bool parkEnter() { - u32 prev = __atomic_fetch_or(&_misc_flags, FLAG_PARKED, __ATOMIC_RELEASE); - return (prev & FLAG_PARKED) == 0; + inline bool parkEnter(u64 start_ticks, const Context& context) { + u32 flags = __atomic_load_n(&_misc_flags, __ATOMIC_ACQUIRE); + while ((flags & FLAG_PARKED) == 0) { + _park_start_ticks = start_ticks; + _park_context = context; + if (__atomic_compare_exchange_n(&_misc_flags, &flags, + flags | FLAG_PARKED, true, + __ATOMIC_RELEASE, __ATOMIC_ACQUIRE)) { + return true; + } + } + return false; } +#ifdef UNIT_TEST + inline bool parkEnter() { return parkEnter(0, Context{}); } +#endif + inline void setParkBlockToken(u64 token) { _park_block_token = token; } @@ -374,16 +398,74 @@ class ProfiledThread : public ThreadLocalData { } // Returns false if the thread was not parked (idempotent). - inline bool parkExit(u64 &park_block_token) { + inline bool parkExit(u64& start_ticks, Context& context, + u64& park_block_token) { u32 prev = __atomic_fetch_and(&_misc_flags, ~FLAG_PARKED, __ATOMIC_ACQ_REL); if ((prev & FLAG_PARKED) == 0) { return false; } + start_ticks = _park_start_ticks; + context = _park_context; park_block_token = _park_block_token; _park_block_token = 0; return true; } +#ifdef UNIT_TEST + inline bool parkExit(u64& park_block_token) { + u64 start_ticks = 0; + Context context{}; + return parkExit(start_ticks, context, park_block_token); + } +#endif + + // Object.wait owns its interval until MonitorWaited, including monitor + // reacquisition. A nested contention callback must not overwrite that state. + inline bool monitorEnter(u64 start_ticks, const Context& context, u64 blocker, + OSThreadState state) { + u32 flags = __atomic_load_n(&_misc_flags, __ATOMIC_ACQUIRE); + if ((flags & FLAG_MONITOR_BLOCKED) != 0) return false; + _monitor_start_ticks = start_ticks; + _monitor_context = context; + _monitor_blocker = blocker; + _monitor_block_token = 0; + _monitor_block_state = state; + __atomic_fetch_or(&_misc_flags, FLAG_MONITOR_BLOCKED, __ATOMIC_RELEASE); + return true; + } + + inline void setMonitorBlockToken(u64 token) { + _monitor_block_token = token; + } + + inline u64 monitorBlockToken() const { return _monitor_block_token; } + + inline void clearMonitorBlock() { + __atomic_fetch_and(&_misc_flags, ~FLAG_MONITOR_BLOCKED, __ATOMIC_ACQ_REL); + _monitor_block_token = 0; + _monitor_block_state = OSThreadState::UNKNOWN; + } + + inline bool monitorExit(OSThreadState expected_state, u64& start_ticks, + Context& context, u64& blocker, + u64& monitor_block_token) { + u32 flags = __atomic_load_n(&_misc_flags, __ATOMIC_ACQUIRE); + if ((flags & FLAG_MONITOR_BLOCKED) == 0 || + _monitor_block_state != expected_state) { + return false; + } + u32 prev = __atomic_fetch_and(&_misc_flags, ~FLAG_MONITOR_BLOCKED, + __ATOMIC_ACQ_REL); + if ((prev & FLAG_MONITOR_BLOCKED) == 0) return false; + start_ticks = _monitor_start_ticks; + context = _monitor_context; + blocker = _monitor_blocker; + monitor_block_token = _monitor_block_token; + _monitor_block_token = 0; + _monitor_block_state = OSThreadState::UNKNOWN; + return true; + } + Context snapshotContext(size_t numAttrs); private: diff --git a/ddprof-lib/src/main/cpp/vmEntry.cpp b/ddprof-lib/src/main/cpp/vmEntry.cpp index 670c54af6..e9998f9a5 100644 --- a/ddprof-lib/src/main/cpp/vmEntry.cpp +++ b/ddprof-lib/src/main/cpp/vmEntry.cpp @@ -8,6 +8,7 @@ #include "vmEntry.h" #include "arguments.h" #include "context.h" +#include "context_api.h" #include "counters.h" #include "j9/j9Support.h" #include "jniHelper.h" @@ -19,6 +20,8 @@ #include "profiler.h" #include "safeAccess.h" #include "threadLocalData.h" +#include "taskBlockRecorder.h" +#include "tsc.h" // Pulls in vmStructs.h plus the definitions of crashProtectionActive()/cast_to() that its inline // accessors odr-use here; the light vmStructs.h alone leaves those unresolved in assertion-enabled // builds (see the note in hotspotStackFrame_aarch64.cpp). @@ -48,6 +51,8 @@ bool VM::_hotspot = false; bool VM::_zing = false; bool VM::_can_sample_objects = false; bool VM::_can_intercept_binding = false; +bool VM::_monitor_events_delegated = false; +bool VM::_native_monitor_events_available = false; bool VM::_is_adaptive_gc_boundary_flag_set = false; jvmtiExtensionFunction VM::_request_stack_trace = nullptr; @@ -67,6 +72,139 @@ static void wakeupHandler(int signo) { // Dummy handler for interrupting syscalls } +static u64 monitorBlockerHash(jvmtiEnv *jvmti, jobject object) { + if (object == NULL) return 0; + jint hash = 0; + if (jvmti->GetObjectHashCode(object, &hash) != JVMTI_ERROR_NONE) return 0; + return static_cast(static_cast(hash)); +} + +static void monitorBlockEnter(jvmtiEnv *jvmti, JNIEnv *jni, jthread thread, + jobject object, OSThreadState state) { + Profiler *profiler = Profiler::instance(); + if (!profiler->taskBlockEnabled() || + !JVMSupport::isPlatformThread(jni, thread)) { + return; + } + ProfiledThread *current = ProfiledThread::current(); + if (current == nullptr) return; + Context context = ContextApi::snapshot(); + if (context.spanId != 0) { + Counters::increment(TASK_BLOCK_SKIPPED_TRACE_CONTEXT); + return; + } + + if (!current->monitorEnter(TSC::ticks(), context, + monitorBlockerHash(jvmti, object), state)) { + u64 token = current->monitorBlockToken(); + ThreadFilter *tf = profiler->threadFilter(); + bool current_owner = false; + if (token != 0) { + ThreadFilter::SlotID slot_id = ThreadFilter::tokenSlotId(token); + BlockRunSnapshot snapshot = tf->snapshotBlockedRun(slot_id); + current_owner = current->filterSlotId() == slot_id && snapshot.active && + snapshot.owner == BlockRunOwner::JVMTI && + snapshot.generation == ThreadFilter::tokenGeneration(token); + } + if (current_owner) { + return; + } + current->clearMonitorBlock(); + if (!current->monitorEnter(TSC::ticks(), context, + monitorBlockerHash(jvmti, object), state)) { + return; + } + } + + ThreadFilter *tf = profiler->threadFilter(); + ThreadFilter::SlotID slot_id = current->filterSlotId(); + if (slot_id < 0) { + slot_id = tf->slotIdByTid(current->tid()); + if (slot_id >= 0) current->setFilterSlotId(slot_id); + } + if (!tf->allThreads() || slot_id < 0) { + current->clearMonitorBlock(); + return; + } + u64 token = + tf->enterBlockedRun(slot_id, state, BlockRunOwner::JVMTI); + if (token == 0) { + ThreadFilter::Slot *slot = tf->slotForId(slot_id); + if (slot != nullptr && slot->inContextWindow()) { + Counters::increment(TASK_BLOCK_SKIPPED_TRACE_CONTEXT); + } + current->clearMonitorBlock(); + return; + } + current->setMonitorBlockToken(token); +} + +static void monitorBlockExit(JNIEnv *jni, jthread thread, OSThreadState state) { + if (!JVMSupport::isPlatformThread(jni, thread)) return; + ProfiledThread *current = ProfiledThread::current(); + if (current == nullptr) return; + + u64 start_ticks = 0; + Context context{}; + u64 blocker = 0; + u64 token = 0; + if (!current->monitorExit(state, start_ticks, context, blocker, token) || + token == 0) { + return; + } + + Profiler *profiler = Profiler::instance(); + bool recording_enabled = profiler->taskBlockEnabled(); + bool activity = profiler->tryEnterTaskBlockActivity(); + if (!activity) profiler->waitForTaskBlockRotation(); + + ThreadFilter *tf = profiler->threadFilter(); + ThreadFilter::SlotID slot_id = ThreadFilter::tokenSlotId(token); + ThreadFilter::SlotID current_slot = current->filterSlotId(); + if (current_slot < 0) current_slot = tf->slotIdByTid(current->tid()); + BlockRunSnapshot snapshot{}; + bool exited = current_slot == slot_id && + tf->snapshotAndExitBlockedRun( + slot_id, ThreadFilter::tokenGeneration(token), &snapshot); + + if (!activity) { + Counters::increment(TASK_BLOCK_DROPPED_ROTATION); + return; + } + if (recording_enabled && exited && snapshot.context_eligible) { + recordTaskBlockIfEligible(current->tid(), thread, 0, start_ticks, + TSC::ticks(), context, blocker, 0, + snapshot.active_state, true); + } else if (recording_enabled && exited && !snapshot.context_eligible) { + Counters::increment(TASK_BLOCK_SKIPPED_TRACE_CONTEXT); + } + profiler->leaveTaskBlockActivity(); +} + +static void JNICALL MonitorContendedEnter(jvmtiEnv *jvmti, JNIEnv *jni, + jthread thread, jobject object) { + monitorBlockEnter(jvmti, jni, thread, object, OSThreadState::MONITOR_WAIT); +} + +static void JNICALL MonitorContendedEntered(jvmtiEnv *jvmti, JNIEnv *jni, + jthread thread, jobject object) { + monitorBlockExit(jni, thread, OSThreadState::MONITOR_WAIT); +} + +static void JNICALL MonitorWait(jvmtiEnv *jvmti, JNIEnv *jni, jthread thread, + jobject object, jlong timeout) { + if (!VM::monitorEventsDelegated()) { + monitorBlockEnter(jvmti, jni, thread, object, OSThreadState::OBJECT_WAIT); + } +} + +static void JNICALL MonitorWaited(jvmtiEnv *jvmti, JNIEnv *jni, jthread thread, + jobject object, jboolean timed_out) { + if (!VM::monitorEventsDelegated()) { + monitorBlockExit(jni, thread, OSThreadState::OBJECT_WAIT); + } +} + static bool isVmRuntimeEntry(const char* blob_name) { return strcmp(blob_name, "_ZNK12MemAllocator8allocateEv") == 0 || strncmp(blob_name, "_Z22post_allocation_notify", 26) == 0 @@ -440,7 +578,8 @@ bool VM::initializeRequestStackTrace() { return false; } -bool VM::initProfilerBridge(JavaVM *vm, bool attach) { +bool VM::initProfilerBridge(JavaVM *vm, bool attach, + bool delegateMonitorEvents) { TEST_LOG("VM::initProfilerBridge"); if (!initShared(vm)) { return false; @@ -479,6 +618,8 @@ bool VM::initProfilerBridge(JavaVM *vm, bool attach) { _can_intercept_binding = potential_capabilities.can_generate_native_method_bind_events && HeapUsage::needsNativeBindingInterception(); + bool can_add_monitor_events = + potential_capabilities.can_generate_monitor_events; jvmtiCapabilities capabilities = {0}; capabilities.can_generate_all_class_hook_events = 1; @@ -495,11 +636,18 @@ bool VM::initProfilerBridge(JavaVM *vm, bool attach) { capabilities.can_get_source_file_name = 1; capabilities.can_get_line_numbers = 1; capabilities.can_generate_compiled_method_load_events = 1; - capabilities.can_generate_monitor_events = 1; + capabilities.can_generate_monitor_events = can_add_monitor_events ? 1 : 0; capabilities.can_tag_objects = 1; _jvmti->AddCapabilities(&capabilities); + jvmtiCapabilities actual_capabilities = {0}; + _jvmti->GetCapabilities(&actual_capabilities); + _native_monitor_events_available = + actual_capabilities.can_generate_monitor_events; + _monitor_events_delegated = + delegateMonitorEvents && _native_monitor_events_available; + if (_hotspot) { probeJFRRequestStackTrace(); } @@ -516,6 +664,12 @@ bool VM::initProfilerBridge(JavaVM *vm, bool attach) { callbacks.SampledObjectAlloc = ObjectSampler::SampledObjectAlloc; callbacks.GarbageCollectionFinish = LivenessTracker::GarbageCollectionFinish; callbacks.NativeMethodBind = VMStructs::NativeMethodBind; + if (_native_monitor_events_available) { + callbacks.MonitorContendedEnter = MonitorContendedEnter; + callbacks.MonitorContendedEntered = MonitorContendedEntered; + callbacks.MonitorWait = MonitorWait; + callbacks.MonitorWaited = MonitorWaited; + } _jvmti->SetEventCallbacks(&callbacks, sizeof(callbacks)); _jvmti->SetEventNotificationMode(JVMTI_ENABLE, JVMTI_EVENT_VM_DEATH, NULL); @@ -571,6 +725,37 @@ bool VM::initProfilerBridge(JavaVM *vm, bool attach) { return true; } +bool VM::setNativeMonitorEventsEnabled(bool enabled) { + if (!_native_monitor_events_available) return false; + + jvmtiEventMode mode = enabled ? JVMTI_ENABLE : JVMTI_DISABLE; + jvmtiError enter = _jvmti->SetEventNotificationMode( + mode, JVMTI_EVENT_MONITOR_CONTENDED_ENTER, NULL); + jvmtiError entered = _jvmti->SetEventNotificationMode( + mode, JVMTI_EVENT_MONITOR_CONTENDED_ENTERED, NULL); + jvmtiError wait = JVMTI_ERROR_NONE; + jvmtiError waited = JVMTI_ERROR_NONE; + // When Java instrumentation owns Object.wait, do not enable the native wait + // notifications at all. Disable still addresses all four events so teardown + // is complete even if ownership was configured before this initialization. + if (!enabled || !_monitor_events_delegated) { + wait = _jvmti->SetEventNotificationMode( + mode, JVMTI_EVENT_MONITOR_WAIT, NULL); + waited = _jvmti->SetEventNotificationMode( + mode, JVMTI_EVENT_MONITOR_WAITED, NULL); + } + + if (enter == JVMTI_ERROR_NONE && entered == JVMTI_ERROR_NONE && + wait == JVMTI_ERROR_NONE && waited == JVMTI_ERROR_NONE) { + return true; + } + + Log::warn("Unable to %s JVMTI monitor events: %d/%d/%d/%d", + enabled ? "enable" : "disable", enter, entered, wait, waited); + if (enabled) setNativeMonitorEventsEnabled(false); + return false; +} + // Run late initialization when JVM is ready. May be called more than once (from // initProfilerBridge() directly, and later from the VMInit JVMTI callback, or from // initLibrary() followed by a JNI-triggered attach) -- the VMStructs init below only diff --git a/ddprof-lib/src/main/cpp/vmEntry.h b/ddprof-lib/src/main/cpp/vmEntry.h index 75725ef15..2a9d74c17 100644 --- a/ddprof-lib/src/main/cpp/vmEntry.h +++ b/ddprof-lib/src/main/cpp/vmEntry.h @@ -147,6 +147,8 @@ class VM { static bool _zing; static bool _can_sample_objects; static bool _can_intercept_binding; + static bool _monitor_events_delegated; + static bool _native_monitor_events_available; static bool _is_adaptive_gc_boundary_flag_set; static CodeCache *_libjvm; @@ -183,7 +185,8 @@ class VM { static JVM_GetManagement _getManagement; static bool initLibrary(JavaVM *vm); - static bool initProfilerBridge(JavaVM *vm, bool attach); + static bool initProfilerBridge(JavaVM *vm, bool attach, + bool delegateMonitorEvents = false); static jvmtiEnv *jvmti() { return _jvmti; } @@ -218,6 +221,13 @@ class VM { static bool canSampleObjects() { return _can_sample_objects; } + static bool monitorEventsDelegated() { return _monitor_events_delegated; } + + static bool nativeMonitorEventsAvailable() { + return _native_monitor_events_available; + } + static bool setNativeMonitorEventsEnabled(bool enabled); + static bool isZing() { return _zing; } static bool isUseAdaptiveGCBoundarySet() { diff --git a/ddprof-lib/src/main/java/com/datadoghq/profiler/JavaProfiler.java b/ddprof-lib/src/main/java/com/datadoghq/profiler/JavaProfiler.java index e451e8ff3..02a6b98e4 100644 --- a/ddprof-lib/src/main/java/com/datadoghq/profiler/JavaProfiler.java +++ b/ddprof-lib/src/main/java/com/datadoghq/profiler/JavaProfiler.java @@ -104,6 +104,25 @@ public static JavaProfiler getInstance(String scratchDir) throws IOException { * @param scratchDir directory where the bundled library will be exploded before linking; ignored when 'libLocation' is {@literal null} */ public static synchronized JavaProfiler getInstance(String libLocation, String scratchDir) throws IOException { + return getInstance(libLocation, scratchDir, false); + } + + /** + * Get a {@linkplain JavaProfiler} instance with explicit monitor-event ownership. + * + *

The first successful initialization fixes this process-wide setting because the native + * profiler is a singleton. When delegation is enabled, Java instrumentation owns + * {@code Object.wait} TaskBlock intervals and native JVMTI wait callbacks are suppressed; + * native JVMTI callbacks continue to own synchronized monitor contention. + * + * @param libLocation the path to the native library to use, or {@literal null} for the bundled library + * @param scratchDir directory where the bundled library will be exploded before linking + * @param delegateMonitorWaitEvents whether Java instrumentation owns {@code Object.wait} intervals + * @return the process-wide profiler instance + * @throws IOException if the native library cannot be loaded + */ + public static synchronized JavaProfiler getInstance(String libLocation, String scratchDir, + boolean delegateMonitorWaitEvents) throws IOException { if (instance != null) { return instance; } @@ -113,12 +132,11 @@ public static synchronized JavaProfiler getInstance(String libLocation, String s if (!result.succeeded) { throw new IOException("Failed to load Datadog Java profiler library", result.error); } - if (isVirtualThread(Thread.currentThread())) { throw new IOException("Cannot initialize profiler on a virtual thread"); } - init0(); + init0(delegateMonitorWaitEvents); instance = profiler; @@ -134,6 +152,16 @@ public static synchronized JavaProfiler getInstance(String libLocation, String s return profiler; } + /** + * Reports whether Java instrumentation, rather than JVMTI callbacks, owns + * {@code Object.wait} TaskBlock intervals. + * + * @return {@code true} when native wait callbacks are delegated + */ + public boolean isMonitorEventsDelegated() { + return monitorEventsDelegated0(); + } + /** * Stop profiling (without dumping results) * @@ -400,7 +428,7 @@ public void recordQueueTime(long startTicks, * @return {@code true} when this call owns a park interval that must be closed */ boolean parkEnter() { - return parkEnter0(); + return parkEnter0(Thread.currentThread()); } /** @@ -408,7 +436,7 @@ boolean parkEnter() { * {@code blocker} and {@code unblockingSpanId} are reserved for park instrumentation. */ void parkExit(long blocker, long unblockingSpanId) { - parkExit0(blocker, unblockingSpanId); + parkExit0(Thread.currentThread(), blocker, unblockingSpanId); } /** @@ -420,14 +448,14 @@ void parkExit(long blocker, long unblockingSpanId) { * @return an opaque token to pass to {@link #blockExit(long)}, or 0 if no state was armed */ long blockEnter(int state) { - return blockEnter0(state); + return blockEnter0(Thread.currentThread(), state); } /** * Clears a blocked interval previously armed by {@link #blockEnter(int)}. */ void blockExit(long token) { - blockExit0(token); + blockExit0(Thread.currentThread(), token); } /** @@ -483,7 +511,7 @@ public Map getDebugCounters() { return counters; } - private static native boolean init0(); + private static native boolean init0(boolean delegateMonitorWaitEvents); private native void stop0() throws IllegalStateException; private native String execute0(String command) throws IllegalArgumentException, IllegalStateException, IOException; @@ -491,6 +519,7 @@ public Map getDebugCounters() { private static native void filterThreadRemove0(); private static native int getTid0(); + private static native boolean monitorEventsDelegated0(); private static native boolean recordTrace0(long rootSpanId, String endpoint, String operation, int sizeLimit); @@ -504,13 +533,13 @@ public Map getDebugCounters() { private static native void recordQueueEnd0(long startTicks, long endTicks, String task, String scheduler, Thread origin, String queueType, int queueLength); - private static native boolean parkEnter0(); + private static native boolean parkEnter0(Thread thread); - private static native void parkExit0(long blocker, long unblockingSpanId); + private static native void parkExit0(Thread thread, long blocker, long unblockingSpanId); - private static native long blockEnter0(int state); + private static native long blockEnter0(Thread thread, int state); - private static native void blockExit0(long token); + private static native void blockExit0(Thread thread, long token); private static native long beginTaskBlock0(Thread thread); diff --git a/ddprof-lib/src/test/cpp/park_state_ut.cpp b/ddprof-lib/src/test/cpp/park_state_ut.cpp index 5a236994e..a7f558fc4 100644 --- a/ddprof-lib/src/test/cpp/park_state_ut.cpp +++ b/ddprof-lib/src/test/cpp/park_state_ut.cpp @@ -137,6 +137,79 @@ TEST(ProfiledThreadParkStateTest, ParkExitReturnsZeroTokenWhenBlockRunWasNotArme EXPECT_EQ(0ULL, park_block_token); } +TEST(ProfiledThreadParkStateTest, ParkExitReturnsEntrySnapshot) { + TestProfiledThread thread = testThread(12351); + Context entered{}; + entered.spanId = 17; + entered.rootSpanId = 18; + ASSERT_TRUE(thread->parkEnter(123, entered)); + thread->setParkBlockToken(456); + + u64 start_ticks = 0; + u64 token = 0; + Context exited{}; + ASSERT_TRUE(thread->parkExit(start_ticks, exited, token)); + EXPECT_EQ(123ULL, start_ticks); + EXPECT_EQ(456ULL, token); + EXPECT_EQ(17ULL, exited.spanId); + EXPECT_EQ(18ULL, exited.rootSpanId); +} + +TEST(ProfiledThreadMonitorStateTest, MatchingExitReturnsEntrySnapshot) { + TestProfiledThread thread = testThread(12352); + Context entered{}; + entered.spanId = 21; + ASSERT_TRUE(thread->monitorEnter( + 100, entered, 200, OSThreadState::MONITOR_WAIT)); + thread->setMonitorBlockToken(300); + + u64 start_ticks = 0; + u64 blocker = 0; + u64 token = 0; + Context exited{}; + ASSERT_TRUE(thread->monitorExit(OSThreadState::MONITOR_WAIT, start_ticks, + exited, blocker, token)); + EXPECT_EQ(100ULL, start_ticks); + EXPECT_EQ(200ULL, blocker); + EXPECT_EQ(300ULL, token); + EXPECT_EQ(21ULL, exited.spanId); +} + +TEST(ProfiledThreadMonitorStateTest, NestedContentionDoesNotReplaceObjectWait) { + TestProfiledThread thread = testThread(12353); + Context context{}; + ASSERT_TRUE(thread->monitorEnter( + 100, context, 200, OSThreadState::OBJECT_WAIT)); + thread->setMonitorBlockToken(300); + EXPECT_FALSE(thread->monitorEnter( + 400, context, 500, OSThreadState::MONITOR_WAIT)); + + u64 start_ticks = 0; + u64 blocker = 0; + u64 token = 0; + Context exited{}; + EXPECT_FALSE(thread->monitorExit(OSThreadState::MONITOR_WAIT, start_ticks, + exited, blocker, token)); + ASSERT_TRUE(thread->monitorExit(OSThreadState::OBJECT_WAIT, start_ticks, + exited, blocker, token)); + EXPECT_EQ(100ULL, start_ticks); + EXPECT_EQ(200ULL, blocker); + EXPECT_EQ(300ULL, token); +} + +TEST(ProfiledThreadMonitorStateTest, ClearAllowsRecoveryFromStaleState) { + TestProfiledThread thread = testThread(12354); + Context context{}; + ASSERT_TRUE(thread->monitorEnter( + 100, context, 200, OSThreadState::OBJECT_WAIT)); + thread->setMonitorBlockToken(300); + thread->clearMonitorBlock(); + + ASSERT_TRUE(thread->monitorEnter( + 400, context, 500, OSThreadState::MONITOR_WAIT)); + EXPECT_EQ(0ULL, thread->monitorBlockToken()); +} + TEST(WallClockOwnedBlockFilterTest, SlotStateTransitions) { ThreadFilter::Slot slot; diff --git a/ddprof-test/src/test/java/com/datadoghq/profiler/JavaProfilerApiSurfaceTest.java b/ddprof-test/src/test/java/com/datadoghq/profiler/JavaProfilerApiSurfaceTest.java index 058bd5294..89c042413 100644 --- a/ddprof-test/src/test/java/com/datadoghq/profiler/JavaProfilerApiSurfaceTest.java +++ b/ddprof-test/src/test/java/com/datadoghq/profiler/JavaProfilerApiSurfaceTest.java @@ -14,6 +14,7 @@ import static org.junit.jupiter.api.Assertions.assertFalse; import static org.junit.jupiter.api.Assertions.assertTrue; +/** Locks the supported public boundary and package-scoped producer hooks. */ public class JavaProfilerApiSurfaceTest { @Test public void taskBlockApiIsPublicButInternalHooksRemainPackageScoped() throws Exception { @@ -31,6 +32,15 @@ public void taskBlockApiIsPublicButInternalHooksRemainPackageScoped() throws Exc .getModifiers())); } + @Test + public void monitorWaitOwnershipIsExplicitPublicApi() throws Exception { + assertTrue(Modifier.isPublic(JavaProfiler.class + .getDeclaredMethod("getInstance", String.class, String.class, boolean.class) + .getModifiers())); + assertTrue(Modifier.isPublic(JavaProfiler.class + .getDeclaredMethod("isMonitorEventsDelegated").getModifiers())); + } + private static void assertNotPublic(Method method) { assertFalse(Modifier.isPublic(method.getModifiers()), method.getName() + " is an internal instrumentation hook"); diff --git a/ddprof-test/src/test/java/com/datadoghq/profiler/wallclock/JvmtiBasedMonitorTaskBlockTest.java b/ddprof-test/src/test/java/com/datadoghq/profiler/wallclock/JvmtiBasedMonitorTaskBlockTest.java new file mode 100644 index 000000000..550d9d22b --- /dev/null +++ b/ddprof-test/src/test/java/com/datadoghq/profiler/wallclock/JvmtiBasedMonitorTaskBlockTest.java @@ -0,0 +1,30 @@ +/* + * Copyright 2026, Datadog, Inc. + * SPDX-License-Identifier: Apache-2.0 + */ + +package com.datadoghq.profiler.wallclock; + +import com.datadoghq.profiler.Platform; +import java.util.Map; +import org.junit.jupiter.api.Assumptions; + +/** Verifies synchronous monitor production when delegated wall-clock stacks are enabled. */ +public class JvmtiBasedMonitorTaskBlockTest extends MonitorTaskBlockTest { + @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"); + } + + @Override + protected void withTestAssumptions() { + Assumptions.assumeTrue(Platform.isJavaVersionAtLeast(11)); + } + + @Override + protected String getProfilerCommand() { + return "wall=1ms,wallprecheck=true,jvmtistacks=true"; + } +} diff --git a/ddprof-test/src/test/java/com/datadoghq/profiler/wallclock/JvmtiBasedParkTaskBlockTest.java b/ddprof-test/src/test/java/com/datadoghq/profiler/wallclock/JvmtiBasedParkTaskBlockTest.java new file mode 100644 index 000000000..d17eae2f7 --- /dev/null +++ b/ddprof-test/src/test/java/com/datadoghq/profiler/wallclock/JvmtiBasedParkTaskBlockTest.java @@ -0,0 +1,30 @@ +/* + * Copyright 2026, Datadog, Inc. + * SPDX-License-Identifier: Apache-2.0 + */ + +package com.datadoghq.profiler.wallclock; + +import com.datadoghq.profiler.Platform; +import java.util.Map; +import org.junit.jupiter.api.Assumptions; + +/** Verifies synchronous park production when delegated wall-clock stacks are enabled. */ +public class JvmtiBasedParkTaskBlockTest extends ParkTaskBlockTest { + @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"); + } + + @Override + protected void withTestAssumptions() { + Assumptions.assumeTrue(Platform.isJavaVersionAtLeast(11)); + } + + @Override + protected String getProfilerCommand() { + return "wall=1ms,wallprecheck=true,jvmtistacks=true"; + } +} diff --git a/ddprof-test/src/test/java/com/datadoghq/profiler/wallclock/MonitorTaskBlockTest.java b/ddprof-test/src/test/java/com/datadoghq/profiler/wallclock/MonitorTaskBlockTest.java new file mode 100644 index 000000000..b2c8bf608 --- /dev/null +++ b/ddprof-test/src/test/java/com/datadoghq/profiler/wallclock/MonitorTaskBlockTest.java @@ -0,0 +1,240 @@ +/* + * Copyright 2026, Datadog, Inc. + * SPDX-License-Identifier: Apache-2.0 + */ + +package com.datadoghq.profiler.wallclock; + +import com.datadoghq.profiler.AbstractProfilerTest; +import java.lang.reflect.Method; +import java.nio.file.Files; +import java.nio.file.Path; +import java.util.concurrent.CountDownLatch; +import java.util.concurrent.TimeUnit; +import java.util.concurrent.atomic.AtomicReference; +import org.junit.jupiter.api.Test; +import org.junit.jupiter.api.Assumptions; +import org.openjdk.jmc.common.item.IItemCollection; + +import static org.junit.jupiter.api.Assertions.assertFalse; +import static org.junit.jupiter.api.Assertions.assertTrue; + +/** Verifies TaskBlock production from native JVMTI monitor callbacks. */ +public class MonitorTaskBlockTest extends AbstractProfilerTest { + @Test + public void objectWaitEmitsTaskBlockOutsideContextWindow() throws Exception { + Object monitor = new Object(); + CountDownLatch entered = new CountDownLatch(1); + AtomicReference failure = new AtomicReference<>(); + Thread worker = new Thread(() -> { + try { + synchronized (monitor) { + entered.countDown(); + monitor.wait(100); + } + } catch (Throwable t) { + failure.set(t); + } + }, "taskblock-object-wait"); + + worker.start(); + assertTrue(entered.await(5, TimeUnit.SECONDS)); + assertCompleted(worker, failure); + stopProfiler(); + + IItemCollection events = verifyEvents("datadog.TaskBlock"); + assertTaskBlockStackReference(events); + TaskBlockAssertions.assertContains(events, 0, 0, identityHash(monitor), 0); + TaskBlockAssertions.assertContainsObservedState(events, "WAITING"); + } + + @Test + public void monitorContentionEmitsTaskBlockOutsideContextWindow() throws Exception { + Object monitor = new Object(); + CountDownLatch attempting = new CountDownLatch(1); + AtomicReference failure = new AtomicReference<>(); + Thread worker; + synchronized (monitor) { + worker = new Thread(() -> { + try { + attempting.countDown(); + synchronized (monitor) { + } + } catch (Throwable t) { + failure.set(t); + } + }, "taskblock-monitor-contention"); + worker.start(); + assertTrue(attempting.await(5, TimeUnit.SECONDS)); + Thread.sleep(100); + } + + assertCompleted(worker, failure); + stopProfiler(); + + IItemCollection events = verifyEvents("datadog.TaskBlock"); + assertTaskBlockStackReference(events); + TaskBlockAssertions.assertContains(events, 0, 0, identityHash(monitor), 0); + TaskBlockAssertions.assertContainsObservedState(events, "CONTENDED"); + } + + @Test + public void contextWindowObjectWaitDoesNotEmitTaskBlock() throws Exception { + Object monitor = new Object(); + AtomicReference failure = new AtomicReference<>(); + Thread worker = new Thread(() -> { + try { + registerCurrentThreadForWallClockProfiling(); + profiler.setContext(0x4400L, 0x4401L, 0L, 0x4401L); + synchronized (monitor) { + monitor.wait(100); + } + } catch (Throwable t) { + failure.set(t); + } finally { + profiler.clearContext(); + profiler.removeThread(); + } + }, "taskblock-traced-object-wait"); + + worker.start(); + assertCompleted(worker, failure); + stopProfiler(); + + assertFalse(TaskBlockAssertions.containsBlocker( + verifyEvents("datadog.TaskBlock", false), identityHash(monitor))); + } + + @Test + public void staleWaitStateIsRecoveredAfterProfilerRestart() throws Exception { + Object waitMonitor = new Object(); + Object contentionMonitor = new Object(); + CountDownLatch waiting = new CountDownLatch(1); + CountDownLatch waitCompleted = new CountDownLatch(1); + CountDownLatch restartReady = new CountDownLatch(1); + CountDownLatch attemptingContention = new CountDownLatch(1); + AtomicReference failure = new AtomicReference<>(); + Thread worker = new Thread(() -> { + try { + synchronized (waitMonitor) { + waiting.countDown(); + waitMonitor.wait(); + } + waitCompleted.countDown(); + assertTrue(restartReady.await(5, TimeUnit.SECONDS)); + attemptingContention.countDown(); + synchronized (contentionMonitor) { + } + } catch (Throwable t) { + failure.set(t); + } + }, "taskblock-monitor-restart"); + + worker.start(); + assertTrue(waiting.await(5, TimeUnit.SECONDS)); + Thread.sleep(50); + stopProfiler(); + synchronized (waitMonitor) { + waitMonitor.notifyAll(); + } + assertTrue(waitCompleted.await(5, TimeUnit.SECONDS)); + + Path recording = Files.createTempFile("MonitorTaskBlockTest-restart-", ".jfr"); + boolean restarted = false; + try { + profiler.execute("start,wall=1ms,wallscope=all,wallprecheck=true,jfr,file=" + + recording.toAbsolutePath()); + restarted = true; + synchronized (contentionMonitor) { + restartReady.countDown(); + assertTrue(attemptingContention.await(5, TimeUnit.SECONDS)); + Thread.sleep(100); + } + assertCompleted(worker, failure); + profiler.stop(); + restarted = false; + + IItemCollection events = verifyEvents(recording, "datadog.TaskBlock", false); + assertTaskBlockStackReference(events); + assertTrue(TaskBlockAssertions.containsBlocker( + events, identityHash(contentionMonitor))); + } finally { + restartReady.countDown(); + synchronized (waitMonitor) { + waitMonitor.notifyAll(); + } + if (restarted) profiler.stop(); + worker.join(5_000); + Files.deleteIfExists(recording); + } + } + + @Test + public void virtualMonitorCallbacksDoNotEmitCarrierTaskBlocks() throws Exception { + Method startVirtualThread; + try { + startVirtualThread = Thread.class.getMethod("startVirtualThread", Runnable.class); + } catch (NoSuchMethodException unavailableBeforeJdk21) { + Assumptions.assumeTrue(false, "virtual threads require JDK 21"); + return; + } + + Object waitMonitor = new Object(); + AtomicReference failure = new AtomicReference<>(); + Thread waiter = (Thread) startVirtualThread.invoke(null, (Runnable) () -> { + try { + synchronized (waitMonitor) { + waitMonitor.wait(100); + } + } catch (Throwable t) { + failure.set(t); + } + }); + assertCompleted(waiter, failure); + + Object contentionMonitor = new Object(); + CountDownLatch attempting = new CountDownLatch(1); + Thread contender; + synchronized (contentionMonitor) { + contender = (Thread) startVirtualThread.invoke(null, (Runnable) () -> { + try { + attempting.countDown(); + synchronized (contentionMonitor) { + } + } catch (Throwable t) { + failure.set(t); + } + }); + assertTrue(attempting.await(5, TimeUnit.SECONDS)); + Thread.sleep(100); + } + assertCompleted(contender, failure); + stopProfiler(); + + IItemCollection events = verifyEvents("datadog.TaskBlock", false); + assertFalse(TaskBlockAssertions.containsBlocker(events, identityHash(waitMonitor))); + assertFalse(TaskBlockAssertions.containsBlocker(events, identityHash(contentionMonitor))); + } + + @Override + protected String getProfilerCommand() { + return "wall=1ms,wallprecheck=true"; + } + + protected void assertTaskBlockStackReference(IItemCollection events) { + TaskBlockAssertions.assertContainsStackTrace(events); + TaskBlockAssertions.assertContainsJavaType(events, "MonitorTaskBlockTest"); + TaskBlockAssertions.assertNoCorrelationId(events); + } + + private static void assertCompleted(Thread thread, AtomicReference failure) + throws InterruptedException { + thread.join(5_000); + assertFalse(thread.isAlive(), "worker did not complete"); + if (failure.get() != null) throw new AssertionError(failure.get()); + } + + private static long identityHash(Object object) { + return Integer.toUnsignedLong(System.identityHashCode(object)); + } +} diff --git a/ddprof-test/src/test/java/com/datadoghq/profiler/wallclock/ParkTaskBlockTest.java b/ddprof-test/src/test/java/com/datadoghq/profiler/wallclock/ParkTaskBlockTest.java new file mode 100644 index 000000000..36dd6790d --- /dev/null +++ b/ddprof-test/src/test/java/com/datadoghq/profiler/wallclock/ParkTaskBlockTest.java @@ -0,0 +1,115 @@ +/* + * Copyright 2026, Datadog, Inc. + * SPDX-License-Identifier: Apache-2.0 + */ + +package com.datadoghq.profiler.wallclock; + +import com.datadoghq.profiler.AbstractProfilerTest; +import com.datadoghq.profiler.ProfilerOwnedBlockHooks; +import java.lang.reflect.Method; +import java.util.concurrent.TimeUnit; +import java.util.concurrent.locks.LockSupport; +import org.junit.jupiter.api.Test; +import org.junit.jupiter.api.Assumptions; +import org.openjdk.jmc.common.item.IItemCollection; + +import static org.junit.jupiter.api.Assertions.assertFalse; + +/** Verifies TaskBlock production from Java-owned platform-thread park hooks. */ +public class ParkTaskBlockTest extends AbstractProfilerTest { + private static final long BLOCKER = 0x3102L; + private static final long UNBLOCKING_SPAN_ID = 0x3103L; + + @Test + public void platformParkEmitsTaskBlockOutsideContextWindow() { + ProfilerOwnedBlockHooks.parkEnter(profiler); + try { + parkForMillis(200); + } finally { + ProfilerOwnedBlockHooks.parkExit(profiler, BLOCKER, UNBLOCKING_SPAN_ID); + } + stopProfiler(); + + IItemCollection events = verifyEvents("datadog.TaskBlock"); + TaskBlockAssertions.assertNoAnchorFields(events); + assertTaskBlockStackReference(events); + TaskBlockAssertions.assertContains(events, 0, 0, BLOCKER, UNBLOCKING_SPAN_ID); + TaskBlockAssertions.assertContainsObservedState(events, "PARKED"); + } + + @Test + public void contextWindowParkDoesNotEmitTaskBlock() { + registerCurrentThreadForWallClockProfiling(); + profiler.setContext(0x3100L, 0x3101L, 0L, 0x3101L); + try { + ProfilerOwnedBlockHooks.parkEnter(profiler); + try { + parkForMillis(200); + } finally { + ProfilerOwnedBlockHooks.parkExit(profiler, BLOCKER, UNBLOCKING_SPAN_ID); + } + } finally { + profiler.clearContext(); + profiler.removeThread(); + } + stopProfiler(); + + assertFalse(verifyEvents("datadog.TaskBlock", false).hasItems(), + "A park inside the context window must remain ordinary wall-clock data"); + } + + @Test + public void virtualParkDoesNotMutateCarrierProducerState() throws Exception { + Method startVirtualThread; + try { + startVirtualThread = Thread.class.getMethod("startVirtualThread", Runnable.class); + } catch (NoSuchMethodException unavailableBeforeJdk21) { + Assumptions.assumeTrue(false, "virtual threads require JDK 21"); + return; + } + + long virtualBlocker = 0x3201L; + Thread virtual = (Thread) startVirtualThread.invoke(null, (Runnable) () -> { + ProfilerOwnedBlockHooks.parkEnter(profiler); + try { + parkForMillis(20); + } finally { + ProfilerOwnedBlockHooks.parkExit(profiler, virtualBlocker, 0); + } + }); + virtual.join(5_000); + assertFalse(virtual.isAlive()); + + ProfilerOwnedBlockHooks.parkEnter(profiler); + try { + parkForMillis(200); + } finally { + ProfilerOwnedBlockHooks.parkExit(profiler, BLOCKER, UNBLOCKING_SPAN_ID); + } + stopProfiler(); + + IItemCollection events = verifyEvents("datadog.TaskBlock"); + assertFalse(TaskBlockAssertions.containsBlocker(events, virtualBlocker)); + TaskBlockAssertions.assertContains(events, 0, 0, BLOCKER, UNBLOCKING_SPAN_ID); + } + + @Override + protected String getProfilerCommand() { + return "wall=1ms,wallprecheck=true"; + } + + protected void assertTaskBlockStackReference(IItemCollection events) { + TaskBlockAssertions.assertContainsStackTrace(events); + TaskBlockAssertions.assertContainsJavaType(events, "ParkTaskBlockTest"); + TaskBlockAssertions.assertNoCorrelationId(events); + } + + private static void parkForMillis(long millis) { + long deadline = System.nanoTime() + TimeUnit.MILLISECONDS.toNanos(millis); + long remaining; + while ((remaining = deadline - System.nanoTime()) > 0) { + LockSupport.parkNanos(remaining); + } + } +} diff --git a/ddprof-test/src/test/java/com/datadoghq/profiler/wallclock/TaskBlockAssertions.java b/ddprof-test/src/test/java/com/datadoghq/profiler/wallclock/TaskBlockAssertions.java index 2752966cd..e5e6f33c3 100644 --- a/ddprof-test/src/test/java/com/datadoghq/profiler/wallclock/TaskBlockAssertions.java +++ b/ddprof-test/src/test/java/com/datadoghq/profiler/wallclock/TaskBlockAssertions.java @@ -131,6 +131,17 @@ static void assertNoCorrelationId(IItemCollection events) { } } + static boolean containsBlocker(IItemCollection events, long blocker) { + for (IItemIterable iterable : events) { + IMemberAccessor accessor = BLOCKER.getAccessor(iterable.getType()); + if (accessor == null) continue; + for (IItem item : iterable) { + if (accessor.getMember(item).longValue() == blocker) return true; + } + } + return false; + } + static void assertNoAnchorFields(IItemCollection events) { for (IItemIterable iterable : events) { assertNull(ANCHOR_SAMPLE_ID.getAccessor(iterable.getType())); diff --git a/ddprof-test/src/test/java/com/datadoghq/profiler/wallclock/WallclockMitigationsCombinedTest.java b/ddprof-test/src/test/java/com/datadoghq/profiler/wallclock/WallclockMitigationsCombinedTest.java index 8ad1b1e9c..7a1679e3c 100644 --- a/ddprof-test/src/test/java/com/datadoghq/profiler/wallclock/WallclockMitigationsCombinedTest.java +++ b/ddprof-test/src/test/java/com/datadoghq/profiler/wallclock/WallclockMitigationsCombinedTest.java @@ -43,7 +43,6 @@ public void contextScopedThreadsRemainSampled() throws Exception { Thread sleeping = new Thread( () -> { - registerCurrentThreadForWallClockProfiling(); ready.countDown(); long token = ProfilerOwnedBlockHooks.blockEnter( profiler, OSTHREAD_STATE_SLEEPING); @@ -59,7 +58,6 @@ public void contextScopedThreadsRemainSampled() throws Exception { Thread parkedBusy = new Thread( () -> { - registerCurrentThreadForWallClockProfiling(); long spanId = 0x1111L; long rootSpanId = 0x2222L; profiler.setTraceContext(rootSpanId, spanId, 0, 0, -1, null, -1, null); @@ -78,7 +76,6 @@ public void contextScopedThreadsRemainSampled() throws Exception { Thread runnable = new Thread( () -> { - registerCurrentThreadForWallClockProfiling(); ready.countDown(); while (!stop.get()) { // keep runnable @@ -124,7 +121,7 @@ public void contextScopedThreadsRemainSampled() throws Exception { @Override protected String getProfilerCommand() { - return "wall=1ms,filter=0,wallprecheck=true"; + return "wall=1ms,wallprecheck=true"; } private Map samplesByThreadName() { From 856ce94703a892744bfa3ff712d244f015a39638 Mon Sep 17 00:00:00 2001 From: Paul Fournillon Date: Thu, 16 Jul 2026 23:20:43 +0200 Subject: [PATCH 2/9] test: verify JVM producers in all-thread scope --- .../JvmtiBasedMonitorTaskBlockTest.java | 2 +- .../JvmtiBasedParkTaskBlockTest.java | 2 +- .../wallclock/MonitorTaskBlockTest.java | 4 +- .../profiler/wallclock/ParkTaskBlockTest.java | 56 ++++++++++++++++++- .../WallclockMitigationsCombinedTest.java | 5 +- 5 files changed, 63 insertions(+), 6 deletions(-) diff --git a/ddprof-test/src/test/java/com/datadoghq/profiler/wallclock/JvmtiBasedMonitorTaskBlockTest.java b/ddprof-test/src/test/java/com/datadoghq/profiler/wallclock/JvmtiBasedMonitorTaskBlockTest.java index 550d9d22b..ff6df5c97 100644 --- a/ddprof-test/src/test/java/com/datadoghq/profiler/wallclock/JvmtiBasedMonitorTaskBlockTest.java +++ b/ddprof-test/src/test/java/com/datadoghq/profiler/wallclock/JvmtiBasedMonitorTaskBlockTest.java @@ -25,6 +25,6 @@ protected void withTestAssumptions() { @Override protected String getProfilerCommand() { - return "wall=1ms,wallprecheck=true,jvmtistacks=true"; + return "wall=1ms,filter=,wallprecheck=true,jvmtistacks=true"; } } diff --git a/ddprof-test/src/test/java/com/datadoghq/profiler/wallclock/JvmtiBasedParkTaskBlockTest.java b/ddprof-test/src/test/java/com/datadoghq/profiler/wallclock/JvmtiBasedParkTaskBlockTest.java index d17eae2f7..63e56c380 100644 --- a/ddprof-test/src/test/java/com/datadoghq/profiler/wallclock/JvmtiBasedParkTaskBlockTest.java +++ b/ddprof-test/src/test/java/com/datadoghq/profiler/wallclock/JvmtiBasedParkTaskBlockTest.java @@ -25,6 +25,6 @@ protected void withTestAssumptions() { @Override protected String getProfilerCommand() { - return "wall=1ms,wallprecheck=true,jvmtistacks=true"; + return "wall=1ms,filter=,wallprecheck=true,jvmtistacks=true"; } } diff --git a/ddprof-test/src/test/java/com/datadoghq/profiler/wallclock/MonitorTaskBlockTest.java b/ddprof-test/src/test/java/com/datadoghq/profiler/wallclock/MonitorTaskBlockTest.java index b2c8bf608..b05f95b6e 100644 --- a/ddprof-test/src/test/java/com/datadoghq/profiler/wallclock/MonitorTaskBlockTest.java +++ b/ddprof-test/src/test/java/com/datadoghq/profiler/wallclock/MonitorTaskBlockTest.java @@ -142,7 +142,7 @@ public void staleWaitStateIsRecoveredAfterProfilerRestart() throws Exception { Path recording = Files.createTempFile("MonitorTaskBlockTest-restart-", ".jfr"); boolean restarted = false; try { - profiler.execute("start,wall=1ms,wallscope=all,wallprecheck=true,jfr,file=" + profiler.execute("start,wall=1ms,filter=,wallprecheck=true,jfr,file=" + recording.toAbsolutePath()); restarted = true; synchronized (contentionMonitor) { @@ -218,7 +218,7 @@ public void virtualMonitorCallbacksDoNotEmitCarrierTaskBlocks() throws Exception @Override protected String getProfilerCommand() { - return "wall=1ms,wallprecheck=true"; + return "wall=1ms,filter=,wallprecheck=true"; } protected void assertTaskBlockStackReference(IItemCollection events) { diff --git a/ddprof-test/src/test/java/com/datadoghq/profiler/wallclock/ParkTaskBlockTest.java b/ddprof-test/src/test/java/com/datadoghq/profiler/wallclock/ParkTaskBlockTest.java index 36dd6790d..bf7d3528b 100644 --- a/ddprof-test/src/test/java/com/datadoghq/profiler/wallclock/ParkTaskBlockTest.java +++ b/ddprof-test/src/test/java/com/datadoghq/profiler/wallclock/ParkTaskBlockTest.java @@ -8,13 +8,17 @@ import com.datadoghq.profiler.AbstractProfilerTest; import com.datadoghq.profiler.ProfilerOwnedBlockHooks; import java.lang.reflect.Method; +import java.util.concurrent.CountDownLatch; import java.util.concurrent.TimeUnit; +import java.util.concurrent.atomic.AtomicBoolean; +import java.util.concurrent.atomic.AtomicReference; import java.util.concurrent.locks.LockSupport; import org.junit.jupiter.api.Test; import org.junit.jupiter.api.Assumptions; import org.openjdk.jmc.common.item.IItemCollection; import static org.junit.jupiter.api.Assertions.assertFalse; +import static org.junit.jupiter.api.Assertions.assertTrue; /** Verifies TaskBlock production from Java-owned platform-thread park hooks. */ public class ParkTaskBlockTest extends AbstractProfilerTest { @@ -94,9 +98,17 @@ public void virtualParkDoesNotMutateCarrierProducerState() throws Exception { TaskBlockAssertions.assertContains(events, 0, 0, BLOCKER, UNBLOCKING_SPAN_ID); } + @Test + public void platformParkSuppressesSignalsAndClearsOwnership() throws Exception { + long baseline = profiler.getDebugCounters() + .getOrDefault("wc_signals_suppressed_owned_block", 0L); + long afterFirstPark = runSuppressedPark(baseline); + runSuppressedPark(afterFirstPark); + } + @Override protected String getProfilerCommand() { - return "wall=1ms,wallprecheck=true"; + return "wall=1ms,filter=,wallprecheck=true"; } protected void assertTaskBlockStackReference(IItemCollection events) { @@ -112,4 +124,46 @@ private static void parkForMillis(long millis) { LockSupport.parkNanos(remaining); } } + + private long runSuppressedPark(long baseline) throws Exception { + CountDownLatch armed = new CountDownLatch(1); + AtomicBoolean release = new AtomicBoolean(); + AtomicReference error = new AtomicReference<>(); + Thread worker = new Thread(() -> { + try { + ProfilerOwnedBlockHooks.parkEnter(profiler); + armed.countDown(); + while (!release.get()) { + Thread.yield(); + } + } catch (Throwable t) { + error.set(t); + } finally { + ProfilerOwnedBlockHooks.parkExit(profiler, BLOCKER, UNBLOCKING_SPAN_ID); + } + }, "taskblock-park-suppression"); + + worker.start(); + assertTrue(armed.await(5, TimeUnit.SECONDS)); + try { + waitForCounterAbove("wc_signals_suppressed_owned_block", baseline, 5_000L); + } finally { + release.set(true); + } + worker.join(5_000L); + assertFalse(worker.isAlive()); + if (error.get() != null) throw new AssertionError(error.get()); + return profiler.getDebugCounters() + .getOrDefault("wc_signals_suppressed_owned_block", 0L); + } + + private void waitForCounterAbove(String name, long baseline, long timeoutMillis) + throws Exception { + long deadline = System.nanoTime() + TimeUnit.MILLISECONDS.toNanos(timeoutMillis); + while (System.nanoTime() < deadline) { + if (profiler.getDebugCounters().getOrDefault(name, 0L) > baseline) return; + Thread.sleep(10L); + } + throw new AssertionError("Counter did not increase: " + name); + } } diff --git a/ddprof-test/src/test/java/com/datadoghq/profiler/wallclock/WallclockMitigationsCombinedTest.java b/ddprof-test/src/test/java/com/datadoghq/profiler/wallclock/WallclockMitigationsCombinedTest.java index 7a1679e3c..8ad1b1e9c 100644 --- a/ddprof-test/src/test/java/com/datadoghq/profiler/wallclock/WallclockMitigationsCombinedTest.java +++ b/ddprof-test/src/test/java/com/datadoghq/profiler/wallclock/WallclockMitigationsCombinedTest.java @@ -43,6 +43,7 @@ public void contextScopedThreadsRemainSampled() throws Exception { Thread sleeping = new Thread( () -> { + registerCurrentThreadForWallClockProfiling(); ready.countDown(); long token = ProfilerOwnedBlockHooks.blockEnter( profiler, OSTHREAD_STATE_SLEEPING); @@ -58,6 +59,7 @@ public void contextScopedThreadsRemainSampled() throws Exception { Thread parkedBusy = new Thread( () -> { + registerCurrentThreadForWallClockProfiling(); long spanId = 0x1111L; long rootSpanId = 0x2222L; profiler.setTraceContext(rootSpanId, spanId, 0, 0, -1, null, -1, null); @@ -76,6 +78,7 @@ public void contextScopedThreadsRemainSampled() throws Exception { Thread runnable = new Thread( () -> { + registerCurrentThreadForWallClockProfiling(); ready.countDown(); while (!stop.get()) { // keep runnable @@ -121,7 +124,7 @@ public void contextScopedThreadsRemainSampled() throws Exception { @Override protected String getProfilerCommand() { - return "wall=1ms,wallprecheck=true"; + return "wall=1ms,filter=0,wallprecheck=true"; } private Map samplesByThreadName() { From e1e8830b0767d075fdc1598414f6d10282090ec2 Mon Sep 17 00:00:00 2001 From: Paul Fournillon Date: Fri, 24 Jul 2026 17:59:57 +0200 Subject: [PATCH 3/9] fix(taskblock): harden JVM producer initialization --- .../native/config/ConfigurationPresets.kt | 17 +++- ddprof-lib/src/main/cpp/javaApi.cpp | 79 +++++---------- ddprof-lib/src/main/cpp/vmEntry.cpp | 80 +++++++-------- ddprof-lib/src/main/cpp/vmEntry.h | 13 ++- .../com/datadoghq/profiler/JavaProfiler.java | 7 +- .../src/test/cpp/taskBlockRecorder_ut.cpp | 97 ++++++++++++++++++ .../datadoghq/profiler/ExternalLauncher.java | 84 ++++++++++++++++ .../datadoghq/profiler/JavaProfilerTest.java | 99 +++++++++++++++++++ 8 files changed, 372 insertions(+), 104 deletions(-) diff --git a/build-logic/conventions/src/main/kotlin/com/datadoghq/native/config/ConfigurationPresets.kt b/build-logic/conventions/src/main/kotlin/com/datadoghq/native/config/ConfigurationPresets.kt index c05b9c9ab..3244dd45c 100644 --- a/build-logic/conventions/src/main/kotlin/com/datadoghq/native/config/ConfigurationPresets.kt +++ b/build-logic/conventions/src/main/kotlin/com/datadoghq/native/config/ConfigurationPresets.kt @@ -1,3 +1,18 @@ +/* + * 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.native.config @@ -149,7 +164,7 @@ object ConfigurationPresets { config.compilerArgs.set( listOf("-O0", "-g", "-DDEBUG") + commonLinuxCompilerArgs(version) ) - config.linkerArgs.set(commonLinuxLinkerArgs()) + config.linkerArgs.set(commonLinuxLinkerArgs() + listOf("-Wl,-z,nodelete")) } Platform.MACOS -> { config.compilerArgs.set( diff --git a/ddprof-lib/src/main/cpp/javaApi.cpp b/ddprof-lib/src/main/cpp/javaApi.cpp index c0b93a27c..2aaa215ce 100644 --- a/ddprof-lib/src/main/cpp/javaApi.cpp +++ b/ddprof-lib/src/main/cpp/javaApi.cpp @@ -78,7 +78,20 @@ Java_com_datadoghq_profiler_JavaProfiler_init0( } // JavaVM* has already been stored when the native library was loaded so we can pass nullptr here - return VM::initProfilerBridge(nullptr, true, delegateMonitorWaitEvents); + ProfilerBridgeInitResult result = + VM::initProfilerBridge(nullptr, true, delegateMonitorWaitEvents); + if (result == ProfilerBridgeInitResult::MONITOR_EVENTS_DELEGATION_CONFLICT) { + throwNew(env, "java/lang/IllegalStateException", + "Monitor-event ownership conflicts with the profiler's " + "process-wide initialization"); + return JNI_FALSE; + } + if (result != ProfilerBridgeInitResult::SUCCESS) { + throwNew(env, "java/lang/IllegalStateException", + "Failed to initialize the profiler bridge"); + return JNI_FALSE; + } + return JNI_TRUE; } extern "C" DLLEXPORT void JNICALL @@ -144,32 +157,6 @@ Java_com_datadoghq_profiler_JavaProfiler_getSamples(JNIEnv *env, return (jlong)Profiler::instance()->total_samples(); } -// 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 @@ -191,7 +178,7 @@ JavaCritical_com_datadoghq_profiler_JavaProfiler_filterThreadAdd0() { return; } - int slot_id = ensureCurrentThreadFilterSlot(thread_filter, current); + int slot_id = thread_filter->ensureCurrentThreadSlot(current); if (unlikely(slot_id < 0)) { return; // Failed to register thread } @@ -385,7 +372,7 @@ Java_com_datadoghq_profiler_JavaProfiler_parkEnter0( ThreadFilter *tf = profiler->threadFilter(); if (context.spanId == 0 && tf->registryActive() && (profiler->taskBlockEnabled() || tf->enabled())) { - ThreadFilter::SlotID slot_id = ensureCurrentThreadFilterSlot(tf, current); + ThreadFilter::SlotID slot_id = tf->ensureCurrentThreadSlot(current); if (slot_id >= 0) { current->setParkBlockToken(tf->enterBlockedRun( slot_id, OSThreadState::CONDVAR_WAIT, BlockRunOwner::JAVA)); @@ -413,32 +400,10 @@ Java_com_datadoghq_profiler_JavaProfiler_parkExit0( return; } Profiler *profiler = Profiler::instance(); - bool recording_enabled = profiler->taskBlockEnabled(); - bool activity = profiler->tryEnterTaskBlockActivity(); - if (!activity) profiler->waitForTaskBlockRotation(); - - ThreadFilter *tf = profiler->threadFilter(); - ThreadFilter::SlotID slot_id = ThreadFilter::tokenSlotId(park_block_token); - ThreadFilter::SlotID current_slot = current->filterSlotId(); - if (current_slot < 0) current_slot = tf->slotIdByTid(current->tid()); - BlockRunSnapshot snapshot{}; - bool exited = current_slot == slot_id && - tf->snapshotAndExitBlockedRun( - slot_id, ThreadFilter::tokenGeneration(park_block_token), &snapshot); - - if (!activity) { - Counters::increment(TASK_BLOCK_DROPPED_ROTATION); - return; - } - if (recording_enabled && exited && snapshot.context_eligible) { - recordTaskBlockIfEligible( - current->tid(), thread, 1, start_ticks, TSC::ticks(), context, - static_cast(blocker), static_cast(unblockingSpanId), - snapshot.active_state, true); - } else if (recording_enabled && exited && !snapshot.context_eligible) { - Counters::increment(TASK_BLOCK_SKIPPED_TRACE_CONTEXT); - } - profiler->leaveTaskBlockActivity(); + finishTaskBlockAtExit( + current, profiler->threadFilter(), thread, 1, park_block_token, + start_ticks, context, static_cast(blocker), + static_cast(unblockingSpanId)); } static bool decodeJavaBlockState(jint state, OSThreadState &decoded) { @@ -470,7 +435,7 @@ Java_com_datadoghq_profiler_JavaProfiler_blockEnter0( if (!profiler->taskBlockEnabled() && !tf->enabled()) { return 0; } - ThreadFilter::SlotID slot_id = ensureCurrentThreadFilterSlot(tf, current); + ThreadFilter::SlotID slot_id = tf->ensureCurrentThreadSlot(current); if (slot_id < 0) return 0; return static_cast(tf->enterBlockedRun(slot_id, decoded)); } @@ -511,7 +476,7 @@ Java_com_datadoghq_profiler_JavaProfiler_beginTaskBlock0( } ThreadFilter *tf = profiler->threadFilter(); if (!tf->unfilteredWallTrackingActive()) return 0; - ThreadFilter::SlotID slot_id = ensureCurrentThreadFilterSlot(tf, current); + ThreadFilter::SlotID slot_id = tf->ensureCurrentThreadSlot(current); if (slot_id < 0) return 0; Context context = ContextApi::snapshot(); diff --git a/ddprof-lib/src/main/cpp/vmEntry.cpp b/ddprof-lib/src/main/cpp/vmEntry.cpp index e9998f9a5..e530558a8 100644 --- a/ddprof-lib/src/main/cpp/vmEntry.cpp +++ b/ddprof-lib/src/main/cpp/vmEntry.cpp @@ -16,6 +16,7 @@ #include "jvmThread.h" #include "libraries.h" #include "log.h" +#include "mutex.h" #include "os.h" #include "profiler.h" #include "safeAccess.h" @@ -55,6 +56,12 @@ bool VM::_monitor_events_delegated = false; bool VM::_native_monitor_events_available = false; bool VM::_is_adaptive_gc_boundary_flag_set = false; +// Serializes the one-time bridge installation and ownership negotiation. +// Callback readers need no synchronization because ownership is assigned +// before callbacks can be enabled and is never changed afterward. +static Mutex profiler_bridge_init_lock; +static bool profiler_bridge_initialized = false; + jvmtiExtensionFunction VM::_request_stack_trace = nullptr; jvmtiExtensionFunction VM::_init_request_stack_trace = nullptr; @@ -86,7 +93,7 @@ static void monitorBlockEnter(jvmtiEnv *jvmti, JNIEnv *jni, jthread thread, !JVMSupport::isPlatformThread(jni, thread)) { return; } - ProfiledThread *current = ProfiledThread::current(); + ProfiledThread *current = ProfiledThread::initCurrentThreadSignalSafe(); if (current == nullptr) return; Context context = ContextApi::snapshot(); if (context.spanId != 0) { @@ -101,10 +108,15 @@ static void monitorBlockEnter(jvmtiEnv *jvmti, JNIEnv *jni, jthread thread, bool current_owner = false; if (token != 0) { ThreadFilter::SlotID slot_id = ThreadFilter::tokenSlotId(token); - BlockRunSnapshot snapshot = tf->snapshotBlockedRun(slot_id); - current_owner = current->filterSlotId() == slot_id && snapshot.active && - snapshot.owner == BlockRunOwner::JVMTI && - snapshot.generation == ThreadFilter::tokenGeneration(token); + ThreadFilter::Slot *slot = current->filterSlotId() == slot_id + ? tf->activeSlotForId(slot_id, current->tid()) + : nullptr; + if (slot != nullptr) { + BlockRunSnapshot snapshot = slot->snapshotBlockRun(); + current_owner = snapshot.active && + snapshot.owner == BlockRunOwner::JVMTI && + snapshot.generation == ThreadFilter::tokenGeneration(token); + } } if (current_owner) { return; @@ -117,12 +129,8 @@ static void monitorBlockEnter(jvmtiEnv *jvmti, JNIEnv *jni, jthread thread, } ThreadFilter *tf = profiler->threadFilter(); - ThreadFilter::SlotID slot_id = current->filterSlotId(); - if (slot_id < 0) { - slot_id = tf->slotIdByTid(current->tid()); - if (slot_id >= 0) current->setFilterSlotId(slot_id); - } - if (!tf->allThreads() || slot_id < 0) { + ThreadFilter::SlotID slot_id = tf->ensureCurrentThreadSlot(current); + if (!tf->unfilteredWallTrackingActive() || slot_id < 0) { current->clearMonitorBlock(); return; } @@ -154,31 +162,8 @@ static void monitorBlockExit(JNIEnv *jni, jthread thread, OSThreadState state) { } Profiler *profiler = Profiler::instance(); - bool recording_enabled = profiler->taskBlockEnabled(); - bool activity = profiler->tryEnterTaskBlockActivity(); - if (!activity) profiler->waitForTaskBlockRotation(); - - ThreadFilter *tf = profiler->threadFilter(); - ThreadFilter::SlotID slot_id = ThreadFilter::tokenSlotId(token); - ThreadFilter::SlotID current_slot = current->filterSlotId(); - if (current_slot < 0) current_slot = tf->slotIdByTid(current->tid()); - BlockRunSnapshot snapshot{}; - bool exited = current_slot == slot_id && - tf->snapshotAndExitBlockedRun( - slot_id, ThreadFilter::tokenGeneration(token), &snapshot); - - if (!activity) { - Counters::increment(TASK_BLOCK_DROPPED_ROTATION); - return; - } - if (recording_enabled && exited && snapshot.context_eligible) { - recordTaskBlockIfEligible(current->tid(), thread, 0, start_ticks, - TSC::ticks(), context, blocker, 0, - snapshot.active_state, true); - } else if (recording_enabled && exited && !snapshot.context_eligible) { - Counters::increment(TASK_BLOCK_SKIPPED_TRACE_CONTEXT); - } - profiler->leaveTaskBlockActivity(); + finishTaskBlockAtExit(current, profiler->threadFilter(), thread, 0, token, + start_ticks, context, blocker, 0); } static void JNICALL MonitorContendedEnter(jvmtiEnv *jvmti, JNIEnv *jni, @@ -578,16 +563,25 @@ bool VM::initializeRequestStackTrace() { return false; } -bool VM::initProfilerBridge(JavaVM *vm, bool attach, - bool delegateMonitorEvents) { +ProfilerBridgeInitResult VM::initProfilerBridge(JavaVM *vm, bool attach, + bool delegateMonitorEvents) { + MutexLocker init_locker(profiler_bridge_init_lock); + if (profiler_bridge_initialized) { + bool requested_delegation = + delegateMonitorEvents && _native_monitor_events_available; + return requested_delegation == _monitor_events_delegated + ? ProfilerBridgeInitResult::SUCCESS + : ProfilerBridgeInitResult::MONITOR_EVENTS_DELEGATION_CONFLICT; + } + TEST_LOG("VM::initProfilerBridge"); if (!initShared(vm)) { - return false; + return ProfilerBridgeInitResult::FAILURE; } CodeCache *lib = openJvmLibrary(); if (lib == nullptr) { - return false; + return ProfilerBridgeInitResult::FAILURE; } // Under Agent_OnLoad (attach == false), this is the first native entry point and @@ -722,7 +716,8 @@ bool VM::initProfilerBridge(JavaVM *vm, bool attach, OS::installSignalHandler(WAKEUP_SIGNAL, NULL, wakeupHandler); - return true; + profiler_bridge_initialized = true; + return ProfilerBridgeInitResult::SUCCESS; } bool VM::setNativeMonitorEventsEnabled(bool enabled) { @@ -887,7 +882,8 @@ Agent_OnLoad(JavaVM* vm, char* options, void* reserved) { return ARGUMENTS_ERROR; } - if (!VM::initProfilerBridge(vm, false)) { + if (VM::initProfilerBridge(vm, false) != + ProfilerBridgeInitResult::SUCCESS) { Log::error("JVM does not support Tool Interface"); return COMMAND_ERROR; } diff --git a/ddprof-lib/src/main/cpp/vmEntry.h b/ddprof-lib/src/main/cpp/vmEntry.h index 2a9d74c17..1d41a654a 100644 --- a/ddprof-lib/src/main/cpp/vmEntry.h +++ b/ddprof-lib/src/main/cpp/vmEntry.h @@ -132,6 +132,15 @@ class JavaVersionAccess { static int get_hotspot_version(char* prop_value); }; +// The profiler bridge is process-wide and initialized exactly once. Later Java +// API initialization may reuse it only with the same effective Object.wait +// ownership. +enum class ProfilerBridgeInitResult { + SUCCESS, + FAILURE, + MONITOR_EVENTS_DELEGATION_CONFLICT, +}; + class VM { friend class VMTestAccessor; @@ -185,8 +194,8 @@ class VM { static JVM_GetManagement _getManagement; static bool initLibrary(JavaVM *vm); - static bool initProfilerBridge(JavaVM *vm, bool attach, - bool delegateMonitorEvents = false); + static ProfilerBridgeInitResult initProfilerBridge( + JavaVM *vm, bool attach, bool delegateMonitorEvents = false); static jvmtiEnv *jvmti() { return _jvmti; } diff --git a/ddprof-lib/src/main/java/com/datadoghq/profiler/JavaProfiler.java b/ddprof-lib/src/main/java/com/datadoghq/profiler/JavaProfiler.java index 02a6b98e4..506623f9f 100644 --- a/ddprof-lib/src/main/java/com/datadoghq/profiler/JavaProfiler.java +++ b/ddprof-lib/src/main/java/com/datadoghq/profiler/JavaProfiler.java @@ -110,8 +110,9 @@ public static synchronized JavaProfiler getInstance(String libLocation, String s /** * Get a {@linkplain JavaProfiler} instance with explicit monitor-event ownership. * - *

The first successful initialization fixes this process-wide setting because the native - * profiler is a singleton. When delegation is enabled, Java instrumentation owns + *

The first successful native bridge initialization fixes this process-wide setting because + * the native profiler is a singleton. This may occur during {@code -agentpath} startup before + * this method is called. When delegation is enabled, Java instrumentation owns * {@code Object.wait} TaskBlock intervals and native JVMTI wait callbacks are suppressed; * native JVMTI callbacks continue to own synchronized monitor contention. * @@ -120,6 +121,8 @@ public static synchronized JavaProfiler getInstance(String libLocation, String s * @param delegateMonitorWaitEvents whether Java instrumentation owns {@code Object.wait} intervals * @return the process-wide profiler instance * @throws IOException if the native library cannot be loaded + * @throws IllegalStateException if monitor ownership conflicts with an earlier native bridge + * initialization */ public static synchronized JavaProfiler getInstance(String libLocation, String scratchDir, boolean delegateMonitorWaitEvents) throws IOException { diff --git a/ddprof-lib/src/test/cpp/taskBlockRecorder_ut.cpp b/ddprof-lib/src/test/cpp/taskBlockRecorder_ut.cpp index 9745609ad..3c90c672f 100644 --- a/ddprof-lib/src/test/cpp/taskBlockRecorder_ut.cpp +++ b/ddprof-lib/src/test/cpp/taskBlockRecorder_ut.cpp @@ -190,6 +190,103 @@ TEST_F(TaskBlockRecorderTest, RotationRejectsEndWithoutStrandingLifecycle) { EXPECT_EQ(1, Counters::getCounter(TASK_BLOCK_DROPPED_ROTATION)); } +TEST_F(TaskBlockRecorderTest, RotationRejectsParkExitWithoutBlockingOrStranding) { + constexpr int tid = 12346; + ThreadFilter filter; + filter.init("", true); + ThreadFilter::SlotID slot_id = filter.registerThread(tid); + ASSERT_GE(slot_id, 0); + + std::unique_ptr current( + ProfiledThread::forTid(tid), ProfiledThread::deleteForTest); + current->setFilterSlotId(slot_id); + Context context{}; + ASSERT_TRUE(current->parkEnter(TSC::ticks(), context)); + u64 token = filter.enterBlockedRun( + slot_id, OSThreadState::CONDVAR_WAIT, BlockRunOwner::JAVA); + ASSERT_NE(0ULL, token); + current->setParkBlockToken(token); + + Profiler* profiler = Profiler::instance(); + profiler->beginTaskBlockRotationForTest(); + std::future result = std::async(std::launch::async, [&]() { + u64 start_ticks = 0; + u64 exit_token = 0; + Context exit_context{}; + if (!current->parkExit(start_ticks, exit_context, exit_token)) return true; + return finishTaskBlockAtExit( + current.get(), &filter, nullptr, 1, exit_token, start_ticks, + exit_context, 0, 0); + }); + + EXPECT_EQ(std::future_status::ready, + result.wait_for(std::chrono::seconds(1))); + ThreadFilter::Slot* slot = filter.slotForId(slot_id); + ASSERT_NE(nullptr, slot); + EXPECT_EQ(BlockRunOwner::NONE, slot->activeBlockOwner()); + EXPECT_EQ(OSThreadState::UNKNOWN, slot->activeBlockState()); + EXPECT_TRUE(current->parkEnter(TSC::ticks(), context)); + u64 ignored_ticks = 0; + u64 ignored_token = 0; + Context ignored_context{}; + EXPECT_TRUE(current->parkExit( + ignored_ticks, ignored_context, ignored_token)); + + profiler->endTaskBlockRotationForTest(); + EXPECT_FALSE(result.get()); + EXPECT_EQ(1, Counters::getCounter(TASK_BLOCK_DROPPED_ROTATION)); +} + +TEST_F(TaskBlockRecorderTest, + RotationRejectsMonitorExitWithoutBlockingOrStranding) { + constexpr int tid = 12347; + ThreadFilter filter; + filter.init("", true); + ThreadFilter::SlotID slot_id = filter.registerThread(tid); + ASSERT_GE(slot_id, 0); + + std::unique_ptr current( + ProfiledThread::forTid(tid), ProfiledThread::deleteForTest); + current->setFilterSlotId(slot_id); + Context context{}; + ASSERT_TRUE(current->monitorEnter( + TSC::ticks(), context, 7, OSThreadState::OBJECT_WAIT)); + u64 token = filter.enterBlockedRun( + slot_id, OSThreadState::OBJECT_WAIT, BlockRunOwner::JVMTI); + ASSERT_NE(0ULL, token); + current->setMonitorBlockToken(token); + + Profiler* profiler = Profiler::instance(); + profiler->beginTaskBlockRotationForTest(); + std::future result = std::async(std::launch::async, [&]() { + u64 start_ticks = 0; + u64 blocker = 0; + u64 exit_token = 0; + Context exit_context{}; + if (!current->monitorExit(OSThreadState::OBJECT_WAIT, start_ticks, + exit_context, blocker, exit_token)) { + return true; + } + return finishTaskBlockAtExit( + current.get(), &filter, nullptr, 0, exit_token, start_ticks, + exit_context, blocker, 0); + }); + + EXPECT_EQ(std::future_status::ready, + result.wait_for(std::chrono::seconds(1))); + ThreadFilter::Slot* slot = filter.slotForId(slot_id); + ASSERT_NE(nullptr, slot); + EXPECT_EQ(BlockRunOwner::NONE, slot->activeBlockOwner()); + EXPECT_EQ(OSThreadState::UNKNOWN, slot->activeBlockState()); + EXPECT_TRUE(current->monitorEnter( + TSC::ticks(), context, 8, OSThreadState::MONITOR_WAIT)); + current->clearMonitorBlock(); + + profiler->endTaskBlockRotationForTest(); + EXPECT_FALSE(result.get()); + EXPECT_EQ(1, Counters::getCounter(TASK_BLOCK_DROPPED_ROTATION)); +} + TEST_F(TaskBlockRecorderTest, StackCaptureFailureIsCountedAndActivityReleased) { g_record_result.store(Profiler::TaskBlockRecordResult::STACK_CAPTURE_FAILED, std::memory_order_release); diff --git a/ddprof-test/src/test/java/com/datadoghq/profiler/ExternalLauncher.java b/ddprof-test/src/test/java/com/datadoghq/profiler/ExternalLauncher.java index 695412dcb..8120e4460 100644 --- a/ddprof-test/src/test/java/com/datadoghq/profiler/ExternalLauncher.java +++ b/ddprof-test/src/test/java/com/datadoghq/profiler/ExternalLauncher.java @@ -9,7 +9,14 @@ import java.lang.management.ManagementFactory; import java.lang.management.ThreadMXBean; import java.lang.reflect.Method; +import java.nio.file.Files; +import java.nio.file.Path; import java.util.Random; +import java.util.concurrent.CountDownLatch; +import java.util.concurrent.ExecutorService; +import java.util.concurrent.Executors; +import java.util.concurrent.Future; +import java.util.concurrent.TimeUnit; import java.util.concurrent.atomic.LongAdder; /** @@ -23,6 +30,10 @@ *

  • profiler [comma delimited profiler command list] - starts the profiler
    • *
  • profiler-work: [comma delimited profiler command list] - starts the profiler and runs a CPU-intensive task
    • *
  • profiler-virtual-thread - calls {@link JavaProfiler#getInstance()} for the first time from a virtual thread
    • + *
  • profiler-agent-compatible - reuses native monitor ownership after agent initialization
    • + *
  • profiler-delegation-conflict - requests delegated monitor ownership after agent initialization
    • + *
  • profiler-preexisting-monitor-wait - exercises Object.wait on a thread created before profiler initialization
    • + *
  • profiler-preexisting-monitor-contention - exercises monitor contention on a thread created before profiler initialization
    • * */ public class ExternalLauncher { @@ -38,6 +49,63 @@ private static Thread startVirtualThread(Runnable task) throws Exception { return (Thread) start.invoke(builder, task); } + /** Runs one native monitor callback lifecycle on a platform thread created before JNI load. */ + private static void runPreExistingMonitorCallback(boolean contention) throws Exception { + ExecutorService executor = Executors.newSingleThreadExecutor(task -> { + Thread thread = new Thread(task, "preexisting-monitor-callback"); + thread.setDaemon(true); + return thread; + }); + executor.submit(Thread::currentThread).get(5, TimeUnit.SECONDS); + + Path recording = Files.createTempFile("preexisting-monitor-callback", ".jfr"); + JavaProfiler profiler = null; + boolean started = false; + try { + profiler = JavaProfiler.getInstance(); + profiler.execute("start,wall=1ms,filter=,wallprecheck=true,jfr,file=" + + recording.toAbsolutePath()); + started = true; + long before = profiler.getDebugCounters().getOrDefault("task_block_emitted", 0L); + Object monitor = new Object(); + + if (contention) { + CountDownLatch attempting = new CountDownLatch(1); + Future blocked; + synchronized (monitor) { + blocked = executor.submit(() -> { + attempting.countDown(); + synchronized (monitor) { + // Acquiring the monitor completes the contended interval. + } + }); + if (!attempting.await(5, TimeUnit.SECONDS)) { + throw new IllegalStateException("Worker did not attempt monitor entry"); + } + Thread.sleep(100L); + } + blocked.get(5, TimeUnit.SECONDS); + } else { + executor.submit(() -> { + synchronized (monitor) { + monitor.wait(100L); + } + return null; + }).get(5, TimeUnit.SECONDS); + } + + long emitted = profiler.getDebugCounters().getOrDefault("task_block_emitted", 0L) - before; + System.out.println("[preexisting-monitor-events] " + emitted); + } finally { + if (started) { + profiler.stop(); + } + executor.shutdownNow(); + executor.awaitTermination(5, TimeUnit.SECONDS); + Files.deleteIfExists(recording); + } + } + public static void main(String[] args) throws Exception { Thread worker = null; try { @@ -58,6 +126,22 @@ public static void main(String[] args) throws Exception { } }); vt.join(); + } else if (args[0].equals("profiler-delegation-conflict")) { + String libraryPath = System.getProperty("ddprof.test.agent.path"); + try { + JavaProfiler.getInstance(libraryPath, null, true); + System.out.println("[delegation-conflict-missed]"); + } catch (IllegalStateException expected) { + System.out.println("[delegation-conflict] " + expected.getMessage()); + } + } else if (args[0].equals("profiler-agent-compatible")) { + String libraryPath = System.getProperty("ddprof.test.agent.path"); + JavaProfiler profiler = JavaProfiler.getInstance(libraryPath, null, false); + System.out.println("[agent-compatible] " + profiler.isMonitorEventsDelegated()); + } else if (args[0].equals("profiler-preexisting-monitor-wait")) { + runPreExistingMonitorCallback(false); + } else if (args[0].equals("profiler-preexisting-monitor-contention")) { + runPreExistingMonitorCallback(true); } else if (args[0].equals("profiler")) { JavaProfiler instance = JavaProfiler.getInstance(); if (args.length == 2) { diff --git a/ddprof-test/src/test/java/com/datadoghq/profiler/JavaProfilerTest.java b/ddprof-test/src/test/java/com/datadoghq/profiler/JavaProfilerTest.java index 2023c4757..154ffec23 100644 --- a/ddprof-test/src/test/java/com/datadoghq/profiler/JavaProfilerTest.java +++ b/ddprof-test/src/test/java/com/datadoghq/profiler/JavaProfilerTest.java @@ -7,8 +7,10 @@ import org.junit.jupiter.api.Test; +import java.io.InputStream; import java.nio.file.Files; import java.nio.file.Path; +import java.nio.file.StandardCopyOption; import java.util.ArrayList; import java.util.Collections; import java.util.List; @@ -18,12 +20,46 @@ import java.util.concurrent.atomic.AtomicReference; import java.util.concurrent.locks.Lock; import java.util.concurrent.locks.LockSupport; +import java.util.function.Function; import static org.junit.jupiter.api.Assertions.*; import static org.junit.jupiter.api.Assumptions.assumeFalse; import static org.junit.jupiter.api.Assumptions.assumeTrue; public class JavaProfilerTest extends AbstractProcessProfilerTest { + /** Extracts the packaged native library so a child JVM can load it through {@code -agentpath}. */ + private static Path extractProfilerLibrary() throws Exception { + OperatingSystem os = OperatingSystem.current(); + String extension = os == OperatingSystem.macos ? "dylib" : "so"; + String qualifier = os == OperatingSystem.linux && os.isMusl() ? "-musl" : ""; + String resource = "/META-INF/native-libs/" + os.name().toLowerCase() + "-" + + Arch.current().name().toLowerCase() + qualifier + "/libjavaProfiler." + extension; + Path library = Files.createTempFile("libjavaProfiler-agent-", "." + extension); + try (InputStream input = JavaProfiler.class.getResourceAsStream(resource)) { + assertNotNull(input, "Profiler library resource not found: " + resource); + Files.copy(input, library, StandardCopyOption.REPLACE_EXISTING); + } + return library; + } + + /** Launches a child JVM whose profiler bridge is initialized before Java application startup. */ + private LaunchResult launchWithProfilerAgent( + String target, Function onStdoutLine) throws Exception { + Path library = extractProfilerLibrary(); + Path recording = Files.createTempFile("agent-initialization-", ".jfr"); + try { + List jvmArgs = new ArrayList<>(); + jvmArgs.add("-agentpath:" + library.toAbsolutePath() + + "=start,wall=10ms,filter=,wallprecheck=true,jfr,file=" + + recording.toAbsolutePath()); + jvmArgs.add("-Dddprof.test.agent.path=" + library.toAbsolutePath()); + return launch(target, jvmArgs, "", onStdoutLine, null); + } finally { + Files.deleteIfExists(recording); + Files.deleteIfExists(library); + } + } + @Test void sanityInitailizationTest() throws Exception { String config = System.getProperty("ddprof_test.config"); @@ -134,6 +170,69 @@ void getInstanceFromVirtualThreadThrowsIOException() throws Exception { "Expected IOException from getInstance() on a virtual thread, got: " + result); } + @Test + void compatibleLateJavaInitializationReusesAgentBridge() throws Exception { + AtomicReference resultLine = new AtomicReference<>(); + LaunchResult result = launchWithProfilerAgent("profiler-agent-compatible", line -> { + if (line.startsWith("[agent-compatible]")) { + resultLine.set(line); + return LineConsumerResult.STOP; + } + return LineConsumerResult.CONTINUE; + }); + + assertTrue(result.inTime); + assertEquals(0, result.exitCode); + assertEquals("[agent-compatible] false", resultLine.get()); + } + + @Test + void conflictingLateMonitorDelegationIsRejected() throws Exception { + AtomicReference resultLine = new AtomicReference<>(); + LaunchResult result = launchWithProfilerAgent("profiler-delegation-conflict", line -> { + if (line.startsWith("[delegation-conflict")) { + resultLine.set(line); + return LineConsumerResult.STOP; + } + return LineConsumerResult.CONTINUE; + }); + + assertTrue(result.inTime); + assertEquals(0, result.exitCode); + assertNotNull(resultLine.get(), "Late delegation request did not report a result"); + assertTrue(resultLine.get().startsWith("[delegation-conflict]"), + "Expected ownership conflict, got: " + resultLine.get()); + } + + @Test + void preExistingThreadObjectWaitUsesNativeMonitorCallbacks() throws Exception { + assertPreExistingMonitorCallback("profiler-preexisting-monitor-wait"); + } + + @Test + void preExistingThreadContentionUsesNativeMonitorCallbacks() throws Exception { + assertPreExistingMonitorCallback("profiler-preexisting-monitor-contention"); + } + + /** Verifies that a pre-JNI-load worker emits a TaskBlock through its first monitor callback. */ + private void assertPreExistingMonitorCallback(String target) throws Exception { + AtomicReference resultLine = new AtomicReference<>(); + LaunchResult result = launch(target, Collections.emptyList(), "", line -> { + if (line.startsWith("[preexisting-monitor-events]")) { + resultLine.set(line); + return LineConsumerResult.STOP; + } + return LineConsumerResult.CONTINUE; + }, null); + + assertTrue(result.inTime); + assertEquals(0, result.exitCode); + assertNotNull(resultLine.get(), "Pre-existing monitor callback did not report a result"); + long emitted = Long.parseLong(resultLine.get().substring( + "[preexisting-monitor-events] ".length())); + assertTrue(emitted > 0, "Pre-existing thread emitted no native monitor TaskBlock event"); + } + @Test void vmStackwalkerCrashRecoveryTest() throws Exception { assumeFalse(Platform.isJ9() || Platform.isZing()); // J9 and Zing do not support vmstructs From 64530933b34d3c2e066921f676cc632396a69813 Mon Sep 17 00:00:00 2001 From: Paul Fournillon Date: Tue, 28 Jul 2026 10:41:46 +0200 Subject: [PATCH 4/9] fix: address review suggestions --- ddprof-lib/src/main/cpp/jvmSupport.cpp | 17 ++++++++++-- .../com/datadoghq/profiler/JavaProfiler.java | 5 ++++ .../datadoghq/profiler/ExternalLauncher.java | 12 +++++++++ .../datadoghq/profiler/JavaProfilerTest.java | 27 +++++++++++++++++++ 4 files changed, 59 insertions(+), 2 deletions(-) diff --git a/ddprof-lib/src/main/cpp/jvmSupport.cpp b/ddprof-lib/src/main/cpp/jvmSupport.cpp index 86dff230c..30dce551c 100644 --- a/ddprof-lib/src/main/cpp/jvmSupport.cpp +++ b/ddprof-lib/src/main/cpp/jvmSupport.cpp @@ -6,6 +6,7 @@ #include "jvmSupport.h" #include "asyncSampleMutex.h" +#include "common.h" #include "frames.h" #include "os.h" #include "profiler.h" @@ -16,6 +17,8 @@ #include +#include + using JniFunction = void (JNICALL*)(); using IsVirtualThreadFunction = jboolean (JNICALL*)(JNIEnv*, jobject); @@ -41,11 +44,21 @@ bool JVMSupport::isPlatformThread(JNIEnv* jni, jthread thread) { const JniFunction* functions = reinterpret_cast(jni->functions); + if (functions == nullptr) return false; IsVirtualThreadFunction is_virtual_thread = reinterpret_cast( functions[IS_VIRTUAL_THREAD_INDEX]); - return is_virtual_thread != nullptr && - is_virtual_thread(jni, thread) == JNI_FALSE; + if (is_virtual_thread == nullptr) { + static std::atomic warning_emitted{false}; + bool expected = false; + if (warning_emitted.compare_exchange_strong(expected, true, + std::memory_order_relaxed)) { + LOG_WARN("JNI version 19 or later does not expose IsVirtualThread; " + "JVM producer callbacks will be ignored"); + } + return false; + } + return is_virtual_thread(jni, thread) == JNI_FALSE; } bool JVMSupport::initialize() { diff --git a/ddprof-lib/src/main/java/com/datadoghq/profiler/JavaProfiler.java b/ddprof-lib/src/main/java/com/datadoghq/profiler/JavaProfiler.java index 506623f9f..55535505b 100644 --- a/ddprof-lib/src/main/java/com/datadoghq/profiler/JavaProfiler.java +++ b/ddprof-lib/src/main/java/com/datadoghq/profiler/JavaProfiler.java @@ -127,6 +127,11 @@ public static synchronized JavaProfiler getInstance(String libLocation, String s public static synchronized JavaProfiler getInstance(String libLocation, String scratchDir, boolean delegateMonitorWaitEvents) throws IOException { if (instance != null) { + if (monitorEventsDelegated0() != delegateMonitorWaitEvents) { + throw new IllegalStateException( + "Monitor-event ownership conflicts with the profiler's " + + "process-wide initialization"); + } return instance; } diff --git a/ddprof-test/src/test/java/com/datadoghq/profiler/ExternalLauncher.java b/ddprof-test/src/test/java/com/datadoghq/profiler/ExternalLauncher.java index 8120e4460..c193daf58 100644 --- a/ddprof-test/src/test/java/com/datadoghq/profiler/ExternalLauncher.java +++ b/ddprof-test/src/test/java/com/datadoghq/profiler/ExternalLauncher.java @@ -32,6 +32,7 @@ *
  • profiler-virtual-thread - calls {@link JavaProfiler#getInstance()} for the first time from a virtual thread
    • *
  • profiler-agent-compatible - reuses native monitor ownership after agent initialization
    • *
  • profiler-delegation-conflict - requests delegated monitor ownership after agent initialization
    • + *
  • profiler-java-delegation-conflict:<initial>:<requested> - verifies conflicting Java singleton ownership requests
    • *
  • profiler-preexisting-monitor-wait - exercises Object.wait on a thread created before profiler initialization
    • *
  • profiler-preexisting-monitor-contention - exercises monitor contention on a thread created before profiler initialization
    • * @@ -134,6 +135,17 @@ public static void main(String[] args) throws Exception { } catch (IllegalStateException expected) { System.out.println("[delegation-conflict] " + expected.getMessage()); } + } else if (args[0].startsWith("profiler-java-delegation-conflict:")) { + String[] delegationModes = args[0].split(":"); + boolean initialDelegation = Boolean.parseBoolean(delegationModes[1]); + boolean requestedDelegation = Boolean.parseBoolean(delegationModes[2]); + JavaProfiler.getInstance(null, null, initialDelegation); + try { + JavaProfiler.getInstance(null, null, requestedDelegation); + System.out.println("[java-delegation-conflict-missed]"); + } catch (IllegalStateException expected) { + System.out.println("[java-delegation-conflict] " + expected.getMessage()); + } } else if (args[0].equals("profiler-agent-compatible")) { String libraryPath = System.getProperty("ddprof.test.agent.path"); JavaProfiler profiler = JavaProfiler.getInstance(libraryPath, null, false); diff --git a/ddprof-test/src/test/java/com/datadoghq/profiler/JavaProfilerTest.java b/ddprof-test/src/test/java/com/datadoghq/profiler/JavaProfilerTest.java index 154ffec23..9c0004861 100644 --- a/ddprof-test/src/test/java/com/datadoghq/profiler/JavaProfilerTest.java +++ b/ddprof-test/src/test/java/com/datadoghq/profiler/JavaProfilerTest.java @@ -204,6 +204,33 @@ void conflictingLateMonitorDelegationIsRejected() throws Exception { "Expected ownership conflict, got: " + resultLine.get()); } + @Test + void conflictingJavaSingletonMonitorDelegationIsRejected() throws Exception { + assertJavaSingletonDelegationConflict(false, true); + assertJavaSingletonDelegationConflict(true, false); + } + + /** Launches a fresh JVM and verifies that a second ownership mode is rejected. */ + private void assertJavaSingletonDelegationConflict(boolean initialDelegation, + boolean requestedDelegation) throws Exception { + AtomicReference resultLine = new AtomicReference<>(); + LaunchResult result = launch( + "profiler-java-delegation-conflict:" + initialDelegation + ":" + requestedDelegation, + Collections.emptyList(), "", line -> { + if (line.startsWith("[java-delegation-conflict")) { + resultLine.set(line); + return LineConsumerResult.STOP; + } + return LineConsumerResult.CONTINUE; + }, null); + + assertTrue(result.inTime); + assertEquals(0, result.exitCode); + assertNotNull(resultLine.get(), "Conflicting Java singleton request did not report a result"); + assertTrue(resultLine.get().startsWith("[java-delegation-conflict]"), + "Expected ownership conflict, got: " + resultLine.get()); + } + @Test void preExistingThreadObjectWaitUsesNativeMonitorCallbacks() throws Exception { assertPreExistingMonitorCallback("profiler-preexisting-monitor-wait"); From 4823197540d30cc50d37733757e24ca811672222 Mon Sep 17 00:00:00 2001 From: Paul Fournillon Date: Thu, 30 Jul 2026 14:22:16 +0200 Subject: [PATCH 5/9] refactor(taskblock): clarify monitor-wait delegation naming --- ddprof-lib/src/main/cpp/javaApi.cpp | 4 ++-- ddprof-lib/src/main/cpp/vmEntry.cpp | 18 +++++++++--------- ddprof-lib/src/main/cpp/vmEntry.h | 8 +++++--- .../com/datadoghq/profiler/JavaProfiler.java | 15 ++++++++------- .../datadoghq/profiler/ExternalLauncher.java | 3 ++- .../profiler/JavaProfilerApiSurfaceTest.java | 2 +- 6 files changed, 27 insertions(+), 23 deletions(-) diff --git a/ddprof-lib/src/main/cpp/javaApi.cpp b/ddprof-lib/src/main/cpp/javaApi.cpp index 2aaa215ce..6548797ba 100644 --- a/ddprof-lib/src/main/cpp/javaApi.cpp +++ b/ddprof-lib/src/main/cpp/javaApi.cpp @@ -109,9 +109,9 @@ Java_com_datadoghq_profiler_JavaProfiler_getTid0(JNIEnv *env, jclass unused) { } extern "C" DLLEXPORT jboolean JNICALL -Java_com_datadoghq_profiler_JavaProfiler_monitorEventsDelegated0( +Java_com_datadoghq_profiler_JavaProfiler_monitorWaitEventsDelegated0( JNIEnv *env, jclass unused) { - return VM::monitorEventsDelegated(); + return VM::monitorWaitEventsDelegated(); } extern "C" DLLEXPORT jstring JNICALL diff --git a/ddprof-lib/src/main/cpp/vmEntry.cpp b/ddprof-lib/src/main/cpp/vmEntry.cpp index e530558a8..eaea90508 100644 --- a/ddprof-lib/src/main/cpp/vmEntry.cpp +++ b/ddprof-lib/src/main/cpp/vmEntry.cpp @@ -52,7 +52,7 @@ bool VM::_hotspot = false; bool VM::_zing = false; bool VM::_can_sample_objects = false; bool VM::_can_intercept_binding = false; -bool VM::_monitor_events_delegated = false; +bool VM::_monitor_wait_events_delegated = false; bool VM::_native_monitor_events_available = false; bool VM::_is_adaptive_gc_boundary_flag_set = false; @@ -178,14 +178,14 @@ static void JNICALL MonitorContendedEntered(jvmtiEnv *jvmti, JNIEnv *jni, static void JNICALL MonitorWait(jvmtiEnv *jvmti, JNIEnv *jni, jthread thread, jobject object, jlong timeout) { - if (!VM::monitorEventsDelegated()) { + if (!VM::monitorWaitEventsDelegated()) { monitorBlockEnter(jvmti, jni, thread, object, OSThreadState::OBJECT_WAIT); } } static void JNICALL MonitorWaited(jvmtiEnv *jvmti, JNIEnv *jni, jthread thread, jobject object, jboolean timed_out) { - if (!VM::monitorEventsDelegated()) { + if (!VM::monitorWaitEventsDelegated()) { monitorBlockExit(jni, thread, OSThreadState::OBJECT_WAIT); } } @@ -564,12 +564,12 @@ bool VM::initializeRequestStackTrace() { } ProfilerBridgeInitResult VM::initProfilerBridge(JavaVM *vm, bool attach, - bool delegateMonitorEvents) { + bool delegateMonitorWaitEvents) { MutexLocker init_locker(profiler_bridge_init_lock); if (profiler_bridge_initialized) { bool requested_delegation = - delegateMonitorEvents && _native_monitor_events_available; - return requested_delegation == _monitor_events_delegated + delegateMonitorWaitEvents && _native_monitor_events_available; + return requested_delegation == _monitor_wait_events_delegated ? ProfilerBridgeInitResult::SUCCESS : ProfilerBridgeInitResult::MONITOR_EVENTS_DELEGATION_CONFLICT; } @@ -639,8 +639,8 @@ ProfilerBridgeInitResult VM::initProfilerBridge(JavaVM *vm, bool attach, _jvmti->GetCapabilities(&actual_capabilities); _native_monitor_events_available = actual_capabilities.can_generate_monitor_events; - _monitor_events_delegated = - delegateMonitorEvents && _native_monitor_events_available; + _monitor_wait_events_delegated = + delegateMonitorWaitEvents && _native_monitor_events_available; if (_hotspot) { probeJFRRequestStackTrace(); @@ -733,7 +733,7 @@ bool VM::setNativeMonitorEventsEnabled(bool enabled) { // When Java instrumentation owns Object.wait, do not enable the native wait // notifications at all. Disable still addresses all four events so teardown // is complete even if ownership was configured before this initialization. - if (!enabled || !_monitor_events_delegated) { + if (!enabled || !_monitor_wait_events_delegated) { wait = _jvmti->SetEventNotificationMode( mode, JVMTI_EVENT_MONITOR_WAIT, NULL); waited = _jvmti->SetEventNotificationMode( diff --git a/ddprof-lib/src/main/cpp/vmEntry.h b/ddprof-lib/src/main/cpp/vmEntry.h index 1d41a654a..6d700f331 100644 --- a/ddprof-lib/src/main/cpp/vmEntry.h +++ b/ddprof-lib/src/main/cpp/vmEntry.h @@ -156,7 +156,7 @@ class VM { static bool _zing; static bool _can_sample_objects; static bool _can_intercept_binding; - static bool _monitor_events_delegated; + static bool _monitor_wait_events_delegated; static bool _native_monitor_events_available; static bool _is_adaptive_gc_boundary_flag_set; static CodeCache *_libjvm; @@ -195,7 +195,7 @@ class VM { static bool initLibrary(JavaVM *vm); static ProfilerBridgeInitResult initProfilerBridge( - JavaVM *vm, bool attach, bool delegateMonitorEvents = false); + JavaVM *vm, bool attach, bool delegateMonitorWaitEvents = false); static jvmtiEnv *jvmti() { return _jvmti; } @@ -230,7 +230,9 @@ class VM { static bool canSampleObjects() { return _can_sample_objects; } - static bool monitorEventsDelegated() { return _monitor_events_delegated; } + static bool monitorWaitEventsDelegated() { + return _monitor_wait_events_delegated; + } static bool nativeMonitorEventsAvailable() { return _native_monitor_events_available; diff --git a/ddprof-lib/src/main/java/com/datadoghq/profiler/JavaProfiler.java b/ddprof-lib/src/main/java/com/datadoghq/profiler/JavaProfiler.java index 55535505b..9d6e408ea 100644 --- a/ddprof-lib/src/main/java/com/datadoghq/profiler/JavaProfiler.java +++ b/ddprof-lib/src/main/java/com/datadoghq/profiler/JavaProfiler.java @@ -127,7 +127,7 @@ public static synchronized JavaProfiler getInstance(String libLocation, String s public static synchronized JavaProfiler getInstance(String libLocation, String scratchDir, boolean delegateMonitorWaitEvents) throws IOException { if (instance != null) { - if (monitorEventsDelegated0() != delegateMonitorWaitEvents) { + if (monitorWaitEventsDelegated0() != delegateMonitorWaitEvents) { throw new IllegalStateException( "Monitor-event ownership conflicts with the profiler's " + "process-wide initialization"); @@ -161,13 +161,14 @@ public static synchronized JavaProfiler getInstance(String libLocation, String s } /** - * Reports whether Java instrumentation, rather than JVMTI callbacks, owns - * {@code Object.wait} TaskBlock intervals. + * Reports whether Java instrumentation owns {@code Object.wait} TaskBlock intervals instead + * of native JVMTI {@code MonitorWait} and {@code MonitorWaited} callbacks. Synchronized-monitor + * contention remains owned by native JVMTI callbacks. * - * @return {@code true} when native wait callbacks are delegated + * @return {@code true} when {@code Object.wait} handling is delegated to Java instrumentation */ - public boolean isMonitorEventsDelegated() { - return monitorEventsDelegated0(); + public boolean isMonitorWaitEventsDelegated() { + return monitorWaitEventsDelegated0(); } /** @@ -527,7 +528,7 @@ public Map getDebugCounters() { private static native void filterThreadRemove0(); private static native int getTid0(); - private static native boolean monitorEventsDelegated0(); + private static native boolean monitorWaitEventsDelegated0(); private static native boolean recordTrace0(long rootSpanId, String endpoint, String operation, int sizeLimit); diff --git a/ddprof-test/src/test/java/com/datadoghq/profiler/ExternalLauncher.java b/ddprof-test/src/test/java/com/datadoghq/profiler/ExternalLauncher.java index c193daf58..89ea8a44d 100644 --- a/ddprof-test/src/test/java/com/datadoghq/profiler/ExternalLauncher.java +++ b/ddprof-test/src/test/java/com/datadoghq/profiler/ExternalLauncher.java @@ -149,7 +149,8 @@ public static void main(String[] args) throws Exception { } else if (args[0].equals("profiler-agent-compatible")) { String libraryPath = System.getProperty("ddprof.test.agent.path"); JavaProfiler profiler = JavaProfiler.getInstance(libraryPath, null, false); - System.out.println("[agent-compatible] " + profiler.isMonitorEventsDelegated()); + System.out.println("[agent-compatible] " + + profiler.isMonitorWaitEventsDelegated()); } else if (args[0].equals("profiler-preexisting-monitor-wait")) { runPreExistingMonitorCallback(false); } else if (args[0].equals("profiler-preexisting-monitor-contention")) { diff --git a/ddprof-test/src/test/java/com/datadoghq/profiler/JavaProfilerApiSurfaceTest.java b/ddprof-test/src/test/java/com/datadoghq/profiler/JavaProfilerApiSurfaceTest.java index 89c042413..c74e26fa2 100644 --- a/ddprof-test/src/test/java/com/datadoghq/profiler/JavaProfilerApiSurfaceTest.java +++ b/ddprof-test/src/test/java/com/datadoghq/profiler/JavaProfilerApiSurfaceTest.java @@ -38,7 +38,7 @@ public void monitorWaitOwnershipIsExplicitPublicApi() throws Exception { .getDeclaredMethod("getInstance", String.class, String.class, boolean.class) .getModifiers())); assertTrue(Modifier.isPublic(JavaProfiler.class - .getDeclaredMethod("isMonitorEventsDelegated").getModifiers())); + .getDeclaredMethod("isMonitorWaitEventsDelegated").getModifiers())); } private static void assertNotPublic(Method method) { From 6f87c74336b8c01c0c093e547bb74fb96133864c Mon Sep 17 00:00:00 2001 From: Paul Fournillon Date: Thu, 30 Jul 2026 14:29:42 +0200 Subject: [PATCH 6/9] fix(taskblock): preserve requested monitor-wait ownership --- ddprof-lib/src/main/cpp/vmEntry.cpp | 25 +-- ddprof-lib/src/main/cpp/vmEntry.h | 6 +- .../com/datadoghq/profiler/JavaProfiler.java | 6 +- ddprof-lib/src/test/cpp/vmEntry_ut.cpp | 167 ++++++++++++++++++ .../datadoghq/profiler/ExternalLauncher.java | 45 ++++- .../datadoghq/profiler/JavaProfilerTest.java | 83 +++++++-- 6 files changed, 303 insertions(+), 29 deletions(-) create mode 100644 ddprof-lib/src/test/cpp/vmEntry_ut.cpp diff --git a/ddprof-lib/src/main/cpp/vmEntry.cpp b/ddprof-lib/src/main/cpp/vmEntry.cpp index eaea90508..cb4bcea74 100644 --- a/ddprof-lib/src/main/cpp/vmEntry.cpp +++ b/ddprof-lib/src/main/cpp/vmEntry.cpp @@ -54,13 +54,13 @@ bool VM::_can_sample_objects = false; bool VM::_can_intercept_binding = false; bool VM::_monitor_wait_events_delegated = false; bool VM::_native_monitor_events_available = false; +bool VM::_profiler_bridge_initialized = false; bool VM::_is_adaptive_gc_boundary_flag_set = false; // Serializes the one-time bridge installation and ownership negotiation. // Callback readers need no synchronization because ownership is assigned // before callbacks can be enabled and is never changed afterward. static Mutex profiler_bridge_init_lock; -static bool profiler_bridge_initialized = false; jvmtiExtensionFunction VM::_request_stack_trace = nullptr; jvmtiExtensionFunction VM::_init_request_stack_trace = nullptr; @@ -563,13 +563,19 @@ bool VM::initializeRequestStackTrace() { return false; } +void VM::configureMonitorEvents(bool delegateMonitorWaitEvents) { + jvmtiCapabilities actual_capabilities = {0}; + _jvmti->GetCapabilities(&actual_capabilities); + _native_monitor_events_available = + actual_capabilities.can_generate_monitor_events; + _monitor_wait_events_delegated = delegateMonitorWaitEvents; +} + ProfilerBridgeInitResult VM::initProfilerBridge(JavaVM *vm, bool attach, bool delegateMonitorWaitEvents) { MutexLocker init_locker(profiler_bridge_init_lock); - if (profiler_bridge_initialized) { - bool requested_delegation = - delegateMonitorWaitEvents && _native_monitor_events_available; - return requested_delegation == _monitor_wait_events_delegated + if (_profiler_bridge_initialized) { + return delegateMonitorWaitEvents == _monitor_wait_events_delegated ? ProfilerBridgeInitResult::SUCCESS : ProfilerBridgeInitResult::MONITOR_EVENTS_DELEGATION_CONFLICT; } @@ -635,12 +641,7 @@ ProfilerBridgeInitResult VM::initProfilerBridge(JavaVM *vm, bool attach, _jvmti->AddCapabilities(&capabilities); - jvmtiCapabilities actual_capabilities = {0}; - _jvmti->GetCapabilities(&actual_capabilities); - _native_monitor_events_available = - actual_capabilities.can_generate_monitor_events; - _monitor_wait_events_delegated = - delegateMonitorWaitEvents && _native_monitor_events_available; + configureMonitorEvents(delegateMonitorWaitEvents); if (_hotspot) { probeJFRRequestStackTrace(); @@ -716,7 +717,7 @@ ProfilerBridgeInitResult VM::initProfilerBridge(JavaVM *vm, bool attach, OS::installSignalHandler(WAKEUP_SIGNAL, NULL, wakeupHandler); - profiler_bridge_initialized = true; + _profiler_bridge_initialized = true; return ProfilerBridgeInitResult::SUCCESS; } diff --git a/ddprof-lib/src/main/cpp/vmEntry.h b/ddprof-lib/src/main/cpp/vmEntry.h index 6d700f331..35268a62a 100644 --- a/ddprof-lib/src/main/cpp/vmEntry.h +++ b/ddprof-lib/src/main/cpp/vmEntry.h @@ -133,8 +133,8 @@ class JavaVersionAccess { }; // The profiler bridge is process-wide and initialized exactly once. Later Java -// API initialization may reuse it only with the same effective Object.wait -// ownership. +// API initialization may reuse it only with the same requested Object.wait +// ownership, independently of native monitor-event availability. enum class ProfilerBridgeInitResult { SUCCESS, FAILURE, @@ -158,6 +158,7 @@ class VM { static bool _can_intercept_binding; static bool _monitor_wait_events_delegated; static bool _native_monitor_events_available; + static bool _profiler_bridge_initialized; static bool _is_adaptive_gc_boundary_flag_set; static CodeCache *_libjvm; @@ -179,6 +180,7 @@ class VM { static void *getLibraryHandle(const char *name); static bool initShared(JavaVM *vm); + static void configureMonitorEvents(bool delegateMonitorWaitEvents); static void probeJFRRequestStackTrace(); static CodeCache* openJvmLibrary(); diff --git a/ddprof-lib/src/main/java/com/datadoghq/profiler/JavaProfiler.java b/ddprof-lib/src/main/java/com/datadoghq/profiler/JavaProfiler.java index 9d6e408ea..10e790621 100644 --- a/ddprof-lib/src/main/java/com/datadoghq/profiler/JavaProfiler.java +++ b/ddprof-lib/src/main/java/com/datadoghq/profiler/JavaProfiler.java @@ -114,7 +114,8 @@ public static synchronized JavaProfiler getInstance(String libLocation, String s * the native profiler is a singleton. This may occur during {@code -agentpath} startup before * this method is called. When delegation is enabled, Java instrumentation owns * {@code Object.wait} TaskBlock intervals and native JVMTI wait callbacks are suppressed; - * native JVMTI callbacks continue to own synchronized monitor contention. + * native JVMTI callbacks continue to own synchronized monitor contention. Ownership is + * preserved independently of whether the JVM provides native monitor-event capability. * * @param libLocation the path to the native library to use, or {@literal null} for the bundled library * @param scratchDir directory where the bundled library will be exploded before linking @@ -163,7 +164,8 @@ public static synchronized JavaProfiler getInstance(String libLocation, String s /** * Reports whether Java instrumentation owns {@code Object.wait} TaskBlock intervals instead * of native JVMTI {@code MonitorWait} and {@code MonitorWaited} callbacks. Synchronized-monitor - * contention remains owned by native JVMTI callbacks. + * contention remains owned by native JVMTI callbacks. This reports the process-wide ownership + * selected during bridge initialization, independently of native monitor-event capability. * * @return {@code true} when {@code Object.wait} handling is delegated to Java instrumentation */ diff --git a/ddprof-lib/src/test/cpp/vmEntry_ut.cpp b/ddprof-lib/src/test/cpp/vmEntry_ut.cpp new file mode 100644 index 000000000..766186a40 --- /dev/null +++ b/ddprof-lib/src/test/cpp/vmEntry_ut.cpp @@ -0,0 +1,167 @@ +/* + * Copyright 2026, Datadog, Inc. + * SPDX-License-Identifier: Apache-2.0 + */ + +#include + +#include "vmEntry.h" + +class VMTestAccessor { + public: + static jvmtiEnv* jvmti() { return VM::_jvmti; } + static void setJvmti(jvmtiEnv* jvmti) { VM::_jvmti = jvmti; } + + static bool nativeMonitorEventsAvailable() { + return VM::_native_monitor_events_available; + } + static void setNativeMonitorEventsAvailable(bool available) { + VM::_native_monitor_events_available = available; + } + + static bool monitorWaitEventsDelegated() { + return VM::_monitor_wait_events_delegated; + } + static void setMonitorWaitEventsDelegated(bool delegated) { + VM::_monitor_wait_events_delegated = delegated; + } + + static bool profilerBridgeInitialized() { + return VM::_profiler_bridge_initialized; + } + static void setProfilerBridgeInitialized(bool initialized) { + VM::_profiler_bridge_initialized = initialized; + } + + static void configureMonitorEvents(bool delegate_monitor_wait_events) { + VM::configureMonitorEvents(delegate_monitor_wait_events); + } +}; + +class MonitorEventConfigurationTest : public ::testing::Test { + protected: + inline static MonitorEventConfigurationTest* active_test = nullptr; + + jvmtiInterface_1_ functions{}; + _jvmtiEnv mock_env{}; + jvmtiEnv* original_jvmti = nullptr; + bool original_initialized = false; + bool original_available = false; + bool original_delegated = false; + bool capability_available = false; + int get_capabilities_calls = 0; + + static jvmtiError JNICALL getCapabilities( + jvmtiEnv*, jvmtiCapabilities* capabilities) { + MonitorEventConfigurationTest* test = active_test; + *capabilities = jvmtiCapabilities{}; + capabilities->can_generate_monitor_events = test->capability_available; + test->get_capabilities_calls++; + return JVMTI_ERROR_NONE; + } + + void SetUp() override { + original_jvmti = VMTestAccessor::jvmti(); + original_initialized = VMTestAccessor::profilerBridgeInitialized(); + original_available = VMTestAccessor::nativeMonitorEventsAvailable(); + original_delegated = VMTestAccessor::monitorWaitEventsDelegated(); + + functions.GetCapabilities = &getCapabilities; + mock_env.functions = &functions; + VMTestAccessor::setJvmti(&mock_env); + VMTestAccessor::setProfilerBridgeInitialized(false); + active_test = this; + } + + void TearDown() override { + active_test = nullptr; + VMTestAccessor::setMonitorWaitEventsDelegated(original_delegated); + VMTestAccessor::setNativeMonitorEventsAvailable(original_available); + VMTestAccessor::setProfilerBridgeInitialized(original_initialized); + VMTestAccessor::setJvmti(original_jvmti); + } +}; + +TEST_F(MonitorEventConfigurationTest, + StoresRequestedOwnershipIndependentlyOfCapability) { + for (bool available : {false, true}) { + for (bool delegated : {false, true}) { + SCOPED_TRACE(::testing::Message() + << "available=" << available + << ", delegated=" << delegated); + capability_available = available; + get_capabilities_calls = 0; + + VMTestAccessor::configureMonitorEvents(delegated); + + EXPECT_EQ(1, get_capabilities_calls); + EXPECT_EQ(available, VMTestAccessor::nativeMonitorEventsAvailable()); + EXPECT_EQ(delegated, VMTestAccessor::monitorWaitEventsDelegated()); + EXPECT_FALSE(VMTestAccessor::profilerBridgeInitialized()); + } + } +} + +class ProfilerBridgeDelegationTest : public ::testing::Test { + protected: + bool original_initialized = false; + bool original_available = false; + bool original_delegated = false; + + void SetUp() override { + original_initialized = VMTestAccessor::profilerBridgeInitialized(); + original_available = VMTestAccessor::nativeMonitorEventsAvailable(); + original_delegated = VMTestAccessor::monitorWaitEventsDelegated(); + VMTestAccessor::setProfilerBridgeInitialized(true); + } + + void TearDown() override { + VMTestAccessor::setMonitorWaitEventsDelegated(original_delegated); + VMTestAccessor::setNativeMonitorEventsAvailable(original_available); + VMTestAccessor::setProfilerBridgeInitialized(original_initialized); + } + + static void expectNegotiation(bool available, bool delegated, + bool requested, + ProfilerBridgeInitResult expected) { + VMTestAccessor::setNativeMonitorEventsAvailable(available); + VMTestAccessor::setMonitorWaitEventsDelegated(delegated); + + EXPECT_EQ(expected, VM::initProfilerBridge(nullptr, true, requested)); + EXPECT_TRUE(VMTestAccessor::profilerBridgeInitialized()); + EXPECT_EQ(available, VMTestAccessor::nativeMonitorEventsAvailable()); + EXPECT_EQ(delegated, VMTestAccessor::monitorWaitEventsDelegated()); + } +}; + +TEST_F(ProfilerBridgeDelegationTest, + ReusesMatchingOwnershipWhenCapabilityIsUnavailable) { + expectNegotiation(false, false, false, ProfilerBridgeInitResult::SUCCESS); + expectNegotiation(false, true, true, ProfilerBridgeInitResult::SUCCESS); +} + +TEST_F(ProfilerBridgeDelegationTest, + RejectsConflictingOwnershipWhenCapabilityIsUnavailable) { + expectNegotiation( + false, false, true, + ProfilerBridgeInitResult::MONITOR_EVENTS_DELEGATION_CONFLICT); + expectNegotiation( + false, true, false, + ProfilerBridgeInitResult::MONITOR_EVENTS_DELEGATION_CONFLICT); +} + +TEST_F(ProfilerBridgeDelegationTest, + ReusesMatchingOwnershipWhenCapabilityIsAvailable) { + expectNegotiation(true, false, false, ProfilerBridgeInitResult::SUCCESS); + expectNegotiation(true, true, true, ProfilerBridgeInitResult::SUCCESS); +} + +TEST_F(ProfilerBridgeDelegationTest, + RejectsConflictingOwnershipWhenCapabilityIsAvailable) { + expectNegotiation( + true, false, true, + ProfilerBridgeInitResult::MONITOR_EVENTS_DELEGATION_CONFLICT); + expectNegotiation( + true, true, false, + ProfilerBridgeInitResult::MONITOR_EVENTS_DELEGATION_CONFLICT); +} diff --git a/ddprof-test/src/test/java/com/datadoghq/profiler/ExternalLauncher.java b/ddprof-test/src/test/java/com/datadoghq/profiler/ExternalLauncher.java index 89ea8a44d..717dbc7f0 100644 --- a/ddprof-test/src/test/java/com/datadoghq/profiler/ExternalLauncher.java +++ b/ddprof-test/src/test/java/com/datadoghq/profiler/ExternalLauncher.java @@ -32,6 +32,9 @@ *
  • profiler-virtual-thread - calls {@link JavaProfiler#getInstance()} for the first time from a virtual thread
    • *
  • profiler-agent-compatible - reuses native monitor ownership after agent initialization
    • *
  • profiler-delegation-conflict - requests delegated monitor ownership after agent initialization
    • + *
  • profiler-java-default-delegation-reuse - verifies explicit native ownership after default initialization
    • + *
  • profiler-java-default-delegation-conflict - verifies delegated ownership conflicts after default initialization
    • + *
  • profiler-java-delegation-reuse:<delegated> - verifies compatible Java singleton ownership reuse
    • *
  • profiler-java-delegation-conflict:<initial>:<requested> - verifies conflicting Java singleton ownership requests
    • *
  • profiler-preexisting-monitor-wait - exercises Object.wait on a thread created before profiler initialization
    • *
  • profiler-preexisting-monitor-contention - exercises monitor contention on a thread created before profiler initialization
    • @@ -127,24 +130,60 @@ public static void main(String[] args) throws Exception { } }); vt.join(); + JavaProfiler initial = JavaProfiler.getInstance(); + JavaProfiler reused = JavaProfiler.getInstance(); + System.out.println("[virtual-thread-recovery] " + + (initial == reused) + " " + reused.isMonitorWaitEventsDelegated()); } else if (args[0].equals("profiler-delegation-conflict")) { String libraryPath = System.getProperty("ddprof.test.agent.path"); try { JavaProfiler.getInstance(libraryPath, null, true); System.out.println("[delegation-conflict-missed]"); } catch (IllegalStateException expected) { - System.out.println("[delegation-conflict] " + expected.getMessage()); + JavaProfiler recovered = + JavaProfiler.getInstance(libraryPath, null, false); + System.out.println("[delegation-conflict] " + + recovered.isMonitorWaitEventsDelegated()); } + } else if (args[0].equals("profiler-java-default-delegation-reuse")) { + JavaProfiler initial = JavaProfiler.getInstance(); + JavaProfiler reused = JavaProfiler.getInstance(null, null, false); + System.out.println("[java-default-delegation-reuse] " + + (initial == reused) + " " + reused.isMonitorWaitEventsDelegated()); + } else if (args[0].equals("profiler-java-default-delegation-conflict")) { + JavaProfiler initial = JavaProfiler.getInstance(); + try { + JavaProfiler.getInstance(null, null, true); + System.out.println("[java-default-delegation-conflict-missed]"); + } catch (IllegalStateException expected) { + JavaProfiler recovered = + JavaProfiler.getInstance(null, null, false); + System.out.println("[java-default-delegation-conflict] " + + (initial == recovered) + " " + + recovered.isMonitorWaitEventsDelegated()); + } + } else if (args[0].startsWith("profiler-java-delegation-reuse:")) { + boolean delegated = Boolean.parseBoolean( + args[0].substring("profiler-java-delegation-reuse:".length())); + JavaProfiler initial = JavaProfiler.getInstance(null, null, delegated); + JavaProfiler reused = JavaProfiler.getInstance(null, null, delegated); + System.out.println("[java-delegation-reuse] " + + (initial == reused) + " " + reused.isMonitorWaitEventsDelegated()); } else if (args[0].startsWith("profiler-java-delegation-conflict:")) { String[] delegationModes = args[0].split(":"); boolean initialDelegation = Boolean.parseBoolean(delegationModes[1]); boolean requestedDelegation = Boolean.parseBoolean(delegationModes[2]); - JavaProfiler.getInstance(null, null, initialDelegation); + JavaProfiler initial = + JavaProfiler.getInstance(null, null, initialDelegation); try { JavaProfiler.getInstance(null, null, requestedDelegation); System.out.println("[java-delegation-conflict-missed]"); } catch (IllegalStateException expected) { - System.out.println("[java-delegation-conflict] " + expected.getMessage()); + JavaProfiler recovered = + JavaProfiler.getInstance(null, null, initialDelegation); + System.out.println("[java-delegation-conflict] " + + (initial == recovered) + " " + + recovered.isMonitorWaitEventsDelegated()); } } else if (args[0].equals("profiler-agent-compatible")) { String libraryPath = System.getProperty("ddprof.test.agent.path"); diff --git a/ddprof-test/src/test/java/com/datadoghq/profiler/JavaProfilerTest.java b/ddprof-test/src/test/java/com/datadoghq/profiler/JavaProfilerTest.java index 9c0004861..02a378d3e 100644 --- a/ddprof-test/src/test/java/com/datadoghq/profiler/JavaProfilerTest.java +++ b/ddprof-test/src/test/java/com/datadoghq/profiler/JavaProfilerTest.java @@ -154,20 +154,26 @@ void testJ9ForceJvmtiSanity() throws Exception { void getInstanceFromVirtualThreadThrowsIOException() throws Exception { assumeTrue(Platform.isJavaVersionAtLeast(21)); - AtomicReference resultLine = new AtomicReference<>(); + AtomicReference attemptLine = new AtomicReference<>(); + AtomicReference recoveryLine = new AtomicReference<>(); boolean val = launch("profiler-virtual-thread", Collections.emptyList(), "", l -> { if (l.startsWith("[virtual-thread-")) { - resultLine.set(l); - return LineConsumerResult.STOP; + if (l.startsWith("[virtual-thread-recovery]")) { + recoveryLine.set(l); + return LineConsumerResult.STOP; + } + attemptLine.set(l); + return LineConsumerResult.CONTINUE; } return LineConsumerResult.CONTINUE; }, null).inTime; assertTrue(val); - String result = resultLine.get(); + String result = attemptLine.get(); assertNotNull(result, "getInstance() did not report a result from the virtual thread"); assertTrue(result.startsWith("[virtual-thread-ioexception]"), "Expected IOException from getInstance() on a virtual thread, got: " + result); + assertEquals("[virtual-thread-recovery] true false", recoveryLine.get()); } @Test @@ -199,9 +205,24 @@ void conflictingLateMonitorDelegationIsRejected() throws Exception { assertTrue(result.inTime); assertEquals(0, result.exitCode); - assertNotNull(resultLine.get(), "Late delegation request did not report a result"); - assertTrue(resultLine.get().startsWith("[delegation-conflict]"), - "Expected ownership conflict, got: " + resultLine.get()); + assertEquals("[delegation-conflict] false", resultLine.get()); + } + + @Test + void defaultJavaSingletonMonitorDelegationIsReused() throws Exception { + assertJavaDelegationScenario( + "profiler-java-default-delegation-reuse", + "[java-default-delegation-reuse]", + "[java-default-delegation-reuse] true false"); + } + + @Test + void conflictingDefaultJavaSingletonMonitorDelegationDoesNotPoisonInstance() + throws Exception { + assertJavaDelegationScenario( + "profiler-java-default-delegation-conflict", + "[java-default-delegation-conflict", + "[java-default-delegation-conflict] true false"); } @Test @@ -210,6 +231,30 @@ void conflictingJavaSingletonMonitorDelegationIsRejected() throws Exception { assertJavaSingletonDelegationConflict(true, false); } + @Test + void compatibleJavaSingletonMonitorDelegationIsReused() throws Exception { + assertJavaSingletonDelegationReuse(false); + assertJavaSingletonDelegationReuse(true); + } + + /** Launches a fresh JVM and verifies that repeated ownership returns the same singleton. */ + private void assertJavaSingletonDelegationReuse(boolean delegated) throws Exception { + AtomicReference resultLine = new AtomicReference<>(); + LaunchResult result = launch( + "profiler-java-delegation-reuse:" + delegated, + Collections.emptyList(), "", line -> { + if (line.startsWith("[java-delegation-reuse]")) { + resultLine.set(line); + return LineConsumerResult.STOP; + } + return LineConsumerResult.CONTINUE; + }, null); + + assertTrue(result.inTime); + assertEquals(0, result.exitCode); + assertEquals("[java-delegation-reuse] true " + delegated, resultLine.get()); + } + /** Launches a fresh JVM and verifies that a second ownership mode is rejected. */ private void assertJavaSingletonDelegationConflict(boolean initialDelegation, boolean requestedDelegation) throws Exception { @@ -226,9 +271,27 @@ private void assertJavaSingletonDelegationConflict(boolean initialDelegation, assertTrue(result.inTime); assertEquals(0, result.exitCode); - assertNotNull(resultLine.get(), "Conflicting Java singleton request did not report a result"); - assertTrue(resultLine.get().startsWith("[java-delegation-conflict]"), - "Expected ownership conflict, got: " + resultLine.get()); + assertEquals( + "[java-delegation-conflict] true " + initialDelegation, + resultLine.get()); + } + + /** Launches a fresh JVM and verifies the exact output of a delegation scenario. */ + private void assertJavaDelegationScenario( + String target, String marker, String expected) throws Exception { + AtomicReference resultLine = new AtomicReference<>(); + LaunchResult result = launch( + target, Collections.emptyList(), "", line -> { + if (line.startsWith(marker)) { + resultLine.set(line); + return LineConsumerResult.STOP; + } + return LineConsumerResult.CONTINUE; + }, null); + + assertTrue(result.inTime); + assertEquals(0, result.exitCode); + assertEquals(expected, resultLine.get()); } @Test From 7f5cafaeb1068142c0c9e303108b8c13bc5b1d9b Mon Sep 17 00:00:00 2001 From: Paul Fournillon Date: Thu, 30 Jul 2026 14:36:10 +0200 Subject: [PATCH 7/9] fix(taskblock): make monitor callback activation transactional --- ddprof-lib/src/main/cpp/profiler.cpp | 34 ++- ddprof-lib/src/main/cpp/profiler.h | 6 +- ddprof-lib/src/main/cpp/vmEntry.cpp | 64 +++-- ddprof-lib/src/test/cpp/vmEntry_ut.cpp | 333 +++++++++++++++++++++++++ 4 files changed, 409 insertions(+), 28 deletions(-) diff --git a/ddprof-lib/src/main/cpp/profiler.cpp b/ddprof-lib/src/main/cpp/profiler.cpp index 986a29693..ced803a4b 100644 --- a/ddprof-lib/src/main/cpp/profiler.cpp +++ b/ddprof-lib/src/main/cpp/profiler.cpp @@ -1435,6 +1435,26 @@ Error Profiler::init() { return Error::OK; } +void Profiler::setTaskBlockEnabled(bool enabled) { + if (enabled) { + // Keep callback admission closed until native setup has either completed + // or rolled back, so partial event enablement cannot create paired state. + bool monitor_events_enabled = + VM::nativeMonitorEventsAvailable() && + VM::setNativeMonitorEventsEnabled(true); + _task_block_monitor_events_enabled.store(monitor_events_enabled, + std::memory_order_release); + _task_block_enabled.store(true, std::memory_order_release); + return; + } + + _task_block_enabled.store(false, std::memory_order_release); + if (_task_block_monitor_events_enabled.exchange( + false, std::memory_order_acq_rel)) { + VM::setNativeMonitorEventsEnabled(false); + } +} + Error Profiler::start(Arguments &args, bool reset) { MutexLocker ml(_state_lock); Error error = checkState(); @@ -1743,12 +1763,8 @@ Error Profiler::start(Arguments &args, bool reset) { // Paired with drainInflight() on the stop side. _cpu_engine->enableEvents(true); - _task_block_enabled.store( - (activated & EM_WALL) && args._wall_precheck && track_unfiltered_wall, - std::memory_order_release); - _task_block_monitor_events_enabled = - taskBlockEnabled() && VM::nativeMonitorEventsAvailable() && - VM::setNativeMonitorEventsEnabled(true); + setTaskBlockEnabled( + (activated & EM_WALL) && args._wall_precheck && track_unfiltered_wall); _state.store(RUNNING, std::memory_order_release); _start_time = time(NULL); __atomic_add_fetch(&_epoch, 1, __ATOMIC_RELAXED); @@ -1773,11 +1789,7 @@ Error Profiler::stop() { if (state() != RUNNING) { return Error("Profiler is not active"); } - _task_block_enabled.store(false, std::memory_order_release); - if (_task_block_monitor_events_enabled) { - VM::setNativeMonitorEventsEnabled(false); - _task_block_monitor_events_enabled = false; - } + setTaskBlockEnabled(false); // Order matters: disable engines first so the _enabled check inside signal // handlers will fail for any new signal delivered from now on. drain() then diff --git a/ddprof-lib/src/main/cpp/profiler.h b/ddprof-lib/src/main/cpp/profiler.h index 8561bb878..754a5f66f 100644 --- a/ddprof-lib/src/main/cpp/profiler.h +++ b/ddprof-lib/src/main/cpp/profiler.h @@ -133,7 +133,7 @@ class alignas(alignof(SpinLock)) Profiler { alignas(DEFAULT_CACHE_LINE_SIZE) u64 _failures[ASGCT_FAILURE_TYPES]; bool _wall_precheck = false; std::atomic _task_block_enabled{false}; - bool _task_block_monitor_events_enabled = false; + std::atomic _task_block_monitor_events_enabled{false}; std::atomic _task_block_rotation{false}; std::atomic _task_block_inflight{0}; @@ -182,6 +182,7 @@ class alignas(alignof(SpinLock)) Profiler { void lockAll(); void unlockAll(); + void setTaskBlockEnabled(bool enabled); void beginTaskBlockRotation(); void endTaskBlockRotation(); @@ -472,6 +473,9 @@ class alignas(alignof(SpinLock)) Profiler { bool taskBlockEnabled() const { return _task_block_enabled.load(std::memory_order_acquire); } + bool nativeMonitorTaskBlockEnabled() const { + return _task_block_monitor_events_enabled.load(std::memory_order_acquire); + } void writeLog(LogLevel level, const char *message); void writeLog(LogLevel level, const char *message, size_t len); void writeDatadogProfilerSetting(int tid, int length, const char *name, diff --git a/ddprof-lib/src/main/cpp/vmEntry.cpp b/ddprof-lib/src/main/cpp/vmEntry.cpp index cb4bcea74..6784b8115 100644 --- a/ddprof-lib/src/main/cpp/vmEntry.cpp +++ b/ddprof-lib/src/main/cpp/vmEntry.cpp @@ -90,6 +90,7 @@ static void monitorBlockEnter(jvmtiEnv *jvmti, JNIEnv *jni, jthread thread, jobject object, OSThreadState state) { Profiler *profiler = Profiler::instance(); if (!profiler->taskBlockEnabled() || + !profiler->nativeMonitorTaskBlockEnabled() || !JVMSupport::isPlatformThread(jni, thread)) { return; } @@ -724,31 +725,62 @@ ProfilerBridgeInitResult VM::initProfilerBridge(JavaVM *vm, bool attach, bool VM::setNativeMonitorEventsEnabled(bool enabled) { if (!_native_monitor_events_available) return false; - jvmtiEventMode mode = enabled ? JVMTI_ENABLE : JVMTI_DISABLE; - jvmtiError enter = _jvmti->SetEventNotificationMode( - mode, JVMTI_EVENT_MONITOR_CONTENDED_ENTER, NULL); - jvmtiError entered = _jvmti->SetEventNotificationMode( - mode, JVMTI_EVENT_MONITOR_CONTENDED_ENTERED, NULL); + jvmtiError enter = JVMTI_ERROR_NONE; + jvmtiError entered = JVMTI_ERROR_NONE; jvmtiError wait = JVMTI_ERROR_NONE; jvmtiError waited = JVMTI_ERROR_NONE; - // When Java instrumentation owns Object.wait, do not enable the native wait - // notifications at all. Disable still addresses all four events so teardown - // is complete even if ownership was configured before this initialization. - if (!enabled || !_monitor_wait_events_delegated) { - wait = _jvmti->SetEventNotificationMode( - mode, JVMTI_EVENT_MONITOR_WAIT, NULL); - waited = _jvmti->SetEventNotificationMode( - mode, JVMTI_EVENT_MONITOR_WAITED, NULL); + + if (enabled) { + // JVMTI enables each event independently and does not queue events that + // occur while disabled. Install every terminal notification before its + // entry notification so an admitted interval always has an exit path. + entered = _jvmti->SetEventNotificationMode( + JVMTI_ENABLE, JVMTI_EVENT_MONITOR_CONTENDED_ENTERED, NULL); + if (entered != JVMTI_ERROR_NONE) goto enable_failed; + + if (!_monitor_wait_events_delegated) { + waited = _jvmti->SetEventNotificationMode( + JVMTI_ENABLE, JVMTI_EVENT_MONITOR_WAITED, NULL); + if (waited != JVMTI_ERROR_NONE) goto enable_failed; + } + + enter = _jvmti->SetEventNotificationMode( + JVMTI_ENABLE, JVMTI_EVENT_MONITOR_CONTENDED_ENTER, NULL); + if (enter != JVMTI_ERROR_NONE) goto enable_failed; + + if (!_monitor_wait_events_delegated) { + wait = _jvmti->SetEventNotificationMode( + JVMTI_ENABLE, JVMTI_EVENT_MONITOR_WAIT, NULL); + if (wait != JVMTI_ERROR_NONE) goto enable_failed; + } + return true; + +enable_failed: + Log::warn("Unable to enable JVMTI monitor events: %d/%d/%d/%d", + enter, entered, wait, waited); + setNativeMonitorEventsEnabled(false); + return false; } + // Stop admitting new intervals before removing the terminal notifications. + // Disable all four events even when Object.wait is delegated so teardown + // also cleans up modes established before ownership was configured. + enter = _jvmti->SetEventNotificationMode( + JVMTI_DISABLE, JVMTI_EVENT_MONITOR_CONTENDED_ENTER, NULL); + wait = _jvmti->SetEventNotificationMode( + JVMTI_DISABLE, JVMTI_EVENT_MONITOR_WAIT, NULL); + entered = _jvmti->SetEventNotificationMode( + JVMTI_DISABLE, JVMTI_EVENT_MONITOR_CONTENDED_ENTERED, NULL); + waited = _jvmti->SetEventNotificationMode( + JVMTI_DISABLE, JVMTI_EVENT_MONITOR_WAITED, NULL); + if (enter == JVMTI_ERROR_NONE && entered == JVMTI_ERROR_NONE && wait == JVMTI_ERROR_NONE && waited == JVMTI_ERROR_NONE) { return true; } - Log::warn("Unable to %s JVMTI monitor events: %d/%d/%d/%d", - enabled ? "enable" : "disable", enter, entered, wait, waited); - if (enabled) setNativeMonitorEventsEnabled(false); + Log::warn("Unable to disable JVMTI monitor events: %d/%d/%d/%d", + enter, entered, wait, waited); return false; } diff --git a/ddprof-lib/src/test/cpp/vmEntry_ut.cpp b/ddprof-lib/src/test/cpp/vmEntry_ut.cpp index 766186a40..fa740cd17 100644 --- a/ddprof-lib/src/test/cpp/vmEntry_ut.cpp +++ b/ddprof-lib/src/test/cpp/vmEntry_ut.cpp @@ -3,8 +3,13 @@ * SPDX-License-Identifier: Apache-2.0 */ +#include +#include +#include + #include +#include "profiler.h" #include "vmEntry.h" class VMTestAccessor { @@ -38,6 +43,25 @@ class VMTestAccessor { } }; +class ProfilerTestAccessor { + public: + static void setTaskBlockEnabled(Profiler* profiler, bool enabled) { + profiler->setTaskBlockEnabled(enabled); + } + + static void setTaskBlockState(Profiler* profiler, bool enabled, + bool monitor_events_enabled) { + profiler->_task_block_enabled.store(enabled, std::memory_order_release); + profiler->_task_block_monitor_events_enabled.store( + monitor_events_enabled, std::memory_order_release); + } + + static bool monitorEventsEnabled(Profiler* profiler) { + return profiler->_task_block_monitor_events_enabled.load( + std::memory_order_acquire); + } +}; + class MonitorEventConfigurationTest : public ::testing::Test { protected: inline static MonitorEventConfigurationTest* active_test = nullptr; @@ -165,3 +189,312 @@ TEST_F(ProfilerBridgeDelegationTest, true, true, false, ProfilerBridgeInitResult::MONITOR_EVENTS_DELEGATION_CONFLICT); } + +class NativeMonitorEventsTest : public ::testing::Test { + protected: + struct EventCall { + jvmtiEventMode mode; + jvmtiEvent event; + bool task_block_enabled; + }; + + static constexpr std::array MONITOR_EVENTS = { + JVMTI_EVENT_MONITOR_CONTENDED_ENTER, + JVMTI_EVENT_MONITOR_CONTENDED_ENTERED, + JVMTI_EVENT_MONITOR_WAIT, + JVMTI_EVENT_MONITOR_WAITED, + }; + + inline static NativeMonitorEventsTest* active_test = nullptr; + + jvmtiInterface_1_ functions{}; + _jvmtiEnv mock_env{}; + std::vector calls; + std::array event_enabled{}; + bool inject_failure = false; + bool fail_all_disables = false; + jvmtiEventMode failure_mode = JVMTI_ENABLE; + jvmtiEvent failure_event = JVMTI_EVENT_MONITOR_CONTENDED_ENTER; + + Profiler* profiler = Profiler::instance(); + jvmtiEnv* original_jvmti = nullptr; + bool original_available = false; + bool original_delegated = false; + bool original_task_block_enabled = false; + bool original_monitor_events_enabled = false; + + static jvmtiError JNICALL setEventNotificationMode( + jvmtiEnv*, jvmtiEventMode mode, jvmtiEvent event, jthread, ...) { + NativeMonitorEventsTest* test = active_test; + test->calls.push_back( + {mode, event, test->profiler->taskBlockEnabled()}); + if (test->inject_failure && mode == test->failure_mode && + event == test->failure_event) { + return JVMTI_ERROR_INTERNAL; + } + if (test->fail_all_disables && mode == JVMTI_DISABLE) { + return JVMTI_ERROR_INTERNAL; + } + + test->event_enabled[test->eventIndex(event)] = mode == JVMTI_ENABLE; + return JVMTI_ERROR_NONE; + } + + void SetUp() override { + original_jvmti = VMTestAccessor::jvmti(); + original_available = VMTestAccessor::nativeMonitorEventsAvailable(); + original_delegated = VMTestAccessor::monitorWaitEventsDelegated(); + original_task_block_enabled = profiler->taskBlockEnabled(); + original_monitor_events_enabled = + ProfilerTestAccessor::monitorEventsEnabled(profiler); + + functions.SetEventNotificationMode = &setEventNotificationMode; + mock_env.functions = &functions; + VMTestAccessor::setJvmti(&mock_env); + VMTestAccessor::setNativeMonitorEventsAvailable(true); + VMTestAccessor::setMonitorWaitEventsDelegated(false); + ProfilerTestAccessor::setTaskBlockState(profiler, false, false); + active_test = this; + } + + void TearDown() override { + active_test = nullptr; + ProfilerTestAccessor::setTaskBlockState( + profiler, original_task_block_enabled, original_monitor_events_enabled); + VMTestAccessor::setMonitorWaitEventsDelegated(original_delegated); + VMTestAccessor::setNativeMonitorEventsAvailable(original_available); + VMTestAccessor::setJvmti(original_jvmti); + } + + static size_t eventIndex(jvmtiEvent event) { + for (size_t i = 0; i < MONITOR_EVENTS.size(); i++) { + if (MONITOR_EVENTS[i] == event) return i; + } + ADD_FAILURE() << "Unexpected JVMTI event " << event; + return 0; + } + + bool eventIsEnabled(jvmtiEvent event) const { + return event_enabled[eventIndex(event)]; + } + + void setAllEventsEnabled(bool enabled) { + event_enabled.fill(enabled); + } + + void resetObservations() { + calls.clear(); + event_enabled.fill(false); + inject_failure = false; + fail_all_disables = false; + } + + void fail(jvmtiEventMode mode, jvmtiEvent event) { + inject_failure = true; + failure_mode = mode; + failure_event = event; + } + + void expectCalls( + const std::vector>& expected) { + ASSERT_EQ(expected.size(), calls.size()); + for (size_t i = 0; i < expected.size(); i++) { + EXPECT_EQ(expected[i].first, calls[i].mode) << "call " << i; + EXPECT_EQ(expected[i].second, calls[i].event) << "call " << i; + } + } + + static std::vector> disableCalls() { + return { + {JVMTI_DISABLE, JVMTI_EVENT_MONITOR_CONTENDED_ENTER}, + {JVMTI_DISABLE, JVMTI_EVENT_MONITOR_WAIT}, + {JVMTI_DISABLE, JVMTI_EVENT_MONITOR_CONTENDED_ENTERED}, + {JVMTI_DISABLE, JVMTI_EVENT_MONITOR_WAITED}, + }; + } +}; + +TEST_F(NativeMonitorEventsTest, EnablesTerminalEventsBeforeEntryEvents) { + EXPECT_TRUE(VM::setNativeMonitorEventsEnabled(true)); + + expectCalls({ + {JVMTI_ENABLE, JVMTI_EVENT_MONITOR_CONTENDED_ENTERED}, + {JVMTI_ENABLE, JVMTI_EVENT_MONITOR_WAITED}, + {JVMTI_ENABLE, JVMTI_EVENT_MONITOR_CONTENDED_ENTER}, + {JVMTI_ENABLE, JVMTI_EVENT_MONITOR_WAIT}, + }); + for (jvmtiEvent event : MONITOR_EVENTS) { + EXPECT_TRUE(eventIsEnabled(event)); + } +} + +TEST_F(NativeMonitorEventsTest, DelegatedEnableOnlyInstallsContendedPair) { + VMTestAccessor::setMonitorWaitEventsDelegated(true); + + EXPECT_TRUE(VM::setNativeMonitorEventsEnabled(true)); + + expectCalls({ + {JVMTI_ENABLE, JVMTI_EVENT_MONITOR_CONTENDED_ENTERED}, + {JVMTI_ENABLE, JVMTI_EVENT_MONITOR_CONTENDED_ENTER}, + }); + EXPECT_TRUE(eventIsEnabled(JVMTI_EVENT_MONITOR_CONTENDED_ENTER)); + EXPECT_TRUE(eventIsEnabled(JVMTI_EVENT_MONITOR_CONTENDED_ENTERED)); + EXPECT_FALSE(eventIsEnabled(JVMTI_EVENT_MONITOR_WAIT)); + EXPECT_FALSE(eventIsEnabled(JVMTI_EVENT_MONITOR_WAITED)); +} + +TEST_F(NativeMonitorEventsTest, DisableRemovesEntriesBeforeTerminalEvents) { + for (bool delegated : {false, true}) { + SCOPED_TRACE(delegated); + resetObservations(); + setAllEventsEnabled(true); + VMTestAccessor::setMonitorWaitEventsDelegated(delegated); + + EXPECT_TRUE(VM::setNativeMonitorEventsEnabled(false)); + + expectCalls(disableCalls()); + for (jvmtiEvent event : MONITOR_EVENTS) { + EXPECT_FALSE(eventIsEnabled(event)); + } + } +} + +TEST_F(NativeMonitorEventsTest, EnableFailureStopsAndRollsBackAllEvents) { + const std::array enable_order = { + JVMTI_EVENT_MONITOR_CONTENDED_ENTERED, + JVMTI_EVENT_MONITOR_WAITED, + JVMTI_EVENT_MONITOR_CONTENDED_ENTER, + JVMTI_EVENT_MONITOR_WAIT, + }; + + for (size_t failure_index = 0; failure_index < enable_order.size(); + failure_index++) { + SCOPED_TRACE(failure_index); + resetObservations(); + fail(JVMTI_ENABLE, enable_order[failure_index]); + + EXPECT_FALSE(VM::setNativeMonitorEventsEnabled(true)); + + std::vector> expected; + for (size_t i = 0; i <= failure_index; i++) { + expected.push_back({JVMTI_ENABLE, enable_order[i]}); + } + std::vector> rollback = + disableCalls(); + expected.insert(expected.end(), rollback.begin(), rollback.end()); + expectCalls(expected); + for (jvmtiEvent event : MONITOR_EVENTS) { + EXPECT_FALSE(eventIsEnabled(event)); + } + } +} + +TEST_F(NativeMonitorEventsTest, DelegatedEnableFailureRollsBackAllEvents) { + VMTestAccessor::setMonitorWaitEventsDelegated(true); + const std::array enable_order = { + JVMTI_EVENT_MONITOR_CONTENDED_ENTERED, + JVMTI_EVENT_MONITOR_CONTENDED_ENTER, + }; + + for (size_t failure_index = 0; failure_index < enable_order.size(); + failure_index++) { + SCOPED_TRACE(failure_index); + resetObservations(); + fail(JVMTI_ENABLE, enable_order[failure_index]); + + EXPECT_FALSE(VM::setNativeMonitorEventsEnabled(true)); + + std::vector> expected; + for (size_t i = 0; i <= failure_index; i++) { + expected.push_back({JVMTI_ENABLE, enable_order[i]}); + } + std::vector> rollback = + disableCalls(); + expected.insert(expected.end(), rollback.begin(), rollback.end()); + expectCalls(expected); + for (jvmtiEvent event : MONITOR_EVENTS) { + EXPECT_FALSE(eventIsEnabled(event)); + } + } +} + +TEST_F(NativeMonitorEventsTest, DisableFailureStillAttemptsEveryEvent) { + for (jvmtiEvent failed_event : MONITOR_EVENTS) { + SCOPED_TRACE(failed_event); + resetObservations(); + setAllEventsEnabled(true); + fail(JVMTI_DISABLE, failed_event); + + EXPECT_FALSE(VM::setNativeMonitorEventsEnabled(false)); + + expectCalls(disableCalls()); + for (jvmtiEvent event : MONITOR_EVENTS) { + EXPECT_EQ(event == failed_event, eventIsEnabled(event)); + } + } +} + +TEST_F(NativeMonitorEventsTest, UnavailableCapabilityDoesNotCallJvmti) { + VMTestAccessor::setNativeMonitorEventsAvailable(false); + + EXPECT_FALSE(VM::setNativeMonitorEventsEnabled(true)); + EXPECT_TRUE(calls.empty()); +} + +TEST_F(NativeMonitorEventsTest, AdmissionRemainsClosedDuringSuccessfulSetup) { + ProfilerTestAccessor::setTaskBlockEnabled(profiler, true); + + ASSERT_FALSE(calls.empty()); + for (const EventCall& call : calls) { + EXPECT_FALSE(call.task_block_enabled); + } + EXPECT_TRUE(profiler->taskBlockEnabled()); + EXPECT_TRUE(ProfilerTestAccessor::monitorEventsEnabled(profiler)); +} + +TEST_F(NativeMonitorEventsTest, + AdmissionRemainsClosedDuringFailedSetupAndRollback) { + fail(JVMTI_ENABLE, JVMTI_EVENT_MONITOR_WAIT); + + ProfilerTestAccessor::setTaskBlockEnabled(profiler, true); + + ASSERT_FALSE(calls.empty()); + for (const EventCall& call : calls) { + EXPECT_FALSE(call.task_block_enabled); + } + EXPECT_TRUE(profiler->taskBlockEnabled()); + EXPECT_FALSE(ProfilerTestAccessor::monitorEventsEnabled(profiler)); + for (jvmtiEvent event : MONITOR_EVENTS) { + EXPECT_FALSE(eventIsEnabled(event)); + } +} + +TEST_F(NativeMonitorEventsTest, + NativeAdmissionRemainsClosedWhenSetupAndRollbackFail) { + fail(JVMTI_ENABLE, JVMTI_EVENT_MONITOR_WAIT); + fail_all_disables = true; + + ProfilerTestAccessor::setTaskBlockEnabled(profiler, true); + + EXPECT_TRUE(profiler->taskBlockEnabled()); + EXPECT_FALSE(profiler->nativeMonitorTaskBlockEnabled()); + EXPECT_FALSE(ProfilerTestAccessor::monitorEventsEnabled(profiler)); + EXPECT_TRUE(eventIsEnabled(JVMTI_EVENT_MONITOR_CONTENDED_ENTER)); + EXPECT_TRUE(eventIsEnabled(JVMTI_EVENT_MONITOR_CONTENDED_ENTERED)); + EXPECT_FALSE(eventIsEnabled(JVMTI_EVENT_MONITOR_WAIT)); + EXPECT_TRUE(eventIsEnabled(JVMTI_EVENT_MONITOR_WAITED)); +} + +TEST_F(NativeMonitorEventsTest, AdmissionClosesBeforeNativeTeardown) { + setAllEventsEnabled(true); + ProfilerTestAccessor::setTaskBlockState(profiler, true, true); + + ProfilerTestAccessor::setTaskBlockEnabled(profiler, false); + + expectCalls(disableCalls()); + for (const EventCall& call : calls) { + EXPECT_FALSE(call.task_block_enabled); + } + EXPECT_FALSE(profiler->taskBlockEnabled()); + EXPECT_FALSE(ProfilerTestAccessor::monitorEventsEnabled(profiler)); +} From 66c4a33fae6e717dcc9e2597c90333bb652f82c5 Mon Sep 17 00:00:00 2001 From: Paul Fournillon Date: Mon, 3 Aug 2026 12:03:25 +0200 Subject: [PATCH 8/9] test: migrate task block context setup --- .../datadoghq/profiler/wallclock/MonitorTaskBlockTest.java | 4 ++-- .../com/datadoghq/profiler/wallclock/ParkTaskBlockTest.java | 4 ++-- 2 files changed, 4 insertions(+), 4 deletions(-) diff --git a/ddprof-test/src/test/java/com/datadoghq/profiler/wallclock/MonitorTaskBlockTest.java b/ddprof-test/src/test/java/com/datadoghq/profiler/wallclock/MonitorTaskBlockTest.java index b05f95b6e..eb77c01cb 100644 --- a/ddprof-test/src/test/java/com/datadoghq/profiler/wallclock/MonitorTaskBlockTest.java +++ b/ddprof-test/src/test/java/com/datadoghq/profiler/wallclock/MonitorTaskBlockTest.java @@ -85,14 +85,14 @@ public void contextWindowObjectWaitDoesNotEmitTaskBlock() throws Exception { Thread worker = new Thread(() -> { try { registerCurrentThreadForWallClockProfiling(); - profiler.setContext(0x4400L, 0x4401L, 0L, 0x4401L); + profiler.setTraceContext(0x4400L, 0x4401L, 0L, 0x4401L, -1, null, -1, null); synchronized (monitor) { monitor.wait(100); } } catch (Throwable t) { failure.set(t); } finally { - profiler.clearContext(); + profiler.clearTraceContext(); profiler.removeThread(); } }, "taskblock-traced-object-wait"); diff --git a/ddprof-test/src/test/java/com/datadoghq/profiler/wallclock/ParkTaskBlockTest.java b/ddprof-test/src/test/java/com/datadoghq/profiler/wallclock/ParkTaskBlockTest.java index bf7d3528b..ccb3331b3 100644 --- a/ddprof-test/src/test/java/com/datadoghq/profiler/wallclock/ParkTaskBlockTest.java +++ b/ddprof-test/src/test/java/com/datadoghq/profiler/wallclock/ParkTaskBlockTest.java @@ -45,7 +45,7 @@ public void platformParkEmitsTaskBlockOutsideContextWindow() { @Test public void contextWindowParkDoesNotEmitTaskBlock() { registerCurrentThreadForWallClockProfiling(); - profiler.setContext(0x3100L, 0x3101L, 0L, 0x3101L); + profiler.setTraceContext(0x3100L, 0x3101L, 0L, 0x3101L, -1, null, -1, null); try { ProfilerOwnedBlockHooks.parkEnter(profiler); try { @@ -54,7 +54,7 @@ public void contextWindowParkDoesNotEmitTaskBlock() { ProfilerOwnedBlockHooks.parkExit(profiler, BLOCKER, UNBLOCKING_SPAN_ID); } } finally { - profiler.clearContext(); + profiler.clearTraceContext(); profiler.removeThread(); } stopProfiler(); From e32a7e78f618d545ad55014d705ff54f7d1099be Mon Sep 17 00:00:00 2001 From: Paul Fournillon Date: Mon, 3 Aug 2026 12:37:31 +0200 Subject: [PATCH 9/9] fix(taskblock): avoid reinitializing active profiler bridge --- ddprof-lib/src/main/cpp/vmEntry.cpp | 5 +++++ 1 file changed, 5 insertions(+) diff --git a/ddprof-lib/src/main/cpp/vmEntry.cpp b/ddprof-lib/src/main/cpp/vmEntry.cpp index 6784b8115..2cb1c0107 100644 --- a/ddprof-lib/src/main/cpp/vmEntry.cpp +++ b/ddprof-lib/src/main/cpp/vmEntry.cpp @@ -506,6 +506,11 @@ bool VM::initShared(JavaVM* vm) { } bool VM::initLibrary(JavaVM *vm) { + MutexLocker init_locker(profiler_bridge_init_lock); + if (_profiler_bridge_initialized) { + return true; + } + TEST_LOG("VM::initLibrary"); if (!initShared(vm)) { return false;