From 781198cf2ebd2948a15fb8ded0fa7f48c716e10d Mon Sep 17 00:00:00 2001 From: Bennie Swart Date: Thu, 20 May 2021 13:59:15 +0200 Subject: [PATCH 1/3] Fix ThreadPoolExecutor unbalanced semaphore count Previously when the worker pool was over-saturated, the idle count would increment after each completed job whereas it would only decrement when it was already greater than 0. --- Lib/concurrent/futures/thread.py | 6 ++++++ Lib/test/test_concurrent_futures.py | 7 +++++++ 2 files changed, 13 insertions(+) diff --git a/Lib/concurrent/futures/thread.py b/Lib/concurrent/futures/thread.py index b7a2cac7f57015b..0330ac9d453aa29 100644 --- a/Lib/concurrent/futures/thread.py +++ b/Lib/concurrent/futures/thread.py @@ -143,6 +143,7 @@ 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) + self._idle_overcount = 0 self._threads = set() self._broken = False self._shutdown = False @@ -172,6 +173,9 @@ def submit(self, fn, /, *args, **kwargs): submit.__doc__ = _base.Executor.submit.__doc__ def _adjust_thread_count(self): + # consume overcounts first, issue #44188 + while self._idle_overcount > 0 and self._idle_semaphore.acquire(timeout=0): + self._idle_overcount -= 1 # if idle threads are available, don't spin new threads if self._idle_semaphore.acquire(timeout=0): return @@ -193,6 +197,8 @@ def weakref_cb(_, q=self._work_queue): t.start() self._threads.add(t) _threads_queues[t] = self._work_queue + else: + self._idle_overcount += 1 def _initializer_failed(self): with self._shutdown_lock: diff --git a/Lib/test/test_concurrent_futures.py b/Lib/test/test_concurrent_futures.py index 99651f5f4ed4da3..f7bff0035ffb9ba 100644 --- a/Lib/test/test_concurrent_futures.py +++ b/Lib/test/test_concurrent_futures.py @@ -886,8 +886,10 @@ def test_default_workers(self): def test_saturation(self): executor = self.executor_type(4) + submissions_done = threading.Semaphore(0) def acquire_lock(lock): lock.acquire() + submissions_done.release() sem = threading.Semaphore(0) for i in range(15 * executor._max_workers): @@ -895,6 +897,11 @@ def acquire_lock(lock): self.assertEqual(len(executor._threads), executor._max_workers) for i in range(15 * executor._max_workers): sem.release() + for i in range(15 * executor._max_workers): + submissions_done.acquire(blocking=True) + executor.submit(acquire_lock, sem) + self.assertEqual(executor._idle_semaphore._value, executor._max_workers - 1) + sem.release() executor.shutdown(wait=True) def test_idle_thread_reuse(self): From b35748a3510dec1b6759a5222f9f873976d1d913 Mon Sep 17 00:00:00 2001 From: Antoine Pitrou Date: Tue, 31 Aug 2021 20:36:38 +0200 Subject: [PATCH 2/3] Use a locked counter, improve tests a bit --- Lib/concurrent/futures/thread.py | 50 ++++++++++++------- Lib/test/test_concurrent_futures.py | 49 +++++++++++++++--- Tools/c-analyzer/c_parser/_state_machine.py | 2 +- .../c_parser/preprocessor/__init__.py | 2 +- 4 files changed, 77 insertions(+), 26 deletions(-) diff --git a/Lib/concurrent/futures/thread.py b/Lib/concurrent/futures/thread.py index 0330ac9d453aa29..e0a212132f0529d 100644 --- a/Lib/concurrent/futures/thread.py +++ b/Lib/concurrent/futures/thread.py @@ -61,14 +61,24 @@ def run(self): def _worker(executor_reference, work_queue, initializer, initargs): + def run_with_executor(func): + """ + Call the given callable with the executor as argument, if it + is still alive. + """ + executor = executor_reference() + if executor is not None: + try: + func(executor) + finally: + executor = None + if initializer is not None: try: initializer(*initargs) except BaseException: _base.LOGGER.critical('Exception in initializer:', exc_info=True) - executor = executor_reference() - if executor is not None: - executor._initializer_failed() + run_with_executor(lambda executor: executor._initializer_failed()) return try: while True: @@ -78,11 +88,9 @@ def _worker(executor_reference, work_queue, initializer, initargs): # 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 + # increment idle thread count + run_with_executor( + lambda executor: executor._increase_idle_count()) continue executor = executor_reference() @@ -142,8 +150,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) - self._idle_overcount = 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 @@ -173,11 +182,8 @@ def submit(self, fn, /, *args, **kwargs): submit.__doc__ = _base.Executor.submit.__doc__ def _adjust_thread_count(self): - # consume overcounts first, issue #44188 - while self._idle_overcount > 0 and self._idle_semaphore.acquire(timeout=0): - self._idle_overcount -= 1 - # 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 @@ -187,6 +193,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, @@ -197,8 +205,6 @@ def weakref_cb(_, q=self._work_queue): t.start() self._threads.add(t) _threads_queues[t] = self._work_queue - else: - self._idle_overcount += 1 def _initializer_failed(self): with self._shutdown_lock: @@ -213,6 +219,16 @@ def _initializer_failed(self): if work_item is not None: work_item.future.set_exception(BrokenThreadPool(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.py b/Lib/test/test_concurrent_futures.py index f7bff0035ffb9ba..22f31c021b408b3 100644 --- a/Lib/test/test_concurrent_futures.py +++ b/Lib/test/test_concurrent_futures.py @@ -869,6 +869,17 @@ def test_free_reference(self): 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 = [] @@ -885,23 +896,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) - submissions_done = threading.Semaphore(0) + self.assertEqual(len(executor._threads), 0) + def acquire_lock(lock): lock.acquire() - submissions_done.release() 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() - for i in range(15 * executor._max_workers): - submissions_done.acquire(blocking=True) - executor.submit(acquire_lock, sem) - self.assertEqual(executor._idle_semaphore._value, executor._max_workers - 1) - 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) def test_idle_thread_reuse(self): diff --git a/Tools/c-analyzer/c_parser/_state_machine.py b/Tools/c-analyzer/c_parser/_state_machine.py index b505b4e3e47240b..53cbb13e7c4edb3 100644 --- a/Tools/c-analyzer/c_parser/_state_machine.py +++ b/Tools/c-analyzer/c_parser/_state_machine.py @@ -23,7 +23,7 @@ def parse(srclines): if isinstance(srclines, str): # a filename raise NotImplementedError - + # This only handles at most 10 nested levels. #MATCHED_PARENS = textwrap.dedent(rf''' diff --git a/Tools/c-analyzer/c_parser/preprocessor/__init__.py b/Tools/c-analyzer/c_parser/preprocessor/__init__.py index f206f694db5a884..8da4d8cadf7d0ee 100644 --- a/Tools/c-analyzer/c_parser/preprocessor/__init__.py +++ b/Tools/c-analyzer/c_parser/preprocessor/__init__.py @@ -91,7 +91,7 @@ def get_file_preprocessor(filename): macros = list(_resolve_file_values(filename, file_macros)) if file_incldirs: incldirs = [v for v, in _resolve_file_values(filename, file_incldirs)] - + def preprocess(**kwargs): if file_macros and 'macros' not in kwargs: kwargs['macros'] = macros From b937e76a1d652e972ecb10b6bfc227d8205c8db8 Mon Sep 17 00:00:00 2001 From: Antoine Pitrou Date: Tue, 31 Aug 2021 20:43:34 +0200 Subject: [PATCH 3/3] Blurb --- .../NEWS.d/next/Library/2021-08-31-20-43-30.bpo-44188.JB6g3G.rst | 1 + 1 file changed, 1 insertion(+) create mode 100644 Misc/NEWS.d/next/Library/2021-08-31-20-43-30.bpo-44188.JB6g3G.rst 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.