gh-152721: Fix quadratic RLE replay time in the profiling binary reader (#152722)

* gh-152721: Fix quadratic RLE replay time in the profiling binary reader

* gh-152721: Clarify the batch-list comment

Reword per review: the list is built with append per element, not
pre-sized; note the old alloc(count - i) + trim per batch was O(count^2).

* Cap RLE batch size to bound the per-batch timestamp list

* Move MAX_RLE_BATCH_SAMPLES comment to its own line (keep under 79 cols)

* Drop the now-redundant emit_batch trim and fix the review nits

The batch list is built to its exact size, so the PyList_SetSlice trim was a no-op; call emit_sample directly. Also point the cap comment at the issue (gh-151378) and drop the stale over-long comment.

* gh-152721: Test replay batches beyond the RLE limit

---------

Co-authored-by: Pablo Galindo Salgado <Pablogsal@gmail.com>

Author: tonghuaroot

Lib/test/test_profiling/test_sampling_profiler/test_binary_format.py

@@ -7,6 +7,7 @@
 import struct
 import tempfile
 import unittest
+from unittest import mock
 from collections import defaultdict
 
 from test.support import captured_stderr
@@ -1503,6 +1504,77 @@ 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):
+        """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)
+
+
+    def test_rle_long_run_splits_batches(self):
+        # Construct a single repeat record larger than the writer's buffer.
+        num_samples = 8193
+        filename = self.create_binary_file([], compression="none")
+        data = bytearray(pathlib.Path(filename).read_bytes())
+        record = (struct.pack("=QIB", 1, 0, 0)  # STACK_REPEAT
+                  + b"\x81\x40"  # 8193 as a varint
+                  + b"\x01\x00" * num_samples)  # delta=1, status=0
+        data[64:64] = record
+        struct.pack_into("=Q", data, 12, 0)  # start timestamp
+        struct.pack_into("=Q", data, 28, num_samples)
+        struct.pack_into("=I", data, 36, 1)  # thread count
+        for offset in (40, 48):  # string and frame table offsets
+            old_offset = struct.unpack_from("=Q", data, offset)[0]
+            struct.pack_into("=Q", data, offset, old_offset + len(record))
+        struct.pack_into("=Q", data, len(data) - 24, len(data))
+        pathlib.Path(filename).write_bytes(data)
+
+        collector = mock.Mock()
+        with BinaryReader(filename) as reader:
+            count = reader.replay_samples(collector)
+        batches = [call.args[1] for call in collector.collect.call_args_list]
+        self.assertEqual(count, num_samples)
+        self.assertEqual([len(batch) for batch in batches], [8192, 1])
+        self.assertEqual([ts for batch in batches for ts in batch],
+                         list(range(1, num_samples + 1)))
+
 
 class TestBinaryStress(BinaryFormatTestBase):
     """Randomized stress tests for binary format."""

Misc/NEWS.d/next/Library/2026-07-01-18-00-00.gh-issue-152721.rlequad.rst

@@ -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.

Modules/_remote_debugging/binary_io_reader.c

@@ -33,6 +33,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
  * ============================================================================ */
@@ -1083,21 +1086,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, uint64_t total)
@@ -1226,17 +1214,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) {
                     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;
                     }
@@ -1249,14 +1238,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);
+                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;