small reorganization to trampoline unwind generation

Author: pablogsal

Lib/test/test_perf_profiler.py

@@ -1,4 +1,3 @@
-import glob
 import struct
 import unittest
 import string
@@ -15,7 +14,6 @@
 )
 from test.support import import_helper
 from test.support.os_helper import temp_dir
-from test import test_tools
 
 
 if not support.has_subprocess_support:
@@ -754,78 +752,86 @@ def _run_and_get_jitdump(self, code):
         # The child prints its pid so we open exactly its own jitdump file
         # rather than whatever another test worker left in /tmp.
         code = "import os, sys\nsys.stdout.write(str(os.getpid()))\n" + code
-        rc, out, err = assert_python_ok("-Xperf_jit", "-c", code, PYTHON_JIT="0")
-        path = f"/tmp/jit-{int(out)}.dump"
-        if not os.path.exists(path):
+        _, out, _ = assert_python_ok("-Xperf_jit", "-c", code, PYTHON_JIT="0")
+        path = pathlib.Path(f"/tmp/jit-{int(out)}.dump")
+        try:
+            data = path.read_bytes()
+        except FileNotFoundError:
             # perf_map_jit_init() gives up silently when it cannot create
             # the file (for example an unwritable /tmp).
             self.skipTest("jitdump file was not created")
-        self.addCleanup(os.unlink, path)
-        with open(path, "rb") as f:
-            data = f.read()
+        self.addCleanup(path.unlink)
         if not data:
             # The file is created before the executable mapping of the
             # jitdump; if that mapping fails (for example a noexec /tmp) the
             # backend gives up silently and never writes the header.
             self.skipTest("jitdump could not be initialized")
         return data
 
+    def _check_code_load(self, data, pos, size):
+        """Validate a code load record and return its name and code size."""
+        code_size = struct.unpack_from(
+            f"{JITDUMP_ENDIAN}Q", data, pos + CODE_LOAD_CODE_SIZE_OFFSET)[0]
+        name_start = pos + CODE_LOAD_NAME_OFFSET
+        name_end = data.find(b"\x00", name_start, pos + size)
+        self.assertGreater(name_end, 0)
+        name = data[name_start:name_end].decode("utf-8", errors="replace")
+        # The machine code follows the name inside the load record.
+        self.assertLessEqual(name_end + 1 + code_size, pos + size, name)
+        return name, code_size
+
+    def _check_unwind_info(self, data, pos, size, name, code_size):
+        """Check the FDE and perf header against their code load record."""
+        unwind_data_size, eh_frame_hdr_size = struct.unpack_from(
+            f"{JITDUMP_ENDIAN}QQ", data, pos + UNWIND_DATA_SIZE_OFFSET)
+        self.assertEqual(eh_frame_hdr_size, EH_FRAME_HDR_SIZE)
+        self.assertLessEqual(UNWIND_EH_FRAME_OFFSET + unwind_data_size, size)
+        eh_frame_size = unwind_data_size - eh_frame_hdr_size
+        self.assertGreater(eh_frame_size, 0)
+        start = pos + UNWIND_EH_FRAME_OFFSET
+        eh_frame = data[start:start + eh_frame_size]
+
+        cie_length, cie_id = struct.unpack_from(f"{JITDUMP_ENDIAN}II", eh_frame, 0)
+        self.assertEqual(cie_id, 0, "first entry must be a CIE")
+        self.assertEqual(eh_frame[8], 1, "CIE version must be 1")
+        self.assertEqual(eh_frame[9:12], b"zR\x00")
+        encoding = _fde_pointer_encoding(eh_frame)
+        if encoding == DW_EH_PE_PCREL_SDATA4:
+            fields = "iI"
+        elif encoding == DW_EH_PE_PCREL_ABSPTR:
+            fields = "qQ"
+        else:
+            self.fail(f"unexpected FDE pointer encoding {encoding:#x}")
+        # jit_unwind.c patches initial_location and address_range for
+        # perf's DSO layout, where the .eh_frame follows the code at
+        # code_size rounded up to 8 bytes.
+        pc_offset = 4 + cie_length + 8
+        initial_location, address_range = struct.unpack_from(
+            f"{JITDUMP_ENDIAN}{fields}", eh_frame, pc_offset)
+        self.assertEqual(address_range, code_size, name)
+        rounded_code_size = (code_size + 7) & ~7
+        self.assertEqual(initial_location, -(rounded_code_size + pc_offset), name)
+        # perf's eh_frame_hdr must point back at the code with the same
+        # rounding as the FDE.
+        hdr_from = struct.unpack_from(
+            f"{JITDUMP_ENDIAN}i", data,
+            start + eh_frame_size + EH_FRAME_HDR_FROM_OFFSET)[0]
+        self.assertEqual(hdr_from, -(rounded_code_size + eh_frame_size), name)
+
     def _check_unwinding_records(self, data):
