diff options
| -rw-r--r-- | Lib/multiprocessing/pool.py | 7 | ||||
| -rw-r--r-- | Lib/test/_test_multiprocessing.py | 28 | ||||
| -rw-r--r-- | Misc/NEWS | 3 |
3 files changed, 37 insertions, 1 deletions
diff --git a/Lib/multiprocessing/pool.py b/Lib/multiprocessing/pool.py index ffdf426..ae8cec4 100644 --- a/Lib/multiprocessing/pool.py +++ b/Lib/multiprocessing/pool.py @@ -128,6 +128,8 @@ def worker(inqueue, outqueue, initializer=None, initargs=(), maxtasks=None, util.debug("Possible encoding error while sending result: %s" % ( wrapped)) put((job, i, (False, wrapped))) + + task = job = result = func = args = kwds = None completed += 1 util.debug('worker exiting after %d tasks' % completed) @@ -402,10 +404,11 @@ class Pool(object): if set_length: util.debug('doing set_length()') set_length(i+1) + finally: + task = taskseq = job = None else: util.debug('task handler got sentinel') - try: # tell result handler to finish when cache is empty util.debug('task handler sending sentinel to result handler') @@ -445,6 +448,7 @@ class Pool(object): cache[job]._set(i, obj) except KeyError: pass + task = job = obj = None while cache and thread._state != TERMINATE: try: @@ -461,6 +465,7 @@ class Pool(object): cache[job]._set(i, obj) except KeyError: pass + task = job = obj = None if hasattr(outqueue, '_reader'): util.debug('ensuring that outqueue is not full') diff --git a/Lib/test/_test_multiprocessing.py b/Lib/test/_test_multiprocessing.py index b5f4782..1d3bb0f 100644 --- a/Lib/test/_test_multiprocessing.py +++ b/Lib/test/_test_multiprocessing.py @@ -18,6 +18,7 @@ import random import logging import struct import operator +import weakref import test.support import test.support.script_helper @@ -1738,6 +1739,19 @@ def raise_large_valuerror(wait): time.sleep(wait) raise ValueError("x" * 1024**2) +def identity(x): + return x + +class CountedObject(object): + n_instances = 0 + + def __new__(cls): + cls.n_instances += 1 + return object.__new__(cls) + + def __del__(self): + type(self).n_instances -= 1 + class SayWhenError(ValueError): pass def exception_throwing_generator(total, when): @@ -1746,6 +1760,7 @@ def exception_throwing_generator(total, when): raise SayWhenError("Somebody said when") yield i + class _TestPool(BaseTestCase): @classmethod @@ -2000,6 +2015,19 @@ class _TestPool(BaseTestCase): # check that we indeed waited for all jobs self.assertGreater(time.time() - t_start, 0.9) + def test_release_task_refs(self): + # Issue #29861: task arguments and results should not be kept + # alive after we are done with them. + objs = [CountedObject() for i in range(10)] + refs = [weakref.ref(o) for o in objs] + self.pool.map(identity, objs) + + del objs + self.assertEqual(set(wr() for wr in refs), {None}) + # With a process pool, copies of the objects are returned, check + # they were released too. + self.assertEqual(CountedObject.n_instances, 0) + def raising(): raise KeyError("key") @@ -287,6 +287,9 @@ Extension Modules Library ------- +- bpo-29861: Release references to tasks, their arguments and their results + as soon as they are finished in multiprocessing.Pool. + - bpo-19930: The mode argument of os.makedirs() no longer affects the file permission bits of newly-created intermediate-level directories. |