Skip to content
Open
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
Original file line number Diff line number Diff line change
Expand Up @@ -1175,6 +1175,49 @@ def test_alternating_threads_status_changes(self):
self.assertEqual(count, 100)
self.assert_samples_equal(samples, collector)

def test_rle_alternating_status_batches_correctly(self):

Copy link
Copy Markdown
Member

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

This nicely checks the batching, but the old code produced the same batches (just slower), so this would pass before the fix too, no? It doesn't really guard against the quadratic behavior coming back. Fine to keep since a timing-based test would be flaky, just noting it.

Copy link
Copy Markdown
Contributor Author

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

Good point — it's really a correctness guard for the append path rather than a timing guard (which would be flaky, as you note), so I kept it. Thanks!

"""A repeat record whose status alternates every sample replays as N
single-status batches with the right cumulative timestamps."""
class BatchCollector:
def __init__(self):
self.batches = []

def collect(self, stack_frames, timestamps_us):
for interp in stack_frames:
for thread in interp.threads:
self.batches.append(
(thread.status, list(timestamps_us))
)

def export(self, filename):
pass

num_samples = 2000
frame = make_frame("rle.py", 42, "rle_func")
with tempfile.NamedTemporaryFile(suffix=".bin", delete=False) as f:
filename = f.name
self.temp_files.append(filename)

writer = BinaryCollector(filename, 1000, compression="none")
expected = []
for i in range(num_samples):
status = THREAD_STATUS_HAS_GIL if i % 2 else 0
ts = 1000 + i
expected.append((status, [ts]))
sample = [
make_interpreter(0, [make_thread(1, [frame], status)])
]
writer.collect(sample, timestamp_us=ts)
writer.export(None)

collector = BatchCollector()
with BinaryReader(filename) as reader:
count = reader.replay_samples(collector)

self.assertEqual(count, num_samples)
self.assertEqual(len(collector.batches), num_samples)
self.assertEqual(collector.batches, expected)


class TestBinaryStress(BinaryFormatTestBase):
"""Randomized stress tests for binary format."""
Expand Down
Original file line number Diff line number Diff line change
@@ -0,0 +1,2 @@
Fix quadratic replay time in the :mod:`profiling.sampling` binary reader when a
profile's run-length-encoded samples alternate thread status.
43 changes: 19 additions & 24 deletions Modules/_remote_debugging/binary_io_reader.c
Original file line number Diff line number Diff line change
Expand Up @@ -28,6 +28,9 @@
/* Progress callback frequency */
#define PROGRESS_CALLBACK_INTERVAL 1000

/* Cap per-batch RLE samples to bound the timestamp list (gh-151378) */
#define MAX_RLE_BATCH_SAMPLES 8192

/* ============================================================================
* BINARY READER IMPLEMENTATION
* ============================================================================ */
Expand Down Expand Up @@ -947,21 +950,6 @@ emit_sample(RemoteDebuggingState *state, PyObject *collector,
return 0;
}

/* Helper to trim timestamp list and emit batch. Returns 0 on success, -1 on error. */
static int
emit_batch(RemoteDebuggingState *state, PyObject *collector,
uint64_t thread_id, uint32_t interpreter_id, uint8_t status,
const uint32_t *frame_indices, size_t stack_depth,
BinaryReader *reader, PyObject *timestamps_list, Py_ssize_t actual_size)
{
/* Trim list to actual size */
if (PyList_SetSlice(timestamps_list, actual_size, PyList_GET_SIZE(timestamps_list), NULL) < 0) {
return -1;
}
return emit_sample(state, collector, thread_id, interpreter_id, status,
frame_indices, stack_depth, reader, timestamps_list);
}

/* Helper to invoke progress callback, returns -1 on error */
static inline int
invoke_progress_callback(PyObject *callback, Py_ssize_t current, uint32_t total)
Expand Down Expand Up @@ -1095,17 +1083,18 @@ binary_reader_replay(BinaryReader *reader, PyObject *collector, PyObject *progre
ts->prev_timestamp += delta;

/* Start new batch on first sample or status change */
if (i == 0 || status != batch_status) {
if (i == 0 || status != batch_status
|| batch_idx >= MAX_RLE_BATCH_SAMPLES) {

Copy link
Copy Markdown
Member

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

Do we have a test that covers the MAX_RLE_BATCH_SAMPLES split? With num_samples = 2000 and alternating status the cap never triggers. A run of >8192 identical-status samples should replay as two batches with the timestamps split between them.

if (timestamps_list) {
int rc = emit_batch(state, collector, thread_id, interpreter_id,
batch_status, ts->current_stack, ts->current_stack_depth,
reader, timestamps_list, batch_idx);
int rc = emit_sample(state, collector, thread_id, interpreter_id,
batch_status, ts->current_stack, ts->current_stack_depth,
reader, timestamps_list);
Py_DECREF(timestamps_list);
if (rc < 0) {
return -1;
}
}
timestamps_list = PyList_New(count - i);
timestamps_list = PyList_New(0);
if (!timestamps_list) {
return -1;
}
Expand All @@ -1118,14 +1107,20 @@ binary_reader_replay(BinaryReader *reader, PyObject *collector, PyObject *progre
Py_DECREF(timestamps_list);
return -1;
}
PyList_SET_ITEM(timestamps_list, batch_idx++, ts_obj);
int append_rc = PyList_Append(timestamps_list, ts_obj);

Copy link
Copy Markdown
Member

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

Now that the list is built to its exact size, the PyList_SetSlice trim in emit_batch is always a no-op, no? We could call emit_sample directly and drop actual_size.

Py_DECREF(ts_obj);
if (append_rc < 0) {
Py_DECREF(timestamps_list);
return -1;
}
batch_idx++;
}

/* Emit final batch */
if (timestamps_list) {
int rc = emit_batch(state, collector, thread_id, interpreter_id,
batch_status, ts->current_stack, ts->current_stack_depth,
reader, timestamps_list, batch_idx);
int rc = emit_sample(state, collector, thread_id, interpreter_id,
batch_status, ts->current_stack, ts->current_stack_depth,
reader, timestamps_list);
Py_DECREF(timestamps_list);
if (rc < 0) {
return -1;
Expand Down
Loading