Log node request processing so operators can see live activity in the console.

This commit is contained in:
Dobromir Popov
2026-07-07 18:12:57 +02:00
parent 1eb1e0baa2
commit aba5fb12fa
4 changed files with 65 additions and 6 deletions

View File

@@ -164,6 +164,9 @@ class RelayHttpBridge:
path = str(payload.get("path") or "/")
headers = payload.get("headers") if isinstance(payload.get("headers"), dict) else {}
req_suffix = f" request_id={request_id}" if request_id else ""
print(f" [node] relay {method} {path}{req_suffix}", flush=True)
# body_base64 carries binary data (e.g. bfloat16 activation tensors) safely.
# Fallback to text "body" for backward-compat with non-binary requests.
body_b64 = payload.get("body_base64")

View File

@@ -113,6 +113,10 @@ class _TorchHandler(http.server.BaseHTTPRequestHandler):
def log_message(self, fmt, *args): # noqa: suppress request logs in tests
pass
def _request_log_suffix(self) -> str:
req_id = self.headers.get("X-Meshnet-Request-Id") or self.headers.get("X-Request-Id")
return f" request_id={req_id}" if req_id else ""
def do_POST(self):
server: _TorchHTTPServer = self.server # type: ignore[assignment]
if self.path == "/forward":
@@ -199,8 +203,18 @@ class _TorchHandler(http.server.BaseHTTPRequestHandler):
return
server.forward_chunk_count += 1
if int(self.headers.get("X-Meshnet-Hop-Index", "0")) > 0:
hop_index = int(self.headers.get("X-Meshnet-Hop-Index", "0"))
if hop_index > 0:
server.received_activations = True
if chunk_index_value == 0:
shard_start = getattr(server.backend, "shard_start", "?")
shard_end = getattr(server.backend, "shard_end", "?")
print(
f" [node] forward hop={hop_index} "
f"layers={shard_start}-{shard_end} "
f"session={session[:8]}{self._request_log_suffix()}",
flush=True,
)
start_layer_header = self.headers.get("X-Meshnet-Start-Layer")
start_layer = int(start_layer_header) if start_layer_header else None
@@ -311,20 +325,53 @@ class _TorchHandler(http.server.BaseHTTPRequestHandler):
temperature = float(body.get("temperature") or 1.0)
top_p = float(body.get("top_p") or 1.0)
print(
f" [node] processing chat model={model_name!r} stream={stream} "
f"max_tokens={max_tokens}{self._request_log_suffix()}",
flush=True,
)
# Fast path: this node owns the complete model — use HF generate() with KV cache.
# Avoids the single-token-per-forward-pass limitation of the distributed path.
if backend.is_head and backend.is_tail:
gen_started = time.monotonic()
try:
if stream:
self._stream_openai_response(
backend.generate_text_streaming(messages, max_tokens, temperature, top_p),
model_name,
token_count = 0
def _counting_stream():
nonlocal token_count
for token_text in backend.generate_text_streaming(
messages, max_tokens, temperature, top_p,
):
if token_text:
token_count += 1
yield token_text
self._stream_openai_response(_counting_stream(), model_name)
print(
f" [node] chat complete (stream) tokens={token_count} "
f"elapsed_s={time.monotonic() - gen_started:.1f}{self._request_log_suffix()}",
flush=True,
)
else:
text = backend.generate_text(messages, max_tokens, temperature, top_p)
completion_tokens = _backend_token_count(
backend, "count_text_tokens", text, fallback=len(text.split()) or 1,
)
print(
f" [node] chat complete tokens={completion_tokens} "
f"elapsed_s={time.monotonic() - gen_started:.1f}{self._request_log_suffix()}",
flush=True,
)
self._send_openai_response(text, model_name, False, messages, backend=backend)
except Exception as exc:
self._record_failed_request()
print(
f" [node] chat failed after {time.monotonic() - gen_started:.1f}s: {exc}"
f"{self._request_log_suffix()}",
flush=True,
)
self._send_json(500, {"error": f"generation failed: {exc}"})
return