From 46812e9e5fce0ce1747d1537575c4bee38e0f288 Mon Sep 17 00:00:00 2001 From: Vlad Scherbich Date: Thu, 23 Jul 2026 22:30:15 -0400 Subject: [PATCH] fix(profiling): maintain fast-copy and foreign-takeover fallback across forks (PROF-14568) Follow-up to #18798: add persistent fast-copy intent flags, fork-safe warmup re-run, sticky foreign-takeover inheritance, prefork sampling pause, and per-cycle handler ownership checks. Publish fast-copy metadata via ProfilerState snapshots and extend fork/warmup coverage in tests. --- .../dd_wrapper/include/profiler_state.hpp | 15 + .../dd_wrapper/include/profiler_stats.hpp | 15 +- .../dd_wrapper/src/profiler_stats.cpp | 108 +++- .../datadog/profiling/stack/__init__.pyi | 15 +- .../datadog/profiling/stack/_stack.pyi | 16 +- .../profiling/stack/echion/echion/interp.h | 5 +- .../profiling/stack/echion/echion/threads.h | 8 +- .../profiling/stack/echion/echion/vm.h | 19 +- .../profiling/stack/include/sampler.hpp | 22 +- .../profiling/stack/src/echion/interp.cc | 8 +- .../profiling/stack/src/echion/threads.cc | 9 +- .../datadog/profiling/stack/src/echion/vm.cc | 15 +- .../datadog/profiling/stack/src/sampler.cpp | 500 ++++++++++-------- .../datadog/profiling/stack/src/stack.cpp | 50 +- ...-handler-warmup-fork-2c9d4e7a1f6b0853.yaml | 7 + .../collector/test_copy_memory_stats.py | 321 ++++++++++- tests/profiling/collector/test_utils.py | 31 ++ 17 files changed, 843 insertions(+), 321 deletions(-) create mode 100644 releasenotes/notes/profiling-stack-sampler-foreign-handler-warmup-fork-2c9d4e7a1f6b0853.yaml diff --git a/ddtrace/internal/datadog/profiling/dd_wrapper/include/profiler_state.hpp b/ddtrace/internal/datadog/profiling/dd_wrapper/include/profiler_state.hpp index 0b9891ad0e9..efc80077d41 100644 --- a/ddtrace/internal/datadog/profiling/dd_wrapper/include/profiler_state.hpp +++ b/ddtrace/internal/datadog/profiling/dd_wrapper/include/profiler_state.hpp @@ -86,6 +86,21 @@ class ProfilerState // every upload in UploaderBuilder::build, which would drop the value. std::string profiler_settings_info_json; + // Process-static fast-copy configuration snapshot. ProfilerStats is swapped on + // every upload, so metadata serialization falls back to this when per-window + // stats have not been seeded yet (e.g. heap upload before a sampling cycle). + struct FastCopyProfilerMetadata + { + bool snapshot_initialized = false; + bool user_disabled = false; + bool capable = false; + bool syscall_fallback = false; + bool enabled = false; + bool desired = false; + bool foreign_takeover = false; + }; + FastCopyProfilerMetadata fast_copy_metadata{}; + // ======================================================================== // Native call tracking state // ======================================================================== diff --git a/ddtrace/internal/datadog/profiling/dd_wrapper/include/profiler_stats.hpp b/ddtrace/internal/datadog/profiling/dd_wrapper/include/profiler_stats.hpp index 3802b83d1f1..1881f66a9c5 100644 --- a/ddtrace/internal/datadog/profiling/dd_wrapper/include/profiler_stats.hpp +++ b/ddtrace/internal/datadog/profiling/dd_wrapper/include/profiler_stats.hpp @@ -40,6 +40,12 @@ class ProfilerStats // Sticky: fell back to syscall copy (init failure, foreign handler, etc.) std::optional fast_copy_memory_syscall_fallback; + // Persistent intent to use safe_memcpy; not toggled by warmup/fallback + std::optional fast_copy_memory_desired; + + // Sticky: foreign handler owns SIGSEGV/SIGBUS; blocks reclaim and re-warm + std::optional fast_copy_memory_foreign_takeover; + // Number of copy_memory errors accumulated since the last profile reset (i.e. since the last upload) size_t copy_memory_error_count = 0; @@ -49,8 +55,7 @@ class ProfilerStats // Samples dropped because the cap was reached (cumulative over tracker lifetime) std::optional heap_tracker_cap_drops; - // Peak number of asyncio tasks seen across sampled threads in any single sampling - // cycle during the current profile period (see set_asyncio_task_count). + // Number of asyncio tasks seen across sampled threads in the last sampling cycle std::optional asyncio_task_count; // Number of greenlets currently tracked by the stack profiler @@ -87,6 +92,12 @@ class ProfilerStats void set_fast_copy_memory_syscall_fallback(bool fallback); std::optional get_fast_copy_memory_syscall_fallback() const; + void set_fast_copy_memory_desired(bool desired); + std::optional get_fast_copy_memory_desired() const; + + void set_fast_copy_memory_foreign_takeover(bool takeover); + std::optional get_fast_copy_memory_foreign_takeover() const; + // fast_copy_memory_* are process-static; carry them across ProfilerStats swaps. void copy_fast_copy_metadata_from(const ProfilerStats& other); diff --git a/ddtrace/internal/datadog/profiling/dd_wrapper/src/profiler_stats.cpp b/ddtrace/internal/datadog/profiling/dd_wrapper/src/profiler_stats.cpp index f0c74973cf0..45efc24726e 100644 --- a/ddtrace/internal/datadog/profiling/dd_wrapper/src/profiler_stats.cpp +++ b/ddtrace/internal/datadog/profiling/dd_wrapper/src/profiler_stats.cpp @@ -1,5 +1,7 @@ #include "profiler_stats.hpp" +#include "profiler_state.hpp" + #include namespace { @@ -12,16 +14,37 @@ append_to_string(std::string& s, size_t value) s.append(buf, ptr); } -void -append_optional_bool(std::string& s, const char* key, const std::optional& value) +std::optional +resolve_fast_copy_bool(const std::optional& stat_value, bool snapshot_value, bool snapshot_initialized) { - if (!value.has_value()) { - return; + if (stat_value.has_value()) { + return stat_value; } + if (snapshot_initialized) { + return snapshot_value; + } + return std::nullopt; +} + +bool +resolve_fast_copy_bool_or_default(const std::optional& stat_value, + bool snapshot_value, + bool snapshot_initialized, + bool default_value) +{ + if (const auto resolved = resolve_fast_copy_bool(stat_value, snapshot_value, snapshot_initialized)) { + return *resolved; + } + return default_value; +} + +void +append_bool(std::string& s, const char* key, bool value) +{ s += '"'; s += key; s += "\": "; - s += *value ? "true" : "false"; + s += value ? "true" : "false"; s += ','; } @@ -115,6 +138,30 @@ Datadog::ProfilerStats::get_fast_copy_memory_syscall_fallback() const return fast_copy_memory_syscall_fallback; } +void +Datadog::ProfilerStats::set_fast_copy_memory_desired(bool desired) +{ + fast_copy_memory_desired = desired; +} + +std::optional +Datadog::ProfilerStats::get_fast_copy_memory_desired() const +{ + return fast_copy_memory_desired; +} + +void +Datadog::ProfilerStats::set_fast_copy_memory_foreign_takeover(bool takeover) +{ + fast_copy_memory_foreign_takeover = takeover; +} + +std::optional +Datadog::ProfilerStats::get_fast_copy_memory_foreign_takeover() const +{ + return fast_copy_memory_foreign_takeover; +} + void Datadog::ProfilerStats::copy_fast_copy_metadata_from(const ProfilerStats& other) { @@ -127,6 +174,12 @@ Datadog::ProfilerStats::copy_fast_copy_metadata_from(const ProfilerStats& other) if (auto value = other.get_fast_copy_memory_syscall_fallback()) { set_fast_copy_memory_syscall_fallback(*value); } + if (auto value = other.get_fast_copy_memory_desired()) { + set_fast_copy_memory_desired(*value); + } + if (auto value = other.get_fast_copy_memory_foreign_takeover()) { + set_fast_copy_memory_foreign_takeover(*value); + } if (auto value = other.get_fast_copy_memory_enabled()) { set_fast_copy_memory_enabled(*value); } @@ -262,17 +315,40 @@ Datadog::ProfilerStats::get_internal_metadata_json() append_to_string(internal_metadata_json, sampling_event_count); internal_metadata_json += ","; - auto maybe_fast_copy_enabled = get_fast_copy_memory_enabled(); - if (maybe_fast_copy_enabled) { - internal_metadata_json += R"("fast_copy_memory_enabled": )"; - internal_metadata_json += *maybe_fast_copy_enabled ? "true" : "false"; - internal_metadata_json += ","; - } - - append_optional_bool(internal_metadata_json, "fast_copy_memory_user_disabled", fast_copy_memory_user_disabled); - append_optional_bool(internal_metadata_json, "fast_copy_memory_capable", fast_copy_memory_capable); - append_optional_bool( - internal_metadata_json, "fast_copy_memory_syscall_fallback", fast_copy_memory_syscall_fallback); + const auto& fast_copy_snapshot = ProfilerState::get().fast_copy_metadata; + append_bool( + internal_metadata_json, + "fast_copy_memory_enabled", + resolve_fast_copy_bool_or_default( + fast_copy_memory_enabled, fast_copy_snapshot.enabled, fast_copy_snapshot.snapshot_initialized, false)); + append_bool(internal_metadata_json, + "fast_copy_memory_user_disabled", + resolve_fast_copy_bool_or_default(fast_copy_memory_user_disabled, + fast_copy_snapshot.user_disabled, + fast_copy_snapshot.snapshot_initialized, + false)); + append_bool( + internal_metadata_json, + "fast_copy_memory_capable", + resolve_fast_copy_bool_or_default( + fast_copy_memory_capable, fast_copy_snapshot.capable, fast_copy_snapshot.snapshot_initialized, false)); + append_bool(internal_metadata_json, + "fast_copy_memory_syscall_fallback", + resolve_fast_copy_bool_or_default(fast_copy_memory_syscall_fallback, + fast_copy_snapshot.syscall_fallback, + fast_copy_snapshot.snapshot_initialized, + false)); + append_bool( + internal_metadata_json, + "fast_copy_memory_desired", + resolve_fast_copy_bool_or_default( + fast_copy_memory_desired, fast_copy_snapshot.desired, fast_copy_snapshot.snapshot_initialized, false)); + append_bool(internal_metadata_json, + "fast_copy_memory_foreign_takeover", + resolve_fast_copy_bool_or_default(fast_copy_memory_foreign_takeover, + fast_copy_snapshot.foreign_takeover, + fast_copy_snapshot.snapshot_initialized, + false)); auto maybe_heap_tracker_count = get_heap_tracker_size(); if (maybe_heap_tracker_count) { diff --git a/ddtrace/internal/datadog/profiling/stack/__init__.pyi b/ddtrace/internal/datadog/profiling/stack/__init__.pyi index b41e62760f2..679e0dd7576 100644 --- a/ddtrace/internal/datadog/profiling/stack/__init__.pyi +++ b/ddtrace/internal/datadog/profiling/stack/__init__.pyi @@ -10,13 +10,8 @@ from ddtrace._trace import span as ddspan # Core stack v2 functions def start(min_interval: float = ...) -> bool: ... def stop() -> None: ... -def is_origin_task_linking_enabled() -> bool: ... def _native_call_registry_size() -> int: ... -# executor worker thread <-> originating asyncio task association -def link_origin_task(task_id: int, task_name: str) -> None: ... -def unlink_origin_task() -> None: ... - # Sampling configuration def set_adaptive_sampling(do_adaptive_sampling: bool = False) -> None: ... def set_target_overhead(target_overhead: float) -> None: ... @@ -42,9 +37,9 @@ def set_interval(new_interval: float) -> None: ... # Memory copy strategy def set_fast_copy(enabled: bool) -> None: ... def is_safe_copy_failed() -> bool: ... -def fast_copy_memory_active() -> bool: ... # test introspection: is safe_memcpy active? +def fast_copy_memory_active() -> bool: ... -# _set_fast_copy_warmup_seconds is test-only; accessed via _stack (import * skips it). +# Test-only APIs live on _stack (import * skips them). def uninstall_segv_handler() -> None: ... def reinstall_segv_handler() -> None: @@ -57,11 +52,7 @@ def reinstall_segv_handler() -> None: ... def segv_handler_installed() -> bool: - """Return True if our handler is the installed disposition for SIGSEGV and SIGBUS. - - Primarily test introspection: it queries the live disposition via sigaction(2) - for both signals on every call, so it is not free. Do not call it on hot paths. - """ + """True if our SIGSEGV/SIGBUS handlers are installed (test introspection; not cheap).""" ... # Pause/resume sampling diff --git a/ddtrace/internal/datadog/profiling/stack/_stack.pyi b/ddtrace/internal/datadog/profiling/stack/_stack.pyi index 50449e05bb1..0c5d22cd28c 100644 --- a/ddtrace/internal/datadog/profiling/stack/_stack.pyi +++ b/ddtrace/internal/datadog/profiling/stack/_stack.pyi @@ -23,16 +23,18 @@ def set_interval(new_interval: float) -> None: ... # Memory copy strategy def set_fast_copy(enabled: bool) -> None: ... def is_safe_copy_failed() -> bool: ... -def fast_copy_memory_active() -> bool: ... # test introspection: is safe_memcpy active? -def _set_fast_copy_warmup_seconds(seconds: float) -> None: ... # test-only; before start +def fast_copy_memory_active() -> bool: ... +def _sampler_running() -> bool: ... +def _sampling_paused() -> bool: ... +def _take_prefork_pause_observation() -> bool: ... +def _set_fast_copy_warmup_seconds(seconds: float) -> None: ... def segv_handler_installed() -> bool: - """Return True if our handler is the installed disposition for SIGSEGV and SIGBUS. - - Primarily test introspection: it queries the live disposition via sigaction(2) - for both signals on every call, so it is costly. Do not call it on hot paths. - """ + """True if our SIGSEGV/SIGBUS handlers are installed (test introspection; not cheap).""" ... +def uninstall_segv_handler() -> None: ... +def reinstall_segv_handler() -> None: ... + # span <-> profile association def link_span( span_id: int, diff --git a/ddtrace/internal/datadog/profiling/stack/echion/echion/interp.h b/ddtrace/internal/datadog/profiling/stack/echion/echion/interp.h index 4f5bc50d460..e9a1a694022 100644 --- a/ddtrace/internal/datadog/profiling/stack/echion/echion/interp.h +++ b/ddtrace/internal/datadog/profiling/stack/echion/echion/interp.h @@ -30,4 +30,7 @@ class InterpreterInfo }; void -for_each_interp(_PyRuntimeState* runtime, const std::function& callback); +for_each_interp( + _PyRuntimeState* runtime, + const std::function& callback, + const std::function& continue_sampling = []() { return true; }); diff --git a/ddtrace/internal/datadog/profiling/stack/echion/echion/threads.h b/ddtrace/internal/datadog/profiling/stack/echion/echion/threads.h index 6a5b764ccd1..cc90edcd074 100644 --- a/ddtrace/internal/datadog/profiling/stack/echion/echion/threads.h +++ b/ddtrace/internal/datadog/profiling/stack/echion/echion/threads.h @@ -129,7 +129,11 @@ class ThreadInfo // ---------------------------------------------------------------------------- -using PyThreadStateCallback = std::function; +using PyThreadStateCallback = std::function; void -for_each_thread(EchionSampler& echion, InterpreterInfo& interp, const PyThreadStateCallback& callback); +for_each_thread( + EchionSampler& echion, + InterpreterInfo& interp, + const PyThreadStateCallback& callback, + const std::function& continue_sampling = []() { return true; }); diff --git a/ddtrace/internal/datadog/profiling/stack/echion/echion/vm.h b/ddtrace/internal/datadog/profiling/stack/echion/echion/vm.h index a615a6a467d..cfaaf0ab132 100644 --- a/ddtrace/internal/datadog/profiling/stack/echion/echion/vm.h +++ b/ddtrace/internal/datadog/profiling/stack/echion/echion/vm.h @@ -51,8 +51,11 @@ inline kern_return_t (*safe_copy)(vm_map_read_t, #endif -// Whether safe_copy is currently set to the memcpy-based wrapper. -inline bool fast_copy_active = false; +// Transient: safe_memcpy is the active copy path. +inline std::atomic fast_copy_active{ false }; + +// User wants fast copy (set at init); survives warmup toggling fast_copy_active. +inline bool fast_copy_requested = false; // User opted out via _DD_PROFILING_STACK_FAST_COPY or set_fast_copy(false). inline bool fast_copy_user_disabled = false; @@ -66,6 +69,18 @@ mark_fast_copy_syscall_fallback() fast_copy_syscall_fallback = true; } +// Persistent intent; not toggled by warmup/fallback; survives fork. +inline std::atomic fast_copy_desired{ false }; +// Sticky: foreign handler owns SIGSEGV/SIGBUS; blocks reclaim and re-warm. +inline std::atomic fast_copy_foreign_takeover{ false }; + +inline bool +fast_copy_handler_ops_enabled() +{ + return fast_copy_desired.load(std::memory_order_relaxed) && + !fast_copy_foreign_takeover.load(std::memory_order_relaxed); +} + // Set at init; survives toggling fast_copy_active. inline bool safe_memcpy_initialized = false; diff --git a/ddtrace/internal/datadog/profiling/stack/include/sampler.hpp b/ddtrace/internal/datadog/profiling/stack/include/sampler.hpp index 9d3b4490c07..fa1e097a671 100644 --- a/ddtrace/internal/datadog/profiling/stack/include/sampler.hpp +++ b/ddtrace/internal/datadog/profiling/stack/include/sampler.hpp @@ -19,6 +19,8 @@ class EchionSampler; namespace Datadog { +class ProfilerStats; + enum class PauseResult : std::uint8_t { Paused, // sampler was running and is now paused @@ -82,10 +84,6 @@ class Sampler std::vector thread_candidates; void adapt_sampling_interval(); - // Captures one sampling cycle across all threads (or a reservoir-sampled subset thereof - // when max_threads_per_sample is set). - void capture_samples(microsecond_t wall_time_us); - // Rolling window for p_stable: ring buffer of process_delta values (us CPU per adapt window). // p_stable is the p-th percentile of this buffer, giving a stable estimate of app CPU usage // that doesn't collapse to zero during brief idle periods. @@ -100,15 +98,12 @@ class Sampler // Percentile (0..1) used for p_stable; configurable, default p95. double p_stable_percentile_frac = 0.95; - // Fast-copy startup warmup in seconds. double fast_copy_warmup_seconds = 15.0; - // Rolling window duration in seconds; controls the ring buffer capacity. uint32_t p_stable_window_s = 600; - // Tracks whether the sampler was running when prefork was called, - // so that postfork_parent/restart_after_fork can restore it. bool was_running_at_fork_{ false }; + bool paused_for_fork_{ false }; void atfork_child(); friend void stack_atfork_prepare(); @@ -145,6 +140,8 @@ class Sampler // self-time, and we're not currently accounting for the echion self-time. void set_interval(double new_interval); bool is_running() const { return thread_running.load(); } + bool is_sampling_paused() const { return paused_.load(std::memory_order_acquire); } + bool take_prefork_pause_observation(); void set_adaptive_sampling(bool value) { do_adaptive_sampling = value; } void set_target_overhead(double value) { target_overhead = value; } void set_max_sampling_period(microsecond_t max_interval_us) @@ -177,12 +174,13 @@ class Sampler // Restart the sampling thread in the parent after fork void postfork_parent(); - // Restart the sampler after fork if it was running. - // Returns true if start() was invoked and succeeded. - bool restart_after_fork(); + // Restart the sampler after fork if it was running + void restart_after_fork(); }; +// Publish the current echion fast-copy globals into ProfilerState and, +// optionally, the active profile stats object. void -seed_fast_copy_profiler_stats(); +publish_fast_copy_profiler_metadata(Datadog::ProfilerStats* stats = nullptr); } // namespace Datadog diff --git a/ddtrace/internal/datadog/profiling/stack/src/echion/interp.cc b/ddtrace/internal/datadog/profiling/stack/src/echion/interp.cc index 384d7046fa6..9496640b652 100644 --- a/ddtrace/internal/datadog/profiling/stack/src/echion/interp.cc +++ b/ddtrace/internal/datadog/profiling/stack/src/echion/interp.cc @@ -1,7 +1,9 @@ #include void -for_each_interp(_PyRuntimeState* runtime, const std::function& callback) +for_each_interp(_PyRuntimeState* runtime, + const std::function& callback, + const std::function& continue_sampling) { InterpreterInfo interpreter_info = { 0 }; @@ -16,6 +18,10 @@ for_each_interp(_PyRuntimeState* runtime, const std::function& continue_sampling) { std::unordered_set threads; std::unordered_set seen_threads; @@ -823,6 +826,10 @@ for_each_thread(EchionSampler& echion, InterpreterInfo& interp, const PyThreadSt threads.insert(static_cast(interp.tstate_head)); while (!threads.empty()) { + if (!continue_sampling()) { + break; + } + // Pop the next thread PyThreadState* tstate_addr = *threads.begin(); threads.erase(threads.begin()); diff --git a/ddtrace/internal/datadog/profiling/stack/src/echion/vm.cc b/ddtrace/internal/datadog/profiling/stack/src/echion/vm.cc index a776e7b191c..bc7c26c0725 100644 --- a/ddtrace/internal/datadog/profiling/stack/src/echion/vm.cc +++ b/ddtrace/internal/datadog/profiling/stack/src/echion/vm.cc @@ -54,7 +54,9 @@ init_safe_copy() // Try safe_memcpy (fast path) first. if (init_segv_catcher() == 0) { safe_copy = safe_memcpy_wrapper; - fast_copy_active = true; + fast_copy_active.store(true, std::memory_order_relaxed); + fast_copy_requested = true; + fast_copy_desired.store(true, std::memory_order_relaxed); safe_memcpy_initialized = true; } else { // std::cerr might not have been fully initialized at this point. @@ -80,7 +82,9 @@ init_safe_copy() if (init_segv_catcher() == 0) { safe_copy = safe_memcpy_wrapper; - fast_copy_active = true; + fast_copy_active.store(true, std::memory_order_relaxed); + fast_copy_requested = true; + fast_copy_desired.store(true, std::memory_order_relaxed); safe_memcpy_initialized = true; return; } @@ -98,7 +102,8 @@ set_fast_copy_enabled(bool enabled) // Fast copy enabled: prefer safe_memcpy, fall back to process_vm_readv. if (safe_memcpy_initialized) { safe_copy = safe_memcpy_wrapper; - fast_copy_active = true; + fast_copy_active.store(true, std::memory_order_relaxed); + fast_copy_requested = true; return true; } fprintf(stderr, @@ -112,7 +117,7 @@ set_fast_copy_enabled(bool enabled) #if defined PL_LINUX if (process_vm_readv_available) { safe_copy = process_vm_readv; - fast_copy_active = false; + fast_copy_active.store(false, std::memory_order_relaxed); return true; } fprintf(stderr, @@ -123,7 +128,7 @@ set_fast_copy_enabled(bool enabled) #elif defined PL_DARWIN // mach_vm_read_overwrite is always available on macOS. safe_copy = mach_vm_read_overwrite; - fast_copy_active = false; + fast_copy_active.store(false, std::memory_order_relaxed); return true; #endif } diff --git a/ddtrace/internal/datadog/profiling/stack/src/sampler.cpp b/ddtrace/internal/datadog/profiling/stack/src/sampler.cpp index c0f36cc2313..fa9d641ddfe 100644 --- a/ddtrace/internal/datadog/profiling/stack/src/sampler.cpp +++ b/ddtrace/internal/datadog/profiling/stack/src/sampler.cpp @@ -2,8 +2,8 @@ #include "constants.hpp" #include "dd_wrapper/include/profiler_state.hpp" +#include "dd_wrapper/include/profiler_stats.hpp" #include "dd_wrapper/include/sample.hpp" -#include "origin_task_links.hpp" #include "thread_span_links.hpp" #include "echion/danger.h" @@ -16,6 +16,7 @@ #include "echion/threads.h" #include "echion/vm.h" +#include #include #include #include @@ -25,21 +26,6 @@ using namespace Datadog; -static void -update_fast_copy_stats(ProfilerStats& stats) -{ - stats.set_fast_copy_memory_user_disabled(fast_copy_user_disabled); - stats.set_fast_copy_memory_capable(safe_memcpy_initialized); - stats.set_fast_copy_memory_syscall_fallback(fast_copy_syscall_fallback); - stats.set_fast_copy_memory_enabled(fast_copy_active); -} - -void -Datadog::seed_fast_copy_profiler_stats() -{ - update_fast_copy_stats(Sample::profile_borrow().stats()); -} - // Helper class for spawning a std::thread with control over its default stack size #ifdef __linux__ #include @@ -96,6 +82,8 @@ create_thread_with_stack(size_t stack_size, Sampler* sampler, uint64_t seq_num) namespace { +std::atomic g_prefork_sampler_paused_observation{ false }; + // Returns the CPU time of the calling thread in microseconds, or 0 on error. uint64_t get_thread_cpu_time_us() @@ -130,6 +118,30 @@ get_thread_cpu_time_us() } // namespace +void +Datadog::publish_fast_copy_profiler_metadata(ProfilerStats* stats) +{ + auto& snapshot = ProfilerState::get().fast_copy_metadata; + snapshot.user_disabled = fast_copy_user_disabled; + snapshot.capable = safe_memcpy_initialized; + snapshot.syscall_fallback = fast_copy_syscall_fallback; + snapshot.enabled = fast_copy_active.load(std::memory_order_relaxed); + snapshot.desired = fast_copy_desired.load(std::memory_order_relaxed); + snapshot.foreign_takeover = fast_copy_foreign_takeover.load(std::memory_order_relaxed); + snapshot.snapshot_initialized = true; + + if (stats == nullptr) { + return; + } + + stats->set_fast_copy_memory_user_disabled(fast_copy_user_disabled); + stats->set_fast_copy_memory_capable(safe_memcpy_initialized); + stats->set_fast_copy_memory_syscall_fallback(fast_copy_syscall_fallback); + stats->set_fast_copy_memory_enabled(fast_copy_active.load(std::memory_order_relaxed)); + stats->set_fast_copy_memory_desired(fast_copy_desired.load(std::memory_order_relaxed)); + stats->set_fast_copy_memory_foreign_takeover(fast_copy_foreign_takeover.load(std::memory_order_relaxed)); +} + void Sampler::adapt_sampling_interval() { @@ -253,110 +265,13 @@ Sampler::adapt_sampling_interval() sampler_thread_count = new_sampler_thread_count; } -void -Sampler::capture_samples(const microsecond_t wall_time_us) -{ - auto* const runtime = &_PyRuntime; - - // When max_threads_per_sample is set, we collect all threads first, then apply - // reservoir sampling (Algorithm R) to select a uniform random subset, and only - // sample the selected threads. This caps the O(n_threads) stack-unwinding cost. - if (max_threads_per_sample == 0) { - for_each_interp(runtime, [&](InterpreterInfo& interp) -> void { - for_each_thread(*echion, interp, [&](PyThreadState* tstate, ThreadInfo& thread) { - auto success = thread.sample(*echion, tstate, wall_time_us); - if (success) { - Sample::profile_borrow().stats().increment_sample_count(); - } - }); - }); - } else { - thread_candidates.clear(); - - for_each_interp(runtime, [&](InterpreterInfo& interp) -> void { - for_each_thread(*echion, interp, [&](PyThreadState* tstate, ThreadInfo& /*thread*/) { - thread_candidates.push_back(*tstate); - }); - }); - - // Algorithm R: if we have more threads than the cap, select a uniform random subset. - // Selected threads are placed in [0, sample_count). Overflow threads remain in - // [sample_count, size) as fallbacks in case a selected thread was unregistered - // between collection and sampling. - // We use Algorithm R rather than the asymptotically faster Algorithm L because we - // already traverse all threads unconditionally above (the CPython thread list is a - // linked list, so discovery always costs O(n)). Algorithm L's advantage is skipping - // elements to reduce random-number generation, but that only pays off when iteration - // itself is expensive — it isn't here. Algorithm R is simpler and sufficient. - size_t sample_count = thread_candidates.size(); - if (sample_count > max_threads_per_sample) { - for (size_t i = max_threads_per_sample; i < sample_count; i++) { - std::uniform_int_distribution dist(0, i); - size_t j = dist(rng); - if (j < max_threads_per_sample) { - std::swap(thread_candidates[j], thread_candidates[i]); - } - } - sample_count = max_threads_per_sample; - } - - // Apply inverse-probability weighting: each sampled thread represents n/k threads, - // so scale wall_time_us up to preserve correct absolute wall-time totals. - // Note: If a thread disappears between snapshot collection and sampling, fewer than - // sample_count threads are actually sampled. The weight per sample is pre-computed - // so the total reported wall time can be slightly under the true value under high - // thread churn. This is a rare edge case. - const size_t n_total = thread_candidates.size(); - const microsecond_t effective_wall_time_us = - (sample_count < n_total) - ? wall_time_us * static_cast(n_total) / static_cast(sample_count) - : wall_time_us; - - size_t fallback_idx = sample_count; - for (size_t i = 0; i < sample_count; i++) { - // The lock is acquired per iteration rather than for the whole loop so that new - // threads can register (which also needs this lock) between stack unwinds. Holding - // it for the entire loop would block thread registration for the full sampling cycle. - const std::lock_guard guard(echion->thread_info_map_lock()); - - // The tstate is a snapshot captured earlier, and thread_info_map is re-looked up - // here by thread_id. Under extreme thread churn a pthread_t could theoretically - // be reused between snapshot collection and this lookup (old thread exits, new - // thread registers with same ID), causing the new ThreadInfo to be paired with - // the old tstate. This window is a few microseconds and pthread_t reuse within - // it is unlikely. - auto it = echion->thread_info_map().find(thread_candidates[i].thread_id); - if (it == echion->thread_info_map().end()) { - // Thread was unregistered; try to fill from overflow - for (; fallback_idx < thread_candidates.size(); ++fallback_idx) { - auto fb_it = echion->thread_info_map().find(thread_candidates[fallback_idx].thread_id); - if (fb_it != echion->thread_info_map().end()) { - thread_candidates[i] = thread_candidates[fallback_idx]; - it = fb_it; - // Advance so this candidate isn't reused on the next fallback search - fallback_idx++; - break; - } - } - if (it == echion->thread_info_map().end()) { - continue; - } - } - auto success = it->second->sample(*echion, &thread_candidates[i], effective_wall_time_us); - if (success) { - Sample::profile_borrow().stats().increment_sample_count(); - } - } - } -} - void Sampler::sampling_thread(const uint64_t seq_num) { // Mark thread as running thread_running.store(true); - seed_fast_copy_profiler_stats(); + publish_fast_copy_profiler_metadata(&Sample::profile_borrow().stats()); // (Re)install our SIGSEGV/SIGBUS handlers once, but ONLY if we still own them. // @@ -369,7 +284,7 @@ Sampler::sampling_thread(const uint64_t seq_num) // If a foreign owner is already authoritative, leave it in place and fall back to // the syscall copy rather than reclaiming on top. static std::once_flag segv_handler_once; - if (fast_copy_active) { + if (fast_copy_handler_ops_enabled()) { std::call_once(segv_handler_once, []() { if (segv_handler_installed()) { init_segv_catcher(); @@ -381,25 +296,61 @@ Sampler::sampling_thread(const uint64_t seq_num) auto sample_time_prev = steady_clock::now(); auto interval_adjust_time_prev = sample_time_prev; - // safe_memcpy recovery needs us to own both handlers (PROF-14568): warm up on the - // syscall copy, upgrade only if we still own them, then re-check and fall back. - const bool fast_copy_desired = fast_copy_active; + // Warm up on syscall copy, upgrade if we still own handlers; see vm.h for intent flags. #if defined PL_LINUX const bool syscall_copy_available = process_vm_readv_available; #else - const bool syscall_copy_available = true; // mach_vm_read_overwrite is always available + const bool syscall_copy_available = true; #endif - // Warm up only when fast copy is wanted and a safe fallback path exists to run on. - const bool fast_copy_warmup = fast_copy_desired && syscall_copy_available; + const bool fast_copy_warmup = fast_copy_handler_ops_enabled() && syscall_copy_available; bool fast_copy_upgraded = !fast_copy_warmup; bool handler_fallback_done = false; const auto fast_copy_warmup_deadline = sample_time_prev + duration_cast(duration(fast_copy_warmup_seconds)); if (fast_copy_warmup) { - // Drop to the safe syscall copy for the startup window. set_fast_copy_enabled(false); } + publish_fast_copy_profiler_metadata(&Sample::profile_borrow().stats()); + + // Probe handler ownership once/cycle (+ before each sample); cheap continue during list walks. + bool stop_sampling = false; + bool fast_copy_ownership_ok_this_cycle = false; + auto ensure_fast_copy_safe = [&](bool force_reprobe = false) -> bool { + if (stop_sampling) { + return false; + } + if (!fast_copy_active.load(std::memory_order_relaxed) || handler_fallback_done) { + return true; + } + if (!force_reprobe && fast_copy_ownership_ok_this_cycle) { + return true; + } + if (segv_handler_installed()) { + fast_copy_ownership_ok_this_cycle = true; + return true; + } + // Foreign takeover: fall back and record sticky flag (vm.h). + handler_fallback_done = true; + mark_fast_copy_syscall_fallback(); + fast_copy_ownership_ok_this_cycle = false; + fast_copy_foreign_takeover.store(true, std::memory_order_relaxed); + std::cerr << "ddtrace stack profiler: SIGSEGV/SIGBUS handler was taken over by another " + "component; falling back to syscall-based memory copy to avoid crashing." + << std::endl; + if (!set_fast_copy_enabled(false)) { + std::cerr << "ddtrace stack profiler: no safe memory-copy fallback available; " + "stopping stack sampling to avoid crashing." + << std::endl; + stop_sampling = true; + return false; + } + return true; + }; + + const auto continue_traversal = [&]() -> bool { return !stop_sampling; }; + + auto* const runtime = &_PyRuntime; while (seq_num == thread_seq_num.load()) { // Check if a pause has been requested (e.g., for signal handler swapping). // Block until resumed or the thread is asked to stop. @@ -416,6 +367,8 @@ Sampler::sampling_thread(const uint64_t seq_num) } } + publish_fast_copy_profiler_metadata(&Sample::profile_borrow().stats()); + // Measure CPU time before acquiring the profile lock so lock-wait time // is not counted as sampling overhead. auto sample_capture_cpu_before = get_thread_cpu_time_us(); @@ -424,114 +377,191 @@ Sampler::sampling_thread(const uint64_t seq_num) auto wall_time_us = duration_cast(sample_time_now - sample_time_prev).count(); sample_time_prev = sample_time_now; - // Foreign handler handling (see notes before the loop); faulthandler's - // transient swaps are safe since the sampler is paused around them. - if (fast_copy_desired) { - if (!fast_copy_upgraded) { - // Warmup window: still on the safe syscall copy. Once it elapses, - // upgrade to safe_memcpy only if we still own the handlers. - if (sample_time_now >= fast_copy_warmup_deadline) { - fast_copy_upgraded = true; // decide once - if (segv_handler_installed()) { - set_fast_copy_enabled(true); - } else { - // Another component already owns a handler; stay on the safe - // syscall copy (already active from warmup) for the life of - // the process. - handler_fallback_done = true; - mark_fast_copy_syscall_fallback(); - std::cerr << "ddtrace stack profiler: another component owns the SIGSEGV/SIGBUS " - "handler; keeping the syscall-based memory copy to avoid crashing." - << std::endl; - } - } - } else if (fast_copy_active && !handler_fallback_done && !segv_handler_installed()) { - // A handler was taken over after upgrading; fall back permanently - // (no debounce). This is not free: it pins the process to the slower - // syscall copy for its remaining lifetime, which can meaningfully - // degrade sample quality (e.g. on asyncio workloads). We still prefer - // it over the alternative, which is crashing under a foreign handler. - handler_fallback_done = true; - mark_fast_copy_syscall_fallback(); - std::cerr << "ddtrace stack profiler: SIGSEGV/SIGBUS handler was taken over by another " - "component; falling back to syscall-based memory copy to avoid crashing." - << std::endl; - if (!set_fast_copy_enabled(false)) { - // No safe fallback available (e.g. process_vm_readv blocked), so - // safe_memcpy is still active; reading under a foreign handler would - // crash - stop sampling instead. - std::cerr << "ddtrace stack profiler: no safe memory-copy fallback available; " - "stopping stack sampling to avoid crashing." + // Foreign handler handling; faulthandler swaps are safe while paused. + if (fast_copy_handler_ops_enabled() && !fast_copy_upgraded) { + if (sample_time_now >= fast_copy_warmup_deadline) { + fast_copy_upgraded = true; + if (segv_handler_installed()) { + set_fast_copy_enabled(true); + } else { + handler_fallback_done = true; + mark_fast_copy_syscall_fallback(); + fast_copy_foreign_takeover.store(true, std::memory_order_relaxed); + std::cerr << "ddtrace stack profiler: another component owns the SIGSEGV/SIGBUS " + "handler; keeping the syscall-based memory copy to avoid crashing." << std::endl; - break; } } } - // Reset per-cycle asyncio task accumulator before iterating sampled threads - echion->reset_asyncio_task_count(); - - try { - capture_samples(wall_time_us); - - // Collect greenlet count before acquiring the profile lock to avoid - // holding two locks simultaneously (greenlet lock then profile lock). - size_t greenlet_count; - { - const std::lock_guard guard(echion->greenlet_info_map_lock()); - greenlet_count = echion->greenlet_info_map().size(); - } + fast_copy_ownership_ok_this_cycle = false; + if (!ensure_fast_copy_safe(true)) { + break; + } - // Drain copy_memory errors accumulated since the last sampling cycle. - auto copy_errors = g_copy_memory_error_count.exchange(0, std::memory_order_relaxed); + echion->reset_asyncio_task_count(); - if (do_adaptive_sampling) { - // Adjust the sampling interval at most every second - if (sample_time_now - interval_adjust_time_prev > microseconds(g_adaptive_sampling_interval_us)) { - adapt_sampling_interval(); - interval_adjust_time_prev = sample_time_now; + // When max_threads_per_sample is set, we collect all threads first, then apply + // reservoir sampling (Algorithm R) to select a uniform random subset, and only + // sample the selected threads. This caps the O(n_threads) stack-unwinding cost. + if (max_threads_per_sample == 0) { + for_each_interp( + runtime, + [&](InterpreterInfo& interp) -> void { + for_each_thread( + *echion, + interp, + [&](const PyThreadState* tstate, ThreadInfo& thread) { + if (!ensure_fast_copy_safe(true)) { + return; + } + auto success = thread.sample(*echion, const_cast(tstate), wall_time_us); + if (success) { + Sample::profile_borrow().stats().increment_sample_count(); + } + }, + continue_traversal); + }, + continue_traversal); + } else { + thread_candidates.clear(); + + for_each_interp( + runtime, + [&](InterpreterInfo& interp) -> void { + for_each_thread( + *echion, + interp, + [&](const PyThreadState* tstate, ThreadInfo& /*thread*/) { thread_candidates.push_back(*tstate); }, + continue_traversal); + }, + continue_traversal); + + // Algorithm R: if we have more threads than the cap, select a uniform random subset. + // Selected threads are placed in [0, sample_count). Overflow threads remain in + // [sample_count, size) as fallbacks in case a selected thread was unregistered + // between collection and sampling. + // We use Algorithm R rather than the asymptotically faster Algorithm L because we + // already traverse all threads unconditionally above (the CPython thread list is a + // linked list, so discovery always costs O(n)). Algorithm L's advantage is skipping + // elements to reduce random-number generation, but that only pays off when iteration + // itself is expensive — it isn't here. Algorithm R is simpler and sufficient. + size_t sample_count = thread_candidates.size(); + if (sample_count > max_threads_per_sample) { + for (size_t i = max_threads_per_sample; i < sample_count; i++) { + std::uniform_int_distribution dist(0, i); + size_t j = dist(rng); + if (j < max_threads_per_sample) { + std::swap(thread_candidates[j], thread_candidates[i]); + } } + sample_count = max_threads_per_sample; } - // Measure CPU time before acquiring the profile lock so lock-wait time is - // not counted as sampling overhead. - auto sample_capture_cpu_after = get_thread_cpu_time_us(); - - // Update all end-of-cycle stats under a single borrow so they always land in - // the same upload window as the samples they describe. Without this, the uploader - // could swap cur_profiler_stats between two separate borrow calls, silently - // shifting some counters (including sample_capture_cpu_time_us) into the next window. - { - auto borrow = Sample::profile_borrow(); - - borrow.stats().increment_sampling_event_count(); - borrow.stats().set_string_table_count(echion->string_table().size()); - update_fast_copy_stats(borrow.stats()); - borrow.stats().set_asyncio_task_count(echion->asyncio_task_count()); - borrow.stats().set_greenlet_count(greenlet_count); - - if (copy_errors > 0) { - borrow.stats().add_copy_memory_error_count(copy_errors); + // Apply inverse-probability weighting: each sampled thread represents n/k threads, + // so scale wall_time_us up to preserve correct absolute wall-time totals. + // Note: If a thread disappears between snapshot collection and sampling, fewer than + // sample_count threads are actually sampled. The weight per sample is pre-computed + // so the total reported wall time can be slightly under the true value under high + // thread churn. This is a rare edge case. + const size_t n_total = thread_candidates.size(); + const microsecond_t effective_wall_time_us = + (sample_count < n_total) + ? wall_time_us * static_cast(n_total) / static_cast(sample_count) + : wall_time_us; + + size_t fallback_idx = sample_count; + for (size_t i = 0; i < sample_count; i++) { + // Re-verify handler ownership before running safe_memcpy for this thread. + // Done before taking the thread_info_map lock so a syscall probe never + // extends the lock hold time. + if (!ensure_fast_copy_safe(true)) { + break; } - - size_t cpu_diff = sample_capture_cpu_after - sample_capture_cpu_before; - if (cpu_diff > 0) { - borrow.stats().add_sample_capture_cpu_time_us(cpu_diff); + // The lock is acquired per iteration rather than for the whole loop so that new + // threads can register (which also needs this lock) between stack unwinds. Holding + // it for the entire loop would block thread registration for the full sampling cycle. + const std::lock_guard guard(echion->thread_info_map_lock()); + + // The tstate is a snapshot captured earlier, and thread_info_map is re-looked up + // here by thread_id. Under extreme thread churn a pthread_t could theoretically + // be reused between snapshot collection and this lookup (old thread exits, new + // thread registers with same ID), causing the new ThreadInfo to be paired with + // the old tstate. This window is a few microseconds and pthread_t reuse within + // it is unlikely. + auto it = echion->thread_info_map().find(thread_candidates[i].thread_id); + if (it == echion->thread_info_map().end()) { + // Thread was unregistered; try to fill from overflow + for (; fallback_idx < thread_candidates.size(); ++fallback_idx) { + auto fb_it = echion->thread_info_map().find(thread_candidates[fallback_idx].thread_id); + if (fb_it != echion->thread_info_map().end()) { + thread_candidates[i] = thread_candidates[fallback_idx]; + it = fb_it; + // Advance so this candidate isn't reused on the next fallback search + fallback_idx++; + break; + } + } + if (it == echion->thread_info_map().end()) { + continue; + } + } + auto success = it->second->sample(*echion, &thread_candidates[i], effective_wall_time_us); + if (success) { + Sample::profile_borrow().stats().increment_sample_count(); } } - } catch (const std::exception& e) { - std::cerr << "Unexpected error in sampling thread: " << e.what() << std::endl; + } - // If the exception interrupted a sample mid-build (after render_thread_begin - // but before render_stack_end), return it to the pool instead of leaking it. - echion->renderer().abort_sample(); + // A mid-cycle handler takeover with no safe fallback means we must stop entirely. + if (stop_sampling) { + break; + } - // Mark the sampler inactive so a subsequent fork does not restart this - // sampler that has stopped due to an error (see prefork). - sampler_active_.store(false); + // Collect greenlet count before acquiring the profile lock to avoid + // holding two locks simultaneously (greenlet lock then profile lock). + size_t greenlet_count; + { + const std::lock_guard guard(echion->greenlet_info_map_lock()); + greenlet_count = echion->greenlet_info_map().size(); + } - // Stop the sampling loop. - break; + // Drain copy_memory errors accumulated since the last sampling cycle. + auto copy_errors = g_copy_memory_error_count.exchange(0, std::memory_order_relaxed); + + if (do_adaptive_sampling) { + // Adjust the sampling interval at most every second + if (sample_time_now - interval_adjust_time_prev > microseconds(g_adaptive_sampling_interval_us)) { + adapt_sampling_interval(); + interval_adjust_time_prev = sample_time_now; + } + } + + // Measure CPU time before acquiring the profile lock so lock-wait time is + // not counted as sampling overhead. + auto sample_capture_cpu_after = get_thread_cpu_time_us(); + + // Update all end-of-cycle stats under a single borrow so they always land in + // the same upload window as the samples they describe. Without this, the uploader + // could swap cur_profiler_stats between two separate borrow calls, silently + // shifting some counters (including sample_capture_cpu_time_us) into the next window. + { + auto borrow = Sample::profile_borrow(); + + borrow.stats().increment_sampling_event_count(); + borrow.stats().set_string_table_count(echion->string_table().size()); + publish_fast_copy_profiler_metadata(&borrow.stats()); + borrow.stats().set_asyncio_task_count(echion->asyncio_task_count()); + borrow.stats().set_greenlet_count(greenlet_count); + + if (copy_errors > 0) { + borrow.stats().add_copy_memory_error_count(copy_errors); + } + + size_t cpu_diff = sample_capture_cpu_after - sample_capture_cpu_before; + if (cpu_diff > 0) { + borrow.stats().add_sample_capture_cpu_time_us(cpu_diff); + } } // Before sleeping, check whether the user has called for this thread to die. @@ -592,6 +622,8 @@ Sampler::postfork_child() paused_.store(false); new (&pause_mutex_) std::mutex(); new (&pause_cv_) std::condition_variable(); + paused_for_fork_ = false; + g_prefork_sampler_paused_observation.store(false, std::memory_order_release); // Clear stale echion state (mutexes, maps) from parent process if (echion) { @@ -628,26 +660,38 @@ void Sampler::prefork() { was_running_at_fork_ = sampler_active_.load(); + paused_for_fork_ = false; + if (was_running_at_fork_) { + if (pause() == PauseResult::Paused) { + paused_for_fork_ = true; + g_prefork_sampler_paused_observation.store(true, std::memory_order_release); + } + } } void Sampler::postfork_parent() { - // The parent's sampling thread survives the fork unchanged; no restart needed. - // Calling start() here would launch a second thread and cause a data race on - // EchionSampler state. + // Parent sampling thread survives fork; resume if prefork paused it. + if (paused_for_fork_) { + resume(); + paused_for_fork_ = false; + } } -bool +void Sampler::restart_after_fork() { - // Restart the sampler if it was running before fork. - // We use the saved flag because postfork_child() resets the live sampler - // state (thread_running, etc.) before this runs. + // was_running_at_fork_ uses saved state; prefork() changed thread_seq_num parity. if (was_running_at_fork_) { - return start(); + start(); } - return false; +} + +bool +Sampler::take_prefork_pause_observation() +{ + return g_prefork_sampler_paused_observation.exchange(false, std::memory_order_acq_rel); } static void @@ -671,9 +715,6 @@ stack_postfork_cleanup() // Reset ThreadSpanLinks state (reset locks, clear span-thread mappings) ThreadSpanLinks::postfork_child(); - // Reset OriginTaskLinks state (reset locks, clear origin-task mappings) - OriginTaskLinks::postfork_child(); - // Clear Sampler state (reset locks, clear mappings, etc.) Sampler::get().postfork_child(); } @@ -685,11 +726,7 @@ stack_atfork_child() stack_postfork_cleanup(); // Restart the sampler if it was running before fork. - // OriginTaskLinks was left disabled in postfork_child; re-enable when the - // child sampler is started again. - if (Sampler::get().restart_after_fork()) { - OriginTaskLinks::get_instance().enable(); - } + Sampler::get().restart_after_fork(); } __attribute__((constructor)) void @@ -697,7 +734,6 @@ stack_init() { _set_pid(getpid()); ThreadSpanLinks::postfork_child(); - OriginTaskLinks::postfork_child(); } void diff --git a/ddtrace/internal/datadog/profiling/stack/src/stack.cpp b/ddtrace/internal/datadog/profiling/stack/src/stack.cpp index b5777dace00..10865efa9f5 100644 --- a/ddtrace/internal/datadog/profiling/stack/src/stack.cpp +++ b/ddtrace/internal/datadog/profiling/stack/src/stack.cpp @@ -1,5 +1,6 @@ #include "cast_to_pyfunc.hpp" #include "dd_wrapper/include/profiler_state.hpp" +#include "dd_wrapper/include/sample.hpp" #include "origin_task_links.hpp" #include "python_headers.hpp" #include "sampler.hpp" @@ -33,7 +34,7 @@ stack_start_impl(PyObject* self, PyObject* args, PyObject* kwargs) Py_BEGIN_ALLOW_THREADS; OriginTaskLinks::get_instance().enable(); Py_END_ALLOW_THREADS; - seed_fast_copy_profiler_stats(); + publish_fast_copy_profiler_metadata(&Sample::profile_borrow().stats()); Py_RETURN_TRUE; } Py_RETURN_FALSE; @@ -855,8 +856,13 @@ stack_set_fast_copy(PyObject* Py_UNUSED(self), PyObject* args) const bool want = static_cast(enabled); if (!want) { fast_copy_user_disabled = true; + fast_copy_requested = false; + } else { + fast_copy_requested = true; } + fast_copy_desired.store(want && safe_memcpy_initialized, std::memory_order_relaxed); set_fast_copy_enabled(want); + publish_fast_copy_profiler_metadata(); Py_RETURN_NONE; } @@ -869,7 +875,8 @@ stack_uninstall_segv_handler(PyObject* Py_UNUSED(self), PyObject* Py_UNUSED(args // faulthandler) install its own handler so it doesn't record ours as its // previous handler (which would create a signal-handler cycle). // Follow with stack_reinstall_segv_handler to reinstall on top. - if (fast_copy_active) { + // Gate on intent (not active path); skip after foreign takeover (vm.h). + if (fast_copy_handler_ops_enabled()) { uninstall_segv_handler(); } Py_RETURN_NONE; @@ -882,7 +889,7 @@ stack_reinstall_segv_handler(PyObject* Py_UNUSED(self), PyObject* Py_UNUSED(args // This is used to reclaim the handler after another component (e.g., Python's // faulthandler module) overwrites it. Our handler chains to the previous one // for non-recovery faults, so both systems coexist correctly. - if (fast_copy_active) { + if (fast_copy_handler_ops_enabled()) { init_segv_catcher(); } Py_RETURN_NONE; @@ -912,7 +919,34 @@ stack_is_safe_copy_failed(PyObject* Py_UNUSED(self), PyObject* Py_UNUSED(args)) static PyObject* stack_fast_copy_memory_active(PyObject* Py_UNUSED(self), PyObject* Py_UNUSED(args)) { - if (fast_copy_active) { + if (fast_copy_active.load(std::memory_order_relaxed)) { + Py_RETURN_TRUE; + } + Py_RETURN_FALSE; +} + +static PyObject* +stack_sampler_running(PyObject* Py_UNUSED(self), PyObject* Py_UNUSED(args)) +{ + if (Sampler::get().is_running()) { + Py_RETURN_TRUE; + } + Py_RETURN_FALSE; +} + +static PyObject* +stack_sampling_paused(PyObject* Py_UNUSED(self), PyObject* Py_UNUSED(args)) +{ + if (Sampler::get().is_sampling_paused()) { + Py_RETURN_TRUE; + } + Py_RETURN_FALSE; +} + +static PyObject* +stack_take_prefork_pause_observation(PyObject* Py_UNUSED(self), PyObject* Py_UNUSED(args)) +{ + if (Sampler::get().take_prefork_pause_observation()) { Py_RETURN_TRUE; } Py_RETURN_FALSE; @@ -1010,10 +1044,16 @@ static PyMethodDef stack_methods[] = { stack_fast_copy_memory_active, METH_NOARGS, "Return True if the fast safe_memcpy copy path is currently active" }, + { "_sampler_running", stack_sampler_running, METH_NOARGS, "Test-only: sampler thread running" }, + { "_sampling_paused", stack_sampling_paused, METH_NOARGS, "Test-only: sampler at pause point" }, + { "_take_prefork_pause_observation", + stack_take_prefork_pause_observation, + METH_NOARGS, + "Test-only: prefork paused sampler" }, { "_set_fast_copy_warmup_seconds", stack_set_fast_copy_warmup_seconds, METH_VARARGS, - "Test-only: set the fast-copy startup warmup duration in seconds (before start)" }, + "Test-only: fast-copy warmup seconds (before start)" }, { "uninstall_segv_handler", stack_uninstall_segv_handler, METH_NOARGS, diff --git a/releasenotes/notes/profiling-stack-sampler-foreign-handler-warmup-fork-2c9d4e7a1f6b0853.yaml b/releasenotes/notes/profiling-stack-sampler-foreign-handler-warmup-fork-2c9d4e7a1f6b0853.yaml new file mode 100644 index 00000000000..b1bcca09337 --- /dev/null +++ b/releasenotes/notes/profiling-stack-sampler-foreign-handler-warmup-fork-2c9d4e7a1f6b0853.yaml @@ -0,0 +1,7 @@ +--- +fixes: + - | + profiling: hardens the stack sampler's fault-handler ownership tracking so it + keeps using the faster fault-recovery memory copy across startup warmup and + ``fork()``, and safely stays on the syscall copy (without later reclaiming the + handler) once another component takes over ``SIGSEGV``/``SIGBUS``. diff --git a/tests/profiling/collector/test_copy_memory_stats.py b/tests/profiling/collector/test_copy_memory_stats.py index b979a503eba..d84abdb6315 100644 --- a/tests/profiling/collector/test_copy_memory_stats.py +++ b/tests/profiling/collector/test_copy_memory_stats.py @@ -1,5 +1,9 @@ +import sys + import pytest +from ddtrace.internal.datadog.profiling import stack as _stack_ext + @pytest.mark.subprocess( env=dict( @@ -39,6 +43,10 @@ def test_copy_memory_error_count_present(): assert "fast_copy_memory_syscall_fallback" in metadata, ( f"Missing fast_copy_memory_syscall_fallback in {f}: {metadata}" ) + assert "fast_copy_memory_desired" in metadata, f"Missing fast_copy_memory_desired in {f}: {metadata}" + assert "fast_copy_memory_foreign_takeover" in metadata, ( + f"Missing fast_copy_memory_foreign_takeover in {f}: {metadata}" + ) @pytest.mark.subprocess( @@ -79,8 +87,12 @@ def test_fast_copy_memory_disabled(): ) assert metadata["fast_copy_memory_user_disabled"] is True, metadata assert metadata["fast_copy_memory_syscall_fallback"] is False, metadata + assert metadata["fast_copy_memory_desired"] is False, metadata + assert metadata["fast_copy_memory_foreign_takeover"] is False, metadata +@pytest.mark.skipif(sys.platform == "win32", reason="stack v2 profiler is not available on Windows") +@pytest.mark.skipif(not _stack_ext.is_available, reason="stack v2 native extension not available") @pytest.mark.subprocess( env=dict( DD_PROFILING_OUTPUT_PPROF="/tmp/test_fast_copy_memory_enabled", @@ -90,49 +102,312 @@ def test_fast_copy_memory_disabled(): err=None, ) def test_fast_copy_memory_enabled() -> None: - """Sampler runs on the syscall copy during warmup, then upgrades to safe_memcpy (PROF-14568).""" + """Warmup on syscall copy, then upgrade to safe_memcpy.""" import glob import json import os import time - # Underscore-prefixed, so only on the _stack submodule (`import *` skips it). from ddtrace.internal.datadog.profiling.stack import _stack from ddtrace.profiling import profiler from ddtrace.trace import tracer + from tests.profiling.collector.test_utils import wait_for_fast_copy_upgrade _stack._set_fast_copy_warmup_seconds(1.0) + output_filename = os.environ["DD_PROFILING_OUTPUT_PPROF"] + "." + str(os.getpid()) + p: profiler.Profiler = profiler.Profiler(tracer=tracer) p.start() - # Require warmup (False) before accepting the upgrade (True), so the brief - # constructor-time True isn't mistaken for it. - saw_warmup: bool = False - saw_upgrade: bool = False - deadline: float = time.monotonic() + 10 - while time.monotonic() < deadline: - active: bool = _stack.fast_copy_memory_active() - if not saw_warmup: - if active is False: - saw_warmup = True - elif active is True: - saw_upgrade = True - break - time.sleep(0.05) + saw_warmup, saw_upgrade = wait_for_fast_copy_upgrade(_stack, require_warmup=False) + + # Wait for a metadata upload cycle while the profiler is still running so + # fast_copy_memory_enabled reflects the post-warmup state before stop(). + metadata = None + if saw_upgrade: + meta_deadline = time.monotonic() + 5 + while time.monotonic() < meta_deadline: + files = sorted(glob.glob(output_filename + ".*.internal_metadata.json")) + if files: + with open(files[-1]) as fp: + candidate = json.load(fp) + if candidate.get("fast_copy_memory_enabled") is True: + metadata = candidate + break + time.sleep(0.1) p.stop() - output_filename = os.environ["DD_PROFILING_OUTPUT_PPROF"] + "." + str(os.getpid()) - files = sorted(glob.glob(output_filename + ".*.internal_metadata.json")) - assert files, "Expected at least one internal_metadata.json file" + if metadata is None: + files = sorted(glob.glob(output_filename + ".*.internal_metadata.json")) + assert files, "Expected at least one internal_metadata.json file" + with open(files[-1]) as fp: + metadata = json.load(fp) - with open(files[-1]) as fp: - metadata = json.load(fp) assert metadata["fast_copy_memory_user_disabled"] is False, metadata assert metadata["fast_copy_memory_capable"] is True, metadata assert metadata["fast_copy_memory_syscall_fallback"] is False, metadata assert metadata["fast_copy_memory_enabled"] is True, metadata + assert metadata["fast_copy_memory_desired"] is True, metadata + assert metadata["fast_copy_memory_foreign_takeover"] is False, metadata + + if saw_warmup: + assert saw_upgrade, "Expected the sampler to upgrade to safe_memcpy after warmup" + else: + assert saw_upgrade, "Expected safe_memcpy to stay active when warmup is skipped" + + +@pytest.mark.skipif(sys.platform == "win32", reason="stack v2 profiler is not available on Windows") +@pytest.mark.skipif(not _stack_ext.is_available, reason="stack v2 native extension not available") +@pytest.mark.subprocess( + env=dict( + DD_PROFILING_OUTPUT_PPROF="/tmp/test_fast_copy_reinstall_during_warmup", + DD_PROFILING_UPLOAD_INTERVAL="1", + _DD_PROFILING_STACK_FAST_COPY="1", + ), + out="OK\n", + err=None, +) +def test_fast_copy_reinstall_segv_handler_works_during_warmup() -> None: + """Reinstall reclaims handler during warmup (keys off fast_copy_desired).""" + import signal + import time + + from ddtrace.internal.datadog.profiling.stack import _stack + from ddtrace.profiling import profiler + from ddtrace.trace import tracer + + _stack._set_fast_copy_warmup_seconds(30.0) + + p: profiler.Profiler = profiler.Profiler(tracer=tracer) + p.start() + try: + deadline: float = time.monotonic() + 5 + while time.monotonic() < deadline and _stack.fast_copy_memory_active() is not False: + time.sleep(0.02) + assert _stack.fast_copy_memory_active() is False, "expected the syscall copy during warmup" + assert _stack.segv_handler_installed() is True + + signal.signal(signal.SIGSEGV, signal.SIG_DFL) + assert _stack.segv_handler_installed() is False + _stack.reinstall_segv_handler() + assert _stack.segv_handler_installed() is True, "reinstall must reclaim the handler during warmup" + finally: + p.stop() + + print("OK") + + +@pytest.mark.skipif(sys.platform == "win32", reason="stack v2 profiler is not available on Windows") +@pytest.mark.skipif(not _stack_ext.is_available, reason="stack v2 native extension not available") +@pytest.mark.subprocess( + env=dict( + DD_PROFILING_OUTPUT_PPROF="/tmp/test_fast_copy_takeover_fallback", + DD_PROFILING_UPLOAD_INTERVAL="1", + _DD_PROFILING_STACK_FAST_COPY="1", + ), + out="OK\n", + err=None, +) +def test_fast_copy_falls_back_and_does_not_reclaim_after_takeover() -> None: + """Foreign takeover forces permanent syscall-copy fallback.""" + import glob + import json + import os + import signal + import time + + from ddtrace.internal.datadog.profiling.stack import _stack + from ddtrace.profiling import profiler + from ddtrace.trace import tracer + from tests.profiling.collector.test_utils import wait_for_fast_copy_upgrade + + _stack._set_fast_copy_warmup_seconds(0.5) + + p: profiler.Profiler = profiler.Profiler(tracer=tracer) + p.start() + try: + wait_for_fast_copy_upgrade(_stack) + assert _stack.segv_handler_installed() is True + + signal.signal(signal.SIGSEGV, signal.SIG_DFL) + assert _stack.segv_handler_installed() is False + + deadline = time.monotonic() + 10 + while time.monotonic() < deadline and _stack.fast_copy_memory_active() is not False: + time.sleep(0.02) + assert _stack.fast_copy_memory_active() is False, "sampler did not fall back after the takeover" + + deadline = time.monotonic() + 10 + while time.monotonic() < deadline: + if _stack.fast_copy_memory_active() is False: + _stack.reinstall_segv_handler() + if _stack.segv_handler_installed() is False: + break + time.sleep(0.05) + else: + assert False, "handler must not be reclaimed after a foreign takeover" + + time.sleep(0.5) + assert _stack.fast_copy_memory_active() is False, "fast copy must stay disabled after a foreign takeover" + finally: + p.stop() + + output_filename = os.environ["DD_PROFILING_OUTPUT_PPROF"] + "." + str(os.getpid()) + files = sorted(glob.glob(output_filename + ".*.internal_metadata.json")) + assert files, "Expected at least one internal_metadata.json file" + with open(files[-1]) as fp: + metadata = json.load(fp) + assert metadata["fast_copy_memory_desired"] is True, metadata + assert metadata["fast_copy_memory_enabled"] is False, metadata + assert metadata["fast_copy_memory_syscall_fallback"] is True, metadata + assert metadata["fast_copy_memory_foreign_takeover"] is True, metadata + + print("OK") + + +@pytest.mark.skipif(sys.platform == "win32", reason="fork not supported on Windows") +@pytest.mark.skipif(not _stack_ext.is_available, reason="stack v2 native extension not available") +@pytest.mark.subprocess( + env=dict( + DD_PROFILING_OUTPUT_PPROF="/tmp/test_fast_copy_fork_rewarm", + DD_PROFILING_UPLOAD_INTERVAL="1", + _DD_PROFILING_STACK_FAST_COPY="1", + ), + err=None, +) +def test_fast_copy_rewarms_in_forked_child_during_warmup() -> None: + """Child forked during warmup re-warms into safe_memcpy.""" + import os + import time + + from ddtrace.internal.datadog.profiling.stack import _stack + from ddtrace.profiling import profiler + from ddtrace.trace import tracer + + _stack._set_fast_copy_warmup_seconds(10.0) + + p: profiler.Profiler = profiler.Profiler(tracer=tracer) + p.start() + try: + deadline: float = time.monotonic() + 5 + while time.monotonic() < deadline and _stack.fast_copy_memory_active() is not False: + time.sleep(0.02) + assert _stack.fast_copy_memory_active() is False, "expected syscall copy during warmup before fork" + + pid: int = os.fork() + if pid == 0: + try: + saw_upgrade: bool = False + child_deadline: float = time.monotonic() + 15 + while time.monotonic() < child_deadline: + if _stack.fast_copy_memory_active() is True: + saw_upgrade = True + break + time.sleep(0.05) + if not saw_upgrade: + os._exit(1) + except Exception: + os._exit(1) + os._exit(0) + else: + _, status = os.waitpid(pid, 0) + assert os.WIFEXITED(status), f"child killed by signal {os.WTERMSIG(status)}" + assert os.WEXITSTATUS(status) == 0, "child did not re-warm into fast copy after fork" + finally: + p.stop() + + +@pytest.mark.skipif(sys.platform == "win32", reason="fork not supported on Windows") +@pytest.mark.skipif(not _stack_ext.is_available, reason="stack v2 native extension not available") +@pytest.mark.subprocess( + env=dict( + DD_PROFILING_OUTPUT_PPROF="/tmp/test_fast_copy_fork_takeover", + DD_PROFILING_UPLOAD_INTERVAL="1", + _DD_PROFILING_STACK_FAST_COPY="1", + ), + err=None, +) +def test_forked_child_inherits_foreign_takeover_after_parent_fallback() -> None: + """A child inherits the sticky foreign-takeover flag and stays on the syscall copy.""" + import os + import signal + import time + + from ddtrace.internal.datadog.profiling.stack import _stack + from ddtrace.profiling import profiler + from ddtrace.trace import tracer + from tests.profiling.collector.test_utils import wait_for_fast_copy_upgrade + + _stack._set_fast_copy_warmup_seconds(0.5) + + p: profiler.Profiler = profiler.Profiler(tracer=tracer) + p.start() + try: + wait_for_fast_copy_upgrade(_stack) + + signal.signal(signal.SIGSEGV, signal.SIG_DFL) + assert _stack.segv_handler_installed() is False + + deadline = time.monotonic() + 10 + while time.monotonic() < deadline and _stack.fast_copy_memory_active() is not False: + time.sleep(0.02) + assert _stack.fast_copy_memory_active() is False, "parent did not fall back after takeover" + + pid: int = os.fork() + if pid == 0: + try: + _stack.reinstall_segv_handler() + if _stack.segv_handler_installed() is not False: + os._exit(1) + if _stack.fast_copy_memory_active() is not False: + os._exit(2) + except Exception: + os._exit(3) + os._exit(0) + else: + _, status = os.waitpid(pid, 0) + assert os.WIFEXITED(status) and os.WEXITSTATUS(status) == 0, ( + f"child did not inherit sticky foreign takeover (status={status})" + ) + finally: + p.stop() + + +@pytest.mark.skipif(sys.platform == "win32", reason="fork not supported on Windows") +@pytest.mark.skipif(not _stack_ext.is_available, reason="stack v2 native extension not available") +@pytest.mark.subprocess( + env=dict( + DD_PROFILING_OUTPUT_PPROF="/tmp/test_fork_pauses_sampler", + DD_PROFILING_UPLOAD_INTERVAL="1", + _DD_PROFILING_STACK_FAST_COPY="1", + ), + err=None, +) +def test_fork_pauses_sampler_before_safe_memcpy() -> None: + """prefork() pauses the sampler so fork() does not span safe_memcpy.""" + import os + + from ddtrace.internal.datadog.profiling.stack import _stack + from ddtrace.profiling import profiler + from ddtrace.trace import tracer + from tests.profiling.collector.test_utils import wait_for_fast_copy_upgrade + + _stack._set_fast_copy_warmup_seconds(0.5) + + p: profiler.Profiler = profiler.Profiler(tracer=tracer) + p.start() + try: + wait_for_fast_copy_upgrade(_stack) - assert saw_warmup, "Expected the sampler to run on the syscall copy during the warmup window" - assert saw_upgrade, "Expected the sampler to upgrade to safe_memcpy after warmup" + pid: int = os.fork() + if pid == 0: + os._exit(0) + else: + _, status = os.waitpid(pid, 0) + assert os.WIFEXITED(status) and os.WEXITSTATUS(status) == 0, f"child failed (status={status})" + assert _stack._take_prefork_pause_observation() is True, "prefork did not pause the sampler" + assert _stack._sampling_paused() is False, "parent sampler must resume after fork" + finally: + p.stop() diff --git a/tests/profiling/collector/test_utils.py b/tests/profiling/collector/test_utils.py index 40e369a39ee..b70828e7855 100644 --- a/tests/profiling/collector/test_utils.py +++ b/tests/profiling/collector/test_utils.py @@ -53,6 +53,37 @@ def uvloop_available() -> bool: return False +def wait_for_fast_copy_upgrade(stack: Any, timeout: float = 10.0, *, require_warmup: bool = True) -> tuple[bool, bool]: + """Wait until warmup (syscall copy) then upgrade to safe_memcpy. + + Returns (saw_warmup, saw_upgrade). When process_vm_readv is unavailable the + sampler may skip warmup and stay on safe_memcpy (saw_warmup=False). + """ + import time + + saw_warmup = False + saw_upgrade = False + deadline = time.monotonic() + timeout + while time.monotonic() < deadline: + active = stack.fast_copy_memory_active() + if not saw_warmup: + if active is False: + saw_warmup = True + elif active is True: + saw_upgrade = True + break + time.sleep(0.05) + + # When process_vm_readv is unavailable the sampler skips warmup and stays on safe_memcpy. + if not saw_upgrade and stack.fast_copy_memory_active(): + saw_upgrade = True + + if require_warmup: + assert saw_warmup, "Expected the sampler to run on the syscall copy during the warmup window" + assert saw_upgrade, "Expected the sampler to upgrade to safe_memcpy after warmup" + return saw_warmup, saw_upgrade + + class ProfilerContextManager: def __init__(self) -> None: self.profiler = profiler.Profiler()