gh-88354: Fix ThreadPoolExecutor unbalanced semaphore count

Lib/concurrent/futures/thread.py

@@ -108,17 +108,21 @@ def _worker(executor_reference, ctx, work_queue):
             try:
                 work_item = work_queue.get_nowait()
             except queue.Empty:
-                # attempt to increment idle count if queue is empty
-                executor = executor_reference()
-                if executor is not None:
-                    executor._idle_semaphore.release()
-                del executor
                 work_item = work_queue.get(block=True)
 
             if work_item is not None:
                 work_item.run(ctx)
                 # Delete references to object. See GH-60488
                 del work_item
+
+                # Increment the idle count.  This is paired with the
+                # unconditional decrement in _adjust_thread_count(), so
+                # that oversubscription is tracked as a negative count
+                # (gh-88354).
+                executor = executor_reference()
+                if executor is not None:
+                    executor._increase_idle_count()
+                del executor
                 continue
 
             executor = executor_reference()
@@ -188,7 +192,9 @@ def __init__(self, max_workers=None, thread_name_prefix='',
 
         self._max_workers = max_workers
         self._work_queue = queue.SimpleQueue()
-        self._idle_semaphore = threading.Semaphore(0)
+        # An atomic counter would have been sufficient, but we don't have one
+        self._idle_lock = threading.Lock()
+        self._idle_count = 0
         self._threads = set()
         self._broken = False
         self._shutdown = False
@@ -217,8 +223,8 @@ 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(timeout=0):
+        # If idle threads are available, don't spin new threads
+        if self._decrease_idle_count():
             return
 
         # When the executor gets lost, the weakref callback will wake up
@@ -228,6 +234,8 @@ def weakref_cb(_, q=self._work_queue):
 
         num_threads = len(self._threads)
         if num_threads < self._max_workers:
+            # Immediately register an idle thread
+            self._increase_idle_count()
             thread_name = '%s_%d' % (self._thread_name_prefix or self,
                                      num_threads)
             t = threading.Thread(name=thread_name, target=_worker,
@@ -251,6 +259,16 @@ def _initializer_failed(self):
                 if work_item is not None:
                     work_item.future.set_exception(self.BROKEN(self._broken))
 
+    def _decrease_idle_count(self):
+        with self._idle_lock:
+            old_idle = self._idle_count
+            self._idle_count = old_idle - 1
+            return old_idle > 0
+
+    def _increase_idle_count(self):
+        with self._idle_lock:
+            self._idle_count += 1
+
     def shutdown(self, wait=True, *, cancel_futures=False):
         with self._shutdown_lock:
             self._shutdown = True

Lib/test/test_concurrent_futures/test_thread_pool.py

@@ -4,6 +4,7 @@
 import multiprocessing.util
 import os
 import threading
+import time
 import unittest
 from concurrent import futures
 from test import support
@@ -14,6 +15,17 @@
 
 
 class ThreadPoolExecutorTest(ThreadPoolMixin, ExecutorTest, BaseTestCase):
+
+    def wait_for_idle_count(self, executor, expected_count):
+        t1 = time.monotonic()
+        while executor._idle_count < expected_count:
+            if time.monotonic() - t1 > 5:
+                self.fail("idle count not increased after 5 s.")
+            time.sleep(0.01)
+        # Make sure it doesn't change again
+        time.sleep(0.01)
+        self.assertEqual(executor._idle_count, expected_count)
+
     def test_map_submits_without_iteration(self):
         """Tests verifying issue 11777."""
         finished = []
@@ -30,16 +42,47 @@ def test_default_workers(self):
         self.assertEqual(executor._max_workers, expected)
 
     def test_saturation(self):
+        """Test adjusting the number of running workers depending on idleness"""
         executor = self.executor_type(4)
+        self.assertEqual(len(executor._threads), 0)
+
         def acquire_lock(lock):
             lock.acquire()
 
         sem = threading.Semaphore(0)
+
+        for i in range(3):
+            # Launch a single task that will block on `sem`
+            executor.submit(acquire_lock, sem)
+            self.assertEqual(len(executor._threads), 1)
+            self.assertEqual(executor._idle_count, 0)
+            sem.release()
+            self.wait_for_idle_count(executor, 1)
+            self.assertEqual(len(executor._threads), 1)
+
+        # Launch many tasks that will all block on `sem`
         for i in range(15 * executor._max_workers):
             executor.submit(acquire_lock, sem)
+        # As many workers as desired were launched, and none is idle
         self.assertEqual(len(executor._threads), executor._max_workers)
+        self.assertLess(executor._idle_count, 0)  # oversubscription
+        # Allow all tasks to finish
         for i in range(15 * executor._max_workers):
             sem.release()
+        # At some point, the idle count will reach the expected value
+        self.wait_for_idle_count(executor, executor._max_workers)
+
+        # Launch a bunch of blocking tasks again
+        for i in range(2):
+            executor.submit(acquire_lock, sem)
+        self.assertEqual(executor._idle_count, executor._max_workers - 2)
+        for i in range(executor._max_workers):
+            executor.submit(acquire_lock, sem)
+        self.assertLess(executor._idle_count, 0)  # oversubscription
+        for i in range(executor._max_workers + 2):
+            sem.release()
+        self.wait_for_idle_count(executor, executor._max_workers)
+
         executor.shutdown(wait=True)
 
     @support.requires_gil_enabled("gh-117344: test is flaky without the GIL")

Misc/NEWS.d/next/Library/2021-08-31-20-43-30.bpo-44188.JB6g3G.rst

@@ -0,0 +1 @@
+Fix ThreadPoolExecutor unbalanced semaphore count.