diff --git a/ddtrace/internal/datadog/profiling/stack/echion/echion/threads.h b/ddtrace/internal/datadog/profiling/stack/echion/echion/threads.h index fb6aa7bbdc8..0b09d2734cb 100644 --- a/ddtrace/internal/datadog/profiling/stack/echion/echion/threads.h +++ b/ddtrace/internal/datadog/profiling/stack/echion/echion/threads.h @@ -33,6 +33,7 @@ #include class EchionSampler; +class ThreadInfoTaskTraversalTest; class ThreadInfo { @@ -119,6 +120,8 @@ class ThreadInfo }; private: + friend class ThreadInfoTaskTraversalTest; + void reset_cycle_state() noexcept; void render_unwound_stacks(EchionSampler&); [[nodiscard]] Result unwind_tasks(EchionSampler&, PyThreadState*, microsecond_t wall_time_us); diff --git a/ddtrace/internal/datadog/profiling/stack/src/echion/threads.cc b/ddtrace/internal/datadog/profiling/stack/src/echion/threads.cc index 2ffbf2675e9..268842ead46 100644 --- a/ddtrace/internal/datadog/profiling/stack/src/echion/threads.cc +++ b/ddtrace/internal/datadog/profiling/stack/src/echion/threads.cc @@ -453,57 +453,55 @@ ThreadInfo::get_tasks_from_linked_list(EchionSampler& echion, uintptr_t head_add return ErrorKind::TaskInfoError; } - // Copy head node struct from remote memory to local memory - struct llist_node head_node_local; - if (copy_type(reinterpret_cast(head_addr), head_node_local)) { + const size_t tasks_start = tasks.size(); + // This traversal only appends to tasks. On structural failure, remove its partial results while preserving entries + // from earlier sources. + auto fail = [&tasks, tasks_start]() -> Result { + tasks.resize(tasks_start); return ErrorKind::TaskInfoError; - } + }; - // Check if list is empty (head points to itself in circular list) - uintptr_t head_addr_uint = head_addr; - uintptr_t next_as_uint = reinterpret_cast(head_node_local.next); - uintptr_t prev_as_uint = reinterpret_cast(head_node_local.prev); - if (next_as_uint == head_addr_uint && prev_as_uint == head_addr_uint) { - return Result::ok(); + struct llist_node head_node; + if (copy_type(reinterpret_cast(head_addr), head_node)) { + return fail(); } + llist_node current_node = head_node; - struct llist_node current_node = head_node_local; // Start with head node - - // Copied from CPython's _remote_debugging_module.c: MAX_ITERATIONS - const size_t MAX_ITERATIONS = 1 << 16; + constexpr size_t max_iterations = 1 << 16; size_t iteration_count = 0; + uintptr_t current_node_addr = head_addr; + std::unordered_set visited; - // Iterate over linked-list. The linked list is circular, so we stop - // when we're back at head. - while (reinterpret_cast(current_node.next) != head_addr_uint) { - // Safety: prevent infinite loops - if (++iteration_count > MAX_ITERATIONS) { - return ErrorKind::TaskInfoError; + // A valid circular list must return to the expected head within the hard bound without null, repeated, unreadable, + // or backward-inconsistent nodes. Any violation rolls back this source. + while (reinterpret_cast(current_node.next) != head_addr) { + if (++iteration_count > max_iterations || current_node.next == nullptr) { + return fail(); } - if (current_node.next == nullptr) { - return ErrorKind::TaskInfoError; // nullptr pointer - invalid list + const uintptr_t next_node_addr = reinterpret_cast(current_node.next); + if (!visited.insert(next_node_addr).second) { + return fail(); } - uintptr_t next_node_addr = reinterpret_cast(current_node.next); - - // Calculate task_addr from current_node.next - size_t task_node_offset_val = offsetof(TaskObj, task_node); - uintptr_t task_addr_uint = next_node_addr - task_node_offset_val; - - // Create TaskInfo for the task - auto maybe_task_info = TaskInfo::create(echion, reinterpret_cast(task_addr_uint)); - if (maybe_task_info) { - auto& task_info = *maybe_task_info; - if (task_info->loop == reinterpret_cast(this->asyncio_loop)) { - tasks.push_back(std::move(task_info)); - } + struct llist_node next_node; + if (copy_type(reinterpret_cast(next_node_addr), next_node) || + reinterpret_cast(next_node.prev) != current_node_addr) { + return fail(); } - // Read next node from current_node.next into current_node - if (copy_type(reinterpret_cast(next_node_addr), current_node)) { - return ErrorKind::TaskInfoError; // Failed to read next node + const uintptr_t task_addr = next_node_addr - offsetof(TaskObj, task_node); + auto maybe_task = TaskInfo::create(echion, reinterpret_cast(task_addr)); + if (maybe_task && (*maybe_task)->loop == reinterpret_cast(this->asyncio_loop)) { + tasks.push_back(std::move(*maybe_task)); } + + current_node_addr = next_node_addr; + current_node = next_node; + } + + if (reinterpret_cast(head_node.prev) != current_node_addr) { + return fail(); } return Result::ok(); @@ -516,24 +514,19 @@ ThreadInfo::get_all_tasks(EchionSampler& echion, PyThreadState* tstate) if (this->asyncio_loop == 0) return tasks; - // Python 3.14+: Native tasks are in linked-list per thread AND per interpreter - // CPython iterates over both: - // 1. Per-thread list: tstate->asyncio_tasks_head (active tasks) - // 2. Per-interpreter list: interp->asyncio_tasks_head (lingering tasks) - // First, get tasks from this thread's linked-list (if tstate_addr is set) - // Note: We continue processing even if one source fails to maximize partial results + // Python 3.14+ task discovery combines four sources: + // - per-thread linked lists for active native Tasks; + // - the per-interpreter linked list for native Tasks surviving thread-state clearing; + // - _scheduled_tasks for third-party Task implementations; + // - _eager_tasks for Tasks executing their first eager step. + // The stack sampler reads Python threads without acquiring the GIL or stopping them. It can therefore observe a + // Task moving between sources, so deduplicate tasks by address below. + // Continue after one source fails to preserve results from the other sources. if (tstate != nullptr && this->tstate_addr != 0) { (void)get_tasks_from_thread_linked_list(echion, tasks); - - // Second, get tasks from interpreter's linked-list (lingering tasks) (void)get_tasks_from_interpreter_linked_list(echion, tstate, tasks); } - // Handle third-party tasks from Python _scheduled_tasks WeakSet - // In Python 3.14+, _scheduled_tasks is a Python-level weakref.WeakSet() that only contains - // tasks that don't inherit from asyncio.Task. Native asyncio.Task instances are stored - // in linked-lists (handled above) and are NOT added to _scheduled_tasks. - // This is typically empty in practice, but we handle it for completeness. auto asyncio_scheduled_tasks = echion.asyncio_scheduled_tasks(); if (asyncio_scheduled_tasks != nullptr) { if (auto maybe_scheduled_tasks_set = MirrorSet::create(asyncio_scheduled_tasks)) { @@ -577,6 +570,10 @@ ThreadInfo::get_all_tasks(EchionSampler& echion, PyThreadState* tstate) } } + // A Task may appear in multiple sources. Keep the earliest snapshot because it is closest to the thread stack + // captured for this sample. + std::unordered_set seen; + std::erase_if(tasks, [&seen](const TaskInfo::Ptr& task) { return !seen.insert(task->origin).second; }); return tasks; } #else diff --git a/ddtrace/internal/datadog/profiling/stack/test/CMakeLists.txt b/ddtrace/internal/datadog/profiling/stack/test/CMakeLists.txt index 173db9145f2..36850396db4 100644 --- a/ddtrace/internal/datadog/profiling/stack/test/CMakeLists.txt +++ b/ddtrace/internal/datadog/profiling/stack/test/CMakeLists.txt @@ -102,6 +102,9 @@ configure_stack_internal_test(test_sample_lifecycle) dd_wrapper_add_test(test_sampling_cycle_state test_sampling_cycle_state.cpp) configure_stack_internal_test(test_sampling_cycle_state) +dd_wrapper_add_test(test_task_traversal test_task_traversal.cpp) +configure_stack_internal_test(test_task_traversal) + dd_wrapper_add_test(test_alt_stack_ownership test_alt_stack_ownership.cpp) # ThreadAltStack lives in the vendored echion header tree. target_include_directories(test_alt_stack_ownership PRIVATE ../echion) diff --git a/ddtrace/internal/datadog/profiling/stack/test/test_task_traversal.cpp b/ddtrace/internal/datadog/profiling/stack/test/test_task_traversal.cpp new file mode 100644 index 00000000000..433c9c49926 --- /dev/null +++ b/ddtrace/internal/datadog/profiling/stack/test/test_task_traversal.cpp @@ -0,0 +1,144 @@ +#include "echion/echion_sampler.h" +#include "echion/threads.h" + +#include + +class ThreadInfoTaskTraversalTest : public ::testing::Test +{ + protected: +#if PY_VERSION_HEX >= 0x030e0000 + // Keep the production traversal private while allowing deterministic linked-list topologies in this test. + static Result traverse(ThreadInfo& thread, + EchionSampler& echion, + uintptr_t head, + std::vector& tasks) + { + return thread.get_tasks_from_linked_list(echion, head, tasks); + } + + static Result> get_all_tasks(ThreadInfo& thread, + EchionSampler& echion, + PyThreadState* tstate) + { + return thread.get_all_tasks(echion, tstate); + } +#endif +}; + +#if PY_VERSION_HEX >= 0x030e0000 +TEST_F(ThreadInfoTaskTraversalTest, RejectsTaskMovedToAnotherList) +{ + // A real asyncio.Task ensures TaskInfo::create follows the same coroutine and name-reading path as production. + Py_Initialize(); + PyObject* globals = PyDict_New(); + ASSERT_NE(globals, nullptr); + ASSERT_EQ(PyDict_SetItemString(globals, "__builtins__", PyEval_GetBuiltins()), 0); + + PyObject* result = PyRun_String(R"( +import asyncio +loop = asyncio.new_event_loop() +async def wait_forever(): + await asyncio.Event().wait() +valid_task = loop.create_task(wait_forever()) +task = loop.create_task(wait_forever()) +)", + Py_file_input, + globals, + globals); + ASSERT_NE(result, nullptr); + Py_DECREF(result); + + auto* loop = PyDict_GetItemString(globals, "loop"); + auto* valid_task = reinterpret_cast(PyDict_GetItemString(globals, "valid_task")); + auto* task = reinterpret_cast(PyDict_GetItemString(globals, "task")); + ASSERT_NE(loop, nullptr); + ASSERT_NE(valid_task, nullptr); + ASSERT_NE(task, nullptr); + + EchionSampler echion; +#if defined PL_LINUX + ThreadInfo thread(1, 1, "test-thread", CLOCK_THREAD_CPUTIME_ID); +#elif defined PL_DARWIN + ThreadInfo thread(1, 1, "test-thread", mach_thread_self()); +#endif + thread.asyncio_loop = reinterpret_cast(loop); + + // Seed the output to verify a failed source preserves tasks previously found by another source. + std::vector tasks; + auto maybe_task = TaskInfo::create(echion, task); + ASSERT_TRUE(maybe_task); + tasks.push_back(std::move(*maybe_task)); + TaskInfo* sentinel = tasks.front().get(); + + // Model Echion reading A and V from A <-> V <-> T before CPython moves T under head B. Reading T afterward + // produces this mixed-time view: + // + // copied nodes: A -> V -> T + // live task: B <-> T + // + // Traversal appends V before T.prev != V reveals the malformed edge and requires source-local rollback. + const llist_node original_valid_task_node = valid_task->task_node; + const llist_node original_task_node = task->task_node; + llist_node expected_head{}; + llist_node moved_head{}; + expected_head.next = &valid_task->task_node; + expected_head.prev = &task->task_node; + valid_task->task_node.prev = &expected_head; + valid_task->task_node.next = &task->task_node; + moved_head.next = moved_head.prev = &task->task_node; + task->task_node.next = task->task_node.prev = &moved_head; + + result = nullptr; + auto traversal = traverse(thread, echion, reinterpret_cast(&expected_head), tasks); + + // Reject the malformed source and roll back only the entries it appended. + EXPECT_FALSE(traversal); + EXPECT_EQ(tasks.size(), 1); + if (!tasks.empty()) { + EXPECT_EQ(tasks.front().get(), sentinel); + } + + valid_task->task_node = original_valid_task_node; + task->task_node = original_task_node; + tasks.clear(); + + // Expose the same Task through a valid thread list and the eager-task set. Cross-source discovery must still + // return one TaskInfo because downstream accounting and wall-time scaling operate on this result. + PyObject* eager_tasks = PySet_New(nullptr); + ASSERT_NE(eager_tasks, nullptr); + ASSERT_EQ(PySet_Add(eager_tasks, reinterpret_cast(task)), 0); + echion.init_asyncio(nullptr, eager_tasks); + + _PyThreadStateImpl remote_tstate{}; + remote_tstate.asyncio_tasks_head.next = remote_tstate.asyncio_tasks_head.prev = &task->task_node; + task->task_node.next = task->task_node.prev = &remote_tstate.asyncio_tasks_head; + thread.tstate_addr = reinterpret_cast(&remote_tstate); + PyThreadState local_tstate{}; + + auto all_tasks = get_all_tasks(thread, echion, &local_tstate); + + // Restore CPython's real links before cancellation or object destruction can inspect them. + task->task_node = original_task_node; + Py_DECREF(eager_tasks); + ASSERT_TRUE(all_tasks); + EXPECT_EQ(all_tasks->size(), 1); + + // Process cancellation and close the loop so the real Task does not remain pending at process exit. + result = PyRun_String(R"( +for pending in (valid_task, task): + pending.cancel() +for pending in (valid_task, task): + try: + loop.run_until_complete(pending) + except asyncio.CancelledError: + pass +loop.close() +)", + Py_file_input, + globals, + globals); + EXPECT_NE(result, nullptr); + Py_XDECREF(result); + Py_DECREF(globals); +} +#endif diff --git a/releasenotes/notes/fix-profiler-asyncio-task-traversal-754a147329fc90cd.yaml b/releasenotes/notes/fix-profiler-asyncio-task-traversal-754a147329fc90cd.yaml new file mode 100644 index 00000000000..ce74e914c24 --- /dev/null +++ b/releasenotes/notes/fix-profiler-asyncio-task-traversal-754a147329fc90cd.yaml @@ -0,0 +1,4 @@ +--- +fixes: + - | + profiling: Fixes an issue where asyncio tasks can be duplicated in profiles on Python 3.14, causing inflated task counts and wall time and increased profiler CPU usage.