wip 2

coverage/pep669_tracer.py

@@ -6,15 +6,17 @@
 from __future__ import annotations
 
 import atexit
+import dataclasses
+import dis
 import inspect
+import re
 import sys
 import threading
 import traceback
 
 from types import CodeType, FrameType, ModuleType
 from typing import Any, Callable, Dict, List, Optional, Set, Tuple, cast
 
-from coverage import env
 from coverage.types import (
     TArc, TFileDisposition, TLineNo, TTraceData, TTraceFileData, TTraceFn,
     TTracer, TWarnFn,
@@ -26,23 +28,66 @@
 THIS_FILE = __file__.rstrip("co")
 
 
-def log(msg):
+def logfile():
     with open("/tmp/pan.out", "a") as f:
-        print(msg, file=f)
+        yield f
 
-def panopticon(meth):
-    def _wrapped(self, *args, **kwargs):
-        assert not kwargs
-        log(f"{meth.__name__}{args!r}")
-        try:
-            return meth(self, *args, **kwargs)
-        except:
-            with open("/tmp/pan.out", "a") as f:
-                traceback.print_exception(sys.exception(), file=f)
-                sys.monitoring.set_events(sys.monitoring.COVERAGE_ID, 0)
-            raise
-    return _wrapped
+def log(msg):
+    for f in logfile():
+        print(msg, file=f)
 
+FILENAME_SUBS = [
+    (r"/private/var/folders/.*/pytest-of-.*/pytest-\d+/", "/tmp/"),
+]
+
+def arg_repr(arg):
+    match arg:
+        case CodeType():
+            filename = arg.co_filename
+            for pat, sub in FILENAME_SUBS:
+                filename = re.sub(pat, sub, filename)
+            arg_repr = f"<name={arg.co_name}, file={filename!r}@{arg.co_firstlineno}>"
+        case _:
+            arg_repr = repr(arg)
+    return arg_repr
+
+def panopticon(*names):
+    def _decorator(meth):
+        def _wrapped(self, *args, **kwargs):
+            assert not kwargs
+            try:
+                args_reprs = []
+                for name, arg in zip(names, args):
+                    if name is None:
+                        continue
+                    args_reprs.append(f"{name}={arg_repr(arg)}")
+                log(f"{meth.__name__}({', '.join(args_reprs)})")
+                return meth(self, *args)
+            except:
+                with open("/tmp/pan.out", "a") as f:
+                    traceback.print_exception(sys.exception(), file=f)
+                    sys.monitoring.set_events(sys.monitoring.COVERAGE_ID, 0)
+                raise
+        return _wrapped
+    return _decorator
+
+
+@dataclasses.dataclass
+class CodeInfo:
+    tracing: bool
+    file_data: Optional[TTraceFileData]
+    byte_to_line: Dict[int, int]
+
+
+def bytes_to_lines(code):
+    b2l = {}
+    cur_line = None
+    for inst in dis.get_instructions(code):
+        if inst.starts_line is not None:
+            cur_line = inst.starts_line
+        b2l[inst.offset] = cur_line
+    log(f"--> bytes_to_lines: {b2l!r}")
+    return b2l
 
 class Pep669Tracer(TTracer):
     """Python implementation of the raw data tracer for PEP669 implementations."""
@@ -64,9 +109,11 @@ def __init__(self) -> None:
         self.cur_file_data: Optional[TTraceFileData] = None
         self.last_line: TLineNo = 0
         self.cur_file_name: Optional[str] = None
-        #self.context: Optional[str] = None
 
-        self.code_cache: Dict[CodeType, Tuple[bool, Optional[TTraceFileData]]] = {}
+        self.code_infos: Dict[CodeType, CodeInfo] = {}
+        self.stats = {
+            "starts": 0,
+        }
 
         # The frame_stack parallels the Python call stack. Each entry is
         # information about an active frame, a three-element tuple:
@@ -134,16 +181,18 @@ def start(self) -> TTraceFn:
         self.myid = sys.monitoring.COVERAGE_ID
         sys.monitoring.use_tool_id(self.myid, "coverage.py")
         events = sys.monitoring.events
-        sys.monitoring.set_events(self.myid, events.PY_START)
+        sys.monitoring.set_events(
+            self.myid,
+            events.PY_START | events.PY_RETURN | events.PY_RESUME | events.PY_YIELD,
+        )
         sys.monitoring.register_callback(self.myid, events.PY_START, self.sysmon_py_start)
-        # Use PY_START globally, then use set_local_event(LINE) for interesting
-        # frames, so i might not need to bookkeep which are the interesting frame.
         sys.monitoring.register_callback(self.myid, events.PY_RESUME, self.sysmon_py_resume)
         sys.monitoring.register_callback(self.myid, events.PY_RETURN, self.sysmon_py_return)
         sys.monitoring.register_callback(self.myid, events.PY_YIELD, self.sysmon_py_yield)
         # UNWIND is like RETURN/YIELD
         sys.monitoring.register_callback(self.myid, events.LINE, self.sysmon_line)
-        #sys.monitoring.register_callback(self.myid, events.BRANCH, self.sysmon_branch)
+        sys.monitoring.register_callback(self.myid, events.BRANCH, self.sysmon_branch)
+        sys.monitoring.register_callback(self.myid, events.JUMP, self.sysmon_jump)
 
     def stop(self) -> None:
         """Stop this Tracer."""
@@ -161,16 +210,23 @@ def reset_activity(self) -> None:
     def get_stats(self) -> Optional[Dict[str, int]]:
         """Return a dictionary of statistics, or None."""
         return None
+        return self.stats | {
+            "codes": len(self.code_infos),
+            "codes_tracing": sum(1 for ci in self.code_infos.values() if ci.tracing),
+        }
 
-    @panopticon
+    @panopticon("code", "@")
     def sysmon_py_start(self, code, instruction_offset: int):
         # Entering a new frame.  Decide if we should trace in this file.
         self._activity = True
+        self.stats["starts"] += 1
 
         self.frame_stack.append((self.cur_file_data, self.cur_file_name, self.last_line))
 
-        if code in self.code_cache:
-            tracing_code, self.cur_file_data = self.code_cache[code]
+        code_info = self.code_infos.get(code)
+        if code_info is not None:
+            tracing_code = code_info.tracing
+            self.cur_file_data = code_info.file_data
         else:
             tracing_code = self.cur_file_data = None
 
@@ -189,47 +245,50 @@ def sysmon_py_start(self, code, instruction_offset: int):
                 if tracename not in self.data:
                     self.data[tracename] = set()    # type: ignore[assignment]
                 self.cur_file_data = self.data[tracename]
+                b2l = bytes_to_lines(code)
             else:
                 self.cur_file_data = None
+                b2l = None
 
-        self.code_cache[code] = (tracing_code, self.cur_file_data)
+            self.code_infos[code] = CodeInfo(
+                tracing=tracing_code,
+                file_data=self.cur_file_data,
+                byte_to_line=b2l,
+            )
 
-        if tracing_code:
-            events = sys.monitoring.events
-            log(f"set_local_events({code!r})")
-            sys.monitoring.set_local_events(
-                self.myid,
-                code,
-                (
+            if tracing_code:
+                events = sys.monitoring.events
+                log(f"set_local_events(code={arg_repr(code)})")
+                sys.monitoring.set_local_events(
+                    self.myid,
+                    code,
                     sys.monitoring.events.LINE |
-                    sys.monitoring.events.PY_RETURN |
-                    sys.monitoring.events.PY_RESUME |
-                    sys.monitoring.events.PY_YIELD
+                    sys.monitoring.events.BRANCH |
+                    sys.monitoring.events.JUMP,
                 )
-            )
 
         self.last_line = -code.co_firstlineno
 
-    @panopticon
+    @panopticon("code", "@")
     def sysmon_py_resume(self, code, instruction_offset: int):
         self.frame_stack.append((self.cur_file_data, self.cur_file_name, self.last_line))
         frame = inspect.currentframe()
         self.last_line = frame.f_lineno
 
-    @panopticon
+    @panopticon("code", "@", None)
     def sysmon_py_return(self, code, instruction_offset: int, retval: object):
         if self.cur_file_data is not None:
-            cast(Set[TArc], self.cur_file_data).add((self.last_line, -code.co_firstlineno))
+            if self.trace_arcs:
+                cast(Set[TArc], self.cur_file_data).add((self.last_line, -code.co_firstlineno))
 
         # Leaving this function, pop the filename stack.
-        self.cur_file_data, self.cur_file_name, self.last_line = (
-            self.frame_stack.pop()
-        )
+        if self.frame_stack:
+            self.cur_file_data, self.cur_file_name, self.last_line =  self.frame_stack.pop()
 
     def sysmon_py_yield(self, code, instruction_offset: int, retval: object):
         ...
 
-    @panopticon
+    @panopticon("code", "line")
     def sysmon_line(self, code, line_number: int):
         #assert self.cur_file_data is not None
         if self.cur_file_data is not None:
@@ -238,8 +297,12 @@ def sysmon_line(self, code, line_number: int):
             else:
                 cast(Set[TLineNo], self.cur_file_data).add(line_number)
             self.last_line = line_number
-        return sys.monitoring.DISABLE
+        #return sys.monitoring.DISABLE
 
-    @panopticon
+    @panopticon("code", "from@", "to@")
     def sysmon_branch(self, code, instruction_offset: int, destination_offset: int):
         ...
+
+    @panopticon("code", "from@", "to@")
+    def sysmon_jump(self, code, instruction_offset: int, destination_offset: int):
+        ...

tests/coveragetest.py

@@ -168,8 +168,6 @@ def check_coverage(
         # Coverage.py wants to deal with things as modules with file names.
         modname = self.get_module_name()
 
-        #import dis,textwrap; dis.dis(textwrap.dedent(text))
-
         self.make_file(modname + ".py", text)
 
         if arcs is None and arcz is not None: