Merge pull request #2545 from vkarak/feat/time-profiler

[feat] Add a lightweight time profiler in the framework

Author: Vasileios Karakasis

reframe/core/decorators.py

@@ -23,7 +23,7 @@
 import reframe.core.hooks as hooks
 from reframe.core.exceptions import ReframeSyntaxError, SkipTestError, what
 from reframe.core.fixtures import FixtureRegistry
-from reframe.core.logging import getlogger
+from reframe.core.logging import getlogger, time_function
 from reframe.core.pipeline import RegressionTest
 from reframe.utility.versioning import VersionValidator
 
@@ -64,6 +64,7 @@ def skip(self, test):
         '''Add a test to the skip set.'''
         self._skip_tests.add(test)
 
+    @time_function
     def instantiate_all(self, reset_sysenv=0):
         '''Instantiate all the registered tests.
 

reframe/core/logging.py

@@ -20,6 +20,7 @@
 import reframe.utility.jsonext as jsonext
 import reframe.utility.osext as osext
 from reframe.core.exceptions import ConfigError, LoggingError
+from reframe.utility.profile import TimeProfiler
 
 
 # Global configuration options for logging
@@ -726,3 +727,31 @@ def getlogger():
 
 def getperflogger(check):
     return LoggerAdapter(_perf_logger, check)
+
+
+# Global framework profiler
+_profiler = TimeProfiler()
+
+
+def getprofiler():
+    return _profiler
+
+
+def time_function(fn):
+    '''Decorator for timing a function using the global profiler'''
+
+    def _fn(*args, **kwargs):
+        with _profiler.time_region(fn.__qualname__):
+            return fn(*args, **kwargs)
+
+    return _fn
+
+
+def time_function_noexit(fn):
+    '''Decorator for timing a function using the global profiler'''
+
+    def _fn(*args, **kwargs):
+        _profiler.enter_region(fn.__qualname__)
+        return fn(*args, **kwargs)
+
+    return _fn

reframe/frontend/cli.py

@@ -47,6 +47,7 @@ def format_env(envvars):
     return ret
 
 
+@logging.time_function
 def list_checks(testcases, printer, detailed=False, concretized=False):
     printer.info('[List of matched checks]')
     unique_checks = set()
@@ -114,6 +115,7 @@ def dep_lines(u, *, prefix, depth=0, lines=None, printed=None):
         printer.info(f'Found {len(unique_checks)} check(s)\n')
 
 
+@logging.time_function
 def describe_checks(testcases, printer):
     records = []
     unique_names = set()
@@ -186,6 +188,7 @@ def calc_verbosity(site_config, quiesce):
     return curr_verbosity - quiesce
 
 
+@logging.time_function_noexit
 def main():
     # Setup command line options
     argparser = argparse.ArgumentParser()
@@ -950,6 +953,8 @@ def print_infoline(param, value):
     print_infoline('output directory', repr(session_info['prefix_output']))
     printer.info('')
     try:
+        logging.getprofiler().enter_region('test processing')
+
         # Need to parse the cli options before loading the tests
         parsed_job_options = []
         for opt in options.job_options:
@@ -1369,6 +1374,7 @@ def module_unuse(*paths):
         sys.exit(1)
     finally:
         try:
+            logging.getprofiler().exit_region()     # region: 'test processing'
             log_files = logging.log_files()
             if site_config.get('general/0/save_log_files'):
                 log_files = logging.save_log_files(rt.output_prefix)
@@ -1379,3 +1385,6 @@ def module_unuse(*paths):
         finally:
             if not restrict_logging():
                 printer.info(logfiles_message())
+
+            logging.getprofiler().exit_region()     # region: 'main'
+            logging.getprofiler().print_report(printer.debug)

reframe/frontend/dependencies.py

@@ -13,9 +13,10 @@
 
 import reframe.utility as util
 from reframe.core.exceptions import DependencyError
-from reframe.core.logging import getlogger
+from reframe.core.logging import getlogger, time_function
 
 
+@time_function
 def build_deps(cases, default_cases=None):
     '''Build dependency graph from test cases.
 
