diff --git a/Lib/concurrent/futures/thread.py b/Lib/concurrent/futures/thread.py index 909359b648709fe..82d50da0f14b1d1 100644 --- a/Lib/concurrent/futures/thread.py +++ b/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 diff --git a/Lib/test/test_concurrent_futures/test_thread_pool.py b/Lib/test/test_concurrent_futures/test_thread_pool.py index 226b7acb64fb295..29729c6bc0fc066 100644 --- a/Lib/test/test_concurrent_futures/test_thread_pool.py +++ b/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") diff --git a/Misc/NEWS.d/next/Library/2021-08-31-20-43-30.bpo-44188.JB6g3G.rst b/Misc/NEWS.d/next/Library/2021-08-31-20-43-30.bpo-44188.JB6g3G.rst new file mode 100644 index 000000000000000..e01e869f631fa14 --- /dev/null +++ b/Misc/NEWS.d/next/Library/2021-08-31-20-43-30.bpo-44188.JB6g3G.rst @@ -0,0 +1 @@ +Fix ThreadPoolExecutor unbalanced semaphore count.