[wip] [feat] Add support for running interdependent tests with the serial execution policy

reframe/frontend/cli.py

@@ -12,6 +12,7 @@
 import reframe.core.runtime as runtime
 import reframe.frontend.argparse as argparse
 import reframe.frontend.check_filters as filters
+import reframe.frontend.dependency as dependency
 import reframe.utility.os_ext as os_ext
 from reframe.core.exceptions import (EnvironError, ConfigError, ReframeError,
                                      ReframeFatalError, format_exception,
@@ -501,6 +502,10 @@ def main():
                                        allowed_environs)
 
         # Act on checks
+        # Build dependency graph and reorder test case accordingly
+        dependency_graph = dependency.build_deps(testcases)
+        dependency.validate_deps(dependency_graph)
+        testcases = dependency.toposort(dependency_graph)
 
         # Unload regression's module and load user-specified modules
         if hasattr(settings, 'reframe_module'):
@@ -543,9 +548,9 @@ def main():
         elif options.run:
             # Setup the execution policy
             if options.exec_policy == 'serial':
-                exec_policy = SerialExecutionPolicy()
+                exec_policy = SerialExecutionPolicy(dependency_graph)
             elif options.exec_policy == 'async':
-                exec_policy = AsynchronousExecutionPolicy()
+                exec_policy = AsynchronousExecutionPolicy(dependency_graph)
             else:
                 # This should not happen, since choices are handled by
                 # argparser

reframe/frontend/dependency.py

@@ -167,3 +167,18 @@ def visit(node, path):
             cases_by_name[c.check.name] = [c]
 
     return list(itertools.chain(*(cases_by_name[n] for n in visited)))
+
+
+def create_ref_count(graph):
+
+    # TODO: create unit test
+    ref_count = {}
+    for node, deps in graph.items():
+        ref_count.setdefault(node, 0)
+        for adj in deps:
+            try:
+                ref_count[adj] += 1
+            except KeyError:
+                ref_count[adj] = 1
+
+    return ref_count

reframe/frontend/executors/__init__.py

@@ -1,5 +1,6 @@
 import abc
 import copy
+import collections
 import sys
 import weakref
 
@@ -9,6 +10,7 @@
 import reframe.core.runtime as runtime
 from reframe.core.exceptions import (AbortTaskError, JobNotStartedError,
                                      ReframeFatalError, TaskExit)
+import reframe.frontend.dependency as dependency
 from reframe.frontend.printer import PrettyPrinter
 from reframe.frontend.statistics import TestStats
 
@@ -255,6 +257,7 @@ def stats(self):
         return self._stats
 
     def runall(self, testcases):
+
         num_checks = len({tc.check.name for tc in testcases})
         self._printer.separator('short double line',
                                 'Running %d check(s)' % num_checks)
