coverage/sysmon.py - external/github.com/nedbat/coveragepy - Git at Google

 # Licensed under the Apache License: http://www.apache.org/licenses/LICENSE-2.0
 # For details: https://github.com/coveragepy/coveragepy/blob/main/NOTICE.txt

 """Callback functions and support for sys.monitoring data collection."""

 from __future__ import annotations

 import collections
 import functools
 import inspect
 import os
 import os.path
 import sys
 import threading
 import tokenize
 import traceback
 from collections.abc import Callable
 from dataclasses import dataclass
 from types import CodeType
 from typing import Any, NewType, Optional, cast

 from coverage import env
 from coverage.bytecode import BranchArcResolver, bytes_to_lines
 from coverage.debug import short_filename, short_stack
 from coverage.exceptions import NoSource
 from coverage.misc import isolate_module
 from coverage.parser import multiline_map_from_text
 from coverage.python import get_python_source
 from coverage.types import (
     AnyCallable,
     TFileDisposition,
     TLineNo,
     TOffset,
     Tracer,
     TShouldStartContextFn,
     TShouldTraceFn,
     TTraceData,
     TTraceFileData,
     TWarnFn,
 )

 # Only needed for some of the commented-out logging:
 # from coverage.debug import ppformat

 os = isolate_module(os)

 # pylint: disable=unused-argument

 # $set_env.py: COVERAGE_SYSMON_LOG - Log sys.monitoring activity
 LOG = bool(int(os.getenv("COVERAGE_SYSMON_LOG", 0)))

 # $set_env.py: COVERAGE_SYSMON_STATS - Collect sys.monitoring stats
 COLLECT_STATS = bool(int(os.getenv("COVERAGE_SYSMON_STATS", 0)))

 # This module will be imported in all versions of Python, but only used in 3.12+
 # It will be type-checked for 3.12, but not for earlier versions.
 sys_monitoring = getattr(sys, "monitoring", None)

 DISABLE_TYPE = NewType("DISABLE_TYPE", object)
 MonitorReturn = Optional[DISABLE_TYPE]
 DISABLE = cast(MonitorReturn, getattr(sys_monitoring, "DISABLE", None))


 if LOG:  # pragma: debugging

     class LoggingWrapper:
         """Wrap a namespace to log all its functions."""

         def __init__(self, wrapped: Any, namespace: str) -> None:
             self.wrapped = wrapped
             self.namespace = namespace

         def __getattr__(self, name: str) -> Callable[..., Any]:
             def _wrapped(*args: Any, **kwargs: Any) -> Any:
                 log(f"{self.namespace}.{name}{args}{kwargs}")
                 return getattr(self.wrapped, name)(*args, **kwargs)

             return _wrapped

     sys_monitoring = LoggingWrapper(sys_monitoring, "sys.monitoring")
     assert sys_monitoring is not None

     short_stack = functools.partial(
         short_stack,
         full=True,
         short_filenames=True,
         frame_ids=True,
     )
     seen_threads: set[int] = set()

     def log(msg: str) -> None:
         """Write a message to our detailed debugging log(s)."""
         # Thread ids are reused across processes?
         # Make a shorter number more likely to be unique.
         pid = os.getpid()
         tid = cast(int, threading.current_thread().ident)
         tslug = f"{(pid * tid) % 9_999_991:07d}"
         if tid not in seen_threads:
             seen_threads.add(tid)
             log(f"New thread {tid} {tslug}:\n{short_stack()}")
         # log_seq = int(os.getenv("PANSEQ", "0"))
         # root = f"/tmp/pan.{log_seq:03d}"
         for filename in [
             "/tmp/foo.out",
             # f"{root}.out",
             # f"{root}-{pid}.out",
             # f"{root}-{pid}-{tslug}.out",
         ]:
             with open(filename, "a", encoding="utf-8") as f:
                 try:
                     print(f"{pid}:{tslug}: {msg}", file=f, flush=True)
                 except UnicodeError:
                     print(f"{pid}:{tslug}: {ascii(msg)}", file=f, flush=True)

     def arg_repr(arg: Any) -> str:
         """Make a customized repr for logged values."""
         if isinstance(arg, CodeType):
             return (
                 f"<code @{id(arg):#x}"
                 + f" name={arg.co_name},"
                 + f" file={short_filename(arg.co_filename)!r}#{arg.co_firstlineno}>"
             )
         return repr(arg)

     def panopticon(*names: str | None) -> AnyCallable:
         """Decorate a function to log its calls."""

         def _decorator(method: AnyCallable) -> AnyCallable:
             @functools.wraps(method)
             def _wrapped(self: Any, *args: Any) -> Any:
                 try:
                     # log(f"{method.__name__}() stack:\n{short_stack()}")
                     args_reprs = []
                     for name, arg in zip(names, args):
                         if name is None:
                             continue
                         args_reprs.append(f"{name}={arg_repr(arg)}")
                     log(f"{id(self):#x}:{method.__name__}({', '.join(args_reprs)})")
                     ret = method(self, *args)
                     # log(f" end {id(self):#x}:{method.__name__}({', '.join(args_reprs)})")
                     return ret
                 except Exception as exc:
                     log(f"!!{exc.__class__.__name__}: {exc}")
                     if 1:
                         log("".join(traceback.format_exception(exc)))
                     try:
                         assert sys_monitoring is not None
                         sys_monitoring.set_events(sys.monitoring.COVERAGE_ID, 0)
                     except ValueError:
                         # We might have already shut off monitoring.
                         log("oops, shutting off events with disabled tool id")
                     raise

             return _wrapped

         return _decorator

 else:

     def log(msg: str) -> None:
         """Write a message to our detailed debugging log(s), but not really."""

     def panopticon(*names: str | None) -> AnyCallable:
         """Decorate a function to log its calls, but not really."""

         def _decorator(meth: AnyCallable) -> AnyCallable:
             return meth

         return _decorator


 @dataclass
 class CodeInfo:
     """The information we want about each code object."""

     tracing: bool
     file_data: TTraceFileData | None
     byte_to_line: dict[TOffset, TLineNo] | None

     # Lazily-created resolver of branch events to arcs, created on the
     # first branch event in the code object.
     branch_resolver: BranchArcResolver | None


 class SysMonitor(Tracer):
     """Python implementation of the raw data tracer for PEP669 implementations."""

     # One of these will be used across threads. Be careful.

     def __init__(self) -> None:
         # Attributes set from the collector:
         self.data: TTraceData
         self.trace_arcs = False
         self.should_trace: TShouldTraceFn
         self.should_trace_cache: dict[str, TFileDisposition | None]
         # TODO: should_start_context and switch_context are unused!
         # Change tests/testenv.py:DYN_CONTEXTS when this is updated.
         self.should_start_context: TShouldStartContextFn | None = None
         self.switch_context: Callable[[str | None], None] | None = None
         self.lock_data: Callable[[], None]
         self.unlock_data: Callable[[], None]
         # TODO: warn is unused.
         self.warn: TWarnFn

         assert sys_monitoring is not None
         # sys.monitoring pre-allocates tool ids, but it's kind of pointless.
         # There's no guarantee that "our" tool id will still be available, so
         # we have to search for a usable one in start() anyway.
         self.myid = sys_monitoring.COVERAGE_ID

         # Map id(code_object) -> CodeInfo
         self.code_infos: dict[int, CodeInfo] = {}
         # A list of code_objects, just to keep them alive so that id's are
         # useful as identity.
         self.code_objects: list[CodeType] = []

         # Map filename:__name__ -> set(id(code_object))
         self.filename_code_ids: dict[str, set[int]] = collections.defaultdict(set)

         # Map filename -> multiline map, so each file is parsed at most once.
         self.multiline_maps: dict[str, dict[TLineNo, TLineNo]] = {}

         self.sysmon_on = False
         self.lock = threading.Lock()

         self.stats: dict[str, int] | None = None
         if COLLECT_STATS:
             self.stats = dict.fromkeys(
                 "starts start_tracing returns line_lines line_arcs branches branch_trails".split(),
                 0,
             )

         self._activity = False

     def __repr__(self) -> str:
         points = sum(len(v) for v in self.data.values())
         files = len(self.data)
         return f"<SysMonitor at {id(self):#x}: {points} data points in {files} files>"

     @panopticon()
     def start(self) -> None:
         """Start this Tracer."""
         with self.lock:
             assert sys_monitoring is not None
             while self.myid <= 5:
                 try:
                     sys_monitoring.use_tool_id(self.myid, "coverage.py")
                     break
                 except ValueError:
                     self.myid += 1
                     continue
             else:
                 raise RuntimeError("No sys.monitoring tool id is available")
             register = functools.partial(sys_monitoring.register_callback, self.myid)
             events = sys.monitoring.events

             sys_monitoring.set_events(self.myid, events.PY_START)
             register(events.PY_START, self.sysmon_py_start)
             if self.trace_arcs:
                 register(events.PY_RETURN, self.sysmon_py_return)
                 register(events.LINE, self.sysmon_line_arcs)
                 if env.PYBEHAVIOR.branch_right_left:
                     register(events.BRANCH_RIGHT, self.sysmon_branch_either)
                     register(events.BRANCH_LEFT, self.sysmon_branch_either)
             else:
                 register(events.LINE, self.sysmon_line_lines)
             sys_monitoring.restart_events()
             self.sysmon_on = True

     @panopticon()
     def stop(self) -> None:
         """Stop this Tracer."""
         with self.lock:
             if not self.sysmon_on:
                 # In forking situations, we might try to stop when we are not
                 # started.  Do nothing in that case.
                 return
             assert sys_monitoring is not None
             sys_monitoring.set_events(self.myid, 0)
             self.sysmon_on = False
             sys_monitoring.free_tool_id(self.myid)

         if LOG:  # pragma: debugging
             items = sorted(
                 self.filename_code_ids.items(),
                 key=lambda item: len(item[1]),
                 reverse=True,
             )
             code_objs = sum(len(code_ids) for _, code_ids in items)
             dupes = code_objs - len(items)
             if dupes:
                 log(f"==== Duplicate code objects: {dupes} duplicates, {code_objs} total")
                 for filename, code_ids in items:
                     if len(code_ids) > 1:
                         log(f"{len(code_ids):>5} objects: {filename}")
             else:
                 log("==== Duplicate code objects: none")

     @panopticon()
     def post_fork(self) -> None:
         """The process has forked, clean up as needed."""
         self.stop()

     def activity(self) -> bool:
         """Has there been any activity?"""
         return self._activity

     def reset_activity(self) -> None:
         """Reset the activity() flag."""
         self._activity = False

     def get_stats(self) -> dict[str, int] | None:
         """Return a dictionary of statistics, or None."""
         return self.stats

     @panopticon("code", "@")
     def sysmon_py_start(self, code: CodeType, instruction_offset: TOffset) -> MonitorReturn:
         """Handle sys.monitoring.events.PY_START events."""
         self._activity = True
         if self.stats is not None:
             self.stats["starts"] += 1

         if code.co_name == "__annotate__":
             # Type annotation code objects don't execute, ignore them.
             return DISABLE

         # Entering a new frame.  Decide if we should trace in this file.
         code_info = self.code_infos.get(id(code))
         tracing_code: bool | None = None
         file_data: TTraceFileData | None = None
         if code_info is not None:
             tracing_code = code_info.tracing
             file_data = code_info.file_data

         if tracing_code is None:
             filename = code.co_filename
             disp = self.should_trace_cache.get(filename)
             if disp is None:
                 frame = inspect.currentframe()
                 if frame is not None:
                     frame = inspect.currentframe().f_back  # type: ignore[union-attr]
                     if LOG:  # pragma: debugging
                         # @panopticon adds a frame.
                         frame = frame.f_back  # type: ignore[union-attr]
                 disp = self.should_trace(filename, frame)  # type: ignore[arg-type]
                 self.should_trace_cache[filename] = disp

             tracing_code = disp.trace
             if tracing_code:
                 tracename = disp.source_filename
                 assert tracename is not None
                 self.lock_data()
                 try:
                     if tracename not in self.data:
                         self.data[tracename] = set()
                 finally:
                     self.unlock_data()
                 file_data = self.data[tracename]
                 # byte_to_line is only read by the arc callbacks
                 b2l = bytes_to_lines(code) if self.trace_arcs else None
             else:
                 file_data = None
                 b2l = None

             code_info = CodeInfo(
                 tracing=tracing_code,
                 file_data=file_data,
                 byte_to_line=b2l,
                 branch_resolver=None,
             )
             self.code_infos[id(code)] = code_info
             self.code_objects.append(code)

             if tracing_code:
                 if self.stats is not None:
                     self.stats["start_tracing"] += 1
                 events = sys.monitoring.events
                 with self.lock:
                     if self.sysmon_on:
                         assert sys_monitoring is not None
                         local_events = events.LINE
                         if self.trace_arcs:
                             assert env.PYBEHAVIOR.branch_right_left
                             local_events |= (
                                 events.PY_RETURN | events.BRANCH_RIGHT | events.BRANCH_LEFT
                             )
                         sys_monitoring.set_local_events(self.myid, code, local_events)

                         if LOG:  # pragma: debugging
                             if code.co_filename not in {"<string>"}:
                                 self.filename_code_ids[f"{code.co_filename}:{code.co_name}"].add(
                                     id(code)
                                 )

         return DISABLE

     @panopticon("code", "@", None)
     def sysmon_py_return(
         self,
         code: CodeType,
         instruction_offset: TOffset,
         retval: object,
     ) -> MonitorReturn:
         """Handle sys.monitoring.events.PY_RETURN events for branch coverage."""
         if self.stats is not None:
             self.stats["returns"] += 1
         code_info = self.code_infos.get(id(code))
         # code_info is not None and code_info.file_data is not None, since we
         # wouldn't have enabled this event if they were.
         last_line = code_info.byte_to_line.get(instruction_offset)  # type: ignore
         if last_line is not None:
             arc = (last_line, -code.co_firstlineno)
             code_info.file_data.add(arc)  # type: ignore
             # log(f"adding {arc=}")
         return DISABLE

     @panopticon("code", "line")
     def sysmon_line_lines(self, code: CodeType, line_number: TLineNo) -> MonitorReturn:
         """Handle sys.monitoring.events.LINE events for line coverage."""
         if self.stats is not None:
             self.stats["line_lines"] += 1
         code_info = self.code_infos.get(id(code))
         # It should be true that code_info is not None and code_info.file_data
         # is not None, since we wouldn't have enabled this event if they were.
         # But somehow code_info can be None here, so we have to check.
         if code_info is not None and code_info.file_data is not None:
             code_info.file_data.add(line_number)  # type: ignore
         # log(f"adding {line_number=}")
         return DISABLE

     @panopticon("code", "line")
     def sysmon_line_arcs(self, code: CodeType, line_number: TLineNo) -> MonitorReturn:
         """Handle sys.monitoring.events.LINE events for branch coverage."""
         if self.stats is not None:
             self.stats["line_arcs"] += 1
         code_info = self.code_infos[id(code)]
         # code_info is not None and code_info.file_data is not None, since we
         # wouldn't have enabled this event if they were.
         arc = (line_number, line_number)
         code_info.file_data.add(arc)  # type: ignore
         # log(f"adding {arc=}")
         return DISABLE

     @panopticon("code", "@", "@")
     def sysmon_branch_either(
         self, code: CodeType, instruction_offset: TOffset, destination_offset: TOffset
     ) -> MonitorReturn:
         """Handle BRANCH_RIGHT and BRANCH_LEFT events."""
         if self.stats is not None:
             self.stats["branches"] += 1
         code_info = self.code_infos[id(code)]
         # code_info is not None and code_info.file_data is not None, since we
         # wouldn't have enabled this event if they were.
         resolver = code_info.branch_resolver
         if resolver is None:
             if self.stats is not None:
                 self.stats["branch_trails"] += 1
             assert code_info.byte_to_line is not None
             resolver = code_info.branch_resolver = BranchArcResolver(
                 code,
                 code_info.byte_to_line,
                 self.get_multiline_map(code.co_filename),
             )
         arc = resolver.resolve(instruction_offset, destination_offset)
         if arc is not None:
             code_info.file_data.add(arc)  # type: ignore
             # log(f"adding {arc=}")
         else:
             # This could be an exception jumping from line to line.
             assert code_info.byte_to_line is not None
             l1 = code_info.byte_to_line.get(instruction_offset)
             if l1 is not None:
                 l2 = code_info.byte_to_line.get(destination_offset)
                 if l2 is not None and l1 != l2:
                     arc = (l1, l2)
                     code_info.file_data.add(arc)  # type: ignore
                     # log(f"adding unforeseen {arc=}")

         return DISABLE

     def get_multiline_map(self, filename: str) -> dict[TLineNo, TLineNo]:
         """Get the multiline map for `filename`, computing it at most once."""
         multiline_map = self.multiline_maps.get(filename)
         if multiline_map is None:
             multiline_map = self.multiline_maps[filename] = compute_multiline_map(filename)
         return multiline_map


 def compute_multiline_map(filename: str) -> dict[TLineNo, TLineNo]:
     """Tokenize `filename` and return its multiline map."""
     try:
         text = get_python_source(filename)
     except (OSError, NoSource):
         # This can happen if open() in python.py fails.
         return {}
     try:
         return multiline_map_from_text(text)
     except (tokenize.TokenError, IndentationError, SyntaxError):
         # The file was not Python. This can happen when the code object refers
         # to an original non-Python source file, like a Jinja template.
         # In that case, just return an empty map, which might lead to slightly
         # wrong branch coverage, but we don't have any better option.
         return {}
	# Licensed under the Apache License: http://www.apache.org/licenses/LICENSE-2.0
	# For details: https://github.com/coveragepy/coveragepy/blob/main/NOTICE.txt

	"""Callback functions and support for sys.monitoring data collection."""

	from __future__ import annotations

	import collections
	import functools
	import inspect
	import os
	import os.path
	import sys
	import threading
	import tokenize
	import traceback
	from collections.abc import Callable
	from dataclasses import dataclass
	from types import CodeType
	from typing import Any, NewType, Optional, cast

	from coverage import env
	from coverage.bytecode import BranchArcResolver, bytes_to_lines
	from coverage.debug import short_filename, short_stack
	from coverage.exceptions import NoSource
	from coverage.misc import isolate_module
	from coverage.parser import multiline_map_from_text
	from coverage.python import get_python_source
	from coverage.types import (
	AnyCallable,
	TFileDisposition,
	TLineNo,
	TOffset,
	Tracer,
	TShouldStartContextFn,
	TShouldTraceFn,
	TTraceData,
	TTraceFileData,
	TWarnFn,
	)

	# Only needed for some of the commented-out logging:
	# from coverage.debug import ppformat

	os = isolate_module(os)

	# pylint: disable=unused-argument

	# $set_env.py: COVERAGE_SYSMON_LOG - Log sys.monitoring activity
	LOG = bool(int(os.getenv("COVERAGE_SYSMON_LOG", 0)))

	# $set_env.py: COVERAGE_SYSMON_STATS - Collect sys.monitoring stats
	COLLECT_STATS = bool(int(os.getenv("COVERAGE_SYSMON_STATS", 0)))

	# This module will be imported in all versions of Python, but only used in 3.12+
	# It will be type-checked for 3.12, but not for earlier versions.
	sys_monitoring = getattr(sys, "monitoring", None)

	DISABLE_TYPE = NewType("DISABLE_TYPE", object)
	MonitorReturn = Optional[DISABLE_TYPE]
	DISABLE = cast(MonitorReturn, getattr(sys_monitoring, "DISABLE", None))


	if LOG: # pragma: debugging

	class LoggingWrapper:
	"""Wrap a namespace to log all its functions."""

	def __init__(self, wrapped: Any, namespace: str) -> None:
	self.wrapped = wrapped
	self.namespace = namespace

	def __getattr__(self, name: str) -> Callable[..., Any]:
	def _wrapped(args: Any, *kwargs: Any) -> Any:
	log(f"{self.namespace}.{name}{args}{kwargs}")
	return getattr(self.wrapped, name)(args, *kwargs)

	return _wrapped

	sys_monitoring = LoggingWrapper(sys_monitoring, "sys.monitoring")
	assert sys_monitoring is not None

	short_stack = functools.partial(
	short_stack,
	full=True,
	short_filenames=True,
	frame_ids=True,
	)
	seen_threads: set[int] = set()

	def log(msg: str) -> None:
	"""Write a message to our detailed debugging log(s)."""
	# Thread ids are reused across processes?
	# Make a shorter number more likely to be unique.
	pid = os.getpid()
	tid = cast(int, threading.current_thread().ident)
	tslug = f"{(pid * tid) % 9_999_991:07d}"
	if tid not in seen_threads:
	seen_threads.add(tid)
	log(f"New thread {tid} {tslug}:\n{short_stack()}")
	# log_seq = int(os.getenv("PANSEQ", "0"))
	# root = f"/tmp/pan.{log_seq:03d}"
	for filename in [
	"/tmp/foo.out",
	# f"{root}.out",
	# f"{root}-{pid}.out",
	# f"{root}-{pid}-{tslug}.out",
	]:
	with open(filename, "a", encoding="utf-8") as f:
	try:
	print(f"{pid}:{tslug}: {msg}", file=f, flush=True)
	except UnicodeError:
	print(f"{pid}:{tslug}: {ascii(msg)}", file=f, flush=True)

	def arg_repr(arg: Any) -> str:
	"""Make a customized repr for logged values."""
	if isinstance(arg, CodeType):
	return (
	f"<code @{id(arg):#x}"
	+ f" name={arg.co_name},"
	+ f" file={short_filename(arg.co_filename)!r}#{arg.co_firstlineno}>"
	)
	return repr(arg)

	def panopticon(*names: str \| None) -> AnyCallable:
	"""Decorate a function to log its calls."""

	def _decorator(method: AnyCallable) -> AnyCallable:
	@functools.wraps(method)
	def _wrapped(self: Any, *args: Any) -> Any:
	try:
	# log(f"{method.__name__}() stack:\n{short_stack()}")
	args_reprs = []
	for name, arg in zip(names, args):
	if name is None:
	continue
	args_reprs.append(f"{name}={arg_repr(arg)}")
	log(f"{id(self):#x}:{method.__name__}({', '.join(args_reprs)})")
	ret = method(self, *args)
	# log(f" end {id(self):#x}:{method.__name__}({', '.join(args_reprs)})")
	return ret
	except Exception as exc:
	log(f"!!{exc.__class__.__name__}: {exc}")
	if 1:
	log("".join(traceback.format_exception(exc)))
	try:
	assert sys_monitoring is not None
	sys_monitoring.set_events(sys.monitoring.COVERAGE_ID, 0)
	except ValueError:
	# We might have already shut off monitoring.
	log("oops, shutting off events with disabled tool id")
	raise

	return _wrapped

	return _decorator

	else:

	def log(msg: str) -> None:
	"""Write a message to our detailed debugging log(s), but not really."""

	def panopticon(*names: str \| None) -> AnyCallable:
	"""Decorate a function to log its calls, but not really."""

	def _decorator(meth: AnyCallable) -> AnyCallable:
	return meth

	return _decorator


	@dataclass
	class CodeInfo:
	"""The information we want about each code object."""

	tracing: bool
	file_data: TTraceFileData \| None
	byte_to_line: dict[TOffset, TLineNo] \| None

	# Lazily-created resolver of branch events to arcs, created on the
	# first branch event in the code object.
	branch_resolver: BranchArcResolver \| None


	class SysMonitor(Tracer):
	"""Python implementation of the raw data tracer for PEP669 implementations."""

	# One of these will be used across threads. Be careful.

	def __init__(self) -> None:
	# Attributes set from the collector:
	self.data: TTraceData
	self.trace_arcs = False
	self.should_trace: TShouldTraceFn
	self.should_trace_cache: dict[str, TFileDisposition \| None]
	# TODO: should_start_context and switch_context are unused!
	# Change tests/testenv.py:DYN_CONTEXTS when this is updated.
	self.should_start_context: TShouldStartContextFn \| None = None
	self.switch_context: Callable[[str \| None], None] \| None = None
	self.lock_data: Callable[[], None]
	self.unlock_data: Callable[[], None]
	# TODO: warn is unused.
	self.warn: TWarnFn

	assert sys_monitoring is not None
	# sys.monitoring pre-allocates tool ids, but it's kind of pointless.
	# There's no guarantee that "our" tool id will still be available, so
	# we have to search for a usable one in start() anyway.
	self.myid = sys_monitoring.COVERAGE_ID

	# Map id(code_object) -> CodeInfo
	self.code_infos: dict[int, CodeInfo] = {}
	# A list of code_objects, just to keep them alive so that id's are
	# useful as identity.
	self.code_objects: list[CodeType] = []

	# Map filename:__name__ -> set(id(code_object))
	self.filename_code_ids: dict[str, set[int]] = collections.defaultdict(set)

	# Map filename -> multiline map, so each file is parsed at most once.
	self.multiline_maps: dict[str, dict[TLineNo, TLineNo]] = {}

	self.sysmon_on = False
	self.lock = threading.Lock()

	self.stats: dict[str, int] \| None = None
	if COLLECT_STATS:
	self.stats = dict.fromkeys(
	"starts start_tracing returns line_lines line_arcs branches branch_trails".split(),
	0,
	)

	self._activity = False

	def __repr__(self) -> str:
	points = sum(len(v) for v in self.data.values())
	files = len(self.data)
	return f"<SysMonitor at {id(self):#x}: {points} data points in {files} files>"

	@panopticon()
	def start(self) -> None:
	"""Start this Tracer."""
	with self.lock:
	assert sys_monitoring is not None
	while self.myid <= 5:
	try:
	sys_monitoring.use_tool_id(self.myid, "coverage.py")
	break
	except ValueError:
	self.myid += 1
	continue
	else:
	raise RuntimeError("No sys.monitoring tool id is available")
	register = functools.partial(sys_monitoring.register_callback, self.myid)
	events = sys.monitoring.events

	sys_monitoring.set_events(self.myid, events.PY_START)
	register(events.PY_START, self.sysmon_py_start)
	if self.trace_arcs:
	register(events.PY_RETURN, self.sysmon_py_return)
	register(events.LINE, self.sysmon_line_arcs)
	if env.PYBEHAVIOR.branch_right_left:
	register(events.BRANCH_RIGHT, self.sysmon_branch_either)
	register(events.BRANCH_LEFT, self.sysmon_branch_either)
	else:
	register(events.LINE, self.sysmon_line_lines)
	sys_monitoring.restart_events()
	self.sysmon_on = True

	@panopticon()
	def stop(self) -> None:
	"""Stop this Tracer."""
	with self.lock:
	if not self.sysmon_on:
	# In forking situations, we might try to stop when we are not
	# started. Do nothing in that case.
	return
	assert sys_monitoring is not None
	sys_monitoring.set_events(self.myid, 0)
	self.sysmon_on = False
	sys_monitoring.free_tool_id(self.myid)

	if LOG: # pragma: debugging
	items = sorted(
	self.filename_code_ids.items(),
	key=lambda item: len(item[1]),
	reverse=True,
	)
	code_objs = sum(len(code_ids) for _, code_ids in items)
	dupes = code_objs - len(items)
	if dupes:
	log(f"==== Duplicate code objects: {dupes} duplicates, {code_objs} total")
	for filename, code_ids in items:
	if len(code_ids) > 1:
	log(f"{len(code_ids):>5} objects: {filename}")
	else:
	log("==== Duplicate code objects: none")

	@panopticon()
	def post_fork(self) -> None:
	"""The process has forked, clean up as needed."""
	self.stop()

	def activity(self) -> bool:
	"""Has there been any activity?"""
	return self._activity

	def reset_activity(self) -> None:
	"""Reset the activity() flag."""
	self._activity = False

	def get_stats(self) -> dict[str, int] \| None:
	"""Return a dictionary of statistics, or None."""
	return self.stats

	@panopticon("code", "@")
	def sysmon_py_start(self, code: CodeType, instruction_offset: TOffset) -> MonitorReturn:
	"""Handle sys.monitoring.events.PY_START events."""
	self._activity = True
	if self.stats is not None:
	self.stats["starts"] += 1

	if code.co_name == "__annotate__":
	# Type annotation code objects don't execute, ignore them.
	return DISABLE

	# Entering a new frame. Decide if we should trace in this file.
	code_info = self.code_infos.get(id(code))
	tracing_code: bool \| None = None
	file_data: TTraceFileData \| None = None
	if code_info is not None:
	tracing_code = code_info.tracing
	file_data = code_info.file_data

	if tracing_code is None:
	filename = code.co_filename
	disp = self.should_trace_cache.get(filename)
	if disp is None:
	frame = inspect.currentframe()
	if frame is not None:
	frame = inspect.currentframe().f_back # type: ignore[union-attr]
	if LOG: # pragma: debugging
	# @panopticon adds a frame.
	frame = frame.f_back # type: ignore[union-attr]
	disp = self.should_trace(filename, frame) # type: ignore[arg-type]
	self.should_trace_cache[filename] = disp

	tracing_code = disp.trace
	if tracing_code:
	tracename = disp.source_filename
	assert tracename is not None
	self.lock_data()
	try:
	if tracename not in self.data:
	self.data[tracename] = set()
	finally:
	self.unlock_data()
	file_data = self.data[tracename]
	# byte_to_line is only read by the arc callbacks
	b2l = bytes_to_lines(code) if self.trace_arcs else None
	else:
	file_data = None
	b2l = None

	code_info = CodeInfo(
	tracing=tracing_code,
	file_data=file_data,
	byte_to_line=b2l,
	branch_resolver=None,
	)
	self.code_infos[id(code)] = code_info
	self.code_objects.append(code)

	if tracing_code:
	if self.stats is not None:
	self.stats["start_tracing"] += 1
	events = sys.monitoring.events
	with self.lock:
	if self.sysmon_on:
	assert sys_monitoring is not None
	local_events = events.LINE
	if self.trace_arcs:
	assert env.PYBEHAVIOR.branch_right_left
	local_events \|= (
	events.PY_RETURN \| events.BRANCH_RIGHT \| events.BRANCH_LEFT
	)
	sys_monitoring.set_local_events(self.myid, code, local_events)

	if LOG: # pragma: debugging
	if code.co_filename not in {"<string>"}:
	self.filename_code_ids[f"{code.co_filename}:{code.co_name}"].add(
	id(code)
	)

	return DISABLE

	@panopticon("code", "@", None)
	def sysmon_py_return(
	self,
	code: CodeType,
	instruction_offset: TOffset,
	retval: object,
	) -> MonitorReturn:
	"""Handle sys.monitoring.events.PY_RETURN events for branch coverage."""
	if self.stats is not None:
	self.stats["returns"] += 1
	code_info = self.code_infos.get(id(code))
	# code_info is not None and code_info.file_data is not None, since we
	# wouldn't have enabled this event if they were.
	last_line = code_info.byte_to_line.get(instruction_offset) # type: ignore
	if last_line is not None:
	arc = (last_line, -code.co_firstlineno)
	code_info.file_data.add(arc) # type: ignore
	# log(f"adding {arc=}")
	return DISABLE

	@panopticon("code", "line")
	def sysmon_line_lines(self, code: CodeType, line_number: TLineNo) -> MonitorReturn:
	"""Handle sys.monitoring.events.LINE events for line coverage."""
	if self.stats is not None:
	self.stats["line_lines"] += 1
	code_info = self.code_infos.get(id(code))
	# It should be true that code_info is not None and code_info.file_data
	# is not None, since we wouldn't have enabled this event if they were.
	# But somehow code_info can be None here, so we have to check.
	if code_info is not None and code_info.file_data is not None:
	code_info.file_data.add(line_number) # type: ignore
	# log(f"adding {line_number=}")
	return DISABLE

	@panopticon("code", "line")
	def sysmon_line_arcs(self, code: CodeType, line_number: TLineNo) -> MonitorReturn:
	"""Handle sys.monitoring.events.LINE events for branch coverage."""
	if self.stats is not None:
	self.stats["line_arcs"] += 1
	code_info = self.code_infos[id(code)]
	# code_info is not None and code_info.file_data is not None, since we
	# wouldn't have enabled this event if they were.
	arc = (line_number, line_number)
	code_info.file_data.add(arc) # type: ignore
	# log(f"adding {arc=}")
	return DISABLE

	@panopticon("code", "@", "@")
	def sysmon_branch_either(
	self, code: CodeType, instruction_offset: TOffset, destination_offset: TOffset
	) -> MonitorReturn:
	"""Handle BRANCH_RIGHT and BRANCH_LEFT events."""
	if self.stats is not None:
	self.stats["branches"] += 1
	code_info = self.code_infos[id(code)]
	# code_info is not None and code_info.file_data is not None, since we
	# wouldn't have enabled this event if they were.
	resolver = code_info.branch_resolver
	if resolver is None:
	if self.stats is not None:
	self.stats["branch_trails"] += 1
	assert code_info.byte_to_line is not None
	resolver = code_info.branch_resolver = BranchArcResolver(
	code,
	code_info.byte_to_line,
	self.get_multiline_map(code.co_filename),
	)
	arc = resolver.resolve(instruction_offset, destination_offset)
	if arc is not None:
	code_info.file_data.add(arc) # type: ignore
	# log(f"adding {arc=}")
	else:
	# This could be an exception jumping from line to line.
	assert code_info.byte_to_line is not None
	l1 = code_info.byte_to_line.get(instruction_offset)
	if l1 is not None:
	l2 = code_info.byte_to_line.get(destination_offset)
	if l2 is not None and l1 != l2:
	arc = (l1, l2)
	code_info.file_data.add(arc) # type: ignore
	# log(f"adding unforeseen {arc=}")

	return DISABLE

	def get_multiline_map(self, filename: str) -> dict[TLineNo, TLineNo]:
	"""Get the multiline map for `filename`, computing it at most once."""
	multiline_map = self.multiline_maps.get(filename)
	if multiline_map is None:
	multiline_map = self.multiline_maps[filename] = compute_multiline_map(filename)
	return multiline_map


	def compute_multiline_map(filename: str) -> dict[TLineNo, TLineNo]:
	"""Tokenize `filename` and return its multiline map."""
	try:
	text = get_python_source(filename)
	except (OSError, NoSource):
	# This can happen if open() in python.py fails.
	return {}
	try:
	return multiline_map_from_text(text)
	except (tokenize.TokenError, IndentationError, SyntaxError):
	# The file was not Python. This can happen when the code object refers
	# to an original non-Python source file, like a Jinja template.
	# In that case, just return an empty map, which might lead to slightly
	# wrong branch coverage, but we don't have any better option.
	return {}