Refactor parallel linting

The previous implementation created new PyLinter objects in the worker
(child) process causing failure when running under Prospector because
Prospector uses a custom PyLinter class (a class inherited from PyLinter)
and PyLint naturally just creates PyLinter object. This caused linting to
fail because there is options for Prospector's IndentChecker which was not
created in the worker process.

The new implementation passes the original PyLinter object into workers
when the workers are created. See https://docs.python.org/3/library/multiprocessing.html#contexts-and-start-methods

The performance has remained about the same based on quick tests done with
Django project containing about 30 000 lines of code; with the old
implementation linting took 26-28 seconds with 8 jobs on quad core i7 and
24-27 seconds with the new implementation.

Author: janneronkko

ChangeLog

@@ -9,8 +9,8 @@ Release date: TBA
 
 * Refactor file checking
 
-  Remove code duplication and prepare for supporting parallel linting
-  under Prospector.
+  Remove code duplication and allow parallel linting using custom PyLinter
+  classes.
 
 * Added a new check, ``invalid-overridden-method``
 

pylint/lint.py

@@ -243,70 +243,8 @@ def _cpu_count() -> int:
     return 1
 
 
-if multiprocessing is not None:
-
-    class ChildLinter(multiprocessing.Process):
-        def run(self):
-            # pylint: disable=no-member, unbalanced-tuple-unpacking
-            tasks_queue, results_queue, self._config = self._args
-
-            self._config["jobs"] = 1  # Child does not parallelize any further.
-            self._python3_porting_mode = self._config.pop("python3_porting_mode", None)
-            self._plugins = self._config.pop("plugins", None)
-
-            # Run linter for received files/modules.
-            for file_or_module in iter(tasks_queue.get, "STOP"):
-                try:
-                    result = self._run_linter(file_or_module[0])
-                    results_queue.put(result)
-                except Exception as ex:
-                    print(
-                        "internal error with sending report for module %s"
-                        % file_or_module,
-                        file=sys.stderr,
-                    )
-                    print(ex, file=sys.stderr)
-                    results_queue.put({})
-
-        def _run_linter(self, file_or_module):
-            linter = PyLinter()
-
-            # Register standard checkers.
-            linter.load_default_plugins()
-            # Load command line plugins.
-            if self._plugins:
-                linter.load_plugin_modules(self._plugins)
-
-            linter.load_configuration_from_config(self._config)
-
-            # Load plugin specific configuration
-            linter.load_plugin_configuration()
-
-            linter.set_reporter(reporters.CollectingReporter())
-
-            # Enable the Python 3 checker mode. This option is
-            # passed down from the parent linter up to here, since
-            # the Python 3 porting flag belongs to the Run class,
-            # instead of the Linter class.
-            if self._python3_porting_mode:
-                linter.python3_porting_mode()
-
-            # Run the checks.
-            linter.check(file_or_module)
-
-            msgs = [_get_new_args(m) for m in linter.reporter.messages]
-            return (
-                file_or_module,
-                linter.file_state.base_name,
-                linter.current_name,
-                msgs,
-                linter.stats,
-                linter.msg_status,
-            )
-
-
 # pylint: disable=too-many-instance-attributes
