williams/llmproxy
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

llm-proxy: propagate client cancellation to the backend (free the slot)

Browse Source 
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 <noreply@anthropic.com>
Claude-Session: https://claude.ai/code/session_01EWpfhDgYNA21tETDP9ueBE
This commit is contained in:
williams
2026-07-20 22:14:40 +00:00
parent 06f20ae752
commit 6cbf352851
1 changed files with 62 additions and 8 deletions
Download Patch File Download Diff File Expand all files Collapse all files
llm-proxy.py
+62 -8
View File
@@ -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: