Spawn workers on demand in ProcessPoolExecutor

aeros · aeros · commit 674574afa29e · 2020-04-09T20:36:00.000-04:00
diff --git a/Doc/whatsnew/3.9.rst b/Doc/whatsnew/3.9.rst
@@ -200,6 +200,11 @@ and :class:`~concurrent.futures.ProcessPoolExecutor`. This improves
 compatibility with subinterpreters and predictability in their shutdown
 processes. (Contributed by Kyle Stanley in :issue:`39812`.)
 
+Workers in :class:`~concurrent.futures.ProcessPoolExecutor` are now spawned on
+demand, only when there are no available idle workers to reuse. This optimizes
+startup overhead and reduces the amount of lost CPU time to idle workers.
+(Contributed by Kyle Stanley in :issue:`39207`.)
+
 curses
 ------
 
diff --git a/Lib/concurrent/futures/process.py b/Lib/concurrent/futures/process.py
@@ -209,7 +209,8 @@ def _sendback_result(result_queue, work_id, result=None, exception=None):
         result_queue.put(_ResultItem(work_id, exception=exc))
 
 
-def _process_worker(call_queue, result_queue, initializer, initargs):
+def _process_worker(call_queue, result_queue, initializer, initargs,
+                    idle_worker_semaphore):
     """Evaluates calls from call_queue and places the results in result_queue.
 
     This worker is run in a separate process.
@@ -221,6 +222,8 @@ def _process_worker(call_queue, result_queue, initializer, initargs):
             to by the worker.
         initializer: A callable initializer, or None
         initargs: A tuple of args for the initializer
+        idle_worker_semaphore: A multiprocessing.Semaphore that is used to
+            prevent new workers from being spawned when there are idle workers.
     """
     if initializer is not None:
         try:
@@ -249,6 +252,8 @@ def _process_worker(call_queue, result_queue, initializer, initargs):
         # open files or shared memory that is not needed anymore
         del call_item
 
+        # increment idle process count after worker finishes job
+        idle_worker_semaphore.release()
 
 class _ExecutorManagerThread(threading.Thread):
     """Manages the communication between this process and the worker processes.
@@ -601,6 +606,7 @@ def __init__(self, max_workers=None, mp_context=None,
         # Shutdown is a two-step process.
         self._shutdown_thread = False
         self._shutdown_lock = threading.Lock()
+        self._idle_worker_semaphore = mp_context.Semaphore(0)
         self._broken = False
         self._queue_count = 0
         self._pending_work_items = {}
@@ -633,20 +639,25 @@ def __init__(self, max_workers=None, mp_context=None,
     def _start_executor_manager_thread(self):
         if self._executor_manager_thread is None:
             # Start the processes so that their sentinels are known.
-            self._adjust_process_count()
             self._executor_manager_thread = _ExecutorManagerThread(self)
             self._executor_manager_thread.start()
             _threads_wakeups[self._executor_manager_thread] = \
                 self._executor_manager_thread_wakeup
 
     def _adjust_process_count(self):
-        for _ in range(len(self._processes), self._max_workers):
+        # if there's an idle process, we don't need to spawn a new one.
+        if self._idle_worker_semaphore.acquire(block=False):
+            return
+
+        process_count = len(self._processes)
+        if process_count < self._max_workers:
             p = self._mp_context.Process(
                 target=_process_worker,
                 args=(self._call_queue,
                       self._result_queue,
                       self._initializer,
-                      self._initargs))
+                      self._initargs,
+                      self._idle_worker_semaphore))
             p.start()
             self._processes[p.pid] = p
 
@@ -669,6 +680,7 @@ def submit(self, fn, /, *args, **kwargs):
             # Wake up queue management thread
             self._executor_manager_thread_wakeup.wakeup()
 
+            self._adjust_process_count()
             self._start_executor_manager_thread()
             return f
     submit.__doc__ = _base.Executor.submit.__doc__
diff --git a/Lib/test/test_concurrent_futures.py b/Lib/test/test_concurrent_futures.py
@@ -486,10 +486,16 @@ def _prime_executor(self):
         pass
 
     def test_processes_terminate(self):
-        self.executor.submit(mul, 21, 2)
-        self.executor.submit(mul, 6, 7)
-        self.executor.submit(mul, 3, 14)
-        self.assertEqual(len(self.executor._processes), 5)
+        def acquire_lock(lock):
+            lock.acquire()
+
+        mp_context = get_context()
+        sem = mp_context.Semaphore(0)
+        for _ in range(3):
+            self.executor.submit(acquire_lock, sem)
+        self.assertEqual(len(self.executor._processes), 3)
+        for _ in range(3):
+            sem.release()
         processes = self.executor._processes
         self.executor.shutdown()
 
@@ -964,6 +970,37 @@ def test_ressources_gced_in_workers(self):
         mgr.shutdown()
         mgr.join()
 
+    def test_saturation(self):
+        executor = self.executor_type(4)
+        def acquire_lock(lock):
+            lock.aquire()
+
+        mp_context = get_context()
+        sem = mp_context.Semaphore(0)
+        job_count = 15 * executor._max_workers
+        for _ in range(job_count):
+            executor.submit(acquire_lock, sem)
+        self.assertEqual(len(executor._processes), executor._max_workers)
+        for _ in range(job_count):
+            sem.release()
+        executor.shutdown()
+
+    def test_idle_process_reuse_one(self):
+        executor = self.executor_type()
+        executor.submit(mul, 21, 2).result()
+        executor.submit(mul, 6, 7).result()
+        executor.submit(mul, 3, 14).result()
+        self.assertEqual(len(executor._processes), 1)
+        executor.shutdown()
+
+    def test_idle_process_reuse_multiple(self):
+        executor = self.executor_type()
+        executor.submit(mul, 12, 7).result()
+        executor.submit(mul, 33, 25)
+        executor.submit(mul, 25, 26).result()
+        executor.submit(mul, 18, 29)
+        self.assertEqual(len(executor._processes), 2)
+        executor.shutdown()
 
 create_executor_tests(ProcessPoolExecutorTest,
                       executor_mixins=(ProcessPoolForkMixin,
