diff options
author | Charles-François Natali <neologix@free.fr> | 2011-10-24 16:47:43 (GMT) |
---|---|---|
committer | Charles-François Natali <neologix@free.fr> | 2011-10-24 16:47:43 (GMT) |
commit | feeb3a366c0971acf1d9a98019a7a60679c41240 (patch) | |
tree | 20fcc7fbd183b5a8493c737ad838568025adf983 | |
parent | b4d8e912d7a9fd2f346aae5038c15f4f1f3dd411 (diff) | |
parent | f8859e1808eff603b727d1adbeb38f745c9fedb5 (diff) | |
download | cpython-feeb3a366c0971acf1d9a98019a7a60679c41240.zip cpython-feeb3a366c0971acf1d9a98019a7a60679c41240.tar.gz cpython-feeb3a366c0971acf1d9a98019a7a60679c41240.tar.bz2 |
Issue #10332: multiprocessing: fix a race condition when a Pool is closed
before all tasks have completed.
-rw-r--r-- | Lib/multiprocessing/pool.py | 6 | ||||
-rw-r--r-- | Lib/test/test_multiprocessing.py | 14 | ||||
-rw-r--r-- | Misc/NEWS | 3 |
3 files changed, 22 insertions, 1 deletions
diff --git a/Lib/multiprocessing/pool.py b/Lib/multiprocessing/pool.py index e450319..04e7c44 100644 --- a/Lib/multiprocessing/pool.py +++ b/Lib/multiprocessing/pool.py @@ -321,7 +321,11 @@ class Pool(object): @staticmethod def _handle_workers(pool): - while pool._worker_handler._state == RUN and pool._state == RUN: + 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): pool._maintain_pool() time.sleep(0.1) # send sentinel to stop workers diff --git a/Lib/test/test_multiprocessing.py b/Lib/test/test_multiprocessing.py index 48a4ff4..0bc056f 100644 --- a/Lib/test/test_multiprocessing.py +++ b/Lib/test/test_multiprocessing.py @@ -1266,6 +1266,20 @@ class _TestPoolWorkerLifetime(BaseTestCase): p.close() p.join() + def test_pool_worker_lifetime_early_close(self): + # Issue #10332: closing a pool whose workers have limited lifetimes + # before all the tasks completed would make join() hang. + p = multiprocessing.Pool(3, maxtasksperchild=1) + results = [] + for i in range(6): + results.append(p.apply_async(sqr, (i, 0.3))) + p.close() + p.join() + # check the results + for (j, res) in enumerate(results): + self.assertEqual(res.get(), sqr(j)) + + # # Test that manager has expected number of shared objects left # @@ -338,6 +338,9 @@ Core and Builtins Library ------- +- Issue #10332: multiprocessing: fix a race condition when a Pool is closed + before all tasks have completed. + - Issue #13255: wrong docstrings in array module. - Issue #8540: Remove deprecated Context._clamp attribute in Decimal module. |