coalib.processes.get_default_actions() - Code Metrics - Inspection of "Makman2/auto-apply2" - coala-analyzer/coala - Measure and Improve Code Quality continuously with Scrutinizer

Completed

Pull Request — master (#1170)

by Mischa

created 2015-12-30 02:48 UTC

coalib.processes.get_default_actions() C

↳ Parent: Project

Complexity

Conditions

Size

Total Lines

Duplication

Lines	0
Ratio	0 %

Metric	Value
cc	7
dl	0
loc	30
rs	5.5

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.OpenEditorAction import OpenEditorAction
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 = [OpenEditorAction,
           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.
    :raise ValueError: Raised when an action is supplied for a bear that does
                       not exist. Invalid actions are passed through the
                       `invalid_actions` dict (with `bearname: actionname`)
                       as a member inside the exception.
    :return:           A dict with the bearname as keys and their default
                       actions as values.
    """
    try:
        default_actions = dict(section["default_actions"])
    except IndexError:
        return {}

    action_dict = {action.get_metadata().name: action for action in ACTIONS}
    invalid_actions = default_actions.values() - action_dict.keys()
    if len(invalid_actions) != 0:
        ex = ValueError("Section " + repr(section.name) + " contains invalid "
                        "values for key 'default_actions'.")
        reverted_default_actions = {value: key
                                    for key, value in default_actions.items()}
        ex.invalid_actions = {reverted_default_actions[action]: action
                              for action in invalid_actions}
        raise ex

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


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 non-processed results.
    """

    try:
        default_actions = get_default_actions(section)
    except ValueError as ex:
        log_printer.err(ex)
        for bearname, actionname in ex.invalid_actions.items():
            log_printer.err("Selected default action {} for bear {} does "
                            "not exist.".format(repr(actionname),
                                                repr(bearname)))
        return results

    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)
        except Exception as ex:
            log_printer.log_exception(
                "Failed to execute action {}. {}".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))

    results = autoapply_actions(results,
                                file_dict,
                                file_diff_dict,
                                section,
                                log_printer)

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


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] = _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', "")),
                                  path_list(section.get('ignore', "")))
    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 = {}
    running_processes = get_running_processes(processes)
    retval = False
    # Number of processes working on local bears
    local_processes = len(processes)
    global_result_buffer = []
    ignore_ranges = list(yield_ignore_ranges(file_dict))

    # One process is the logger thread
    while local_processes > 1 and running_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.LOCAL:
                assert local_processes != 0
                retval = print_result(local_result_dict[index],
                                      file_dict,
                                      retval,
                                      print_results,
                                      section,
                                      log_printer,
                                      file_diff_dict,
                                      ignore_ranges)
            elif control_elem == CONTROL_ELEMENT.GLOBAL:
                global_result_buffer.append(index)
        except queue.Empty:
            running_processes = get_running_processes(processes)

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

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

            if control_elem == CONTROL_ELEMENT.GLOBAL:
                retval = print_result(global_result_dict[index],
                                      file_dict,
                                      retval,
                                      print_results,
                                      section,
                                      log_printer,
                                      file_diff_dict,
                                      ignore_ranges)
            else:
                assert control_elem == CONTROL_ELEMENT.GLOBAL_FINISHED
                running_processes = get_running_processes(processes)

        except queue.Empty:
            running_processes = get_running_processes(processes)

    return retval


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

coala-analyzer / coala

Pull Request — master (#1170)

coalib.processes.get_default_actions() C

Complexity

Size

Duplication

Duplication Side-by-Side

Filter issues like