Reduce the coordinator's per-mutant overhead

HISTORY.rst

@@ -4,6 +4,8 @@ Changelog
 Unreleased
 ~~~~~~~~~~
 
+* Performance: pre-compile the mutant tree after generation, remove the quadratic recount behind the status line, and batch the ``.meta`` writes
+
 * Fix ``# pragma: no mutate block`` being silently ignored when placed on an ``else``, ``except``, ``except*`` or ``finally`` header, and on the ``try`` line of a ``try``/``except*``
 
 * Fix mutants being reported as survived when a test uses ``patch.dict(os.environ, ..., clear=True)`` (`#511`)

src/mutmut/__main__.py

@@ -17,6 +17,7 @@
     )
     sys.exit(1)
 import ast
+import compileall
 import fnmatch
 import gc
 import hashlib
@@ -39,6 +40,7 @@
 from multiprocessing import set_start_method
 from os import makedirs
 from pathlib import Path
+from time import monotonic
 from time import process_time
 from types import TracebackType
 
@@ -63,6 +65,7 @@
 from mutmut.runners.harness import PytestRunner
 from mutmut.runners.harness import TestRunner
 from mutmut.runners.harness import collected_test_names
+from mutmut.stats import ProgressCounter
 from mutmut.stats import calculate_summary_stats
 from mutmut.stats import emoji_by_status
 from mutmut.stats import load_stats
@@ -161,6 +164,17 @@ def create_mutants(max_children: int) -> MutantGenerationStats:
     return stats
 
 
+def precompile_mutants(max_children: int) -> None:
+    """Byte-compile the mutated tree with a worker pool, so the single-process stats run does
+    not compile it serially on first import. Up-to-date .pyc files are skipped."""
+    for path in config().source_paths:
+        mutated = Path("mutants") / path
+        if mutated.is_dir():
+            compileall.compile_dir(mutated, quiet=1, workers=max_children)
+        elif mutated.is_file():
+            compileall.compile_file(mutated, quiet=1)
+
+
 def create_file_mutants(path: Path) -> FileMutationResult:
     try:
         print(path)
@@ -991,6 +1005,7 @@ def _run(mutant_names: tuple[str, ...] | list[str], max_children: int | None) ->
         setup_source_paths()
         store_lines_covered_by_tests()
         stats = create_mutants(max_children)
