From 188aabccd2d1252edb434bf6fe77ac91ba96329f Mon Sep 17 00:00:00 2001
From: Antoine Pitrou <pitrou@free.fr>
Date: Fri, 24 Mar 2017 13:52:11 +0100
Subject: [PATCH 1/2] bpo-29861: release references to multiprocessing Pool
 tasks (#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 8988945cdc27ffa86ba8c624e095b51c459f5154)
---
 Lib/multiprocessing/pool.py      |  7 ++++++-
 Lib/test/test_multiprocessing.py | 28 ++++++++++++++++++++++++++++
 Misc/NEWS                        |  5 +++++
 3 files changed, 39 insertions(+), 1 deletion(-)

diff --git a/Lib/multiprocessing/pool.py b/Lib/multiprocessing/pool.py
index ceb93aab8624ef..a47cd0f58a05af 100644
--- a/Lib/multiprocessing/pool.py
+++ b/Lib/multiprocessing/pool.py
@@ -120,6 +120,8 @@ def worker(inqueue, outqueue, initializer=None, initargs=(), maxtasks=None):
             debug("Possible encoding error while sending result: %s" % (
                 wrapped))
             put((job, i, (False, wrapped)))
+
+        task = job = result = func = args = kwds = None
         completed += 1
     debug('worker exiting after %d tasks' % completed)
 
@@ -362,10 +364,11 @@ def _handle_tasks(taskqueue, put, outqueue, pool, cache):
                 if set_length:
                     debug('doing set_length()')
                     set_length(i+1)
+            finally:
+                task = taskseq = job = None
         else:
             debug('task handler got sentinel')
 
-
         try:
             # tell result handler to finish when cache is empty
             debug('task handler sending sentinel to result handler')
@@ -405,6 +408,7 @@ def _handle_results(outqueue, get, cache):
                 cache[job]._set(i, obj)
             except KeyError:
                 pass
+            task = job = obj = None
 
         while cache and thread._state != TERMINATE:
             try:
@@ -421,6 +425,7 @@ def _handle_results(outqueue, get, cache):
                 cache[job]._set(i, obj)
             except KeyError:
                 pass
+            task = job = obj = None
 
         if hasattr(outqueue, '_reader'):
             debug('ensuring that outqueue is not full')
diff --git a/Lib/test/test_multiprocessing.py b/Lib/test/test_multiprocessing.py
index c66727cf3ae52e..163c42f1d10a17 100644
--- a/Lib/test/test_multiprocessing.py
+++ b/Lib/test/test_multiprocessing.py
@@ -14,6 +14,7 @@
 import random
 import logging
 import errno
+import weakref
 import test.script_helper
 from test import test_support
 from StringIO import StringIO
@@ -1123,6 +1124,19 @@ def sqr(x, wait=0.0):
     time.sleep(wait)
     return x*x
 
+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):
@@ -1268,6 +1282,20 @@ def test_empty_iterable(self):
         p.close()
         p.join()
 
+    def test_release_task_refs(self):
+        # Issue #29861: task arguments and results should not be kept
+        # alive after we are done with them.
+        objs = list(CountedObject() for i in range(10))
+        refs = list(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 unpickleable_result():
     return lambda: 42
 
diff --git a/Misc/NEWS b/Misc/NEWS
index 2b492f406cb85a..54634d7a96bfaf 100644
--- a/Misc/NEWS
+++ b/Misc/NEWS
@@ -3122,6 +3122,11 @@ Library
 
 - Issue #13163: Rename operands in smtplib.SMTP._get_socket to correct names;
   fixes otherwise misleading output in tracebacks and when when debug is on.
+- 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.
 
 - Issue #17526: fix an IndexError raised while passing code without filename to
   inspect.findsource().  Initial patch by Tyler Doyle.

From fb030c0d8a684970968176d283f0af9706ad4d4f Mon Sep 17 00:00:00 2001
From: Antoine Pitrou <antoine@python.org>
Date: Fri, 24 Mar 2017 15:54:26 +0100
Subject: [PATCH 2/2] Fix Misc/NEWS ?

---
 Misc/NEWS | 8 +++-----
 1 file changed, 3 insertions(+), 5 deletions(-)

diff --git a/Misc/NEWS b/Misc/NEWS
index 54634d7a96bfaf..de5ef7dc484c37 100644
--- a/Misc/NEWS
+++ b/Misc/NEWS
@@ -39,6 +39,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-27880: Fixed integer overflow in cPickle when pickle large strings or
   too many objects.
 
@@ -3122,11 +3125,6 @@ Library
 
 - Issue #13163: Rename operands in smtplib.SMTP._get_socket to correct names;
   fixes otherwise misleading output in tracebacks and when when debug is on.
-- 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.
 
 - Issue #17526: fix an IndexError raised while passing code without filename to
   inspect.findsource().  Initial patch by Tyler Doyle.