@@ -325,7 +328,7 @@ class ExecutionPolicy(abc.ABC):
 
     An execution policy implements the regression check pipeline.'''
 
-    def __init__(self):
+    def __init__(self, dependency_graph=None):
         # Options controlling the check execution
         self.skip_system_check = False
         self.force_local = False
@@ -351,6 +354,16 @@ def __init__(self):
 
         self.stats = None
 
+        # Check dependencies data
+        self.dependency_graph = dependency_graph
+        # For each check to be executed in a refame run
+        # ref_count[c] stores the number of checks depending
+        # on check c
+        if dependency_graph is not None:
+            self.ref_count = dependency.create_ref_count(dependency_graph)
+        else:
+            self.ref_count = None
+
     def __repr__(self):
         return debug.repr(self)
 

reframe/frontend/executors/policies.py

@@ -10,8 +10,8 @@
 
 
 class SerialExecutionPolicy(ExecutionPolicy):
-    def __init__(self):
-        super().__init__()
+    def __init__(self, dependency_graph=None):
+        super().__init__(dependency_graph)
         self._tasks = []
 
     def runcase(self, case):
@@ -45,7 +45,22 @@ def runcase(self, case):
             if not self.skip_performance_check:
                 task.performance()
 
-            task.cleanup(not self.keep_stage_files)
+            # Execute cleanup of current case if all dependent cases have been
+            # executed
+            if self.ref_count is None or self.ref_count[case] == 0:
+                task.cleanup(not self.keep_stage_files)
+
+            # Execute cleanup of dependencies if all dependent cases have been
+            # executed
+            if self.dependency_graph is not None:
+                for dep in self.dependency_graph[case]:
+                    self.ref_count[dep] -= 1
+                    if self.ref_count[dep] == 0:
+                        # Check if dep has failed before cleaning
+                        for t in self.stats.tasks():
+                            if t.testcase == dep and t.failed is False:
+                                t.cleanup(not self.keep_stage_files)
+                                break
 
         except TaskExit:
             return
@@ -92,9 +107,9 @@ def __call__(self, x, init_rate):
 
 
 class AsynchronousExecutionPolicy(ExecutionPolicy, TaskEventListener):
-    def __init__(self):
+    def __init__(self, dependency_graph=None):
 
-        super().__init__()
+        super().__init__(dependency_graph)
 
         # All currently running tasks
         self._running_tasks = []

unittests/resources/checks_unlisted/cleanup_checks.py

@@ -0,0 +1,127 @@
+#
+# Checks for testing the cleanup procedure with dependencies
+#
+
+import reframe as rfm
+import reframe.utility.sanity as sn
+
+
+@rfm.simple_test
+class DependencyT0(rfm.RunOnlyRegressionTest):
+    def __init__(self):
+        self.local = True
+        self.executable = 'echo DependencyT0'
+        self.sanity_patterns = sn.assert_found('Dependency', self.stdout)
+        self.valid_systems = ['*']
+        self.valid_prog_environs = ['*']
+        self.depends_on('DependencyT1')
+
+
+@rfm.simple_test
+class DependencyT1(rfm.RunOnlyRegressionTest):
+    def __init__(self):
+        self.local = True
+        self.executable = 'echo DependencyT1'
+        self.sanity_patterns = sn.assert_found('Dependency', self.stdout)
+        self.valid_systems = ['*']
+        self.valid_prog_environs = ['*']
+        self.depends_on('DependencyT2')
+
+
+@rfm.simple_test
+class DependencyT2(rfm.RunOnlyRegressionTest):
+    def __init__(self):
+        self.local = True
+        self.executable = 'echo DependencyT2'
+        self.sanity_patterns = sn.assert_found('Dependency', self.stdout)
+        self.valid_systems = ['*']
+        self.valid_prog_environs = ['*']
+        self.depends_on('DependencyT3')
+
+
+@rfm.simple_test
+class DependencyT3(rfm.RunOnlyRegressionTest):
+    def __init__(self):
+        self.local = True
+        self.executable = 'echo DependencyT3'
+        self.sanity_patterns = sn.assert_found('Dependency', self.stdout)
+        self.valid_systems = ['*']
+        self.valid_prog_environs = ['*']
+
+
+@rfm.simple_test
+class MultiDependencyT0(rfm.RunOnlyRegressionTest):
+    def __init__(self):
+        self.local = True
+        self.executable = 'echo MultiDependencyT0'
+        self.sanity_patterns = sn.assert_found('Dependency', self.stdout)
+        self.valid_systems = ['*']
+        self.valid_prog_environs = ['*']
+        self.depends_on('MultiDependencyT1')
+        self.depends_on('MultiDependencyT2')
+
+
+@rfm.simple_test
+class MultiDependencyT1(rfm.RunOnlyRegressionTest):
+    def __init__(self):
+        self.local = True
+        self.executable = 'echo MultiDependencyT1'
+        self.sanity_patterns = sn.assert_found('Dependency', self.stdout)
+        self.valid_systems = ['*']
+        self.valid_prog_environs = ['*']
+        self.depends_on('MultiDependencyT6')
+
+
+@rfm.simple_test
+class MultiDependencyT2(rfm.RunOnlyRegressionTest):
+    def __init__(self):
+        self.local = True
+        self.executable = 'echo MultiDependencyT2'
+        self.sanity_patterns = sn.assert_found('Dependency', self.stdout)
+        self.valid_systems = ['*']
+        self.valid_prog_environs = ['*']
+        self.depends_on('MultiDependencyT3')
+        self.depends_on('MultiDependencyT4')
+
+
+@rfm.simple_test
+class MultiDependencyT3(rfm.RunOnlyRegressionTest):
+    def __init__(self):
+        self.local = True
+        self.executable = 'echo MultiDependencyT3'
+        self.sanity_patterns = sn.assert_found('Dependency', self.stdout)
+        self.valid_systems = ['*']
+        self.valid_prog_environs = ['*']
+        self.depends_on('MultiDependencyT6')
+        self.depends_on('MultiDependencyT5')
+
+
+@rfm.simple_test
+class MultiDependencyT4(rfm.RunOnlyRegressionTest):
+    def __init__(self):
+        self.local = True
+        self.executable = 'echo MultiDependencyT4'
+        self.sanity_patterns = sn.assert_found('Dependency', self.stdout)
+        self.valid_systems = ['*']
+        self.valid_prog_environs = ['*']
+        self.depends_on('MultiDependencyT5')
+
+
+@rfm.simple_test
+class MultiDependencyT5(rfm.RunOnlyRegressionTest):
+    def __init__(self):
+        self.local = True
+        self.executable = 'echo MultiDependencyT5'
+        self.sanity_patterns = sn.assert_found('Dependency', self.stdout)
+        self.valid_systems = ['*']
+        self.valid_prog_environs = ['*']
+
+
+@rfm.simple_test
+class MultiDependencyT6(rfm.RunOnlyRegressionTest):
+    def __init__(self):
+        self.local = True
+        self.executable = 'echo MultiDependencyT6'
+        self.sanity_patterns = sn.assert_found('Dependency', self.stdout)
+        self.valid_systems = ['*']
+        self.valid_prog_environs = ['*']

unittests/resources/checks_unlisted/test_cleanup.py

@@ -0,0 +1,22 @@
+import unittest
+from unittests.test_cli import TestFrontend
+
+
+class TestCleanup(TestFrontend):
+    def test_dependency_cli(self):
+        self.checkpath = [
+            'unittests/resources/checks_unlisted/cleanup_checks.py']
+        self.action = 'run'
+        self.more_options = ['-n', 'Dependency']
+        returncode, stdout, _ = self._run_reframe()
+        self.assertIn('Running 4 check(s)', stdout)
+        self.assertEqual(0, returncode)
+
+    def test_multi_dependency_cli(self):
+        self.checkpath = [
+            'unittests/resources/checks_unlisted/cleanup_checks.py']
+        self.action = 'run'
+        self.more_options = ['-n', 'MultiDependency']
+        returncode, stdout, _ = self._run_reframe()
+        self.assertIn('Running 7 check(s)', stdout)
+        self.assertEqual(0, returncode)