-        """Check every unwinding record against the code load record it
-        describes; return {name: code_size} for the regions seen."""
+        """Return {name: code_size} after checking each unwind/load pair."""
         records = list(_jitdump_records(data))
         regions = {}
         for index, (event, pos, size) in enumerate(records):
             if event != PERF_UNWINDING_INFO:
                 continue
-            # The unwinding info record is immediately followed by the
-            # code load record it describes.
+            # Unwinding info immediately precedes the code it describes.
             self.assertLess(index + 1, len(records))
             load_event, load_pos, load_size = records[index + 1]
             self.assertEqual(load_event, PERF_LOAD)
-            code_size = struct.unpack_from(
-                f"{JITDUMP_ENDIAN}Q", data, load_pos + CODE_LOAD_CODE_SIZE_OFFSET)[0]
-            name_start = load_pos + CODE_LOAD_NAME_OFFSET
-            name_end = data.find(b"\x00", name_start, load_pos + load_size)
-            self.assertGreater(name_end, 0)
-            name = data[name_start:name_end].decode("utf-8", errors="replace")
-            # The machine code follows the name inside the load record.
-            self.assertLessEqual(name_end + 1 + code_size, load_pos + load_size, name)
-            unwind_data_size, eh_frame_hdr_size = struct.unpack_from(
-                f"{JITDUMP_ENDIAN}QQ", data, pos + UNWIND_DATA_SIZE_OFFSET)
-            self.assertEqual(eh_frame_hdr_size, EH_FRAME_HDR_SIZE)
-            self.assertLessEqual(UNWIND_EH_FRAME_OFFSET + unwind_data_size, size)
-            eh_frame_size = unwind_data_size - eh_frame_hdr_size
-            self.assertGreater(eh_frame_size, 0)
-            start = pos + UNWIND_EH_FRAME_OFFSET
-            eh_frame = data[start:start + eh_frame_size]
-
-            cie_length, cie_id = struct.unpack_from(f"{JITDUMP_ENDIAN}II", eh_frame, 0)
-            self.assertEqual(cie_id, 0, "first entry must be a CIE")
-            self.assertEqual(eh_frame[8], 1, "CIE version must be 1")
-            self.assertEqual(eh_frame[9:12], b"zR\x00")
-            encoding = _fde_pointer_encoding(eh_frame)
-            if encoding == DW_EH_PE_PCREL_SDATA4:
-                fields = "iI"
-            elif encoding == DW_EH_PE_PCREL_ABSPTR:
-                fields = "qQ"
-            else:
-                self.fail(f"unexpected FDE pointer encoding {encoding:#x}")
-            # jit_unwind.c patches initial_location and address_range for
-            # perf's DSO layout, where the .eh_frame follows the code at
-            # code_size rounded up to 8 bytes.
-            pc_offset = 4 + cie_length + 8
-            initial_location, address_range = struct.unpack_from(
-                f"{JITDUMP_ENDIAN}{fields}", eh_frame, pc_offset)
-            self.assertEqual(address_range, code_size, name)
-            rounded_code_size = (code_size + 7) & ~7
-            self.assertEqual(initial_location, -(rounded_code_size + pc_offset), name)
-            # perf's eh_frame_hdr must point back at the code with the same
-            # rounding as the FDE.
-            hdr_from = struct.unpack_from(
-                f"{JITDUMP_ENDIAN}i", data,
-                start + eh_frame_size + EH_FRAME_HDR_FROM_OFFSET)[0]
-            self.assertEqual(hdr_from, -(rounded_code_size + eh_frame_size), name)
+            name, code_size = self._check_code_load(data, load_pos, load_size)
+            with self.subTest(region=name):
+                self._check_unwind_info(data, pos, size, name, code_size)
             regions[name] = code_size
         self.assertTrue(regions, "no CodeUnwindingInfoEvent found")
         return regions
