Skip to content

bpo-39207: Spawn workers on demand in ProcessPoolExecutor #19453

New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Merged
miss-islington merged 7 commits into from
Apr 19, 2020
Merged

bpo-39207: Spawn workers on demand in ProcessPoolExecutor #19453

Show file tree
Hide file tree
Changes from 4 commits
Commits
Show all changes
7 commits
Select commit Hold shift + click to select a range
674574a
Spawn workers on demand in ProcessPoolExecutor
aeros Apr 9, 2020
5e5e4f1
Set _idle_worker_semaphore to None in executor.shutdown()
aeros Apr 10, 2020
04e5927
📜🤖 Added by blurb_it.
blurb-it[bot] Apr 10, 2020
28bb669
Use executor weakref to access semaphore
aeros Apr 10, 2020
04e6b29
Fix race condition in unit tests
aeros Apr 11, 2020
75690c7
Change mp.Semaphore to threading.Semaphore
aeros Apr 18, 2020
d390e51
Several improvements to ProcessPoolExecutor tests.
aeros Apr 18, 2020
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
5 changes: 5 additions & 0 deletions Doc/whatsnew/3.9.rst
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -200,6 +200,11 @@ and :class:`~concurrent.futures.ProcessPoolExecutor`. This improves
compatibility with subinterpreters and predictability in their shutdown
processes. (Contributed by Kyle Stanley in :issue:`39812`.)

Workers in :class:`~concurrent.futures.ProcessPoolExecutor` are now spawned on
demand, only when there are no available idle workers to reuse. This optimizes
startup overhead and reduces the amount of lost CPU time to idle workers.
(Contributed by Kyle Stanley in :issue:`39207`.)

curses
------

Expand Down
16 changes: 14 additions & 2 deletions Lib/concurrent/futures/process.py
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -318,6 +318,12 @@ def run(self):
# while waiting on new results.
del result_item

# attempt to increment idle process count
executor = self.executor_reference()
if executor is not None:
executor._idle_worker_semaphore.release()
del executor

if self.is_shutting_down():
self.flag_executor_shutting_down()

Expand Down Expand Up @@ -601,6 +607,7 @@ def __init__(self, max_workers=None, mp_context=None,
# Shutdown is a two-step process.
self._shutdown_thread = False
self._shutdown_lock = threading.Lock()
self._idle_worker_semaphore = mp_context.Semaphore(0)
Comment thread
aeros marked this conversation as resolved.
Outdated
self._broken = False
self._queue_count = 0
self._pending_work_items = {}
Expand Down Expand Up @@ -633,14 +640,18 @@ def __init__(self, max_workers=None, mp_context=None,
def _start_executor_manager_thread(self):
if self._executor_manager_thread is None:
# Start the processes so that their sentinels are known.
self._adjust_process_count()
self._executor_manager_thread = _ExecutorManagerThread(self)
self._executor_manager_thread.start()
_threads_wakeups[self._executor_manager_thread] = \
self._executor_manager_thread_wakeup

def _adjust_process_count(self):
for _ in range(len(self._processes), self._max_workers):
# if there's an idle process, we don't need to spawn a new one.
if self._idle_worker_semaphore.acquire(block=False):
return

process_count = len(self._processes)
if process_count < self._max_workers:
p = self._mp_context.Process(
target=_process_worker,
args=(self._call_queue,
Expand Down Expand Up @@ -669,6 +680,7 @@ def submit(self, fn, /, *args, **kwargs):
# Wake up queue management thread
self._executor_manager_thread_wakeup.wakeup()

self._adjust_process_count()
self._start_executor_manager_thread()
return f
submit.__doc__ = _base.Executor.submit.__doc__
Expand Down
45 changes: 41 additions & 4 deletions Lib/test/test_concurrent_futures.py
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -486,10 +486,16 @@ def _prime_executor(self):
pass

def test_processes_terminate(self):
self.executor.submit(mul, 21, 2)
self.executor.submit(mul, 6, 7)
self.executor.submit(mul, 3, 14)
self.assertEqual(len(self.executor._processes), 5)
def acquire_lock(lock):
lock.acquire()

mp_context = get_context()
sem = mp_context.Semaphore(0)
for _ in range(3):
self.executor.submit(acquire_lock, sem)
Comment thread
aeros marked this conversation as resolved.
self.assertEqual(len(self.executor._processes), 3)
for _ in range(3):
sem.release()
processes = self.executor._processes
self.executor.shutdown()

Expand Down Expand Up @@ -964,6 +970,37 @@ def test_ressources_gced_in_workers(self):
mgr.shutdown()
mgr.join()

def test_saturation(self):
executor = self.executor_type(4)
def acquire_lock(lock):
lock.aquire()

mp_context = get_context()
sem = mp_context.Semaphore(0)
job_count = 15 * executor._max_workers
for _ in range(job_count):
executor.submit(acquire_lock, sem)
self.assertEqual(len(executor._processes), executor._max_workers)
for _ in range(job_count):
sem.release()
executor.shutdown()
Comment thread
aeros marked this conversation as resolved.
Outdated

def test_idle_process_reuse_one(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._processes), 1)
executor.shutdown()

def test_idle_process_reuse_multiple(self):
executor = self.executor_type()
executor.submit(mul, 12, 7).result()
executor.submit(mul, 33, 25)
executor.submit(mul, 25, 26).result()
executor.submit(mul, 18, 29)
self.assertEqual(len(executor._processes), 2)
Copy link
Copy Markdown
Contributor Author

@aeros aeros Apr 10, 2020 •
edited
Loading

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

This test might be subject to race conditions, as indicated in https://buildbot.python.org/all/#/builders/296/builds/46:

======================================================================
FAIL: test_idle_process_reuse_multiple (test.test_concurrent_futures.ProcessPoolSpawnProcessPoolExecutorTest)
----------------------------------------------------------------------
Traceback (most recent call last):
  File "/root/buildarea/pull_request.angelico-debian-amd64/build/Lib/test/test_concurrent_futures.py", line 1002, in test_idle_process_reuse_multiple
    self.assertEqual(len(executor._processes), 2)
AssertionError: 1 != 2

Instead, I think it would probably make more sense to check self.assertTrue(len(executor._processes) <= 2), after explicitly setting max workers to a specific amount, e.g. executor = self.executor_type(4), and then submitting _max_workers jobs. It's perfectly okay if there are less than 2 workers as a result of the jobs being completed quickly, but there should never be more than two if the idle workers are being properly reused (since we directly waited for the result() on two of them).

I'll take a look at this again tomorrow, after the buildbot tests have completed.

executor.shutdown()

create_executor_tests(ProcessPoolExecutorTest,
executor_mixins=(ProcessPoolForkMixin,
Expand Down
4 changes: 4 additions & 0 deletions Misc/NEWS.d/next/Library/2020-04-10-01-24-58.bpo-39207.2dE5Ox.rst
View file
Open in desktop
Original file line number Diff line number Diff line change
@@ -0,0 +1,4 @@
Workers in :class:`~concurrent.futures.ProcessPoolExecutor` are now spawned on
demand, only when there are no available idle workers to reuse. This optimizes
startup overhead and reduces the amount of lost CPU time to idle workers.
Patch by Kyle Stanley.