bpo-19675: Terminate processes if construction of a pool is failing.

Lib/multiprocessing/pool.py

@@ -173,7 +173,15 @@ def __init__(self, processes=None, initializer=None, initargs=(),
 
         self._processes = processes
         self._pool = []
-        self._repopulate_pool()
+        try:
+            self._repopulate_pool()
+        except Exception:
+            for p in self._pool:
+                if p.exitcode is None:
+                    p.terminate()
+            for p in self._pool:
+                p.join()
+            raise
 
         self._worker_handler = threading.Thread(
             target=Pool._handle_workers,
@@ -235,10 +243,10 @@ def _repopulate_pool(self):
                                    self._initargs, self._maxtasksperchild,
                                    self._wrap_exception)
                             )
-            self._pool.append(w)
             w.name = w.name.replace('Process', 'PoolWorker')
             w.daemon = True
             w.start()
+            self._pool.append(w)
             util.debug('added worker')
 
     def _maintain_pool(self):

Lib/test/_test_multiprocessing.py

@@ -3,6 +3,7 @@
 #
 
 import unittest
+import unittest.mock
 import queue as pyqueue
 import contextlib
 import time
@@ -4493,6 +4494,50 @@ def test_empty(self):
 
         proc.join()
 
+
+class TestPoolNotLeakOnFailure(unittest.TestCase):
+
+    def test_release_unused_processes(self):
+        # Issue #19675: During pool creation, if we can't create a process,
+        # don't leak already created ones.
+        will_fail_in = 3
+        forked_processes = []
+
+        class FailingForkProcess():
+            def __init__(self, **kwargs):
+                self.name = 'Fake Process'
+                self.exitcode = None
+                self.state = None
+                forked_processes.append(self)
+
+            def start(self):
+                nonlocal will_fail_in
+                if will_fail_in <= 0:
+                    raise OSError("Manually inducted OSError")
+                will_fail_in -= 1
+                self.state = 'started'
+
+            def terminate(self):
+                self.state = 'stopped'
+
+            def join(self):
+                pass
+
+            def is_alive(self):
+                return self.state == 'started'
+
+        try:
+            p = multiprocessing.pool.Pool(5, context=unittest.mock.MagicMock(
+                Process=FailingForkProcess))
+            p.close()
+            p.join()
+        except OSError as err:
+            if str(err) != 'Manually inducted OSError':
+                raise
+        self.assertFalse(
+            any(process.is_alive() for process in forked_processes))
+
+
 #
 # Mixins
 #

Misc/NEWS.d/next/Library/2018-02-10-23-41-05.bpo-19675.-dj35-.rst

@@ -0,0 +1,2 @@
+multiprocessing.Pool does no longer leak processes if the construction of
+the pool fails.