coalib.processes.simplify_section_result() - Code Metrics - Inspection of "Minor tweaks in circleci" - coala-analyzer/coala - Measure and Improve Code Quality continuously with Scrutinizer

Completed

Pull Request — master (#1487)

by Lasse

created 2016-02-15 11:22 UTC

coalib.processes.simplify_section_result() B

↳ Parent: Project

Complexity

Conditions

Size

Total Lines

Duplication

Lines	0
Ratio	0 %

Metric	Value
cc	4
dl	0
loc	25
rs	8.5806

from itertools import chain
import multiprocessing
import queue
import os
import platform
import subprocess

from coalib.collecting.Collectors import collect_files
from coalib.collecting import Dependencies
from coalib.misc.StringConverter import StringConverter
from coalib.output.printers import LOG_LEVEL
from coalib.processes.BearRunning import run
from coalib.processes.CONTROL_ELEMENT import CONTROL_ELEMENT
from coalib.results.Result import Result
from coalib.results.RESULT_SEVERITY import RESULT_SEVERITY
from coalib.results.SourceRange import SourceRange
from coalib.settings.Setting import path_list
from coalib.processes.LogPrinterThread import LogPrinterThread
from coalib.results.result_actions.ApplyPatchAction import ApplyPatchAction
from coalib.results.result_actions.PrintDebugMessageAction import (
    PrintDebugMessageAction)
from coalib.results.result_actions.ShowPatchAction import ShowPatchAction


ACTIONS = [ApplyPatchAction,
           PrintDebugMessageAction,
           ShowPatchAction]


def get_cpu_count():
    try:
        return multiprocessing.cpu_count()
    # cpu_count is not implemented for some CPU architectures/OSes
    except NotImplementedError:  # pragma: no cover
        return 2


def fill_queue(queue_fill, any_list):
    """
    Takes element from a list and populates a queue with those elements.

    :param queue_fill: The queue to be filled.
    :param any_list:   List containing the elements.
    """
    for elem in any_list:
        queue_fill.put(elem)


def get_running_processes(processes):
    return sum((1 if process.is_alive() else 0) for process in processes)


def create_process_group(command_array, **kwargs):
    if platform.system() == "Windows":  # pragma: no cover
        proc = subprocess.Popen(
            command_array,
            creationflags=subprocess.CREATE_NEW_PROCESS_GROUP,
            **kwargs)
    else:
        proc = subprocess.Popen(command_array,
                                preexec_fn=os.setsid,
                                **kwargs)
    return proc


def get_default_actions(section):
    """
    Parses the key `default_actions` in the given section.

    :param section:    The section where to parse from.
    :return:           A dict with the bearname as keys and their default
                       actions as values and another dict that contains bears
                       and invalid action names.
    """
    try:
        default_actions = dict(section["default_actions"])
    except IndexError:
        return {}, {}

    action_dict = {action.get_metadata().name: action for action in ACTIONS}
    invalid_action_set = default_actions.values() - action_dict.keys()
    invalid_actions = {}
    if len(invalid_action_set) != 0:
        invalid_actions = {
            bear: action
            for bear, action in default_actions.items()
            if action in invalid_action_set}
        for invalid in invalid_actions.keys():
            del default_actions[invalid]

    actions = {bearname: action_dict[action_name]
               for bearname, action_name in default_actions.items()}
    return actions, invalid_actions


def autoapply_actions(results,
                      file_dict,
                      file_diff_dict,
                      section,
                      log_printer):
    """
    Auto-applies actions like defined in the given section.

    :param results:        A list of results.
    :param file_dict:      A dictionary containing the name of files and its
                           contents.
    :param file_diff_dict: A dictionary that contains filenames as keys and
                           diff objects as values.
    :param section:        The section.
    :param log_printer:    A log printer instance to log messages on.
    :return:               A list of unprocessed results.
    """

    default_actions, invalid_actions = get_default_actions(section)

    for bearname, actionname in invalid_actions.items():
        log_printer.warn("Selected default action {} for bear {} does "
                         "not exist. Ignoring action.".format(
                             repr(actionname),
                             repr(bearname)))

    if len(default_actions) == 0:
        # There's nothing to auto-apply.
        return results

    not_processed_results = []
    for result in results:
        try:
            action = default_actions[result.origin]
        except KeyError:
            not_processed_results.append(result)
            continue

        if not action.is_applicable(result, file_dict, file_diff_dict):
            log_printer.warn("Selected default action {} for bear {} is not "
                             "applicable. Action not applied.".format(
                                 repr(action.get_metadata().name),
                                 repr(result.origin)))
            not_processed_results.append(result)
            continue

        try:
            action().apply_from_section(result,
                                        file_dict,
                                        file_diff_dict,
                                        section)
            log_printer.info("Applied {} on {} from {}.".format(
                repr(action.get_metadata().name),
                result.location_repr(),
                repr(result.origin)))
        except Exception as ex:
            not_processed_results.append(result)
            log_printer.log_exception(
                "Failed to execute action {} with error: {}.".format(
                    repr(action.get_metadata().name),
                    ex),
                ex)
            log_printer.debug("-> for result " + repr(result) + ".")

    return not_processed_results


def print_result(results,
                 file_dict,
                 retval,
                 print_results,
                 section,
                 log_printer,
                 file_diff_dict,
                 ignore_ranges):
    """
    Takes the results produced by each bear and gives them to the print_results
    method to present to the user.

    :param results:        A list of results.
    :param file_dict:      A dictionary containing the name of files and its
                           contents.
    :param retval:         It is True if no results were yielded ever before.
                           If it is False this function will return False no
                           matter what happens. Else it depends on if this
                           invocation yields results.
    :param print_results:  A function that prints all given results appropriate
                           to the output medium.
    :param file_diff_dict: A dictionary that contains filenames as keys and
                           diff objects as values.
    :param ignore_ranges:  A list of SourceRanges. Results that affect code in
                           any of those ranges will be ignored.
    :return:               Returns False if any results were yielded. Else
                           True.
    """
    min_severity_str = str(section.get('min_severity', 'INFO')).upper()
    min_severity = RESULT_SEVERITY.str_dict.get(min_severity_str, 'INFO')
    results = list(filter(lambda result:
                          type(result) is Result and
                          result.severity >= min_severity and
                          not result.to_ignore(ignore_ranges),
                          results))

    if bool(section.get('autoapply', 'true')):
        patched_results = autoapply_actions(results,
                                            file_dict,
                                            file_diff_dict,
                                            section,
                                            log_printer)
    else:
        patched_results = results

    print_results(log_printer,
                  section,
                  patched_results,
                  file_dict,
                  file_diff_dict)
    return retval or len(results) > 0, patched_results


def get_file_dict(filename_list, log_printer):
    """
    Reads all files into a dictionary.

    :param filename_list: List of names of paths to files to get contents of.
    :param log_printer:   The logger which logs errors.
    :return:              Reads the content of each file into a dictionary
                          with filenames as keys.
    """
    file_dict = {}
    for filename in filename_list:
        try:
            with open(filename, "r", encoding="utf-8") as _file:
                file_dict[filename] = tuple(_file.readlines())
        except UnicodeDecodeError:
            log_printer.warn("Failed to read file '{}'. It seems to contain "
                             "non-unicode characters. Leaving it "
                             "out.".format(filename))
        except Exception as exception:  # pragma: no cover
            log_printer.log_exception("Failed to read file '{}' because of "
                                      "an unknown error. Leaving it "
                                      "out.".format(filename),
                                      exception,
                                      log_level=LOG_LEVEL.WARNING)

    return file_dict


def filter_raising_callables(it, exception, *args, **kwargs):
    """
    Filters all callable items inside the given iterator that raise the
    given exceptions.

    :param it:        The iterator to filter.
    :param exception: The (tuple of) exception(s) to filter for.
    :param args:      Positional arguments to pass to the callable.
    :param kwargs:    Keyword arguments to pass to the callable.
    """
    for elem in it:
        try:
            yield elem(*args, **kwargs)
        except exception:
            pass


def instantiate_bears(section,
                      local_bear_list,
                      global_bear_list,
                      file_dict,
                      message_queue):
    """
    Instantiates each bear with the arguments it needs.

    :param section:          The section the bears belong to.
    :param local_bear_list:  List of local bear classes to instantiate.
    :param global_bear_list: List of global bear classes to instantiate.
    :param file_dict:        Dictionary containing filenames and their
                             contents.
    :param message_queue:    Queue responsible to maintain the messages
                             delivered by the bears.
    :return:                 The local and global bear instance lists.
    """
    local_bear_list = [bear
                       for bear in filter_raising_callables(
                           local_bear_list,
                           RuntimeError,
                           section,
                           message_queue,
                           timeout=0.1)]

    global_bear_list = [bear
                        for bear in filter_raising_callables(
                            global_bear_list,
                            RuntimeError,
                            file_dict,
                            section,
                            message_queue,
                            timeout=0.1)]

    return local_bear_list, global_bear_list


def instantiate_processes(section,
                          local_bear_list,
                          global_bear_list,
                          job_count,
                          log_printer):
    """
    Instantiate the number of processes that will run bears which will be
    responsible for running bears in a multiprocessing environment.

    :param section:          The section the bears belong to.
    :param local_bear_list:  List of local bears belonging to the section.
    :param global_bear_list: List of global bears belonging to the section.
    :param job_count:        Max number of processes to create.
    :param log_printer:      The log printer to warn to.
    :return:                 A tuple containing a list of processes,
                             and the arguments passed to each process which are
                             the same for each object.
    """
    filename_list = collect_files(
        path_list(section.get('files', "")),
        ignored_file_paths=path_list(section.get('ignore', "")),
        limit_file_paths=path_list(section.get('limit_files', "")))
    file_dict = get_file_dict(filename_list, log_printer)

    manager = multiprocessing.Manager()
    global_bear_queue = multiprocessing.Queue()
    filename_queue = multiprocessing.Queue()
    local_result_dict = manager.dict()
    global_result_dict = manager.dict()
    message_queue = multiprocessing.Queue()
    control_queue = multiprocessing.Queue()

    bear_runner_args = {"file_name_queue": filename_queue,
                        "local_bear_list": local_bear_list,
                        "global_bear_list": global_bear_list,
                        "global_bear_queue": global_bear_queue,
                        "file_dict": file_dict,
                        "local_result_dict": local_result_dict,
                        "global_result_dict": global_result_dict,
                        "message_queue": message_queue,
                        "control_queue": control_queue,
                        "timeout": 0.1}

    local_bear_list[:], global_bear_list[:] = instantiate_bears(
        section,
        local_bear_list,
        global_bear_list,
        file_dict,
        message_queue)

    fill_queue(filename_queue, file_dict.keys())
    fill_queue(global_bear_queue, range(len(global_bear_list)))

    return ([multiprocessing.Process(target=run, kwargs=bear_runner_args)
             for i in range(job_count)],
            bear_runner_args)


def get_ignore_scope(line, keyword):
    """
    Retrieves the bears that are to be ignored defined in the given line.
    :param line:    The line containing the ignore declaration.
    :param keyword: The keyword that was found. Everything after the rightmost
                    occurrence of it will be considered for the scope.
    :return:        A list of lower cased bearnames or an empty list (-> "all")
    """
    toignore = line[line.rfind(keyword) + len(keyword):]
    if toignore.startswith("all"):
        return []
    else:
        return list(StringConverter(toignore, list_delimiters=', '))


def yield_ignore_ranges(file_dict):
    """
    Yields tuples of affected bears and a SourceRange that shall be ignored for
    those.

    :param file_dict: The file dictionary.
    """
    for filename, file in file_dict.items():
        start = None
        bears = []
        for line_number, line in enumerate(file, start=1):
            line = line.lower()
            if "start ignoring " in line:
                start = line_number
                bears = get_ignore_scope(line, "start ignoring ")
            elif "stop ignoring" in line:
                if start:
                    yield (bears,
                           SourceRange.from_values(filename,
                                                   start,
                                                   end_line=line_number))
            elif "ignore " in line:
                yield (get_ignore_scope(line, "ignore "),
                       SourceRange.from_values(filename,
                                               line_number,
                                               end_line=line_number+1))


def process_queues(processes,
                   control_queue,
                   local_result_dict,
                   global_result_dict,
                   file_dict,
                   print_results,
                   section,
                   log_printer):
    """
    Iterate the control queue and send the results recieved to the print_result
    method so that they can be presented to the user.

    :param processes:          List of processes which can be used to run
                               Bears.
    :param control_queue:      Containing control elements that indicate
                               whether there is a result available and which
                               bear it belongs to.
    :param local_result_dict:  Dictionary containing results respective to
                               local bears. It is modified by the processes
                               i.e. results are added to it by multiple
                               processes.
    :param global_result_dict: Dictionary containing results respective to
                               global bears. It is modified by the processes
                               i.e. results are added to it by multiple
                               processes.
    :param file_dict:          Dictionary containing file contents with
                               filename as keys.
    :param print_results:      Prints all given results appropriate to the
                               output medium.
    :return:                   Return True if all bears execute succesfully and
                               Results were delivered to the user. Else False.
    """
    file_diff_dict = {}
    retval = False
    # Number of processes working on local/global bears. They are count down
    # when the last queue element of that process is processed which may be
    # *after* the process has ended!
    local_processes = len(processes)
    global_processes = len(processes)
    global_result_buffer = []
    ignore_ranges = list(yield_ignore_ranges(file_dict))

    # One process is the logger thread
    while local_processes > 1:
        try:
            control_elem, index = control_queue.get(timeout=0.1)

            if control_elem == CONTROL_ELEMENT.LOCAL_FINISHED:
                local_processes -= 1
            elif control_elem == CONTROL_ELEMENT.GLOBAL_FINISHED:
                global_processes -= 1
            elif control_elem == CONTROL_ELEMENT.LOCAL:
                assert local_processes != 0
                retval, res = print_result(local_result_dict[index],
                                           file_dict,
                                           retval,
                                           print_results,
                                           section,
                                           log_printer,
                                           file_diff_dict,
                                           ignore_ranges)
                local_result_dict[index] = res
            else:
                assert control_elem == CONTROL_ELEMENT.GLOBAL
                global_result_buffer.append(index)
        except queue.Empty:
            if get_running_processes(processes) < 2:  # pragma: no cover
                # Recover silently, those branches are only
                # nondeterministically covered.
                break

    # Flush global result buffer
    for elem in global_result_buffer:
        retval, res = print_result(global_result_dict[elem],
                                   file_dict,
                                   retval,
                                   print_results,
                                   section,
                                   log_printer,
                                   file_diff_dict,
                                   ignore_ranges)
        global_result_dict[elem] = res

    # One process is the logger thread
    while global_processes > 1:
        try:
            control_elem, index = control_queue.get(timeout=0.1)

            if control_elem == CONTROL_ELEMENT.GLOBAL:
                retval, res = print_result(global_result_dict[index],
                                           file_dict,
                                           retval,
                                           print_results,
                                           section,
                                           log_printer,
                                           file_diff_dict,
                                           ignore_ranges)
                global_result_dict[index] = res
            else:
                assert control_elem == CONTROL_ELEMENT.GLOBAL_FINISHED
                global_processes -= 1
        except queue.Empty:
            if get_running_processes(processes) < 2:  # pragma: no cover
                # Recover silently, those branches are only
                # nondeterministically covered.
                break

    return retval


def simplify_section_result(section_result):
    """
    Takes in a section's result from `execute_section` and simplifies it
    for easy usage in other functions.

    :param section_result: The result of a section which was executed.
    :return:               Tuple containing:
                            - bool - True if results were yielded
                            - bool - True if unfixed results were yielded
                            - list - Results from all bears (local and global)
    """
    section_yielded_result = section_result[0]
    results_for_section = []
    for value in chain(section_result[1].values(),
                       section_result[2].values()):
        if value is None:
            continue

        for result in value:
            results_for_section.append(result)
    section_yielded_unfixed_results = len(results_for_section) > 0

    return (section_yielded_result,
            section_yielded_unfixed_results,
            results_for_section)


def execute_section(section,
                    global_bear_list,
                    local_bear_list,
                    print_results,
                    log_printer):
    """
    Executes the section with the given bears.

    The execute_section method does the following things:
    1. Prepare a Process
      * Load files
      * Create queues
    2. Spawn up one or more Processes
    3. Output results from the Processes
    4. Join all processes

    :param section:          The section to execute.
    :param global_bear_list: List of global bears belonging to the section.
    :param local_bear_list:  List of local bears belonging to the section.
    :param print_results:    Prints all given results appropriate to the
                             output medium.
    :param log_printer:      The log_printer to warn to.
    :return:                 Tuple containing a bool (True if results were
                             yielded, False otherwise), a Manager.dict
                             containing all local results(filenames are key)
                             and a Manager.dict containing all global bear
                             results (bear names are key) as well as the
                             file dictionary.
    """
    local_bear_list = Dependencies.resolve(local_bear_list)
    global_bear_list = Dependencies.resolve(global_bear_list)

    try:
        running_processes = int(section['jobs'])
    except ValueError:
        log_printer.warn("Unable to convert setting 'jobs' into a number. "
                         "Falling back to CPU count.")
        running_processes = get_cpu_count()
    except IndexError:
        running_processes = get_cpu_count()

    processes, arg_dict = instantiate_processes(section,
                                                local_bear_list,
                                                global_bear_list,
                                                running_processes,
                                                log_printer)

    logger_thread = LogPrinterThread(arg_dict["message_queue"],
                                     log_printer)
    # Start and join the logger thread along with the processes to run bears
    processes.append(logger_thread)

    for runner in processes:
        runner.start()

    try:
        return (process_queues(processes,
                               arg_dict["control_queue"],
                               arg_dict["local_result_dict"],
                               arg_dict["global_result_dict"],
                               arg_dict["file_dict"],
                               print_results,
                               section,
                               log_printer),
                arg_dict["local_result_dict"],
                arg_dict["global_result_dict"],
                arg_dict["file_dict"])
    finally:
        logger_thread.running = False

        for runner in processes:
            runner.join()


1			from itertools import chain
2			import multiprocessing
3			import queue
4			import os
5			import platform
6			import subprocess
7
8			from coalib.collecting.Collectors import collect_files
9			from coalib.collecting import Dependencies
10			from coalib.misc.StringConverter import StringConverter
11			from coalib.output.printers import LOG_LEVEL
12			from coalib.processes.BearRunning import run
13			from coalib.processes.CONTROL_ELEMENT import CONTROL_ELEMENT
14			from coalib.results.Result import Result
15			from coalib.results.RESULT_SEVERITY import RESULT_SEVERITY
16			from coalib.results.SourceRange import SourceRange
17			from coalib.settings.Setting import path_list
18			from coalib.processes.LogPrinterThread import LogPrinterThread
19			from coalib.results.result_actions.ApplyPatchAction import ApplyPatchAction
20			from coalib.results.result_actions.PrintDebugMessageAction import (
21			PrintDebugMessageAction)
22			from coalib.results.result_actions.ShowPatchAction import ShowPatchAction
23
24
25			ACTIONS = [ApplyPatchAction,
26			PrintDebugMessageAction,
27			ShowPatchAction]
28
29
30			def get_cpu_count():
31			try:
32			return multiprocessing.cpu_count()
33			# cpu_count is not implemented for some CPU architectures/OSes
34			except NotImplementedError: # pragma: no cover
35			return 2
36
37
38			def fill_queue(queue_fill, any_list):
39			"""
40			Takes element from a list and populates a queue with those elements.
41
42			:param queue_fill: The queue to be filled.
43			:param any_list: List containing the elements.
44			"""
45			for elem in any_list:
46			queue_fill.put(elem)
47
48
49			def get_running_processes(processes):
50			return sum((1 if process.is_alive() else 0) for process in processes)
51
52
53			def create_process_group(command_array, **kwargs):
54			if platform.system() == "Windows": # pragma: no cover
55			proc = subprocess.Popen(
56			command_array,
57			creationflags=subprocess.CREATE_NEW_PROCESS_GROUP,
58			**kwargs)
59			else:
60			proc = subprocess.Popen(command_array,
61			preexec_fn=os.setsid,
62			**kwargs)
63			return proc
64
65
66			def get_default_actions(section):
67			"""
68			Parses the key `default_actions` in the given section.
69
70			:param section: The section where to parse from.
71			:return: A dict with the bearname as keys and their default
72			actions as values and another dict that contains bears
73			and invalid action names.
74			"""
75			try:
76			default_actions = dict(section["default_actions"])
77			except IndexError:
78			return {}, {}
79
80			action_dict = {action.get_metadata().name: action for action in ACTIONS}
81			invalid_action_set = default_actions.values() - action_dict.keys()
82			invalid_actions = {}
83			if len(invalid_action_set) != 0:
84			invalid_actions = {
85			bear: action
86			for bear, action in default_actions.items()
87			if action in invalid_action_set}
88			for invalid in invalid_actions.keys():
89			del default_actions[invalid]
90
91			actions = {bearname: action_dict[action_name]
92			for bearname, action_name in default_actions.items()}
93			return actions, invalid_actions
94
95
96			def autoapply_actions(results,
97			file_dict,
98			file_diff_dict,
99			section,
100			log_printer):
101			"""
102			Auto-applies actions like defined in the given section.
103
104			:param results: A list of results.
105			:param file_dict: A dictionary containing the name of files and its
106			contents.
107			:param file_diff_dict: A dictionary that contains filenames as keys and
108			diff objects as values.
109			:param section: The section.
110			:param log_printer: A log printer instance to log messages on.
111			:return: A list of unprocessed results.
112			"""
113
114			default_actions, invalid_actions = get_default_actions(section)
115
116			for bearname, actionname in invalid_actions.items():
117			log_printer.warn("Selected default action {} for bear {} does "
118			"not exist. Ignoring action.".format(
119			repr(actionname),
120			repr(bearname)))
121
122			if len(default_actions) == 0:
123			# There's nothing to auto-apply.
124			return results
125
126			not_processed_results = []
127			for result in results:
128			try:
129			action = default_actions[result.origin]
130			except KeyError:
131			not_processed_results.append(result)
132			continue
133
134			if not action.is_applicable(result, file_dict, file_diff_dict):
135			log_printer.warn("Selected default action {} for bear {} is not "
136			"applicable. Action not applied.".format(
137			repr(action.get_metadata().name),
138			repr(result.origin)))
139			not_processed_results.append(result)
140			continue
141
142			try:
143			action().apply_from_section(result,
144			file_dict,
145			file_diff_dict,
146			section)
147			log_printer.info("Applied {} on {} from {}.".format(
148			repr(action.get_metadata().name),
149			result.location_repr(),
150			repr(result.origin)))
151			except Exception as ex:
152			not_processed_results.append(result)
153			log_printer.log_exception(
154			"Failed to execute action {} with error: {}.".format(
155			repr(action.get_metadata().name),
156			ex),
157			ex)
158			log_printer.debug("-> for result " + repr(result) + ".")
159
160			return not_processed_results
161
162
163			def print_result(results,
164			file_dict,
165			retval,
166			print_results,
167			section,
168			log_printer,
169			file_diff_dict,
170			ignore_ranges):
171			"""
172			Takes the results produced by each bear and gives them to the print_results
173			method to present to the user.
174
175			:param results: A list of results.
176			:param file_dict: A dictionary containing the name of files and its
177			contents.
178			:param retval: It is True if no results were yielded ever before.
179			If it is False this function will return False no
180			matter what happens. Else it depends on if this
181			invocation yields results.
182			:param print_results: A function that prints all given results appropriate
183			to the output medium.
184			:param file_diff_dict: A dictionary that contains filenames as keys and
185			diff objects as values.
186			:param ignore_ranges: A list of SourceRanges. Results that affect code in
187			any of those ranges will be ignored.
188			:return: Returns False if any results were yielded. Else
189			True.
190			"""
191			min_severity_str = str(section.get('min_severity', 'INFO')).upper()
192			min_severity = RESULT_SEVERITY.str_dict.get(min_severity_str, 'INFO')
193			results = list(filter(lambda result:
194			type(result) is Result and
195			result.severity >= min_severity and
196			not result.to_ignore(ignore_ranges),
197			results))
198
199			if bool(section.get('autoapply', 'true')):
200			patched_results = autoapply_actions(results,
201			file_dict,
202			file_diff_dict,
203			section,
204			log_printer)
205			else:
206			patched_results = results
207
208			print_results(log_printer,
209			section,
210			patched_results,
211			file_dict,
212			file_diff_dict)
213			return retval or len(results) > 0, patched_results
214
215
216			def get_file_dict(filename_list, log_printer):
217			"""
218			Reads all files into a dictionary.
219
220			:param filename_list: List of names of paths to files to get contents of.
221			:param log_printer: The logger which logs errors.
222			:return: Reads the content of each file into a dictionary
223			with filenames as keys.
224			"""
225			file_dict = {}
226			for filename in filename_list:
227			try:
228			with open(filename, "r", encoding="utf-8") as _file:
229			file_dict[filename] = tuple(_file.readlines())
230			except UnicodeDecodeError:
231			log_printer.warn("Failed to read file '{}'. It seems to contain "
232			"non-unicode characters. Leaving it "
233			"out.".format(filename))
234			except Exception as exception: # pragma: no cover
235			log_printer.log_exception("Failed to read file '{}' because of "
236			"an unknown error. Leaving it "
237			"out.".format(filename),
238			exception,
239			log_level=LOG_LEVEL.WARNING)
240
241			return file_dict
242
243
244			def filter_raising_callables(it, exception, args, *kwargs):
245			"""
246			Filters all callable items inside the given iterator that raise the
247			given exceptions.
248
249			:param it: The iterator to filter.
250			:param exception: The (tuple of) exception(s) to filter for.
251			:param args: Positional arguments to pass to the callable.
252			:param kwargs: Keyword arguments to pass to the callable.
253			"""
254			for elem in it:
255			try:
256			yield elem(args, *kwargs)
257			except exception:
258			pass
259
260
261			def instantiate_bears(section,
262			local_bear_list,
263			global_bear_list,
264			file_dict,
265			message_queue):
266			"""
267			Instantiates each bear with the arguments it needs.
268
269			:param section: The section the bears belong to.
270			:param local_bear_list: List of local bear classes to instantiate.
271			:param global_bear_list: List of global bear classes to instantiate.
272			:param file_dict: Dictionary containing filenames and their
273			contents.
274			:param message_queue: Queue responsible to maintain the messages
275			delivered by the bears.
276			:return: The local and global bear instance lists.
277			"""
278			local_bear_list = [bear
279			for bear in filter_raising_callables(
280			local_bear_list,
281			RuntimeError,
282			section,
283			message_queue,
284			timeout=0.1)]
285
286			global_bear_list = [bear
287			for bear in filter_raising_callables(
288			global_bear_list,
289			RuntimeError,
290			file_dict,
291			section,
292			message_queue,
293			timeout=0.1)]
294
295			return local_bear_list, global_bear_list
296
297
298			def instantiate_processes(section,
299			local_bear_list,
300			global_bear_list,
301			job_count,
302			log_printer):
303			"""
304			Instantiate the number of processes that will run bears which will be
305			responsible for running bears in a multiprocessing environment.
306
307			:param section: The section the bears belong to.
308			:param local_bear_list: List of local bears belonging to the section.
309			:param global_bear_list: List of global bears belonging to the section.
310			:param job_count: Max number of processes to create.
311			:param log_printer: The log printer to warn to.
312			:return: A tuple containing a list of processes,
313			and the arguments passed to each process which are
314			the same for each object.
315			"""
316			filename_list = collect_files(
317			path_list(section.get('files', "")),
318			ignored_file_paths=path_list(section.get('ignore', "")),
319			limit_file_paths=path_list(section.get('limit_files', "")))
320			file_dict = get_file_dict(filename_list, log_printer)
321
322			manager = multiprocessing.Manager()
323			global_bear_queue = multiprocessing.Queue()
324			filename_queue = multiprocessing.Queue()
325			local_result_dict = manager.dict()
326			global_result_dict = manager.dict()
327			message_queue = multiprocessing.Queue()
328			control_queue = multiprocessing.Queue()
329
330			bear_runner_args = {"file_name_queue": filename_queue,
331			"local_bear_list": local_bear_list,
332			"global_bear_list": global_bear_list,
333			"global_bear_queue": global_bear_queue,
334			"file_dict": file_dict,
335			"local_result_dict": local_result_dict,
336			"global_result_dict": global_result_dict,
337			"message_queue": message_queue,
338			"control_queue": control_queue,
339			"timeout": 0.1}
340
341			local_bear_list[:], global_bear_list[:] = instantiate_bears(
342			section,
343			local_bear_list,
344			global_bear_list,
345			file_dict,
346			message_queue)
347
348			fill_queue(filename_queue, file_dict.keys())
349			fill_queue(global_bear_queue, range(len(global_bear_list)))
350
351			return ([multiprocessing.Process(target=run, kwargs=bear_runner_args)
352			for i in range(job_count)],
353			bear_runner_args)
354
355
356			def get_ignore_scope(line, keyword):
357			"""
358			Retrieves the bears that are to be ignored defined in the given line.
359			:param line: The line containing the ignore declaration.
360			:param keyword: The keyword that was found. Everything after the rightmost
361			occurrence of it will be considered for the scope.
362			:return: A list of lower cased bearnames or an empty list (-> "all")
363			"""
364			toignore = line[line.rfind(keyword) + len(keyword):]
365			if toignore.startswith("all"):
366			return []
367			else:
368			return list(StringConverter(toignore, list_delimiters=', '))
369
370
371			def yield_ignore_ranges(file_dict):
372			"""
373			Yields tuples of affected bears and a SourceRange that shall be ignored for
374			those.
375
376			:param file_dict: The file dictionary.
377			"""
378			for filename, file in file_dict.items():
379			start = None
380			bears = []
381			for line_number, line in enumerate(file, start=1):
382			line = line.lower()
383			if "start ignoring " in line:
384			start = line_number
385			bears = get_ignore_scope(line, "start ignoring ")
386			elif "stop ignoring" in line:
387			if start:
388			yield (bears,
389			SourceRange.from_values(filename,
390			start,
391			end_line=line_number))
392			elif "ignore " in line:
393			yield (get_ignore_scope(line, "ignore "),
394			SourceRange.from_values(filename,
395			line_number,
396			end_line=line_number+1))
397
398
399			def process_queues(processes,
400			control_queue,
401			local_result_dict,
402			global_result_dict,
403			file_dict,
404			print_results,
405			section,
406			log_printer):
407			"""
408			Iterate the control queue and send the results recieved to the print_result
409			method so that they can be presented to the user.
410
411			:param processes: List of processes which can be used to run
412			Bears.
413			:param control_queue: Containing control elements that indicate
414			whether there is a result available and which
415			bear it belongs to.
416			:param local_result_dict: Dictionary containing results respective to
417			local bears. It is modified by the processes
418			i.e. results are added to it by multiple
419			processes.
420			:param global_result_dict: Dictionary containing results respective to
421			global bears. It is modified by the processes
422			i.e. results are added to it by multiple
423			processes.
424			:param file_dict: Dictionary containing file contents with
425			filename as keys.
426			:param print_results: Prints all given results appropriate to the
427			output medium.
428			:return: Return True if all bears execute succesfully and
429			Results were delivered to the user. Else False.
430			"""
431			file_diff_dict = {}
432			retval = False
433			# Number of processes working on local/global bears. They are count down
434			# when the last queue element of that process is processed which may be
435			# after the process has ended!
436			local_processes = len(processes)
437			global_processes = len(processes)
438			global_result_buffer = []
439			ignore_ranges = list(yield_ignore_ranges(file_dict))
440
441			# One process is the logger thread
442			while local_processes > 1:
443			try:
444			control_elem, index = control_queue.get(timeout=0.1)
445
446			if control_elem == CONTROL_ELEMENT.LOCAL_FINISHED:
447			local_processes -= 1
448			elif control_elem == CONTROL_ELEMENT.GLOBAL_FINISHED:
449			global_processes -= 1
450			elif control_elem == CONTROL_ELEMENT.LOCAL:
451			assert local_processes != 0
452			retval, res = print_result(local_result_dict[index],
453			file_dict,
454			retval,
455			print_results,
456			section,
457			log_printer,
458			file_diff_dict,
459			ignore_ranges)
460			local_result_dict[index] = res
461			else:
462			assert control_elem == CONTROL_ELEMENT.GLOBAL
463			global_result_buffer.append(index)
464			except queue.Empty:
465			if get_running_processes(processes) < 2: # pragma: no cover
466			# Recover silently, those branches are only
467			# nondeterministically covered.
468			break
469
470			# Flush global result buffer
471			for elem in global_result_buffer:
472			retval, res = print_result(global_result_dict[elem],
473			file_dict,
474			retval,
475			print_results,
476			section,
477			log_printer,
478			file_diff_dict,
479			ignore_ranges)
480			global_result_dict[elem] = res
481
482			# One process is the logger thread
483			while global_processes > 1:
484			try:
485			control_elem, index = control_queue.get(timeout=0.1)
486
487			if control_elem == CONTROL_ELEMENT.GLOBAL:
488			retval, res = print_result(global_result_dict[index],
489			file_dict,
490			retval,
491			print_results,
492			section,
493			log_printer,
494			file_diff_dict,
495			ignore_ranges)
496			global_result_dict[index] = res
497			else:
498			assert control_elem == CONTROL_ELEMENT.GLOBAL_FINISHED
499			global_processes -= 1
500			except queue.Empty:
501			if get_running_processes(processes) < 2: # pragma: no cover
502			# Recover silently, those branches are only
503			# nondeterministically covered.
504			break
505
506			return retval
507
508
509			def simplify_section_result(section_result):
510			"""
511			Takes in a section's result from `execute_section` and simplifies it
512			for easy usage in other functions.
513
514			:param section_result: The result of a section which was executed.
515			:return: Tuple containing:
516			- bool - True if results were yielded
517			- bool - True if unfixed results were yielded
518			- list - Results from all bears (local and global)
519			"""
520			section_yielded_result = section_result[0]
521			results_for_section = []
522			for value in chain(section_result[1].values(),
523			section_result[2].values()):
524			if value is None:
525			continue
526
527			for result in value:
528			results_for_section.append(result)
529			section_yielded_unfixed_results = len(results_for_section) > 0
530
531			return (section_yielded_result,
532			section_yielded_unfixed_results,
533			results_for_section)
534
535
536			def execute_section(section,
537			global_bear_list,
538			local_bear_list,
539			print_results,
540			log_printer):
541			"""
542			Executes the section with the given bears.
543
544			The execute_section method does the following things:
545			1. Prepare a Process
546			* Load files
547			* Create queues
548			2. Spawn up one or more Processes
549			3. Output results from the Processes
550			4. Join all processes
551
552			:param section: The section to execute.
553			:param global_bear_list: List of global bears belonging to the section.
554			:param local_bear_list: List of local bears belonging to the section.
555			:param print_results: Prints all given results appropriate to the
556			output medium.
557			:param log_printer: The log_printer to warn to.
558			:return: Tuple containing a bool (True if results were
559			yielded, False otherwise), a Manager.dict
560			containing all local results(filenames are key)
561			and a Manager.dict containing all global bear
562			results (bear names are key) as well as the
563			file dictionary.
564			"""
565			local_bear_list = Dependencies.resolve(local_bear_list)
566			global_bear_list = Dependencies.resolve(global_bear_list)
567
568			try:
569			running_processes = int(section['jobs'])
570			except ValueError:
571			log_printer.warn("Unable to convert setting 'jobs' into a number. "
572			"Falling back to CPU count.")
573			running_processes = get_cpu_count()
574			except IndexError:
575			running_processes = get_cpu_count()
576
577			processes, arg_dict = instantiate_processes(section,
578			local_bear_list,
579			global_bear_list,
580			running_processes,
581			log_printer)
582
583			logger_thread = LogPrinterThread(arg_dict["message_queue"],
584			log_printer)
585			# Start and join the logger thread along with the processes to run bears
586			processes.append(logger_thread)
587
588			for runner in processes:
589			runner.start()
590
591			try:
592			return (process_queues(processes,
593			arg_dict["control_queue"],
594			arg_dict["local_result_dict"],
595			arg_dict["global_result_dict"],
596			arg_dict["file_dict"],
597			print_results,
598			section,
599			log_printer),
600			arg_dict["local_result_dict"],
601			arg_dict["global_result_dict"],
602			arg_dict["file_dict"])
603			finally:
604			logger_thread.running = False
605
606			for runner in processes:
607			runner.join()
608

coala-analyzer / coala

Pull Request — master (#1487)

coalib.processes.simplify_section_result() B

Complexity

Size

Duplication

Duplication Side-by-Side

Filter issues like