Skip to content
Draft
Show file tree
Hide file tree
Changes from all commits
Commits
File filter

Filter by extension

Filter by extension

Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
35 changes: 35 additions & 0 deletions Lib/test/test_external_inspection.py
Original file line number Diff line number Diff line change
Expand Up @@ -36,6 +36,7 @@
THREAD_STATUS_ON_CPU = 1 << 1
THREAD_STATUS_UNKNOWN = 1 << 2
THREAD_STATUS_HAS_EXCEPTION = 1 << 4
THREAD_STATUS_MAIN_THREAD = 1 << 5

# Maximum number of retry attempts for operations that may fail transiently
MAX_TRIES = 10
Expand Down Expand Up @@ -2596,6 +2597,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.assertEqual(suspended.status, THREAD_STATUS_MAIN_THREAD)
break

def _run_thread_status_test(self, mode, check_condition):
"""
Common pattern for thread status detection tests.
Expand Down
Original file line number Diff line number Diff line change
@@ -0,0 +1,2 @@
Fix :mod:`profiling.sampling` attributing CPU samples to suspended thread states
when multiple thread states share the same native thread.
5 changes: 4 additions & 1 deletion Modules/_remote_debugging/_remote_debugging.h
Original file line number Diff line number Diff line change
Expand Up @@ -179,6 +179,9 @@ typedef enum _WIN32_THREADSTATE {
#define THREAD_STATUS_HAS_EXCEPTION (1 << 4)
#define THREAD_STATUS_MAIN_THREAD (1 << 5)

// Internal marker, removed before returning the sample.
#define THREAD_STATUS_UNBOUND (1 << 6)

/* Exception cause macro */
#define set_exception_cause(unwinder, exc_type, message) \
do { \
Expand Down Expand Up @@ -686,7 +689,7 @@ extern int get_thread_status(RemoteUnwinderObject *unwinder, uint64_t tid, uint6
extern PyObject* unwind_stack_for_thread(
RemoteUnwinderObject *unwinder,
uintptr_t *current_tstate,
uintptr_t gil_holder_tstate,
int *has_unbound,
uintptr_t gc_frame,
uintptr_t main_thread_tstate,
const RemoteReadPrefetch *prefetch
Expand Down
69 changes: 68 additions & 1 deletion Modules/_remote_debugging/module.c
Original file line number Diff line number Diff line change
Expand Up @@ -650,6 +650,66 @@ read_interp_state_and_maybe_thread_frame(
interp_state_buffer);
}

static int
filter_unbound_cpu_samples(PyObject *result, int skip)
{
PyObject *bound = PySet_New(NULL);
if (bound == NULL) {
return -1;
}
// An unbound stack may still be doing native work after a callback.
// Exclude it only when a bound stack shares its native thread ID.
for (Py_ssize_t i = 0; i < PyList_GET_SIZE(result); i++) {
PyObject *threads = PyStructSequence_GET_ITEM(PyList_GET_ITEM(result, i), 1);
for (Py_ssize_t j = 0; j < PyList_GET_SIZE(threads); j++) {
PyObject *thread = PyList_GET_ITEM(threads, j);
long status = PyLong_AsLong(PyStructSequence_GET_ITEM(thread, 1));
if (!(status & THREAD_STATUS_UNBOUND) &&
PyList_GET_SIZE(PyStructSequence_GET_ITEM(thread, 2)) != 0 &&
PySet_Add(bound, PyStructSequence_GET_ITEM(thread, 0)) < 0) {
goto error;
}
}
}
for (Py_ssize_t i = 0; i < PyList_GET_SIZE(result); i++) {
PyObject *threads = PyStructSequence_GET_ITEM(PyList_GET_ITEM(result, i), 1);
for (Py_ssize_t j = PyList_GET_SIZE(threads); j-- > 0;) {
PyObject *thread = PyList_GET_ITEM(threads, j);
long status = PyLong_AsLong(PyStructSequence_GET_ITEM(thread, 1));
if (!(status & THREAD_STATUS_UNBOUND)) {
continue;
}
status &= ~THREAD_STATUS_UNBOUND;
int found = PySet_Contains(bound, PyStructSequence_GET_ITEM(thread, 0));
if (found < 0) {
goto error;
}
if (found) {
status &= ~THREAD_STATUS_ON_CPU;
}
if (skip && !(status & THREAD_STATUS_ON_CPU)) {
if (PySequence_DelItem(threads, j) < 0) {
goto error;
}
}
else {
PyObject *py_status = PyLong_FromLong(status);
if (py_status == NULL) {
goto error;
}
Py_DECREF(PyStructSequence_GET_ITEM(thread, 1));
PyStructSequence_SET_ITEM(thread, 1, py_status);
}
}
}
Py_DECREF(bound);
return 0;

error:
Py_DECREF(bound);
return -1;
}

/*[clinic input]
@critical_section
_remote_debugging.RemoteUnwinder.get_stack_trace
Expand Down Expand Up @@ -715,6 +775,8 @@ _remote_debugging_RemoteUnwinder_get_stack_trace_impl(RemoteUnwinderObject *self
return NULL;
}

int has_unbound = 0;

// Iterate over all interpreters
uintptr_t current_interpreter = self->interpreter_addr;
while (current_interpreter != 0) {
Expand Down Expand Up @@ -803,7 +865,7 @@ _remote_debugging_RemoteUnwinder_get_stack_trace_impl(RemoteUnwinderObject *self
while (current_tstate != 0) {
uintptr_t prev_tstate = current_tstate;
PyObject* frame_info = unwind_stack_for_thread(self, &current_tstate,
gil_holder_tstate,
&has_unbound,
gc_frame,
main_thread_tstate,
&prefetch);
Expand Down Expand Up @@ -890,6 +952,11 @@ _remote_debugging_RemoteUnwinder_get_stack_trace_impl(RemoteUnwinderObject *self
}
}

if (has_unbound && filter_unbound_cpu_samples(result,
self->mode == PROFILING_MODE_CPU && self->skip_non_matching_threads) < 0) {
Py_CLEAR(result);
}

exit:
// Invalidate cache entries for threads not seen in this sample.
// Only do this every 1024 iterations to avoid performance overhead.
Expand Down
9 changes: 8 additions & 1 deletion Modules/_remote_debugging/threads.c
Original file line number Diff line number Diff line change
Expand Up @@ -332,7 +332,7 @@ PyObject*
unwind_stack_for_thread(
RemoteUnwinderObject *unwinder,
uintptr_t *current_tstate,
uintptr_t gil_holder_tstate,
int *has_unbound,
uintptr_t gc_frame,
uintptr_t main_thread_tstate,
const RemoteReadPrefetch *prefetch
Expand Down Expand Up @@ -611,6 +611,13 @@ unwind_stack_for_thread(
goto error;
}

if ((unwinder->mode == PROFILING_MODE_CPU || unwinder->mode == PROFILING_MODE_ALL) &&
!GET_MEMBER(_thread_status, ts, unwinder->debug_offsets.thread_state.status).bound_gilstate &&
PyList_GET_SIZE(frame_info) != 0) {
status_flags |= THREAD_STATUS_UNBOUND;
*has_unbound = 1;
}

// Always use status_flags
PyObject *py_status = PyLong_FromLong(status_flags);
if (py_status == NULL) {
Expand Down
Loading