Skip to content

Commit 188aabc

Browse files
committed
bpo-29861: release references to multiprocessing Pool tasks (python#743)
* bpo-29861: release references to multiprocessing Pool tasks Release references to tasks, their arguments and their results as soon as they are finished, instead of keeping them alive until another task arrives. * Comments in test (cherry picked from commit 8988945)
1 parent de65804 commit 188aabc

3 files changed

Lines changed: 39 additions & 1 deletion

File tree

‎Lib/multiprocessing/pool.py‎

Lines changed: 6 additions & 1 deletion
Original file line numberDiff line numberDiff line change
@@ -120,6 +120,8 @@ def worker(inqueue, outqueue, initializer=None, initargs=(), maxtasks=None):
120120
debug("Possible encoding error while sending result: %s" % (
121121
wrapped))
122122
put((job, i, (False, wrapped)))
123+
124+
task = job = result = func = args = kwds = None
123125
completed += 1
124126
debug('worker exiting after %d tasks' % completed)
125127

@@ -362,10 +364,11 @@ def _handle_tasks(taskqueue, put, outqueue, pool, cache):
362364
if set_length:
363365
debug('doing set_length()')
364366
set_length(i+1)
367+
finally:
368+
task = taskseq = job = None
365369
else:
366370
debug('task handler got sentinel')
367371

368-
369372
try:
370373
# tell result handler to finish when cache is empty
371374
debug('task handler sending sentinel to result handler')
@@ -405,6 +408,7 @@ def _handle_results(outqueue, get, cache):
405408
cache[job]._set(i, obj)
406409
except KeyError:
407410
pass
411+
task = job = obj = None
408412

409413
while cache and thread._state != TERMINATE:
410414
try:
@@ -421,6 +425,7 @@ def _handle_results(outqueue, get, cache):
421425
cache[job]._set(i, obj)
422426
except KeyError:
423427
pass
428+
task = job = obj = None
424429

425430
if hasattr(outqueue, '_reader'):
426431
debug('ensuring that outqueue is not full')

‎Lib/test/test_multiprocessing.py‎

Lines changed: 28 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -14,6 +14,7 @@
1414
import random
1515
import logging
1616
import errno
17+
import weakref
1718
import test.script_helper
1819
from test import test_support
1920
from StringIO import StringIO
@@ -1123,6 +1124,19 @@ def sqr(x, wait=0.0):
11231124
time.sleep(wait)
11241125
return x*x
11251126

1127+
def identity(x):
1128+
return x
1129+
1130+
class CountedObject(object):
1131+
n_instances = 0
1132+
1133+
def __new__(cls):
1134+
cls.n_instances += 1
1135+
return object.__new__(cls)
1136+
1137+
def __del__(self):
1138+
type(self).n_instances -= 1
1139+
11261140
class SayWhenError(ValueError): pass
11271141

11281142
def exception_throwing_generator(total, when):
@@ -1268,6 +1282,20 @@ def test_empty_iterable(self):
12681282
p.close()
12691283
p.join()
12701284

1285+
def test_release_task_refs(self):
1286+
# Issue #29861: task arguments and results should not be kept
1287+
# alive after we are done with them.
1288+
objs = list(CountedObject() for i in range(10))
1289+
refs = list(weakref.ref(o) for o in objs)
1290+
self.pool.map(identity, objs)
1291+
1292+
del objs
1293+
self.assertEqual(set(wr() for wr in refs), {None})
1294+
# With a process pool, copies of the objects are returned, check
1295+
# they were released too.
1296+
self.assertEqual(CountedObject.n_instances, 0)
1297+
1298+
12711299
def unpickleable_result():
12721300
return lambda: 42
12731301

‎Misc/NEWS‎

Lines changed: 5 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -3122,6 +3122,11 @@ Library
31223122

31233123
- Issue #13163: Rename operands in smtplib.SMTP._get_socket to correct names;
31243124
fixes otherwise misleading output in tracebacks and when when debug is on.
3125+
- bpo-29861: Release references to tasks, their arguments and their results
3126+
as soon as they are finished in multiprocessing.Pool.
3127+
3128+
- bpo-19930: The mode argument of os.makedirs() no longer affects the file
3129+
permission bits of newly-created intermediate-level directories.
31253130

31263131
- Issue #17526: fix an IndexError raised while passing code without filename to
31273132
inspect.findsource(). Initial patch by Tyler Doyle.

0 commit comments

Comments
 (0)