vibefpmgui.parallel_tests

  1"""Running tests in parallel (Build options → Parallel tests).
  2
  3fpm runs tests one after another, so for a parallel run fpm only builds
  4them: ``fpm test --runner "test_runner.py --collect"`` prints each test's
  5command line instead of running it (see `vibefpmgui.test_runner`). Then
  6`ParallelTests` runs those commands itself, a few at a time, each in its own
  7process with its own output, so the outputs never mix.
  8"""
  9
 10import codecs
 11import time
 12from collections.abc import Callable
 13
 14from qtpy.QtCore import QObject, QProcess, QProcessEnvironment, Signal
 15
 16
 17class ParallelTests(QObject):
 18    """Run ``(name, argv)`` commands, at most ``workers`` at a time. Signals
 19    say when each starts, what it prints and how it ended; `done` comes once
 20    all have ended, or once the ones running when `stop` was called have."""
 21
 22    started = Signal(str)                    # name
 23    output = Signal(str, str)                # name, text
 24    finished = Signal(str, int, float, bool)  # name, exit code, seconds, stopped
 25    done = Signal(bool)                      # stopped
 26
 27    def __init__(self, commands: list[tuple[str, list[str]]], workers: int, cwd: str,
 28                 env: QProcessEnvironment, inner: list[str], kill: Callable[[QProcess], None],
 29                 parent=None):
 30        """``inner``: a runner put in front of each command (fpm's --runner,
 31        e.g. ``mpirun -np 2``); ``kill`` stops a process and what it started."""
 32        super().__init__(parent)
 33        self.queue = list(commands)
 34        self.workers = max(1, workers)
 35        self.cwd, self.env, self.inner, self.kill = cwd, env, inner, kill
 36        self.running: dict[str, QProcess] = {}
 37        self.stopping = False
 38        self._started_at: dict[str, float] = {}
 39        self._decoders: dict[str, codecs.IncrementalDecoder] = {}
 40
 41    def start(self):
 42        self._fill()
 43
 44    def stop(self):
 45        """Run no more, and kill the ones running (`done` follows)."""
 46        self.stopping = True
 47        self.queue.clear()
 48        for process in list(self.running.values()):
 49            self.kill(process)
 50        if not self.running:
 51            self.done.emit(True)
 52
 53    def wait(self, msecs: int = 3000):
 54        for process in list(self.running.values()):
 55            process.waitForFinished(msecs)
 56
 57    def command(self, argv: list[str]) -> list[str]:
 58        return self.inner + argv
 59
 60    def _fill(self):
 61        while self.queue and len(self.running) < self.workers and not self.stopping:
 62            name, argv = self.queue.pop(0)
 63            self._launch(name, self.command(argv))
 64        if not self.running and not self.queue:
 65            self.done.emit(self.stopping)
 66
 67    def _launch(self, name: str, command: list[str]):
 68        process = QProcess(self)
 69        process.setWorkingDirectory(self.cwd)
 70        process.setProcessEnvironment(self.env)
 71        process.setProcessChannelMode(QProcess.MergedChannels)
 72        # (Lambdas are fine here: a QProcess's signals come in the GUI thread.)
 73        process.readyReadStandardOutput.connect(lambda: self._read(name, process))
 74        process.finished.connect(lambda code, status: self._ended(name, process, code, status))
 75        process.errorOccurred.connect(lambda error: self._error(name, process, error, command[0]))
 76        self.running[name] = process
 77        self._started_at[name] = time.monotonic()
 78        self._decoders[name] = codecs.getincrementaldecoder("utf-8")(errors="replace")
 79        self.started.emit(name)
 80        process.start(command[0], command[1:])
 81
 82    def _read(self, name: str, process: QProcess, final: bool = False):
 83        text = self._decoders[name].decode(bytes(process.readAllStandardOutput()), final=final)
 84        if text:
 85            text = text.replace("\r\n", "\n")
 86            # (Progress lines redrawn with \r: only their last state.)
 87            text = "\n".join(line.rsplit("\r", 1)[-1] for line in text.split("\n"))
 88            self.output.emit(name, text)
 89
 90    def _ended(self, name: str, process: QProcess, code: int, status):
 91        if name not in self.running:
 92            return
 93        self._read(name, process, final=True)
 94        stopped = self.stopping
 95        if status == QProcess.CrashExit and not stopped:
 96            code = code or -1  # (a crash: the exit code isn't the program's)
 97        self._finish(name, process, code, stopped)
 98
 99    def _error(self, name: str, process: QProcess, error, program: str):
100        if error == QProcess.FailedToStart and name in self.running:
101            self.output.emit(name, f"could not run {program}: {process.errorString()}\n")
102            self._finish(name, process, 127, False)
103
104    def _finish(self, name: str, process: QProcess, code: int, stopped: bool):
105        del self.running[name]
106        seconds = time.monotonic() - self._started_at.pop(name, time.monotonic())
107        self._decoders.pop(name, None)
108        process.deleteLater()
109        self.finished.emit(name, code, seconds, stopped)
110        self._fill()