0
# Event Loop Blocking Detection
1
2
Comprehensive detection of event loop blocking operations with detailed stack trace information. Event loop blocking detection monitors asyncio event loop responsiveness to catch synchronous operations that destroy performance and can cause timeouts.
3
4
## Capabilities
5
6
### Basic Event Loop Blocking Detection
7
8
The primary function for detecting event loop blocking within a specific scope.
9
10
```python { .api }
11
def no_event_loop_blocking(
12
action: LeakAction = LeakAction.WARN,
13
logger: Optional[logging.Logger] = None,
14
*,
15
threshold: float = 0.2,
16
check_interval: float = 0.05,
17
caller_context: CallerContext | None = None,
18
):
19
"""
20
Context manager/decorator that detects event loop blocking within its scope.
21
22
Args:
23
action: Action to take when blocking is detected (LeakAction enum or string)
24
logger: Optional logger instance
25
threshold: Minimum blocking duration to report (seconds)
26
check_interval: How often to check for blocks (seconds)
27
caller_context: Context information for filtering stack traces
28
29
Returns:
30
_EventLoopBlockContextManager: Context manager that can also be used as decorator
31
32
Example:
33
# Basic usage
34
async def main():
35
with no_event_loop_blocking(threshold=0.05):
36
time.sleep(0.1) # This will be detected with stack trace
37
38
# Handle blocking with detailed stack information
39
try:
40
with no_event_loop_blocking(action="raise"):
41
requests.get("https://httpbin.org/delay/1") # Synchronous HTTP call
42
except EventLoopBlockError as e:
43
print(f"Event loop blocked {e.block_count} times")
44
print(e.get_block_summary())
45
46
# As decorator
47
@no_event_loop_blocking(action="raise")
48
async def my_async_function():
49
requests.get("https://example.com") # Synchronous HTTP call
50
"""
51
```
52
53
### Event Loop Block Error Handling
54
55
Exception class for event loop blocking errors with detailed information about blocking events.
56
57
```python { .api }
58
class EventLoopBlockError(LeakError):
59
"""
60
Raised when event loop blocking is detected and action is set to RAISE.
61
62
Attributes:
63
blocking_events: List of EventLoopBlock objects with detailed information
64
block_count: Number of blocking events detected
65
"""
66
def __init__(self, message: str, blocking_events: list[EventLoopBlock])
67
68
blocking_events: list[EventLoopBlock]
69
block_count: int
70
71
def get_block_summary(self) -> str:
72
"""Get a summary of all blocking events."""
73
74
def __str__(self) -> str:
75
"""String representation including blocking event details."""
76
```
77
78
### Event Loop Block Information
79
80
Data class containing detailed information about blocking events.
81
82
```python { .api }
83
class EventLoopBlock:
84
"""
85
Information about an event loop blocking event.
86
87
Attributes:
88
block_id: Unique identifier for the blocking event
89
duration: How long the event loop was blocked (seconds)
90
threshold: The threshold that was exceeded (seconds)
91
timestamp: When the blocking occurred
92
blocking_stack: Stack trace showing what code caused the blocking
93
"""
94
block_id: int
95
duration: float
96
threshold: float
97
timestamp: float
98
blocking_stack: list[traceback.FrameSummary] | None = None
99
100
def format_blocking_stack(self) -> str:
101
"""Format the blocking stack trace as a string."""
102
103
def __str__(self) -> str:
104
"""String representation of the blocking event."""
105
```
106
107
### Caller Context
108
109
Data class for filtering stack traces to relevant code.
110
111
```python { .api }
112
class CallerContext:
113
"""
114
Context information about a caller for stack trace filtering.
115
116
Attributes:
117
filename: File where the caller is located
118
name: Function or method name
119
lineno: Line number (optional)
120
"""
121
filename: str
122
name: str
123
lineno: int | None = None
124
125
def __str__(self) -> str:
126
"""String representation: filename:name:lineno"""
127
```
128
129
## Usage Examples
130
131
### Basic Detection
132
133
```python
134
import asyncio
135
import time
136
from pyleak import no_event_loop_blocking
137
138
async def main():
139
with no_event_loop_blocking(threshold=0.1):
140
# This will be detected as blocking
141
time.sleep(0.2)
142
143
# This will not be detected (below threshold)
144
time.sleep(0.05)
145
```
146
147
### Exception Handling with Stack Traces
148
149
```python
150
import asyncio
151
import time
152
from pyleak import EventLoopBlockError, no_event_loop_blocking
153
154
async def some_function_with_blocking_code():
155
print("starting")
156
time.sleep(1) # Blocking operation
157
print("done")
158
159
async def main():
160
try:
161
with no_event_loop_blocking(action="raise", threshold=0.1):
162
await some_function_with_blocking_code()
163
except EventLoopBlockError as e:
164
print(f"Found {e.block_count} blocking events")
165
for block in e.blocking_events:
166
print(f"Block {block.block_id}: {block.duration:.3f}s")
167
print("Stack trace:")
168
print(block.format_blocking_stack())
169
```
170
171
### Detecting Synchronous HTTP Calls
172
173
```python
174
import asyncio
175
import requests
176
import httpx
177
from pyleak import no_event_loop_blocking
178
179
async def test_sync_vs_async_http():
180
# This will detect blocking
181
with no_event_loop_blocking(action="warn"):
182
response = requests.get("https://httpbin.org/delay/1") # Synchronous!
183
184
# This will not detect blocking
185
with no_event_loop_blocking(action="warn"):
186
async with httpx.AsyncClient() as client:
187
response = await client.get("https://httpbin.org/delay/1") # Asynchronous!
188
```
189
190
### CPU Intensive Operations
191
192
```python
193
import asyncio
194
from pyleak import EventLoopBlockError, no_event_loop_blocking
195
196
async def process_user_data(user_id: int):
197
"""CPU intensive work that blocks the event loop."""
198
print(f"Processing user {user_id}...")
199
return sum(i * i for i in range(100_000_000)) # Blocking computation
200
201
async def main():
202
try:
203
with no_event_loop_blocking(action="raise", threshold=0.5):
204
user1 = await process_user_data(1)
205
user2 = await process_user_data(2)
206
except EventLoopBlockError as e:
207
print(f"Found {e.block_count} blocking events")
208
print("Consider using asyncio.to_thread() or concurrent.futures for CPU-intensive work")
209
```
210
211
### Action Modes
212
213
```python
214
# Warn mode (default) - issues ResourceWarning
215
with no_event_loop_blocking(action="warn"):
216
time.sleep(0.3)
217
218
# Log mode - writes to logger
219
with no_event_loop_blocking(action="log"):
220
time.sleep(0.3)
221
222
# Cancel mode - warns that blocking can't be cancelled
223
with no_event_loop_blocking(action="cancel"):
224
time.sleep(0.3) # Will warn about inability to cancel blocking
225
226
# Raise mode - raises EventLoopBlockError
227
with no_event_loop_blocking(action="raise"):
228
time.sleep(0.3)
229
```
230
231
### Threshold and Check Interval Configuration
232
233
```python
234
from pyleak import no_event_loop_blocking
235
236
# Very sensitive detection (low threshold, frequent checks)
237
with no_event_loop_blocking(threshold=0.01, check_interval=0.001):
238
time.sleep(0.02) # Will be detected
239
240
# Less sensitive detection (higher threshold, less frequent checks)
241
with no_event_loop_blocking(threshold=1.0, check_interval=0.1):
242
time.sleep(0.5) # Will not be detected
243
244
# Default settings (good for most use cases)
245
with no_event_loop_blocking(): # threshold=0.2, check_interval=0.05
246
time.sleep(0.3) # Will be detected
247
```
248
249
### Decorator Usage
250
251
```python
252
@no_event_loop_blocking(action="raise", threshold=0.1)
253
async def my_async_function():
254
# Any blocking operations will cause EventLoopBlockError to be raised
255
time.sleep(0.2) # This will raise an exception
256
```
257
258
### Testing Event Loop Blocking
259
260
```python
261
import pytest
262
import time
263
from pyleak import no_event_loop_blocking, EventLoopBlockError
264
265
@pytest.mark.asyncio
266
async def test_no_blocking():
267
"""Test that ensures no event loop blocking occurs."""
268
with pytest.raises(EventLoopBlockError):
269
with no_event_loop_blocking(action="raise", threshold=0.1):
270
time.sleep(0.2) # This should be detected
271
272
@pytest.mark.asyncio
273
async def test_proper_async_usage():
274
"""Test that properly async code doesn't trigger blocking."""
275
# This should not raise an exception
276
with no_event_loop_blocking(action="raise", threshold=0.1):
277
await asyncio.sleep(0.2) # Proper async operation
278
```
279
280
### Complex Blocking Detection
281
282
```python
283
import asyncio
284
import time
285
from pyleak import EventLoopBlockError, no_event_loop_blocking
286
287
async def debug_blocking():
288
"""Example showing how to debug complex blocking scenarios."""
289
290
def cpu_intensive_work():
291
return sum(i * i for i in range(1_000_000))
292
293
def io_blocking_work():
294
time.sleep(0.1)
295
return "done"
296
297
try:
298
with no_event_loop_blocking(action="raise", threshold=0.05):
299
# Multiple different types of blocking
300
result1 = cpu_intensive_work()
301
result2 = io_blocking_work()
302
303
except EventLoopBlockError as e:
304
print(f"Found {e.block_count} blocking events:")
305
306
for i, block in enumerate(e.blocking_events):
307
print(f"\nBlocking Event {i+1}:")
308
print(f" Duration: {block.duration:.3f}s")
309
print(f" Timestamp: {block.timestamp}")
310
print(" Caused by:")
311
print(" " + "\n ".join(
312
block.format_blocking_stack().strip().split("\n")
313
))
314
315
if __name__ == "__main__":
316
asyncio.run(debug_blocking())
317
```
318
319
### Best Practices
320
321
```python
322
import asyncio
323
import concurrent.futures
324
from pyleak import no_event_loop_blocking
325
326
async def good_async_patterns():
327
"""Examples of proper async patterns that won't block."""
328
329
with no_event_loop_blocking(threshold=0.1):
330
# Use asyncio.sleep instead of time.sleep
331
await asyncio.sleep(0.5)
332
333
# Use asyncio.to_thread for CPU-intensive work
334
result = await asyncio.to_thread(lambda: sum(i*i for i in range(1_000_000)))
335
336
# Use thread pool executor for blocking I/O
337
loop = asyncio.get_event_loop()
338
with concurrent.futures.ThreadPoolExecutor() as executor:
339
result = await loop.run_in_executor(executor, time.sleep, 0.5)
340
341
# Use async libraries for HTTP requests
342
async with httpx.AsyncClient() as client:
343
response = await client.get("https://example.com")
344
```