@@ -839,164 +845,6 @@ def test_jitdump_unwinding_info(self):
         self.assertTrue(any("my_test_func" in name for name in regions))
 
 
-try:
-    with test_tools.imports_under_tool("jit"):
-        import _trampoline_ehframe
-except ImportError:
-    # Installed Python without the Tools directory.
-    _trampoline_ehframe = None
-
-
-def _fake_cie(*, version=1, augmentation=b"zR", ra_column=16,
-              encoding=DW_EH_PE_PCREL_SDATA4, cie_id=0):
-    """A CIE like the assembler's: code align 1, data align -8, one
-    DW_CFA_def_cfa instruction, padded with DW_CFA_nop to 8 bytes."""
-    body = bytes([version]) + augmentation + b"\x00"
-    body += bytes([1, 0x78, ra_column, 1, encoding])
-    body += bytes([0x0C, 7, 8])  # DW_CFA_def_cfa: r7 (rsp) ofs 8
-    body += b"\x00" * (-(8 + len(body)) % 8)
-    return struct.pack("<II", 4 + len(body), cie_id) + body
-
-
-def _fake_fde(cie_total, *, field_size=4, address_range=8,
-              instructions=b"\x41\x0e\x10\x86\x02"):
-    """An FDE right after a CIE of cie_total bytes, padded to 8 bytes."""
-    body = struct.pack("<I", cie_total + 4)  # CIE pointer, relative to itself
-    # initial_location as an assembler would leave it, the parser zeroes it.
-    body += (-40).to_bytes(field_size, "little", signed=True)
-    body += address_range.to_bytes(field_size, "little")
-    body += b"\x00"  # augmentation data length
-    body += instructions
-    body += b"\x00" * (-(4 + len(body)) % 8)
-    return struct.pack("<I", len(body)) + body
-
-
-@unittest.skipIf(_trampoline_ehframe is None,
-                 "Tools/jit/_trampoline_ehframe.py not found")
-class TestTrampolineEhframeScript(unittest.TestCase):
-    """Tests for Tools/jit/_trampoline_ehframe.py."""
-
-    ehframe = _trampoline_ehframe
-
-    def parse(self, data, text_size=8):
-        return self.ehframe.parse_ehframe(bytes(data), "<", text_size)
-
-    def test_parse(self):
-        """Both FDE pointer encodings: ELF sdata4 and Darwin absptr."""
-        cases = [(DW_EH_PE_PCREL_SDATA4, 4, 16, 8), (DW_EH_PE_PCREL_ABSPTR, 8, 30, 20)]
-        for encoding, field_size, ra_column, text_size in cases:
-            with self.subTest(encoding=hex(encoding)):
-                cie = _fake_cie(encoding=encoding, ra_column=ra_column)
-                fde = _fake_fde(len(cie), field_size=field_size,
-                                address_range=text_size)
-                result = self.parse(cie + fde, text_size)
-                self.assertEqual(result.field_size, field_size)
-                self.assertEqual(result.fde_pc_offset, len(cie) + 8)
-                self.assertEqual(result.fde_range_offset, len(cie) + 8 + field_size)
-                # Both patchable fields zeroed, everything else untouched.
-                expected = bytearray(cie + fde)
-                expected[len(cie) + 8:len(cie) + 8 + 2 * field_size] = bytes(2 * field_size)
-                self.assertEqual(result.data, bytes(expected))
-
-    def test_parse_rejects_malformed(self):
-        cie = _fake_cie()
-        fde = _fake_fde(len(cie))
-        cases = [
-            ("version", _fake_cie(version=3) + fde, 8),
-            ("augmentation", _fake_cie(augmentation=b"zPLR") + fde, 8),
-            ("encoding", _fake_cie(encoding=0x1A) + fde, 8),
-            ("exactly one FDE", cie + fde + fde, 8),
-            ("address_range", cie + fde, 12),
-            ("no FDE", cie, 8),
-        ]
-        for message, data, text_size in cases:
-            with self.subTest(message):
-                with self.assertRaisesRegex(ValueError, message):
-                    self.parse(data, text_size)
-
-    def _build_trampoline_objects(self):
-        """The object(s) the Makefile fed to the generator."""
-        builddir = sysconfig.get_config_var("abs_builddir") or "."
-        universal2 = os.path.join(builddir, "Python", "asm_trampoline_universal2.o")
-        if os.path.exists(universal2):
-            return [universal2]
-        return sorted(
-            path for path in glob.glob(
-                os.path.join(builddir, "Python", "asm_trampoline_*.o"))
-            if "apple-darwin" not in os.path.basename(path))
-
-    def test_macho_thin_and_fat(self):
-        """Mach-O objects and fat containers are parsed with no external tools."""
-        E = self.ehframe
-
-        def macho(cputype, text, eh_frame):
-            # A minimal MH_OBJECT: one __TEXT segment with __text and
-            # __eh_frame sections, section data right after the load command.
-            segment_size = 72 + 2 * 80
-            text_offset = 32 + segment_size
-            eh_offset = text_offset + len(text)
-            sections = b""
-            for name, size, offset in (("__text", len(text), text_offset),
-                                       ("__eh_frame", len(eh_frame), eh_offset)):
-                sections += struct.pack("<16s16sQQIIIIIIII", name.encode(),
-                                        b"__TEXT", 0, size, offset,
-                                        0, 0, 0, 0, 0, 0, 0)
-            segment = struct.pack("<II16sQQQQIIII", E._LC_SEGMENT_64,
-                                  segment_size, b"__TEXT", 0,
-                                  len(text) + len(eh_frame), text_offset,
-                                  len(text) + len(eh_frame), 7, 5, 2, 0)
-            header = struct.pack("<IIIIIIII", E._MH_MAGIC_64, cputype, 0,
-                                 1, 1, segment_size, 0, 0)
-            return header + segment + sections + text + eh_frame
-
-        x86 = macho(E._CPU_TYPE_X86_64, b"\x55\xc3", b"x86 eh_frame")
-        arm = macho(E._CPU_TYPE_ARM64, b"\xc0\x03\x5f\xd6", b"arm64 eh_frame")
-        # The fat header and its fat_arch entries are big-endian.
-        blobs = [(E._CPU_TYPE_X86_64, x86), (E._CPU_TYPE_ARM64, arm)]
-        offset = 8 + 20 * len(blobs)
-        entries = b""
-        body = b""
-        for cputype, blob in blobs:
-            entries += struct.pack(">IIIII", cputype, 0, offset + len(body),
-                                   len(blob), 0)
-            body += blob
-        fat = struct.pack(">II", E._FAT_MAGIC, len(blobs)) + entries + body
-
-        with temp_dir() as tmp:
-            thin_path = os.path.join(tmp, "thin.o")
-            fat_path = os.path.join(tmp, "fat.o")
-            with open(thin_path, "wb") as f:
-                f.write(arm)
-            with open(fat_path, "wb") as f:
-                f.write(fat)
-            (thin,) = E.load_object(thin_path)
-            fat_slices = E.load_object(fat_path)
-
-        self.assertEqual(thin.arch_macro, "__aarch64__")
-        self.assertEqual(thin.sections[".text"], b"\xc0\x03\x5f\xd6")
-        self.assertEqual(thin.sections[".eh_frame"], b"arm64 eh_frame")
-        self.assertEqual([s.arch_macro for s in fat_slices],
-                         ["__x86_64__", "__aarch64__"])
-        self.assertEqual(fat_slices[0].sections[".eh_frame"], b"x86 eh_frame")
-        self.assertEqual(fat_slices[1].sections[".text"], b"\xc0\x03\x5f\xd6")
-
-    def test_generated_header_is_current(self):
-        """The header in the build directory matches a fresh generation."""
-        objects = self._build_trampoline_objects()
-        builddir = sysconfig.get_config_var("abs_builddir") or "."
-        header = os.path.join(builddir, "trampoline_ehframe.h")
-        if not objects or not os.path.exists(header):
-            self.skipTest("trampoline object or generated header not found")
-        with open(header) as f:
-            current = f.read()
-        with temp_dir() as tmp:
-            fresh_path = os.path.join(tmp, "trampoline_ehframe.h")
-            self.ehframe.generate(objects, fresh_path)
-            with open(fresh_path) as f:
-                fresh = f.read()
-        self.assertEqual(current, fresh)
-
-
 class TestTrampolineEhframeHeader(unittest.TestCase):
     """Structural checks on the generated trampoline_ehframe.h data."""