cpython/Lib/test/test_perf_profiler.py

355 lines
11 KiB
Python
Raw Normal View History

import unittest
import string
import subprocess
import sys
import sysconfig
import os
import pathlib
from test import support
from test.support.script_helper import (
make_script,
assert_python_failure,
assert_python_ok,
)
from test.support.os_helper import temp_dir
if not support.has_subprocess_support:
raise unittest.SkipTest("test module requires subprocess")
def supports_trampoline_profiling():
perf_trampoline = sysconfig.get_config_var("PY_HAVE_PERF_TRAMPOLINE")
if not perf_trampoline:
return False
return int(perf_trampoline) == 1
if not supports_trampoline_profiling():
raise unittest.SkipTest("perf trampoline profiling not supported")
class TestPerfTrampoline(unittest.TestCase):
def setUp(self):
super().setUp()
self.perf_files = set(pathlib.Path("/tmp/").glob("perf-*.map"))
def tearDown(self) -> None:
super().tearDown()
files_to_delete = (
set(pathlib.Path("/tmp/").glob("perf-*.map")) - self.perf_files
)
for file in files_to_delete:
file.unlink()
def test_trampoline_works(self):
code = """if 1:
def foo():
pass
def bar():
foo()
def baz():
bar()
baz()
"""
with temp_dir() as script_dir:
script = make_script(script_dir, "perftest", code)
with subprocess.Popen(
[sys.executable, "-Xperf", script],
text=True,
stderr=subprocess.PIPE,
stdout=subprocess.PIPE,
) as process:
stdout, stderr = process.communicate()
self.assertEqual(stderr, "")
self.assertEqual(stdout, "")
perf_file = pathlib.Path(f"/tmp/perf-{process.pid}.map")
self.assertTrue(perf_file.exists())
perf_file_contents = perf_file.read_text()
perf_lines = perf_file_contents.splitlines();
expected_symbols = [f"py::foo:{script}", f"py::bar:{script}", f"py::baz:{script}"]
for expected_symbol in expected_symbols:
perf_line = next((line for line in perf_lines if expected_symbol in line), None)
self.assertIsNotNone(perf_line, f"Could not find {expected_symbol} in perf file")
perf_addr = perf_line.split(" ")[0]
self.assertFalse(perf_addr.startswith("0x"), "Address should not be prefixed with 0x")
self.assertTrue(set(perf_addr).issubset(string.hexdigits), "Address should contain only hex characters")
def test_trampoline_works_with_forks(self):
code = """if 1:
import os, sys
def foo_fork():
pass
def bar_fork():
foo_fork()
def baz_fork():
bar_fork()
def foo():
pid = os.fork()
if pid == 0:
print(os.getpid())
baz_fork()
else:
_, status = os.waitpid(-1, 0)
sys.exit(status)
def bar():
foo()
def baz():
bar()
baz()
"""
with temp_dir() as script_dir:
script = make_script(script_dir, "perftest", code)
with subprocess.Popen(
[sys.executable, "-Xperf", script],
text=True,
stderr=subprocess.PIPE,
stdout=subprocess.PIPE,
) as process:
stdout, stderr = process.communicate()
self.assertEqual(process.returncode, 0)
self.assertEqual(stderr, "")
child_pid = int(stdout.strip())
perf_file = pathlib.Path(f"/tmp/perf-{process.pid}.map")
perf_child_file = pathlib.Path(f"/tmp/perf-{child_pid}.map")
self.assertTrue(perf_file.exists())
self.assertTrue(perf_child_file.exists())
perf_file_contents = perf_file.read_text()
self.assertIn(f"py::foo:{script}", perf_file_contents)
self.assertIn(f"py::bar:{script}", perf_file_contents)
self.assertIn(f"py::baz:{script}", perf_file_contents)
child_perf_file_contents = perf_child_file.read_text()
self.assertIn(f"py::foo_fork:{script}", child_perf_file_contents)
self.assertIn(f"py::bar_fork:{script}", child_perf_file_contents)
self.assertIn(f"py::baz_fork:{script}", child_perf_file_contents)
def test_sys_api(self):
code = """if 1:
import sys
def foo():
pass
def spam():
pass
def bar():
sys.deactivate_stack_trampoline()
foo()
sys.activate_stack_trampoline("perf")
spam()
def baz():
bar()
sys.activate_stack_trampoline("perf")
baz()
"""
with temp_dir() as script_dir:
script = make_script(script_dir, "perftest", code)
with subprocess.Popen(
[sys.executable, script],
text=True,
stderr=subprocess.PIPE,
stdout=subprocess.PIPE,
) as process:
stdout, stderr = process.communicate()
self.assertEqual(stderr, "")
self.assertEqual(stdout, "")
perf_file = pathlib.Path(f"/tmp/perf-{process.pid}.map")
self.assertTrue(perf_file.exists())
perf_file_contents = perf_file.read_text()
self.assertNotIn(f"py::foo:{script}", perf_file_contents)
self.assertIn(f"py::spam:{script}", perf_file_contents)
self.assertIn(f"py::bar:{script}", perf_file_contents)
self.assertIn(f"py::baz:{script}", perf_file_contents)
def test_sys_api_with_existing_trampoline(self):
code = """if 1:
import sys
sys.activate_stack_trampoline("perf")
sys.activate_stack_trampoline("perf")
"""
assert_python_ok("-c", code)
def test_sys_api_with_invalid_trampoline(self):
code = """if 1:
import sys
sys.activate_stack_trampoline("invalid")
"""
rc, out, err = assert_python_failure("-c", code)
self.assertIn("invalid backend: invalid", err.decode())
def test_sys_api_get_status(self):
code = """if 1:
import sys
sys.activate_stack_trampoline("perf")
assert sys.is_stack_trampoline_active() is True
sys.deactivate_stack_trampoline()
assert sys.is_stack_trampoline_active() is False
"""
assert_python_ok("-c", code)
def is_unwinding_reliable():
cflags = sysconfig.get_config_var("PY_CORE_CFLAGS")
if not cflags:
return False
return "no-omit-frame-pointer" in cflags
def perf_command_works():
try:
cmd = ["perf", "--help"]
stdout = subprocess.check_output(cmd, text=True)
except (subprocess.SubprocessError, OSError):
return False
# perf version does not return a version number on Fedora. Use presence
# of "perf.data" in help as indicator that it's perf from Linux tools.
if "perf.data" not in stdout:
return False
# Check that we can run a simple perf run
with temp_dir() as script_dir:
try:
output_file = script_dir + "/perf_output.perf"
cmd = (
"perf",
"record",
"-g",
"--call-graph=fp",
"-o",
output_file,
"--",
sys.executable,
"-c",
'print("hello")',
)
stdout = subprocess.check_output(
cmd, cwd=script_dir, text=True, stderr=subprocess.STDOUT
)
except (subprocess.SubprocessError, OSError):
return False
if "hello" not in stdout:
return False
return True
def run_perf(cwd, *args, **env_vars):
if env_vars:
env = os.environ.copy()
env.update(env_vars)
else:
env = None
output_file = cwd + "/perf_output.perf"
base_cmd = ("perf", "record", "-g", "--call-graph=fp", "-o", output_file, "--")
proc = subprocess.run(
base_cmd + args,
stdout=subprocess.PIPE,
stderr=subprocess.PIPE,
env=env,
)
if proc.returncode:
print(proc.stderr)
raise ValueError(f"Perf failed with return code {proc.returncode}")
base_cmd = ("perf", "script")
proc = subprocess.run(
("perf", "script", "-i", output_file),
stdout=subprocess.PIPE,
stderr=subprocess.PIPE,
env=env,
check=True,
)
return proc.stdout.decode("utf-8", "replace"), proc.stderr.decode(
"utf-8", "replace"
)
@unittest.skipUnless(perf_command_works(), "perf command doesn't work")
@unittest.skipUnless(is_unwinding_reliable(), "Unwinding is unreliable")
@support.skip_if_sanitizer(address=True, memory=True, ub=True)
class TestPerfProfiler(unittest.TestCase):
def setUp(self):
super().setUp()
self.perf_files = set(pathlib.Path("/tmp/").glob("perf-*.map"))
def tearDown(self) -> None:
super().tearDown()
files_to_delete = (
set(pathlib.Path("/tmp/").glob("perf-*.map")) - self.perf_files
)
for file in files_to_delete:
file.unlink()
def test_python_calls_appear_in_the_stack_if_perf_activated(self):
with temp_dir() as script_dir:
code = """if 1:
def foo(n):
x = 0
for i in range(n):
x += i
def bar(n):
foo(n)
def baz(n):
bar(n)
baz(10000000)
"""
script = make_script(script_dir, "perftest", code)
stdout, stderr = run_perf(script_dir, sys.executable, "-Xperf", script)
self.assertEqual(stderr, "")
self.assertIn(f"py::foo:{script}", stdout)
self.assertIn(f"py::bar:{script}", stdout)
self.assertIn(f"py::baz:{script}", stdout)
def test_python_calls_do_not_appear_in_the_stack_if_perf_activated(self):
with temp_dir() as script_dir:
code = """if 1:
def foo(n):
x = 0
for i in range(n):
x += i
def bar(n):
foo(n)
def baz(n):
bar(n)
baz(10000000)
"""
script = make_script(script_dir, "perftest", code)
stdout, stderr = run_perf(script_dir, sys.executable, script)
self.assertEqual(stderr, "")
self.assertNotIn(f"py::foo:{script}", stdout)
self.assertNotIn(f"py::bar:{script}", stdout)
self.assertNotIn(f"py::baz:{script}", stdout)
if __name__ == "__main__":
unittest.main()