gh-138122: Add blocking mode for accurate stack traces in Tachyon

Non-blocking sampling reads process memory while the target continues
running, which can produce torn stacks when generators or coroutines
rapidly switch between yield points. Blocking mode uses atomic process
suspension (task_suspend on macOS, NtSuspendProcess on Windows,
PTRACE_SEIZE on Linux) to stop the target during each sample, ensuring
consistent snapshots.

Use blocking mode with longer intervals (1ms+) to avoid impacting the
target too much. The default non-blocking mode remains best for most
cases since it has zero overhead.

Author: pablogsal

Doc/library/profiling.sampling.rst

@@ -312,6 +312,8 @@ The default configuration works well for most use cases:
      - Disabled
    * - Default for ``--subprocesses``
      - Disabled
+   * - Default for ``--blocking``
+     - Disabled (non-blocking sampling)
 
 
 Sampling interval and duration
@@ -362,6 +364,50 @@ This option is particularly useful when investigating concurrency issues or
 when work is distributed across a thread pool.
 
 
+.. _blocking-mode:
+
+Blocking mode
+-------------
+
+By default, Tachyon reads the target process's memory without stopping it.
+This non-blocking approach is ideal for most profiling scenarios because it
+imposes virtually zero overhead on the target application: the profiled
+program runs at full speed and is unaware it is being observed.
+
+However, non-blocking sampling can occasionally produce incomplete or
+inconsistent stack traces in applications with many generators or coroutines
+that rapidly switch between yield points, or in programs with very fast-changing
+call stacks where functions enter and exit between the start and end of a single
+stack read, resulting in reconstructed stacks that mix frames from different
+execution states or that never actually existed.
+
+For these cases, the :option:`--blocking` option stops the target process during
+each sample::
+
+   python -m profiling.sampling run --blocking script.py
+   python -m profiling.sampling attach --blocking 12345
+
+When blocking mode is enabled, the profiler suspends the target process,
+reads its stack, then resumes it. This guarantees that each captured stack
+represents a real, consistent snapshot of what the process was doing at that
+instant. The trade-off is that the target process runs slower because it is
+repeatedly paused.
+
+.. warning::
+
+   Do not use very high sample rates (low ``--interval`` values) with blocking
+   mode. Suspending and resuming a process takes time, and if the sampling
+   interval is too short, the target will spend more time stopped than running.
+   For blocking mode, intervals of 1000 microseconds (1 millisecond) or higher
+   are recommended. The default 100 microsecond interval may cause noticeable
+   slowdown in the target application.
+
+Use blocking mode only when you observe inconsistent stacks in your profiles,
+particularly with generator-heavy or coroutine-heavy code. For most
+applications, the default non-blocking mode provides accurate results with
+zero impact on the target process.
+
+
 Special frames
 --------------
 
@@ -1296,6 +1342,13 @@ Sampling options
    Also profile subprocesses. Each subprocess gets its own profiler
    instance and output file. Incompatible with ``--live``.
 
+.. option:: --blocking
+
+   Stop the target process during each sample. This ensures consistent
+   stack traces at the cost of slowing down the target. Use with longer
+   intervals (1000 µs or higher) to minimize impact. See :ref:`blocking-mode`
+   for details.
+
 
 Mode options
 ------------

Lib/profiling/sampling/cli.py

@@ -342,6 +342,13 @@ def _add_sampling_options(parser):
         action="store_true",
         help="Also profile subprocesses. Each subprocess gets its own profiler and output file.",
     )
+    sampling_group.add_argument(
+        "--blocking",
+        action="store_true",
+        help="Stop all threads in target process before sampling to get consistent snapshots. "
+        "Uses thread_suspend on macOS and ptrace on Linux. Adds overhead but ensures memory "
+        "reads are from a frozen state.",
+    )
 
 
 def _add_mode_options(parser):
@@ -778,6 +785,7 @@ def _handle_attach(args):
             native=args.native,
             gc=args.gc,
             opcodes=args.opcodes,
+            blocking=args.blocking,
         )
         _handle_output(collector, args, args.pid, mode)
 
@@ -848,6 +856,7 @@ def _handle_run(args):
                 native=args.native,
                 gc=args.gc,
                 opcodes=args.opcodes,
+                blocking=args.blocking,
             )
             _handle_output(collector, args, process.pid, mode)
         finally:
@@ -893,6 +902,7 @@ def _handle_live_attach(args, pid):
         native=args.native,
         gc=args.gc,
         opcodes=args.opcodes,
+        blocking=args.blocking,
     )
 
 
@@ -940,6 +950,7 @@ def _handle_live_run(args):
             native=args.native,
             gc=args.gc,
             opcodes=args.opcodes,
+            blocking=args.blocking,
         )
     finally:
         # Clean up the subprocess

Lib/profiling/sampling/sample.py

@@ -1,4 +1,5 @@
 import _remote_debugging
+import contextlib
 import os
 import pstats
 import statistics
@@ -9,6 +10,21 @@
 from _colorize import ANSIColors
 
 from .pstats_collector import PstatsCollector
