util/dvsim/LsfLauncher.py

# Copyright lowRISC contributors.
# Licensed under the Apache License, Version 2.0, see LICENSE for details.
# SPDX-License-Identifier: Apache-2.0

import logging as log
import os
import re
import subprocess
import tarfile
from pathlib import Path

from Launcher import ErrorMessage, Launcher, LauncherError
from utils import VERBOSE, clean_odirs


class LsfLauncher(Launcher):

    # A hidden directory specific to a cfg, where we put individual 'job'
    # scripts.
    jobs_dir = {}

    # All launcher instances available for lookup.
    jobs = {}

    # When the job completes, we try to read the job script output to determine
    # the outcome. It may not have been completely written the first time we
    # read it so we retry on the next poll, no more than 10 times.
    max_poll_retries = 10

    # TODO: Add support for build/run/cov job specific resource requirements:
    #       cpu, mem, disk, stack.
    # TODO: Allow site-specific job resource usage setting using
    #       `DVSIM_LSF_CFG` environment variable.

    @staticmethod
    def prepare_workspace(project, repo_top, args):
        # Since we dispatch to remote machines, a project specific python
        # virtualenv is exists, needs to be activated when launching the job.
        Launcher.set_pyvenv(project)
        if Launcher.pyvenv is None:
            return

        # If it is already a dir, then nothing to be done.
        if os.path.isdir(Launcher.pyvenv):
            return

        # If not, then it needs to be a valid tarball. Extract it in the
        # scratch area if it does not exist.
        stem = Path(Launcher.pyvenv).stem
        if stem.endswith("tar"):
            stem = stem[:-4]
        path = Path(args.scratch_root, stem)
        if not path.is_dir():
            log.info("[prepare_workspace]: [pyvenv]: Extracting %s",
                     Launcher.pyvenv)
            with tarfile.open(Launcher.pyvenv, mode='r') as tar:
                tar.extractall(args.scratch_root)
            log.info("[prepare_workspace]: [pyvenv]: Done: %s", path)
        Launcher.pyvenv = path

    @staticmethod
    def prepare_workspace_for_cfg(cfg):
        # Create the job dir.
        LsfLauncher.jobs_dir[cfg] = Path(cfg.scratch_path, "lsf",
                                         cfg.timestamp)
        clean_odirs(odir=LsfLauncher.jobs_dir[cfg], max_odirs=2)
        os.makedirs(Path(LsfLauncher.jobs_dir[cfg]), exist_ok=True)

    @staticmethod
    def make_job_script(cfg, job_name):
        """Creates the job script.

        Once all jobs in the array are launched, the job script can be created.
        It is a bash script that takes the job index as a single argument.
        This index is set in the bsub command as '$LSB_JOBINDEX', which bsub
        sets as the actual index when launching that job in the array. This
        script is super simple - it is just a giant case statement that
        switches on the job index to run that specific job. This preferred over
        creating individual scripts for each job which incurs additional file
        I/O overhead when the scratch area is on NFS, causing a slowdown.

        Returns the path to the job script.
        """

        lines = ["#!/usr/bin/env bash\nset -e\n"]

        # Activate the python virtualenv if it exists.
        if Launcher.pyvenv:
            lines += ["source {}/bin/activate\n".format(Launcher.pyvenv)]

        lines += ["case $1 in\n"]
        for job in LsfLauncher.jobs[cfg][job_name]:
            # Redirect the job's stdout and stderr to its log file.
            cmd = "{} > {} 2>&1".format(job.deploy.cmd,
                                        job.deploy.get_log_path())
            lines += ["  {})\n".format(job.index), "    {};;\n".format(cmd)]

        # Throw error as a sanity check if the job index is invalid.
        lines += [
            "  *)\n",
            "    echo \"ERROR: Illegal job index: $1\" 1>&2; exit 1;;\n",
            "esac\n"
        ]
        if Launcher.pyvenv:
            lines += ["deactivate\n"]

        job_script = Path(LsfLauncher.jobs_dir[cfg], job_name)
        try:
            with open(job_script, "w", encoding='utf-8') as f:
                f.writelines(lines)
        except IOError as e:
            err_msg = "ERROR: Failed to write {}:\n{}".format(job_script, e)
            LsfLauncher._post_finish_job_array(cfg, job_name, err_msg)
            raise LauncherError(err_msg)

        log.log(VERBOSE, "[job_script]: %s", job_script)
        return job_script

    def __init__(self, deploy):
        super().__init__(deploy)

        # Maintain the job script output as an instance variable for polling
        # and cleanup.
        self.bsub_out = None

        # If we already opened the job script output file (but have not
        # determined the outcome), then we maintain the file descriptor rather
        # then reopening it and starting all over again on the next poll.
        self.bsub_out_fd = None
        self.bsub_out_err_msg = []
        self.bsub_out_err_msg_found = False

        # Set the job id.
        self.job_id = None

        # Polling retry counter..
        self.num_poll_retries = 0

        # Add self to the list of jobs.
        cfg_dict = LsfLauncher.jobs.setdefault(deploy.sim_cfg, {})
        job_name_list = cfg_dict.setdefault(deploy.job_name, [])
        job_name_list.append(self)

        # Job's index in the array.
        self.index = len(job_name_list)

    def _do_launch(self):
        # Add self to the list of jobs.
        job_name = self.deploy.job_name
        cfg = self.deploy.sim_cfg
        job_total = len(LsfLauncher.jobs[cfg][job_name])

        # The actual launching of the bsub command cannot happen until the
        # Scheduler has dispatched ALL jobs in the array.
        if self.index < job_total:
            return

        job_script = self.make_job_script(cfg, job_name)

        # Update the shell's env vars with self.exports. Values in exports must
        # replace the values in the shell's env vars if the keys match.
        exports = os.environ.copy()
        if self.deploy.exports:
            exports.update(self.deploy.exports)

        # Clear the magic MAKEFLAGS variable from exports if necessary. This
        # variable is used by recursive Make calls to pass variables from one
        # level to the next. Here, self.cmd is a call to Make but it's
        # logically a top-level invocation: we don't want to pollute the flow's
        # Makefile with Make variables from any wrapper that called dvsim.
        if 'MAKEFLAGS' in exports:
            del exports['MAKEFLAGS']

        self._dump_env_vars(exports)

        # TODO: Arbitrarily set the max slot-limit to 100.
        job_array = "{}[1-{}]".format(job_name, job_total)
        if job_total > 100:
            job_array += "%100"

        # TODO: This needs to be moved to a HJson.
        if self.deploy.sim_cfg.tool == "vcs":
            job_rusage = "\'rusage[vcssim=1,vcssim_dynamic=1:duration=1]\'"

        elif self.deploy.sim_cfg.tool == "xcelium":
            job_rusage = "\'rusage[xcelium=1,xcelium_dynamic=1:duration=1]\'"

        else:
            job_rusage = None

        # Launch the job array.
        cmd = [
            "bsub",
            # TODO: LSF project name could be site specific!
            "-P",
            cfg.project,
            "-J",
            job_array,
            "-oo",
            "{}.%I.out".format(job_script),
            "-eo",
            "{}.%I.out".format(job_script)
        ]

        if job_rusage:
            cmd += ["-R", job_rusage]

        cmd += ["/usr/bin/bash {} $LSB_JOBINDEX".format(job_script)]

        try:
            p = subprocess.run(cmd,
                               check=True,
                               timeout=60,
                               stdout=subprocess.PIPE,
                               stderr=subprocess.PIPE,
                               env=exports)
        except subprocess.CalledProcessError as e:
            # Need to mark all jobs in this range with this fail pattern.
            err_msg = e.stderr.decode("utf-8").strip()
            self._post_finish_job_array(cfg, job_name, err_msg)
            raise LauncherError(err_msg)

        # Extract the job ID.
        result = p.stdout.decode("utf-8").strip()
        job_id = result.split('Job <')[1].split('>')[0]
        if not job_id:
            self._post_finish_job_array(cfg, job_name, "Job ID not found!")
            raise LauncherError(err_msg)

        for job in LsfLauncher.jobs[cfg][job_name]:
            job.bsub_out = Path("{}.{}.out".format(job_script, job.index))
            job.job_id = "{}[{}]".format(job_id, job.index)
            job._link_odir("D")

    def poll(self):
        # It is possible we may have determined the status already.
        if self.status:
            return self.status

        if not self.bsub_out_fd:
            # If job id is not set, the bsub command has not been sent yet.
            if not self.job_id:
                return 'D'

            # We redirect the job's output to the log file, so the job script
            # output remains empty until the point it finishes. This is a very
            # quick way to check if the job has completed. If nothing has been
            # written to the job script output yet (or if it is not yet
            # created), then the job is still running.
            try:
                if not self.bsub_out.stat().st_size:
                    return "D"
            except FileNotFoundError:
                return "D"

            # If we got to this point,  we can now open the job script output
            # file for reading.
            try:
                self.bsub_out_fd = open(self.bsub_out, "r")
            except IOError as e:
                self._post_finish(
                    "F",
                    ErrorMessage(
                        line_number=None,
                        message="ERROR: Failed to open {}\n{}.".format(
                            self.bsub_out, e),
                        context=[]))
                return "F"

        # Now that the job has completed, we need to determine its status.
        #
        # If the job successfully launched and it failed, the failure message
        # will appear in its log file (because of the stderr redirection).
        # But, in some cases, if there is something wrong with the command
        # itself, bsub might return immediately with an error message, which
        # will appear in the job script output file. We want to retrieve that
        # so that we can report the status accurately.
        #
        # At this point, we could run bjobs or bhist to determine the status,
        # but it has been found to be too slow, expecially when running 1000s
        # of jobs. Plus, we have to read the job script output anyway to look
        # for those error messages.
        #
        # So we just read this file to determine both, the status and extract
        # the error message, rather than running bjobs or bhist. But there is
        # one more complication to deal with - if we read the file now, it is
        # possible that it may not have been fully updated. We will try reading
        # it anyway. If we are unable to find what we are looking for, then we
        # will resume reading it again at the next poll. We will do this upto
        # max_poll_retries times before giving up and flagging an error.
        #
        # TODO: Consider using the IBM Plarform LSF Python APIs instead.
        #       (deferred due to shortage of time / resources).
        # TODO: Parse job telemetry data for performance insights.

        exit_code = self._get_job_exit_code()
        if exit_code is not None:
            self.exit_code = exit_code
            status, err_msg = self._check_status()
            # Prioritize error messages from bsub over the job's log file.
            if self.bsub_out_err_msg:
                err_msg = ErrorMessage(line_number=None,
                                       message=self.bsub_out_err_msg,
                                       context=[])
            self._post_finish(status, err_msg)
            return status

        else:
            self.num_poll_retries += 1
            # Fail the test if we have reached the max polling retries.
            if self.num_poll_retries == LsfLauncher.max_poll_retries:
                self._post_finish(
                    "F", "ERROR: Reached max retries while "
                    "reading job script output {} to determine"
                    " the outcome.".format(self.bsub_out))
                return "F"

        return "D"

    def _get_job_exit_code(self):
        '''Read the job script output to retrieve the exit code.

        Also read the error message if any, which will appear at the beginning
        of the log file followed by bsub's standard 'email' format output. It
        looks something like this:

            <stderr messages>
            ------------------------------------------------------------
            Sender: LSF System <...>
            Subject: ...
            ...

            Successfully completed.
            <OR>
            Exited with exit code 1.

            ...

        The extracted stderr messages are logged to self.fail_msg. The line
        indicating whether it was successful or it failed with an exit code
        is used to return the exit code.

        Returns the exit code if found, else None.
        '''

        # Job script output must have been opened already.
        assert self.bsub_out_fd

        for line in self.bsub_out_fd:
            if not self.bsub_out_err_msg_found:
                m = re.match("^Sender", line)
                if m:
                    # Pop the line before the sender line.
                    self.bsub_out_err_msg = "\n".join(
                        self.bsub_out_err_msg[:-1])
                    self.bsub_out_err_msg_found = True
                else:
                    self.bsub_out_err_msg.append(line.strip())

            else:
                m = re.match(r"^Exited with exit code (\d+).\n$", line)
                if m:
                    return m.group(1)

                if not self.bsub_out_err_msg:
                    m = re.match(r"^Successfully completed.\n$", line)
                    if m:
                        return 0
        return None

    def kill(self):
        if self.job_id:
            try:
                subprocess.run(["bkill", "-s", "SIGTERM", self.job_id],
                               check=True,
                               stdout=subprocess.PIPE,
                               stderr=subprocess.PIPE)
            except subprocess.CalledProcessError as e:
                log.error("Failed to kill job: {}".format(
                    e.stderr.decode("utf-8").strip()))
        else:
            log.error("Job ID for %s not found", self.deploy.full_name)

        self._post_finish('K', "Job killed!")

    def _post_finish(self, status, err_msg):
        if self.bsub_out_fd:
            self.bsub_out_fd.close()
        if self.exit_code is None:
            self.exit_code = 0 if status == 'P' else 1
        super()._post_finish(status, err_msg)

    @staticmethod
    def _post_finish_job_array(cfg, job_name, err_msg):
        '''On LSF error, mark all jobs in this array as killed.

        err_msg is the error message indicating the cause of failure.'''

        for job in LsfLauncher.jobs[cfg][job_name]:
            job._post_finish("F", ErrorMessage(line_number=None,
                                               message=err_msg,
                                               context=[]))