Method _get_result

Lib/test/libregrtest/run_workers.py:528–554

(self)

            worker.wait_stopped(start_time)

    def _get_result(self) -> QueueOutput | None:
        pgo = self.runtests.pgo
        use_faulthandler = (self.timeout is not None)

        # bpo-46205: check the status of workers every iteration to avoid
        # waiting forever on an empty queue.
        while self.live_worker_count > 0:
            if use_faulthandler:
                faulthandler.dump_traceback_later(MAIN_PROCESS_TIMEOUT,
                                                  exit=True)

            # wait for a thread
            try:
                result = self.output.get(timeout=PROGRESS_UPDATE)
                if isinstance(result, WorkerThreadExited):
                    self.live_worker_count -= 1
                    continue
                return result
            except queue.Empty:
                pass

            if not pgo:
                # display progress
                running = get_running(self.workers)
                if running:
                    self.log(running)
        return None

    def display_result(self, mp_result: MultiprocessResult) -> None:
        result = mp_result.result