Merge pull request #1124 from ekouts/bugfix/failed_cleanup

[bugfix] Fix infinite loop bug when cleanup phase fails

Author: Vasileios Karakasis

reframe/frontend/executors/policies.py

@@ -1,7 +1,9 @@
+import contextlib
 import itertools
 import math
 import sys
 import time
+
 from datetime import datetime
 
 from reframe.core.exceptions import (TaskDependencyError, TaskExit)
@@ -10,6 +12,16 @@
                                         TaskEventListener, ABORT_REASONS)
 
 
+def _cleanup_all(tasks, *args, **kwargs):
+    for task in tasks:
+        if task.ref_count == 0:
+            with contextlib.suppress(TaskExit):
+                task.cleanup(*args, **kwargs)
+
+    # Remove cleaned up tests
+    tasks[:] = [t for t in tasks if t.ref_count]
+
+
 class SerialExecutionPolicy(ExecutionPolicy, TaskEventListener):
     def __init__(self):
         super().__init__()
@@ -66,19 +78,6 @@ def runcase(self, case):
             raise
         except BaseException:
             task.fail(sys.exc_info())
-        finally:
-            self.printer.status('FAIL' if task.failed else 'OK',
-                                task.check.info(), just='right')
-
-    def _cleanup_all(self):
-        for task in self._retired_tasks:
-            if task.ref_count == 0:
-                task.cleanup(not self.keep_stage_files)
-
-        # Remove cleaned up tests
-        self._retired_tasks[:] = [
-            t for t in self._retired_tasks if t.ref_count
-        ]
 
     def on_task_setup(self, task):
         pass
@@ -90,18 +89,22 @@ def on_task_exit(self, task):
         pass
 
     def on_task_failure(self, task):
-        pass
+        if task.failed_stage == 'cleanup':
+            self.printer.status('ERROR', task.check.info(), just='right')
+        else:
+            self.printer.status('FAIL', task.check.info(), just='right')
 
     def on_task_success(self, task):
+        self.printer.status('OK', task.check.info(), just='right')
         # update reference count of dependencies
         for c in task.testcase.deps:
             self._task_index[c].ref_count -= 1
 
-        self._cleanup_all()
+        _cleanup_all(self._retired_tasks, not self.keep_stage_files)
 
     def exit(self):
         # Clean up all remaining tasks
-        self._cleanup_all()
+        _cleanup_all(self._retired_tasks, not self.keep_stage_files)
 
 
 class PollRateFunction:
@@ -196,8 +199,11 @@ def on_task_run(self, task):
         self._running_tasks.append(task)
 
     def on_task_failure(self, task):
-        self._remove_from_running(task)
-        self.printer.status('FAIL', task.check.info(), just='right')
+        if task.failed_stage == 'cleanup':
+            self.printer.status('ERROR', task.check.info(), just='right')
+        else:
+            self._remove_from_running(task)
+            self.printer.status('FAIL', task.check.info(), just='right')
 
     def on_task_success(self, task):
         self.printer.status('OK', task.check.info(), just='right')
@@ -290,16 +296,6 @@ def runcase(self, case):
             self._failall(e)
             raise
 
-    def _cleanup_all(self):
-        for task in self._retired_tasks:
-            if task.ref_count == 0:
-                task.cleanup(not self.keep_stage_files)
-
-        # Remove cleaned up tests
-        self._retired_tasks[:] = [
-            t for t in self._retired_tasks if t.ref_count
-        ]
-
     def _poll_tasks(self):
         '''Update the counts of running checks per partition.'''
         getlogger().debug('updating counts for running test cases')
@@ -324,10 +320,8 @@ def _finalize_all(self):
                 break
 
             getlogger().debug('finalizing task: %s' % task.check.info())
-            try:
+            with contextlib.suppress(TaskExit):
                 self._finalize_task(task)
-            except TaskExit:
-                pass
 
     def _finalize_task(self, task):
         if not self.skip_sanity_check:
@@ -395,7 +389,7 @@ def exit(self):
                 self._finalize_all()
                 self._setup_all()
                 self._reschedule_all()
