[Python-checkins] cpython: Use WeakSets rather than manual pruning to prevent unbounded growth of dead (original) (raw)

brian.quinlan python-checkins at python.org
Sun Mar 20 03:33:11 CET 2011

Previous message: [Python-checkins] cpython (merge 3.2 -> default): Merge NEWS entry edit.
Next message: [Python-checkins] cpython (merge default -> default): Branch merge
Messages sorted by: [ date ] [ thread ] [ subject ] [ author ]

http://hg.python.org/cpython/rev/a3b061697b18 changeset: 68700:a3b061697b18 parent: 68691:00459310b136 user: Brian Quinlan <brian at sweetapp.com> date: Sun Mar 20 13:11:11 2011 +1100 summary: Use WeakSets rather than manual pruning to prevent unbounded growth of dead thread references.

files: Lib/concurrent/futures/process.py Lib/concurrent/futures/thread.py

diff --git a/Lib/concurrent/futures/process.py b/Lib/concurrent/futures/process.py --- a/Lib/concurrent/futures/process.py +++ b/Lib/concurrent/futures/process.py @@ -66,28 +66,14 @@ # workers to exit when their work queues are empty and then waits until the # threads/processes finish. -_thread_references = set() +_live_threads = weakref.WeakSet() _shutdown = False def _python_exit(): global _shutdown _shutdown = True - for thread_reference in _thread_references: - thread = thread_reference() - if thread is not None: - thread.join()

-def _remove_dead_thread_references(): - """Remove inactive threads from _thread_references.

Should be called periodically to prevent memory leaks in scenarios such as:
while True:
... t = ThreadPoolExecutor(max_workers=5)
... t.map(int, ['1', '2', '3', '4', '5'])
"""
for thread_reference in set(_thread_references):
```
   if thread_reference() is None:
```

       _thread_references.discard(thread_reference)

for thread in _live_threads:
```
   thread.join()
```

Controls how many more calls than processes will be queued in the call queue.

A smaller number will mean that processes spend more time idle waiting for

@@ -279,7 +265,6 @@ worker processes will be created as the machine has processors. """ _check_system_limits() - _remove_dead_thread_references() if max_workers is None: self._max_workers = multiprocessing.cpu_count() @@ -316,7 +301,7 @@ self._shutdown_process_event)) self._queue_management_thread.daemon = True self._queue_management_thread.start() - _thread_references.add(weakref.ref(self._queue_management_thread)) + _live_threads.add(self._queue_management_thread) def _adjust_process_count(self): for _ in range(len(self._processes), self._max_workers): diff --git a/Lib/concurrent/futures/thread.py b/Lib/concurrent/futures/thread.py --- a/Lib/concurrent/futures/thread.py +++ b/Lib/concurrent/futures/thread.py @@ -25,29 +25,14 @@ # workers to exit when their work queues are empty and then waits until the # threads finish. -_thread_references = set() +_live_threads = weakref.WeakSet() _shutdown = False def _python_exit(): global _shutdown _shutdown = True - for thread_reference in _thread_references: - thread = thread_reference() - if thread is not None: - thread.join()

-def _remove_dead_thread_references(): - """Remove inactive threads from _thread_references.

Should be called periodically to prevent memory leaks in scenarios such as:
while True:
... t = ThreadPoolExecutor(max_workers=5)
... t.map(int, ['1', '2', '3', '4', '5'])
"""
for thread_reference in set(_thread_references):
```
   if thread_reference() is None:
```

       _thread_references.discard(thread_reference)

for thread in _live_threads:
```
   thread.join()
```

atexit.register(_python_exit) class _WorkItem(object): @@ -95,8 +80,6 @@ max_workers: The maximum number of threads that can be used to execute the given calls. """ - _remove_dead_thread_references()

     self._max_workers = max_workers
     self._work_queue = queue.Queue()
     self._threads = set()

@@ -125,7 +108,7 @@ t.daemon = True t.start() self._threads.add(t)

       _thread_references.add(weakref.ref(t))

```
       _live_threads.add(t)
```
def shutdown(self, wait=True): with self._shutdown_lock:

-- Repository URL: http://hg.python.org/cpython

Previous message: [Python-checkins] cpython (merge 3.2 -> default): Merge NEWS entry edit.
Next message: [Python-checkins] cpython (merge default -> default): Branch merge
Messages sorted by: [ date ] [ thread ] [ subject ] [ author ]

More information about the Python-checkins mailing list