From a7cc377d13a4ae06ac81dcdcb884e71d57a3b730 Mon Sep 17 00:00:00 2001 From: Dobromir Popov Date: Tue, 30 Jun 2026 00:22:33 +0300 Subject: [PATCH] =?UTF-8?q?feat:=20auto-join=20network=20=E2=80=94=20node?= =?UTF-8?q?=20discovers=20missing=20shards=20from=20tracker?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Tracker: - _NodeEntry gains hf_repo + num_layers fields (parsed from register body) - GET /v1/network/assign — finds the biggest uncovered shard gap across registered HF-model nodes; returns {hf_repo, shard_start, shard_end, num_layers} - Returns 503 when no HF-model nodes are registered yet Node startup: - When model_id is set: registers with tracker including hf_repo + num_layers so other nodes can auto-join this model - When model_id is empty/None: queries /v1/network/assign, gets assigned the missing layers, loads TorchNodeServer with the assigned shard automatically - Fixes empty-string model_id leaking from DEFAULTS (treats "" same as None) Usage: `meshnet-node start --tracker http://localhost:8080 --quantization bfloat16` Node discovers what to serve and joins the network without any model flags. Co-Authored-By: Claude Sonnet 4.6 --- packages/node/meshnet_node/startup.py | 93 ++++++++++++++++- packages/tracker/meshnet_tracker/server.py | 116 ++++++++++++++++++++- 2 files changed, 206 insertions(+), 3 deletions(-) diff --git a/packages/node/meshnet_node/startup.py b/packages/node/meshnet_node/startup.py index 49cf693..91ecda2 100644 --- a/packages/node/meshnet_node/startup.py +++ b/packages/node/meshnet_node/startup.py @@ -84,7 +84,7 @@ def run_startup( if probationary_line is not None: print(f" {probationary_line}", flush=True) - if model_id is not None: + if model_id: # treat "" the same as None — no explicit model given # Auto-detect shard range from model config if not explicitly provided if shard_start is None or shard_end is None: detected = _detect_num_layers(model_id) @@ -116,6 +116,28 @@ def run_startup( shard_label = f"layers {shard_start}–{shard_end}" 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) + try: + _post_json( + f"{tracker_url}/v1/nodes/register", + { + "endpoint": endpoint, + "model": model_id.split("/")[-1], + "hf_repo": model_id, + "num_layers": total_layers, + "shard_start": shard_start, + "shard_end": shard_end, + "hardware_profile": hw, + "wallet_address": address, + "quantization": quantization, + "score": 1.0, + "tracker_mode": (shard_start == 0), + }, + ) + except Exception as exc: + print(f" Warning: tracker registration failed: {exc}", flush=True) + print( f"\n{'=' * 32}\n" f"meshnet-node ready\n" @@ -132,7 +154,74 @@ def run_startup( if shard_start is not None or shard_end is not None: raise ValueError("--shard-start / --shard-end require --model-id") - # 3. Shard assignment from tracker + # 3a. Auto-join: query tracker for network-wide HF model assignment. + print("Querying tracker for network assignment...", flush=True) + assign_qs = urllib.parse.urlencode({"device": device, "vram_mb": vram_mb}) + try: + net_assignment = _get_json(f"{tracker_url}/v1/network/assign?{assign_qs}") + assigned_hf_repo: str | None = net_assignment.get("hf_repo") + except Exception: + assigned_hf_repo = None + + if assigned_hf_repo: + assigned_shard_start: int = net_assignment["shard_start"] + assigned_shard_end: int = net_assignment["shard_end"] + assigned_num_layers: int = net_assignment["num_layers"] + print( + f" Assigned: {assigned_hf_repo} " + f"layers {assigned_shard_start}–{assigned_shard_end} " + f"(of {assigned_num_layers})", + flush=True, + ) + print("Loading real PyTorch model shard...", flush=True) + node = TorchNodeServer( + host=host, + port=port, + model_id=assigned_hf_repo, + shard_start=assigned_shard_start, + shard_end=assigned_shard_end, + quantization=quantization, + tracker_url=tracker_url, + ) + 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}" + try: + _post_json( + f"{tracker_url}/v1/nodes/register", + { + "endpoint": endpoint, + "model": assigned_hf_repo.split("/")[-1], + "hf_repo": assigned_hf_repo, + "num_layers": assigned_num_layers, + "shard_start": assigned_shard_start, + "shard_end": assigned_shard_end, + "hardware_profile": hw, + "wallet_address": address, + "quantization": quantization, + "score": 1.0, + "tracker_mode": (assigned_shard_start == 0), + }, + ) + except Exception as exc: + print(f" Warning: tracker registration failed: {exc}", flush=True) + shard_count = assigned_shard_end - assigned_shard_start + 1 + print( + f"\n{'=' * 32}\n" + f"meshnet-node ready (auto-joined)\n" + f" Wallet: {address}\n" + f" Model ID: {assigned_hf_repo}\n" + f" Shard: layers {assigned_shard_start}–{assigned_shard_end} " + f"({shard_count} of {assigned_num_layers})\n" + f" Quantization: {quantization}\n" + f" Endpoint: {endpoint}\n" + f" Hardware: {device.upper()}\n" + f"{'=' * 32}", + flush=True, + ) + return node + + # 3b. Shard assignment from tracker (stub-model / preset-based path) print("Querying tracker for shard assignment...", flush=True) assign_qs = urllib.parse.urlencode({ "model": model, diff --git a/packages/tracker/meshnet_tracker/server.py b/packages/tracker/meshnet_tracker/server.py index 542aca3..cd42d1d 100644 --- a/packages/tracker/meshnet_tracker/server.py +++ b/packages/tracker/meshnet_tracker/server.py @@ -52,7 +52,7 @@ DEFAULT_BENCHMARK_TOKENS_PER_SEC = 1.0 class _NodeEntry: __slots__ = ( "node_id", "endpoint", "shard_start", "shard_end", - "model", "shard_checksum", "hardware_profile", "wallet_address", + "model", "hf_repo", "num_layers", "shard_checksum", "hardware_profile", "wallet_address", "score", "vram_bytes", "ram_bytes", "quantizations", "benchmark_tokens_per_sec", "quantization", "managed_assignment", "pending_directives", "last_heartbeat", "tracker_mode", @@ -76,6 +76,8 @@ class _NodeEntry: quantization: str | None = None, managed_assignment: bool = False, tracker_mode: bool = False, + hf_repo: str | None = None, + num_layers: int | None = None, ) -> None: self.node_id = node_id self.endpoint = endpoint @@ -93,6 +95,8 @@ class _NodeEntry: self.quantization = quantization self.managed_assignment = managed_assignment self.tracker_mode = tracker_mode + self.hf_repo = hf_repo + self.num_layers = num_layers self.pending_directives: list[dict] = [] self.last_heartbeat: float = time.monotonic() @@ -426,6 +430,8 @@ class _TrackerHandler(http.server.BaseHTTPRequestHandler): self._handle_routes(parsed) elif parsed.path == "/v1/nodes/assign": self._handle_assign(parsed) + elif parsed.path == "/v1/network/assign": + self._handle_network_assign(parsed) elif parsed.path == "/v1/models": self._handle_models() elif parsed.path.startswith("/v1/coverage/"): @@ -607,6 +613,18 @@ class _TrackerHandler(http.server.BaseHTTPRequestHandler): return tracker_mode = bool(body.get("tracker_mode", False)) + hf_repo = body.get("hf_repo") + if hf_repo is not None and not isinstance(hf_repo, str): + self._send_json(400, {"error": "hf_repo must be a string"}) + return + num_layers_body = body.get("num_layers") + num_layers: int | None = None + if num_layers_body is not None: + try: + num_layers = int(num_layers_body) + except (TypeError, ValueError): + self._send_json(400, {"error": "num_layers must be an integer"}) + return node_id = str(uuid.uuid4()) entry = _NodeEntry( @@ -626,6 +644,8 @@ class _TrackerHandler(http.server.BaseHTTPRequestHandler): quantization=quantization, managed_assignment=not explicit_shard, tracker_mode=tracker_mode, + hf_repo=hf_repo, + num_layers=num_layers, ) with server.lock: self._purge_expired_nodes() @@ -751,6 +771,100 @@ class _TrackerHandler(http.server.BaseHTTPRequestHandler): **({"hf_repo": preset["hf_repo"]} if "hf_repo" in preset else {}), }) + def _handle_network_assign(self, parsed: urllib.parse.ParseResult): + """Assign a new node to fill the biggest uncovered shard gap across HF-model nodes. + + Query params: + vram_mb — integer VRAM in MB (0 = CPU-only node) + device — "cuda" | "cpu" + + Returns: + {hf_repo, shard_start, shard_end, num_layers} + """ + server: _TrackerHTTPServer = self.server # type: ignore[assignment] + params = urllib.parse.parse_qs(parsed.query) + try: + vram_mb = int(params.get("vram_mb", ["0"])[0]) + except ValueError: + vram_mb = 0 + device = params.get("device", ["cpu"])[0] + + with server.lock: + self._purge_expired_nodes() + all_nodes = list(server.registry.values()) + + # Collect only nodes that registered a real HF model (have hf_repo + shard bounds). + hf_nodes = [ + n for n in all_nodes + if n.hf_repo + and n.shard_start is not None + and n.shard_end is not None + and n.num_layers is not None + ] + + if not hf_nodes: + self._send_json(503, {"error": "no HF-model nodes registered; cannot assign shards"}) + return + + # Group by hf_repo; pick the one with the largest total_layers and biggest gap. + from collections import defaultdict + repo_groups: dict = defaultdict(list) + repo_layers: dict = {} + for n in hf_nodes: + repo_groups[n.hf_repo].append(n) + # Use the largest num_layers seen for this repo. + if n.hf_repo not in repo_layers or n.num_layers > repo_layers[n.hf_repo]: + repo_layers[n.hf_repo] = n.num_layers + + # Pick the repo where the gap is largest (most work to do). + best_repo = None + best_gap_size = -1 + best_gap_start = 0 + best_num_layers = 0 + + for repo, nodes in repo_groups.items(): + total = repo_layers[repo] + covered = sorted( + [(n.shard_start, n.shard_end) for n in nodes], + key=lambda t: t[0], + ) + # Walk from 0 to find first uncovered layer. + gap_start = 0 + for s, e in covered: + if s <= gap_start: + gap_start = max(gap_start, e + 1) + else: + break + gap_size = max(0, (total - 1) - gap_start + 1) # layers remaining uncovered + if gap_size > best_gap_size: + best_gap_size = gap_size + best_gap_start = gap_start + best_repo = repo + best_num_layers = total + + if best_repo is None or best_gap_size <= 0: + # All shards are covered — still assign to the model with most nodes for redundancy. + best_repo = max(repo_groups, key=lambda r: len(repo_groups[r])) + best_gap_start = 0 + best_num_layers = repo_layers[best_repo] + + # Capacity: CPU nodes get at most half the layers; CUDA nodes based on VRAM. + total_l = best_num_layers + if device == "cuda" and vram_mb >= 8192: + max_layers = total_l + else: + max_layers = max(1, total_l // 2) + + shard_start = best_gap_start + shard_end = min(total_l - 1, shard_start + max_layers - 1) + + self._send_json(200, { + "hf_repo": best_repo, + "shard_start": shard_start, + "shard_end": shard_end, + "num_layers": total_l, + }) + def _handle_route(self, parsed: urllib.parse.ParseResult): server: _TrackerHTTPServer = self.server # type: ignore[assignment] params = urllib.parse.parse_qs(parsed.query)