From d2f61ecf275f4a85aa69f1b754a369a73cdaf1b5 Mon Sep 17 00:00:00 2001
From: bmartinn <>
Date: Sun, 10 May 2020 00:03:38 +0300
Subject: [PATCH 01/20] Add parallel (multi-process) install, if
 multiprocessing Pool is not available fall-back to serial installation

---
 src/pip/_internal/req/__init__.py | 101 ++++++++++++++++++++----------
 1 file changed, 67 insertions(+), 34 deletions(-)

diff --git a/src/pip/_internal/req/__init__.py b/src/pip/_internal/req/__init__.py
index 06f0a0823f1..31d088cfd92 100644
--- a/src/pip/_internal/req/__init__.py
+++ b/src/pip/_internal/req/__init__.py
@@ -3,11 +3,17 @@
 from typing import Iterator, List, Optional, Sequence, Tuple
 
 from pip._internal.utils.logging import indent_log
+from pip._internal.utils.typing import MYPY_CHECK_RUNNING
 
 from .req_file import parse_requirements
 from .req_install import InstallRequirement
 from .req_set import RequirementSet
 
+try:
+    from multiprocessing.pool import Pool
+except ImportError:  # Platform-specific: No multiprocessing available
+    Pool = None
+
 __all__ = [
     "RequirementSet", "InstallRequirement",
     "parse_requirements", "install_given_reqs",
@@ -60,39 +66,66 @@ def install_given_reqs(
             ', '.join(to_install.keys()),
         )
 
-    installed = []
+    # pre allocate installed package names
+    installed = [None] * len(to_install)
+    install_args = [install_options, global_options, dict(
+    	root=root, home=home, prefix=prefix, warn_script_location=warn_script_location,
+    	use_user_site=use_user_site, pycompile=pycompile)]
+
+    if Pool is not None:
+        # first let's try to install in parallel, if we fail we do it by order.
+        pool = Pool()
+        try:
+            pool_result = pool.starmap_async(__single_install, [(install_args, r) for r in to_install])
+            # python 2.7 timeout=None will not catch KeyboardInterrupt
+            installed = pool_result.get(timeout=999999)
+        except (KeyboardInterrupt, SystemExit):
+            pool.terminate()
+            raise
+        except Exception:
+            # we will reinstall sequentially
+            pass
+        pool.close()
+        pool.join()
 
     with indent_log():
-        for req_name, requirement in to_install.items():
-            if requirement.should_reinstall:
-                logger.info('Attempting uninstall: %s', req_name)
-                with indent_log():
-                    uninstalled_pathset = requirement.uninstall(
-                        auto_confirm=True
-                    )
-            else:
-                uninstalled_pathset = None
-
-            try:
-                requirement.install(
-                    install_options,
-                    global_options,
-                    root=root,
-                    home=home,
-                    prefix=prefix,
-                    warn_script_location=warn_script_location,
-                    use_user_site=use_user_site,
-                    pycompile=pycompile,
-                )
-            except Exception:
-                # if install did not succeed, rollback previous uninstall
-                if uninstalled_pathset and not requirement.install_succeeded:
-                    uninstalled_pathset.rollback()
-                raise
-            else:
-                if uninstalled_pathset and requirement.install_succeeded:
-                    uninstalled_pathset.commit()
-
-            installed.append(InstallationResult(req_name))
-
-    return installed
+        for i, requirement in enumerate(to_install):
+            if installed[i] is None:
+                installed[i] = __single_install(install_args, requirement, allow_raise=True)
+
+    return [i for i in installed if i is not None]
+
+
+def __single_install(args, a_requirement, allow_raise=False):
+    if a_requirement.should_reinstall:
+        logger.info('Attempting uninstall: %s', a_requirement.name)
+        with indent_log():
+            uninstalled_pathset = a_requirement.uninstall(
+                auto_confirm=True
+            )
+    try:
+        a_requirement.install(
+            args[0],   # install_options,
+            args[1],   # global_options,
+            **args[2]  # **kwargs
+        )
+    except Exception:
+        should_rollback = (
+                a_requirement.should_reinstall and
+                not a_requirement.install_succeeded
+        )
+        # if install did not succeed, rollback previous uninstall
+        if should_rollback:
+            uninstalled_pathset.rollback()
+        if allow_raise:
+            raise
+    else:
+        should_commit = (
+                a_requirement.should_reinstall and
+                a_requirement.install_succeeded
+        )
+        if should_commit:
+            uninstalled_pathset.commit()
+        return InstallationResult(a_requirement.name)
+
+    return None

From dd9cbf98803a6700cb928279669890dfea9b4f39 Mon Sep 17 00:00:00 2001
From: bmartinn <>
Date: Sun, 10 May 2020 00:16:55 +0300
Subject: [PATCH 02/20] Add multiprocessing feature news

---
 news/8187.feature | 3 +++
 1 file changed, 3 insertions(+)
 create mode 100644 news/8187.feature

diff --git a/news/8187.feature b/news/8187.feature
new file mode 100644
index 00000000000..af693aaf2f5
--- /dev/null
+++ b/news/8187.feature
@@ -0,0 +1,3 @@
+Run the wheel install in a multiprocessing Pool, this has x1.5 speedup factor when installing cached packages.
+Packages that could not be installed (exception raised), will be installed serially once the Pool is done.
+If multiprocessing.Pool is not supported by the platform, fall-back to serial installation.

From 02301f17b6a86068fa29cf910afe8e1c186affb5 Mon Sep 17 00:00:00 2001
From: bmartinn <>
Date: Sun, 10 May 2020 02:30:22 +0300
Subject: [PATCH 03/20] CI pep8, flake8, mypy, py27

---
 news/8187.feature | 8 +++++---
 1 file changed, 5 insertions(+), 3 deletions(-)

diff --git a/news/8187.feature b/news/8187.feature
index af693aaf2f5..ba5654a31e3 100644
--- a/news/8187.feature
+++ b/news/8187.feature
@@ -1,3 +1,5 @@
-Run the wheel install in a multiprocessing Pool, this has x1.5 speedup factor when installing cached packages.
-Packages that could not be installed (exception raised), will be installed serially once the Pool is done.
-If multiprocessing.Pool is not supported by the platform, fall-back to serial installation.
+Run the wheel install in a multiprocessing Pool, this has x1.5 speedup factor
+when installing cached packages. Packages that could not be installed
+(exception raised), will be installed serially once the Pool is done.
+If multiprocessing.Pool is not supported by the platform,
+fall-back to serial installation.

From 8c471790fd7df37c54d9ae636dbedfd5f085b160 Mon Sep 17 00:00:00 2001
From: bmartinn <>
Date: Mon, 1 Jun 2020 00:24:10 +0300
Subject: [PATCH 04/20] Restored code after bad rebase

---
 src/pip/_internal/req/__init__.py | 137 ++++++++++++++++++++----------
 1 file changed, 92 insertions(+), 45 deletions(-)

diff --git a/src/pip/_internal/req/__init__.py b/src/pip/_internal/req/__init__.py
index 31d088cfd92..090769d2924 100644
--- a/src/pip/_internal/req/__init__.py
+++ b/src/pip/_internal/req/__init__.py
@@ -1,6 +1,8 @@
 import collections
 import logging
 from typing import Iterator, List, Optional, Sequence, Tuple
+import sys
+from functools import partial
 
 from pip._internal.utils.logging import indent_log
 from pip._internal.utils.typing import MYPY_CHECK_RUNNING
@@ -10,9 +12,9 @@
 from .req_set import RequirementSet
 
 try:
-    from multiprocessing.pool import Pool
+    from multiprocessing.pool import Pool  # noqa
 except ImportError:  # Platform-specific: No multiprocessing available
-    Pool = None
+    Pool = None   # type: ignore
 
 __all__ = [
     "RequirementSet", "InstallRequirement",
@@ -67,65 +69,110 @@ def install_given_reqs(
         )
 
     # pre allocate installed package names
-    installed = [None] * len(to_install)
+    installed = [None] * len(
+        to_install
+    )  # type: List[Union[None, InstallationResult, BaseException]]
+
     install_args = [install_options, global_options, dict(
-    	root=root, home=home, prefix=prefix, warn_script_location=warn_script_location,
-    	use_user_site=use_user_site, pycompile=pycompile)]
+        root=root, home=home, prefix=prefix,
+        warn_script_location=warn_script_location,
+        use_user_site=use_user_site, pycompile=pycompile)]
+
+    with indent_log():
+        # first try to install in parallel
+        installed_pool = __safe_pool_map(
+            partial(__single_install, install_args, in_subprocess=True),
+            to_install)
+        if installed_pool:
+            installed = installed_pool
+
+        for i, requirement in enumerate(to_install):
+            if installed[i] is None:
+                installed[i] = __single_install(
+                    install_args, requirement, in_subprocess=False)
+            elif isinstance(installed[i], BaseException):
+                raise installed[i]   # type: ignore
+
+    return [i for i in installed if isinstance(i, InstallationResult)]
 