@@ -139,6 +140,7 @@ def _reduce_deps(graph):
     return ret
 
 
+@time_function
 def validate_deps(graph):
     '''Validate dependency graph.'''
 
@@ -181,6 +183,7 @@ def validate_deps(graph):
         sources -= visited
 
 
+@time_function
 def prune_deps(graph, testcases, max_depth=None):
     '''Prune the graph so that it contains only the specified cases and their
     dependencies up to max_depth.
@@ -205,6 +208,7 @@ def prune_deps(graph, testcases, max_depth=None):
     return pruned_graph
 
 
+@time_function
 def toposort(graph, is_subgraph=False):
     '''Return a list of the graph nodes topologically sorted.
 

reframe/frontend/executors/__init__.py

@@ -106,6 +106,7 @@ def clone(self):
         return TestCase(self._check_orig, self._partition, self._environ)
 
 
+@logging.time_function
 def generate_testcases(checks, prepare=False):
     '''Generate concrete test cases from checks.
 
@@ -313,22 +314,27 @@ def __exit__(this, exc_type, exc_value, traceback):
             self.fail()
             raise TaskExit from e
 
+    @logging.time_function
     def setup(self, *args, **kwargs):
         self.testcase.prepare()
         self._safe_call(self.check.setup, *args, **kwargs)
         self._notify_listeners('on_task_setup')
 
+    @logging.time_function
     def compile(self):
         self._safe_call(self.check.compile)
         self._notify_listeners('on_task_compile')
 
+    @logging.time_function
     def compile_wait(self):
         self._safe_call(self.check.compile_wait)
 
+    @logging.time_function
     def run(self):
         self._safe_call(self.check.run)
         self._notify_listeners('on_task_run')
 
+    @logging.time_function
     def run_complete(self):
         done = self._safe_call(self.check.run_complete)
         if done:
@@ -337,23 +343,28 @@ def run_complete(self):
 
         return done
 
+    @logging.time_function
     def compile_complete(self):
         done = self._safe_call(self.check.compile_complete)
         if done:
             self._notify_listeners('on_task_compile_exit')
 
         return done
 
+    @logging.time_function
     def run_wait(self):
         self._safe_call(self.check.run_wait)
         self.zombie = False
 
+    @logging.time_function
     def sanity(self):
         self._safe_call(self.check.sanity)
 
+    @logging.time_function
     def performance(self):
         self._safe_call(self.check.performance)
 
+    @logging.time_function
     def finalize(self):
         try:
             jsonfile = os.path.join(self.check.stagedir, '.rfm_testcase.json')
@@ -367,6 +378,7 @@ def finalize(self):
         self._current_stage = 'finalize'
         self._notify_listeners('on_task_success')
 
+    @logging.time_function
     def cleanup(self, *args, **kwargs):
         self._safe_call(self.check.cleanup, *args, **kwargs)
 
@@ -479,6 +491,7 @@ def policy(self):
     def stats(self):
         return self._stats
 
+    @logging.time_function
     def runall(self, testcases, restored_cases=None):
         num_checks = len({tc.check.unique_name for tc in testcases})
         self._printer.separator('short double line',

reframe/frontend/loader.py

@@ -18,7 +18,7 @@
 import reframe.utility as util
 import reframe.utility.osext as osext
 from reframe.core.exceptions import NameConflictError, is_severe, what
-from reframe.core.logging import getlogger
+from reframe.core.logging import getlogger, time_function
 
 
 class RegressionCheckValidator(ast.NodeVisitor):
@@ -267,6 +267,7 @@ def load_from_dir(self, dirname, recurse=False, force=False):
 
         return checks
 
+    @time_function
     def load_all(self, force=False):
         '''Load all checks in self._load_path.
 

reframe/frontend/testgenerators.py

