Method run

Lib/test/libregrtest/run_workers.py:601–627

(self)

        return result

    def run(self) -> None:
        fail_fast = self.runtests.fail_fast
        fail_env_changed = self.runtests.fail_env_changed

        self.start_workers()

        self.test_index = 0
        try:
            while True:
                item = self._get_result()
                if item is None:
                    break

                result = self._process_result(item)
                if result.must_stop(fail_fast, fail_env_changed):
                    break
        except KeyboardInterrupt:
            print()
            self.results.interrupted = True
        finally:
            if self.timeout is not None:
                faulthandler.cancel_dump_traceback_later()

            # Always ensure that all worker processes are no longer
            # worker when we exit this function
            self.pending.stop()
            self.stop_workers()