From 6cbf3528518f293600e17c837f0e18e85568a975 Mon Sep 17 00:00:00 2001 From: williams Date: Mon, 20 Jul 2026 22:14:40 +0000 Subject: [PATCH] llm-proxy: propagate client cancellation to the backend (free the slot) MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit A pi-agent stop did not reach models served through the proxy: the forward loop only noticed the client was gone when it next tried to WRITE, so during the blocking non-streaming getresponse() (full generation) the backend ran to completion holding the single-slot admission slot — a cancelled request kept burning the NPU/DGX. Add a client-liveness watchdog: while a backend request is in flight, select() on the client socket; on EOF close the backend conn so it aborts and the slot frees. New _try_backend return "aborted" (no failover; client is gone). _stream_response read loop hardened to stop cleanly when the watchdog closes the conn mid-read. Live-verified: non-streaming request cut at 3s -> CLIENT-CANCELLED logged, slot freed, followup 200 in 4s. Co-Authored-By: Claude Opus 4.8 Claude-Session: https://claude.ai/code/session_01EWpfhDgYNA21tETDP9ueBE --- llm-proxy.py | 70 ++++++++++++++++++++++++++++++++++++++++++++++------ 1 file changed, 62 insertions(+), 8 deletions(-) diff --git a/llm-proxy.py b/llm-proxy.py index 4d6d37a..8173039 100755 --- a/llm-proxy.py +++ b/llm-proxy.py @@ -42,6 +42,8 @@ Signals: import http.server import http.client import ssl +import select +import socket import sys import os import os.path @@ -640,13 +642,16 @@ class Handler(http.server.BaseHTTPRequestHandler): self.wfile.write(chunk) self.wfile.flush() bytes_out += len(chunk) - except (BrokenPipeError, ConnectionResetError): - pass # client disconnected, stop forwarding except http.client.IncompleteRead as e: if e.partial: - self.wfile.write(e.partial) - self.wfile.flush() - bytes_out += len(e.partial) + try: + self.wfile.write(e.partial) + self.wfile.flush() + bytes_out += len(e.partial) + except (OSError, ValueError): + pass + except (OSError, ValueError): + pass # client disconnected, or backend conn closed by the cancel watchdog dt_total = time.monotonic() - t0 dt_ttfb = t_hdr - t_try log(f"{method} {path} -> {backend_name} status={resp.status} ttfb={dt_ttfb:.2f}s total={dt_total:.2f}s bytes={bytes_out} stream={is_streaming}{hint}") @@ -661,6 +666,44 @@ class Handler(http.server.BaseHTTPRequestHandler): except (BrokenPipeError, ConnectionResetError): pass + def _client_watchdog(self, get_conn, gone_evt, stop_evt): + """Watch the CLIENT socket while a backend request is in flight. If the client + hangs up (pi-agent stop / tab close), close the backend connection so the backend + aborts generation and the admission slot is freed — instead of running a single-slot + backend to completion for a request nobody is waiting on. Covers both the streaming + read loop and the blocking non-streaming getresponse(). Returns the started thread.""" + cs = self.connection + + def _run(): + while not stop_evt.is_set(): + try: + r, _, _ = select.select([cs], [], [], 1.0) + except (OSError, ValueError): + return + if stop_evt.is_set() or not r: + continue + try: + peek = cs.recv(1, socket.MSG_PEEK) + except OSError: + peek = b"" + if peek: + continue # client sent data (pipelined) — still there + gone_evt.set() # EOF -> client hung up; cut the backend loose + for _ in range(100): # conn may not exist yet; wait up to ~10s + if stop_evt.is_set(): + return + c = get_conn() + if c is not None: + try: c.close() + except Exception: pass + return + time.sleep(0.1) + return + + t = threading.Thread(target=_run, daemon=True) + t.start() + return t + def _try_backend(self, backend, method, path, body, fwd, t0, hint, is_last, is_streaming=False): name, host, port = backend if not _admit(name): @@ -668,6 +711,9 @@ class Handler(http.server.BaseHTTPRequestHandler): return "busy" t_try = time.monotonic() conn = None + gone = threading.Event() + stop = threading.Event() + self._client_watchdog(lambda: conn, gone, stop) # frees the slot if the client bails try: conn = http.client.HTTPConnection(host, port, timeout=CONNECT_TIMEOUT if not is_last else READ_TIMEOUT) conn.request(method, path, body=body, headers=fwd) @@ -675,16 +721,20 @@ class Handler(http.server.BaseHTTPRequestHandler): self._stream_response(resp, name, t0, t_try, hint, method, path, is_streaming=is_streaming) conn.close() return "ok" - except (ConnectionRefusedError, TimeoutError, OSError) as e: - dt = time.monotonic() - t_try - log(f"{method} {path} -> {name} FAIL {type(e).__name__} {dt:.2f}s") + except (ConnectionRefusedError, TimeoutError, OSError, http.client.HTTPException, ValueError) as e: try: if conn: conn.close() except Exception: pass + if gone.is_set(): # the watchdog closed conn: client cancelled, not a backend fault + log(f"{method} {path} -> {name} CLIENT-CANCELLED, backend cut loose, slot freed after {time.monotonic()-t_try:.2f}s") + return "aborted" + dt = time.monotonic() - t_try + log(f"{method} {path} -> {name} FAIL {type(e).__name__} {dt:.2f}s") return "down" finally: + stop.set() _release(name) def _proxy_local(self, method, path, body, hint, t0, model=None, is_streaming=False): @@ -711,6 +761,8 @@ class Handler(http.server.BaseHTTPRequestHandler): r = self._try_backend(chosen, method, path, body, fwd, t0, hint + " (routed)", is_last=True, is_streaming=is_streaming) if r == "ok": return + if r == "aborted": + return # client hung up; nothing to send, don't fail over if r == "busy": self._send_error(503, b'{"error":{"message":"backend busy (admission gate), retry shortly","type":"backend_busy"}}') return @@ -723,6 +775,8 @@ class Handler(http.server.BaseHTTPRequestHandler): r = self._try_backend(b, method, path, body, fwd, t0, hint, is_last, is_streaming=is_streaming) if r == "ok": return + if r == "aborted": + return # client hung up; nothing to send, don't fail over if r == "busy": busy_seen = True if busy_seen: