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 18:38 UTC

coalib.processes.get_default_actions() D

↳ Parent: Project

Complexity

Conditions

Size

Total Lines

Duplication

Lines	0
Ratio	0 %

Metric	Value
cc	8
dl	0
loc	28
rs	4

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

coala-analyzer / coala

Pull Request — master (#1170)

coalib.processes.get_default_actions() D

Complexity

Size

Duplication

Duplication Side-by-Side

Filter issues like