24b605e8ec48b9b67e507ca959eaed4ff4afc6a6,reframe/frontend/executors/policies.py,AsynchronousExecutionPolicy,run_check,#AsynchronousExecutionPolicy#Any#Any#Any#,205

Before Change

        self.printer.status(status, msg)

    def run_check(self, check, partition, environ):
        try:
            executor = RegressionTestExecutor(check, self.strict_check)
            testcase = TestCase(executor)

            executor.setup(
                partition=partition,
                environ=environ,
                sched_account=self.sched_account,
                sched_partition=self.sched_partition,
                sched_reservation=self.sched_reservation,
                sched_nodelist=self.sched_nodelist,
                sched_exclude_nodelist=self.sched_exclude_nodelist,
                sched_options=self.sched_options
            )

            ready_testcase = RunningTestCase(testcase, EnvironmentSnapshot())
            partname = partition.fullname
            if self._running_cases_counts[partname] >= partition.max_jobs:
                // Make sure that we still exceeded the job limit
                getlogger().debug("reached job limit (%s) for partition %s" %
                                  (partition.max_jobs, partname))
                self._update_running_counts()

            if self._running_cases_counts[partname] < partition.max_jobs:
                // Test"s environment is already loaded; no need to be reloaded
                self._reschedule(ready_testcase, load_env=False)
            else:
                self._print_executor_status("HOLD", executor)
                self._ready_cases[partname].append(ready_testcase)

        except (KeyboardInterrupt, ReframeFatalError, AssertionError):
            if not testcase.failed():
                // test case failed during setup
                testcase.fail(sys.exc_info())
            self._failall()
            raise
        except BaseException:
            // Here we are sure that test case has failed during setup, since
            // _compile_and_run() handles already non-fatal exceptions. Though
            // we check again the testcase, just in case.
            if not testcase.failed():
                testcase.fail(sys.exc_info())
        finally:
            if testcase.valid() and testcase.failed_stage == "setup":
                // We need to print the result here only if the setup stage has
                // finished, since otherwise _compile_and_run() prints it
                self.printer.result(executor.check, partition, environ,
                                    not testcase.failed())

            self._test_cases.append(testcase)
            self.environ_snapshot.load()

    def _update_running_counts(self):
        Update the counts of running checks per partition.
        getlogger().debug("updating counts for running test cases")
        freed_slots = {}

After Change

        super().run_check(check, partition, environ)
        task = RegressionTask(check, self.task_listeners)
        self._tasks.append(task)
        try:
            task.setup(partition, environ,
                       sched_account=self.sched_account,
                       sched_partition=self.sched_partition,
                       sched_reservation=self.sched_reservation,
                       sched_nodelist=self.sched_nodelist,
                       sched_exclude_nodelist=self.sched_exclude_nodelist,
                       sched_options=self.sched_options)

            self.printer.status("RUN", task.check.info())
            partname = partition.fullname
            if self._running_tasks_counts[partname] >= partition.max_jobs:
                // Make sure that we still exceeded the job limit
                getlogger().debug("reached job limit (%s) for partition %s" %
                                  (partition.max_jobs, partname))
                self._poll_tasks()

            if self._running_tasks_counts[partname] < partition.max_jobs:
                // Test"s environment is already loaded; no need to be reloaded
                self._reschedule(task, load_env=False)
            else:
                self.printer.status("HOLD", task.check.info(), just="right")
                self._ready_tasks[partname].append(task)
        except TaskExit:
            return
        except ABORT_REASONS as e:
            if not task.failed:
                // Abort was caused due to failure elsewhere, abort current
                // task as well
                task.abort(e)

            self._failall(e)
            raise

    def _poll_tasks(self):
        Update the counts of running checks per partition.
        getlogger().debug("updating counts for running test cases")
        getlogger().debug("polling %s task(s)" % len(self._running_tasks))

Instances