Merge pull request #1109 from ekouts/bugfix/inf_loop_async

[bugfix] Fix dependency bug in async policy

Author: Vasileios Karakasis

reframe/frontend/executors/policies.py

@@ -168,7 +168,9 @@ def __init__(self):
         self.task_listeners.append(self)
 
     def _remove_from_running(self, task):
-        getlogger().debug('removing task: %s' % task.check.info())
+        getlogger().debug(
+            'removing task from running list: %s' % task.check.info()
+        )
         try:
             self._running_tasks.remove(task)
         except ValueError:
@@ -210,6 +212,30 @@ def on_task_exit(self, task):
         self._remove_from_running(task)
         self._completed_tasks.append(task)
 
+    def _setup_task(self, task):
+        if self.deps_succeeded(task):
+            try:
+                task.setup(task.testcase.partition,
+                           task.testcase.environ,
+                           sched_flex_alloc_nodes=self.sched_flex_alloc_nodes,
+                           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)
+            except TaskExit:
+                return False
+            else:
+                return True
+        elif self.deps_failed(task):
+            exc = TaskDependencyError('dependencies failed')
+            task.fail((type(exc), exc, None))
+            return False
+        else:
+            # Not all dependencies have finished yet
+            return False
+
     def runcase(self, case):
         super().runcase(case)
         check, partition, environ = case
@@ -228,29 +254,17 @@ def runcase(self, case):
         )
         try:
             partname = partition.fullname
-            if self.deps_failed(task):
-                exc = TaskDependencyError('dependencies failed')
-                task.fail((type(exc), exc, None))
-                return
+            if not self._setup_task(task):
+                if not task.failed:
+                    self.printer.status(
+                        'DEP', '%s on %s using %s' %
+                        (check.name, partname, environ.name),
+                        just='right'
+                    )
+                    self._waiting_tasks.append(task)
 
-            if not self.deps_succeeded(task):
-                self.printer.status(
-                    'DEP', '%s on %s using %s' %
-                    (check.name, partname, environ.name),
-                    just='right'
-                )
-                self._waiting_tasks.append(task)
                 return
 
-            task.setup(partition, environ,
-                       sched_flex_alloc_nodes=self.sched_flex_alloc_nodes,
-                       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)
-
             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' %
@@ -296,20 +310,7 @@ def _poll_tasks(self):
     def _setup_all(self):
         still_waiting = []
         for task in self._waiting_tasks:
-            if self.deps_failed(task):
-                exc = TaskDependencyError('dependencies failed')
-                task.fail((type(exc), exc, None))
-            elif self.deps_succeeded(task):
-                task.setup(task.testcase.partition,
-                           task.testcase.environ,
-                           sched_flex_alloc_nodes=self.sched_flex_alloc_nodes,
-                           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)
-            else:
+            if not self._setup_task(task) and not task.failed:
                 still_waiting.append(task)
 
         self._waiting_tasks[:] = still_waiting

unittests/resources/checks_unlisted/deps_complex.py

@@ -11,7 +11,7 @@
 #       |
 #   +-->t4<--+
 #   |        |
-#   t5<------t1
+#   t5<------t1<--t8<--t9
 #   ^        ^
 #   |        |
 #   +---t6---+
@@ -150,3 +150,37 @@ def __init__(self):
     def prepend_output(self, T2):
         with open(os.path.join(T2().stagedir, 'out.txt')) as fp:
             self._count += int(fp.read())
+
+
+@rfm.simple_test
+class T8(BaseTest):
+    def __init__(self):
+        super().__init__()
+        self.depends_on('T1')
+        self.sanity_patterns = sn.assert_eq(self.count, 22)
+
+    @rfm.require_deps
+    def prepend_output(self, T1):
+        with open(os.path.join(T1().stagedir, 'out.txt')) as fp:
+            self._count += int(fp.read())
+
+    @rfm.run_after('setup')
+    def fail(self):
+        # Make this test fail on purpose
+        raise Exception
+
+
+@rfm.simple_test
+class T9(BaseTest):
+    # This tests fails because of T8. It is added to make sure that
+    # all tests are accounted for in the summary.
+
+    def __init__(self):
+        super().__init__()
+        self.depends_on('T8')
+        self.sanity_patterns = sn.assert_eq(self.count, 31)
+
+    @rfm.require_deps
+    def prepend_output(self, T8):
+        with open(os.path.join(T8().stagedir, 'out.txt')) as fp:
+            self._count += int(fp.read())

unittests/test_policies.py

@@ -58,6 +58,11 @@ def runall(self, checks, sort=False, *args, **kwargs):
 
         self.runner.runall(cases)
 
+    def assertRunall(self):
+        # Make sure that all cases finished or failed
+        for t in self.runner.stats.tasks():
+            assert t.succeeded or t.failed
+
     def _num_failures_stage(self, stage):
         stats = self.runner.stats
         return len([t for t in stats.failures() if t.failed_stage == stage])
@@ -77,6 +82,7 @@ def test_runall(self):
 
         stats = self.runner.stats
         self.assertEqual(7, stats.num_cases())
+        self.assertRunall()
         self.assertEqual(4, len(stats.failures()))
         self.assertEqual(2, self._num_failures_stage('setup'))
         self.assertEqual(1, self._num_failures_stage('sanity'))
@@ -87,6 +93,7 @@ def test_runall_skip_system_check(self):
 
         stats = self.runner.stats
         self.assertEqual(8, stats.num_cases())
+        self.assertRunall()
         self.assertEqual(4, len(stats.failures()))
         self.assertEqual(2, self._num_failures_stage('setup'))
         self.assertEqual(1, self._num_failures_stage('sanity'))
