Issue #16284: Prevent keeping unnecessary references to worker functions in concurren...

diff --git a/Lib/concurrent/futures/process.py b/Lib/concurrent/futures/process.py
index 94e02897f39a9867e38978ace726b675d53cce4e..3c20b934c25432e0f5bc8669faa9aab53576597b 100644 (file)
--- a/Lib/concurrent/futures/process.py
+++ b/Lib/concurrent/futures/process.py
@@ -240,6 +240,8 @@ def _queue_management_worker(executor_reference,
                         "terminated abruptly while the future was "
                         "running or pending."
                     ))
+                # Delete references to object. See issue16284
+                del work_item
             pending_work_items.clear()
             # Terminate remaining workers forcibly: the queues or their
             # locks may be in a dirty state and block forever.
@@ -264,6 +266,8 @@ def _queue_management_worker(executor_reference,
                     work_item.future.set_exception(result_item.exception)
                 else:
                     work_item.future.set_result(result_item.result)
+                # Delete references to object. See issue16284
+                del work_item
         # Check whether we should start shutting down.
         executor = executor_reference()
         # No more work items can be added if:

diff --git a/Lib/concurrent/futures/thread.py b/Lib/concurrent/futures/thread.py
index 95bb682565cb7f53f7a30f9066aea8c3a160f2e2..f9beb0f7f7ed3439b1dbf1c0dff861125e27e871 100644 (file)
--- a/Lib/concurrent/futures/thread.py
+++ b/Lib/concurrent/futures/thread.py
@@ -63,6 +63,8 @@ def _worker(executor_reference, work_queue):
             work_item = work_queue.get(block=True)
             if work_item is not None:
                 work_item.run()
+                # Delete references to object. See issue16284
+                del work_item
                 continue
             executor = executor_reference()
             # Exit if:

diff --git a/Lib/multiprocessing/queues.py b/Lib/multiprocessing/queues.py
index 37271fb4eb05f290037a3ffde1ab6329d1cbc5a0..f6f02b66656bd453252b644389e6ffc0f6dbea9c 100644 (file)
--- a/Lib/multiprocessing/queues.py
+++ b/Lib/multiprocessing/queues.py
@@ -243,10 +243,14 @@ class Queue(object):
 
                         if wacquire is None:
                             send(obj)
+                            # Delete references to object. See issue16284
+                            del obj
                         else:
                             wacquire()
                             try:
                                 send(obj)
+                                # Delete references to object. See issue16284
+                                del obj
                             finally:
                                 wrelease()
                 except IndexError:

diff --git a/Lib/test/test_concurrent_futures.py b/Lib/test/test_concurrent_futures.py
index 6ae450df066e9d37d077ac8c8322029974d2dd26..4ad33091b0f1c66cd94b7b628ff2864ef9abcaa4 100644 (file)
--- a/Lib/test/test_concurrent_futures.py
+++ b/Lib/test/test_concurrent_futures.py
@@ -15,6 +15,7 @@ import sys
 import threading
 import time
 import unittest
+import weakref
 
 from concurrent import futures
 from concurrent.futures._base import (
@@ -52,6 +53,11 @@ def sleep_and_print(t, msg):
     sys.stdout.flush()
 
 
+class MyObject(object):
+    def my_method(self):
+        pass
+
+
 class ExecutorMixin:
     worker_count = 5
 
@@ -396,6 +402,22 @@ class ExecutorTest(unittest.TestCase):
         self.executor.map(str, [2] * (self.worker_count + 1))
         self.executor.shutdown()
 
+    @test.support.cpython_only
+    def test_no_stale_references(self):
+        # Issue #16284: check that the executors don't unnecessarily hang onto
+        # references.
+        my_object = MyObject()
+        my_object_collected = threading.Event()
+        my_object_callback = weakref.ref(
+            my_object, lambda obj: my_object_collected.set())
+        # Deliberately discarding the future.
+        self.executor.submit(my_object.my_method)
+        del my_object
+
+        collected = my_object_collected.wait(timeout=5.0)
+        self.assertTrue(collected,
+                        "Stale reference not collected within timeout.")
+
 
 class ThreadPoolExecutorTest(ThreadPoolMixin, ExecutorTest):
     def test_map_submits_without_iteration(self):

diff --git a/Misc/ACKS b/Misc/ACKS
index 96d8178b6348debbb00478d4558e7eb990b26c73..bc5fc613b98317ce370e16011ef9eb689a210266 100644 (file)
--- a/Misc/ACKS
+++ b/Misc/ACKS
@@ -726,6 +726,7 @@ Lukas Lueg
 Loren Luke
 Fredrik Lundh
 Mark Lutz
+Taras Lyapun
 Jim Lynch
 Mikael Lyngvig
 Martin von Löwis

diff --git a/Misc/NEWS b/Misc/NEWS
index c6ba38b0a4a8964a87dde7e60e8bf4e8c23f9546..5ca7e4df6ae36fb622468ac68f8526d0b2521b78 100644 (file)
--- a/Misc/NEWS
+++ b/Misc/NEWS
@@ -88,6 +88,9 @@ Core and Builtins
 Library
 -------
 
+- Issue #16284: Prevent keeping unnecessary references to worker functions
+  in concurrent.futures ThreadPoolExecutor.
+
 - Issue #1207589: Add Cut/Copy/Paste items to IDLE right click Context Menu
   Patch by Todd Rovito.