gh-159312: Avoid CPU attribution to suspended thread states

Lib/test/test_external_inspection.py

@@ -2596,6 +2596,40 @@ def test_unsupported_platform_error(self):
 
 @requires_remote_subprocess_debugging()
 class TestDetectionOfThreadStatus(RemoteInspectionTestBase):
+    @skip_if_not_supported
+    def test_cpu_status_with_multiple_thread_states(self):
+        import_module("_testinternalcapi")
+        import_module("_hashlib")
+        code = textwrap.dedent("""
+            import _hashlib
+            def active():
+                while True:
+                    _hashlib.pbkdf2_hmac('sha256', b'password', b'salt', 10_000_000)
+            active()
+        """)
+        script = f"""
+            import _testinternalcapi
+            def suspended():
+                sock.sendall(b"ready")
+                _testinternalcapi.exec_interpreter(0, {code!r})
+            suspended()
+        """
+        with self._target_process(script) as (p, sock, _):
+            _wait_for_signal(sock, b"ready")
+            unwinder = RemoteUnwinder(p.pid, all_threads=True, mode=PROFILING_MODE_ALL)
+            for _ in busy_retry(SHORT_TIMEOUT):
+                trace = _get_stack_trace_with_retry(unwinder)
+                active = self._find_thread_with_frame(trace, lambda f: f.funcname == "active")
+                if active is None or not active.status & THREAD_STATUS_ON_CPU:
+                    continue
+                if active.status & THREAD_STATUS_HAS_GIL:
+                    continue
+                suspended = self._find_thread_with_frame(trace, lambda f: f.funcname == "suspended")
+                self.assertIsNotNone(suspended)
+                self.assertEqual(suspended.thread_id, active.thread_id)
+                self.assertFalse(suspended.status & THREAD_STATUS_ON_CPU)
+                break
+
     def _run_thread_status_test(self, mode, check_condition):
         """
         Common pattern for thread status detection tests.

Misc/NEWS.d/next/Library/2026-10-13-18-00-00.gh-issue-159312.Lk3w9a.rst

@@ -0,0 +1,2 @@
+Fix :mod:`profiling.sampling` attributing CPU samples to suspended thread states
+when multiple thread states share the same native thread.

Modules/_remote_debugging/threads.c

@@ -293,6 +293,62 @@ typedef struct {
     unsigned int :24;
 } _thread_status;
 
+static int
+has_bound_stack(RemoteUnwinderObject *unwinder, long tid)
+{
+    uintptr_t interp, tstate;
+    int remaining = 8192;
+    if (read_ptr(unwinder, unwinder->runtime_start_address +
+                 unwinder->debug_offsets.runtime_state.interpreters_head, &interp) < 0) {
+        goto unavailable;
+    }
+    while (interp && remaining-- > 0) {
+        if (read_ptr(unwinder, interp +
+                     unwinder->debug_offsets.interpreter_state.threads_head, &tstate) < 0) {
+            goto unavailable;
+        }
+        while (tstate && remaining-- > 0) {
+            char ts[SIZEOF_THREAD_STATE];
+            if (_Py_RemoteDebug_ReadRemoteMemory(&unwinder->handle, tstate,
+                    unwinder->debug_offsets.thread_state.size, ts) < 0) {
+                goto unavailable;
+            }
+            if (GET_MEMBER(long, ts, unwinder->debug_offsets.thread_state.native_thread_id) == tid &&
+                GET_MEMBER(_thread_status, ts, unwinder->debug_offsets.thread_state.status).bound_gilstate) {
+                uintptr_t frame = GET_MEMBER(uintptr_t, ts, unwinder->debug_offsets.thread_state.current_frame);
+                uintptr_t base = GET_MEMBER(uintptr_t, ts, unwinder->debug_offsets.thread_state.base_frame);
+                while (frame && frame != base && remaining-- > 0) {
+                    char data[SIZEOF_INTERP_FRAME];
+                    if (_Py_RemoteDebug_ReadRemoteMemory(&unwinder->handle, frame,
+                                                        sizeof(data), data) < 0) {
+                        goto unavailable;
+                    }
+                    int valid = is_frame_valid(unwinder, (uintptr_t)data,
+                        GET_MEMBER_NO_TAG(uintptr_t, data, unwinder->debug_offsets.interpreter_frame.executable));
+                    if (valid < 0) {
+                        goto unavailable;
+                    }
+                    if (valid == 1) {
+                        return 1;
+                    }
+                    frame = GET_MEMBER(uintptr_t, data, unwinder->debug_offsets.interpreter_frame.previous);
+                }
+            }
+            tstate = GET_MEMBER(uintptr_t, ts, unwinder->debug_offsets.thread_state.next);
+        }
+        if (read_ptr(unwinder, interp +
+                     unwinder->debug_offsets.interpreter_state.next, &interp) < 0) {
+            goto unavailable;
+        }
+    }
+    return 0;
+
+unavailable:
+    // Keep the sample when remote state changes during the lookup.
+    PyErr_Clear();
+    return 0;
+}
+
 static int
 read_thread_state_and_maybe_frame(
     RemoteUnwinderObject *unwinder,
@@ -501,7 +557,11 @@ unwind_stack_for_thread(
 
     if (cpu_status == THREAD_STATE_UNKNOWN) {
         status_flags |= THREAD_STATUS_UNKNOWN;
-    } else if (cpu_status == THREAD_STATE_RUNNING) {
+    } else if (cpu_status == THREAD_STATE_RUNNING &&
+               (GET_MEMBER(_thread_status, ts, unwinder->debug_offsets.thread_state.status).bound_gilstate ||
+                !has_bound_stack(unwinder, tid))) {
+        // An unbound state may still be doing native work after a callback.
+        // Exclude it only if another bound stack shares the same OS thread.
         status_flags |= THREAD_STATUS_ON_CPU;
     }