From e44abc910d1d2b0111c6a17670011be7a5097e2c Mon Sep 17 00:00:00 2001 From: Dobromir Popov Date: Wed, 8 Jul 2026 18:48:50 +0200 Subject: [PATCH] routing --- packages/node/meshnet_node/startup.py | 3 + packages/node/meshnet_node/torch_server.py | 91 ++++++++++++++++--- .../tracker/meshnet_tracker/dashboard.html | 9 +- packages/tracker/meshnet_tracker/server.py | 37 +++++++- tests/test_dashboard.py | 3 +- tests/test_dynamic_routing.py | 16 ++++ tests/test_tracker_routing.py | 51 +++++++++++ 7 files changed, 191 insertions(+), 19 deletions(-) diff --git a/packages/node/meshnet_node/startup.py b/packages/node/meshnet_node/startup.py index 14ebac6..c119474 100644 --- a/packages/node/meshnet_node/startup.py +++ b/packages/node/meshnet_node/startup.py @@ -829,6 +829,7 @@ def run_startup( shard_label = _format_shard_label(shard_start, shard_end, total_layers) public_host = advertise_host or (socket.getfqdn() if host == "0.0.0.0" else host) endpoint = f"http://{public_host}:{actual_port}" + node.set_advertised_endpoint(endpoint) local_base_url = f"http://127.0.0.1:{actual_port}" relay_bridge, relay_fields = _start_relay_bridge_if_available( tracker_url, @@ -977,6 +978,7 @@ def run_startup( actual_port = node.start() public_host = advertise_host or (socket.getfqdn() if host == "0.0.0.0" else host) endpoint = f"http://{public_host}:{actual_port}" + node.set_advertised_endpoint(endpoint) local_base_url = f"http://127.0.0.1:{actual_port}" relay_bridge, relay_fields = _start_relay_bridge_if_available( tracker_url, @@ -1154,6 +1156,7 @@ def run_startup( shard_label = f"{shard_label} (pinned)" public_host = advertise_host or (socket.getfqdn() if host == "0.0.0.0" else host) endpoint = f"http://{public_host}:{actual_port}" + node.set_advertised_endpoint(endpoint) local_base_url = f"http://127.0.0.1:{actual_port}" relay_bridge, relay_fields = _start_relay_bridge_if_available( tracker_url, diff --git a/packages/node/meshnet_node/torch_server.py b/packages/node/meshnet_node/torch_server.py index 2218e3b..3bc8605 100644 --- a/packages/node/meshnet_node/torch_server.py +++ b/packages/node/meshnet_node/torch_server.py @@ -31,6 +31,52 @@ from .server import ( ) +def _endpoint_key(url: str) -> str: + """Normalize http(s) endpoints for host:port comparison.""" + parsed = urllib.parse.urlparse(url.rstrip("/")) + host = (parsed.hostname or "").lower() + if not host: + return url.rstrip("/").lower() + port = parsed.port + if port is None: + port = 443 if parsed.scheme == "https" else 80 + return f"{host}:{port}" + + +def _own_endpoint_key(server: _TorchHTTPServer) -> str: + advertised = getattr(server, "advertised_endpoint", None) + if advertised: + return _endpoint_key(advertised) + host, port = server.server_address + return _endpoint_key(f"http://{host}:{port}") + + +def _clamp_downstream_hops( + hops: list[dict], + backend: TorchModelShard | None, +) -> list[dict]: + """Ensure downstream start_layer continues after this shard's layers.""" + if not hops or backend is None: + return hops + shard_end = getattr(backend, "shard_end", None) + if shard_end is None: + return hops + min_start = int(shard_end) + 1 + clamped: list[dict] = [] + for hop in hops: + adjusted = dict(hop) + if int(adjusted.get("start_layer", 0)) < min_start: + adjusted["start_layer"] = min_start + clamped.append(adjusted) + return clamped + + +def _format_downstream_route(hops: list[dict]) -> str: + return ", ".join( + f"{h['endpoint']}@{h.get('start_layer', 0)}" for h in hops + ) + + def _write_progress_line(state: list[bool], message: str, *, final: bool = False) -> None: """Rewrite one in-place progress line (\\r) or finish with a newline.""" if final: @@ -104,6 +150,7 @@ class _TorchHTTPServer(http.server.HTTPServer): self.route_timeout = route_timeout self.debug = debug self.max_loaded_shards = max(1, max_loaded_shards) + self.advertised_endpoint: str | None = None self.total_requests: int = 0 self.failed_requests: int = 0 self.queue_depth: int = 0 @@ -572,6 +619,9 @@ class _TorchHandler(http.server.BaseHTTPRequestHandler): start_layer tells each downstream node which layer to begin from, enabling correct execution when shard ranges overlap. """ + server: _TorchHTTPServer = self.server # type: ignore[assignment] + active_backend = backend or server.backend + # Fast path: tracker pre-resolved the downstream route and injected it as a header. injected = self.headers.get("X-Meshnet-Route") if injected: @@ -590,14 +640,16 @@ class _TorchHandler(http.server.BaseHTTPRequestHandler): hops.append(hop) elif isinstance(item, str): hops.append({"endpoint": item, "start_layer": 0}) - print(f" [node] using injected downstream route: {[h['endpoint'] for h in hops]}", flush=True) + hops = _clamp_downstream_hops(hops, active_backend) + print( + f" [node] using injected downstream route: {_format_downstream_route(hops)}", + flush=True, + ) return hops except (json.JSONDecodeError, TypeError, KeyError): pass # Slow path: query the tracker (direct node-to-node calls, or tracker didn't inject). - server: _TorchHTTPServer = self.server # type: ignore[assignment] - active_backend = backend or server.backend if server.tracker_url is None: return [] route_model = getattr(active_backend, "model_id", None) or model @@ -605,23 +657,31 @@ class _TorchHandler(http.server.BaseHTTPRequestHandler): url = f"{server.tracker_url}/v1/route?model={urllib.parse.quote(route_model)}" with urllib.request.urlopen(url, timeout=server.route_timeout) as r: route_resp = json.loads(r.read()) - own_port = server.server_address[1] + own_key = _own_endpoint_key(server) nodes_info = route_resp.get("nodes", []) - hops = [] - covered_up_to: int | None = None + hops: list[dict] = [] + passed_self = False for node_info in nodes_info: ep = node_info.get("endpoint", "") - if ep.rstrip("/").endswith(f":{own_port}"): - covered_up_to = node_info.get("shard_end") + if not ep: continue - if covered_up_to is None: - covered_up_to = (node_info.get("shard_start") or 1) - 1 - hop = {"endpoint": ep, "start_layer": covered_up_to + 1} + if _endpoint_key(ep) == own_key: + passed_self = True + continue + if not passed_self: + continue + hop = { + "endpoint": ep, + "start_layer": int(node_info.get("start_layer", 0)), + } if node_info.get("relay_addr"): hop["relay_addr"] = str(node_info["relay_addr"]) hops.append(hop) - covered_up_to = node_info.get("shard_end", covered_up_to) - print(f" [node] tracker downstream route: {[h['endpoint'] for h in hops]}", flush=True) + hops = _clamp_downstream_hops(hops, active_backend) + print( + f" [node] tracker downstream route: {_format_downstream_route(hops)}", + flush=True, + ) return hops except Exception as exc: print(f" [node] WARNING: route lookup failed for {route_model!r}: {exc}", flush=True) @@ -1039,6 +1099,11 @@ class TorchNodeServer: self._thread.start() return self.port + def set_advertised_endpoint(self, endpoint: str) -> None: + """Set the LAN-facing endpoint used for route self-detection.""" + if self._server is not None: + self._server.advertised_endpoint = endpoint + def stop(self) -> None: if self._server is None: return diff --git a/packages/tracker/meshnet_tracker/dashboard.html b/packages/tracker/meshnet_tracker/dashboard.html index 9fec492..bfa2d58 100644 --- a/packages/tracker/meshnet_tracker/dashboard.html +++ b/packages/tracker/meshnet_tracker/dashboard.html @@ -293,14 +293,14 @@ function buildModelAliasMap(map) { if (!display) return; for (const name of names) { const key = modelAliasKey(name); - if (key) byAlias.set(key, display); + if (key && !byAlias.has(key)) byAlias.set(key, display); } }; for (const entry of (map && map.recommended_models) || []) { register(entry.id, entry.id, entry.hf_repo, ...(entry.aliases || [])); } for (const entry of availableModels || []) { - register(entry.name || entry.id, entry.id, entry.name, ...(entry.aliases || [])); + register(entry.id, entry.id, entry.name, entry.hf_repo, ...(entry.aliases || [])); } return byAlias; } @@ -311,7 +311,9 @@ function resolveModelGroup(node, aliasMap) { const hit = aliasMap.get(modelAliasKey(candidate)); if (hit) return hit; } - return node.hf_repo || node.model || "?"; + const raw = node.hf_repo || node.model || "?"; + const key = modelAliasKey(raw); + return aliasMap.get(key) || key || "?"; } async function fetchJson(path) { @@ -1575,6 +1577,7 @@ async function refresh() { availableModels = ((models && models.data) || []).map(model => ({ id: model.id, name: model.name || model.id, + hf_repo: model.hf_repo, recommended: Boolean(model.recommended), aliases: model.aliases || [], })).filter(model => model.id); diff --git a/packages/tracker/meshnet_tracker/server.py b/packages/tracker/meshnet_tracker/server.py index 48f0216..45522d9 100644 --- a/packages/tracker/meshnet_tracker/server.py +++ b/packages/tracker/meshnet_tracker/server.py @@ -159,6 +159,18 @@ DEFAULT_CALLER_CREDIT_USDT = 1.0 DEFAULT_DEVNET_TOPUP_USDT = 1.0 +def _endpoint_key(url: str) -> str: + """Normalize http(s) endpoints for host:port comparison.""" + parsed = urllib.parse.urlparse(url.rstrip("/")) + host = (parsed.hostname or "").lower() + if not host: + return url.rstrip("/").lower() + port = parsed.port + if port is None: + port = 443 if parsed.scheme == "https" else 80 + return f"{host}:{port}" + + def _model_aliases(model: str | None) -> set[str]: """Return stable lookup aliases for a model repo or display name.""" if not model: @@ -3168,7 +3180,10 @@ class _TrackerHandler(http.server.BaseHTTPRequestHandler): node_work.append((rn.wallet_address, max(0, effective_end - covered_up_to))) covered_up_to = effective_end # Strip the first-shard node we're about to proxy to — it's already handling the request. - downstream_hops = [h for h in route_hops if h["endpoint"].rstrip("/") != node.endpoint.rstrip("/")] + downstream_hops = [ + h for h in route_hops + if _endpoint_key(h["endpoint"]) != _endpoint_key(node.endpoint) + ] downstream_urls = json.dumps(downstream_hops) route_debug = " -> ".join( f"{n.node_id}@{n.endpoint}[{n.shard_start}-{n.shard_end}]" @@ -5447,7 +5462,25 @@ class _TrackerHandler(http.server.BaseHTTPRequestHandler): if not node.wallet_address or not server.contracts.registry.get_wallet(node.wallet_address).banned ] - route, error = _select_route(alive, required_start, required_end, contracts=server.contracts) + route_model = resolved_name or model + candidates = _enumerate_routes( + alive, + required_start, + required_end, + model=route_model, + contracts=server.contracts, + max_candidates=server.route_stats.config.max_candidate_routes, + ) + if candidates: + # Prefer a distributed multi-hop route when available. Greedy + # _select_route alone would pick a single full-shard node and omit + # partial head shards that share the same port on another host. + route = max(candidates, key=lambda cand: len(cand.nodes)).nodes + error = "" + else: + route, error = _select_route( + alive, required_start, required_end, contracts=server.contracts, + ) if error: _tracker_log( server, diff --git a/tests/test_dashboard.py b/tests/test_dashboard.py index 632f019..fea342f 100644 --- a/tests/test_dashboard.py +++ b/tests/test_dashboard.py @@ -28,9 +28,10 @@ def test_dashboard_served_with_all_panels(): ).read().decode() for panel in PANELS: assert panel in html - assert "