Backport pylint-dev#7834 to v2.13

pylint/lint/parallel.py

@@ -29,10 +29,15 @@
 except ImportError:
     multiprocessing = None  # type: ignore[assignment]
 
+try:
+    from concurrent.futures import ProcessPoolExecutor
+except ImportError:
+    ProcessPoolExecutor = None  # type: ignore[assignment,misc]
+
 if TYPE_CHECKING:
     from pylint.lint import PyLinter
 
-# PyLinter object used by worker processes when checking files using multiprocessing
+# PyLinter object used by worker processes when checking files using parallel mode
 # should only be used by the worker processes
 _worker_linter = None
 
@@ -52,7 +57,7 @@ def _get_new_args(message):
 def _worker_initialize(
     linter: bytes, arguments: Union[None, str, Sequence[str]] = None
 ) -> None:
-    """Function called to initialize a worker for a Process within a multiprocessing Pool.
+    """Function called to initialize a worker for a Process within a concurrent Pool.
 
     :param linter: A linter-class (PyLinter) instance pickled with dill
     :param arguments: File or module name(s) to lint and to be added to sys.path
@@ -144,9 +149,9 @@ def check_parallel(
     # is identical to the linter object here. This is required so that
     # a custom PyLinter object can be used.
     initializer = functools.partial(_worker_initialize, arguments=arguments)
-    with multiprocessing.Pool(
-        jobs, initializer=initializer, initargs=[dill.dumps(linter)]
-    ) as pool:
+    with ProcessPoolExecutor(
+        max_workers=jobs, initializer=initializer, initargs=(dill.dumps(linter),)
+    ) as executor:
         linter.open()
         all_stats = []
         all_mapreduce_data = collections.defaultdict(list)
@@ -163,7 +168,7 @@ def check_parallel(
             stats,
             msg_status,
             mapreduce_data,
-        ) in pool.imap_unordered(_worker_check_single_file, files):
+        ) in executor.map(_worker_check_single_file, files):
             linter.file_state.base_name = base_name
             linter.set_current_module(module, file_path)
             for msg in messages:

pylint/lint/run.py

@@ -19,6 +19,11 @@
 except ImportError:
     multiprocessing = None  # type: ignore[assignment]
 
+try:
+    from concurrent.futures import ProcessPoolExecutor
+except ImportError:
+    ProcessPoolExecutor = None  # type: ignore[assignment,misc]
+
 
 def _cpu_count() -> int:
     """Use sched_affinity if available for virtualized or containerized environments."""
@@ -336,9 +341,9 @@ def __init__(
             )
             sys.exit(32)
         if linter.config.jobs > 1 or linter.config.jobs == 0:
-            if multiprocessing is None:
+            if ProcessPoolExecutor is None:
                 print(
-                    "Multiprocessing library is missing, fallback to single process",
+                    "concurrent.futures module is missing, fallback to single process",
                     file=sys.stderr,
                 )
                 linter.set_option("jobs", 1)

tests/test_check_parallel.py

@@ -7,8 +7,9 @@
 # pylint: disable=protected-access,missing-function-docstring,no-self-use
 
 import argparse
-import multiprocessing
 import os
+from concurrent.futures import ProcessPoolExecutor
+from concurrent.futures.process import BrokenProcessPool
 from typing import List
 
 import dill
@@ -184,10 +185,10 @@ def test_worker_initialize_pickling(self) -> None:
         """
         linter = PyLinter(reporter=Reporter())
         linter.attribute = argparse.ArgumentParser()  # type: ignore[attr-defined]
-        with multiprocessing.Pool(
-            2, initializer=worker_initialize, initargs=[dill.dumps(linter)]
-        ) as pool:
-            pool.imap_unordered(print, [1, 2])
+        with ProcessPoolExecutor(
+            max_workers=2, initializer=worker_initialize, initargs=(dill.dumps(linter),)
+        ) as executor:
+            executor.map(print, [1, 2])
 
     def test_worker_check_single_file_uninitialised(self) -> None:
         pylint.lint.parallel._worker_linter = None