bpo-35424: emit ResourceWarning at multiprocessing.Pool destruction (GH-10974)
multiprocessing.Pool destructor now emits ResourceWarning
if the pool is still running.
diff --git a/Lib/multiprocessing/pool.py b/Lib/multiprocessing/pool.py
index 1e26a9b..bfb2769 100644
--- a/Lib/multiprocessing/pool.py
+++ b/Lib/multiprocessing/pool.py
@@ -13,13 +13,14 @@
# Imports
#
-import threading
-import queue
-import itertools
import collections
+import itertools
import os
+import queue
+import threading
import time
import traceback
+import warnings
# If threading is available then ThreadPool should be provided. Therefore
# we avoid top-level imports which are liable to fail on some systems.
@@ -30,6 +31,7 @@
# Constants representing the state of a pool
#
+INIT = "INIT"
RUN = "RUN"
CLOSE = "CLOSE"
TERMINATE = "TERMINATE"
@@ -154,11 +156,15 @@
def __init__(self, processes=None, initializer=None, initargs=(),
maxtasksperchild=None, context=None):
+ # Attributes initialized early to make sure that they exist in
+ # __del__() if __init__() raises an exception
+ self._pool = []
+ self._state = INIT
+
self._ctx = context or get_context()
self._setup_queues()
self._taskqueue = queue.SimpleQueue()
self._cache = {}
- self._state = RUN
self._maxtasksperchild = maxtasksperchild
self._initializer = initializer
self._initargs = initargs
@@ -172,7 +178,6 @@
raise TypeError('initializer must be a callable')
self._processes = processes
- self._pool = []
try:
self._repopulate_pool()
except Exception:
@@ -216,6 +221,14 @@
self._result_handler, self._cache),
exitpriority=15
)
+ self._state = RUN
+
+ # Copy globals as function locals to make sure that they are available
+ # during Python shutdown when the Pool is destroyed.
+ def __del__(self, _warn=warnings.warn, RUN=RUN):
+ if self._state == RUN:
+ _warn(f"unclosed running multiprocessing pool {self!r}",
+ ResourceWarning, source=self)
def __repr__(self):
cls = self.__class__
diff --git a/Lib/test/_test_multiprocessing.py b/Lib/test/_test_multiprocessing.py
index b5597d5..7341131 100644
--- a/Lib/test/_test_multiprocessing.py
+++ b/Lib/test/_test_multiprocessing.py
@@ -2577,6 +2577,22 @@
pass
pool.join()
+ def test_resource_warning(self):
+ if self.TYPE == 'manager':
+ self.skipTest("test not applicable to manager")
+
+ pool = self.Pool(1)
+ pool.terminate()
+ pool.join()
+
+ # force state to RUN to emit ResourceWarning in __del__()
+ pool._state = multiprocessing.pool.RUN
+
+ with support.check_warnings(('unclosed running multiprocessing pool',
+ ResourceWarning)):
+ pool = None
+ support.gc_collect()
+
def raising():
raise KeyError("key")