undertaker 1.2-1 / usr / bin / undertaker-calc-coverage

#!/usr/bin/env python

import os
import shutil
import sys
import glob
import re
import shlex
import logging
from subprocess import *

class ModelReader(dict):
    def __init__(self, fd):

        self.meta_info = dict()

        for l in fd:
            if l.startswith('I: '):
                continue
            if l.startswith('UNDERTAKER_SET'):
                row = shlex.split(l)[1:]
                if len(row) > 0:
                    self.meta_info[row[0]] = row[1:]
                continue
            row = shlex.split(l)
            if len(row) > 0:
                self[row[0]] = row[1:]

model = None

std_configs = ["allyesconfig"]

class ExpansionError(RuntimeError):
    """ Base class of all sort of Expansion errors """
    pass

class ExpansionSanityCheckError(ExpansionError):
    """ Internal kernel config sanity checks failed, like `make silentoldconfig` """
    pass

class ZizlerError(RuntimeError):
    pass

def execute(command, echo=True):
    """
    executes 'command' in a shell

    returns a tuple with 
     1. the command's standard output as list of lines
     2. the exitcode
    """

    if echo:
        logging.debug("executing: %s" % command)
    p = Popen(command, stdout=PIPE, stderr=STDOUT, shell=True)
    (stdout, stderr) = p.communicate()
    return (stdout.rsplit('\n'), p.returncode)

def switch_config(config):
    """
    switches the current tree to the given config. throws an exception in case make fails
    """
    execute("find include -name autoconf.h -exec rm {} +", echo = True)

    (output, returncode) = execute("make %s" % config, echo=False)
    if returncode != 0:
        raise RuntimeError("Failed to switch to config %s" % config)
    (out2, ret2) = execute("make silentoldconfig", echo=False)
    if ret2 != 0:
        raise RuntimeError("silentoldconfig failed while switching to config %s" % config)
    return output

def switch_config_path(filename):
    """
    similar to switch_config, but takes a filename instead of a config target

    As sideffect, the current .config file is overwritten
    """

    # now replace the old .config with our 'expanded' one
    shutil.copyfile(filename, '.config')
    arch = "x86"
    m = re.search("arch/([^/]*)/", filename)
    if m:
        arch = m.group(1)
    satconf_cmd = 'yes "" | make oldconfig ARCH="%s" KERNELVERSION=`git describe`' % arch

    (output, returncode) = execute(satconf_cmd, echo=False)
    if returncode != 0:
        raise ExpansionError("command used: " + satconf_cmd)

    cmd = 'yes "" | make silentoldconfig ARCH="%s" KERNELVERSION=`git describe`' % arch
    (out2, ret2) = execute(cmd, echo=False)
    if ret2 != 0:
        # with the satconfig approach, this always worked
        raise ExpansionSanityCheckError("silentoldconfig failed while switching to config %s (rc: %d)" % (filename, ret2))

def call_sparse(filename):
    """
    run sparse with current configuration

    @return output of sparse
    """

    logging.debug("Creating %s" % filename + ".sparse")

    coverage_cmd = "make %s C=2" % filename.replace(".c", ".o")
    p = Popen(coverage_cmd, stdout=PIPE, stderr=STDOUT, shell=True)
    (stdout, dummy) = p.communicate()

    return stdout

def get_covered_lines(filename):
    coverage_cmd = "make %s C=2 CHECK=sparse-coverage" % filename.replace(".c", ".o")
    (sparse_coverage, returncode) = execute(coverage_cmd, echo=False)

    lines = []
    for line in sparse_coverage:
        tokens = line.split(" ")
        if tokens[0] != "" and tokens[0] in filename:
            lines.append(int(tokens[1]))
    return lines

def default_configs(filename, results):

    if not filename in results:
        results[filename] = {}

    for config in std_configs:
        logging.info("Switching to configuration preset %s" % config)
        switch_config(config)
        results[filename][config] = len(get_conditional_blocks(filename, selected_only=True))
    return results

