[New-bugs-announce] [issue19096] multiprocessing.Pool._terminate_pool restarts workers during shutdown

[Edward Catmur]

New submission from Edward Catmur:

There is a race condition in multiprocessing.Pool._terminate_pool that can result in workers being restarted during shutdown (process shutdown or pool.terminate()).

        worker_handler._state = TERMINATE    # <~~~~ race from here
        task_handler._state = TERMINATE

        debug('helping task handler/workers to finish')
        cls._help_stuff_finish(inqueue, task_handler, len(pool))

        assert result_handler.is_alive() or len(cache) == 0

        result_handler._state = TERMINATE
        outqueue.put(None)                  # sentinel

        # We must wait for the worker handler to exit before terminating
        # workers because we don't want workers to be restarted behind our back.
        debug('joining worker handler')
        worker_handler.join()                # <~~~~~ race to here

At any point between setting worker_handler._state = TERMINATE and joining the worker handler, if the intervening code causes a worker to exit then it is possible for the worker handler to fail to notice that it has been shutdown and so attempt to restart the worker:

    @staticmethod
    def _handle_workers(pool):
        thread = threading.current_thread()

        # Keep maintaining workers until the cache gets drained, unless the pool
        # is terminated.
        while thread._state == RUN or (pool._cache and thread._state != TERMINATE):
            # <~~~~~~ race here
            pool._maintain_pool()
            time.sleep(0.1)
        # send sentinel to stop workers
        pool._taskqueue.put(None)
        util.debug('worker handler exiting')

We noticed this initially because in the absence of the fix to #14881 a ThreadPool trying to restart a worker fails and hangs the process.  In the presence of the fix to #14881 there is no immediate issue, but trying to restart a worker process/thread on pool shutdown is clearly unwanted and could result in bad things happening e.g. at process shutdown.

To trigger the race with ThreadPool, it is enough just to pause the _handle_workers thread after checking its state and before calling _maintain_pool:

import multiprocessing.pool
import time
class ThreadPool(multiprocessing.pool.ThreadPool):
    def _maintain_pool(self):
        time.sleep(1)
        super(ThreadPool, self)._maintain_pool()
    def _repopulate_pool(self):
        assert self._state == multiprocessing.pool.RUN
        super(ThreadPool, self)._repopulate_pool()
pool = ThreadPool(4)
pool.map(lambda x: x, range(5))
pool.terminate()
pool.join()

Exception in thread Thread-5:
Traceback (most recent call last):
  File ".../cpython/Lib/threading.py", line 657, in _bootstrap_inner
    self.run()
  File ".../cpython/Lib/threading.py", line 605, in run
    self._target(*self._args, **self._kwargs)
  File ".../cpython/Lib/multiprocessing/pool.py", line 358, in _handle_workers
    pool._maintain_pool()
  File ".../bug.py", line 6, in _maintain_pool
    super(ThreadPool, self)._maintain_pool()
  File ".../cpython/Lib/multiprocessing/pool.py", line 232, in _maintain_pool
    self._repopulate_pool()
  File ".../bug.py", line 8, in _repopulate_pool
    assert self._state == multiprocessing.pool.RUN
AssertionError

In this case, the race occurs when ThreadPool._help_stuff_finish puts sentinels on inqueue to make the workers finish.

It is also possible to trigger the bug with multiprocessing.pool.Pool:

import multiprocessing.pool
import time
class Pool(multiprocessing.pool.Pool):
    def _maintain_pool(self):
        time.sleep(2)
        super(Pool, self)._maintain_pool()
    def _repopulate_pool(self):
        assert self._state == multiprocessing.pool.RUN
        super(Pool, self)._repopulate_pool()
    @staticmethod
    def _handle_tasks(taskqueue, put, outqueue, pool):
        time.sleep(1)
        _real_handle_tasks(taskqueue, put, outqueue, pool)
_real_handle_tasks = multiprocessing.pool.Pool._handle_tasks
multiprocessing.pool.Pool._handle_tasks = Pool._handle_tasks
pool = Pool(4)
pool.map(str, range(10))
pool.map_async(str, range(10))
pool.terminate()
pool.join()

In this case, the race occurs when _handle_tasks checks thread._state, breaks out of its first loop, and sends sentinels to the workers.

The terminate/join can be omitted, in which case the bug will occur at gc or process shutdown when the pool's atexit handler runs.  The bug is avoided if terminate is replaced with close, and we are using this workaround.

----------
components: Library (Lib)
messages: 198432
nosy: ecatmur
priority: normal
severity: normal
status: open
title: multiprocessing.Pool._terminate_pool restarts workers during shutdown
type: behavior
versions: Python 2.7, Python 3.5

_______________________________________
Python tracker <report at bugs.python.org>
<http://bugs.python.org/issue19096>
_______________________________________