| # Copyright lowRISC contributors. |
| # Licensed under the Apache License, Version 2.0, see LICENSE for details. |
| # SPDX-License-Identifier: Apache-2.0 |
| |
| import collections |
| import datetime |
| import logging as log |
| import os |
| import re |
| import sys |
| from pathlib import Path |
| |
| from utils import VERBOSE, clean_odirs, mk_symlink, rm_path |
| |
| |
| class LauncherError(Exception): |
| |
| def __init__(self, msg): |
| self.msg = msg |
| |
| |
| class ErrorMessage( |
| collections.namedtuple( |
| 'ErrorMessage', |
| ['line_number', 'message', 'context'], |
| )): |
| """Contains error-related information. |
| |
| This support classification of failures into buckets. The message field |
| is used to generate the bucket, and context contains a list of lines in |
| the failing log that can be useful for quick diagnostics. |
| """ |
| pass |
| |
| |
| class Launcher: |
| """ |
| Abstraction for launching and maintaining a job. |
| |
| An abstract class that provides methods to prepare a job's environment, |
| launch the job, poll for its completion and finally do some cleanup |
| activities. This class is not meant to be instantiated directly. Each |
| launcher object holds an instance of the deploy object. |
| """ |
| |
| # Type of launcher used as string. |
| variant = None |
| |
| # Max jobs running at one time |
| max_parallel = sys.maxsize |
| |
| # Max jobs polled at one time |
| max_poll = 10000 |
| |
| # Poll job's completion status every this many seconds |
| poll_freq = 1 |
| |
| # Points to the python virtual env area. |
| pyvenv = None |
| |
| # If a history of previous invocations is to be maintained, then keep no |
| # more than this many directories. |
| max_odirs = 5 |
| |
| # Flag indicating the workspace preparation steps are complete. |
| workspace_prepared = False |
| workspace_prepared_for_cfg = set() |
| |
| # Jobs that are not run when one of their dependent jobs fail are |
| # considered killed. All non-passing jobs are required to have an |
| # an associated fail_msg attribute (an object of class ErrorMessage) |
| # reflecting the appropriate message. This class attribute thus serves |
| # as a catch-all for all those jobs that are not even run. If a job |
| # instance runs and fails, the fail_msg attribute is overridden by |
| # the instance with the correct message in _post_finish(). |
| fail_msg = ErrorMessage( |
| line_number=None, |
| message="Job killed most likely because its dependent job failed.", |
| context=[]) |
| |
| @staticmethod |
| def set_pyvenv(project): |
| '''Activate a python virtualenv if available. |
| |
| The env variable <PROJECT>_PYTHON_VENV if set, points to the path |
| containing the python virtualenv created specifically for this |
| project. We can activate it if needed, before launching jobs using |
| external compute machines. |
| |
| This is not applicable when running jobs locally on the user's machine. |
| ''' |
| |
| if Launcher.pyvenv is not None: |
| return |
| |
| # If project-specific python virtualenv path is set, then activate it |
| # before running downstream tools. This is more relevant when not |
| # launching locally, but on external machines in a compute farm, which |
| # may not have access to the default python installation area on the |
| # host machine. |
| # |
| # The code below allows each launcher variant to set its own virtualenv |
| # because the loading / activating mechanism could be different between |
| # them. |
| Launcher.pyvenv = os.environ.get("{}_PYVENV_{}".format( |
| project.upper(), Launcher.variant.upper())) |
| |
| if not Launcher.pyvenv: |
| Launcher.pyvenv = os.environ.get("{}_PYVENV".format( |
| project.upper())) |
| |
| @staticmethod |
| def prepare_workspace(project, repo_top, args): |
| '''Prepare the workspace based on the chosen launcher's needs. |
| |
| This is done once for the entire duration for the flow run. |
| 'project' is the name of the project. |
| 'repo_top' is the path to the repository. |
| 'args' are the command line args passed to dvsim. |
| ''' |
| pass |
| |
| @staticmethod |
| def prepare_workspace_for_cfg(cfg): |
| '''Prepare the workspace for a cfg. |
| |
| This is invoked once for each cfg. |
| 'cfg' is the flow configuration object. |
| ''' |
| pass |
| |
| def __str__(self): |
| return self.deploy.full_name + ":launcher" |
| |
| def __init__(self, deploy): |
| cfg = deploy.sim_cfg |
| |
| # One-time preparation of the workspace. |
| if not Launcher.workspace_prepared: |
| self.prepare_workspace(cfg.project, cfg.proj_root, cfg.args) |
| Launcher.workspace_prepared = True |
| |
| # One-time preparation of the workspace, specific to the cfg. |
| if cfg not in Launcher.workspace_prepared_for_cfg: |
| self.prepare_workspace_for_cfg(cfg) |
| Launcher.workspace_prepared_for_cfg.add(cfg) |
| |
| # Store the deploy object handle. |
| self.deploy = deploy |
| |
| # Status of the job. This is primarily determined by the |
| # _check_status() method, but eventually updated by the _post_finish() |
| # method, in case any of the cleanup tasks fails. This value is finally |
| # returned to the Scheduler by the poll() method. |
| self.status = None |
| |
| # Return status of the process running the job. |
| self.exit_code = None |
| |
| # Flag to indicate whether to 'overwrite' if odir already exists, |
| # or to backup the existing one and create a new one. |
| # For builds, we want to overwrite existing to leverage the tools' |
| # incremental / partition compile features. For runs, we may want to |
| # create a new one. |
| self.renew_odir = False |
| |
| # The actual job runtime computed by dvsim, in seconds. |
| self.job_runtime_secs = 0 |
| |
| def _make_odir(self): |
| """Create the output directory.""" |
| |
| # If renew_odir flag is True - then move it. |
| if self.renew_odir: |
| clean_odirs(odir=self.deploy.odir, max_odirs=self.max_odirs) |
| os.makedirs(self.deploy.odir, exist_ok=True) |
| |
| def _link_odir(self, status): |
| """Soft-links the job's directory based on job's status. |
| |
| The dispatched, passed and failed directories in the scratch area |
| provide a quick way to get to the job that was executed. |
| """ |
| |
| dest = Path(self.deploy.sim_cfg.links[status], self.deploy.qual_name) |
| mk_symlink(self.deploy.odir, dest) |
| |
| # Delete the symlink from dispatched directory if it exists. |
| if status != "D": |
| old = Path(self.deploy.sim_cfg.links['D'], self.deploy.qual_name) |
| rm_path(old) |
| |
| def _dump_env_vars(self, exports): |
| """Write env vars to a file for ease of debug. |
| |
| Each extended class computes the list of exports and invokes this |
| method right before launching the job. |
| """ |
| |
| with open(self.deploy.odir + "/env_vars", |
| "w", |
| encoding="UTF-8", |
| errors="surrogateescape") as f: |
| for var in sorted(exports.keys()): |
| f.write("{}={}\n".format(var, exports[var])) |
| |
| def _pre_launch(self): |
| """Do pre-launch activities. |
| |
| Examples include such as preparing the job's environment, clearing |
| old runs, creating the output directory, dumping all env variables |
| etc. This method is already invoked by launch() as the first step. |
| """ |
| |
| self.deploy.pre_launch() |
| self._make_odir() |
| self.start_time = datetime.datetime.now() |
| |
| def _do_launch(self): |
| """Launch the job.""" |
| |
| raise NotImplementedError() |
| |
| def launch(self): |
| """Launch the job.""" |
| |
| self._pre_launch() |
| self._do_launch() |
| |
| def poll(self): |
| """Poll the launched job for completion. |
| |
| Invokes _check_status() and _post_finish() when the job completes. |
| """ |
| |
| raise NotImplementedError() |
| |
| def kill(self): |
| """Terminate the job.""" |
| |
| raise NotImplementedError() |
| |
| def _check_status(self): |
| """Determine the outcome of the job (P/F if it ran to completion). |
| |
| Returns (status, err_msg) extracted from the log, where the status is |
| "P" if the it passed, "F" otherwise. This is invoked by poll() just |
| after the job finishes. err_msg is an instance of the named tuple |
| ErrorMessage. |
| """ |
| |
| def _find_patterns(patterns, line): |
| """Helper function that returns the pattern if any of the given |
| patterns is found, else None.""" |
| |
| assert patterns |
| for pattern in patterns: |
| match = re.search(r"{}".format(pattern), line) |
| if match: |
| return pattern |
| return None |
| |
| if self.deploy.dry_run: |
| return "P", None |
| |
| # Only one fail pattern needs to be seen. |
| chk_failed = bool(self.deploy.fail_patterns) |
| |
| # All pass patterns need to be seen, so we replicate the list and remove |
| # patterns as we encounter them. |
| pass_patterns = self.deploy.pass_patterns.copy() |
| chk_passed = bool(pass_patterns) and (self.exit_code == 0) |
| |
| try: |
| with open(self.deploy.get_log_path(), |
| "r", |
| encoding="UTF-8", |
| errors="surrogateescape") as f: |
| lines = f.readlines() |
| except OSError as e: |
| return "F", ErrorMessage( |
| line_number=None, |
| message="Error opening file {}:\n{}".format( |
| self.deploy.get_log_path(), e), |
| context=[], |
| ) |
| |
| # Since the log file is already opened and read to assess the job's |
| # status, use this opportunity to also extract other pieces of |
| # information. |
| self.deploy.extract_info_from_log(lines) |
| |
| if chk_failed or chk_passed: |
| for cnt, line in enumerate(lines): |
| if chk_failed: |
| if _find_patterns(self.deploy.fail_patterns, line): |
| # If failed, then nothing else to do. Just return. |
| # Provide some extra lines for context. |
| return "F", ErrorMessage(line_number=cnt + 1, |
| message=line.strip(), |
| context=lines[cnt:cnt + 5]) |
| |
| if chk_passed: |
| pattern = _find_patterns(pass_patterns, line) |
| if pattern: |
| pass_patterns.remove(pattern) |
| chk_passed = bool(pass_patterns) |
| |
| # If no fail patterns were seen, but the job returned with non-zero |
| # exit code for whatever reason, then show the last 10 lines of the log |
| # as the failure message, which might help with the debug. |
| if self.exit_code != 0: |
| return "F", ErrorMessage(line_number=None, |
| message="Job returned non-zero exit code", |
| context=lines[-10:]) |
| if chk_passed: |
| return "F", ErrorMessage( |
| line_number=None, |
| message=f"Some pass patterns missing: {pass_patterns}", |
| context=lines[-10:], |
| ) |
| return "P", None |
| |
| def _post_finish(self, status, err_msg): |
| """Do post-completion activities, such as preparing the results. |
| |
| Must be invoked by poll(), after the job outcome is determined. |
| |
| status is the status of the job, either 'P', 'F' or 'K'. |
| err_msg is an instance of the named tuple ErrorMessage. |
| """ |
| |
| assert status in ['P', 'F', 'K'] |
| self._link_odir(status) |
| log.debug("Item %s has completed execution: %s", self, status) |
| |
| try: |
| # Run the target-specific cleanup tasks regardless of the job's |
| # outcome. |
| self.deploy.post_finish(status) |
| except Exception as e: |
| # If the job had already failed, then don't do anything. If it's |
| # cleanup task failed, then mark the job as failed. |
| if status == "P": |
| status = "F" |
| err_msg = ErrorMessage(line_number=None, |
| message=f"{e}", |
| context=[f"{e}"]) |
| |
| self.status = status |
| if self.status != "P": |
| assert err_msg and isinstance(err_msg, ErrorMessage) |
| self.fail_msg = err_msg |
| log.log(VERBOSE, err_msg.message) |