# e.g. taken from the makefiles
extra_dependencies = {
    './drivers/cpuidle/cpuidle.c': ["CONFIG_CPU_IDLE=y", "CONFIG_ARCH_HAS_DEFAULT_IDLE=y"],
}

def expand_by_copy(config):
    logging.debug("Trying to expand configuration %s" % config)

    if not os.path.exists(config):
        raise RuntimeError("Partial configuration %s does not exist" % config)

    shutil.copy(config, config + '.config')
    return config + '.config'


def expand_by_filtering(config):
    """
    configuration 'config' is a partial configuration, i.e., it
    contains an incomplete set of config options. The strategy
    here is to start with a default config ("allnoconfig"), and
    then filter out configs for which we have an entry in the
    partial selection. Finally, we verify and complete the
    "hacked" selection with "make oldconfig"

    @return the path to the expanded configuration
    """

    logging.debug("Trying to expand configuration %s" % config)

    if not os.path.exists(config):
        raise RuntimeError("Partial configuration %s does not exist" % config)

    switch_config('allnoconfig')

    with open(config + '.config', 'w+') as newconfigf:
        with open('.config', 'r') as oldconfigf:
            with open(config) as configf:
                found_items = []
                for l in configf:
                    if l.startswith('CONFIG_'):
                        item = l.split('=')[0]
                        found_items.append(item)

                for l in oldconfigf:
                    item = l.split('=')[0]
                    if item not in found_items:
                        newconfigf.write(l)

#            if filename in extra_dependencies.keys():
#                newconfigf.write("\n".join(extra_dependencies[filename]))

    return config + '.config'

def verify_config(partial_config):
    """
    verifies that the current .config file satisfies the constraints of the
    given partial configuration.

    @return a dictionary with items that violate the partial selection
    """

    value_errors = {}
    # magic values induced by our models
    config_whitelist = ('CONFIG_n', 'CONFIG_y', 'CONFIG_m')

    with open(partial_config) as partial_configf:
        for l in partial_configf:
            if l.startswith('CONFIG_'):
                (item, value) = l.split('=')
                if model and item in model.meta_info['ALWAYS_ON']:
                    continue
                if item in config_whitelist:
                    continue
                value_errors[item] = value.rstrip()

    with open('.config') as configf:
        found_items = {}
        for l in configf:
            if l.startswith('CONFIG_'):
                (item, value) = l.split('=')
                found_items[item] = value.rstrip()
                if model and item in model.meta_info['ALWAYS_ON']:
                    continue
                if (item in value_errors):
                    if (value_errors[item] == value.rstrip()):
                        del value_errors[item]

        for (item, value) in value_errors.items():
            if value == 'n':
                if not item in found_items:
                    del value_errors[item]

    return value_errors

def selected_items(partial_config):
    """
    counts how many items got selected by the given partial config
    """
    selected = 0 

    with open(partial_config) as partial_configf:
        for l in partial_configf:
            if l.startswith('CONFIG_'):
                (item, value) = l.split('=')
                if value != 'n':
                    selected += 1

    return selected
    
autoconf = None
def find_autoconf():
    """ returns the path to the autoconf.h file in this linux tree
    """
    global autoconf
    if autoconf == None:
        switch_config('allyesconfig')
        (autoconf, rc) = execute("find include -name autoconf.h", echo = True)
        autoconf = filter(lambda x: len(x) > 0, autoconf)
        if rc != 0 or len(autoconf) != 1:
            logging.error("Not exactly one autoconf.h was found (%s)" % ", ".join(autoconf))
            raise RuntimeError("Not exactly one autoconf.h was found")
    return autoconf[0]

