Revert "bpo-34172: multiprocessing.Pool leaks resources after being d… · python/cpython@9dfc754 (original) (raw)
`@@ -149,9 +149,8 @@ class Pool(object):
`
149
149
` '''
`
150
150
`_wrap_exception = True
`
151
151
``
152
``
`-
@staticmethod
`
153
``
`-
def Process(ctx, *args, **kwds):
`
154
``
`-
return ctx.Process(*args, **kwds)
`
``
152
`+
def Process(self, *args, **kwds):
`
``
153
`+
return self._ctx.Process(*args, **kwds)
`
155
154
``
156
155
`def init(self, processes=None, initializer=None, initargs=(),
`
157
156
`maxtasksperchild=None, context=None):
`
`@@ -186,15 +185,13 @@ def init(self, processes=None, initializer=None, initargs=(),
`
186
185
``
187
186
`self._worker_handler = threading.Thread(
`
188
187
`target=Pool._handle_workers,
`
189
``
`-
args=(self._cache, self._taskqueue, self._ctx, self.Process,
`
190
``
`-
self._processes, self._pool, self._inqueue, self._outqueue,
`
191
``
`-
self._initializer, self._initargs, self._maxtasksperchild,
`
192
``
`-
self._wrap_exception)
`
``
188
`+
args=(self, )
`
193
189
` )
`
194
190
`self._worker_handler.daemon = True
`
195
191
`self._worker_handler._state = RUN
`
196
192
`self._worker_handler.start()
`
197
193
``
``
194
+
198
195
`self._task_handler = threading.Thread(
`
199
196
`target=Pool._handle_tasks,
`
200
197
`args=(self._taskqueue, self._quick_put, self._outqueue,
`
`@@ -220,62 +217,43 @@ def init(self, processes=None, initializer=None, initargs=(),
`
220
217
`exitpriority=15
`
221
218
` )
`
222
219
``
223
``
`-
@staticmethod
`
224
``
`-
def _join_exited_workers(pool):
`
``
220
`+
def _join_exited_workers(self):
`
225
221
`"""Cleanup after any worker processes which have exited due to reaching
`
226
222
` their specified lifetime. Returns True if any workers were cleaned up.
`
227
223
` """
`
228
224
`cleaned = False
`
229
``
`-
for i in reversed(range(len(pool))):
`
230
``
`-
worker = pool[i]
`
``
225
`+
for i in reversed(range(len(self._pool))):
`
``
226
`+
worker = self._pool[i]
`
231
227
`if worker.exitcode is not None:
`
232
228
`# worker exited
`
233
229
`util.debug('cleaning up worker %d' % i)
`
234
230
`worker.join()
`
235
231
`cleaned = True
`
236
``
`-
del pool[i]
`
``
232
`+
del self._pool[i]
`
237
233
`return cleaned
`
238
234
``
239
235
`def _repopulate_pool(self):
`
240
``
`-
return self._repopulate_pool_static(self._ctx, self.Process,
`
241
``
`-
self._processes,
`
242
``
`-
self._pool, self._inqueue,
`
243
``
`-
self._outqueue, self._initializer,
`
244
``
`-
self._initargs,
`
245
``
`-
self._maxtasksperchild,
`
246
``
`-
self._wrap_exception)
`
247
``
-
248
``
`-
@staticmethod
`
249
``
`-
def _repopulate_pool_static(ctx, Process, processes, pool, inqueue,
`
250
``
`-
outqueue, initializer, initargs,
`
251
``
`-
maxtasksperchild, wrap_exception):
`
252
236
`"""Bring the number of pool processes up to the specified number,
`
253
237
` for use after reaping workers which have exited.
`
254
238
` """
`
255
``
`-
for i in range(processes - len(pool)):
`
256
``
`-
w = Process(ctx, target=worker,
`
257
``
`-
args=(inqueue, outqueue,
`
258
``
`-
initializer,
`
259
``
`-
initargs, maxtasksperchild,
`
260
``
`-
wrap_exception)
`
261
``
`-
)
`
``
239
`+
for i in range(self._processes - len(self._pool)):
`
``
240
`+
w = self.Process(target=worker,
`
``
241
`+
args=(self._inqueue, self._outqueue,
`
``
242
`+
self._initializer,
`
``
243
`+
self._initargs, self._maxtasksperchild,
`
``
244
`+
self._wrap_exception)
`
``
245
`+
)
`
262
246
`w.name = w.name.replace('Process', 'PoolWorker')
`
263
247
`w.daemon = True
`
264
248
`w.start()
`
265
``
`-
pool.append(w)
`
``
249
`+
self._pool.append(w)
`
266
250
`util.debug('added worker')
`
267
251
``
268
``
`-
@staticmethod
`
269
``
`-
def _maintain_pool(ctx, Process, processes, pool, inqueue, outqueue,
`
270
``
`-
initializer, initargs, maxtasksperchild,
`
271
``
`-
wrap_exception):
`
``
252
`+
def _maintain_pool(self):
`
272
253
`"""Clean up any exited workers and start replacements for them.
`
273
254
` """
`
274
``
`-
if Pool._join_exited_workers(pool):
`
275
``
`-
Pool._repopulate_pool_static(ctx, Process, processes, pool,
`
276
``
`-
inqueue, outqueue, initializer,
`
277
``
`-
initargs, maxtasksperchild,
`
278
``
`-
wrap_exception)
`
``
255
`+
if self._join_exited_workers():
`
``
256
`+
self._repopulate_pool()
`
279
257
``
280
258
`def _setup_queues(self):
`
281
259
`self._inqueue = self._ctx.SimpleQueue()
`
`@@ -433,20 +411,16 @@ def _map_async(self, func, iterable, mapper, chunksize=None, callback=None,
`
433
411
`return result
`
434
412
``
435
413
`@staticmethod
`
436
``
`-
def _handle_workers(cache, taskqueue, ctx, Process, processes, pool,
`
437
``
`-
inqueue, outqueue, initializer, initargs,
`
438
``
`-
maxtasksperchild, wrap_exception):
`
``
414
`+
def _handle_workers(pool):
`
439
415
`thread = threading.current_thread()
`
440
416
``
441
417
`# Keep maintaining workers until the cache gets drained, unless the pool
`
442
418
`# is terminated.
`
443
``
`-
while thread._state == RUN or (cache and thread._state != TERMINATE):
`
444
``
`-
Pool._maintain_pool(ctx, Process, processes, pool, inqueue,
`
445
``
`-
outqueue, initializer, initargs,
`
446
``
`-
maxtasksperchild, wrap_exception)
`
``
419
`+
while thread._state == RUN or (pool._cache and thread._state != TERMINATE):
`
``
420
`+
pool._maintain_pool()
`
447
421
`time.sleep(0.1)
`
448
422
`# send sentinel to stop workers
`
449
``
`-
taskqueue.put(None)
`
``
423
`+
pool._taskqueue.put(None)
`
450
424
`util.debug('worker handler exiting')
`
451
425
``
452
426
`@staticmethod
`
`@@ -828,7 +802,7 @@ class ThreadPool(Pool):
`
828
802
`_wrap_exception = False
`
829
803
``
830
804
`@staticmethod
`
831
``
`-
def Process(ctx, *args, **kwds):
`
``
805
`+
def Process(*args, **kwds):
`
832
806
`from .dummy import Process
`
833
807
`return Process(*args, **kwds)
`
834
808
``