From c27d0bdc7e863d2710ce3999d8e48d8ae372428f Mon Sep 17 00:00:00 2001 From: "erwan.viollet" Date: Fri, 3 Jul 2026 10:21:08 +0200 Subject: [PATCH 1/2] [PROF-15268] refactor(profiler): per-thread JFR-inflight counter (POC) MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit POC follow-up to #614 evaluating whether moving the SignalInflight counter from a global atomic to per-thread storage on ProfiledThread is a viable alternative. - ProfiledThread: adds _jfr_inflight (atomic RMW on owner-thread write, ACQUIRE-read from drain), _registry_next intrusive pointer, and a spinlock-protected registry head. initCurrentThread / freeKey / current() insert / remove around the pthread_key lifecycle. - SignalInflight: enter/exit prefer the current ProfiledThread's per- thread counter; fall back to the existing global counter for threads that fire signals before initCurrentThread runs. drain() iterates the registry summing per-thread counters plus the fallback. Cache-line contention on the counter is eliminated on the fast path. The J9 longjmp leak documented in signalInflight.h is not yet closed by this commit (needs segvHandler/busHandler hooks to reset the current thread's slot before chaining) — left for a follow-up if the design survives review. Verified: buildDebug, compileRelease --rerun-tasks, ShutdownTest, JavaProfilerTest, CollapsingSleepTest, SmokeWallTest (all cstack modes). --- ddprof-lib/src/main/cpp/signalInflight.cpp | 54 +++++++++++++++++++--- ddprof-lib/src/main/cpp/signalInflight.h | 27 ++++++----- ddprof-lib/src/main/cpp/thread.cpp | 43 +++++++++++++++++ ddprof-lib/src/main/cpp/thread.h | 35 +++++++++++++- 4 files changed, 137 insertions(+), 22 deletions(-) diff --git a/ddprof-lib/src/main/cpp/signalInflight.cpp b/ddprof-lib/src/main/cpp/signalInflight.cpp index 2925dfd26c..57cafb4214 100644 --- a/ddprof-lib/src/main/cpp/signalInflight.cpp +++ b/ddprof-lib/src/main/cpp/signalInflight.cpp @@ -16,20 +16,58 @@ #include "signalInflight.h" #include "log.h" +#include "thread.h" #include #include #include -alignas(64) int SignalInflight::_counter = 0; +alignas(64) int SignalInflight::_fallback_counter = 0; // 200 ms: long enough for any legitimate signal handler to finish, // short enough to avoid a perceptible hang if a handler is somehow stuck. static const long DRAIN_TIMEOUT_NS = 200000000L; +void SignalInflight::enter() { + ProfiledThread *pt = ProfiledThread::currentSignalSafe(); + if (pt != nullptr) { + // Fast path: increment the owning thread's counter. No cross-core write + // traffic because only this thread ever writes here. + pt->enterJfrInflight(); + } else { + // Fallback for threads without a ProfiledThread (early init, native + // threads that fire a signal before initCurrentThread runs). + __atomic_fetch_add(&_fallback_counter, 1, __ATOMIC_ACQUIRE); + } +} + +void SignalInflight::exit() { + ProfiledThread *pt = ProfiledThread::currentSignalSafe(); + if (pt != nullptr) { + pt->exitJfrInflight(); + } else { + __atomic_fetch_sub(&_fallback_counter, 1, __ATOMIC_RELEASE); + } +} + +// Sum callback used by drain(); ctx points to an int accumulator. +static void addInflight(ProfiledThread *pt, void *ctx) { + *static_cast(ctx) += pt->jfrInflight(); +} + +static int totalInflight() { + int total = __atomic_load_n(&SignalInflight::_fallback_counter, __ATOMIC_ACQUIRE); + ProfiledThread::forEachRegistered(addInflight, &total); + return total; +} + +bool SignalInflight::hasInflight() { + return totalInflight() > 0; +} + bool SignalInflight::drain() { - if (__atomic_load_n(&_counter, __ATOMIC_ACQUIRE) == 0) { - return true; // fast path: nothing in flight + if (totalInflight() == 0) { + return true; // fast path } struct timespec deadline; @@ -44,7 +82,11 @@ bool SignalInflight::drain() { deadline.tv_nsec -= 1000000000L; } - while (__atomic_load_n(&_counter, __ATOMIC_ACQUIRE) > 0) { + while (true) { + int total = totalInflight(); + if (total == 0) { + return true; + } struct timespec now; if (clock_gettime(CLOCK_MONOTONIC, &now) != 0) { Log::error("SignalInflight::drain: clock_gettime(CLOCK_MONOTONIC) failed (errno=%d). " @@ -53,14 +95,12 @@ bool SignalInflight::drain() { } if (now.tv_sec > deadline.tv_sec || (now.tv_sec == deadline.tv_sec && now.tv_nsec >= deadline.tv_nsec)) { - int remaining = __atomic_load_n(&_counter, __ATOMIC_ACQUIRE); Log::error("SignalInflight::drain: timed out after %ldms waiting for " "%d in-flight signal handler(s). Skipping JFR teardown to " "prevent use-after-free. This indicates a stuck signal handler.", - DRAIN_TIMEOUT_NS / 1000000L, remaining); + DRAIN_TIMEOUT_NS / 1000000L, total); return false; } sched_yield(); } - return true; } diff --git a/ddprof-lib/src/main/cpp/signalInflight.h b/ddprof-lib/src/main/cpp/signalInflight.h index 2855af1c2e..845277166a 100644 --- a/ddprof-lib/src/main/cpp/signalInflight.h +++ b/ddprof-lib/src/main/cpp/signalInflight.h @@ -50,25 +50,24 @@ // to gate is exactly the safety net for this case. class SignalInflight { public: - // ACQUIRE on increment / RELEASE on decrement: drain() observes all - // handler-side writes before observing the counter at zero. - static void enter() { - __atomic_fetch_add(&_counter, 1, __ATOMIC_ACQUIRE); - } - static void exit() { - __atomic_fetch_sub(&_counter, 1, __ATOMIC_RELEASE); - } - static bool hasInflight() { - return __atomic_load_n(&_counter, __ATOMIC_ACQUIRE) > 0; - } + // Increment / decrement the inflight tally. Prefers per-thread storage on + // the current ProfiledThread when one is available; falls back to the + // global counter otherwise (early-init threads, native threads with no PT). + static void enter(); + static void exit(); - // Spin until the counter reaches zero or DRAIN_TIMEOUT_NS elapses. + // True iff any thread (per-thread or global-fallback) has a positive count. + static bool hasInflight(); + + // Spin until every counter reaches zero or DRAIN_TIMEOUT_NS elapses. // Returns true on clean drain, false on timeout. Callers MUST NOT proceed // with JFR teardown when this returns false. static bool drain(); -private: - alignas(64) static int _counter; + // Global fallback counter (still cache-line-aligned to keep it off engine + // _enabled lines). Not private so InflightGuard's inline path can access + // it if needed; the recommended entry point remains enter()/exit(). + alignas(64) static int _fallback_counter; }; // RAII guard for signal-handler in-flight tracking. Construct as the first diff --git a/ddprof-lib/src/main/cpp/thread.cpp b/ddprof-lib/src/main/cpp/thread.cpp index 16f482fc04..bf487644c5 100644 --- a/ddprof-lib/src/main/cpp/thread.cpp +++ b/ddprof-lib/src/main/cpp/thread.cpp @@ -13,6 +13,45 @@ pthread_key_t ProfiledThread::_tls_key; bool ProfiledThread::_tls_key_initialized = false; +ProfiledThread *ProfiledThread::_registry_head = nullptr; +int ProfiledThread::_registry_lock = 0; + +// Simple test-and-set spinlock. Registry ops are rare (thread create / +// destroy / drain) so we don't need anything fancier. +#define REGISTRY_LOCK() \ + while (__atomic_test_and_set(&_registry_lock, __ATOMIC_ACQUIRE)) { \ + __asm__ __volatile__("" ::: "memory"); \ + } +#define REGISTRY_UNLOCK() \ + __atomic_clear(&_registry_lock, __ATOMIC_RELEASE) + +void ProfiledThread::registryInsert(ProfiledThread *pt) { + REGISTRY_LOCK(); + pt->_registry_next = _registry_head; + _registry_head = pt; + REGISTRY_UNLOCK(); +} + +void ProfiledThread::registryRemove(ProfiledThread *pt) { + REGISTRY_LOCK(); + ProfiledThread **cur = &_registry_head; + while (*cur != nullptr && *cur != pt) { + cur = &((*cur)->_registry_next); + } + if (*cur == pt) { + *cur = pt->_registry_next; + pt->_registry_next = nullptr; + } + REGISTRY_UNLOCK(); +} + +void ProfiledThread::forEachRegistered(void (*visit)(ProfiledThread *, void *), void *ctx) { + REGISTRY_LOCK(); + for (ProfiledThread *p = _registry_head; p != nullptr; p = p->_registry_next) { + visit(p, ctx); + } + REGISTRY_UNLOCK(); +} void ProfiledThread::initTLSKey() { static pthread_once_t tls_initialized = PTHREAD_ONCE_INIT; @@ -32,6 +71,7 @@ inline void ProfiledThread::freeKey(void *key) { ProfiledThread *tls_ref = (ProfiledThread *)(key); if (tls_ref != NULL) { SignalBlocker blocker; + registryRemove(tls_ref); delete tls_ref; } } @@ -48,6 +88,7 @@ void ProfiledThread::initCurrentThread() { int tid = OS::threadId(); ProfiledThread *tls = ProfiledThread::forTid(tid); + registryInsert(tls); pthread_setspecific(_tls_key, (const void *)tls); } @@ -60,6 +101,7 @@ void ProfiledThread::release() { if (tls != NULL) { SignalBlocker blocker; pthread_setspecific(key, NULL); + registryRemove(tls); delete tls; } } @@ -80,6 +122,7 @@ ProfiledThread *ProfiledThread::current() { // Lazy allocation - safe since current() is never called from signal handlers int tid = OS::threadId(); tls = ProfiledThread::forTid(tid); + registryInsert(tls); pthread_setspecific(_tls_key, (const void *)tls); } return tls; diff --git a/ddprof-lib/src/main/cpp/thread.h b/ddprof-lib/src/main/cpp/thread.h index a15cf8fc1c..4f30450147 100644 --- a/ddprof-lib/src/main/cpp/thread.h +++ b/ddprof-lib/src/main/cpp/thread.h @@ -42,6 +42,12 @@ class ProfiledThread : public ThreadLocalData { static pthread_key_t _tls_key; static bool _tls_key_initialized; + // Intrusive singly-linked list of live ProfiledThreads. Protected by + // _registry_lock. Iterated by SignalInflight::drain() to sum per-thread + // JFR-inflight counters across all threads. + static ProfiledThread *_registry_head; + static int _registry_lock; // spinlock; simple RMW acquire/release + static void initTLSKey(); static void doInitTLSKey(); static inline void freeKey(void *key); @@ -60,6 +66,13 @@ class ProfiledThread : public ThreadLocalData { 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) + // Per-thread JFR-inflight counter. Incremented by InflightGuard at signal- + // handler entry, decremented on all exit paths. drain() reads it across + // threads via ACQUIRE. + int _jfr_inflight; + // Intrusive singly-linked registry pointer. Threads insert themselves in + // initCurrentThread() and are removed from freeKey() at thread exit. + ProfiledThread *_registry_next; UnwindFailures _unwind_failures; bool _otel_ctx_initialized; bool _crash_protection_active; @@ -79,7 +92,7 @@ class ProfiledThread : public ThreadLocalData { : ThreadLocalData(), _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), _filter_slot_id(-1), _init_window(0), - _signal_depth(0), + _signal_depth(0), _jfr_inflight(0), _registry_next(nullptr), _otel_ctx_initialized(false), _crash_protection_active(false), _otel_ctx_record{}, _otel_tag_encodings{}, _otel_local_root_span_id(0) {}; @@ -190,6 +203,26 @@ class ProfiledThread : public ThreadLocalData { inline void enterSignalScope() { ++_signal_depth; } inline void exitSignalScope() { if (_signal_depth > 0) --_signal_depth; } + // JFR-inflight tracking (used by InflightGuard). Only the owning thread + // writes; drain() reads from other threads with ACQUIRE. + inline void enterJfrInflight() { + __atomic_fetch_add(&_jfr_inflight, 1, __ATOMIC_ACQUIRE); + } + inline void exitJfrInflight() { + __atomic_fetch_sub(&_jfr_inflight, 1, __ATOMIC_RELEASE); + } + inline int jfrInflight() const { + return __atomic_load_n(&_jfr_inflight, __ATOMIC_ACQUIRE); + } + + // Registry iteration for SignalInflight::drain(). Callback is invoked under + // the registry lock; it must be short and must not touch the registry. + static void forEachRegistered(void (*visit)(ProfiledThread *, void *), void *ctx); + + // Registry insert/remove — called from initCurrentThread() / freeKey(). + static void registryInsert(ProfiledThread *pt); + static void registryRemove(ProfiledThread *pt); + UnwindFailures* unwindFailures(bool reset = true) { if (reset) { _unwind_failures.clear(); From 63a7ec5aee622cff54c8e894f561e1e00fa305f8 Mon Sep 17 00:00:00 2001 From: "copilot-swe-agent[bot]" <198982749+Copilot@users.noreply.github.com> Date: Thu, 9 Jul 2026 17:06:04 +0000 Subject: [PATCH 2/2] fix: resolve merge conflict - port per-thread JFR-inflight additions from deleted thread.h to threadLocalData.h PR #615 (origin/main) renamed thread.cpp to threadLocalData.cpp and deleted thread.h, absorbing ProfiledThread into threadLocalData.h. Our branch had modified thread.h to add per-thread JFR-inflight tracking. Resolution: - Delete thread.h (accept deletion from main) - Port _registry_head/_registry_lock static members to threadLocalData.h - Port _jfr_inflight/_registry_next instance members to threadLocalData.h - Port enterJfrInflight/exitJfrInflight/jfrInflight methods to threadLocalData.h - Port forEachRegistered/registryInsert/registryRemove declarations to threadLocalData.h - Update signalInflight.cpp to include threadLocalData.h instead of thread.h - threadLocalData.cpp already has implementation (correctly auto-merged) --- ddprof-lib/src/main/cpp/signalInflight.cpp | 2 +- ddprof-lib/src/main/cpp/thread.h | 334 --------------------- ddprof-lib/src/main/cpp/threadLocalData.h | 35 ++- 3 files changed, 35 insertions(+), 336 deletions(-) delete mode 100644 ddprof-lib/src/main/cpp/thread.h diff --git a/ddprof-lib/src/main/cpp/signalInflight.cpp b/ddprof-lib/src/main/cpp/signalInflight.cpp index 57cafb4214..ede42bbd08 100644 --- a/ddprof-lib/src/main/cpp/signalInflight.cpp +++ b/ddprof-lib/src/main/cpp/signalInflight.cpp @@ -16,7 +16,7 @@ #include "signalInflight.h" #include "log.h" -#include "thread.h" +#include "threadLocalData.h" #include #include diff --git a/ddprof-lib/src/main/cpp/thread.h b/ddprof-lib/src/main/cpp/thread.h deleted file mode 100644 index 4f30450147..0000000000 --- a/ddprof-lib/src/main/cpp/thread.h +++ /dev/null @@ -1,334 +0,0 @@ -/* - * Copyright 2025, 2026, Datadog, Inc. - * SPDX-License-Identifier: Apache-2.0 - */ - -#ifndef _THREAD_H -#define _THREAD_H - -#include "context.h" -#include "otel_context.h" -#include "os.h" -#include "threadLocalData.h" -#include "threadState.h" -#include "unwindStats.h" -#include -#include -#include -#include -#include -#include -#include - -class ProfiledThread : public ThreadLocalData { -public: - enum ThreadType : u32 { - TYPE_UNKNOWN = 0, - TYPE_JAVA_THREAD = 0x1, - TYPE_NOT_JAVA_THREAD = 0x2, - TYPE_MASK = TYPE_JAVA_THREAD | TYPE_NOT_JAVA_THREAD - }; - - static constexpr u32 FLAG_PARKED = 0x4u; // next free bit after TYPE_MASK (0x1|0x2) - -private: - // We are allowing several levels of nesting because we can be - // eg. in a crash handler when wallclock signal kicks in, - // catching sigseg while also triggering CPU signal handler - // which would also potentially trigger sigseg we need to handle. - // This means 3 levels but we allow for some wiggling space, just in case. - // Even with 5 levels cap we will need any highly recursing signal handlers - static constexpr u32 CRASH_HANDLER_NESTING_LIMIT = 5; - static pthread_key_t _tls_key; - static bool _tls_key_initialized; - - // Intrusive singly-linked list of live ProfiledThreads. Protected by - // _registry_lock. Iterated by SignalInflight::drain() to sum per-thread - // JFR-inflight counters across all threads. - static ProfiledThread *_registry_head; - static int _registry_lock; // spinlock; simple RMW acquire/release - - static void initTLSKey(); - static void doInitTLSKey(); - static inline void freeKey(void *key); - - u64 _pc; - u64 _sp; - u64 _span_id; // Wall-clock collapsing cache: last-seen span ID (not a context store — read from _otel_ctx_record on each signal, cached here to detect "same as last time") - volatile u32 _crash_depth; - int _tid; - u32 _cpu_epoch; - u32 _wall_epoch; - u64 _call_trace_id; - u32 _recording_epoch; - u32 _misc_flags; - u64 _park_block_token; - 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) - // Per-thread JFR-inflight counter. Incremented by InflightGuard at signal- - // handler entry, decremented on all exit paths. drain() reads it across - // threads via ACQUIRE. - int _jfr_inflight; - // Intrusive singly-linked registry pointer. Threads insert themselves in - // initCurrentThread() and are removed from freeKey() at thread exit. - ProfiledThread *_registry_next; - UnwindFailures _unwind_failures; - bool _otel_ctx_initialized; - bool _crash_protection_active; - // alignas(8) + sizeof(OtelThreadContextRecord)==640 (multiple of 8) guarantee - // _otel_tag_encodings sits at +640 with no padding, so the three fields form one - // 688-byte contiguous region exposed as a combined DirectByteBuffer. - alignas(8) OtelThreadContextRecord _otel_ctx_record; - // These two fields MUST be contiguous and 8-byte aligned — the JNI layer - // exposes them as a single DirectByteBuffer (sidecar), and VarHandle long - // views require 8-byte alignment for the buffer base address. - // Read invariant: sidecar readers must gate on record->valid (see ContextApi::get). - // ThreadContext.restore() relies on this to perform a bulk memcpy under valid=0. - alignas(8) u32 _otel_tag_encodings[DD_TAGS_CAPACITY]; - u64 _otel_local_root_span_id; - - ProfiledThread(int tid) - : ThreadLocalData(), _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), _filter_slot_id(-1), _init_window(0), - _signal_depth(0), _jfr_inflight(0), _registry_next(nullptr), - _otel_ctx_initialized(false), _crash_protection_active(false), - _otel_ctx_record{}, _otel_tag_encodings{}, _otel_local_root_span_id(0) {}; - - virtual ~ProfiledThread() { } -public: - static ProfiledThread *forTid(int tid) { return new ProfiledThread(tid); } - - static void initCurrentThread(); - static void release(); -#ifdef UNIT_TEST - // Simulates the moment inside release() after pthread_setspecific(NULL) but - // before delete — the race window the clearCurrentThreadTLS fix covers. - // Returns the detached pointer so the caller can delete it after assertions. - static ProfiledThread* clearCurrentThreadTLS() { - if (__atomic_load_n(&_tls_key_initialized, __ATOMIC_ACQUIRE)) { - ProfiledThread *pt = (ProfiledThread *)pthread_getspecific(_tls_key); - pthread_setspecific(_tls_key, nullptr); - return pt; - } - return nullptr; - } - // Deletes a ProfiledThread returned by clearCurrentThreadTLS(). - // Needed because the destructor is private. - static void deleteForTest(ProfiledThread *pt) { delete pt; } -#endif - - static ProfiledThread *current(); - static ProfiledThread *currentSignalSafe(); // Signal-safe version that never allocates - static int currentTid(); - - inline int tid() { return _tid; } - - inline u64 noteCPUSample(u32 recording_epoch) { - _recording_epoch = recording_epoch; - return ++_cpu_epoch; - } - - /** - * Attempts to reuse a cached call trace ID for wallclock sample collapsing. - * Collapsing is allowed only when the execution state (PC, SP) and trace - * context (spanId, rootSpanId) are identical to the previous sample. - * - * @param pc Program counter from ucontext - * @param sp Stack pointer from ucontext - * @param recording_epoch Current profiling session epoch - * @param context_valid True if the OTEP valid flag was set; controls whether _otel_local_root_span_id is updated - * @param span_id Current trace span ID - * @param root_span_id Current trace root span ID - * @return Cached call_trace_id if collapsing is allowed, 0 otherwise - */ - u64 lookupWallclockCallTraceId(u64 pc, u64 sp, u32 recording_epoch, - bool context_valid, u64 span_id, u64 root_span_id) { - if (_pc == pc && _sp == sp && _span_id == span_id && - _otel_local_root_span_id == root_span_id && _recording_epoch == recording_epoch && - _call_trace_id != 0) { - return _call_trace_id; - } - _pc = pc; - _sp = sp; - _span_id = span_id; - // Only update the sidecar when context is valid (valid=1). If the signal fires - // between detach() and attach() in Java, ContextApi::get returns valid=0 with - // root_span_id=0; writing that would clobber the value Java just stored. - if (context_valid) { - // Plain store is safe: naturally-aligned u64 stores/loads are atomic on - // x86-64 and aarch64 (the only supported targets). The Java writer uses - // sidecarBuffer.putLong() which is a single aligned 8-byte store. - _otel_local_root_span_id = root_span_id; - } - _recording_epoch = recording_epoch; - return 0; - } - - inline void recordCallTraceId(u64 call_trace_id) { - _call_trace_id = call_trace_id; - } - - // this is called in the crash handler to avoid recursing - bool enterCrashHandler() { - u32 prev = _crash_depth; - // This is thread local; no need for atomic cmpxchg - if (prev < CRASH_HANDLER_NESTING_LIMIT) { - _crash_depth++; - return true; - } - return false; - } - - // needs to be called when the crash handler exits - void exitCrashHandler() { - // failsafe check - do not attempt to decrement if there are no crash handlers on stack - if (_crash_depth > 0) _crash_depth--; - } - - void resetCrashHandler() { - _crash_depth = 0; - } - - bool isDeepCrashHandler() { - return _crash_depth > CRASH_HANDLER_NESTING_LIMIT; - } - - // Signal-handler depth counter used by SignalHandlerScope (guards.h). All - // access happens on the owning thread (signal handlers are delivered to the - // thread that's interrupted), so plain reads/writes are AS-safe — no locks, - // no malloc, no syscalls. See guards.h for the public API. - inline uint8_t signalDepth() const { return _signal_depth; } - inline void enterSignalScope() { ++_signal_depth; } - inline void exitSignalScope() { if (_signal_depth > 0) --_signal_depth; } - - // JFR-inflight tracking (used by InflightGuard). Only the owning thread - // writes; drain() reads from other threads with ACQUIRE. - inline void enterJfrInflight() { - __atomic_fetch_add(&_jfr_inflight, 1, __ATOMIC_ACQUIRE); - } - inline void exitJfrInflight() { - __atomic_fetch_sub(&_jfr_inflight, 1, __ATOMIC_RELEASE); - } - inline int jfrInflight() const { - return __atomic_load_n(&_jfr_inflight, __ATOMIC_ACQUIRE); - } - - // Registry iteration for SignalInflight::drain(). Callback is invoked under - // the registry lock; it must be short and must not touch the registry. - static void forEachRegistered(void (*visit)(ProfiledThread *, void *), void *ctx); - - // Registry insert/remove — called from initCurrentThread() / freeKey(). - static void registryInsert(ProfiledThread *pt); - static void registryRemove(ProfiledThread *pt); - - UnwindFailures* unwindFailures(bool reset = true) { - if (reset) { - _unwind_failures.clear(); - } - return &_unwind_failures; - } - - int filterSlotId() { return _filter_slot_id; } - void setFilterSlotId(int slotId) { _filter_slot_id = slotId; } - - // JVM thread init race window (PROF-13072): skip at most one signal that fires - // between Profiler::registerThread() and the JVM's pd_set_thread() call. - // Pure native threads (e.g. NativeThreadCreator) also see nullptr from - // JVMThread::current(), so the window auto-expires after one skip, allowing - // their subsequent samples through. - inline bool inInitWindow() const { return _init_window > 0; } - inline void startInitWindow() { _init_window = 1; } - inline void tickInitWindow() { if (_init_window > 0) --_init_window; } - - // Signal handler reentrancy protection - bool tryEnterCriticalSection() { - // Uses GCC atomic builtin (no malloc, async-signal-safe) - bool expected = false; - return __atomic_compare_exchange_n(&_in_critical_section, &expected, true, false, __ATOMIC_ACQUIRE, __ATOMIC_RELAXED); - } - void exitCriticalSection() { - // Uses GCC atomic builtin (no malloc, async-signal-safe) - __atomic_store_n(&_in_critical_section, false, __ATOMIC_RELEASE); - } - - // Context TLS (OTEP #4947) - inline void markContextInitialized() { - _otel_ctx_initialized = true; - } - - inline bool isContextInitialized() { - return _otel_ctx_initialized; - } - - inline OtelThreadContextRecord* getOtelContextRecord() { - return &_otel_ctx_record; - } - - // CAS RMW to update only TYPE_MASK bits without clobbering FLAG_PARKED, which - // is managed independently by the Java park hooks on the owning thread. - inline void setJavaThread(bool is_java) { - const u32 type_bits = is_java ? static_cast(TYPE_JAVA_THREAD) : static_cast(TYPE_NOT_JAVA_THREAD); - u32 cur = __atomic_load_n(&_misc_flags, __ATOMIC_RELAXED); - u32 desired; - do { - desired = (cur & ~static_cast(TYPE_MASK)) | type_bits; - } while (!__atomic_compare_exchange_n(&_misc_flags, &cur, desired, - /*weak=*/true, - __ATOMIC_ACQ_REL, __ATOMIC_RELAXED)); - } - - inline enum ThreadType threadType() const { - u32 flags = __atomic_load_n(&_misc_flags, __ATOMIC_ACQUIRE); - return static_cast(flags & TYPE_MASK); - } - - inline bool isCrashProtectionActive() const { return _crash_protection_active; } - inline void setCrashProtectionActive(bool active) { _crash_protection_active = active; } - - // JFR tag encoding sidecar — populated by JNI thread, read by signal handler - // (flightRecorder.cpp writeCurrentContext / wallClock.cpp collapsing). - inline u32* getOtelTagEncodingsPtr() { return _otel_tag_encodings; } - inline u32 getOtelTagEncoding(u32 idx) const { - return idx < DD_TAGS_CAPACITY ? _otel_tag_encodings[idx] : 0; - } - inline u64 getOtelLocalRootSpanId() const { return _otel_local_root_span_id; } - - inline void clearOtelSidecar() { - memset(_otel_tag_encodings, 0, sizeof(_otel_tag_encodings)); - _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 void setParkBlockToken(u64 token) { - _park_block_token = token; - } - - // Returns false if the thread was not parked (idempotent). - inline bool parkExit(u64 &park_block_token) { - u32 prev = __atomic_fetch_and(&_misc_flags, ~FLAG_PARKED, __ATOMIC_ACQ_REL); - if ((prev & FLAG_PARKED) == 0) { - return false; - } - park_block_token = _park_block_token; - _park_block_token = 0; - return true; - } - - Context snapshotContext(size_t numAttrs); - -private: - // Atomic flag for signal handler reentrancy protection within the same thread - // Must be atomic because a signal handler can interrupt normal execution mid-instruction, - // and both contexts may attempt to enter the critical section. Without atomic exchange(), - // both could see the flag as false and both would think they successfully entered. - // The atomic exchange() is uninterruptible, ensuring only one context succeeds. - bool _in_critical_section{false}; -}; - -#endif // _THREAD_H diff --git a/ddprof-lib/src/main/cpp/threadLocalData.h b/ddprof-lib/src/main/cpp/threadLocalData.h index e8b9a30c46..a9e9abfe1e 100644 --- a/ddprof-lib/src/main/cpp/threadLocalData.h +++ b/ddprof-lib/src/main/cpp/threadLocalData.h @@ -56,6 +56,12 @@ class ProfiledThread : public ThreadLocalData { static pthread_key_t _tls_key; static bool _tls_key_initialized; + // Intrusive singly-linked list of live ProfiledThreads. Protected by + // _registry_lock. Iterated by SignalInflight::drain() to sum per-thread + // JFR-inflight counters across all threads. + static ProfiledThread *_registry_head; + static int _registry_lock; // spinlock; simple RMW acquire/release + static void initTLSKey(); static void doInitTLSKey(); static inline void freeKey(void *key); @@ -83,6 +89,13 @@ class ProfiledThread : public ThreadLocalData { 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) + // Per-thread JFR-inflight counter. Incremented by InflightGuard at signal- + // handler entry, decremented on all exit paths. drain() reads it across + // threads via ACQUIRE. + int _jfr_inflight; + // Intrusive singly-linked registry pointer. Threads insert themselves in + // initCurrentThread() and are removed from freeKey() at thread exit. + ProfiledThread *_registry_next; UnwindFailures _unwind_failures; bool _otel_ctx_initialized; // alignas(8) + sizeof(OtelThreadContextRecord)==640 (multiple of 8) guarantee @@ -101,7 +114,7 @@ class ProfiledThread : public ThreadLocalData { : 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), _filter_slot_id(-1), _init_window(0), - _signal_depth(0), + _signal_depth(0), _jfr_inflight(0), _registry_next(nullptr), _otel_ctx_initialized(false), _otel_ctx_record{}, _otel_tag_encodings{}, _otel_local_root_span_id(0) {}; @@ -224,6 +237,26 @@ class ProfiledThread : public ThreadLocalData { inline void enterSignalScope() { ++_signal_depth; } inline void exitSignalScope() { if (_signal_depth > 0) --_signal_depth; } + // JFR-inflight tracking (used by InflightGuard). Only the owning thread + // writes; drain() reads from other threads with ACQUIRE. + inline void enterJfrInflight() { + __atomic_fetch_add(&_jfr_inflight, 1, __ATOMIC_ACQUIRE); + } + inline void exitJfrInflight() { + __atomic_fetch_sub(&_jfr_inflight, 1, __ATOMIC_RELEASE); + } + inline int jfrInflight() const { + return __atomic_load_n(&_jfr_inflight, __ATOMIC_ACQUIRE); + } + + // Registry iteration for SignalInflight::drain(). Callback is invoked under + // the registry lock; it must be short and must not touch the registry. + static void forEachRegistered(void (*visit)(ProfiledThread *, void *), void *ctx); + + // Registry insert/remove — called from initCurrentThread() / freeKey(). + static void registryInsert(ProfiledThread *pt); + static void registryRemove(ProfiledThread *pt); + UnwindFailures* unwindFailures(bool reset = true) { if (reset) { _unwind_failures.clear();