@@ -97,6 +104,7 @@ def test_runall_skip_prgenv_check(self):
 
         stats = self.runner.stats
         self.assertEqual(8, stats.num_cases())
+        self.assertRunall()
         self.assertEqual(4, len(stats.failures()))
         self.assertEqual(2, self._num_failures_stage('setup'))
         self.assertEqual(1, self._num_failures_stage('sanity'))
@@ -108,6 +116,7 @@ def test_runall_skip_sanity_check(self):
 
         stats = self.runner.stats
         self.assertEqual(7, stats.num_cases())
+        self.assertRunall()
         self.assertEqual(3, len(stats.failures()))
         self.assertEqual(2, self._num_failures_stage('setup'))
         self.assertEqual(0, self._num_failures_stage('sanity'))
@@ -119,6 +128,7 @@ def test_runall_skip_performance_check(self):
 
         stats = self.runner.stats
         self.assertEqual(7, stats.num_cases())
+        self.assertRunall()
         self.assertEqual(3, len(stats.failures()))
         self.assertEqual(2, self._num_failures_stage('setup'))
         self.assertEqual(1, self._num_failures_stage('sanity'))
@@ -130,6 +140,7 @@ def test_strict_performance_check(self):
 
         stats = self.runner.stats
         self.assertEqual(7, stats.num_cases())
+        self.assertRunall()
         self.assertEqual(5, len(stats.failures()))
         self.assertEqual(2, self._num_failures_stage('setup'))
         self.assertEqual(1, self._num_failures_stage('sanity'))
@@ -138,6 +149,7 @@ def test_strict_performance_check(self):
     def test_force_local_execution(self):
         self.runner.policy.force_local = True
         self.runall([HelloTest()])
+        self.assertRunall()
         stats = self.runner.stats
         for t in stats.tasks():
             self.assertTrue(t.check.local)
@@ -165,6 +177,7 @@ def test_retries_bad_check(self):
 
         # Ensure that the test was retried #max_retries times and failed.
         self.assertEqual(2, self.runner.stats.num_cases())
+        self.assertRunall()
         self.assertEqual(max_retries, rt.runtime().current_run)
         self.assertEqual(2, len(self.runner.stats.failures()))
 
@@ -179,6 +192,7 @@ def test_retries_good_check(self):
 
         # Ensure that the test passed without retries.
         self.assertEqual(1, self.runner.stats.num_cases())
+        self.assertRunall()
         self.assertEqual(0, rt.runtime().current_run)
         self.assertEqual(0, len(self.runner.stats.failures()))
 
@@ -196,6 +210,7 @@ def test_pass_in_retries(self):
 
         # Ensure that the test passed after retries in run #run_to_pass.
         self.assertEqual(1, self.runner.stats.num_cases())
+        self.assertRunall()
         self.assertEqual(1, len(self.runner.stats.failures(run=0)))
         self.assertEqual(run_to_pass, rt.runtime().current_run)
         self.assertEqual(0, len(self.runner.stats.failures()))
@@ -210,13 +225,14 @@ def test_dependencies(self):
         self.checks = self.loader.load_all()
         self.runall(self.checks, sort=True)
 
+        self.assertRunall()
         stats = self.runner.stats
-        assert stats.num_cases(0) == 8
-        assert len(stats.failures()) == 2
+        assert stats.num_cases(0) == 10
+        assert len(stats.failures()) == 4
         for tf in stats.failures():
             check = tf.testcase.check
-            exc_type, exc_value, _ = tf.exc_info
-            if check.name == 'T7':
+            _, exc_value, _ = tf.exc_info
+            if check.name == 'T7' or check.name == 'T9':
                 assert isinstance(exc_value, TaskDependencyError)
 
         # Check that cleanup is executed properly for successful tests as well
@@ -309,6 +325,7 @@ def test_concurrency_unlimited(self):
 
         # Ensure that all tests were run and without failures.
         self.assertEqual(len(checks), self.runner.stats.num_cases())
+        self.assertRunall()
         self.assertEqual(0, len(self.runner.stats.failures()))
 
         # Ensure that maximum concurrency was reached as fast as possible
@@ -334,6 +351,7 @@ def test_concurrency_limited(self):
 
         # Ensure that all tests were run and without failures.
         self.assertEqual(len(checks), self.runner.stats.num_cases())
+        self.assertRunall()
         self.assertEqual(0, len(self.runner.stats.failures()))
 
         # Ensure that maximum concurrency was reached as fast as possible
@@ -372,6 +390,7 @@ def test_concurrency_none(self):
 
         # Ensure that all tests were run and without failures.
         self.assertEqual(len(checks), self.runner.stats.num_cases())
+        self.assertRunall()
         self.assertEqual(0, len(self.runner.stats.failures()))
 
         # Ensure that a single task was running all the time
@@ -391,6 +410,7 @@ def _run_checks(self, checks, max_jobs):
         self.assertRaises(KeyboardInterrupt, self.runall, checks)
 
         self.assertEqual(4, self.runner.stats.num_cases())
+        self.assertRunall()
         self.assertEqual(4, len(self.runner.stats.failures()))
         self.assert_all_dead()
 
@@ -427,6 +447,7 @@ def test_poll_fails_main_loop(self):
         self.runall(checks)
         stats = self.runner.stats
         self.assertEqual(num_tasks, stats.num_cases())
+        self.assertRunall()
         self.assertEqual(num_tasks, len(stats.failures()))
 
     def test_poll_fails_busy_loop(self):
@@ -437,6 +458,7 @@ def test_poll_fails_busy_loop(self):
         self.runall(checks)
         stats = self.runner.stats
         self.assertEqual(num_tasks, stats.num_cases())
+        self.assertRunall()
         self.assertEqual(num_tasks, len(stats.failures()))