-    if Pool is not None:
-        # first let's try to install in parallel, if we fail we do it by order.
+
+def __safe_pool_map(
+        func,               # type: Callable[[Any], Any]
+        iterable,           # type: Iterable[Any]
+):
+    # type: (...) -> Optional[List[Any]]
+    """
+    Safe call to Pool map, if Pool is not available return None
+    """
+    # Disable multiprocessing on Windows python 2.7
+    if sys.platform == 'win32' and sys.version_info.major == 2:
+        return None
+
+    if not iterable or Pool is None:
+        return None
+
+    # first let's try to install in parallel,
+    # if we fail we do it by order.
+    try:
+        # Pool context would have been nice, but not supported on Python 2.7
+        # Once officially dropped, switch to context to avoid close/join calls
         pool = Pool()
+    except ImportError:
+        return [func(i) for i in iterable]
+    else:
         try:
-            pool_result = pool.starmap_async(__single_install, [(install_args, r) for r in to_install])
             # python 2.7 timeout=None will not catch KeyboardInterrupt
-            installed = pool_result.get(timeout=999999)
+            results = pool.map_async(func, iterable).get(timeout=999999)
         except (KeyboardInterrupt, SystemExit):
             pool.terminate()
             raise
-        except Exception:
-            # we will reinstall sequentially
-            pass
-        pool.close()
-        pool.join()
+        else:
+            pool.close()
+            pool.join()
+            return results
 
-    with indent_log():
-        for i, requirement in enumerate(to_install):
-            if installed[i] is None:
-                installed[i] = __single_install(install_args, requirement, allow_raise=True)
-
-    return [i for i in installed if i is not None]
 
+def __single_install(
+        install_args,           # type: List[Any]
+        requirement,            # type: InstallRequirement
+        in_subprocess=False,    # type: bool
+):
+    # type: (...) -> Union[None, InstallationResult, BaseException]
+    """
+    Install a single requirement, returns InstallationResult
+    (to be called per requirement, either in parallel or serially)
+    """
+    if (in_subprocess and
+            (requirement.should_reinstall or not requirement.is_wheel)):
+        return None
 
-def __single_install(args, a_requirement, allow_raise=False):
-    if a_requirement.should_reinstall:
-        logger.info('Attempting uninstall: %s', a_requirement.name)
+    if requirement.should_reinstall:
+        logger.info('Attempting uninstall: %s', requirement.name)
         with indent_log():
