refactor(profiling): change how GenInfo::is_running works (#15508)

## Description

This change replicates P403n1x87/echion#202.

This PR changes the way `GenInfo::is_running` works. Previously, it
would indicate whether the _current_ coroutine was on CPU; now, it
indicates whether the current coroutine _or the coroutine it
(recursively) awaits_ is on CPU.
Making that change also allows us to do less work when we check whether
the current coroutine is on CPU or not. Because a Coroutine / Generator
/ `GenInfo` can only be running if it is not awaiting another Generator,
we do not need to compute `is_running` if we have `await != nullptr`
(and we take `await->is_running` for the value in that case).

Making that change makes it easier to check whether a Task is currently
on-CPU; and allows to do less work when we decide how to unwind
`asyncio` Tasks (cf the changes in `TaskInfo` which doesn't need the
`is_on_cpu` method that iterates on the `await` chain anymore).

Note that I checked whether `GenInfo::is_running` was used in any other
way than the one I describe and simplify; it is not, so I do think this
change is safe to make as-is.

**Note** This PR makes sense on its own, but it is in the context of
making P403n1x87/echion#198 simpler.

**Note** This doesn't need a changelog entry as it makes no difference
to the user, it's purely a refactor.

Author: KowalskiThomas

ddtrace/internal/datadog/profiling/stack_v2/echion/echion/tasks.h

@@ -71,11 +71,13 @@ class GenInfo
   public:
     typedef std::unique_ptr<GenInfo> Ptr;
 
-    PyObject* origin = NULL;
-    PyObject* frame = NULL;
+    PyObject* origin = nullptr;
+    PyObject* frame = nullptr;
 
+    // The coroutine awaited by this coroutine, if any
     GenInfo::Ptr await = nullptr;
 
+    // Whether the coroutine, or the coroutine it awaits, is currently running (on CPU)
     bool is_running = false;
 
     [[nodiscard]] static Result<GenInfo::Ptr> create(PyObject* gen_addr);
@@ -149,13 +151,21 @@ GenInfo::create(PyObject* gen_addr)
         }
     }
 
+    // A coroutine awaiting another coroutine is never running itself,
+    // so when the coroutine is awaiting another coroutine, we use the running state of the awaited coroutine.
+    // Otherwise, we use the running state of the coroutine itself.
+    bool is_running = false;
+    if (await) {
+        is_running = await->is_running;
+    } else {
 #if PY_VERSION_HEX >= 0x030b0000
-    auto is_running = (gen.gi_frame_state == FRAME_EXECUTING);
+        is_running = (gen.gi_frame_state == FRAME_EXECUTING);
 #elif PY_VERSION_HEX >= 0x030a0000
-    auto is_running = (frame != NULL) ? _PyFrame_IsExecuting(&f) : false;
+        is_running = (frame != NULL) ? _PyFrame_IsExecuting(&f) : false;
 #else
-    auto is_running = gen.gi_running;
+        is_running = gen.gi_running;
 #endif
+    }
 
     recursion_depth--;
     return std::make_unique<GenInfo>(gen_addr, frame, std::move(await), is_running);
@@ -178,7 +188,7 @@ class TaskInfo
 
     // Information to reconstruct the async stack as best as we can
     TaskInfo::Ptr waiter = nullptr;
-    std::optional<bool> is_on_cpu_ = std::nullopt;
+    bool is_on_cpu = false;
 
     [[nodiscard]] static Result<TaskInfo::Ptr> create(TaskObj*);
     TaskInfo(PyObject* origin, PyObject* loop, GenInfo::Ptr coro, StringTable::Key name, TaskInfo::Ptr waiter)
@@ -187,32 +197,12 @@ class TaskInfo
       , coro(std::move(coro))
       , name(name)
       , waiter(std::move(waiter))
+      , is_on_cpu(this->coro && this->coro->is_running)
     {
     }
 
     [[nodiscard]] static Result<TaskInfo::Ptr> current(PyObject*);
     inline size_t unwind(FrameStack&);
-
-    // Check if any coroutine in the chain is currently running (on CPU)
-    inline bool is_on_cpu()
-    {
-        if (is_on_cpu_.has_value()) {
-            return *is_on_cpu_;
-        }
-
-        auto* last_coro = static_cast<GenInfo*>(nullptr);
-        auto* current_coro = this->coro.get();
-
-        // Check if the innermost coroutine is running.
-        // We don't need to test all the other coroutines as they are awaiting, by definition.
-        while (current_coro) {
-            last_coro = current_coro;
-            current_coro = current_coro->await.get();
-        }
-
-        is_on_cpu_ = last_coro && last_coro->is_running;
-        return *is_on_cpu_;
-    }
 };
 
 inline std::unordered_map<PyObject*, PyObject*> task_link_map;
@@ -242,8 +232,6 @@ TaskInfo::create(TaskObj* task_addr)
         return ErrorKind::TaskInfoGeneratorError;
     }
 
-    auto origin = reinterpret_cast<PyObject*>(task_addr);
-
     auto maybe_name = string_table.key(task.task_name);
     if (!maybe_name) {
         recursion_depth--;
@@ -262,7 +250,8 @@ TaskInfo::create(TaskObj* task_addr)
     }
 
     recursion_depth--;
-    return std::make_unique<TaskInfo>(origin, loop, std::move(*maybe_coro), name, std::move(waiter));
+    return std::make_unique<TaskInfo>(
+      reinterpret_cast<PyObject*>(task_addr), loop, std::move(*maybe_coro), name, std::move(waiter));
 }
 
 // ----------------------------------------------------------------------------

ddtrace/internal/datadog/profiling/stack_v2/echion/echion/threads.h

@@ -252,25 +252,15 @@ ThreadInfo::unwind_tasks()
         }
     }
 
-    // Only one Task can be on CPU at a time.
-    // Since determining if a task is on CPU is somewhat costly, we
-    // stop checking if Tasks are on CPU after seeing the first one.
-    bool on_cpu_task_seen = false;
     for (auto& leaf_task : leaf_tasks) {
-        bool on_cpu = false;
-        if (!on_cpu_task_seen) {
-            on_cpu = leaf_task.get().is_on_cpu();
-            on_cpu_task_seen = on_cpu;
-        }
-
-        auto stack_info = std::make_unique<StackInfo>(leaf_task.get().name, on_cpu);
+        auto stack_info = std::make_unique<StackInfo>(leaf_task.get().name, leaf_task.get().is_on_cpu);
         auto& stack = stack_info->stack;
         for (auto current_task = leaf_task;;) {
             auto& task = current_task.get();
 
             size_t stack_size = task.unwind(stack);
 
-            if (on_cpu) {
+            if (task.is_on_cpu) {
                 // Undo the stack unwinding
                 // TODO[perf]: not super-efficient :(
                 for (size_t i = 0; i < stack_size; i++)