mirrors/cpython
1
0
Fork 0
mirror of https://github.com/python/cpython.git synced 2025-08-03 16:39:00 +00:00
Code Issues Projects Releases Packages Wiki Activity Actions 8

bpo-24882: Let ThreadPoolExecutor reuse idle threads before creating new thread (#6375)

Browse source 
* Fixes issue 24882

* Add news file entry for change.

* Change test_concurrent_futures.ThreadPoolShutdownTest

Adjust the shutdown test so that, after submitting three jobs
to the executor, the test checks for less than three threads,
instead of looking for exactly three threads.

If idle threads are being recycled properly, then we should have
less than three threads.

* 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.

* Updates tests as requested by pitrou.

* Correct minor whitespace error.

* Make test_saturation faster
This commit is contained in:
Sean 2019-05-22 14:29:58 -07:00 • committed by Antoine Pitrou
parent 2a3a2ece50
commit 904e34d4e6
3 changed files with 43 additions and 5 deletions
Download patch file Download diff file Expand all files Collapse all files

32
Lib/test/test_concurrent_futures.py
View file

@ -346,10 +346,15 @@ class ThreadPoolShutdownTest(ThreadPoolMixin, ExecutorShutdownTest, BaseTestCase
pass
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()
@ -753,6 +758,27 @@ class ThreadPoolExecutorTest(ThreadPoolMixin, ExecutorTest, BaseTestCase):
self.assertEqual(executor._max_workers,
(os.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(ExecutorTest):