0
# AsyncIO Task Detection
1
2
Comprehensive detection of asyncio task leaks with detailed stack trace information. Task leak detection helps identify unfinished tasks that could cause memory leaks, prevent graceful shutdown, or make debugging difficult.
3
4
## Capabilities
5
6
### Basic Task Leak Detection
7
8
The primary function for detecting asyncio task leaks within a specific scope.
9
10
```python { .api }
11
def no_task_leaks(
12
action: Union[LeakAction, str] = LeakAction.WARN,
13
name_filter: Optional[Union[str, re.Pattern]] = None,
14
logger: Optional[logging.Logger] = None,
15
*,
16
enable_creation_tracking: bool = False,
17
):
18
"""
19
Context manager/decorator that detects task 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 task names (string or regex)
24
logger: Optional logger instance
25
enable_creation_tracking: Whether to enable automatic task creation tracking
26
27
Returns:
28
_AsyncTaskLeakContextManager: Context manager that can also be used as decorator
29
30
Example:
31
# As context manager
32
async with no_task_leaks():
33
await some_async_function()
34
35
# As decorator
36
@no_task_leaks()
37
async def my_function():
38
await some_async_function()
39
40
# Handle exceptions with full stack traces
41
try:
42
async with no_task_leaks(action="raise"):
43
# Code that leaks tasks
44
pass
45
except TaskLeakError as e:
46
print(f"Found {e.task_count} leaked tasks")
47
for task_info in e.leaked_tasks:
48
if task_info.task_ref and not task_info.task_ref.done():
49
task_info.task_ref.cancel()
50
"""
51
```
52
53
### Task Leak Error Handling
54
55
Exception class for task leak errors with detailed information about leaked tasks.
56
57
```python { .api }
58
class TaskLeakError(LeakError):
59
"""
60
Raised when task leaks are detected and action is set to RAISE.
61
62
Attributes:
63
leaked_tasks: List of LeakedTask objects with detailed information
64
task_count: Number of leaked tasks detected
65
"""
66
def __init__(self, message: str, leaked_tasks: List[LeakedTask])
67
68
leaked_tasks: List[LeakedTask]
69
task_count: int
70
71
def get_stack_summary(self) -> str:
72
"""Get a summary of all stack traces."""
73
74
def __str__(self) -> str:
75
"""String representation including stack traces."""
76
```
77
78
### Leaked Task Information
79
80
Data class containing detailed information about leaked tasks.
81
82
```python { .api }
83
class LeakedTask:
84
"""
85
Information about a leaked asyncio task.
86
87
Attributes:
88
task_id: Unique identifier for the task
89
name: Task name (from task.get_name())
90
state: Current task state (RUNNING, CANCELLED, DONE)
91
current_stack: Stack trace showing where task is currently executing
92
creation_stack: Stack trace showing where task was created (if tracking enabled)
93
task_ref: Reference to the actual asyncio.Task object
94
"""
95
task_id: int
96
name: str
97
state: TaskState
98
current_stack: Optional[List[traceback.FrameSummary]] = None
99
creation_stack: Optional[List[traceback.FrameSummary]] = None
100
task_ref: Optional[asyncio.Task] = None
101
102
@classmethod
103
def from_task(cls, task: asyncio.Task) -> "LeakedTask":
104
"""Create a LeakedTask object from an asyncio.Task."""
105
106
def format_current_stack(self) -> str:
107
"""Format the current stack trace as a string."""
108
109
def format_creation_stack(self) -> str:
110
"""Format the creation stack trace as a string."""
111
112
def __str__(self) -> str:
113
"""String representation of the leaked task."""
114
```
115
116
### Task States
117
118
Enumeration of possible asyncio task states.
119
120
```python { .api }
121
class TaskState(str, Enum):
122
"""State of an asyncio task."""
123
RUNNING = "running"
124
CANCELLED = "cancelled"
125
DONE = "done"
126
```
127
128
## Usage Examples
129
130
### Basic Detection
131
132
```python
133
import asyncio
134
from pyleak import no_task_leaks
135
136
async def main():
137
async with no_task_leaks():
138
# This task will be detected as leaked
139
asyncio.create_task(asyncio.sleep(10))
140
await asyncio.sleep(0.1)
141
```
142
143
### Exception Handling with Stack Traces
144
145
```python
146
import asyncio
147
from pyleak import TaskLeakError, no_task_leaks
148
149
async def leaky_function():
150
async def background_task():
151
print("background task started")
152
await asyncio.sleep(10)
153
154
print("creating a long running task")
155
asyncio.create_task(background_task())
156
157
async def main():
158
try:
159
async with no_task_leaks(action="raise"):
160
await leaky_function()
161
except TaskLeakError as e:
162
print(f"Found {e.task_count} leaked tasks")
163
print(e.get_stack_summary())
164
```
165
166
### Creation Stack Tracking
167
168
```python
169
async def main():
170
try:
171
async with no_task_leaks(action="raise", enable_creation_tracking=True):
172
await leaky_function()
173
except TaskLeakError as e:
174
for task_info in e.leaked_tasks:
175
print(f"Task: {task_info.name}")
176
print("Currently executing:")
177
print(task_info.format_current_stack())
178
print("Created at:")
179
print(task_info.format_creation_stack())
180
```
181
182
### Name Filtering
183
184
```python
185
import re
186
from pyleak import no_task_leaks
187
188
# Filter by exact name
189
async with no_task_leaks(name_filter="background-worker"):
190
pass
191
192
# Filter by regex pattern
193
async with no_task_leaks(name_filter=re.compile(r"worker-\d+")):
194
pass
195
```
196
197
### Action Modes
198
199
```python
200
# Warn mode (default) - issues ResourceWarning
201
async with no_task_leaks(action="warn"):
202
pass
203
204
# Log mode - writes to logger
205
async with no_task_leaks(action="log"):
206
pass
207
208
# Cancel mode - automatically cancels leaked tasks
209
async with no_task_leaks(action="cancel"):
210
pass
211
212
# Raise mode - raises TaskLeakError
213
async with no_task_leaks(action="raise"):
214
pass
215
```
216
217
### Decorator Usage
218
219
```python
220
@no_task_leaks(action="raise")
221
async def my_async_function():
222
# Any leaked tasks will cause TaskLeakError to be raised
223
await some_async_operation()
224
```
225
226
### Manual Task Cleanup
227
228
```python
229
try:
230
async with no_task_leaks(action="raise"):
231
# Code that might leak tasks
232
pass
233
except TaskLeakError as e:
234
print(f"Found {e.task_count} leaked tasks")
235
236
# Cancel leaked tasks manually
237
for task_info in e.leaked_tasks:
238
if task_info.task_ref and not task_info.task_ref.done():
239
task_info.task_ref.cancel()
240
try:
241
await task_info.task_ref
242
except asyncio.CancelledError:
243
pass
244
```