From 369097c3f91a36a504af97ffc62058541d85cba4 Mon Sep 17 00:00:00 2001 From: Taegyun Kim Date: Thu, 20 Aug 2026 16:25:36 +0000 Subject: [PATCH 1/8] test(profiling): reproduce asyncio task list migration --- .../profiling/stack/echion/echion/threads.h | 3 + .../profiling/stack/test/CMakeLists.txt | 3 + .../stack/test/test_task_traversal.cpp | 87 +++++++++++++++++++ 3 files changed, 93 insertions(+) create mode 100644 ddtrace/internal/datadog/profiling/stack/test/test_task_traversal.cpp 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/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..4ee05397a85 --- /dev/null +++ b/ddtrace/internal/datadog/profiling/stack/test/test_task_traversal.cpp @@ -0,0 +1,87 @@ +#include "echion/echion_sampler.h" +#include "echion/threads.h" + +#include + +class ThreadInfoTaskTraversalTest : public ::testing::Test +{ + protected: +#if PY_VERSION_HEX >= 0x030e0000 + static Result + traverse(ThreadInfo& thread, EchionSampler& echion, uintptr_t head, std::vector& tasks) + { + return thread.get_tasks_from_linked_list(echion, head, tasks); + } +#endif +}; + +#if PY_VERSION_HEX >= 0x030e0000 +TEST_F(ThreadInfoTaskTraversalTest, RejectsTaskMovedToAnotherList) +{ + 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() +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* task = reinterpret_cast(PyDict_GetItemString(globals, "task")); + ASSERT_NE(loop, 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); + + std::vector tasks; + auto maybe_task = TaskInfo::create(echion, task); + ASSERT_TRUE(maybe_task); + tasks.push_back(std::move(*maybe_task)); + + const llist_node original_task_node = task->task_node; + llist_node expected_head{}; + llist_node moved_head{}; + expected_head.next = expected_head.prev = &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); + + EXPECT_FALSE(traversal); + EXPECT_EQ(tasks.size(), 1); + + task->task_node = original_task_node; + tasks.clear(); + result = PyRun_String(R"( +task.cancel() +try: + loop.run_until_complete(task) +except asyncio.CancelledError: + pass +loop.close() +)", + Py_file_input, + globals, + globals); + EXPECT_NE(result, nullptr); + Py_XDECREF(result); + Py_DECREF(globals); +} +#endif From 7474aea631489fcdb5e1df78acf7f98506dc1183 Mon Sep 17 00:00:00 2001 From: Taegyun Kim Date: Thu, 20 Aug 2026 16:39:34 +0000 Subject: [PATCH 2/8] fix(profiling): reject inconsistent asyncio task lists --- .../profiling/stack/src/echion/threads.cc | 72 +++++++++---------- .../stack/test/test_task_traversal.cpp | 32 ++++++++- 2 files changed, 64 insertions(+), 40 deletions(-) diff --git a/ddtrace/internal/datadog/profiling/stack/src/echion/threads.cc b/ddtrace/internal/datadog/profiling/stack/src/echion/threads.cc index 2ffbf2675e9..a98f5b14288 100644 --- a/ddtrace/internal/datadog/profiling/stack/src/echion/threads.cc +++ b/ddtrace/internal/datadog/profiling/stack/src/echion/threads.cc @@ -453,57 +453,51 @@ 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(); + 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(); } + struct 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; + 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(); @@ -577,6 +571,8 @@ ThreadInfo::get_all_tasks(EchionSampler& echion, PyThreadState* tstate) } } + 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/test_task_traversal.cpp b/ddtrace/internal/datadog/profiling/stack/test/test_task_traversal.cpp index 4ee05397a85..70e896dcc5f 100644 --- a/ddtrace/internal/datadog/profiling/stack/test/test_task_traversal.cpp +++ b/ddtrace/internal/datadog/profiling/stack/test/test_task_traversal.cpp @@ -7,11 +7,20 @@ class ThreadInfoTaskTraversalTest : public ::testing::Test { protected: #if PY_VERSION_HEX >= 0x030e0000 - static Result - traverse(ThreadInfo& thread, EchionSampler& echion, uintptr_t head, std::vector& tasks) + 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 }; @@ -69,6 +78,25 @@ task = loop.create_task(wait_forever()) task->task_node = original_task_node; tasks.clear(); + + 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); + + task->task_node = original_task_node; + Py_DECREF(eager_tasks); + ASSERT_TRUE(all_tasks); + EXPECT_EQ(all_tasks->size(), 1); + result = PyRun_String(R"( task.cancel() try: From 6560b44f217542708c7c74b6d729597b4ed5a5a9 Mon Sep 17 00:00:00 2001 From: Taegyun Kim Date: Thu, 20 Aug 2026 16:43:49 +0000 Subject: [PATCH 3/8] docs(profiling): add asyncio traversal release note --- .../fix-profiler-asyncio-task-traversal-754a147329fc90cd.yaml | 4 ++++ 1 file changed, 4 insertions(+) create mode 100644 releasenotes/notes/fix-profiler-asyncio-task-traversal-754a147329fc90cd.yaml 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. From e7c8caf2e4d4dc14cf3bf2e91b4635bb9c670e2e Mon Sep 17 00:00:00 2001 From: Taegyun Kim Date: Thu, 20 Aug 2026 17:09:08 +0000 Subject: [PATCH 4/8] docs(profiling): clarify asyncio traversal invariants --- .../profiling/stack/src/echion/threads.cc | 27 ++++++++++--------- .../stack/test/test_task_traversal.cpp | 14 ++++++++++ 2 files changed, 28 insertions(+), 13 deletions(-) diff --git a/ddtrace/internal/datadog/profiling/stack/src/echion/threads.cc b/ddtrace/internal/datadog/profiling/stack/src/echion/threads.cc index a98f5b14288..fd02b3eeb19 100644 --- a/ddtrace/internal/datadog/profiling/stack/src/echion/threads.cc +++ b/ddtrace/internal/datadog/profiling/stack/src/echion/threads.cc @@ -454,6 +454,8 @@ ThreadInfo::get_tasks_from_linked_list(EchionSampler& echion, uintptr_t head_add } 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; @@ -470,6 +472,8 @@ ThreadInfo::get_tasks_from_linked_list(EchionSampler& echion, uintptr_t head_add uintptr_t current_node_addr = head_addr; std::unordered_set visited; + // 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(); @@ -510,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)) { @@ -571,6 +570,8 @@ 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; diff --git a/ddtrace/internal/datadog/profiling/stack/test/test_task_traversal.cpp b/ddtrace/internal/datadog/profiling/stack/test/test_task_traversal.cpp index 70e896dcc5f..8fad0a2807c 100644 --- a/ddtrace/internal/datadog/profiling/stack/test/test_task_traversal.cpp +++ b/ddtrace/internal/datadog/profiling/stack/test/test_task_traversal.cpp @@ -7,6 +7,7 @@ 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, @@ -27,6 +28,7 @@ class ThreadInfoTaskTraversalTest : public ::testing::Test #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); @@ -58,11 +60,18 @@ task = loop.create_task(wait_forever()) #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)); + // Echion copied head A while A <-> T. CPython then moved task T under head B, leaving the copied A.next stale: + // + // copied head: A -> T + // live list: B <-> T + // + // Following live links from T would cycle through T -> B -> T without ever returning to A. const llist_node original_task_node = task->task_node; llist_node expected_head{}; llist_node moved_head{}; @@ -73,12 +82,15 @@ task = loop.create_task(wait_forever()) 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); 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); @@ -92,11 +104,13 @@ task = loop.create_task(wait_forever()) 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"( task.cancel() try: From 4ed528197f2260e44c9e2ccdc07cd3f179c564f4 Mon Sep 17 00:00:00 2001 From: Taegyun Kim Date: Fri, 21 Aug 2026 10:41:03 -0400 Subject: [PATCH 5/8] Update ddtrace/internal/datadog/profiling/stack/src/echion/threads.cc Co-authored-by: Thomas Kowalski --- ddtrace/internal/datadog/profiling/stack/src/echion/threads.cc | 2 +- 1 file changed, 1 insertion(+), 1 deletion(-) diff --git a/ddtrace/internal/datadog/profiling/stack/src/echion/threads.cc b/ddtrace/internal/datadog/profiling/stack/src/echion/threads.cc index fd02b3eeb19..41db3eeefa9 100644 --- a/ddtrace/internal/datadog/profiling/stack/src/echion/threads.cc +++ b/ddtrace/internal/datadog/profiling/stack/src/echion/threads.cc @@ -465,7 +465,7 @@ ThreadInfo::get_tasks_from_linked_list(EchionSampler& echion, uintptr_t head_add if (copy_type(reinterpret_cast(head_addr), head_node)) { return fail(); } - struct llist_node current_node = head_node; + llist_node current_node = head_node; constexpr size_t max_iterations = 1 << 16; size_t iteration_count = 0; From 50a43c3ab08d931b600d05207e312f555913b765 Mon Sep 17 00:00:00 2001 From: Taegyun Kim Date: Fri, 21 Aug 2026 10:41:10 -0400 Subject: [PATCH 6/8] Update ddtrace/internal/datadog/profiling/stack/src/echion/threads.cc Co-authored-by: Thomas Kowalski --- ddtrace/internal/datadog/profiling/stack/src/echion/threads.cc | 2 +- 1 file changed, 1 insertion(+), 1 deletion(-) diff --git a/ddtrace/internal/datadog/profiling/stack/src/echion/threads.cc b/ddtrace/internal/datadog/profiling/stack/src/echion/threads.cc index 41db3eeefa9..a1e94715a74 100644 --- a/ddtrace/internal/datadog/profiling/stack/src/echion/threads.cc +++ b/ddtrace/internal/datadog/profiling/stack/src/echion/threads.cc @@ -514,7 +514,7 @@ ThreadInfo::get_all_tasks(EchionSampler& echion, PyThreadState* tstate) if (this->asyncio_loop == 0) return tasks; - // Python 3.14 task discovery combines four sources: + // 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; From 0e3726cd99a881a7879134d698aa88b104912bdd Mon Sep 17 00:00:00 2001 From: Taegyun Kim Date: Fri, 21 Aug 2026 19:15:39 +0000 Subject: [PATCH 7/8] style(profiling): fix Echion formatting --- ddtrace/internal/datadog/profiling/stack/src/echion/threads.cc | 2 +- 1 file changed, 1 insertion(+), 1 deletion(-) diff --git a/ddtrace/internal/datadog/profiling/stack/src/echion/threads.cc b/ddtrace/internal/datadog/profiling/stack/src/echion/threads.cc index a1e94715a74..268842ead46 100644 --- a/ddtrace/internal/datadog/profiling/stack/src/echion/threads.cc +++ b/ddtrace/internal/datadog/profiling/stack/src/echion/threads.cc @@ -465,7 +465,7 @@ ThreadInfo::get_tasks_from_linked_list(EchionSampler& echion, uintptr_t head_add if (copy_type(reinterpret_cast(head_addr), head_node)) { return fail(); } - llist_node current_node = head_node; + llist_node current_node = head_node; constexpr size_t max_iterations = 1 << 16; size_t iteration_count = 0; From dc5090ffdc89e19883bdfc5f1ddb393f034aa940 Mon Sep 17 00:00:00 2001 From: Taegyun Kim Date: Fri, 21 Aug 2026 20:01:01 +0000 Subject: [PATCH 8/8] test(profiling): verify task traversal rollback --- .../stack/test/test_task_traversal.cpp | 35 +++++++++++++------ 1 file changed, 25 insertions(+), 10 deletions(-) diff --git a/ddtrace/internal/datadog/profiling/stack/test/test_task_traversal.cpp b/ddtrace/internal/datadog/profiling/stack/test/test_task_traversal.cpp index 8fad0a2807c..433c9c49926 100644 --- a/ddtrace/internal/datadog/profiling/stack/test/test_task_traversal.cpp +++ b/ddtrace/internal/datadog/profiling/stack/test/test_task_traversal.cpp @@ -39,6 +39,7 @@ 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, @@ -48,8 +49,10 @@ task = loop.create_task(wait_forever()) 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; @@ -65,17 +68,23 @@ task = loop.create_task(wait_forever()) auto maybe_task = TaskInfo::create(echion, task); ASSERT_TRUE(maybe_task); tasks.push_back(std::move(*maybe_task)); + TaskInfo* sentinel = tasks.front().get(); - // Echion copied head A while A <-> T. CPython then moved task T under head B, leaving the copied A.next stale: + // 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 head: A -> T - // live list: B <-> T + // copied nodes: A -> V -> T + // live task: B <-> T // - // Following live links from T would cycle through T -> B -> T without ever returning to A. + // 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 = expected_head.prev = &task->task_node; + 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; @@ -85,7 +94,11 @@ task = loop.create_task(wait_forever()) // 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(); @@ -112,11 +125,13 @@ task = loop.create_task(wait_forever()) // Process cancellation and close the loop so the real Task does not remain pending at process exit. result = PyRun_String(R"( -task.cancel() -try: - loop.run_until_complete(task) -except asyncio.CancelledError: - pass +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,