0
# Thread Leak Detection
1
2
Comprehensive detection of thread leaks to help prevent resource exhaustion and ensure proper application termination. Thread leak detection monitors thread lifecycle to identify threads that aren't properly cleaned up.
3
4
## Capabilities
5
6
### Basic Thread Leak Detection
7
8
The primary function for detecting thread leaks within a specific scope.
9
10
```python { .api }
11
def no_thread_leaks(
12
action: str = "warn",
13
name_filter: Optional[Union[str, re.Pattern]] = DEFAULT_THREAD_NAME_FILTER,
14
logger: Optional[logging.Logger] = None,
15
exclude_daemon: bool = True,
16
grace_period: float = 0.1,
17
):
18
"""
19
Context manager/decorator that detects thread leaks within its scope.
20
21
Args:
22
action: Action to take when leaks are detected ("warn", "log", "cancel", "raise")
23
name_filter: Optional filter for thread names (string or regex)
24
logger: Optional logger instance
25
exclude_daemon: Whether to exclude daemon threads from detection
26
grace_period: Time to wait for threads to finish naturally (seconds)
27
28
Returns:
29
_ThreadLeakContextManager: Context manager that can also be used as decorator
30
31
Example:
32
# As context manager
33
with no_thread_leaks():
34
threading.Thread(target=some_function).start()
35
36
# As decorator
37
@no_thread_leaks(action="raise")
38
def my_function():
39
threading.Thread(target=some_work).start()
40
"""
41
```
42
43
### Thread Leak Error Handling
44
45
Exception class for thread leak errors.
46
47
```python { .api }
48
class ThreadLeakError(LeakError):
49
"""Raised when thread leaks are detected and action is set to RAISE."""
50
```
51
52
### Default Thread Name Filter
53
54
Pre-configured regex pattern for filtering thread names.
55
56
```python { .api }
57
DEFAULT_THREAD_NAME_FILTER: re.Pattern
58
# Compiled regex pattern: re.compile(r"^(?!asyncio_\d+$).*")
59
# Excludes asyncio internal threads (names matching "asyncio_\d+")
60
```
61
62
## Usage Examples
63
64
### Basic Detection
65
66
```python
67
import threading
68
import time
69
from pyleak import no_thread_leaks
70
71
def worker_function():
72
time.sleep(5)
73
74
def main():
75
with no_thread_leaks():
76
# This thread will be detected as leaked if it's still running
77
threading.Thread(target=worker_function).start()
78
time.sleep(0.1) # Brief pause, not enough for thread to complete
79
```
80
81
### Exception Handling
82
83
```python
84
import threading
85
from pyleak import ThreadLeakError, no_thread_leaks
86
87
def long_running_worker():
88
time.sleep(10)
89
90
def main():
91
try:
92
with no_thread_leaks(action="raise"):
93
thread = threading.Thread(target=long_running_worker)
94
thread.start()
95
# Thread is still running when context exits
96
except ThreadLeakError as e:
97
print(f"Thread leak detected: {e}")
98
```
99
100
### Name Filtering
101
102
```python
103
import re
104
from pyleak import no_thread_leaks
105
106
# Filter by exact name
107
with no_thread_leaks(name_filter="worker-thread"):
108
thread = threading.Thread(target=some_work, name="worker-thread")
109
thread.start()
110
111
# Filter by regex pattern
112
with no_thread_leaks(name_filter=re.compile(r"background-.*")):
113
thread = threading.Thread(target=some_work, name="background-processor")
114
thread.start()
115
116
# Use default filter (excludes asyncio threads)
117
with no_thread_leaks(): # DEFAULT_THREAD_NAME_FILTER is used
118
pass
119
```
120
121
### Action Modes
122
123
```python
124
# Warn mode (default) - issues ResourceWarning
125
with no_thread_leaks(action="warn"):
126
pass
127
128
# Log mode - writes to logger
129
with no_thread_leaks(action="log"):
130
pass
131
132
# Cancel mode - warns that threads can't be force-stopped
133
with no_thread_leaks(action="cancel"):
134
pass # Will warn about inability to force-stop threads
135
136
# Raise mode - raises ThreadLeakError
137
with no_thread_leaks(action="raise"):
138
pass
139
```
140
141
### Daemon Thread Handling
142
143
```python
144
import threading
145
from pyleak import no_thread_leaks
146
147
def daemon_worker():
148
time.sleep(10)
149
150
def regular_worker():
151
time.sleep(10)
152
153
# Exclude daemon threads (default behavior)
154
with no_thread_leaks(exclude_daemon=True):
155
# This daemon thread will be ignored
156
daemon_thread = threading.Thread(target=daemon_worker, daemon=True)
157
daemon_thread.start()
158
159
# This regular thread will be detected if leaked
160
regular_thread = threading.Thread(target=regular_worker)
161
regular_thread.start()
162
163
# Include daemon threads in detection
164
with no_thread_leaks(exclude_daemon=False):
165
# Both daemon and regular threads will be monitored
166
daemon_thread = threading.Thread(target=daemon_worker, daemon=True)
167
daemon_thread.start()
168
```
169
170
### Grace Period Configuration
171
172
```python
173
from pyleak import no_thread_leaks
174
175
def quick_worker():
176
time.sleep(0.05) # Very quick work
177
178
# Short grace period
179
with no_thread_leaks(grace_period=0.01):
180
threading.Thread(target=quick_worker).start()
181
# May detect thread as leaked due to short grace period
182
183
# Longer grace period
184
with no_thread_leaks(grace_period=0.2):
185
threading.Thread(target=quick_worker).start()
186
# Thread has time to complete naturally
187
```
188
189
### Decorator Usage
190
191
```python
192
@no_thread_leaks(action="raise")
193
def my_threaded_function():
194
# Any leaked threads will cause ThreadLeakError to be raised
195
thread = threading.Thread(target=some_background_work)
196
thread.start()
197
thread.join() # Proper cleanup
198
```
199
200
### Proper Thread Cleanup
201
202
```python
203
import threading
204
from pyleak import no_thread_leaks
205
206
def background_task():
207
time.sleep(2)
208
209
# Proper cleanup pattern
210
def main():
211
with no_thread_leaks(action="raise"):
212
thread = threading.Thread(target=background_task)
213
thread.start()
214
thread.join() # Wait for thread to complete
215
# No leak detected because thread is properly joined
216
```
217
218
### Testing Thread Cleanup
219
220
```python
221
import pytest
222
import threading
223
from pyleak import no_thread_leaks, ThreadLeakError
224
225
def test_thread_cleanup():
226
"""Test that ensures threads are properly cleaned up."""
227
def worker():
228
time.sleep(0.5)
229
230
with pytest.raises(ThreadLeakError):
231
with no_thread_leaks(action="raise", grace_period=0.1):
232
# This thread won't finish in time
233
threading.Thread(target=worker).start()
234
235
def test_proper_thread_management():
236
"""Test that properly managed threads don't trigger leaks."""
237
def worker():
238
time.sleep(0.1)
239
240
# This should not raise an exception
241
with no_thread_leaks(action="raise", grace_period=0.2):
242
thread = threading.Thread(target=worker)
243
thread.start()
244
thread.join()
245
```
246
247
### Custom Thread Name Patterns
248
249
```python
250
import re
251
from pyleak import no_thread_leaks
252
253
# Only monitor specific thread patterns
254
custom_filter = re.compile(r"^(worker|processor|handler)-.*")
255
256
with no_thread_leaks(name_filter=custom_filter, action="raise"):
257
# These threads will be monitored
258
threading.Thread(target=work, name="worker-1").start()
259
threading.Thread(target=process, name="processor-main").start()
260
261
# This thread will be ignored
262
threading.Thread(target=utility, name="utility-helper").start()
263
```