# 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.
# Constants representing the state of a pool
#
+INIT = "INIT"
RUN = "RUN"
CLOSE = "CLOSE"
TERMINATE = "TERMINATE"
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
raise TypeError('initializer must be a callable')
self._processes = processes
- self._pool = []
try:
self._repopulate_pool()
except Exception:
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__
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")