icmp_win: bound trace_hops_win to a total timeout budget and stop it blocking process exit

subprocess_seconds was being applied per hop instead of as a total trace
budget, so a path with several non-responding hops could block for up to
max_hops * subprocess_seconds. The blocking call also ran on a
ThreadPoolExecutor worker via asyncio.to_thread, which concurrent.futures
joins at process exit - so an orphaned hop stuck past its timeout kept the
whole CLI alive long after diagnose() had already written its report.

Divide the budget across hops (capped at 4s per hop) with an early exit
once the total deadline passes, and dispatch the call on a raw daemon
thread instead so it can never hold the process open even in the worst
case.
This commit is contained in:
Andrew Pimenov 2026-08-09 03:33:12 +03:00
parent 3bf021e1f1
commit 21b71a4056
4 changed files with 164 additions and 4 deletions

View file

@ -145,11 +145,23 @@ def ping_samples_win(host: str, count: int, interval: float, timeout: float) ->
return samples return samples
# A single non-responding hop must never be allowed to eat the whole trace budget:
# cap its timeout even if that leaves room for very few hops.
_MAX_HOP_TIMEOUT_MS = 4000
def trace_hops_win(dest: str, max_hops: int, timeout_ms: int) -> list[tuple[int, IcmpReply]]: def trace_hops_win(dest: str, max_hops: int, timeout_ms: int) -> list[tuple[int, IcmpReply]]:
# timeout_ms is the TOTAL budget for the whole trace (all hops combined), not
# a per-hop timeout - a caller-supplied 60s budget across 30 hops must not turn
# into 30 minutes of blocking IcmpSendEcho2 calls.
address = socket.gethostbyname(dest) address = socket.gethostbyname(dest)
hops: list[tuple[int, IcmpReply]] = [] hops: list[tuple[int, IcmpReply]] = []
per_hop_timeout_ms = max(1, min(timeout_ms // max(max_hops, 1), _MAX_HOP_TIMEOUT_MS))
deadline = time.monotonic() + timeout_ms / 1000.0
for ttl in range(1, max_hops + 1): for ttl in range(1, max_hops + 1):
reply = echo_once(address, ttl=ttl, timeout_ms=timeout_ms) if time.monotonic() >= deadline:
break
reply = echo_once(address, ttl=ttl, timeout_ms=per_hop_timeout_ms)
hops.append((ttl, reply)) hops.append((ttl, reply))
if classify_status(reply.status) == "ok": if classify_status(reply.status) == "ok":
break break

View file

@ -1,15 +1,42 @@
from __future__ import annotations from __future__ import annotations
import asyncio import asyncio
import concurrent.futures
import json import json
import platform import platform
import shutil import shutil
from typing import Awaitable, Callable import threading
from typing import Awaitable, Callable, TypeVar
from netcheck.models import Capabilities, TraceHop, TraceResult from netcheck.models import Capabilities, TraceHop, TraceResult
from netcheck.probes.icmp_win import IcmpReply, classify_status from netcheck.probes.icmp_win import IcmpReply, classify_status
from netcheck.traceparse import build_trace_result, finalize_hop from netcheck.traceparse import build_trace_result, finalize_hop
_T = TypeVar("_T")
def _run_in_daemon_thread(func: Callable[..., _T], *args: object) -> "asyncio.Future[_T]":
# asyncio.to_thread()/loop.run_in_executor(None, ...) queue work onto the
# ThreadPoolExecutor-backed default executor, whose worker threads are non-daemon
# and get joined by concurrent.futures' atexit handler. If the underlying call
# (a blocking WinAPI IcmpSendEcho2) ever outlives its own timeout budget, that
# join blocks process exit indefinitely. A raw daemon thread is never tracked
# by that handler, so it can never hold the process open.
result: concurrent.futures.Future = concurrent.futures.Future()
def _target() -> None:
if not result.set_running_or_notify_cancel():
return
try:
value = func(*args)
except BaseException as exc: # noqa: BLE001 - propagated to the awaiting coroutine
result.set_exception(exc)
else:
result.set_result(value)
threading.Thread(target=_target, name="netcheck-icmp-win", daemon=True).start()
return asyncio.wrap_future(result)
def tier_order(caps: Capabilities, tcp_trace: bool = False) -> list[str]: def tier_order(caps: Capabilities, tcp_trace: bool = False) -> list[str]:
# An explicit --tcp-trace means the user already knows ICMP is being filtered, # An explicit --tcp-trace means the user already knows ICMP is being filtered,
@ -104,7 +131,7 @@ async def _tier_mtr(target: str, binary: str, cycles: int, max_hops: int, timeou
async def _tier_icmp_win(target: str, max_hops: int, timeout: float) -> TraceResult: async def _tier_icmp_win(target: str, max_hops: int, timeout: float) -> TraceResult:
from netcheck.probes.icmp_win import trace_hops_win from netcheck.probes.icmp_win import trace_hops_win
replies = await asyncio.to_thread(trace_hops_win, target, max_hops, int(timeout * 1000)) replies = await _run_in_daemon_thread(trace_hops_win, target, max_hops, int(timeout * 1000))
hops = hops_from_win_replies(replies) hops = hops_from_win_replies(replies)
return TraceResult( return TraceResult(
target=target, target=target,

View file

@ -0,0 +1,78 @@
from __future__ import annotations
import time
from unittest.mock import patch
from netcheck.probes.icmp_win import IP_REQ_TIMED_OUT, IcmpReply, trace_hops_win
def _fake_gethostbyname(host: str) -> str:
return "203.0.113.1"
def test_trace_hops_win_total_elapsed_is_bounded_by_the_total_budget_not_hops_times_timeout():
# Every hop times out; each fake call sleeps for its per-hop timeout to
# simulate a genuinely unresponsive router. Old (buggy) code handed the
# *entire* 100ms total budget to every one of the 30 hops as its own
# per-hop timeout, i.e. ~3s worst case for this scenario. The fix must
# divide that budget across hops instead, keeping total wall time close
# to the 100ms total budget.
call_timeouts: list[int] = []
def fake_echo_once(dest: str, ttl: int, timeout_ms: int, payload: bytes = b"netcheck") -> IcmpReply:
call_timeouts.append(timeout_ms)
time.sleep(timeout_ms / 1000.0)
return IcmpReply(address=None, status=IP_REQ_TIMED_OUT, rtt_ms=None, ttl=0)
with (
patch("netcheck.probes.icmp_win.socket.gethostbyname", side_effect=_fake_gethostbyname),
patch("netcheck.probes.icmp_win.echo_once", side_effect=fake_echo_once),
):
started = time.monotonic()
trace_hops_win("example.invalid", max_hops=30, timeout_ms=100)
elapsed = time.monotonic() - started
# Comfortably under the ~3s the old per-hop-gets-the-whole-budget bug would
# have produced here, while still generous about scheduling jitter.
assert elapsed < 1.0
# Every individual hop got a fraction of the total budget, not the whole thing.
assert call_timeouts
assert all(t == 100 // 30 for t in call_timeouts)
def test_trace_hops_win_per_hop_timeout_is_capped_even_for_a_huge_total_budget():
call_timeouts: list[int] = []
def fake_echo_once(dest: str, ttl: int, timeout_ms: int, payload: bytes = b"netcheck") -> IcmpReply:
call_timeouts.append(timeout_ms)
return IcmpReply(address=None, status=IP_REQ_TIMED_OUT, rtt_ms=None, ttl=0)
with (
patch("netcheck.probes.icmp_win.socket.gethostbyname", side_effect=_fake_gethostbyname),
patch("netcheck.probes.icmp_win.echo_once", side_effect=fake_echo_once),
):
# A single-hop trace with a huge total budget (e.g. subprocess_seconds=60s
# divided across a small max_hops) must still not hand the whole budget
# to one hop - it's capped at _MAX_HOP_TIMEOUT_MS (4s).
trace_hops_win("example.invalid", max_hops=1, timeout_ms=60_000)
assert call_timeouts == [4000]
def test_trace_hops_win_happy_path_hop_count_and_ok_break_are_unchanged():
from netcheck.probes.icmp_win import IP_SUCCESS
def fake_echo_once(dest: str, ttl: int, timeout_ms: int, payload: bytes = b"netcheck") -> IcmpReply:
if ttl < 3:
return IcmpReply(address=f"10.0.0.{ttl}", status=IP_REQ_TIMED_OUT, rtt_ms=None, ttl=0)
return IcmpReply(address="203.0.113.1", status=IP_SUCCESS, rtt_ms=5.0, ttl=64)
with (
patch("netcheck.probes.icmp_win.socket.gethostbyname", side_effect=_fake_gethostbyname),
patch("netcheck.probes.icmp_win.echo_once", side_effect=fake_echo_once),
):
hops = trace_hops_win("example.invalid", max_hops=30, timeout_ms=60_000)
# Fast-responding hops still stop the loop at the first success, same as before.
assert [ttl for ttl, _ in hops] == [1, 2, 3]
assert hops[-1][1].status == IP_SUCCESS

View file

@ -1,12 +1,13 @@
from __future__ import annotations from __future__ import annotations
import asyncio import asyncio
import threading
import pytest import pytest
from netcheck.models import Capabilities, TraceHop, TraceResult from netcheck.models import Capabilities, TraceHop, TraceResult
from netcheck.probes.icmp_win import IP_REQ_TIMED_OUT, IP_SUCCESS, IP_TTL_EXPIRED_TRANSIT, IcmpReply from netcheck.probes.icmp_win import IP_REQ_TIMED_OUT, IP_SUCCESS, IP_TTL_EXPIRED_TRANSIT, IcmpReply
from netcheck.probes.traceroute import hops_from_win_replies, run_cascade, tier_order from netcheck.probes.traceroute import _run_in_daemon_thread, hops_from_win_replies, run_cascade, tier_order
def caps(**kw) -> Capabilities: def caps(**kw) -> Capabilities:
@ -161,3 +162,45 @@ def test_win_replies_become_hops_with_ttl_expiry_handled():
assert hops[1].loss_pct == 100.0 assert hops[1].loss_pct == 100.0
assert hops[2].ip == "1.1.1.1" assert hops[2].ip == "1.1.1.1"
assert hops[2].avg_ms == 12.0 assert hops[2].avg_ms == 12.0
async def test_run_in_daemon_thread_returns_the_function_result():
result = await _run_in_daemon_thread(lambda a, b: a + b, 2, 3)
assert result == 5
async def test_run_in_daemon_thread_propagates_exceptions():
def boom():
raise ValueError("nope")
with pytest.raises(ValueError, match="nope"):
await _run_in_daemon_thread(boom)
def test_run_in_daemon_thread_never_creates_a_non_daemon_worker_thread():
# This is the crux of the process-exit hang fix: asyncio.to_thread()/the
# default executor spawn non-daemon ThreadPoolExecutor workers that get
# joined by concurrent.futures' atexit handler, which is exactly what let
# an orphaned, uncancellable IcmpSendEcho2 call keep the whole process
# alive. We can't unit-test "process exit is never blocked" directly (that
# needs a real subprocess), but we can pin the one property that makes it
# true: every thread this helper spawns is a plain daemon thread, never
# routed through a ThreadPoolExecutor.
seen: list[threading.Thread] = []
original_init = threading.Thread.__init__
def spy_init(self, *args, **kwargs):
original_init(self, *args, **kwargs)
seen.append(self)
async def run():
await _run_in_daemon_thread(lambda: None)
threading.Thread.__init__ = spy_init
try:
asyncio.run(run())
finally:
threading.Thread.__init__ = original_init
assert seen
assert all(t.daemon for t in seen)