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 b2f1028fa..6548797ba 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,20 @@ 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); + 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 @@ -94,6 +108,12 @@ Java_com_datadoghq_profiler_JavaProfiler_getTid0(JNIEnv *env, jclass unused) { return OS::threadId(); } +extern "C" DLLEXPORT jboolean JNICALL +Java_com_datadoghq_profiler_JavaProfiler_monitorWaitEventsDelegated0( + JNIEnv *env, jclass unused) { + return VM::monitorWaitEventsDelegated(); +} + extern "C" DLLEXPORT jstring JNICALL Java_com_datadoghq_profiler_JavaProfiler_execute0(JNIEnv *env, jobject unused, jstring command) { @@ -137,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 @@ -184,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 } @@ -360,43 +354,56 @@ 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()) { - ThreadFilter::SlotID slot_id = ensureCurrentThreadFilterSlot(tf, current); + 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 = tf->ensureCurrentThreadSlot(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(); + 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) { @@ -410,9 +417,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(); @@ -427,16 +435,16 @@ 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)); } 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(); @@ -468,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/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/cpp/profiler.cpp b/ddprof-lib/src/main/cpp/profiler.cpp index 7147e298c..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,9 +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); + 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); @@ -1770,7 +1789,7 @@ Error Profiler::stop() { if (state() != RUNNING) { return Error("Profiler is not active"); } - _task_block_enabled.store(false, std::memory_order_release); + 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 310fbe804..754a5f66f 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}; + std::atomic _task_block_monitor_events_enabled{false}; std::atomic _task_block_rotation{false}; std::atomic _task_block_inflight{0}; @@ -181,6 +182,7 @@ class alignas(alignof(SpinLock)) Profiler { void lockAll(); void unlockAll(); + void setTaskBlockEnabled(bool enabled); void beginTaskBlockRotation(); void endTaskBlockRotation(); @@ -471,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/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..2cb1c0107 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" @@ -15,10 +16,13 @@ #include "jvmThread.h" #include "libraries.h" #include "log.h" +#include "mutex.h" #include "os.h" #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,8 +52,16 @@ bool VM::_hotspot = false; bool VM::_zing = false; 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; + jvmtiExtensionFunction VM::_request_stack_trace = nullptr; jvmtiExtensionFunction VM::_init_request_stack_trace = nullptr; @@ -67,6 +79,118 @@ 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() || + !profiler->nativeMonitorTaskBlockEnabled() || + !JVMSupport::isPlatformThread(jni, thread)) { + return; + } + ProfiledThread *current = ProfiledThread::initCurrentThreadSignalSafe(); + 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); + 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; + } + current->clearMonitorBlock(); + if (!current->monitorEnter(TSC::ticks(), context, + monitorBlockerHash(jvmti, object), state)) { + return; + } + } + + ThreadFilter *tf = profiler->threadFilter(); + ThreadFilter::SlotID slot_id = tf->ensureCurrentThreadSlot(current); + if (!tf->unfilteredWallTrackingActive() || 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(); + finishTaskBlockAtExit(current, profiler->threadFilter(), thread, 0, token, + start_ticks, context, blocker, 0); +} + +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::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::monitorWaitEventsDelegated()) { + 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 @@ -382,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; @@ -440,15 +569,31 @@ bool VM::initializeRequestStackTrace() { return false; } -bool VM::initProfilerBridge(JavaVM *vm, bool attach) { +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) { + return delegateMonitorWaitEvents == _monitor_wait_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 @@ -479,6 +624,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 +642,13 @@ 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); + configureMonitorEvents(delegateMonitorWaitEvents); + if (_hotspot) { probeJFRRequestStackTrace(); } @@ -516,6 +665,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); @@ -568,7 +723,70 @@ 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) { + if (!_native_monitor_events_available) return false; + + jvmtiError enter = JVMTI_ERROR_NONE; + jvmtiError entered = JVMTI_ERROR_NONE; + jvmtiError wait = JVMTI_ERROR_NONE; + jvmtiError waited = JVMTI_ERROR_NONE; + + 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 disable JVMTI monitor events: %d/%d/%d/%d", + enter, entered, wait, waited); + return false; } // Run late initialization when JVM is ready. May be called more than once (from @@ -702,7 +920,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 75725ef15..35268a62a 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 requested Object.wait +// ownership, independently of native monitor-event availability. +enum class ProfilerBridgeInitResult { + SUCCESS, + FAILURE, + MONITOR_EVENTS_DELEGATION_CONFLICT, +}; + class VM { friend class VMTestAccessor; @@ -147,6 +156,9 @@ class VM { static bool _zing; static bool _can_sample_objects; 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; @@ -168,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(); @@ -183,7 +196,8 @@ class VM { static JVM_GetManagement _getManagement; static bool initLibrary(JavaVM *vm); - static bool initProfilerBridge(JavaVM *vm, bool attach); + static ProfilerBridgeInitResult initProfilerBridge( + JavaVM *vm, bool attach, bool delegateMonitorWaitEvents = false); static jvmtiEnv *jvmti() { return _jvmti; } @@ -218,6 +232,15 @@ class VM { static bool canSampleObjects() { return _can_sample_objects; } + static bool monitorWaitEventsDelegated() { + return _monitor_wait_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..10e790621 100644 --- a/ddprof-lib/src/main/java/com/datadoghq/profiler/JavaProfiler.java +++ b/ddprof-lib/src/main/java/com/datadoghq/profiler/JavaProfiler.java @@ -104,7 +104,35 @@ 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 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. 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 + * @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 { if (instance != null) { + if (monitorWaitEventsDelegated0() != delegateMonitorWaitEvents) { + throw new IllegalStateException( + "Monitor-event ownership conflicts with the profiler's " + + "process-wide initialization"); + } return instance; } @@ -113,12 +141,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 +161,18 @@ public static synchronized JavaProfiler getInstance(String libLocation, String s return profiler; } + /** + * 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. 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 + */ + public boolean isMonitorWaitEventsDelegated() { + return monitorWaitEventsDelegated0(); + } + /** * Stop profiling (without dumping results) * @@ -400,7 +439,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 +447,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 +459,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 +522,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 +530,7 @@ public Map getDebugCounters() { private static native void filterThreadRemove0(); private static native int getTid0(); + private static native boolean monitorWaitEventsDelegated0(); private static native boolean recordTrace0(long rootSpanId, String endpoint, String operation, int sizeLimit); @@ -504,13 +544,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-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-lib/src/test/cpp/vmEntry_ut.cpp b/ddprof-lib/src/test/cpp/vmEntry_ut.cpp new file mode 100644 index 000000000..fa740cd17 --- /dev/null +++ b/ddprof-lib/src/test/cpp/vmEntry_ut.cpp @@ -0,0 +1,500 @@ +/* + * Copyright 2026, Datadog, Inc. + * SPDX-License-Identifier: Apache-2.0 + */ + +#include +#include +#include + +#include + +#include "profiler.h" +#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 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; + + 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); +} + +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)); +} 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..717dbc7f0 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,14 @@ *

  • 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-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
    • * */ public class ExternalLauncher { @@ -38,6 +53,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 +130,70 @@ 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) { + 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 initial = + JavaProfiler.getInstance(null, null, initialDelegation); + try { + JavaProfiler.getInstance(null, null, requestedDelegation); + System.out.println("[java-delegation-conflict-missed]"); + } catch (IllegalStateException expected) { + 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"); + JavaProfiler profiler = JavaProfiler.getInstance(libraryPath, null, false); + 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")) { + 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/JavaProfilerApiSurfaceTest.java b/ddprof-test/src/test/java/com/datadoghq/profiler/JavaProfilerApiSurfaceTest.java index 058bd5294..c74e26fa2 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("isMonitorWaitEventsDelegated").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/JavaProfilerTest.java b/ddprof-test/src/test/java/com/datadoghq/profiler/JavaProfilerTest.java index 2023c4757..02a378d3e 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"); @@ -118,20 +154,173 @@ 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 + 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); + 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 + void conflictingJavaSingletonMonitorDelegationIsRejected() throws Exception { + assertJavaSingletonDelegationConflict(false, true); + 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 { + 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); + 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 + 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 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..ff6df5c97 --- /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,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 new file mode 100644 index 000000000..63e56c380 --- /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,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 new file mode 100644 index 000000000..eb77c01cb --- /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.setTraceContext(0x4400L, 0x4401L, 0L, 0x4401L, -1, null, -1, null); + synchronized (monitor) { + monitor.wait(100); + } + } catch (Throwable t) { + failure.set(t); + } finally { + profiler.clearTraceContext(); + 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,filter=,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,filter=,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..ccb3331b3 --- /dev/null +++ b/ddprof-test/src/test/java/com/datadoghq/profiler/wallclock/ParkTaskBlockTest.java @@ -0,0 +1,169 @@ +/* + * 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.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 { + 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.setTraceContext(0x3100L, 0x3101L, 0L, 0x3101L, -1, null, -1, null); + try { + ProfilerOwnedBlockHooks.parkEnter(profiler); + try { + parkForMillis(200); + } finally { + ProfilerOwnedBlockHooks.parkExit(profiler, BLOCKER, UNBLOCKING_SPAN_ID); + } + } finally { + profiler.clearTraceContext(); + 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); + } + + @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,filter=,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); + } + } + + 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/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()));