@@ -9,12 +9,13 @@
 import reframe.utility as util
 
 from reframe.core.decorators import TestRegistry
-from reframe.core.logging import getlogger
+from reframe.core.logging import getlogger, time_function
 from reframe.core.meta import make_test
 from reframe.core.schedulers import Job
 from reframe.frontend.executors import generate_testcases
 
 
+@time_function
 def getallnodes(state='all', jobs_cli_options=None):
     rt = runtime.runtime()
     nodes = {}
@@ -72,6 +73,7 @@ def _rfm_set_valid_systems(obj):
     return _rfm_set_valid_systems
 
 
+@time_function
 def distribute_tests(testcases, node_map):
     '''Returns new testcases that will be parameterized to run in node of
     their partitions based on the nodemap
@@ -126,6 +128,7 @@ def distribute_tests(testcases, node_map):
     return generate_testcases(new_checks)
 
 
+@time_function
 def repeat_tests(testcases, num_repeats):
     '''Returns new test cases parameterized over their repetition number'''
 

reframe/utility/profile.py

@@ -0,0 +1,96 @@
+# Copyright 2016-2022 Swiss National Supercomputing Centre (CSCS/ETH Zurich)
+# ReFrame Project Developers. See the top-level LICENSE file for details.
+#
+# SPDX-License-Identifier: BSD-3-Clause
+
+# A lightweight time profiler
+
+import time
+import sys
+
+from collections import OrderedDict
+
+
+class ProfilerError(Exception):
+    pass
+
+
+class time_region:
+    '''Context manager for timing a code region'''
+
+    def __init__(self, region, profiler=None):
+        self._profiler = profiler or TimeProfiler()
+        self._region = region
+
+    def __enter__(self):
+        self._profiler.enter_region(self._region)
+        return self._profiler
+
+    def __exit__(self, exc_type, exc_val, exc_tb):
+        self._profiler.exit_region()
+
+
+class TimeProfiler:
+    def __init__(self):
+        self._region_stack = ['root']
+        if sys.version_info[:2] < (3, 8):
+            self._region_times = OrderedDict()
+        else:
+            self._region_times = {}
+
+    @property
+    def current_region(self):
+        return self._region_stack[-1]
+
+    def enter_region(self, region_name):
+        timestamp = time.time()
+        region_fullname = f'{self.current_region}:{region_name}'
+        if region_fullname in self._region_times:
+            elapsed = self._region_times[region_fullname][1]
+        else:
+            elapsed = 0.0
+
+        self._region_times[region_fullname] = (timestamp, elapsed)
+        self._region_stack.append(region_fullname)
+
+    def exit_region(self):
+        timestamp = time.time()
+        region = self.current_region
+        t_start, elapsed = self._region_times[region]
+        self._region_times[region] = (None, elapsed + timestamp - t_start)
+        self._region_stack.pop()
+
+    def total_time(self, region_name):
+        for region in reversed(self._region_times.keys()):
+            if (region == region_name or
+                region.rsplit(':', maxsplit=1)[-1] == region_name):
+                timestamp, elapsed = self._region_times[region]
+                if timestamp:
+                    raise ProfilerError(
+                        f'region {region_name!r} has not exited'
+                    )
+
+                return elapsed
+
+        raise ProfilerError(f'unknown region: {region_name!r}')
+
+    def time_region(self, region):
+        return globals()['time_region'](region, self)
+
+    def print_report(self, print_fn=None):
+        if print_fn is None:
+            print_fn = print
+
+        print_fn('>>> profiler report [start] <<<')
+        for name, t_info in self._region_times.items():
+            # Remove the root prefix
+            levels = name.count(':')
+            indent = ' '*4*(levels - 1)
+            region_name = name.rsplit(':', maxsplit=1)[-1]
+            msg = f'{indent}{region_name}: {t_info[1]:.6f} s'
+            if t_info[0]:
+                msg += ' <incomplete>'
+
+            print_fn(msg)
+
+        print_fn('>>> profiler report [ end ] <<<')