bpo-33966, multiprocessing: Pool wait for worker start

Lib/multiprocessing/pool.py

@@ -91,7 +91,7 @@ def __repr__(self):
 
 
 def worker(inqueue, outqueue, initializer=None, initargs=(), maxtasks=None,
-           wrap_exception=False):
+           wrap_exception=False, event=None):
     if (maxtasks is not None) and not (isinstance(maxtasks, int)
                                        and maxtasks >= 1):
         raise AssertionError("Maxtasks {!r} is not valid".format(maxtasks))
@@ -104,6 +104,9 @@ def worker(inqueue, outqueue, initializer=None, initargs=(), maxtasks=None,
     if initializer is not None:
         initializer(*initargs)
 
+    if event is not None:
+        event.set()
+
     completed = 0
     while maxtasks is None or (maxtasks and completed < maxtasks):
         try:
@@ -228,19 +231,31 @@ def _repopulate_pool(self):
         """Bring the number of pool processes up to the specified number,
         for use after reaping workers which have exited.
         """
+        events = []
         for i in range(self._processes - len(self._pool)):
+            event = self._ctx.Event()
+            events.append(event)
             w = self.Process(target=worker,
                              args=(self._inqueue, self._outqueue,
                                    self._initializer,
                                    self._initargs, self._maxtasksperchild,
-                                   self._wrap_exception)
+                                   self._wrap_exception,
+                                   event)
                             )
             self._pool.append(w)
             w.name = w.name.replace('Process', 'PoolWorker')
             w.daemon = True
             w.start()
             util.debug('added worker')
 
+        # bpo-33966: On Windows, if the worker is terminated too quickly,
+        # handles created by DupHandle() on reduction.dump() remain open in
+        # the parent process, causing a handles leak. Wait until all new
+        # workers started to make sure that the workers closed handles
+        # inherited by DupHandle.
+        for event in events:
+            event.wait()
+
     def _maintain_pool(self):
         """Clean up any exited workers and start replacements for them.
         """

Misc/NEWS.d/next/Library/2018-06-27-14-26-32.bpo-33966.aLZJuF.rst

@@ -0,0 +1,4 @@
+multiprocess: On Windows, if a worker is terminated too quickly, handles
+created by DupHandle() on reduction.dump() remain open in the parent process,
+causing a handles leak. Pool now waits using events until all new workers
+started  to make sure that the workers closed handles inherited by DupHandle.