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

fix inference

Browse Source
This commit is contained in:
Dobromir Popov
2026-06-30 13:01:29 +03:00
parent dade97ee67
commit c691e8d5d1
4 changed files with 105 additions and 9 deletions
Download Patch File Download Diff File Expand all files Collapse all files

4
packages/node/meshnet_node/startup.py
View File

@@ -204,7 +204,7 @@ def run_startup(
tracker_url=tracker_url,
)
actual_port = node.start()
total_layers = getattr(node.backend, "total_layers", None)
total_layers = getattr(getattr(node, "backend", None), "total_layers", None)
if isinstance(total_layers, int) and total_layers > 0:
layer_count = shard_end - shard_start + 1
shard_label = f"layers {shard_start}–{shard_end}; {layer_count} of {total_layers}"
@@ -213,7 +213,7 @@ def run_startup(
public_host = advertise_host or (socket.getfqdn() if host == "0.0.0.0" else host)
endpoint = f"http://{public_host}:{actual_port}"
# Register with tracker so other nodes can auto-join this model.
total_layers = getattr(node.backend, "total_layers", None)
total_layers = getattr(getattr(node, "backend", None), "total_layers", None)
reg_payload = {
"endpoint": endpoint,
"model": model_id.split("/")[-1],

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

@@ -246,6 +246,11 @@ class _TorchHandler(http.server.BaseHTTPRequestHandler):
# but correct. Each step: head encodes current sequence → forwards through route
# → tail returns the next token string → append → repeat.
remaining_route = self._get_remaining_route(model_name)
print(
f" [node] chat route model={model_name!r} max_tokens={max_tokens} "
f"downstream={remaining_route}",
flush=True,
)
if not remaining_route:
self._send_openai_response(
"error: no downstream route — check tracker connectivity",
@@ -300,7 +305,9 @@ class _TorchHandler(http.server.BaseHTTPRequestHandler):
try:
route = json.loads(injected)
if isinstance(route, list):
return [str(ep) for ep in route]
resolved = [str(ep) for ep in route]
print(f" [node] using injected downstream route: {resolved}", flush=True)
return resolved
except (json.JSONDecodeError, TypeError):
pass
@@ -315,7 +322,9 @@ class _TorchHandler(http.server.BaseHTTPRequestHandler):
route_resp = json.loads(r.read())
route = route_resp.get("route", [])
own_port = server.server_address[1]
return [ep for ep in route if not ep.rstrip("/").endswith(f":{own_port}")]
resolved = [ep for ep in route if not ep.rstrip("/").endswith(f":{own_port}")]
print(f" [node] tracker downstream route: {resolved}", flush=True)
return resolved
except Exception as exc:
print(f" [node] WARNING: route lookup failed for {route_model!r}: {exc}", flush=True)
return []
@@ -346,6 +355,7 @@ class _TorchHandler(http.server.BaseHTTPRequestHandler):
current_pos = pos_ids
for hop_index, node_url in enumerate(route):
print(f" [node] pipeline hop {hop_index}: {node_url}", flush=True)
headers: dict[str, str] = {
"Content-Type": "application/octet-stream",
"X-Meshnet-Wire": _WIRE_VERSION,
@@ -371,12 +381,15 @@ class _TorchHandler(http.server.BaseHTTPRequestHandler):
resp_body = r.read()
resp_headers = {k.lower(): v for k, v in r.headers.items()}
except Exception as exc:
print(f" [node] pipeline hop {hop_index} failed at {node_url}: {exc}", flush=True)
return f"pipeline error at {node_url}: {exc}"
content_type = resp_headers.get("content-type", "")
if "application/json" in content_type:
try:
data = json.loads(resp_body)
return str(data.get("text", ""))
text = str(data.get("text", ""))
print(f" [node] pipeline hop {hop_index} returned text={text!r}", flush=True)
return text
except json.JSONDecodeError:
return resp_body.decode("utf-8", errors="replace")
# Binary activation — update and forward to next node

28
packages/tracker/meshnet_tracker/server.py
View File

@@ -24,6 +24,7 @@ HTTP API contract:
import http.server
import hashlib
import json
import socketserver
import threading
import time
import urllib.parse
@@ -291,7 +292,7 @@ def _purge_expired_nodes_locked(server: "_TrackerHTTPServer") -> list[str]:
for node_id in expired_ids:
entry = server.registry.pop(node_id)
print(
f"[tracker] node expired: {node_id[:8]} {entry.endpoint} "
f"[tracker] node expired: {node_id} {entry.endpoint} "
f"(no heartbeat for >{server.heartbeat_timeout:.0f}s)",
flush=True,
)
@@ -418,7 +419,9 @@ def _node_id_for_registration(
return f"{wallet_prefix}-{digest}"
class _TrackerHTTPServer(http.server.HTTPServer):
class _TrackerHTTPServer(socketserver.ThreadingMixIn, http.server.HTTPServer):
daemon_threads = True
def __init__(
self,
addr: tuple,
@@ -706,6 +709,7 @@ class _TrackerHandler(http.server.BaseHTTPRequestHandler):
# Simple round-robin via list length modulo (stateless, good enough)
node = candidates[int(time.time() * 1000) % len(candidates)]
target_url = f"{node.endpoint}/v1/chat/completions"
request_id = str(body.get("id") or f"req-{time.time_ns():x}")
# Pre-resolve the downstream route so the first-shard node skips its own
# tracker query. We already hold the full registry picture — no need for
@@ -726,6 +730,16 @@ class _TrackerHandler(http.server.BaseHTTPRequestHandler):
route_nodes, _ = _select_route(all_nodes, rs, re)
# Strip the first-shard node we're about to proxy to — it's already handling the request.
downstream_urls = json.dumps([n.endpoint for n in route_nodes if n.endpoint != node.endpoint])
route_debug = " -> ".join(
f"{n.node_id}@{n.endpoint}[{n.shard_start}-{n.shard_end}]"
for n in route_nodes
)
print(
f"[tracker] proxy route {request_id}: model={model!r} "
f"head={node.node_id}@{node.endpoint} downstream={downstream_urls} "
f"route={route_debug or '<empty>'}",
flush=True,
)
req = urllib.request.Request(
target_url,
@@ -743,6 +757,7 @@ class _TrackerHandler(http.server.BaseHTTPRequestHandler):
try:
upstream = urllib.request.urlopen(req, timeout=300.0)
print(f"[tracker] proxy connected {request_id}: {target_url}", flush=True)
except urllib.error.HTTPError as exc:
# Relay error status + body from node
err_body = exc.read()
@@ -756,6 +771,7 @@ class _TrackerHandler(http.server.BaseHTTPRequestHandler):
pass
return
except Exception as exc:
print(f"[tracker] proxy failed {request_id}: {target_url}: {exc}", flush=True)
self._send_json(503, {"error": {
"message": f"upstream node unreachable: {exc}",
"type": "service_unavailable",
@@ -783,6 +799,10 @@ class _TrackerHandler(http.server.BaseHTTPRequestHandler):
else:
# Non-streaming: buffer and relay
resp_body = upstream.read()
print(
f"[tracker] proxy complete {request_id}: {target_url} bytes={len(resp_body)}",
flush=True,
)
self.send_response(200)
self.send_header("Content-Type", content_type)
self.send_header("Content-Length", str(len(resp_body)))
@@ -966,7 +986,7 @@ class _TrackerHandler(http.server.BaseHTTPRequestHandler):
for eid in stale_ids:
old = server.registry.pop(eid)
print(
f"[tracker] node re-registered: replaced {eid[:8]} with {node_id[:8]}"
f"[tracker] node re-registered: replaced {eid} with {node_id}"
f" {old.endpoint}",
flush=True,
)
@@ -980,7 +1000,7 @@ class _TrackerHandler(http.server.BaseHTTPRequestHandler):
shard_info = f"layers {shard_start}-{shard_end}" if shard_start is not None else "unsharded"
repo_info = f" [{hf_repo}]" if hf_repo else ""
print(
f"[tracker] node registered: {node_id[:8]} {endpoint} {model}{repo_info} {shard_info}",
f"[tracker] node registered: {node_id} {endpoint} {model}{repo_info} {shard_info}",
flush=True,
)