Switched idle count to semaphor, Updated tests

As suggested by reviewer tomMoral, swapped lock-protected counter
with a semaphore to track the number of unused threads.

Adjusted test_threads_terminate to wait for completiton of the
previous future before submitting a new one (and checking the
number of threads used).

Also added a new test to confirm the thread pool can be saturated.

Author: iUnknwn

Lib/concurrent/futures/thread.py

@@ -84,7 +84,7 @@ def _worker(executor_reference, work_queue, initializer, initargs):
                 # attempt to increment idle count
                 executor = executor_reference()
                 if executor is not None:
-                    executor._increase_idle_count()
+                    executor._idle_semaphore.release()
                 del executor
                 continue
 
@@ -140,8 +140,7 @@ def __init__(self, max_workers=None, thread_name_prefix='',
 
         self._max_workers = max_workers
         self._work_queue = queue.SimpleQueue()
-        self._idle_lock = threading.Lock()
-        self._idle_count = 0
+        self._idle_semaphore = threading.Semaphore(0)
         self._threads = set()
         self._broken = False
         self._shutdown = False
@@ -188,10 +187,8 @@ def submit(*args, **kwargs):
 
     def _adjust_thread_count(self):
         #if idle threads are available, don't spin new threads
-        with self._idle_lock:
-            if self._idle_count > 0:
-                self._idle_count -= 1
-                return
+        if self._idle_semaphore.acquire(timeout=0):
+            return
 
         # When the executor gets lost, the weakref callback will wake up
         # the worker threads.
@@ -212,10 +209,6 @@ def weakref_cb(_, q=self._work_queue):
             self._threads.add(t)
             _threads_queues[t] = self._work_queue
 
-    def _increase_idle_count(self):
-        with self._idle_lock:
-            self._idle_count += 1
-
     def _initializer_failed(self):
         with self._shutdown_lock:
             self._broken = ('A thread initializer failed, the thread pool '

Lib/test/test_concurrent_futures.py

@@ -346,10 +346,10 @@ def _prime_executor(self):
         pass
 
     def test_threads_terminate(self):
-        self.executor.submit(mul, 21, 2)
-        self.executor.submit(mul, 6, 7)
-        self.executor.submit(mul, 3, 14)
-        self.assertTrue(len(self.executor._threads) < 3)
+        self.executor.submit(mul, 21, 2).result()
+        self.executor.submit(mul, 6, 7).result()
+        self.executor.submit(mul, 3, 14).result()
+        self.assertTrue(len(self.executor._threads) == 1)
         self.executor.shutdown()
         for t in self.executor._threads:
             t.join()
@@ -753,6 +753,10 @@ def test_default_workers(self):
         self.assertEqual(executor._max_workers,
                          (os.cpu_count() or 1) * 5)
 
+    def test_saturation(self):
+        list(self.executor.map(lambda x: mul(0, x), range(100 * self.executor._max_workers)))
+        self.assertEqual(len(self.executor._threads), self.executor._max_workers)
+
 
 class ProcessPoolExecutorTest(ExecutorTest):