Don't spin up new threads if there are existing idle ones

Fixes #88.

Author: agronholm

CHANGES.rst

@@ -1,3 +1,9 @@
+3.3.0
+=====
+
+- Backported bpo-24882: Let ThreadPoolExecutor reuse idle threads before creating new thread
+
+
 3.2.0
 =====
 

concurrent/futures/thread.py

@@ -75,6 +75,12 @@ def _worker(executor_reference, work_queue):
                 work_item.run()
                 # Delete references to object. See issue16284
                 del work_item
+
+                # attempt to increment idle count
+                executor = executor_reference()
+                if executor is not None:
+                    executor._idle_semaphore.release()
+                del executor
                 continue
             executor = executor_reference()
             # Exit if:
@@ -112,6 +118,7 @@ def __init__(self, max_workers=None, thread_name_prefix=''):
 
         self._max_workers = max_workers
         self._work_queue = queue.Queue()
+        self._idle_semaphore = threading.Semaphore(0)
         self._threads = set()
         self._shutdown = False
         self._shutdown_lock = threading.Lock()
@@ -132,12 +139,15 @@ def submit(self, fn, *args, **kwargs):
     submit.__doc__ = _base.Executor.submit.__doc__
 
     def _adjust_thread_count(self):
+        # if idle threads are available, don't spin new threads
+        if self._idle_semaphore.acquire(False):
+            return
+
         # When the executor gets lost, the weakref callback will wake up
         # the worker threads.
         def weakref_cb(_, q=self._work_queue):
             q.put(None)
-        # TODO(bquinlan): Should avoid creating new threads if there are more
-        # idle threads than items in the work queue.
+
         num_threads = len(self._threads)
         if num_threads < self._max_workers:
             thread_name = '%s_%d' % (self._thread_name_prefix or self,

setup.py

@@ -27,7 +27,7 @@
     readme = f.read()
 
 setup(name='futures',
-      version='3.2.0',
+      version='3.3.0',
       description='Backport of the concurrent.futures package from Python 3',
       long_description=readme,
       author='Brian Quinlan',

test_futures.py

@@ -206,7 +206,15 @@ def test_threads_terminate(self):
         self.executor.submit(mul, 21, 2)
         self.executor.submit(mul, 6, 7)
         self.executor.submit(mul, 3, 14)
+        def acquire_lock(lock):
+            lock.acquire()
+
+        sem = threading.Semaphore(0)
+        for i in range(3):
+            self.executor.submit(acquire_lock, sem)
         self.assertEqual(len(self.executor._threads), 3)
+        for i in range(3):
+            sem.release()
         self.executor.shutdown()
         for t in self.executor._threads:
             t.join()
@@ -529,6 +537,27 @@ def test_default_workers(self):
         self.assertEqual(executor._max_workers,
                          (cpu_count() or 1) * 5)
 
+    def test_saturation(self):
+        executor = self.executor_type(4)
+        def acquire_lock(lock):
+            lock.acquire()
+
+        sem = threading.Semaphore(0)
+        for i in range(15 * executor._max_workers):
+            executor.submit(acquire_lock, sem)
+        self.assertEqual(len(executor._threads), executor._max_workers)
+        for i in range(15 * executor._max_workers):
+            sem.release()
+        executor.shutdown(wait=True)
+
+    def test_idle_thread_reuse(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._threads), 1)
+        executor.shutdown(wait=True)
+
 
 class ProcessPoolExecutorTest(ProcessPoolMixin, ExecutorTest):
     pass