-            uninstalled_pathset = a_requirement.uninstall(
+            uninstalled_pathset = requirement.uninstall(
                 auto_confirm=True
             )
     try:
-        a_requirement.install(
-            args[0],   # install_options,
-            args[1],   # global_options,
-            **args[2]  # **kwargs
-        )
-    except Exception:
-        should_rollback = (
-                a_requirement.should_reinstall and
-                not a_requirement.install_succeeded
+        requirement.install(
+            install_args[0],   # install_options,
+            install_args[1],   # global_options,
+            **install_args[2]  # **kwargs
         )
+    except (KeyboardInterrupt, SystemExit):
+        # always raise, we catch it in external loop
+        raise
+    except BaseException as ex:
+        should_rollback = (requirement.should_reinstall and
+                           not requirement.install_succeeded)
         # if install did not succeed, rollback previous uninstall
         if should_rollback:
             uninstalled_pathset.rollback()
-        if allow_raise:
-            raise
-    else:
-        should_commit = (
-                a_requirement.should_reinstall and
-                a_requirement.install_succeeded
-        )
-        if should_commit:
-            uninstalled_pathset.commit()
-        return InstallationResult(a_requirement.name)
-
-    return None
+        if in_subprocess:
+            return ex
+        raise
+
+    should_commit = (requirement.should_reinstall and
+                     requirement.install_succeeded)
+    if should_commit:
+        uninstalled_pathset.commit()
+    return InstallationResult(requirement.name)

From 8133130746082ee21761ae5700a962851740f602 Mon Sep 17 00:00:00 2001
From: bmartinn <>
Date: Fri, 12 Mar 2021 01:36:47 +0200
Subject: [PATCH 05/20] Fix rebase

---
 src/pip/_internal/req/__init__.py | 24 +++++++++++-------------
 1 file changed, 11 insertions(+), 13 deletions(-)

diff --git a/src/pip/_internal/req/__init__.py b/src/pip/_internal/req/__init__.py
index 090769d2924..9fea9bf44c4 100644
--- a/src/pip/_internal/req/__init__.py
+++ b/src/pip/_internal/req/__init__.py
@@ -1,11 +1,11 @@
 import collections
 import logging
-from typing import Iterator, List, Optional, Sequence, Tuple
+from typing import Iterator, List, Optional, Sequence, Tuple, Union, Callable, \
+    Iterable, Any
 import sys
 from functools import partial
 
-from pip._internal.utils.logging import indent_log
-from pip._internal.utils.typing import MYPY_CHECK_RUNNING
+from ..._internal.utils.logging import indent_log
 
 from .req_file import parse_requirements
 from .req_install import InstallRequirement
@@ -69,9 +69,7 @@ def install_given_reqs(
         )
 
     # pre allocate installed package names
-    installed = [None] * len(
-        to_install
-    )  # type: List[Union[None, InstallationResult, BaseException]]
+    installed = {name: None for name in to_install}
 
     install_args = [install_options, global_options, dict(
         root=root, home=home, prefix=prefix,
@@ -82,16 +80,16 @@ def install_given_reqs(
         # first try to install in parallel
         installed_pool = __safe_pool_map(
             partial(__single_install, install_args, in_subprocess=True),
-            to_install)
+            list(to_install.values()))
         if installed_pool:
-            installed = installed_pool
+            installed = dict(zip(to_install.keys(), to_install.values()))
 
-        for i, requirement in enumerate(to_install):
-            if installed[i] is None:
-                installed[i] = __single_install(
+        for name, requirement in to_install.items():
+            if installed[name] is None:
+                installed[name] = __single_install(
                     install_args, requirement, in_subprocess=False)
-            elif isinstance(installed[i], BaseException):
-                raise installed[i]   # type: ignore
+            elif isinstance(installed[name], BaseException):
+                raise installed[name]   # type: ignore
 
     return [i for i in installed if isinstance(i, InstallationResult)]
 

From 1fac1ede3882552e98d41043a07afe2f15dfdbd0 Mon Sep 17 00:00:00 2001
From: bmartinn <>
Date: Fri, 12 Mar 2021 02:52:21 +0200
Subject: [PATCH 06/20] Fix linter

---
 news/{8187.feature => 8187.feature.rst} |  0
 src/pip/_internal/req/__init__.py       | 28 +++++++++++++++++--------
 2 files changed, 19 insertions(+), 9 deletions(-)
 rename news/{8187.feature => 8187.feature.rst} (100%)

diff --git a/news/8187.feature b/news/8187.feature.rst
similarity index 100%
rename from news/8187.feature
rename to news/8187.feature.rst
diff --git a/src/pip/_internal/req/__init__.py b/src/pip/_internal/req/__init__.py
index 9fea9bf44c4..2d911b40ccd 100644
--- a/src/pip/_internal/req/__init__.py
+++ b/src/pip/_internal/req/__init__.py
@@ -1,12 +1,20 @@
 import collections
 import logging
-from typing import Iterator, List, Optional, Sequence, Tuple, Union, Callable, \
-    Iterable, Any
 import sys
 from functools import partial
+from typing import (
+    Any,
+    Callable,
+    Iterable,
+    Iterator,
+    List,
+    Optional,
+    Sequence,
+    Tuple,
+    Union,
+)
 
 from ..._internal.utils.logging import indent_log
-
 from .req_file import parse_requirements
 from .req_install import InstallRequirement
 from .req_set import RequirementSet
@@ -69,7 +77,7 @@ def install_given_reqs(
         )
 
     # pre allocate installed package names
-    installed = {name: None for name in to_install}
+    installed = collections.OrderedDict({name: None for name in to_install})
 
     install_args = [install_options, global_options, dict(
         root=root, home=home, prefix=prefix,
@@ -82,12 +90,14 @@ def install_given_reqs(
             partial(__single_install, install_args, in_subprocess=True),
             list(to_install.values()))
         if installed_pool:
-            installed = dict(zip(to_install.keys(), to_install.values()))
+            installed = collections.OrderedDict(
+                zip(list(to_install.keys()), installed_pool))
 
         for name, requirement in to_install.items():
             if installed[name] is None:
-                installed[name] = __single_install(
+                installed_req = __single_install(
                     install_args, requirement, in_subprocess=False)
+                installed[name] = installed_req  # type: ignore
             elif isinstance(installed[name], BaseException):
                 raise installed[name]   # type: ignore
 
@@ -163,7 +173,7 @@ def __single_install(
         should_rollback = (requirement.should_reinstall and
                            not requirement.install_succeeded)
         # if install did not succeed, rollback previous uninstall
-        if should_rollback:
+        if should_rollback and uninstalled_pathset:
             uninstalled_pathset.rollback()
         if in_subprocess:
             return ex
@@ -171,6 +181,6 @@ def __single_install(
 
     should_commit = (requirement.should_reinstall and
                      requirement.install_succeeded)
-    if should_commit:
+    if should_commit and uninstalled_pathset:
         uninstalled_pathset.commit()
-    return InstallationResult(requirement.name)
+    return InstallationResult(requirement.name or '')

From f57a66d3e08696d4fe2d156d5dd467bf4f3abfaf Mon Sep 17 00:00:00 2001
From: bmartinn <>
Date: Fri, 12 Mar 2021 04:40:27 +0200
Subject: [PATCH 07/20] Fix rebase

---
 src/pip/_internal/req/__init__.py | 14 +++++++-------
 1 file changed, 7 insertions(+), 7 deletions(-)

diff --git a/src/pip/_internal/req/__init__.py b/src/pip/_internal/req/__init__.py
index 2d911b40ccd..2af4cc27bcb 100644
--- a/src/pip/_internal/req/__init__.py
+++ b/src/pip/_internal/req/__init__.py
@@ -101,7 +101,7 @@ def install_given_reqs(
             elif isinstance(installed[name], BaseException):
                 raise installed[name]   # type: ignore
 
-    return [i for i in installed if isinstance(i, InstallationResult)]
+    return [i for i in installed.values() if isinstance(i, InstallationResult)]
 
 
 def __safe_pool_map(
@@ -160,6 +160,9 @@ def __single_install(
             uninstalled_pathset = requirement.uninstall(
                 auto_confirm=True
             )
+    else:
+        uninstalled_pathset = None
+
     try:
         requirement.install(
             install_args[0],   # install_options,
@@ -170,17 +173,14 @@ def __single_install(
         # always raise, we catch it in external loop
         raise
     except BaseException as ex:
-        should_rollback = (requirement.should_reinstall and
-                           not requirement.install_succeeded)
         # if install did not succeed, rollback previous uninstall
-        if should_rollback and uninstalled_pathset:
+        if uninstalled_pathset and not requirement.install_succeeded:
             uninstalled_pathset.rollback()
         if in_subprocess:
             return ex
         raise
 
-    should_commit = (requirement.should_reinstall and
-                     requirement.install_succeeded)
-    if should_commit and uninstalled_pathset:
+    if uninstalled_pathset and requirement.install_succeeded:
         uninstalled_pathset.commit()
+
     return InstallationResult(requirement.name or '')

From c8e8a41f41ac40f68f23f635a93e3398f0da8b1b Mon Sep 17 00:00:00 2001
From: bmartinn <>
Date: Fri, 12 Mar 2021 05:20:40 +0200
Subject: [PATCH 08/20] Add map_multiprocess_ordered to internal.utils.parallel

---
 src/pip/_internal/req/__init__.py   | 38 ++++++-----------------------
 src/pip/_internal/utils/parallel.py | 37 ++++++++++++++++++++++++++--
 2 files changed, 42 insertions(+), 33 deletions(-)

diff --git a/src/pip/_internal/req/__init__.py b/src/pip/_internal/req/__init__.py
index 2af4cc27bcb..c45a275b1e5 100644
--- a/src/pip/_internal/req/__init__.py
+++ b/src/pip/_internal/req/__init__.py
@@ -1,6 +1,5 @@
 import collections
 import logging
-import sys
 from functools import partial
 from typing import (
     Any,
@@ -15,15 +14,11 @@
 )
 
 from ..._internal.utils.logging import indent_log
+from ..utils.parallel import map_multiprocess_ordered
 from .req_file import parse_requirements
 from .req_install import InstallRequirement
 from .req_set import RequirementSet
 
-try:
-    from multiprocessing.pool import Pool  # noqa
-except ImportError:  # Platform-specific: No multiprocessing available
-    Pool = None   # type: ignore
-
 __all__ = [
     "RequirementSet", "InstallRequirement",
     "parse_requirements", "install_given_reqs",
@@ -101,7 +96,10 @@ def install_given_reqs(
             elif isinstance(installed[name], BaseException):
                 raise installed[name]   # type: ignore
 
-    return [i for i in installed.values() if isinstance(i, InstallationResult)]
+    return [
+        i for i in installed.values()  # type: ignore
+        if isinstance(i, InstallationResult)
+    ]
 
 
 def __safe_pool_map(
@@ -112,32 +110,10 @@ def __safe_pool_map(
     """
     Safe call to Pool map, if Pool is not available return None
     """
-    # Disable multiprocessing on Windows python 2.7
-    if sys.platform == 'win32' and sys.version_info.major == 2:
-        return None
-
-    if not iterable or Pool is None:
+    if not iterable:
         return None
 
-    # first let's try to install in parallel,
-    # if we fail we do it by order.
-    try:
-        # Pool context would have been nice, but not supported on Python 2.7
-        # Once officially dropped, switch to context to avoid close/join calls
-        pool = Pool()
-    except ImportError:
-        return [func(i) for i in iterable]
-    else:
-        try:
-            # python 2.7 timeout=None will not catch KeyboardInterrupt
-            results = pool.map_async(func, iterable).get(timeout=999999)
-        except (KeyboardInterrupt, SystemExit):
-            pool.terminate()
-            raise
-        else:
-            pool.close()
-            pool.join()
-            return results
+    return map_multiprocess_ordered(func, iterable)
 
 
 def __single_install(
diff --git a/src/pip/_internal/utils/parallel.py b/src/pip/_internal/utils/parallel.py
index de91dc8abc8..c0cae5e356a 100644
--- a/src/pip/_internal/utils/parallel.py
+++ b/src/pip/_internal/utils/parallel.py
@@ -16,13 +16,17 @@
   than using the default value of 1.
 """
 
-__all__ = ["map_multiprocess", "map_multithread"]
+__all__ = [
+    "map_multiprocess",
+    "map_multithread",
+    "map_multiprocess_ordered",
+]
 
 from contextlib import contextmanager
 from multiprocessing import Pool as ProcessPool
 from multiprocessing import pool
 from multiprocessing.dummy import Pool as ThreadPool
-from typing import Callable, Iterable, Iterator, TypeVar, Union
+from typing import Callable, Iterable, Iterator, List, TypeVar, Union
 
 from pip._vendor.requests.adapters import DEFAULT_POOLSIZE
 
@@ -49,6 +53,9 @@ def closing(pool):
     """Return a context manager making sure the pool closes properly."""
     try:
         yield pool
+    except (KeyboardInterrupt, SystemExit):
+        pool.terminate()
+        raise
     finally:
         # For Pool.imap*, close and join are needed
         # for the returned iterator to begin yielding.
@@ -68,6 +75,17 @@ def _map_fallback(func, iterable, chunksize=1):
     return map(func, iterable)
 
 
+def _map_ordered_fallback(func, iterable, chunksize=1):
+    # type: (Callable[[S], T], Iterable[S], int) -> List[T]
+    """Make a list applying func to each element in iterable.
+
+    This function is the sequential fallback either on Python 2
+    where Pool.imap* doesn't react to KeyboardInterrupt
+    or when sem_open is unavailable.
+    """
+    return list(map(func, iterable))
+
+
 def _map_multiprocess(func, iterable, chunksize=1):
     # type: (Callable[[S], T], Iterable[S], int) -> Iterator[T]
     """Chop iterable into chunks and submit them to a process pool.
@@ -81,6 +99,19 @@ def _map_multiprocess(func, iterable, chunksize=1):
         return pool.imap_unordered(func, iterable, chunksize)
 
 
+def _map_multiprocess_ordered(func, iterable, chunksize=1):
+    # type: (Callable[[S], T], Iterable[S], int) -> List[T]
+    """Chop iterable into chunks and submit them to a process pool.
+
+    For very long iterables using a large value for chunksize can make
+    the job complete much faster than using the default value of 1.
+
+    Return an ordered list of the results.
+    """
+    with closing(ProcessPool()) as pool:
+        return pool.map(func, iterable, chunksize)
+
+
 def _map_multithread(func, iterable, chunksize=1):
     # type: (Callable[[S], T], Iterable[S], int) -> Iterator[T]
     """Chop iterable into chunks and submit them to a thread pool.
@@ -96,6 +127,8 @@ def _map_multithread(func, iterable, chunksize=1):
 
 if LACK_SEM_OPEN:
     map_multiprocess = map_multithread = _map_fallback
+    map_multiprocess_ordered = _map_ordered_fallback
 else:
     map_multiprocess = _map_multiprocess
     map_multithread = _map_multithread
+    map_multiprocess_ordered = _map_multiprocess_ordered

From 78851eafe81e79dcbfb1297deabd6d7fed5b4674 Mon Sep 17 00:00:00 2001
From: bmartinn <>
Date: Fri, 12 Mar 2021 06:29:45 +0200
Subject: [PATCH 09/20] Rename internal function __ to _

---
 src/pip/_internal/req/__init__.py | 13 ++++++++-----
 1 file changed, 8 insertions(+), 5 deletions(-)

diff --git a/src/pip/_internal/req/__init__.py b/src/pip/_internal/req/__init__.py
index c45a275b1e5..e1dc5275290 100644
--- a/src/pip/_internal/req/__init__.py
+++ b/src/pip/_internal/req/__init__.py
@@ -81,8 +81,8 @@ def install_given_reqs(
 
     with indent_log():
         # first try to install in parallel
-        installed_pool = __safe_pool_map(
-            partial(__single_install, install_args, in_subprocess=True),
+        installed_pool = _safe_pool_map(
+            partial(_single_install, install_args, in_subprocess=True),
             list(to_install.values()))
         if installed_pool:
             installed = collections.OrderedDict(
@@ -90,7 +90,7 @@ def install_given_reqs(
 
         for name, requirement in to_install.items():
             if installed[name] is None:
-                installed_req = __single_install(
+                installed_req = _single_install(
                     install_args, requirement, in_subprocess=False)
                 installed[name] = installed_req  # type: ignore
             elif isinstance(installed[name], BaseException):
@@ -102,7 +102,7 @@ def install_given_reqs(
     ]
 
 
-def __safe_pool_map(
+def _safe_pool_map(
         func,               # type: Callable[[Any], Any]
         iterable,           # type: Iterable[Any]
 ):
@@ -116,7 +116,7 @@ def __safe_pool_map(
     return map_multiprocess_ordered(func, iterable)
 
 
-def __single_install(
+def _single_install(
         install_args,           # type: List[Any]
         requirement,            # type: InstallRequirement
         in_subprocess=False,    # type: bool
@@ -126,6 +126,9 @@ def __single_install(
     Install a single requirement, returns InstallationResult
     (to be called per requirement, either in parallel or serially)
     """
+
+    # if we are running inside a subprocess,
+    # then only clean wheel installation is supported
     if (in_subprocess and
             (requirement.should_reinstall or not requirement.is_wheel)):
         return None

From 29c614e1a72caea53862f169ab2ca53ae2d8752f Mon Sep 17 00:00:00 2001
From: bmartinn <>
Date: Fri, 12 Mar 2021 15:49:06 +0200
Subject: [PATCH 10/20] Fix comments

---
 news/8187.feature.rst               | 2 +-
 src/pip/_internal/utils/parallel.py | 4 ++--
 2 files changed, 3 insertions(+), 3 deletions(-)

diff --git a/news/8187.feature.rst b/news/8187.feature.rst
index ba5654a31e3..bb5886d65f6 100644
--- a/news/8187.feature.rst
+++ b/news/8187.feature.rst
@@ -1,4 +1,4 @@
-Run the wheel install in a multiprocessing Pool, this has x1.5 speedup factor
+Run the wheel install in a multiprocessing Pool, this has significant performance gain
 when installing cached packages. Packages that could not be installed
 (exception raised), will be installed serially once the Pool is done.
 If multiprocessing.Pool is not supported by the platform,
diff --git a/src/pip/_internal/utils/parallel.py b/src/pip/_internal/utils/parallel.py
index c0cae5e356a..eaa571b6753 100644
--- a/src/pip/_internal/utils/parallel.py
+++ b/src/pip/_internal/utils/parallel.py
@@ -68,7 +68,7 @@ def _map_fallback(func, iterable, chunksize=1):
     # type: (Callable[[S], T], Iterable[S], int) -> Iterator[T]
     """Make an iterator applying func to each element in iterable.
 
-    This function is the sequential fallback either on Python 2
+    This function is the sequential fallback
     where Pool.imap* doesn't react to KeyboardInterrupt
     or when sem_open is unavailable.
     """
@@ -79,7 +79,7 @@ def _map_ordered_fallback(func, iterable, chunksize=1):
     # type: (Callable[[S], T], Iterable[S], int) -> List[T]
     """Make a list applying func to each element in iterable.
 
-    This function is the sequential fallback either on Python 2
+    This function is the sequential fallback
     where Pool.imap* doesn't react to KeyboardInterrupt
     or when sem_open is unavailable.
     """

From e1e2018919c933c103f461c723e9ef2f405b7970 Mon Sep 17 00:00:00 2001
From: bmartinn <>
Date: Fri, 12 Mar 2021 15:58:17 +0200
Subject: [PATCH 11/20] Refactor requirements installation loop

---
 src/pip/_internal/req/__init__.py | 95 ++++++++++++++-----------------
 1 file changed, 43 insertions(+), 52 deletions(-)

diff --git a/src/pip/_internal/req/__init__.py b/src/pip/_internal/req/__init__.py
index e1dc5275290..3c89c037b5b 100644
--- a/src/pip/_internal/req/__init__.py
+++ b/src/pip/_internal/req/__init__.py
@@ -1,20 +1,11 @@
 import collections
 import logging
 from functools import partial
-from typing import (
-    Any,
-    Callable,
-    Iterable,
-    Iterator,
-    List,
-    Optional,
-    Sequence,
-    Tuple,
-    Union,
-)
+from typing import Iterator, List, Optional, Sequence, Tuple, Union
+
+from pip._internal.utils.logging import indent_log
+from pip._internal.utils.parallel import map_multiprocess_ordered
 
-from ..._internal.utils.logging import indent_log
-from ..utils.parallel import map_multiprocess_ordered
 from .req_file import parse_requirements
 from .req_install import InstallRequirement
 from .req_set import RequirementSet
@@ -37,6 +28,12 @@ def __repr__(self):
         return f"InstallationResult(name={self.name!r})"
 
 
+_InstallArgs = collections.namedtuple(
+    '_InstallArgs',
+    ['install_options', 'global_options', 'kwargs']
+)
+
+
 def _validate_requirements(
     requirements,  # type: List[InstallRequirement]
 ):
@@ -72,52 +69,46 @@ def install_given_reqs(
         )
 
     # pre allocate installed package names
-    installed = collections.OrderedDict({name: None for name in to_install})
-
-    install_args = [install_options, global_options, dict(
-        root=root, home=home, prefix=prefix,
-        warn_script_location=warn_script_location,
-        use_user_site=use_user_site, pycompile=pycompile)]
+    installed = []  # type: List[InstallationResult]
+
+    # store install arguments
+    install_args = _InstallArgs(
+        install_options=install_options,
+        global_options=global_options,
+        kwargs=dict(
+            root=root, home=home, prefix=prefix,
+            warn_script_location=warn_script_location,
+            use_user_site=use_user_site, pycompile=pycompile
+        )
+    )
 
     with indent_log():
         # first try to install in parallel
-        installed_pool = _safe_pool_map(
+        installed_pool = map_multiprocess_ordered(
             partial(_single_install, install_args, in_subprocess=True),
-            list(to_install.values()))
-        if installed_pool:
-            installed = collections.OrderedDict(
-                zip(list(to_install.keys()), installed_pool))
-
-        for name, requirement in to_install.items():
-            if installed[name] is None:
+            requirements)
+
+        # check the results from the parallel installation,
+        # and fill-in missing installations or raise exception
+        for installed_req, requested_req in zip(installed_pool, requirements):
+            # if the requirement was not installed by the parallel pool,
+            # install serially here
+            if installed_req is None:
                 installed_req = _single_install(
-                    install_args, requirement, in_subprocess=False)
-                installed[name] = installed_req  # type: ignore
-            elif isinstance(installed[name], BaseException):
-                raise installed[name]   # type: ignore
-
-    return [
-        i for i in installed.values()  # type: ignore
-        if isinstance(i, InstallationResult)
-    ]
+                    install_args, requested_req, in_subprocess=False)
 
+            if isinstance(installed_req, BaseException):
+                # Raise an exception if we caught one
+                # during the parallel installation
+                raise installed_req
+            elif isinstance(installed_req, InstallationResult):
+                installed.append(installed_req)
 
-def _safe_pool_map(
-        func,               # type: Callable[[Any], Any]
-        iterable,           # type: Iterable[Any]
-):
-    # type: (...) -> Optional[List[Any]]
-    """
-    Safe call to Pool map, if Pool is not available return None
-    """
-    if not iterable:
-        return None
-
-    return map_multiprocess_ordered(func, iterable)
+    return installed
 
 
 def _single_install(
-        install_args,           # type: List[Any]
+        install_args,           # type: _InstallArgs
         requirement,            # type: InstallRequirement
         in_subprocess=False,    # type: bool
 ):
@@ -144,9 +135,9 @@ def _single_install(
 
     try:
         requirement.install(
-            install_args[0],   # install_options,
-            install_args[1],   # global_options,
-            **install_args[2]  # **kwargs
+            install_args.install_options,
+            install_args.global_options,
+            **install_args.kwargs
         )
     except (KeyboardInterrupt, SystemExit):
         # always raise, we catch it in external loop

From db2794e3db7b53015af684bfb1b7a47e2f6250b4 Mon Sep 17 00:00:00 2001
From: "Martin.B" <51887611+bmartinn@users.noreply.github.com>
Date: Sat, 13 Mar 2021 18:21:32 +0200
Subject: [PATCH 12/20] Update news/8187.feature.rst

Co-authored-by: Tzu-ping Chung <uranusjr@gmail.com>
---
 news/8187.feature.rst | 2 +-
 1 file changed, 1 insertion(+), 1 deletion(-)

diff --git a/news/8187.feature.rst b/news/8187.feature.rst
index bb5886d65f6..94cb8def243 100644
--- a/news/8187.feature.rst
+++ b/news/8187.feature.rst
@@ -1,5 +1,5 @@
 Run the wheel install in a multiprocessing Pool, this has significant performance gain
 when installing cached packages. Packages that could not be installed
-(exception raised), will be installed serially once the Pool is done.
+(exception raised) will be installed serially once the Pool is done.
 If multiprocessing.Pool is not supported by the platform,
 fall-back to serial installation.

From fe81fa2e4072b222390fa25480dc040d88f96569 Mon Sep 17 00:00:00 2001
From: bmartinn <>
Date: Sat, 13 Mar 2021 18:45:31 +0200
Subject: [PATCH 13/20] Refactor create parallel package installation index
 list

---
 src/pip/_internal/req/__init__.py | 73 ++++++++++++++++++-------------
 1 file changed, 43 insertions(+), 30 deletions(-)

diff --git a/src/pip/_internal/req/__init__.py b/src/pip/_internal/req/__init__.py
index 3c89c037b5b..9930aaff775 100644
--- a/src/pip/_internal/req/__init__.py
+++ b/src/pip/_internal/req/__init__.py
@@ -30,7 +30,8 @@ def __repr__(self):
 
 _InstallArgs = collections.namedtuple(
     '_InstallArgs',
-    ['install_options', 'global_options', 'kwargs']
+    ['install_options', 'global_options', 'root', 'home', 'prefix',
+     'warn_script_location', 'use_user_site', 'pycompile']
 )
 
 
@@ -75,28 +76,43 @@ def install_given_reqs(
     install_args = _InstallArgs(
         install_options=install_options,
         global_options=global_options,
-        kwargs=dict(
-            root=root, home=home, prefix=prefix,
-            warn_script_location=warn_script_location,
-            use_user_site=use_user_site, pycompile=pycompile
-        )
+        root=root,
+        home=home,
+        prefix=prefix,
+        warn_script_location=warn_script_location,
+        use_user_site=use_user_site,
+        pycompile=pycompile,
     )
 
     with indent_log():
-        # first try to install in parallel
-        installed_pool = map_multiprocess_ordered(
-            partial(_single_install, install_args, in_subprocess=True),
-            requirements)
+        # get a list of packages we can install in parallel
+        parallel_reqs_index = [
+            i for i, req in enumerate(requirements)
+            if not req.should_reinstall and req.is_wheel]
+
+        # install packages parallel
+        if parallel_reqs_index:
+            parallel_reqs = map_multiprocess_ordered(
+                partial(_single_install, install_args, suppress_exception=True),
+                [requirements[i] for i in parallel_reqs_index])
+        else:
+            parallel_reqs = []
 
         # check the results from the parallel installation,
         # and fill-in missing installations or raise exception
-        for installed_req, requested_req in zip(installed_pool, requirements):
-            # if the requirement was not installed by the parallel pool,
-            # install serially here
-            if installed_req is None:
+        for i, req in enumerate(requirements):
+
+            # select the install result from the parallel installation
+            # or install serially now
+            if parallel_reqs_index and parallel_reqs_index[0] == i:
+                installed_req = parallel_reqs.pop(0)
+                parallel_reqs_index.pop(0)
+            else:
                 installed_req = _single_install(
-                    install_args, requested_req, in_subprocess=False)
+                    install_args, req, suppress_exception=False)
 
+            # Now processes the installation result,
+            # throw exception or add into installed packages
             if isinstance(installed_req, BaseException):
                 # Raise an exception if we caught one
                 # during the parallel installation
@@ -108,22 +124,17 @@ def install_given_reqs(
 
 
 def _single_install(
-        install_args,           # type: _InstallArgs
-        requirement,            # type: InstallRequirement
-        in_subprocess=False,    # type: bool
+        install_args,                # type: _InstallArgs
+        requirement,                 # type: InstallRequirement
+        suppress_exception=False,    # type: bool
 ):
     # type: (...) -> Union[None, InstallationResult, BaseException]
     """
     Install a single requirement, returns InstallationResult
-    (to be called per requirement, either in parallel or serially)
+    (to be called per requirement, either in parallel or serially).
+    Notice the two lists are of the same length
     """
 
-    # if we are running inside a subprocess,
-    # then only clean wheel installation is supported
-    if (in_subprocess and
-            (requirement.should_reinstall or not requirement.is_wheel)):
-        return None
-
     if requirement.should_reinstall:
         logger.info('Attempting uninstall: %s', requirement.name)
         with indent_log():
@@ -135,18 +146,20 @@ def _single_install(
 
     try:
         requirement.install(
-            install_args.install_options,
-            install_args.global_options,
-            **install_args.kwargs
+            **install_args._asdict()
         )
     except (KeyboardInterrupt, SystemExit):
         # always raise, we catch it in external loop
         raise
-    except BaseException as ex:
+    except Exception as ex:
+        # Notice we might need to catch BaseException as this function
+        # can be executed from a subprocess.
+        # For the time being we keep the original catch Exception
+
         # if install did not succeed, rollback previous uninstall
         if uninstalled_pathset and not requirement.install_succeeded:
             uninstalled_pathset.rollback()
-        if in_subprocess:
+        if suppress_exception:
             return ex
         raise
 

From 50dae8d18ebaba52f126d233ee32d80dade54f6c Mon Sep 17 00:00:00 2001
From: bmartinn <>
Date: Sat, 13 Mar 2021 22:53:15 +0200
Subject: [PATCH 14/20] Remove redundant exception and refactor
 should_parallel_reqs

---
 src/pip/_internal/req/__init__.py | 37 +++++++++++++++++--------------
 1 file changed, 20 insertions(+), 17 deletions(-)

diff --git a/src/pip/_internal/req/__init__.py b/src/pip/_internal/req/__init__.py
index 9930aaff775..11880f659c4 100644
--- a/src/pip/_internal/req/__init__.py
+++ b/src/pip/_internal/req/__init__.py
@@ -86,17 +86,24 @@ def install_given_reqs(
 
     with indent_log():
         # get a list of packages we can install in parallel
-        parallel_reqs_index = [
-            i for i, req in enumerate(requirements)
-            if not req.should_reinstall and req.is_wheel]
-
-        # install packages parallel
-        if parallel_reqs_index:
-            parallel_reqs = map_multiprocess_ordered(
-                partial(_single_install, install_args, suppress_exception=True),
-                [requirements[i] for i in parallel_reqs_index])
+        should_parallel_reqs = [
+            (i, req) for i, req in enumerate(requirements)
+            if not req.should_reinstall and req.is_wheel
+        ]
+
+        if should_parallel_reqs:
+            # install packages in parallel
+            should_parallel_indexes, should_parallel_values = zip(
+                *should_parallel_reqs)
+            parallel_reqs_dict = dict(
+                zip(should_parallel_indexes,
+                    map_multiprocess_ordered(
+                        partial(_single_install,
+                                install_args,
+                                suppress_exception=True),
+                        should_parallel_values)))
         else:
-            parallel_reqs = []
+            parallel_reqs_dict = {}
 
         # check the results from the parallel installation,
         # and fill-in missing installations or raise exception
@@ -104,10 +111,9 @@ def install_given_reqs(
 
             # select the install result from the parallel installation
             # or install serially now
-            if parallel_reqs_index and parallel_reqs_index[0] == i:
-                installed_req = parallel_reqs.pop(0)
-                parallel_reqs_index.pop(0)
-            else:
+            try:
+                installed_req = parallel_reqs_dict[i]
+            except KeyError:
                 installed_req = _single_install(
                     install_args, req, suppress_exception=False)
 
@@ -148,9 +154,6 @@ def _single_install(
         requirement.install(
             **install_args._asdict()
         )
-    except (KeyboardInterrupt, SystemExit):
-        # always raise, we catch it in external loop
-        raise
     except Exception as ex:
         # Notice we might need to catch BaseException as this function
         # can be executed from a subprocess.

From 647be535f440a0ceb72d9faf507e564a34efad25 Mon Sep 17 00:00:00 2001
From: bmartinn <>
Date: Mon, 15 Mar 2021 19:21:06 +0200
Subject: [PATCH 15/20] Improve code readability

---
 src/pip/_internal/req/__init__.py | 15 ++++++++-------
 1 file changed, 8 insertions(+), 7 deletions(-)

diff --git a/src/pip/_internal/req/__init__.py b/src/pip/_internal/req/__init__.py
index 11880f659c4..9f3a2bdc46a 100644
--- a/src/pip/_internal/req/__init__.py
+++ b/src/pip/_internal/req/__init__.py
@@ -92,16 +92,17 @@ def install_given_reqs(
         ]
 
         if should_parallel_reqs:
-            # install packages in parallel
+            # prepare for parallel execution
             should_parallel_indexes, should_parallel_values = zip(
                 *should_parallel_reqs)
+            # install packages in parallel
+            executed_parallel_reqs = map_multiprocess_ordered(
+                partial(_single_install, install_args, suppress_exception=True),
+                should_parallel_values
+            )
+            # collect back results
             parallel_reqs_dict = dict(
-                zip(should_parallel_indexes,
-                    map_multiprocess_ordered(
-                        partial(_single_install,
-                                install_args,
-                                suppress_exception=True),
-                        should_parallel_values)))
+                zip(should_parallel_indexes, executed_parallel_reqs))
         else:
             parallel_reqs_dict = {}
 

From de4d416cee554b8d5341cc56d8c0032905c9e180 Mon Sep 17 00:00:00 2001
From: bmartinn <>
Date: Sat, 20 Mar 2021 00:13:31 +0200
Subject: [PATCH 16/20] Add --use-feature=parallel-install (default off), debug
 log, simple test

---
 src/pip/_internal/cli/cmdoptions.py    |  2 +-
 src/pip/_internal/commands/install.py  |  1 +
 src/pip/_internal/req/__init__.py      | 16 ++++++++++++----
 tests/functional/test_install_wheel.py | 18 ++++++++++++++++++
 4 files changed, 32 insertions(+), 5 deletions(-)

diff --git a/src/pip/_internal/cli/cmdoptions.py b/src/pip/_internal/cli/cmdoptions.py
index f71c0b02011..a44adf52624 100644
--- a/src/pip/_internal/cli/cmdoptions.py
+++ b/src/pip/_internal/cli/cmdoptions.py
@@ -965,7 +965,7 @@ def check_list_path_option(options):
     metavar="feature",
     action="append",
     default=[],
-    choices=["2020-resolver", "fast-deps", "in-tree-build"],
+    choices=["2020-resolver", "fast-deps", "in-tree-build", "parallel-install"],
     help="Enable new functionality, that may be backward incompatible.",
 )  # type: Callable[..., Option]
 
diff --git a/src/pip/_internal/commands/install.py b/src/pip/_internal/commands/install.py
index dc637d87635..bf152f600b9 100644
--- a/src/pip/_internal/commands/install.py
+++ b/src/pip/_internal/commands/install.py
@@ -397,6 +397,7 @@ def run(self, options, args):
                 warn_script_location=warn_script_location,
                 use_user_site=options.use_user_site,
                 pycompile=options.compile,
+                parallel_install="parallel-install" in options.features_enabled
             )
 
             lib_locations = get_lib_location_guesses(
diff --git a/src/pip/_internal/req/__init__.py b/src/pip/_internal/req/__init__.py
index 9f3a2bdc46a..1082fa45933 100644
--- a/src/pip/_internal/req/__init__.py
+++ b/src/pip/_internal/req/__init__.py
@@ -54,6 +54,7 @@ def install_given_reqs(
     warn_script_location,  # type: bool
     use_user_site,  # type: bool
     pycompile,  # type: bool
+    parallel_install=False,  # type: bool
 ):
     # type: (...) -> List[InstallationResult]
     """
@@ -86,10 +87,13 @@ def install_given_reqs(
 
     with indent_log():
         # get a list of packages we can install in parallel
-        should_parallel_reqs = [
-            (i, req) for i, req in enumerate(requirements)
-            if not req.should_reinstall and req.is_wheel
-        ]
+        if parallel_install:
+            should_parallel_reqs = [
+                (i, req) for i, req in enumerate(requirements)
+                if not req.should_reinstall and req.is_wheel
+            ]
+        else:
+            should_parallel_reqs = []
 
         if should_parallel_reqs:
             # prepare for parallel execution
@@ -114,6 +118,9 @@ def install_given_reqs(
             # or install serially now
             try:
                 installed_req = parallel_reqs_dict[i]
+                if isinstance(installed_req, InstallationResult):
+                    logger.debug(
+                        'Successfully installed %s in parallel', req.name)
             except KeyError:
                 installed_req = _single_install(
                     install_args, req, suppress_exception=False)
@@ -126,6 +133,7 @@ def install_given_reqs(
                 raise installed_req
             elif isinstance(installed_req, InstallationResult):
                 installed.append(installed_req)
+                logger.debug('Successfully installed %s serially', req.name)
 
     return installed
 
diff --git a/tests/functional/test_install_wheel.py b/tests/functional/test_install_wheel.py
index 8df208bb7da..894af6f79ff 100644
--- a/tests/functional/test_install_wheel.py
+++ b/tests/functional/test_install_wheel.py
@@ -707,3 +707,21 @@ def test_wheel_with_unknown_subdir_in_data_dir_has_reasonable_error(
         "install", "--no-index", str(wheel_path), expect_error=True
     )
     assert "simple-0.1.0.data/unknown/hello.txt" in result.stderr
+
+
+def test_install_from_wheel_in_parallel_installs_deps(script, data, tmpdir):
+    """
+    Test can install dependencies of wheels in parellel
+    """
+    # 'requires_source' depends on the 'source' project
+    package = data.packages.joinpath(
+        "requires_source-1.0-py2.py3-none-any.whl"
+    )
+    shutil.copy(data.packages / "source-1.0.tar.gz", tmpdir)
+    result = script.pip(
+        '-vvv',
+        'install',
+        '--use-feature=parallel-install',
+        '--no-index', '--find-links', tmpdir, package,
+    )
+    result.assert_installed('source', editable=False)

From 18d16d9f9fc270d0c97c45c44884e2ac6475a923 Mon Sep 17 00:00:00 2001
From: bmartinn <>
Date: Sat, 27 Mar 2021 00:10:47 +0300
Subject: [PATCH 17/20] Add test parallel install debug logs

---
 src/pip/_internal/req/__init__.py      |  2 +-
 tests/functional/test_install_wheel.py | 11 +++++++++--
 2 files changed, 10 insertions(+), 3 deletions(-)

diff --git a/src/pip/_internal/req/__init__.py b/src/pip/_internal/req/__init__.py
index 1082fa45933..c39fe70f17a 100644
--- a/src/pip/_internal/req/__init__.py
+++ b/src/pip/_internal/req/__init__.py
@@ -124,6 +124,7 @@ def install_given_reqs(
             except KeyError:
                 installed_req = _single_install(
                     install_args, req, suppress_exception=False)
+                logger.debug('Successfully installed %s serially', req.name)
 
             # Now processes the installation result,
             # throw exception or add into installed packages
@@ -133,7 +134,6 @@ def install_given_reqs(
                 raise installed_req
             elif isinstance(installed_req, InstallationResult):
                 installed.append(installed_req)
-                logger.debug('Successfully installed %s serially', req.name)
 
     return installed
 
diff --git a/tests/functional/test_install_wheel.py b/tests/functional/test_install_wheel.py
index 894af6f79ff..e5c98ad3b34 100644
--- a/tests/functional/test_install_wheel.py
+++ b/tests/functional/test_install_wheel.py
@@ -711,7 +711,7 @@ def test_wheel_with_unknown_subdir_in_data_dir_has_reasonable_error(
 
 def test_install_from_wheel_in_parallel_installs_deps(script, data, tmpdir):
     """
-    Test can install dependencies of wheels in parellel
+    Test can install dependencies of wheels in parallel
     """
     # 'requires_source' depends on the 'source' project
     package = data.packages.joinpath(
@@ -723,5 +723,12 @@ def test_install_from_wheel_in_parallel_installs_deps(script, data, tmpdir):
         'install',
         '--use-feature=parallel-install',
         '--no-index', '--find-links', tmpdir, package,
-    )
+        expect_stderr=True,
+    )
+    log_messages = [
+        msg for msg in result.stdout.split('\n')
+        if msg.lstrip().startswith('Successfully installed ')]
+    assert len(log_messages) == 3
+    assert 'serially' in log_messages[0]
+    assert 'parallel' in log_messages[1]
     result.assert_installed('source', editable=False)

From 5a710628b942b95b2f1e405214d9bc0c61acc6ac Mon Sep 17 00:00:00 2001
From: bmartinn <>
Date: Sun, 4 Apr 2021 00:09:45 +0300
Subject: [PATCH 18/20] Rename variables, add documentation

---
 src/pip/_internal/req/__init__.py | 26 ++++++++++++++++----------
 1 file changed, 16 insertions(+), 10 deletions(-)

diff --git a/src/pip/_internal/req/__init__.py b/src/pip/_internal/req/__init__.py
index c39fe70f17a..0d8d057fef7 100644
--- a/src/pip/_internal/req/__init__.py
+++ b/src/pip/_internal/req/__init__.py
@@ -86,7 +86,13 @@ def install_given_reqs(
     )
 
     with indent_log():
-        # get a list of packages we can install in parallel
+        # get a list of packages we can install in parallel.
+        # we will only select packaged wheels that do not require uninstalling
+        # previous version. This ensures that no console outputs is printed,
+        # making the stdout consistent in parallel execution. It also decreases
+        # the chance of package install failure, as we are only unzipping a
+        # wheel file (no compilation or script execution involved)
+
         if parallel_install:
             should_parallel_reqs = [
                 (i, req) for i, req in enumerate(requirements)
@@ -105,10 +111,10 @@ def install_given_reqs(
                 should_parallel_values
             )
             # collect back results
-            parallel_reqs_dict = dict(
+            parallel_install_results = dict(
                 zip(should_parallel_indexes, executed_parallel_reqs))
         else:
-            parallel_reqs_dict = {}
+            parallel_install_results = {}
 
         # check the results from the parallel installation,
         # and fill-in missing installations or raise exception
@@ -117,23 +123,23 @@ def install_given_reqs(
             # select the install result from the parallel installation
             # or install serially now
             try:
-                installed_req = parallel_reqs_dict[i]
-                if isinstance(installed_req, InstallationResult):
+                install_result = parallel_install_results[i]
+                if isinstance(install_result, InstallationResult):
                     logger.debug(
                         'Successfully installed %s in parallel', req.name)
             except KeyError:
-                installed_req = _single_install(
+                install_result = _single_install(
                     install_args, req, suppress_exception=False)
                 logger.debug('Successfully installed %s serially', req.name)
 
             # Now processes the installation result,
             # throw exception or add into installed packages
-            if isinstance(installed_req, BaseException):
+            if isinstance(install_result, BaseException):
                 # Raise an exception if we caught one
                 # during the parallel installation
-                raise installed_req
-            elif isinstance(installed_req, InstallationResult):
-                installed.append(installed_req)
+                raise install_result
+            elif isinstance(install_result, InstallationResult):
+                installed.append(install_result)
 
     return installed
 

From 5e7332ebf561fed90202a1a55970e3a417c6230b Mon Sep 17 00:00:00 2001
From: bmartinn <>
Date: Wed, 7 Apr 2021 11:53:58 +0300
Subject: [PATCH 19/20] Move freeze_support from __main__ to main()

---
 src/pip/_internal/cli/main.py | 6 ++++++
 1 file changed, 6 insertions(+)

diff --git a/src/pip/_internal/cli/main.py b/src/pip/_internal/cli/main.py
index 7ae074b59d5..6a4da54c6fb 100644
--- a/src/pip/_internal/cli/main.py
+++ b/src/pip/_internal/cli/main.py
@@ -44,6 +44,12 @@
 
 def main(args=None):
     # type: (Optional[List[str]]) -> int
+
+    # windows multiprocessing support
+    from multiprocessing import freeze_support
+
+    freeze_support()
+
     if args is None:
         args = sys.argv[1:]
 

From 274b1081f4728febb0742e31a5df507f4f2a96a6 Mon Sep 17 00:00:00 2001
From: bmartinn <>
Date: Tue, 13 Apr 2021 03:56:36 +0300
Subject: [PATCH 20/20] Remove old comment

---
 src/pip/_internal/req/__init__.py | 1 -
 1 file changed, 1 deletion(-)

diff --git a/src/pip/_internal/req/__init__.py b/src/pip/_internal/req/__init__.py
index 0d8d057fef7..34bcdf2bfba 100644
--- a/src/pip/_internal/req/__init__.py
+++ b/src/pip/_internal/req/__init__.py
@@ -153,7 +153,6 @@ def _single_install(
     """
     Install a single requirement, returns InstallationResult
     (to be called per requirement, either in parallel or serially).
-    Notice the two lists are of the same length
     """
 
     if requirement.should_reinstall: