popov/neuron-tai
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

fix: silence BrokenPipeError on slow CPU inference, raise downstream timeout

Browse Source 
BrokenPipeError on wfile.write is harmless — the client disconnected
before the response arrived. Suppress it instead of printing a full
traceback to stderr.

Raise the downstream pipeline timeout from 10s to 120s so the Windows
node (or any caller) waits long enough for CPU-mode inference to
complete before giving up.

Co-Authored-By: Claude Sonnet 4.6 <noreply@anthropic.com>
This commit is contained in:
Dobromir Popov
2026-06-30 10:01:37 +03:00
parent 748d535c46
commit b95e25a5c3
1 changed files with 25 additions and 10 deletions
Download Patch File Download Diff File Expand all files Collapse all files

35
packages/node/meshnet_node/torch_server.py
View File

@@ -206,7 +206,10 @@ class _TorchHandler(http.server.BaseHTTPRequestHandler):
self.send_header("Content-Type", "application/json")
self.send_header("Content-Length", str(len(payload)))
self.end_headers()
self.wfile.write(payload)
try:
self.wfile.write(payload)
except BrokenPipeError:
pass # client disconnected before we could respond — not an error
def _handle_chat_completions(self) -> None:
server: _TorchHTTPServer = self.server # type: ignore[assignment]
@@ -356,7 +359,7 @@ class _TorchHandler(http.server.BaseHTTPRequestHandler):
method="POST",
)
try:
with urllib.request.urlopen(req, timeout=10.0) as r:
with urllib.request.urlopen(req, timeout=120.0) as r:
resp_body = r.read()
resp_headers = {k.lower(): v for k, v in r.headers.items()}
except Exception as exc:
@@ -386,8 +389,11 @@ class _TorchHandler(http.server.BaseHTTPRequestHandler):
self.end_headers()
def _emit(data: str) -> None:
self.wfile.write(f"data: {data}\n\n".encode())
self.wfile.flush()
try:
self.wfile.write(f"data: {data}\n\n".encode())
self.wfile.flush()
except BrokenPipeError:
pass
_emit(json.dumps({
"id": chunk_id, "object": "chat.completion.chunk", "created": created,
@@ -407,8 +413,11 @@ class _TorchHandler(http.server.BaseHTTPRequestHandler):
"model": model,
"choices": [{"index": 0, "delta": {}, "finish_reason": "stop"}],
}))
self.wfile.write(b"data: [DONE]\n\n")
self.wfile.flush()
try:
self.wfile.write(b"data: [DONE]\n\n")
self.wfile.flush()
except BrokenPipeError:
pass
def _send_openai_response(
self,
@@ -440,8 +449,11 @@ class _TorchHandler(http.server.BaseHTTPRequestHandler):
self.end_headers()
def _emit(data: str) -> None:
self.wfile.write(f"data: {data}\n\n".encode())
self.wfile.flush()
try:
self.wfile.write(f"data: {data}\n\n".encode())
self.wfile.flush()
except BrokenPipeError:
pass
_emit(json.dumps({
"id": chunk_id, "object": "chat.completion.chunk", "created": created,
@@ -458,8 +470,11 @@ class _TorchHandler(http.server.BaseHTTPRequestHandler):
"model": model,
"choices": [{"index": 0, "delta": {}, "finish_reason": "stop"}],
}))
self.wfile.write(b"data: [DONE]\n\n")
self.wfile.flush()
try:
self.wfile.write(b"data: [DONE]\n\n")
self.wfile.flush()
except BrokenPipeError:
pass
def _usage_for_response(backend: object, messages: list[dict], completion_text: str) -> dict[str, int]: