# -*- coding: utf-8 -*- # Part of the LLVM Project, under the Apache License v2.0 with LLVM Exceptions. # See https://llvm.org/LICENSE.txt for license information. # SPDX-License-Identifier: Apache-2.0 WITH LLVM-exception """ This module is a collection of methods commonly used in this project. """ import collections import functools import json import logging import os import os.path import re import shlex import subprocess import sys ENVIRONMENT_KEY = 'INTERCEPT_BUILD' Execution = collections.namedtuple('Execution', ['pid', 'cwd', 'cmd']) CtuConfig = collections.namedtuple('CtuConfig', ['collect', 'analyze', 'dir', 'extdef_map_cmd']) def duplicate_check(method): """ Predicate to detect duplicated entries. Unique hash method can be use to detect duplicates. Entries are represented as dictionaries, which has no default hash method. This implementation uses a set datatype to store the unique hash values. This method returns a method which can detect the duplicate values. """ def predicate(entry): entry_hash = predicate.unique(entry) if entry_hash not in predicate.state: predicate.state.add(entry_hash) return False return True predicate.unique = method predicate.state = set() return predicate def run_build(command, *args, **kwargs): """ Run and report build command execution :param command: array of tokens :return: exit code of the process """ environment = kwargs.get('env', os.environ) logging.debug('run build %s, in environment: %s', command, environment) exit_code = subprocess.call(command, *args, **kwargs) logging.debug('build finished with exit code: %d', exit_code) return exit_code def run_command(command, cwd=None): """ Run a given command and report the execution. :param command: array of tokens :param cwd: the working directory where the command will be executed :return: output of the command """ def decode_when_needed(result): """ check_output returns bytes or string depend on python version """ return result.decode('utf-8') if isinstance(result, bytes) else result try: directory = os.path.abspath(cwd) if cwd else os.getcwd() logging.debug('exec command %s in %s', command, directory) output = subprocess.check_output(command, cwd=directory, stderr=subprocess.STDOUT) return decode_when_needed(output).splitlines() except subprocess.CalledProcessError as ex: ex.output = decode_when_needed(ex.output).splitlines() raise ex def reconfigure_logging(verbose_level): """ Reconfigure logging level and format based on the verbose flag. :param verbose_level: number of `-v` flags received by the command :return: no return value """ # Exit when nothing to do. if verbose_level == 0: return root = logging.getLogger() # Tune logging level. level = logging.WARNING - min(logging.WARNING, (10 * verbose_level)) root.setLevel(level) # Be verbose with messages. if verbose_level <= 3: fmt_string = '%(name)s: %(levelname)s: %(message)s' else: fmt_string = '%(name)s: %(levelname)s: %(funcName)s: %(message)s' handler = logging.StreamHandler(sys.stdout) handler.setFormatter(logging.Formatter(fmt=fmt_string)) root.handlers = [handler] def command_entry_point(function): """ Decorator for command entry methods. The decorator initialize/shutdown logging and guard on programming errors (catch exceptions). The decorated method can have arbitrary parameters, the return value will be the exit code of the process. """ @functools.wraps(function) def wrapper(*args, **kwargs): """ Do housekeeping tasks and execute the wrapped method. """ try: logging.basicConfig(format='%(name)s: %(message)s', level=logging.WARNING, stream=sys.stdout) # This hack to get the executable name as %(name). logging.getLogger().name = os.path.basename(sys.argv[0]) return function(*args, **kwargs) except KeyboardInterrupt: logging.warning('Keyboard interrupt') return 130 # Signal received exit code for bash. except Exception: logging.exception('Internal error.') if logging.getLogger().isEnabledFor(logging.DEBUG): logging.error("Please report this bug and attach the output " "to the bug report") else: logging.error("Please run this command again and turn on " "verbose mode (add '-vvvv' as argument).") return 64 # Some non used exit code for internal errors. finally: logging.shutdown() return wrapper def compiler_wrapper(function): """ Implements compiler wrapper base functionality. A compiler wrapper executes the real compiler, then implement some functionality, then returns with the real compiler exit code. :param function: the extra functionality what the wrapper want to do on top of the compiler call. If it throws exception, it will be caught and logged. :return: the exit code of the real compiler. The :param function: will receive the following arguments: :param result: the exit code of the compilation. :param execution: the command executed by the wrapper. """ def is_cxx_compiler(): """ Find out was it a C++ compiler call. Compiler wrapper names contain the compiler type. C++ compiler wrappers ends with `c++`, but might have `.exe` extension on windows. """ wrapper_command = os.path.basename(sys.argv[0]) return re.match(r'(.+)c\+\+(.*)', wrapper_command) def run_compiler(executable): """ Execute compilation with the real compiler. """ command = executable + sys.argv[1:] logging.debug('compilation: %s', command) result = subprocess.call(command) logging.debug('compilation exit code: %d', result) return result # Get relevant parameters from environment. parameters = json.loads(os.environ[ENVIRONMENT_KEY]) reconfigure_logging(parameters['verbose']) # Execute the requested compilation. Do crash if anything goes wrong. cxx = is_cxx_compiler() compiler = parameters['cxx'] if cxx else parameters['cc'] result = run_compiler(compiler) # Call the wrapped method and ignore it's return value. try: call = Execution( pid=os.getpid(), cwd=os.getcwd(), cmd=['c++' if cxx else 'cc'] + sys.argv[1:]) function(result, call) except: logging.exception('Compiler wrapper failed complete.') finally: # Always return the real compiler exit code. return result def wrapper_environment(args): """ Set up environment for interpose compiler wrapper.""" return { ENVIRONMENT_KEY: json.dumps({ 'verbose': args.verbose, 'cc': shlex.split(args.cc), 'cxx': shlex.split(args.cxx) }) }