def get_conditional_blocks(filename, selected_only=False):
    """
    Counts the conditional blocks in the given source file

    The calculation is done using the 'zizler' binary from the system
    path.  If the parameter 'actually_selected' is set, then the source
    file is preprocessed with 'cpp' while using current configuration as
    per 'include/generated/autoconf.h'

    @return a non-empty list of blocks found in the source file
    
    """

    if selected_only:
        cmd = 'zizler -cC "%s" | cpp -include %s' % (filename, find_autoconf())
    else:
        cmd = 'zizler -cC "%s"' % filename
    (blocks, rc) = execute(cmd, echo=True)

    blocks = filter(lambda x: len(x) != 0 and x.startswith("B"), blocks)
    if rc != 0:
        logging.warning("zizler/cpp failed with exitcode: %d" % rc)
    return blocks


def get_loc_coverage(filename):
    """
    Returns LOC of the given file taking the current configuration into account
    """

    cmd = "grep -v -E '^\s*#\s*include' %s | cpp -include %s" % (filename, find_autoconf())
    (lines, rc) = execute(cmd, echo=True)
    return len(lines)


def vamos_coverage(filename, results):
    covered = set()
    found_configs = configurations(filename)
    logging.info("found %d configurations for %s" % (len(found_configs), filename))

    if not filename in results:
        results[filename] = {}

    for c in found_configs:
        logging.debug(c)

    results[filename]['fail_count'] = 0

    for config in found_configs:
        expanded_config = expand_by_copy(config)
        results[filename][config] = {}

        try:
            switch_config_path(expanded_config)
        except ExpansionError as error:
            logging.error("Config %s failed to apply, skipping" % config)
            results[filename]['fail_count'] = 1 + results[filename]['fail_count']
            logging.error(error)
            continue
        except ExpansionSanityCheckError as error:
            results[filename]['fail_count'] = 1 + results[filename]['fail_count']
            logging.warning("sanity check failed, but continuing anyways")
            logging.warning(error)

        value_errors = verify_config(config)
        if len(value_errors) > 0:
            logging.info("Failed to set %d items from partial config %s" % (len(value_errors), config))
            mismatches = list()
            for c in value_errors.keys():
                mismatches.append("%s != %s" % (c, value_errors[c]))
            #logging.debug("".join(mismatches))

        results[filename][config]['value_errors'] = value_errors
        results[filename][config]['selected_items'] = selected_items(config)

        # writeback the expanded configuration
        shutil.copy('.config', config + '.config')

        # now check what blocks are actually set
        old_len = len(covered)
        covered_blocks = set(get_conditional_blocks(filename, selected_only=True))
        covered |= covered_blocks
        logging.info("Config %s added %d additional blocks" % (config, len(covered) - old_len))
        results[filename][config]['covered_blocks'] = covered_blocks


    total_blocks = results[filename]['total_blocks']
    results[filename]['uncovered_blocks'] = ", ".join(total_blocks - covered)
    results[filename]['vamos'] = len(covered)
    return results

def configurations(filename):
    l = glob.glob(filename + ".config[0-9]")
    l += (glob.glob(filename + ".config[0-9][0-9]"))
    return sorted(l)