-class PyLinter(
+class PyLinter(  # pylint: disable=too-many-public-methods
     config.OptionsManagerMixIn,
     MessagesHandlerMixIn,
     reporters.ReportsHandlerMixIn,
@@ -971,16 +909,20 @@ def should_analyze_file(modname, path, is_argument=False):
 
     # pylint: enable=unused-argument
 
-    def check(self, files_or_modules):
-        """main checking entry: check a list of files or modules from their
-        name.
-        """
+    def initialize(self):
         # initialize msgs_state now that all messages have been registered into
         # the store
         for msg in self.msgs_store.messages:
             if not msg.may_be_emitted():
                 self._msgs_state[msg.msgid] = False
 
+    def check(self, files_or_modules):
+        """main checking entry: check a list of files or modules from their
+        name.
+        """
+
+        self.initialize()
+
         if not isinstance(files_or_modules, (list, tuple)):
             files_or_modules = (files_or_modules,)
 
@@ -998,100 +940,21 @@ def check(self, files_or_modules):
         elif self.config.jobs == 1:
             self._check_files(self.get_ast, self._iterate_file_descrs(files_or_modules))
         else:
-            self._parallel_check(files_or_modules)
-
-    def _get_jobs_config(self):
-        child_config = collections.OrderedDict()
-        filter_options = {"long-help"}
-        filter_options.update((opt_name for opt_name, _ in self._external_opts))
-        for opt_providers in self._all_options.values():
-            for optname, optdict, val in opt_providers.options_and_values():
-                if optdict.get("deprecated"):
-                    continue
-
-                if optname not in filter_options:
-                    child_config[optname] = utils._format_option_value(optdict, val)
-        child_config["python3_porting_mode"] = self._python3_porting_mode
-        child_config["plugins"] = self._dynamic_plugins
-        return child_config
-
-    def _parallel_task(self, files_or_modules):
-        # Prepare configuration for child linters.
-        child_config = self._get_jobs_config()
-
-        children = []
-        manager = multiprocessing.Manager()
-        tasks_queue = manager.Queue()
-        results_queue = manager.Queue()
-
-        # Send files to child linters.
-        expanded_files = []
-        for descr in self._expand_files(files_or_modules):
-            modname, filepath, is_arg = descr["name"], descr["path"], descr["isarg"]
-            if self.should_analyze_file(modname, filepath, is_argument=is_arg):
-                expanded_files.append(descr)
-
-        # do not start more jobs than needed
-        for _ in range(min(self.config.jobs, len(expanded_files))):
-            child_linter = ChildLinter(args=(tasks_queue, results_queue, child_config))
-            child_linter.start()
-            children.append(child_linter)
-
-        for files_or_module in expanded_files:
-            path = files_or_module["path"]
-            tasks_queue.put([path])
-
-        # collect results from child linters
-        failed = False
-        for _ in expanded_files:
-            try:
-                result = results_queue.get()
-            except Exception as ex:
-                print(
-                    "internal error while receiving results from child linter",
-                    file=sys.stderr,
-                )
-                print(ex, file=sys.stderr)
-                failed = True
-                break
-            yield result
-
-        # Stop child linters and wait for their completion.
-        for _ in range(self.config.jobs):
-            tasks_queue.put("STOP")
-        for child in children:
-            child.join()
-
-        if failed:
-            print("Error occurred, stopping the linter.", file=sys.stderr)
-            sys.exit(32)
-
-    def _parallel_check(self, files_or_modules):
-        # Reset stats.
-        self.open()
-
-        all_stats = []
-        module = None
-        for result in self._parallel_task(files_or_modules):
-            if not result:
-                continue
-            (_, self.file_state.base_name, module, messages, stats, msg_status) = result
-
-            for msg in messages:
-                msg = Message(*msg)
-                self.set_current_module(module)
-                self.reporter.handle_message(msg)
+            check_parallel(
+                self, self.config.jobs, self._iterate_file_descrs(files_or_modules)
+            )
 
-            all_stats.append(stats)
-            self.msg_status |= msg_status
+    def check_single_file(self, name, filepath, modname):
+        """Check single file
 
-        self.stats = _merge_stats(all_stats)
-        self.current_name = module
+        The arguments are the same that are documented in _check_files
 
-        # Insert stats data to local checkers.
-        for checker in self.get_checkers():
-            if checker is not self:
-                checker.stats = self.stats
+        The initialize() method should be called before calling this method
+        """
+        with self._astroid_module_checker() as check_astroid_module:
+            self._check_file(
+                self.get_ast, check_astroid_module, name, filepath, modname
+            )
 
     def _check_files(self, get_ast, file_descrs):
         """Check all files from file_descrs
@@ -1313,6 +1176,73 @@ def _report_evaluation(self):
             self.reporter.display_reports(sect)
 
 
+def check_parallel(linter, jobs, files):
+    """Use the given linter to lint the files with given amount of workers (jobs)
+    """
+    original_reporter = linter.reporter
+
+    # Configure linter on the parent process side for the workers
+    linter.set_reporter(reporters.CollectingReporter())
+    linter.open()
+
+    # The linter is inherited by all the pool's workers, i.e. the linter
+    # is identical to the linter object here. This is requirde so that
+    # a custom PyLinter object (inherited from PyLinter) can be used.
+    # See https://github.com/PyCQA/prospector/issues/320
+    with multiprocessing.Pool(
+        jobs, initializer=_worker_initialize, initargs=[linter]
+    ) as pool:
+        # ..and now when the workers have inherited the linter, the actual reporter
+        # can be set back here on the parent process so that results get stored into
+        # correct reporter
+        linter.set_reporter(original_reporter)
+
+        all_stats = []
+
+        for module, messages, stats, msg_status in pool.imap_unordered(
+            _worker_check_single_file, files
+        ):
+            linter.set_current_module(module)
+            for msg in messages:
+                msg = Message(*msg)
+                linter.reporter.handle_message(msg)
+
+            all_stats.append(stats)
+            linter.msg_status |= msg_status
+
+    linter.stats = _merge_stats(all_stats)
+
+    # Insert stats data to local checkers.
+    for checker in linter.get_checkers():
+        if checker is not linter:
+            checker.stats = linter.stats
+
+
+# PyLinter object used by worker processes when checking files using multiprocessing
+# should only be used by the worker processes
+_worker_linter = None
+
+
+def _worker_initialize(linter):
+    global _worker_linter  # pylint: disable=global-statement
+    _worker_linter = linter
+
+
+def _worker_check_single_file(file_item):
+    name, filepath, modname = file_item
+
+    _worker_linter.open()
+    _worker_linter.check_single_file(name, filepath, modname)
+
+    msgs = [_get_new_args(m) for m in _worker_linter.reporter.messages]
+    return (
+        _worker_linter.current_name,
+        msgs,
+        _worker_linter.stats,
+        _worker_linter.msg_status,
+    )
+
+
 # some reporting functions ####################################################