+        precompile_mutants(max_children)
 
     time = datetime.now() - start
     print(
@@ -1035,12 +1050,27 @@ def _run(mutant_names: tuple[str, ...] | list[str], max_children: int | None) ->
 
     runner.prepare_main_test_run()
 
+    progress = ProgressCounter(source_file_mutation_data_by_path)
+    last_flush = monotonic()
+
+    def flush_results(*, force: bool, interval_s: float = 2.0) -> None:
+        """Write the meta files with unsaved results, at most every ``interval_s`` seconds unless forced."""
+        nonlocal last_flush
+        if force or monotonic() - last_flush >= interval_s:
+            for m in source_file_mutation_data_by_path.values():
+                m.save_if_dirty()
+            last_flush = monotonic()
+
     def read_one_child_exit_status() -> None:
         pid, wait_status = os.wait()
         exit_code = os.waitstatus_to_exitcode(wait_status)
         if config().debug:
             print("    worker exit code", exit_code)
-        source_file_mutation_data_by_pid[pid].register_result(pid=pid, exit_code=exit_code)
+        data = source_file_mutation_data_by_pid.pop(pid)
+        old_exit_code = data.exit_code_by_key[data.key_by_pid[pid]]
+        data.register_result(pid=pid, exit_code=exit_code)
+        progress.record(old_exit_code, exit_code)
+        flush_results(force=False)
 
     source_file_mutation_data_by_pid: dict[int, SourceFileMutationData] = {}  # many pids map to one MutationData
     running_children = 0
@@ -1059,22 +1089,22 @@ def read_one_child_exit_status() -> None:
             tests = state().tests_by_mangled_function_name.get(mangled_name_from_mutant_name(mutant_name), set())
             estimated_time_of_tests = sum(state().duration_by_test[test_name] for test_name in tests)
             mutation_data.estimated_time_of_tests_by_mutant[mutant_name] = estimated_time_of_tests
-            print_stats(source_file_mutation_data_by_path)
+            progress.print()
 
             # Rerun mutant if it's explicitly mentioned, but otherwise let the result stand
             if not mutant_names and result is not None:
                 continue
 
             if not tests:
-                mutation_data.exit_code_by_key[mutant_name] = 33
-                mutation_data.save()
+                mutation_data.set_result(mutant_name, 33)
+                progress.record(result, 33)
                 continue
 
             failed_type_check_mutant = mutants_caught_by_type_checker.get(mutant_name)
             if failed_type_check_mutant:
-                mutation_data.exit_code_by_key[mutant_name] = 37
+                mutation_data.set_result(mutant_name, 37)
                 mutation_data.type_check_error_by_key[mutant_name] = failed_type_check_mutant.error.error_description
-                mutation_data.save()
+                progress.record(result, 37)
                 continue
 
             cfg = config()
@@ -1126,6 +1156,7 @@ def read_one_child_exit_status() -> None:
         stop_all_children(mutants)
     finally:
         gc.unfreeze()
+        flush_results(force=True)
 
     elapsed_time = datetime.now() - start
 

src/mutmut/mutation/data.py

@@ -108,6 +108,7 @@ def __init__(self, *, path: Path | str) -> None:
         self.durations_by_key: dict[str, float] = {}
         self.start_time_by_pid: dict[int, datetime] = {}
         self.type_check_error_by_key: dict[str, str | None] = {}
+        self.dirty = False  # unsaved results; see save_if_dirty()
 
     def load(self) -> None:
         try:
@@ -127,21 +128,33 @@ def register_pid(self, *, pid: int, key: str) -> None:
         self.key_by_pid[pid] = key
         self.start_time_by_pid[pid] = datetime.now()
 
-    def register_result(self, *, pid: int, exit_code: int) -> None:
+    def register_result(self, *, pid: int, exit_code: int) -> str:
+        """Record the verdict of child ``pid`` and return the mutant's key.
+
+        The meta file is not written here (that would be once per mutant); the caller flushes
+        with ``save_if_dirty``."""
         assert self.key_by_pid[pid] in self.exit_code_by_key
         key = self.key_by_pid[pid]
-        self.exit_code_by_key[key] = exit_code
+        self.set_result(key, exit_code)
         self.durations_by_key[key] = (datetime.now() - self.start_time_by_pid[pid]).total_seconds()
-        # TODO: maybe rate limit this? Saving on each result can slow down mutation testing a lot if the test run is fast.
         del self.key_by_pid[pid]
         del self.start_time_by_pid[pid]
-        self.save()
+        return key
+
+    def set_result(self, key: str, exit_code: int) -> None:
+        self.exit_code_by_key[key] = exit_code
+        self.dirty = True
+
+    def save_if_dirty(self) -> None:
+        if self.dirty:
+            self.save()
 
     def stop_children(self) -> None:
         for pid in self.key_by_pid.keys():
             os.kill(pid, signal.SIGTERM)
 
     def save(self) -> None:
+        self.dirty = False
         with open(self.meta_path, "w") as f:
             json.dump(
                 {

src/mutmut/stats.py

@@ -3,6 +3,8 @@
 import json
 from collections import defaultdict
 from dataclasses import dataclass
+from datetime import datetime
+from datetime import timedelta
 from json import JSONDecodeError
 
 from mutmut.configuration import config
@@ -93,14 +95,45 @@ def calculate_summary_stats(source_file_mutation_data_by_path: dict[str, SourceF
     )
 
 
+def format_stats(s: Stat) -> str:
+    return f"{(s.total - s.not_checked)}/{s.total}  🎉 {s.killed} 🫥 {s.no_tests}  ⏰ {s.timeout}  🤔 {s.suspicious}  🙁 {s.survived}  🔇 {s.skipped}  🧙 {s.caught_by_type_check}"
+
+
 def print_stats(
     source_file_mutation_data_by_path: dict[str, SourceFileMutationData], force_output: bool = False
 ) -> None:
-    s = calculate_summary_stats(source_file_mutation_data_by_path)
-    print_status(
-        f"{(s.total - s.not_checked)}/{s.total}  🎉 {s.killed} 🫥 {s.no_tests}  ⏰ {s.timeout}  🤔 {s.suspicious}  🙁 {s.survived}  🔇 {s.skipped}  🧙 {s.caught_by_type_check}",
-        force_output=force_output,
-    )
+    print_status(format_stats(calculate_summary_stats(source_file_mutation_data_by_path)), force_output=force_output)
+
+
+def _stat_field(exit_code: int | None) -> str:
+    """The ``Stat`` field that counts mutants with this exit code."""
+    return status_by_exit_code[exit_code].replace(" ", "_")
+
+
+class ProgressCounter:
+    """Running totals for the status line during mutation testing: recounting every mutant
+    per update is quadratic over a run, so the totals are computed once and adjusted per verdict."""
+
+    def __init__(self, source_file_mutation_data_by_path: dict[str, SourceFileMutationData]) -> None:
+        self.stat = calculate_summary_stats(source_file_mutation_data_by_path)
+        self._last_print = datetime(1900, 1, 1)
+
+    def record(self, old_exit_code: int | None, new_exit_code: int | None) -> None:
+        """Move one mutant from the status of ``old_exit_code`` to that of ``new_exit_code``."""
+        old_field = _stat_field(old_exit_code)
+        new_field = _stat_field(new_exit_code)
+        if old_field == new_field:
+            return
+        setattr(self.stat, old_field, getattr(self.stat, old_field) - 1)
+        setattr(self.stat, new_field, getattr(self.stat, new_field) + 1)
+
+    def print(self, force_output: bool = False) -> None:
+        """Print the status line, at most every 100ms unless forced."""
+        now = datetime.now()
+        if not force_output and now - self._last_print < timedelta(seconds=0.1):
+            return
+        self._last_print = now
+        print_status(format_stats(self.stat), force_output=True)
 
 
 def load_stats() -> bool:

tests/mutation/test_mutation_data.py

@@ -0,0 +1,57 @@
+"""Results are written to the meta file in batches, not once per mutant."""
+
+import json
+from pathlib import Path
+
+from mutmut.mutation.data import SourceFileMutationData
+
+
+def _data(tmp_path: Path, monkeypatch) -> SourceFileMutationData:
+    monkeypatch.chdir(tmp_path)
+    (tmp_path / "mutants" / "src").mkdir(parents=True)
+    data = SourceFileMutationData(path=Path("src/module.py"))
+    data.exit_code_by_key = {"module.x_f__mutmut_1": None, "module.x_f__mutmut_2": None}
+    return data
+
+
+def _saved_results(data: SourceFileMutationData) -> dict[str, int | None]:
+    with open(data.meta_path) as f:
+        return json.load(f)["exit_code_by_key"]
+
+
+def test_registering_a_result_marks_the_file_dirty_without_writing_it(tmp_path, monkeypatch):
+    data = _data(tmp_path, monkeypatch)
+    data.register_pid(pid=1234, key="module.x_f__mutmut_1")
+
+    assert data.register_result(pid=1234, exit_code=1) == "module.x_f__mutmut_1"
+
+    assert data.exit_code_by_key["module.x_f__mutmut_1"] == 1
+    assert data.durations_by_key["module.x_f__mutmut_1"] >= 0
+    assert 1234 not in data.key_by_pid
+    assert data.dirty
+    assert not data.meta_path.exists()
+
+
+def test_save_if_dirty_writes_once_and_clears_the_flag(tmp_path, monkeypatch):
+    data = _data(tmp_path, monkeypatch)
+    data.save_if_dirty()
+    assert not data.meta_path.exists()
+
+    data.set_result("module.x_f__mutmut_2", 33)
+    data.save_if_dirty()
+    assert _saved_results(data) == {"module.x_f__mutmut_1": None, "module.x_f__mutmut_2": 33}
+    assert not data.dirty
+
+    data.meta_path.unlink()
+    data.save_if_dirty()
+    assert not data.meta_path.exists()
+
+
+def test_save_always_writes_and_clears_the_flag(tmp_path, monkeypatch):
+    data = _data(tmp_path, monkeypatch)
+    data.set_result("module.x_f__mutmut_1", 0)
+
+    data.save()
+
+    assert not data.dirty
+    assert _saved_results(data) == {"module.x_f__mutmut_1": 0, "module.x_f__mutmut_2": None}

tests/test_precompile.py

@@ -0,0 +1,25 @@
+from pathlib import Path
+from unittest.mock import Mock
+
+import mutmut.__main__
+from mutmut.__main__ import precompile_mutants
+from mutmut.configuration import Config
+
+
+def test_precompile_compiles_every_mutated_source_path(tmp_path, monkeypatch):
+    monkeypatch.chdir(tmp_path)
+    package = tmp_path / "mutants" / "src" / "pkg"
+    package.mkdir(parents=True)
+    (package / "mod.py").write_text("def f():\n    return 1\n")
+    (tmp_path / "mutants" / "single.py").write_text("x = 1\n")
+    (tmp_path / "mutants" / "untouched.py").write_text("y = 1\n")
+
+    cfg = Mock(spec=Config)
+    cfg.source_paths = [Path("src"), Path("single.py"), Path("absent.py")]
+    monkeypatch.setattr(mutmut.__main__, "config", lambda: cfg)
+
+    precompile_mutants(max_children=2)
+
+    assert list((package / "__pycache__").glob("mod.*.pyc"))
+    assert list((tmp_path / "mutants" / "__pycache__").glob("single.*.pyc"))
+    assert not list((tmp_path / "mutants" / "__pycache__").glob("untouched.*.pyc"))

tests/test_progress_counter.py

@@ -0,0 +1,38 @@
+from mutmut.mutation.data import SourceFileMutationData
+from mutmut.stats import ProgressCounter
+from mutmut.stats import calculate_summary_stats
+
+
+def _data(path: str, results: dict[str, int | None]) -> SourceFileMutationData:
+    data = SourceFileMutationData(path=path)
+    data.exit_code_by_key = dict(results)
+    return data
+
+
+def test_counter_matches_a_full_recount_after_verdicts_change():
+    by_path = {
+        "a.py": _data("a.py", {"a.x_f__mutmut_1": None, "a.x_f__mutmut_2": None, "a.x_f__mutmut_3": 1}),
+        "b.py": _data("b.py", {"b.x_g__mutmut_1": 0}),
+    }
+    counter = ProgressCounter(by_path)
+    assert counter.stat == calculate_summary_stats(by_path)
+
+    changes = [("a.py", "a.x_f__mutmut_1", 1), ("a.py", "a.x_f__mutmut_2", 36), ("b.py", "b.x_g__mutmut_1", 37)]
+    for path, key, exit_code in changes:
+        old = by_path[path].exit_code_by_key[key]
+        by_path[path].set_result(key, exit_code)
+        counter.record(old, exit_code)
+
+    assert counter.stat == calculate_summary_stats(by_path)
+    assert counter.stat.not_checked == 0
+    assert counter.stat.killed == 2
+    assert counter.stat.timeout == 1
+    assert counter.stat.caught_by_type_check == 1
+    assert counter.stat.total == 4
+
+
+def test_recording_the_same_status_twice_is_a_no_op():
+    by_path = {"a.py": _data("a.py", {"a.x_f__mutmut_1": 1})}
+    counter = ProgressCounter(by_path)
+    counter.record(1, 3)  # both "killed"
+    assert counter.stat == calculate_summary_stats(by_path)