Issue #10332: multiprocessing: fix a race condition when a Pool is closed
before all tasks have completed.
This commit is contained in:
parent
22dabb6ffa
commit
46f990e58c
|
@ -294,7 +294,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
|
||||
|
|
|
@ -1168,6 +1168,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
|
||||
#
|
||||
|
|
|
@ -66,6 +66,9 @@ Core and Builtins
|
|||
Library
|
||||
-------
|
||||
|
||||
- Issue #10332: multiprocessing: fix a race condition when a Pool is closed
|
||||
before all tasks have completed.
|
||||
|
||||
- Issue #1548891: The cStringIO.StringIO() constructor now encodes unicode
|
||||
arguments with the system default encoding just like the write() method
|
||||
does, instead of converting it to a raw buffer. This also fixes handling of
|
||||
|
|
Loading…
Reference in New Issue