coalib.processes.autoapply_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 16:38 UTC

coalib.processes.autoapply_actions() C

↳ Parent: Project

Complexity

Conditions

Size

Total Lines

Duplication

Lines	0
Ratio	0 %

Metric	Value
cc	7
dl	0
loc	63
rs	6.1545

How to fix Long Method

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

coala-analyzer / coala

Pull Request — master (#1170)

coalib.processes.autoapply_actions() C

Complexity

Size

Duplication

How to fix Long Method

Long Method

Duplication Side-by-Side

Filter issues like