+
+
+@contextlib.contextmanager
+def _pause_threads(unwinder, blocking):
+    """Context manager to pause/resume threads around sampling if blocking is True."""
+    if blocking:
+        unwinder.pause_threads()
+        try:
+            yield
+        finally:
+            unwinder.resume_threads()
+    else:
+        yield
+
+
 from .stack_collector import CollapsedStackCollector, FlamegraphCollector
 from .heatmap_collector import HeatmapCollector
 from .gecko_collector import GeckoCollector
@@ -28,12 +44,13 @@
 
 
 class SampleProfiler:
-    def __init__(self, pid, sample_interval_usec, all_threads, *, mode=PROFILING_MODE_WALL, native=False, gc=True, opcodes=False, skip_non_matching_threads=True, collect_stats=False):
+    def __init__(self, pid, sample_interval_usec, all_threads, *, mode=PROFILING_MODE_WALL, native=False, gc=True, opcodes=False, skip_non_matching_threads=True, collect_stats=False, blocking=False):
         self.pid = pid
         self.sample_interval_usec = sample_interval_usec
         self.all_threads = all_threads
         self.mode = mode  # Store mode for later use
         self.collect_stats = collect_stats
+        self.blocking = blocking
         try:
             self.unwinder = self._new_unwinder(native, gc, opcodes, skip_non_matching_threads)
         except RuntimeError as err:
@@ -63,12 +80,11 @@ def sample(self, collector, duration_sec=10, *, async_aware=False):
         running_time = 0
         num_samples = 0
         errors = 0
+        interrupted = False
         start_time = next_time = time.perf_counter()
         last_sample_time = start_time
         realtime_update_interval = 1.0  # Update every second
         last_realtime_update = start_time
-        interrupted = False
-
         try:
             while running_time < duration_sec:
                 # Check if live collector wants to stop
@@ -78,20 +94,22 @@ def sample(self, collector, duration_sec=10, *, async_aware=False):
                 current_time = time.perf_counter()
                 if next_time < current_time:
                     try:
-                        if async_aware == "all":
-                            stack_frames = self.unwinder.get_all_awaited_by()
-                        elif async_aware == "running":
-                            stack_frames = self.unwinder.get_async_stack_trace()
-                        else:
-                            stack_frames = self.unwinder.get_stack_trace()
-                        collector.collect(stack_frames)
-                    except ProcessLookupError:
+                        with _pause_threads(self.unwinder, self.blocking):
+                            if async_aware == "all":
+                                stack_frames = self.unwinder.get_all_awaited_by()
+                            elif async_aware == "running":
+                                stack_frames = self.unwinder.get_async_stack_trace()
+                            else:
+                                stack_frames = self.unwinder.get_stack_trace()
+                            collector.collect(stack_frames)
+                    except ProcessLookupError as e:
                         duration_sec = current_time - start_time
                         break
-                    except (RuntimeError, UnicodeDecodeError, MemoryError, OSError):
+                    except (RuntimeError, UnicodeDecodeError, MemoryError, OSError) as e:
                         collector.collect_failed_sample()
                         errors += 1
                     except Exception as e:
+                        print(e)
                         if not _is_process_running(self.pid):
                             break
                         raise e from None
@@ -303,6 +321,7 @@ def sample(
     native=False,
     gc=True,
     opcodes=False,
+    blocking=False,
 ):
     """Sample a process using the provided collector.
 
@@ -318,6 +337,7 @@ def sample(
         native: Whether to include native frames
         gc: Whether to include GC frames
         opcodes: Whether to include opcode information
+        blocking: Whether to stop all threads before sampling for consistent snapshots
 
     Returns:
         The collector with collected samples
@@ -343,6 +363,7 @@ def sample(
         opcodes=opcodes,
         skip_non_matching_threads=skip_non_matching_threads,
         collect_stats=realtime_stats,
+        blocking=blocking,
     )
     profiler.realtime_stats = realtime_stats
 
@@ -364,6 +385,7 @@ def sample_live(
     native=False,
     gc=True,
     opcodes=False,
+    blocking=False,
 ):
     """Sample a process in live/interactive mode with curses TUI.
 
