https://github.com/python/cpython/commit/3439df909caff05c0297d370ae4bb4e9cd02ef1a
commit: 3439df909caff05c0297d370ae4bb4e9cd02ef1a
branch: main
author: Maurycy Pawłowski-Wieroński <[email protected]>
committer: pablogsal <[email protected]>
date: 2026-09-24T17:00:23+01:00
summary:
gh-153804: `_remote_debugging`: Tachyon Oracle (#153806)
* tachyon oracle
* news
* news in tools, not library
* subprocess.DEVNULL
* validate args.run
* do not sample past deadline
* show raw_tvd
* do not loop forever
* comment on the limitation
* docstring
* simple gen testing
* test all classifiers
* better no lineno
* test both ways
* merge test_classify_gen
* time.perf_counter
* docstring
* -stderr=subprocess.DEVNULL,
files:
A Lib/test/test_tools/test_inspection.py
A Misc/NEWS.d/next/Tools-Demos/2026-07-16-11-38-15.gh-issue-153804.KSTYg7.rst
A Tools/inspection/oracle_external_inspection.py
A Tools/inspection/snippets.py
M Tools/inspection/benchmark_external_inspection.py
diff --git a/Lib/test/test_tools/test_inspection.py
b/Lib/test/test_tools/test_inspection.py
new file mode 100644
index 00000000000000..f4a0d163adbc56
--- /dev/null
+++ b/Lib/test/test_tools/test_inspection.py
@@ -0,0 +1,111 @@
+"""Tests for snippets in Tools/inspection."""
+
+import unittest
+from types import SimpleNamespace
+
+from test.test_tools import imports_under_tool, skip_if_missing
+
+
+skip_if_missing("inspection")
+with imports_under_tool("inspection"):
+ import snippets
+
+
+def frame(funcname, *, filename="", lineno=None):
+ location = SimpleNamespace(lineno=lineno) if lineno is not None else None
+ return SimpleNamespace(
+ funcname=funcname,
+ filename=filename,
+ location=location,
+ )
+
+
+def frames(*names):
+ return [frame(name) for name in names]
+
+
+class ClassifierTests(unittest.TestCase):
+ def test_classifiers(self):
+ flat_lines = snippets.FLAT_ALTERNATING_LINES
+ short_line = min(snippets.SHARED_LEAF_SHORT_LINES)
+ flat_a = [
+ frame("leaf_a", lineno=flat_lines["leaf_a"]),
+ frame("hot_a", lineno=flat_lines["hot_a"]),
+ ]
+ flat_crossed = [
+ frame("hot_a", lineno=flat_lines["hot_a"]),
+ frame("hot_b", lineno=flat_lines["hot_b"]),
+ ]
+ shared_a = [
+ frame("shared_leaf", lineno=short_line),
+ frame("a_wrapper"),
+ ]
+ shared_crossed = [
+ frame("shared_leaf", lineno=short_line),
+ frame("b_wrapper"),
+ ]
+ cases = [
+ (snippets.classify_flat, flat_a, False),
+ (snippets.classify_flat, flat_crossed, True),
+ (
+ snippets.classify_nested,
+ frames("burn_a", "a_leaf", "a_parent"),
+ False,
+ ),
+ (
+ snippets.classify_nested,
+ frames("a_parent", "a_leaf", "burn_a"),
+ True,
+ ),
+ (snippets.classify_shared, shared_a, False),
+ (snippets.classify_shared, shared_crossed, True),
+ (snippets.classify_gen, frames("agen", "drv_a"), False),
+ (snippets.classify_gen, frames("agen", "drv_b"), True),
+ (snippets.classify_gen, frames("bgen", "drv_a"), True),
+ (snippets.classify_gen, frames("agen"), False),
+ (
+ snippets.classify_gen,
+ frames("agen", "agen", "drv_a"),
+ False,
+ ),
+
+ (snippets.classify_recursion, frames("a", "a"), False),
+ (snippets.classify_recursion, frames("a", "b"), True),
+ (
+ snippets.classify_async_running_task,
+ (None, "hot", None, frames("leaf_hot")),
+ False,
+ ),
+ (
+ snippets.classify_async_running_task,
+ (None, "hot", None, frames("leaf_rare")),
+ True,
+ ),
+ (
+ snippets.classify_code_object_reuse,
+ [frame("func_a", filename="A_file.py")],
+ False,
+ ),
+ (
+ snippets.classify_code_object_reuse,
+ [frame("func_a", filename="B_file.py")],
+ True,
+ ),
+ (
+ snippets.classify_oversized_chunk,
+ [frame("big_a", filename="a.py")],
+ False,
+ ),
+ (
+ snippets.classify_oversized_chunk,
+ [frame("big_b", filename="a.py")],
+ True,
+ ),
+ ]
+ for classifier, sample, expected in cases:
+ with self.subTest(classifier=classifier.__name__, sample=sample):
+ self.assertEqual(classifier(sample), expected)
+
+
+if __name__ == "__main__":
+ unittest.main()
diff --git
a/Misc/NEWS.d/next/Tools-Demos/2026-07-16-11-38-15.gh-issue-153804.KSTYg7.rst
b/Misc/NEWS.d/next/Tools-Demos/2026-07-16-11-38-15.gh-issue-153804.KSTYg7.rst
new file mode 100644
index 00000000000000..7d0f287a72445b
--- /dev/null
+++
b/Misc/NEWS.d/next/Tools-Demos/2026-07-16-11-38-15.gh-issue-153804.KSTYg7.rst
@@ -0,0 +1,4 @@
+Add ``Tools/inspection/oracle_external_inspection.py``, a harness for
+measuring the accuracy of Tachyon. It reports impossible-stack rates, speed,
+error rates and the statistical distance from the blocking non-cached best
+reference on selected snippets. Patch by Maurycy Pawłowski-Wieroński.
diff --git a/Tools/inspection/benchmark_external_inspection.py
b/Tools/inspection/benchmark_external_inspection.py
index b7aa0e5de7ed99..5c491e3cfc6211 100644
--- a/Tools/inspection/benchmark_external_inspection.py
+++ b/Tools/inspection/benchmark_external_inspection.py
@@ -7,206 +7,8 @@
import argparse
from _colorize import get_colors, can_colorize
-CODE = '''\
-import time
-import os
-import sys
-import math
-
-def slow_fibonacci(n):
- """Intentionally slow recursive fibonacci - should show up prominently in
profiler"""
- if n <= 1:
- return n
- return slow_fibonacci(n-1) + slow_fibonacci(n-2)
-
-def medium_computation():
- """Medium complexity function"""
- result = 0
- for i in range(1000):
- result += math.sqrt(i) * math.sin(i)
- return result
-
-def fast_loop():
- """Fast simple loop"""
- total = 0
- for i in range(100):
- total += i
- return total
-
-def string_operations():
- """String manipulation that should be visible in profiler"""
- text = "hello world " * 100
- words = text.split()
- return " ".join(reversed(words))
-
-def nested_calls():
- """Nested function calls to test call stack depth"""
- def level1():
- def level2():
- def level3():
- return medium_computation()
- return level3()
- return level2()
- return level1()
-
-def main_loop():
- """Main computation loop with different execution paths"""
- iteration = 0
-
- while True:
- iteration += 1
-
- # Different execution paths with different frequencies
- if iteration % 50 == 0:
- # Expensive operation - should show high per-call time
- result = slow_fibonacci(20)
-
- elif iteration % 10 == 0:
- # Medium operation
- result = nested_calls()
-
- elif iteration % 5 == 0:
- # String operations
- result = string_operations()
-
- else:
- # Fast operation - most common
- result = fast_loop()
-
- # Small delay to make sampling more interesting
- time.sleep(0.001)
-
-if __name__ == "__main__":
- main_loop()
-'''
-
-DEEP_STATIC_CODE = """\
-import time
-def factorial(n):
- if n <= 1:
- time.sleep(10000)
- return 1
- return n * factorial(n-1)
-
-factorial(900)
-"""
+from snippets import CODE_EXAMPLES, CODE
-CODE_WITH_TONS_OF_THREADS = '''\
-import time
-import threading
-import random
-import math
-
-def cpu_intensive_work():
- """Do some CPU intensive calculations"""
- result = 0
- for _ in range(10000):
- result += math.sin(random.random()) * math.cos(random.random())
- return result
-
-def io_intensive_work():
- """Simulate IO intensive work with sleeps"""
- time.sleep(0.1)
-
-def mixed_workload():
- """Mix of CPU and IO work"""
- while True:
- if random.random() < 0.3:
- cpu_intensive_work()
- else:
- io_intensive_work()
-
-def create_threads(n):
- """Create n threads doing mixed workloads"""
- threads = []
- for _ in range(n):
- t = threading.Thread(target=mixed_workload, daemon=True)
- t.start()
- threads.append(t)
- return threads
-
-# Start with 5 threads
-active_threads = create_threads(5)
-thread_count = 5
-
-# Main thread manages threads and does work
-while True:
- # Randomly add or remove threads
- if random.random() < 0.1: # 10% chance each iteration
- if random.random() < 0.5 and thread_count < 100:
- # Add 1-5 new threads
- new_count = random.randint(1, 5)
- new_threads = create_threads(new_count)
- active_threads.extend(new_threads)
- thread_count += new_count
- elif thread_count > 10:
- # Remove 1-3 threads
- remove_count = random.randint(1, 5)
- # The threads will terminate naturally since they're daemons
- active_threads = active_threads[remove_count:]
- thread_count -= remove_count
-
- cpu_intensive_work()
- time.sleep(0.05)
-'''
-
-ASYNC_CODE = '''\
-import asyncio
-import contextlib
-import math
-
-def compute_slice(seed):
- result = 0.0
- for i in range(2000):
- result += math.sin(seed + i) * math.sqrt(i + 1)
- return result
-
-async def leaf_task(seed):
- total = 0.0
- while True:
- total += compute_slice(seed)
- await asyncio.sleep(0)
-
-async def parent_task(seed):
- child = asyncio.create_task(leaf_task(seed + 1000), name=f"leaf-{seed}")
- try:
- while True:
- compute_slice(seed)
- await asyncio.sleep(0.001)
- finally:
- child.cancel()
- with contextlib.suppress(asyncio.CancelledError):
- await child
-
-async def main():
- tasks = [
- asyncio.create_task(parent_task(i), name=f"parent-{i}")
- for i in range(8)
- ]
- await asyncio.gather(*tasks)
-
-if __name__ == "__main__":
- asyncio.run(main())
-'''
-
-CODE_EXAMPLES = {
- "basic": {
- "code": CODE,
- "description": "Mixed workload with fibonacci, computations, and
string operations",
- },
- "deep_static": {
- "code": DEEP_STATIC_CODE,
- "description": "Deep recursive call stack with 900+ frames
(factorial)",
- },
- "threads": {
- "code": CODE_WITH_TONS_OF_THREADS,
- "description": "Tons of threads doing mixed CPU/IO work",
- },
- "asyncio": {
- "code": ASYNC_CODE,
- "description": "Asyncio tasks with active and awaited coroutine
chains",
- },
-}
OPERATIONS = {
"stack_trace": {
diff --git a/Tools/inspection/oracle_external_inspection.py
b/Tools/inspection/oracle_external_inspection.py
new file mode 100644
index 00000000000000..ec5060a1c9a4a5
--- /dev/null
+++ b/Tools/inspection/oracle_external_inspection.py
@@ -0,0 +1,436 @@
+"""Compare external inspection modes against a reference mode ("Oracle").
+
+This script reports the following validation metrics:
+
+- impossible: Number of stacks matching a known impossible pattern. The
+ classifiers are not exhaustive.
+
+- raw_tvd: Total variation distance between this mode's stack distribution
+ and the reference's distribution. Stacks classified as impossible are
+ excluded.
+
+- tvd_excess: raw_tvd minus tvd_floor.
+
+- tvd_floor: IID heuristic for the TVD expected from finite sampling of both
+ distributions. This is not a lower bound.
+"""
+
+import argparse
+import contextlib
+import math
+import os
+import random
+import subprocess
+import statistics
+import sys
+import tempfile
+import time
+from collections import Counter
+
+import _remote_debugging
+
+from snippets import CASES, _get_lineno
+
+
+TRANSIENT_ERRORS = (OSError, RuntimeError, UnicodeDecodeError)
+
+MODES = {
+ "live-cache": (False, True),
+ "live-nocache": (False, False),
+ "blocking-cache": (True, True),
+ "blocking-nocache": (True, False),
+}
+
+
+def collapse_cache(mode):
+ blocking, _ = MODES[mode]
+ return "blocking-nocache" if blocking else "live-nocache"
+
+
+def tvd(left, right):
+ lt, rt = sum(left.values()), sum(right.values())
+ if not lt or not rt:
+ return None
+ return 0.5 * sum(
+ abs(left[k] / lt - right[k] / rt) for k in set(left) | set(right)
+ )
+
+
+def tvd_floor(reference_obs, n_live):
+ n_ref = sum(reference_obs.values())
+ if not n_ref or not n_live:
+ return None
+ spread = sum(
+ math.sqrt(p * (1 - p))
+ for p in (c / n_ref for c in reference_obs.values())
+ )
+ return (
+ 0.5
+ * math.sqrt(2 / math.pi)
+ * math.sqrt(1 / n_live + 1 / n_ref)
+ * spread
+ )
+
+
+def print_run_info(args, cases):
+ print(sys.version.replace("\n", " "))
+ print(
+ f"cases={','.join(cases)} runs={args.runs} "
+ f"duration={args.duration} "
+ f"rate_khz={args.rate_khz} warmup={args.warmup} "
+ f"poisson_sampling={args.poisson_sampling}"
+ )
+
+
+def terminate_process(proc):
+ if proc.poll() is not None:
+ return
+ proc.terminate()
+ try:
+ proc.wait(timeout=5)
+ except subprocess.TimeoutExpired:
+ proc.kill()
+ proc.wait()
+
+
[email protected]
+def target_process(code, warmup):
+ with tempfile.NamedTemporaryFile("w", suffix=".py", delete=False) as tmp:
+ tmp.write(code)
+ tmp.flush()
+ tmp_name = tmp.name
+ proc = None
+ try:
+ proc = subprocess.Popen(
+ [sys.executable, tmp_name],
+ stdout=subprocess.DEVNULL,
+ )
+ time.sleep(warmup)
+ if proc.poll() is not None:
+ raise RuntimeError(
+ f"target exited unexpectedly with code {proc.returncode}"
+ )
+ yield proc
+ finally:
+ with contextlib.suppress(Exception):
+ if proc is not None:
+ terminate_process(proc)
+ with contextlib.suppress(OSError):
+ os.unlink(tmp_name)
+
+
+def get_trace(unwinder, blocking, op="get_stack_trace"):
+ call = getattr(unwinder, op)
+ if not blocking:
+ return call()
+ unwinder.pause_threads()
+ try:
+ return call()
+ finally:
+ unwinder.resume_threads()
+
+
+def iter_units(raw, op):
+ if op == "get_stack_trace":
+ for interp in raw:
+ for thread in interp.threads:
+ yield (
+ thread.thread_id,
+ None,
+ thread.status,
+ thread.frame_info,
+ )
+ else:
+ for awaited_info in raw:
+ for task in awaited_info.awaited_by:
+ frames = [
+ frame
+ for coro in task.coroutine_stack
+ for frame in coro.call_stack
+ ]
+ if frames:
+ yield (task.task_id, task.task_name, None, frames)
+
+
+def run_mode(case, mode_name, args):
+ code, classify, *rest = case
+ op = rest[0] if rest else "get_stack_trace"
+ classify_units = op != "get_stack_trace"
+ blocking, cache_frames = MODES[mode_name]
+ result = {
+ "attempts": 0,
+ "samples": 0,
+ "stacks": 0,
+ "errors": 0,
+ "observations": Counter(),
+ "impossible": 0,
+ "work_time": 0.0,
+ }
+ with target_process(code, args.warmup) as proc:
+ unwinder = _remote_debugging.RemoteUnwinder(
+ proc.pid,
+ all_threads=True,
+ cache_frames=cache_frames,
+ )
+ rate_hz = args.rate_khz * 1000
+ period = 1.0 / rate_hz if rate_hz else 0
+ next_sample = time.perf_counter()
+ deadline = next_sample + args.duration
+
+ while time.perf_counter() < deadline:
+ if period:
+ if args.poisson_sampling:
+ next_sample += random.expovariate(rate_hz)
+ if next_sample >= deadline:
+ break
+ now = time.perf_counter()
+ if next_sample > now:
+ time.sleep(next_sample - now)
+ if not args.poisson_sampling:
+ next_sample += period
+
+ if time.perf_counter() >= deadline:
+ break
+
+ result["attempts"] += 1
+ work_start = time.perf_counter()
+ try:
+ trace = get_trace(unwinder, blocking, op)
+ except TRANSIENT_ERRORS:
+ trace = None
+ result["errors"] += 1
+ result["work_time"] += time.perf_counter() - work_start
+
+ if not trace:
+ continue
+
+ result["samples"] += 1
+ for unit in iter_units(trace, op):
+ frames = unit[3]
+ impossible = classify is not None and (
+ classify(unit) if classify_units else classify(frames)
+ )
+ result["stacks"] += 1
+ if impossible:
+ result["impossible"] += 1
+ else:
+ result["observations"][
+ ";".join(
+ f"{frame.funcname}:{_get_lineno(frame)}"
+ for frame in frames
+ )
+ ] += 1
+ return result
+
+
+def result_metrics(result, reference_obs, is_reference, op):
+ skip_tvd = is_reference or op != "get_stack_trace"
+ n_live = sum(result["observations"].values())
+ raw_tvd = None if skip_tvd else tvd(result["observations"], reference_obs)
+ # IID heuristic, not a calibrated noise bound.
+ floor = None if skip_tvd else tvd_floor(reference_obs, n_live)
+ return {
+ "samples": result["samples"],
+ "stacks": result["stacks"],
+ "empty": result["attempts"] - result["samples"] - result["errors"],
+ "impossible": result["impossible"],
+ "error_percent": (
+ 100.0 * result["errors"] / result["attempts"]
+ if result["attempts"]
+ else 0.0
+ ),
+ "impossible_percent": (
+ 100.0 * result["impossible"] / result["stacks"]
+ if result["stacks"]
+ else 0.0
+ ),
+ "avg_us": (
+ 1e6 * result["work_time"] / result["attempts"]
+ if result["attempts"]
+ else 0.0
+ ),
+ "raw_tvd": raw_tvd,
+ "tvd_floor": floor,
+ "tvd_excess": (
+ None if (raw_tvd is None or floor is None) else raw_tvd - floor
+ ),
+ }
+
+
+def fmt_stat(values, precision):
+ vals = [value for value in values if value is not None]
+ if not vals:
+ return "n/a"
+ mean = statistics.mean(vals)
+ if len(vals) > 1:
+ return f"{mean:.{precision}f}±{statistics.stdev(vals):.{precision}f}"
+ return f"{mean:.{precision}f}"
+
+
+def fmt_floor(values):
+ vals = [value for value in values if value is not None]
+ return "n/a" if not vals else f"{statistics.median(vals):.3f}"
+
+
+def print_results(
+ case_name, run_results, modes, reference_mode, op, has_classify
+):
+ is_sync = op == "get_stack_trace"
+ ref_keys = len(run_results[0][reference_mode]["observations"])
+ rows = {}
+ for mode in modes:
+ metrics = [
+ result_metrics(
+ results[mode],
+ results[reference_mode]["observations"],
+ mode == reference_mode,
+ op,
+ )
+ for results in run_results
+ ]
+ rows[mode] = {
+ "mode": mode,
+ "samples": sum(item["samples"] for item in metrics),
+ "stacks": sum(item["stacks"] for item in metrics),
+ "empty": sum(item["empty"] for item in metrics),
+ "impossible": sum(item["impossible"] for item in metrics),
+ "errors": fmt_stat([item["error_percent"] for item in metrics], 2),
+ "us": fmt_stat([item["avg_us"] for item in metrics], 2),
+ "impossible_pct": fmt_stat(
+ [item["impossible_percent"] for item in metrics], 2
+ ),
+ "raw_tvd": "ref"
+ if mode == reference_mode
+ else fmt_stat([item["raw_tvd"] for item in metrics], 3),
+ "tvd_excess": "ref"
+ if mode == reference_mode
+ else fmt_stat([item["tvd_excess"] for item in metrics], 3),
+ "tvd_floor": "-"
+ if mode == reference_mode
+ else fmt_floor([item["tvd_floor"] for item in metrics]),
+ }
+
+ show_stacks = any(row["stacks"] != row["samples"] for row in rows.values())
+ ordered = [reference_mode] + [m for m in modes if m != reference_mode]
+
+ columns = [("mode", "<18", "mode"), ("samples", ">9", "samples")]
+ if show_stacks:
+ columns.append(("stacks", ">9", "stacks"))
+ if not is_sync:
+ columns.append(("empty", ">9", "empty"))
+ columns += [("µs", ">12", "us"), ("errors%", ">12", "errors")]
+ if has_classify:
+ columns += [
+ ("impossible", ">10", "impossible"),
+ ("impossible%", ">12", "impossible_pct"),
+ ]
+ if is_sync:
+ columns += [
+ ("raw_tvd", ">12", "raw_tvd"),
+ ("tvd_excess", ">12", "tvd_excess"),
+ ("tvd_floor", ">10", "tvd_floor"),
+ ]
+
+ print(f"\n{case_name} ({op}) ref_keys={ref_keys}")
+ print(" ".join(f"{label:{spec}}" for label, spec, _ in columns))
+ for mode in ordered:
+ row = rows[mode]
+ print(" ".join(f"{row[key]:{spec}}" for _, spec, key in columns))
+
+
+def parse_args():
+ parser = argparse.ArgumentParser(
+ formatter_class=argparse.ArgumentDefaultsHelpFormatter
+ )
+ parser.add_argument(
+ "--snippet",
+ action="append",
+ help="snippet name or Python file; may be passed more than once",
+ )
+ parser.add_argument(
+ "--mode",
+ choices=sorted(MODES),
+ action="append",
+ help="mode to run; may be passed more than once; omit to run all
modes",
+ )
+ parser.add_argument(
+ "--reference-mode",
+ choices=sorted(MODES),
+ default="blocking-nocache",
+ help="mode used as the distribution reference",
+ )
+ parser.add_argument(
+ "--duration",
+ type=float,
+ default=3.0,
+ help="seconds to sample each mode",
+ )
+ parser.add_argument(
+ "--runs",
+ type=int,
+ default=1,
+ help="number of independent runs per case",
+ )
+ parser.add_argument(
+ "--rate-khz",
+ type=float,
+ default=100.0,
+ help="target sampling rate in kHz; 0 samples as fast as possible",
+ )
+ parser.add_argument(
+ "--warmup",
+ type=float,
+ default=0.7,
+ help="seconds to let the target run before sampling",
+ )
+ parser.add_argument(
+ "--poisson-sampling",
+ action="store_true",
+ help=(
+ "sample with exponential inter-arrival times instead of a fixed "
+ "period"
+ ),
+ )
+ args = parser.parse_args()
+ if args.runs < 1:
+ parser.error("--runs must be greater than zero")
+ return args
+
+
+def main():
+ args = parse_args()
+ cases = list(args.snippet) if args.snippet else sorted(CASES)
+ modes = list(args.mode) if args.mode else sorted(MODES)
+ if args.reference_mode not in modes:
+ modes.append(args.reference_mode)
+
+ print_run_info(args, cases)
+
+ for name in cases:
+ if name in CASES:
+ case = CASES[name]
+ else:
+ with open(name, encoding="utf-8") as file:
+ case = (file.read(), None)
+ op = case[2] if len(case) > 2 else "get_stack_trace"
+ case_ref = args.reference_mode
+ case_modes = list(modes)
+ if op != "get_stack_trace":
+ case_ref = collapse_cache(case_ref)
+ case_modes = list(
+ dict.fromkeys(collapse_cache(m) for m in case_modes)
+ )
+ if case_ref not in case_modes:
+ case_modes.append(case_ref)
+ run_results = [
+ {mode: run_mode(case, mode, args) for mode in case_modes}
+ for _ in range(args.runs)
+ ]
+ print_results(
+ name, run_results, case_modes, case_ref, op, case[1] is not None
+ )
+ return 0
+
+
+if __name__ == "__main__":
+ sys.exit(main())
diff --git a/Tools/inspection/snippets.py b/Tools/inspection/snippets.py
new file mode 100644
index 00000000000000..47173c4fd12543
--- /dev/null
+++ b/Tools/inspection/snippets.py
@@ -0,0 +1,601 @@
+"""Scripts and classifiers for external inspection validation.
+
+Classifiers detect only recognized impossible patterns. They do not validate
+entire stacks. False does not imply that the stack as a whole is valid.
+"""
+import os
+
+
+def _get_lineno(frame, default=None):
+ if frame is None:
+ return default
+ loc = getattr(frame, "location", None)
+ return getattr(loc, "lineno", default) if loc is not None else default
+
+
+CODE = '''\
+import time
+import os
+import sys
+import math
+
+def slow_fibonacci(n):
+ """Intentionally slow recursive fibonacci - should show up prominently in
profiler"""
+ if n <= 1:
+ return n
+ return slow_fibonacci(n-1) + slow_fibonacci(n-2)
+
+def medium_computation():
+ """Medium complexity function"""
+ result = 0
+ for i in range(1000):
+ result += math.sqrt(i) * math.sin(i)
+ return result
+
+def fast_loop():
+ """Fast simple loop"""
+ total = 0
+ for i in range(100):
+ total += i
+ return total
+
+def string_operations():
+ """String manipulation that should be visible in profiler"""
+ text = "hello world " * 100
+ words = text.split()
+ return " ".join(reversed(words))
+
+def nested_calls():
+ """Nested function calls to test call stack depth"""
+ def level1():
+ def level2():
+ def level3():
+ return medium_computation()
+ return level3()
+ return level2()
+ return level1()
+
+def main_loop():
+ """Main computation loop with different execution paths"""
+ iteration = 0
+
+ while True:
+ iteration += 1
+
+ # Different execution paths with different frequencies
+ if iteration % 50 == 0:
+ # Expensive operation - should show high per-call time
+ result = slow_fibonacci(20)
+
+ elif iteration % 10 == 0:
+ # Medium operation
+ result = nested_calls()
+
+ elif iteration % 5 == 0:
+ # String operations
+ result = string_operations()
+
+ else:
+ # Fast operation - most common
+ result = fast_loop()
+
+ # Small delay to make sampling more interesting
+ time.sleep(0.001)
+
+if __name__ == "__main__":
+ main_loop()
+'''
+
+DEEP_STATIC_CODE = """\
+import time
+def factorial(n):
+ if n <= 1:
+ time.sleep(10000)
+ return 1
+ return n * factorial(n-1)
+
+factorial(900)
+"""
+
+CODE_WITH_TONS_OF_THREADS = '''\
+import time
+import threading
+import random
+import math
+
+def cpu_intensive_work():
+ """Do some CPU intensive calculations"""
+ result = 0
+ for _ in range(10000):
+ result += math.sin(random.random()) * math.cos(random.random())
+ return result
+
+def io_intensive_work():
+ """Simulate IO intensive work with sleeps"""
+ time.sleep(0.1)
+
+def mixed_workload():
+ """Mix of CPU and IO work"""
+ while True:
+ if random.random() < 0.3:
+ cpu_intensive_work()
+ else:
+ io_intensive_work()
+
+def create_threads(n):
+ """Create n threads doing mixed workloads"""
+ threads = []
+ for _ in range(n):
+ t = threading.Thread(target=mixed_workload, daemon=True)
+ t.start()
+ threads.append(t)
+ return threads
+
+# Start with 5 threads
+active_threads = create_threads(5)
+
+# Main thread manages threads and does work
+while True:
+ # Randomly add threads up to the limit
+ if random.random() < 0.1: # 10% chance each iteration
+ if random.random() < 0.5 and len(active_threads) < 100:
+ new_count = min(
+ random.randint(1, 5),
+ 100 - len(active_threads),
+ )
+ new_threads = create_threads(new_count)
+ active_threads.extend(new_threads)
+
+ cpu_intensive_work()
+ time.sleep(0.05)
+'''
+
+ASYNC_CODE = '''\
+import asyncio
+import contextlib
+import math
+
+def compute_slice(seed):
+ result = 0.0
+ for i in range(2000):
+ result += math.sin(seed + i) * math.sqrt(i + 1)
+ return result
+
+async def leaf_task(seed):
+ total = 0.0
+ while True:
+ total += compute_slice(seed)
+ await asyncio.sleep(0)
+
+async def parent_task(seed):
+ child = asyncio.create_task(leaf_task(seed + 1000), name=f"leaf-{seed}")
+ try:
+ while True:
+ compute_slice(seed)
+ await asyncio.sleep(0.001)
+ finally:
+ child.cancel()
+ with contextlib.suppress(asyncio.CancelledError):
+ await child
+
+async def main():
+ tasks = [
+ asyncio.create_task(parent_task(i), name=f"parent-{i}")
+ for i in range(8)
+ ]
+ await asyncio.gather(*tasks)
+
+if __name__ == "__main__":
+ asyncio.run(main())
+'''
+
+
+FLAT_ALTERNATING_CODE = """\
+def leaf_a(): return sum(range(50))
+def leaf_b(): return sum(range(50))
+def hot_a(): return leaf_a()
+def hot_b(): return leaf_b()
+while True:
+ hot_a(); hot_b()
+"""
+
+
+def _expected_lines(code):
+ expected = {}
+ for number, line in enumerate(code.splitlines(), 1):
+ stripped = line.strip()
+ if stripped.startswith("def "):
+ expected[stripped[4:].split("(")[0].strip()] = number
+ return expected
+
+
+FLAT_ALTERNATING_LINES = _expected_lines(FLAT_ALTERNATING_CODE)
+
+
+def classify_flat(frames):
+ present = {}
+ for frame in frames:
+ if frame.funcname in FLAT_ALTERNATING_LINES:
+ present[frame.funcname] = _get_lineno(frame, -1)
+ if not present:
+ return False
+ for name, lineno in present.items():
+ if lineno != FLAT_ALTERNATING_LINES[name]:
+ return True
+ hot_a, hot_b = "hot_a" in present, "hot_b" in present
+ leaf_a, leaf_b = "leaf_a" in present, "leaf_b" in present
+ any_a = leaf_a or hot_a
+ any_b = leaf_b or hot_b
+ return (any_a and any_b) or (leaf_a and not hot_a) or (leaf_b and not
hot_b)
+
+
+NESTED_ALTERNATING_CODE = """\
+def burn_a():
+ total = 0
+ for i in range(20000):
+ total += i
+ return total
+
+def burn_b():
+ total = 0
+ for i in range(20000):
+ total += i
+ return total
+
+def a_leaf():
+ return burn_a()
+
+def b_leaf():
+ return burn_b()
+
+def a_parent():
+ return a_leaf()
+
+def b_parent():
+ return b_leaf()
+
+while True:
+ a_parent()
+ b_parent()
+"""
+
+
+NESTED_ALTERNATING_BRANCHES = {
+ "a": ["a_parent", "a_leaf", "burn_a"],
+ "b": ["b_parent", "b_leaf", "burn_b"],
+}
+
+
+def classify_nested(frames):
+ frame_names = [frame.funcname for frame in frames]
+ names = set(frame_names)
+ present = [
+ family
+ for family, chain in NESTED_ALTERNATING_BRANCHES.items()
+ if names.intersection(chain)
+ ]
+ if len(present) > 1:
+ return True
+ if not present:
+ return False
+ chain = NESTED_ALTERNATING_BRANCHES[present[0]]
+ active = [name for name in chain if name in names]
+ depth = chain.index(active[-1])
+ if len(active) != depth + 1:
+ return True
+ indices = [frame_names.index(name) for name in reversed(active)]
+ return indices != sorted(indices)
+
+
+SHARED_LEAF_CODE = """\
+def shared_leaf(long_run):
+ total = 0
+ if long_run:
+ for i in range(50000):
+ total += i
+ else:
+ for i in range(200):
+ total += i
+ return total
+
+def a_wrapper():
+ return shared_leaf(False)
+
+def b_wrapper():
+ return shared_leaf(True)
+
+while True:
+ a_wrapper()
+ b_wrapper()
+"""
+
+
+def _branch_lines(code, marker):
+ for number, line in enumerate(code.splitlines(), 1):
+ if marker in line:
+ return {number, number + 1}
+ return set()
+
+
+SHARED_LEAF_LONG_LINES = _branch_lines(SHARED_LEAF_CODE, "range(50000)")
+SHARED_LEAF_SHORT_LINES = _branch_lines(SHARED_LEAF_CODE, "range(200)")
+
+
+def classify_shared(frames):
+ frame_names = [frame.funcname for frame in frames]
+ names = set(frame_names)
+ if "a_wrapper" in names and "b_wrapper" in names:
+ return True
+ if "shared_leaf" not in names:
+ return False
+ index = frame_names.index("shared_leaf")
+ parent = frame_names[index + 1] if index + 1 < len(frame_names) else None
+ if parent not in ("a_wrapper", "b_wrapper"):
+ return True
+ lineno = _get_lineno(frames[index], -1)
+ if lineno in SHARED_LEAF_LONG_LINES:
+ return parent != "b_wrapper"
+ if lineno in SHARED_LEAF_SHORT_LINES:
+ return parent != "a_wrapper"
+ return False
+
+
+GEN_ALTERNATING_CODE = """\
+def agen(n):
+ total = 0
+ for i in range(n):
+ total += i
+ yield i
+
+def bgen(n):
+ total = 0
+ for i in range(n):
+ total += i
+ yield i
+
+def drv_a():
+ for _ in agen(60):
+ pass
+
+def drv_b():
+ for _ in bgen(60):
+ pass
+
+while True:
+ drv_a()
+ drv_b()
+"""
+
+
+def classify_gen(frames):
+ names = {frame.funcname for frame in frames}
+ return ("agen" in names and "drv_b" in names) or (
+ "bgen" in names and "drv_a" in names
+ )
+
+
+DEEP_RECURSION_CODE = """\
+def leaf():
+ total = 0
+ for i in range(40):
+ total += i
+
+def a(n):
+ return a(n - 1) if n else leaf()
+
+def b(n):
+ return b(n - 1) if n else leaf()
+
+while True:
+ a(300)
+ b(300)
+"""
+
+
+def classify_recursion(frames):
+ names = {frame.funcname for frame in frames}
+ return "a" in names and "b" in names
+
+
+ASYNC_RUNNING_TASK_CODE = """\
+import asyncio
+
+
+def leaf_hot(n):
+ return sum(range(n))
+
+
+def leaf_rare(n):
+ return sum(range(n))
+
+
+async def run_hot():
+ while True:
+ leaf_hot(50000)
+ await asyncio.sleep(0)
+
+
+async def run_rare(k):
+ while True:
+ leaf_rare(500)
+ await asyncio.sleep(0)
+
+
+async def main():
+ tasks = [asyncio.create_task(run_hot(), name="hot")]
+ for k in range(8):
+ tasks.append(asyncio.create_task(run_rare(k), name=f"rare{k}"))
+ await asyncio.gather(*tasks)
+
+
+asyncio.run(main())
+"""
+
+
+def _name_tag(label):
+ label = (label or "").lower()
+ return "hot" if "hot" in label else "rare" if "rare" in label else None
+
+
+def _frame_tag(frames):
+ fns = {frame.funcname for frame in frames}
+ hot = bool(fns & {"run_hot", "leaf_hot"})
+ rare = bool(fns & {"run_rare", "leaf_rare"})
+ return (
+ "mixed"
+ if (hot and rare)
+ else "hot"
+ if hot
+ else "rare"
+ if rare
+ else None
+ )
+
+
+def classify_async_running_task(unit):
+ name = _name_tag(unit[1])
+ frame = _frame_tag(unit[3])
+ return name is not None and frame is not None and frame != name
+
+
+CODE_OBJECT_REUSE_CODE = """\
+SRC_A = "def func_a(n):\\n total=0\\n for i in range(n): total+=i*i\\n return
total\\n"
+SRC_B = "def func_b(n):\\n total=0\\n for i in range(n): total+=i*i\\n return
total\\n"
+WORK = 60000
+
+
+def build_a():
+ ns = {}
+ code = compile(SRC_A, "A_file.py", "exec")
+ exec(code, ns)
+ return ns["func_a"], code
+
+
+def build_b():
+ ns = {}
+ code = compile(SRC_B, "B_file.py", "exec")
+ exec(code, ns)
+ return ns["func_b"], code
+
+
+while True:
+ fa, ca = build_a()
+ fa(WORK) # call_a
+ del fa, ca
+ fb, cb = build_b()
+ fb(WORK) # call_b
+ del fb, cb
+"""
+
+
+def _marker_line(code, marker):
+ for number, line in enumerate(code.splitlines(), 1):
+ if marker in line:
+ return number
+ return None
+
+
+CALL_A_LINE = _marker_line(CODE_OBJECT_REUSE_CODE, "# call_a")
+CALL_B_LINE = _marker_line(CODE_OBJECT_REUSE_CODE, "# call_b")
+
+
+def classify_code_object_reuse(frames):
+ real = [f for f in frames if f.funcname != "<GC>"]
+ leaf = next((f for f in real if f.funcname in ("func_a", "func_b")), None)
+ if leaf is None:
+ return False
+ base = os.path.basename(leaf.filename)
+ fn = leaf.funcname
+ if (fn == "func_a" and base == "B_file.py") or (
+ fn == "func_b" and base == "A_file.py"
+ ):
+ return True
+ index = real.index(leaf)
+ caller = real[index + 1] if index + 1 < len(real) else None
+ line = _get_lineno(caller)
+ if line == CALL_A_LINE and fn == "func_b":
+ return True
+ if line == CALL_B_LINE and fn == "func_a":
+ return True
+ return False
+
+
+OVERSIZED_CHUNK_CODE = """\
+NLOCALS = 1800
+
+def make(name, tag, hotbody):
+ params = ", ".join(f"x{i}=0" for i in range(NLOCALS))
+ src = (
+ f"def hot_{tag}():\\n{hotbody}\\n"
+ f"def {name}({params}):\\n return hot_{tag}()\\n"
+ )
+ exec(compile(src, f"{tag}.py", "exec"), globals())
+
+make("big_a", "a", " s=0\\n for i in range(2000):\\n s+=i*3\\n
return s")
+make("big_b", "b", " s=1\\n for i in range(2000):\\n s^=(i<<1)\\n
return s")
+
+while True:
+ big_a()
+ big_b()
+"""
+
+
+OVERSIZED_A_FUNCS = {"big_a", "hot_a"}
+OVERSIZED_B_FUNCS = {"big_b", "hot_b"}
+
+
+def classify_oversized_chunk(frames):
+ saw_a = saw_b = False
+ for frame in frames:
+ base = os.path.basename(frame.filename)
+ fn = frame.funcname
+ if base == "a.py":
+ if fn in OVERSIZED_A_FUNCS:
+ saw_a = True
+ elif fn in OVERSIZED_B_FUNCS:
+ return True
+ elif base == "b.py":
+ if fn in OVERSIZED_B_FUNCS:
+ saw_b = True
+ elif fn in OVERSIZED_A_FUNCS:
+ return True
+ return saw_a and saw_b
+
+
+CODE_EXAMPLES = {
+ "basic": {
+ "code": CODE,
+ "description": "Mixed workload with fibonacci, computations, and
string operations",
+ },
+ "deep_static": {
+ "code": DEEP_STATIC_CODE,
+ "description": "Deep recursive call stack with 900+ frames
(factorial)",
+ },
+ "threads": {
+ "code": CODE_WITH_TONS_OF_THREADS,
+ "description": "Tons of threads doing mixed CPU/IO work",
+ },
+ "asyncio": {
+ "code": ASYNC_CODE,
+ "description": "Asyncio tasks with active and awaited coroutine
chains",
+ },
+}
+
+CASES = {
+ "basic": (CODE, None),
+ "deep_static": (DEEP_STATIC_CODE, None),
+ "threads": (CODE_WITH_TONS_OF_THREADS, None),
+ "asyncio": (ASYNC_CODE, None, "get_async_stack_trace"),
+ "flat_alternating": (FLAT_ALTERNATING_CODE, classify_flat),
+ "nested_alternating": (NESTED_ALTERNATING_CODE, classify_nested),
+ "shared_leaf": (SHARED_LEAF_CODE, classify_shared),
+ "gen_alternating": (GEN_ALTERNATING_CODE, classify_gen),
+ "deep_recursion": (DEEP_RECURSION_CODE, classify_recursion),
+ "async_running_task": (
+ ASYNC_RUNNING_TASK_CODE,
+ classify_async_running_task,
+ "get_async_stack_trace",
+ ),
+ "code_object_reuse": (CODE_OBJECT_REUSE_CODE, classify_code_object_reuse),
+ "oversized_chunk": (OVERSIZED_CHUNK_CODE, classify_oversized_chunk),
+}
_______________________________________________
Python-checkins mailing list -- [email protected]
To unsubscribe send an email to [email protected]
https://mail.python.org/mailman3//lists/python-checkins.python.org
Member address: [email protected]