-                self._cleanup_all()
+                _cleanup_all(self._retired_tasks, not self.keep_stage_files)
                 t_elapsed = (datetime.now() - t_start).total_seconds()
                 real_rate = num_polls / t_elapsed
                 getlogger().debug(

reframe/frontend/printer.py

@@ -41,23 +41,13 @@ def status(self, status, message='', just=None, level=logging.INFO):
             status_stripped = status.strip().lower()
             if status_stripped == 'skip':
                 status = color.colorize(status, color.YELLOW)
-            elif status_stripped in ['fail', 'failed']:
+            elif status_stripped in ['fail', 'failed', 'error']:
                 status = color.colorize(status, color.RED)
             else:
                 status = color.colorize(status, color.GREEN)
 
         logging.getlogger().log(level, '[ %s ] %s' % (status, message))
 
-    def result(self, check, partition, environ, success):
-        if success:
-            result_str = 'OK'
-        else:
-            result_str = 'FAIL'
-
-        self.status(
-            result_str, '%s on %s using %s' %
-            (check.name, partition.fullname, environ.name), just='right')
-
     def timestamp(self, msg='', separator=None):
         msg = '%s %s' % (msg, datetime.datetime.today().strftime('%c %Z'))
         if separator:

unittests/resources/checks/frontend_checks.py

@@ -130,6 +130,21 @@ def wait(self):
         sys.exit(1)
 
 
+@rfm.simple_test
+class CleanupFailTest(rfm.RunOnlyRegressionTest):
+    def __init__(self):
+        self.valid_systems = ['*']
+        self.valid_prog_environs = ['*']
+        self.sourcesdir = None
+        self.executable = 'echo foo'
+        self.sanity_patterns = sn.assert_found(r'foo', self.stdout)
+
+    @rfm.run_before('cleanup')
+    def fail(self):
+        # Make this test fail on purpose
+        raise Exception
+
+
 class SleepCheck(BaseFrontendCheck):
     _next_id = 0
 

unittests/test_loader.py

@@ -32,11 +32,11 @@ def test_load_file_absolute(self):
     def test_load_recursive(self):
         checks = self.loader.load_from_dir('unittests/resources/checks',
                                            recurse=True)
-        self.assertEqual(11, len(checks))
+        self.assertEqual(12, len(checks))
 
     def test_load_all(self):
         checks = self.loader_with_path.load_all()
-        self.assertEqual(10, len(checks))
+        self.assertEqual(11, len(checks))
 
     def test_load_all_with_prefix(self):
         checks = self.loader_with_prefix.load_all()

unittests/test_policies.py

@@ -81,70 +81,76 @@ def test_runall(self):
         self.runall(self.checks)
 
         stats = self.runner.stats
-        self.assertEqual(7, stats.num_cases())
+        self.assertEqual(8, stats.num_cases())
         self.assertRunall()
-        self.assertEqual(4, len(stats.failures()))
+        self.assertEqual(5, len(stats.failures()))
         self.assertEqual(2, self._num_failures_stage('setup'))
         self.assertEqual(1, self._num_failures_stage('sanity'))
         self.assertEqual(1, self._num_failures_stage('performance'))
+        self.assertEqual(1, self._num_failures_stage('cleanup'))
 
     def test_runall_skip_system_check(self):
         self.runall(self.checks, skip_system_check=True)
 
         stats = self.runner.stats
-        self.assertEqual(8, stats.num_cases())
+        self.assertEqual(9, stats.num_cases())
         self.assertRunall()
-        self.assertEqual(4, len(stats.failures()))
+        self.assertEqual(5, len(stats.failures()))
         self.assertEqual(2, self._num_failures_stage('setup'))
         self.assertEqual(1, self._num_failures_stage('sanity'))
         self.assertEqual(1, self._num_failures_stage('performance'))
+        self.assertEqual(1, self._num_failures_stage('cleanup'))
 
     def test_runall_skip_prgenv_check(self):
         self.runall(self.checks, skip_environ_check=True)
 
         stats = self.runner.stats
-        self.assertEqual(8, stats.num_cases())
+        self.assertEqual(9, stats.num_cases())
         self.assertRunall()
-        self.assertEqual(4, len(stats.failures()))
+        self.assertEqual(5, len(stats.failures()))
         self.assertEqual(2, self._num_failures_stage('setup'))
         self.assertEqual(1, self._num_failures_stage('sanity'))
         self.assertEqual(1, self._num_failures_stage('performance'))
+        self.assertEqual(1, self._num_failures_stage('cleanup'))
 
     def test_runall_skip_sanity_check(self):
         self.runner.policy.skip_sanity_check = True
         self.runall(self.checks)
 
         stats = self.runner.stats
-        self.assertEqual(7, stats.num_cases())
+        self.assertEqual(8, stats.num_cases())
         self.assertRunall()
-        self.assertEqual(3, len(stats.failures()))
+        self.assertEqual(4, len(stats.failures()))
         self.assertEqual(2, self._num_failures_stage('setup'))
         self.assertEqual(0, self._num_failures_stage('sanity'))
         self.assertEqual(1, self._num_failures_stage('performance'))
+        self.assertEqual(1, self._num_failures_stage('cleanup'))
 
     def test_runall_skip_performance_check(self):
         self.runner.policy.skip_performance_check = True
         self.runall(self.checks)
 
         stats = self.runner.stats
-        self.assertEqual(7, stats.num_cases())
+        self.assertEqual(8, stats.num_cases())
         self.assertRunall()
-        self.assertEqual(3, len(stats.failures()))
+        self.assertEqual(4, len(stats.failures()))
         self.assertEqual(2, self._num_failures_stage('setup'))
         self.assertEqual(1, self._num_failures_stage('sanity'))
         self.assertEqual(0, self._num_failures_stage('performance'))
+        self.assertEqual(1, self._num_failures_stage('cleanup'))
 
     def test_strict_performance_check(self):
         self.runner.policy.strict_check = True
         self.runall(self.checks)
 
         stats = self.runner.stats
-        self.assertEqual(7, stats.num_cases())
+        self.assertEqual(8, stats.num_cases())
         self.assertRunall()
-        self.assertEqual(5, len(stats.failures()))
+        self.assertEqual(6, len(stats.failures()))
         self.assertEqual(2, self._num_failures_stage('setup'))
         self.assertEqual(1, self._num_failures_stage('sanity'))
         self.assertEqual(2, self._num_failures_stage('performance'))
+        self.assertEqual(1, self._num_failures_stage('cleanup'))
 
     def test_force_local_execution(self):
         self.runner.policy.force_local = True