def main():
    from optparse import OptionParser

    parser = OptionParser()
    parser.add_option("-u", "--run-undertaker", dest="do_undertaker",
                      action="store_true", default=False,
                      help="run undertaker to generate partial configs (default: no)")
    parser.add_option("-s", "--skip-configs", dest="skip_configs",
                      action="store_true", default=False,
                      help="Skip analyzing configurations (defaut: use existing configs)")
    parser.add_option('-v', '--verbose', dest='verbose', action='count',
                      help="Increase verbosity (specify multiple times for more)")
    parser.add_option('-m', '--model', dest='model', action='store',
                      help="load the model and configuration was generated with model")
    parser.add_option('-F', '--fast', dest='min_strategy', action='store_false', default=True,
                      help="Use a faster (but maybe more naive) strategy for coverage analysis, (implies -u, default: no)")

    parser.add_option("", '--strategy', dest='strategy', action='store', type="choice",
                      choices=["simple", "minimal"], default="minimal",
                      help="Configuration was generated by $strategy. This will only effect the direct output")
    parser.add_option("", '--expanded-config', dest='expanded_config', action='store_true', default=False,
                      help="Configuration was generated on expanded source file. This will only effect the direct output")
    parser.add_option("", '--expanded-source', dest='expanded_source', action='store_true', default=False,
                      help="Configuration is tested on expaned source file. This will only effect the direct output")

    (options, args) = parser.parse_args()

    log_level = logging.WARNING # default
    if options.verbose == 1:
        log_level = logging.INFO
    elif options.verbose >= 2:
        log_level = logging.DEBUG

    logging.basicConfig(level=log_level)

    if not options.min_strategy: options.do_undertaker = True

    if options.model and os.path.exists(options.model):
        with open(options.model) as fd:
            global model
            model = ModelReader(fd)
            logging.info("Loaded %d options from Model %s" % (len(model), options.model))
            logging.info("%d items are always on" % len(model.meta_info['ALWAYS_ON']))

    if len(args) == 0:
        logging.critical("please specify a worklist")
        sys.exit(1)

    worklist = args[0]
    try:
        with open(worklist) as f:
            filenames = [x.strip() for x in f.readlines()]
    except IOError as error:
        logging.critical("failed to open worklist: " + error.__str__())
        sys.exit(1)

    logging.debug("Worklist contains %d items" % len(filenames))

    if (options.do_undertaker):
        model_param = ""
        strategy = ""
        if (options.min_strategy):
            strategy = "-C min"
        if not options.model:
            logging.info("Running without model")
        else:
            model_param = "-m %s" % options.model
        execute("undertaker -j coverage %s %s -b %s" % (model_param, strategy, worklist), echo=True)
    
    results = {}

    for filename in filenames:
        if not os.path.exists(filename):
            if filename.endswith('.pi'):
                # we're working on partially preprocessed files, replace
                # them here with the actual files
                base = filename[:-3]
                logging.debug("working on %s instead of %s" % ( base + '.c', filename))
                filename = base + '.c'
            else:
                logging.debug("%s does not exist, skipping" % filename)
                continue

        logging.info("processing %s" % filename)

        default_configs(filename, results)
        results[filename]['total_blocks'] = set(get_conditional_blocks(filename))

        print "\nRESULT"
        print "filename: %s" % filename
        print "expanded: %s" % options.expanded_config
        print "expanded_source: %s" % options.expanded_source
        print "kconfig: %s"  % (options.model is not None)
        print "min_strategy: %s" % (options.strategy == "minimal")
        print "loc: %d" % get_loc_coverage(filename)
        print "total_blocks: %d"     % len(results[filename]['total_blocks'])
        print "allyes_coverage: %d" % results[filename]['allyesconfig']

        try:
            if options.skip_configs:
                continue

            if len(results[filename]['total_blocks']) == 0:
                continue

            vamos_coverage(filename, results)
            print "covered_blocks: %d"   % results[filename]['vamos']
            print "uncovered_blocks: %s" % results[filename]['uncovered_blocks']
            print "expansion_errors: %d" % results[filename]['fail_count']

            for config in results[filename].keys():
                if config.startswith(filename):                
                    print "\nCONFIG"
                    print "config_path: %s" % config
                    try:
                        print "item_mismatches: %d" % len(results[filename][config]['value_errors'])
                        print "wrong_items: %s"     % ", ".join(results[filename][config]['value_errors'])
                        print "selected_items: %d"  % results[filename][config]['selected_items']
                        print "runtime: 0.0"
                        print "covered_blocks: %d"  % len(results[filename][config]['covered_blocks'])
                    except KeyError:
                        pass

        except ZizlerError as error:
            continue
        except RuntimeError as error:
            logging.error(error.__str__())
            logging.error("Skipping file " + filename)
            continue
    
if __name__ == "__main__":
    main()