fix(abort): force-wake blocked stream recv via socket _real_close; interruptible retry backoff
This commit is contained in:
1 parent
13eef203f7
commit
3d62523d40
2 files changed
+23
-2
No files matched your search
@@ -116,6 +116,17 @@ class GenericAgent:
|
||||
if self.handler is not None: self.handler.code_stop_signal.append(1)
|
||||
for sess in getattr(self.llmclient.backend, '_sessions', [self.llmclient.backend]):
|
||||
sess.should_stop = lambda: self.stop_sig # live read; cleared by run()'s finally
|
||||
try: # wake a recv() blocked in another thread. Verified on Windows: shutdown()/close() do NOT
|
||||
# wake it (makefile refcount defers real closesocket); _real_close() does -> ChunkedEncodingError
|
||||
import socket as _socket
|
||||
raw = sess.active_response.raw
|
||||
fp = getattr(getattr(raw, '_fp', None), 'fp', None) # http.client response -> buffered socket file
|
||||
sock = fp.raw._sock if fp else raw.connection.sock # SocketIO._sock (SSL-wrapped OK); fallback urllib3 conn
|
||||
try: sock.shutdown(_socket.SHUT_RDWR) # for non-Windows semantics
|
||||
except OSError: pass
|
||||
try: sock._real_close() # CPython internal; bypasses refcount -> actual closesocket
|
||||
except AttributeError: sock.close()
|
||||
except Exception: pass
|
||||
try: sess.active_response.close()
|
||||
except Exception: pass
|
||||
|
||||
|
||||
+12
-2
@@ -446,7 +446,15 @@ def _stream_with_retry(sess, url, headers, payload, parse_fn):
|
||||
try: ra = float((resp.headers or {}).get("retry-after"))
|
||||
except: ra = None
|
||||
return None if ra and ra > cap else max(0.5, ra or min(30.0, 3.0 * (2 ** attempt)))
|
||||
def _stopped(): return getattr(sess, 'should_stop', None) and sess.should_stop()
|
||||
def _sleep(d): # interruptible sleep; True if aborted
|
||||
end = time.time() + d
|
||||
while time.time() < end:
|
||||
if _stopped(): return True
|
||||
time.sleep(0.2)
|
||||
return _stopped()
|
||||
for attempt in range(sess.max_retries + 1):
|
||||
if _stopped(): return []
|
||||
streamed = False
|
||||
STATS.update(t_start=time.time(), t_ttft=None)
|
||||
if not sess.stream: STATS['t_ttft'] = STATS['t_start']
|
||||
@@ -459,7 +467,8 @@ def _stream_with_retry(sess, url, headers, payload, parse_fn):
|
||||
d = _delay(r, attempt) if r.status_code in _RETRYABLE and attempt < sess.max_retries else None
|
||||
if d is not None:
|
||||
print(f"[LLM Retry] HTTP {r.status_code}, retry in {d:.1f}s ({attempt+1}/{sess.max_retries+1})")
|
||||
time.sleep(d); continue
|
||||
if _sleep(d): return []
|
||||
continue
|
||||
try: body = r.text.strip()[:500]
|
||||
except: body = ""
|
||||
err = f"!!!Error: HTTP {r.status_code}" + (f" (retry-after > {cap:.0f}s)" if d is None and r.status_code in _RETRYABLE and attempt < sess.max_retries else "") + (f": {body}" if body else "")
|
||||
@@ -483,7 +492,8 @@ def _stream_with_retry(sess, url, headers, payload, parse_fn):
|
||||
if attempt < sess.max_retries:
|
||||
d = _delay(None, attempt)
|
||||
print(f"[LLM Retry] {type(e).__name__}, retry in {d:.1f}s ({attempt+1}/{sess.max_retries+1})")
|
||||
time.sleep(d); continue
|
||||
if _sleep(d): return []
|
||||
continue
|
||||
yield err; return [{"type": "text", "text": err}]
|
||||
except Exception as e:
|
||||
err = f"\n\n[!!! 流异常中断 {type(e).__name__}: {e} !!!]" if streamed else f"!!!Error: {type(e).__name__}: {e}"
|
||||
|
||||
Reference in new issue
Block a user