Source code for

import json
import os
import shutil

from import BuilderBase
from buildtest.cli.compilers import BuildtestCompilers
from buildtest.exceptions import BuildTestError
from import get_module_commands
from buildtest.utils.file import resolve_path
from import deep_get

[docs]class CompilerBuilder(BuilderBase): """This is a subclass of BuilderBase used for building test that uses ``type: compiler`` in the buildspec.""" type = "compiler" # Fortran Extensions Links: # # Fortran Extensions: lang_ext_table = { ".c": "C", ".cc": "C++", ".cxx": "C++", ".cpp": "C++", ".c++": "C++", ".f90": "Fortran", ".F90": "Fortran", ".f95": "Fortran", ".f03": "Fortran", ".f": "Fortran", ".F": "Fortran", ".FOR": "Fortran", ".for": "Fortran", ".FTN": "Fortran", ".ftn": "Fortran", } cc = None cxx = None fc = None ldflags = None cflags = None cxxflags = None fflags = None cppflags = None def __init__( self, name, recipe, buildspec, buildexecutor, executor, configuration, compiler=None, testdir=None, numprocs=None, numnodes=None, ): super().__init__( name=name, recipe=recipe, buildspec=buildspec, executor=executor, buildexecutor=buildexecutor, testdir=testdir, numprocs=numprocs, numnodes=numnodes, compiler=compiler, ) self.compiler = compiler self.configuration = configuration self.metadata["compiler"] = compiler self.compiler_section = self.recipe["compilers"] self.sourcefile = self.recipe["source"]
[docs] def setup(self): """The setup method is responsible for process compiler section, getting modules pre_build, post_build, pre_run, post_run section and generate compilation and run command. This method invokes other methods and set values in class variables. This method is called by self.generate_script method. """ self._resolve_source() self.lang = self._detect_lang(self.sourcefile) # set executable name and assign to self.executable self.executable = "%s.exe" % os.path.basename(self.sourcefile) self.exec_variable = f"_EXEC={self.executable}" self._process_compiler_config() self.metrics = deep_get( self.recipe, "executors", self.executor, "metrics" ) or self.recipe.get("metrics") # get environment variables self.envvars = ( deep_get(self.compiler_section, "config", self.compiler, "env") or deep_get(self.compiler_section, "default", self.compiler_group, "env") or deep_get(self.compiler_section, "default", "all", "env") ) # get environment variables self.vars = ( deep_get(self.compiler_section, "config", self.compiler, "vars") or deep_get(self.compiler_section, "default", self.compiler_group, "vars") or deep_get(self.compiler_section, "default", "all", "vars") ) # get status self.status = ( deep_get(self.compiler_section, "config", self.compiler, "status") or deep_get(self.compiler_section, "default", self.compiler_group, "status") or deep_get(self.compiler_section, "default", "all", "status") ) # compiler set in compilers 'config' section, we try to get module lines using self._get_modules self.modules = get_module_commands( deep_get(self.compiler_section, "config", self.compiler, "module") ) if not self.modules: self.modules = get_module_commands(self.bc_compiler.get("module")) self.pre_build = ( deep_get(self.compiler_section, "config", self.compiler, "pre_build") or deep_get( self.compiler_section, "default", self.compiler_group, "pre_build" ) or deep_get(self.compiler_section, "default", "all", "pre_build") ) self.post_build = ( deep_get(self.compiler_section, "config", self.compiler, "post_build") or deep_get( self.compiler_section, "default", self.compiler_group, "post_build" ) or deep_get(self.compiler_section, "default", "all", "post_build") ) self.pre_run = ( deep_get(self.compiler_section, "config", self.compiler, "pre_run") or deep_get( self.compiler_section, "default", self.compiler_group, "pre_run" ) or deep_get(self.compiler_section, "default", "all", "pre_run") ) self.post_run = ( deep_get(self.compiler_section, "config", self.compiler, "post_run") or deep_get( self.compiler_section, "default", self.compiler_group, "post_run" ) or deep_get(self.compiler_section, "default", "all", "post_run") ) self.compile_cmd = self._compile_cmd() self.run_cmd = self._run_cmd()
# Compiler schema needs 'source' key in order to compile test if 'build' is specified. Schema needs # 'binary' if one just wants to run test without compilation # if mode: [build, run] - compile and run test # if mode: [build ] - compile test # if mode: [run ] - run test
[docs] def generate_script(self): """This method is responsible for generating test script for compiler schema. The method ``generate_script`` is implemented in each subclass because implementation on test generation differs across schema types. This method will add the lines into list which comprise content of test. The method will return a list containing lines of test script. """ self.setup() # every test starts with shebang line lines = [self.shebang] batch_dict = {} cray_dict = {} # get sbatch, bsub, cobalt, pbs, batch property and store in batch dictionary. # The order of lookup is in order of precedence for batch in ["sbatch", "bsub", "cobalt", "pbs"]: batch_dict[batch] = ( deep_get(self.compiler_section, "config", self.compiler, batch) or deep_get( self.compiler_section, "default", self.compiler_group, batch ) or deep_get(self.compiler_section, "default", "all", batch) ) # setting these values override values from Builder.sched_init() method self.sbatch = batch_dict["sbatch"] self.bsub = batch_dict["bsub"] self.pbs = batch_dict["pbs"] self.cobalt = batch_dict["cobalt"] sched_lines = self.get_job_directives() if sched_lines: lines += sched_lines # get cray burst buffer (BB) and datawarp (DW) fields in order of precedence. for name in ["BB", "DW"]: cray_dict[name] = ( deep_get(self.compiler_section, "config", self.compiler, name) or deep_get(self.compiler_section, "default", self.compiler_group, name) or deep_get(self.compiler_section, "default", "all", name) ) burst_buffer_lines = self._get_burst_buffer(cray_dict["BB"]) if burst_buffer_lines: lines.append("### START OF BURST BUFFER DIRECTIVES ###") lines += burst_buffer_lines lines.append("### END OF BURST BUFFER DIRECTIVES ###") data_warp_lines = self._get_data_warp(cray_dict["DW"]) if data_warp_lines: lines.append("### START OF DATAWARP DIRECTIVES ###") lines += data_warp_lines lines.append("### END OF DATAWARP DIRECTIVES ###") lines.append("\n") lines.append(self._emit_set_command()) lines.append("# name of executable") lines += [self.exec_variable] env_lines = self._get_environment(self.envvars) if env_lines: lines += env_lines # get variables var_lines = self._get_variables(self.vars) if var_lines: lines += var_lines # if 'module' defined in Buildspec add modules to test if self.modules: lines.append("# Loading modules") lines += self.modules if self.pre_build: lines.append("### START OF PRE BUILD SECTION ###") lines.append(self.pre_build) lines.append("### END OF PRE BUILD SECTION ###") lines.append("\n") lines.append("# Compilation Line") lines.append(self.compile_cmd) lines.append("\n") if self.post_build: lines.append("### START OF POST BUILD SECTION ###") lines.append(self.post_build) lines.append("### END OF POST BUILD SECTION ###") lines.append("\n") if self.pre_run: lines.append("### START OF PRE RUN SECTION ###") lines.append(self.pre_run) lines.append("### END OF PRE RUN SECTION ###") lines.append("\n") # add run command lines.append("# Run executable") lines.append(self.run_cmd) lines.append("\n") if self.post_run: lines.append("### START OF POST RUN SECTION ###") lines.append(self.post_run) lines.append("### END OF POST RUN SECTION ###") lines.append("\n") return lines
[docs] def _resolve_source(self): """This method resolves full path to source file, it checks for absolute path first before checking relative path that is relative to Buildspec recipe. """ # attempt to resolve path based on 'source' field. # 1. The source file can be absolute path and if exists we use this # 2. The source file can be relative path to where buildspec is located self.abspath_sourcefile = resolve_path(self.sourcefile) or resolve_path( os.path.join(os.path.dirname(self.buildspec), self.sourcefile) ) # raise error if we can't find source file to compile if not self.abspath_sourcefile: raise BuildTestError( f"Failed to resolve path specified in field 'source': {self.sourcefile}" )
[docs] def _detect_lang(self, sourcefile): """This method will return the Programming Language based by looking up file extension of source file. """ self.logger.debug( f"[{}]: Detecting programming language for source file: {sourcefile}" ) ext = os.path.splitext(sourcefile)[1] self.logger.debug( f"Found file extension: {ext}, now we will attempt to lookup programming language based on extension" ) # if ext not in self.lang_ext_table then raise an error. This table consist of all file extensions that map to a Programming Language if ext not in self.lang_ext_table: raise BuildTestError( f"[{}]: Unable to detect Program Language based on extension: {ext} in source: {sourcefile}" ) # Set Programming Language based on ext. Programming Language could be (C, C++, Fortran) lang = self.lang_ext_table[ext] self.logger.debug( f"[{}]: Based on extension: {ext} the programming language is: {lang}" ) return lang
[docs] def _compile_cmd(self): """This method generates the compilation line and returns the output as a list. The compilation line depends on the the language detected that is stored in variable ``self.lang``. """ cmd = [] # Generate C compilation line if self.lang == "C": cmd = [, self.cppflags, self.cflags, "-o $_EXEC", self.abspath_sourcefile, self.ldflags, ] # Generate C++ compilation line elif self.lang == "C++": cmd = [ self.cxx, self.cppflags, self.cxxflags, "-o $_EXEC", self.abspath_sourcefile, self.ldflags, ] # Generate Fortran compilation line elif self.lang == "Fortran": cmd = [ self.fc, self.cppflags, self.fflags, "-o $_EXEC", self.abspath_sourcefile, self.ldflags, ] # remove any None from list cmd = list(filter(None, cmd)) cmd = " ".join(cmd) return cmd
[docs] def _run_cmd(self): """This method builds the run command which refers to how to run the generated binary after compilation. """ # order of precedence on how to generate run line when executing binary. # 1. Check in 'config' section within compiler # 2. Check in 'default' section within compiler group # 3. Check in 'default' section within 'all' section # 4. Last resort run binary standalone run_line = ( deep_get(self.compiler_section, "config", self.compiler, "run") or deep_get(self.compiler_section, "default", self.compiler_group, "run") or deep_get(self.compiler_section, "default", "all", "run") or "./$_EXEC" ) return run_line
[docs] def _process_compiler_config(self): """This method is responsible for setting cc, fc, cxx class variables based on compiler selection. The order of precedence is ``config``, ``default``, then buildtest setting. Compiler settings in 'config' takes highest precedence, this overrides any configuration in 'default'. Finally we resort to compiler configuration in buildtest setting if none defined. This method is responsible for setting cc, fc, cxx, cflags, cxxflags, fflags, ldflags, and cppflags. """ bc = BuildtestCompilers(configuration=self.configuration) self.compiler_group = bc.compiler_name_to_group[self.compiler] self.logger.debug( f"[{}]: compiler: {self.compiler} belongs to compiler group: {self.compiler_group}" ) # compiler from buildtest settings self.bc_compiler = self.configuration.target_config["compilers"]["compiler"][ self.compiler_group ][self.compiler] self.logger.debug(self.bc_compiler) # set compiler values based on 'default' property in buildspec. This can override # compiler setting defined in configuration file. If default is not set we load from buildtest settings for appropriate compiler. # set compiler variables to ones defined in buildtest configuration = self.bc_compiler["cc"] self.cxx = self.bc_compiler["cxx"] self.fc = self.bc_compiler["fc"] self.logger.debug( f"[{}]: Compiler setting for {self.compiler} from configuration file" ) self.logger.debug( f"[{}]: {self.compiler}: {json.dumps(self.bc_compiler, indent=2)}" ) # if default compiler setting provided in buildspec let's assign it. if deep_get(self.compiler_section, "default", self.compiler_group): = ( self.compiler_section["default"][self.compiler_group].get("cc") or ) self.fc = ( self.compiler_section["default"][self.compiler_group].get("fc") or self.fc ) self.cxx = ( self.compiler_section["default"][self.compiler_group].get("cxx") or self.cxx ) self.cflags = self.compiler_section["default"][self.compiler_group].get( "cflags" ) self.cxxflags = self.compiler_section["default"][self.compiler_group].get( "cxxflags" ) self.fflags = self.compiler_section["default"][self.compiler_group].get( "fflags" ) self.ldflags = self.compiler_section["default"][self.compiler_group].get( "ldflags" ) self.cppflags = self.compiler_section["default"][self.compiler_group].get( "cppflags" ) # if compiler instance defined in config section read from buildspec. This overrides default section if specified if deep_get(self.compiler_section, "config", self.compiler): self.logger.debug( f"[{}]: Detected compiler: {self.compiler} in 'config' scope overriding default compiler group setting for: {self.compiler_group}" ) = ( self.compiler_section["config"][self.compiler].get("cc") or ) self.fc = ( self.compiler_section["config"][self.compiler].get("fc") or self.fc ) self.cxx = ( self.compiler_section["config"][self.compiler].get("cxx") or self.cxx ) self.cflags = ( self.compiler_section["config"][self.compiler].get("cflags") or self.cflags ) self.cxxflags = ( self.compiler_section["config"][self.compiler].get("cxxflags") or self.cxxflags ) self.fflags = ( self.compiler_section["config"][self.compiler].get("fflags") or self.fflags ) self.cppflags = ( self.compiler_section["config"][self.compiler].get("cppflags") or self.cppflags ) self.ldflags = ( self.compiler_section["config"][self.compiler].get("ldflags") or self.ldflags ) self.logger.debug( f"cc: {}, cxx: {self.cxx} fc: {self.fc} cppflags: {self.cppflags} cflags: {self.cflags} fflags: {self.fflags} ldflags: {self.ldflags}" ) # this condition is a safety check before compiling code to ensure if all C, C++, Fortran compiler not set we raise error if not and not self.cxx and not self.fc: raise BuildTestError( "Unable to set C, C++, and Fortran compiler wrapper, please specify 'cc', 'cxx','fc' in your compiler settings in buildtest configuration or specify in buildspec file. " )
[docs] def set_cc(self, cc): = cc
[docs] def set_cxx(self, cxx): self.cxx = cxx
[docs] def set_fc(self, fc): self.fc = fc
[docs] def set_cflags(self, cflags): self.cflags = cflags
[docs] def set_fflags(self, fflags): self.fflags = fflags
[docs] def set_cxxflags(self, cxxflags): self.cxxflags = cxxflags
[docs] def set_cppflags(self, cppflags): self.cppflags = cppflags
[docs] def set_ldflags(self, ldflags): self.ldflags = ldflags
[docs] def get_cc(self): return
[docs] def get_cxx(self): return self.cxx
[docs] def get_fc(self): return self.fc
[docs] def get_cflags(self): return self.cflags
[docs] def get_cxxflags(self): return self.cxxflags
[docs] def get_fflags(self): return self.fflags
[docs] def get_cppfilags(self): return self.cppflags
[docs] def get_ldflags(self): return self.ldflags
[docs] def get_path(self): """This method returns the full path for C, C++, Fortran compilers""" path = { shutil.which(, self.cxx: shutil.which(self.cxx), self.fc: shutil.which(self.fc), } return path