@@ -379,6 +401,7 @@ def sample_live(
         native: Whether to include native frames
         gc: Whether to include GC frames
         opcodes: Whether to include opcode information
+        blocking: Whether to stop all threads before sampling for consistent snapshots
 
     Returns:
         The collector with collected samples
@@ -404,6 +427,7 @@ def sample_live(
         opcodes=opcodes,
         skip_non_matching_threads=skip_non_matching_threads,
         collect_stats=realtime_stats,
+        blocking=blocking,
     )
     profiler.realtime_stats = realtime_stats
 

Lib/test/test_profiling/test_sampling_profiler/test_blocking.py

@@ -0,0 +1,140 @@
+"""Tests for blocking mode sampling profiler."""
+
+import io
+import unittest
+from unittest import mock
+
+try:
+    import _remote_debugging  # noqa: F401
+    import profiling.sampling
+    import profiling.sampling.sample
+    from profiling.sampling.stack_collector import CollapsedStackCollector
+except ImportError:
+    raise unittest.SkipTest(
+        "Test only runs when _remote_debugging is available"
+    )
+
+from test.support import requires_remote_subprocess_debugging
+
+from .helpers import test_subprocess
+
+# Duration for profiling in tests
+PROFILING_DURATION_SEC = 1
+
+
+@requires_remote_subprocess_debugging()
+class TestBlockingModeStackAccuracy(unittest.TestCase):
+    """Test that blocking mode produces accurate stack traces.
+
+    When using blocking mode, the target process is stopped during sampling.
+    This ensures that we see accurate stack traces where functions appear
+    in the correct caller/callee relationship.
+
+    These tests verify that generator functions are correctly shown at the
+    top of the stack when they are actively executing, and not incorrectly
+    shown under their caller's code.
+    """
+
+    @classmethod
+    def setUpClass(cls):
+        # Test script that uses a generator consumed in a loop.
+        # When consume_generator is on the arithmetic lines (temp1, temp2, etc.),
+        # fibonacci_generator should NOT be in the stack at all.
+        # Line numbers are important here - see ARITHMETIC_LINES below.
+        cls.generator_script = '''
+def fibonacci_generator(n):
+    a, b = 0, 1
+    for _ in range(n):
+        yield a
+        a, b = b, a + b
+
+def consume_generator():
+    gen = fibonacci_generator(10000)
+    for value in gen:
+        temp1 = value + 1
+        temp2 = value * 2
+        temp3 = value - 1
+        result = temp1 + temp2 + temp3
+
+def main():
+    while True:
+        consume_generator()
+
+_test_sock.sendall(b"working")
+main()
+'''
+        # Line numbers of the arithmetic operations in consume_generator.
+        # These are the lines where fibonacci_generator should NOT be in the stack.
+        # The socket injection code adds 7 lines before our script.
+        # temp1 = value + 1  -> line 17
+        # temp2 = value * 2  -> line 18
+        # temp3 = value - 1  -> line 19
+        # result = ...       -> line 20
+        cls.ARITHMETIC_LINES = {17, 18, 19, 20}
+
+    def test_generator_not_under_consumer_arithmetic(self):
+        """Test that fibonacci_generator doesn't appear when consume_generator does arithmetic.
+
+        When consume_generator is executing arithmetic lines (temp1, temp2, etc.),
+        fibonacci_generator should NOT be anywhere in the stack - it's not being
+        called at that point.
+
+        Valid stacks:
+          - consume_generator at 'for value in gen:' line WITH fibonacci_generator
+            at the top (generator is yielding)
+          - consume_generator at arithmetic lines WITHOUT fibonacci_generator
+            (we're just doing math, not calling the generator)
+
+        Invalid stacks (indicate torn/inconsistent reads):
+          - consume_generator at arithmetic lines WITH fibonacci_generator
+            anywhere in the stack
+
+        Note: call_tree is ordered from bottom (index 0) to top (index -1).
+        """
+        with test_subprocess(self.generator_script, wait_for_working=True) as subproc:
+            collector = CollapsedStackCollector(sample_interval_usec=100, skip_idle=False)
+
+            with (
+                io.StringIO() as captured_output,
+                mock.patch("sys.stdout", captured_output),
+            ):
+                profiling.sampling.sample.sample(
+                    subproc.process.pid,
+                    collector,
+                    duration_sec=PROFILING_DURATION_SEC,
+                    blocking=True,
+                )
+
+        # Analyze collected stacks
+        total_samples = 0
+        invalid_stacks = 0
+        arithmetic_samples = 0
+
+        for (call_tree, _thread_id), count in collector.stack_counter.items():
+            total_samples += count
+
+            if not call_tree:
+                continue
+
+            # Find consume_generator in the stack and check its line number
+            for i, (filename, lineno, funcname) in enumerate(call_tree):
+                if funcname == "consume_generator" and lineno in self.ARITHMETIC_LINES:
+                    arithmetic_samples += count
+                    # Check if fibonacci_generator appears anywhere in this stack
+                    func_names = [frame[2] for frame in call_tree]
+                    if "fibonacci_generator" in func_names:
+                        invalid_stacks += count
+                    break
+
+        self.assertGreater(total_samples, 10,
+            f"Expected at least 10 samples, got {total_samples}")
+
+        # We should have some samples on the arithmetic lines
+        self.assertGreater(arithmetic_samples, 0,
+            f"Expected some samples on arithmetic lines, got {arithmetic_samples}")
+
+        self.assertEqual(invalid_stacks, 0,
+            f"Found {invalid_stacks}/{arithmetic_samples} invalid stacks where "
+            f"fibonacci_generator appears in the stack when consume_generator "
+            f"is on an arithmetic line. This indicates torn/inconsistent stack "
+            f"traces are being captured.")

Misc/NEWS.d/next/Library/2025-12-20-02-33-05.gh-issue-138122.m3EF9E.rst

@@ -0,0 +1 @@
+TODO