
This is an ongoing series of commits that are reformatting our Python code. Reformatting is done with `black`. If you end up having problems merging this commit because you have made changes to a python file, the best way to handle that is to run git checkout --ours <yourfile> and then reformat it with black. If you run into any problems, post to discourse about it and we will try to help. RFC Thread below: https://discourse.llvm.org/t/rfc-document-and-standardize-python-code-style Reviewed By: MatzeB Differential Revision: https://reviews.llvm.org/D150761
216 lines
7.1 KiB
Python
216 lines
7.1 KiB
Python
# -*- 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),
|
|
}
|
|
)
|
|
}
|