diff --git a/packages/node/meshnet_node/dashboard.py b/packages/node/meshnet_node/dashboard.py index f8e97c9..a12ab05 100644 --- a/packages/node/meshnet_node/dashboard.py +++ b/packages/node/meshnet_node/dashboard.py @@ -94,6 +94,19 @@ def _make_bar(pct: float, width: int = 10) -> str: return "█" * filled + "░" * (width - filled) +def _node_stats(node) -> dict: + total = int(getattr(node, "total_requests", getattr(node, "chat_completion_count", 0)) or 0) + failed = int(getattr(node, "failed_requests", 0) or 0) + queue_depth = int(getattr(node, "queue_depth", 0) or 0) + success_rate = ((total - failed) / total * 100.0) if total else 100.0 + return { + "total_requests": total, + "failed_requests": failed, + "queue_depth": queue_depth, + "success_rate": success_rate, + } + + def run_dashboard(node, config: dict, start_time: float) -> None: """Start the live dashboard. Blocks until Ctrl-C. Returns cleanly.""" if not is_interactive_tty(): @@ -117,7 +130,8 @@ def _build_rich_renderable( from rich.text import Text # type: ignore[import] uptime = time.monotonic() - start_time - req_count = getattr(node, "chat_completion_count", 0) + stats = _node_stats(node) + req_count = stats["total_requests"] # Tokens/sec EMA (approximate: 20 tokens per request heuristic when no real counter) delta_req = req_count - prev_req[0] @@ -163,6 +177,7 @@ def _build_rich_renderable( stats_lines = [ f"Tokens/sec {tps_bar} {tps:.1f} t/s (EMA)", f"Requests {req_count:,} served", + f"Success {stats['success_rate']:.1f}% failed {stats['failed_requests']:,} queue {stats['queue_depth']}", f"Peers 0 connected (gossip: US-017)", f"TAI earned 0.00 TAI (payments: US-006)", f"Uptime {_format_uptime(uptime)}", @@ -205,14 +220,17 @@ def _run_plain_loop(node, config: dict, start_time: float) -> None: try: while True: uptime = time.monotonic() - start_time - req = getattr(node, "chat_completion_count", 0) + stats = _node_stats(node) + req = stats["total_requests"] gpu_stats = _gpu_stats() vram_str = "" if gpu_stats: g = gpu_stats[0] vram_str = f" VRAM{g['used_gb']:.1f}GB" print( - f"[{model_name} req{req}{vram_str} up{_format_uptime(uptime)}]", + f"[{model_name} req{req} ok{stats['success_rate']:.1f}% " + f"fail{stats['failed_requests']} q{stats['queue_depth']}" + f"{vram_str} up{_format_uptime(uptime)}]", flush=True, ) time.sleep(2) diff --git a/packages/node/meshnet_node/hardware.py b/packages/node/meshnet_node/hardware.py index 33724dd..a392fd2 100644 --- a/packages/node/meshnet_node/hardware.py +++ b/packages/node/meshnet_node/hardware.py @@ -1,10 +1,23 @@ """GPU hardware detection with graceful CPU fallback.""" +import os import subprocess +import time + + +def _detect_ram_mb() -> int: + """Return host physical RAM in MB, or 0 if unavailable.""" + try: + pages = os.sysconf("SC_PHYS_PAGES") + page_size = os.sysconf("SC_PAGE_SIZE") + return int((pages * page_size) // (1024 * 1024)) + except (AttributeError, OSError, ValueError): + return 0 def detect_hardware() -> dict: """Detect GPU model and available VRAM. Returns hardware profile dict.""" + ram_mb = _detect_ram_mb() try: import torch # type: ignore[import] if torch.cuda.is_available(): @@ -12,7 +25,7 @@ def detect_hardware() -> dict: name = torch.cuda.get_device_name(idx) props = torch.cuda.get_device_properties(idx) vram_mb = props.total_memory // (1024 * 1024) - return {"device": "cuda", "gpu_name": name, "vram_mb": vram_mb} + return {"device": "cuda", "gpu_name": name, "vram_mb": vram_mb, "ram_mb": ram_mb} except ImportError: pass @@ -26,8 +39,54 @@ def detect_hardware() -> dict: parts = line.split(",", 1) gpu_name = parts[0].strip() vram_mb = int(parts[1].strip()) if len(parts) > 1 else 0 - return {"device": "cuda", "gpu_name": gpu_name, "vram_mb": vram_mb} + return {"device": "cuda", "gpu_name": gpu_name, "vram_mb": vram_mb, "ram_mb": ram_mb} except (FileNotFoundError, subprocess.TimeoutExpired, ValueError, IndexError): pass - return {"device": "cpu", "gpu_name": None, "vram_mb": 0} + return {"device": "cpu", "gpu_name": None, "vram_mb": 0, "ram_mb": ram_mb} + + +def benchmark_throughput(device_str: str = "cpu") -> float: + """ + Estimate compute throughput via a synthetic transformer GEMM benchmark. + + Runs hidden_size × (hidden_size*4) matmul — the dominant op in FFN layers — + and returns iterations/second as a relative speed index. Higher = faster. + The value is used as benchmark_tokens_per_sec in tracker registration for + routing tiebreaks; it is not an absolute token rate. + + Falls back to 1.0 if torch is unavailable. + """ + try: + import torch # type: ignore[import] + + device = torch.device(device_str) + # bfloat16 on CUDA matches real inference dtype; float32 on CPU avoids + # precision-downcast surprises on older hardware without bfloat16 support. + dtype = torch.bfloat16 if device_str == "cuda" else torch.float32 + + # hidden_size=2048 is representative of a mid-sized model; large enough + # that BLAS finds an efficient kernel on both GPU and CPU. + hidden_size = 2048 + a = torch.randn(1, hidden_size, dtype=dtype, device=device) + b = torch.randn(hidden_size, hidden_size * 4, dtype=dtype, device=device) + + def _sync() -> None: + if device_str == "cuda": + torch.cuda.synchronize() + + # Warmup: prime caches and JIT compilation. + for _ in range(10): + torch.matmul(a, b) + _sync() + + n_iters = 50 + t0 = time.perf_counter() + for _ in range(n_iters): + torch.matmul(a, b) + _sync() + elapsed = time.perf_counter() - t0 + + return round(n_iters / max(elapsed, 1e-9), 2) + except Exception: + return 1.0 diff --git a/packages/node/meshnet_node/startup.py b/packages/node/meshnet_node/startup.py index 5a02977..a5db230 100644 --- a/packages/node/meshnet_node/startup.py +++ b/packages/node/meshnet_node/startup.py @@ -14,7 +14,7 @@ from pathlib import Path from typing import Any from .downloader import compute_shard_checksum, download_shard -from .hardware import detect_hardware +from .hardware import detect_hardware, benchmark_throughput from .relay_bridge import RelayHttpBridge, peer_id_from_wallet from .server import StubNodeServer from .torch_server import TorchNodeServer @@ -359,12 +359,18 @@ def run_startup( print(f" GPU: {gpu_name} ({vram_mb / 1024:.1f} GB VRAM, {ram_mb / 1024:.1f} GB RAM)", flush=True) memory_budget_mb, memory_budget_source = _memory_budget(vram_mb, ram_mb) - print(f" Memory budget: {memory_budget_mb} MB {memory_budget_source}", flush=True) + print(f" Memory budget: {memory_budget_mb / 1024:.1f} GB {memory_budget_source}", flush=True) + + print("Benchmarking compute...", flush=True) + bench_tps = benchmark_throughput(device) + device_label = "GPU" if device == "cuda" else "CPU" + print(f" {device_label} throughput index: {bench_tps:,.0f}", flush=True) registration_capabilities = { "vram_bytes": max(0, int(vram_mb)) * 1024 * 1024, "ram_bytes": max(0, int(ram_mb)) * 1024 * 1024, "max_loaded_shards": max_loaded_shards, + "benchmark_tokens_per_sec": bench_tps, } # 2. Wallet print("Loading wallet...", flush=True) @@ -479,6 +485,7 @@ def run_startup( f" Endpoint: {endpoint}\n" f" Node ID: {tracker_node_id or 'unregistered'}\n" f" Hardware: {device.upper()}\n" + f" Benchmark: {bench_tps:,.0f} (throughput index)\n" f"{'=' * 32}", flush=True, ) @@ -571,6 +578,7 @@ def run_startup( f" Endpoint: {endpoint}\n" f" Node ID: {tracker_node_id or 'unregistered'}\n" f" Hardware: {device.upper()}\n" + f" Benchmark: {bench_tps:,.0f} (throughput index)\n" f"{'=' * 32}", flush=True, ) @@ -671,6 +679,7 @@ def run_startup( f" Endpoint: {endpoint}\n" f" Node ID: {node_id}\n" f" Hardware: {hw_str}\n" + f" Benchmark: {bench_tps:,.0f} (throughput index)\n" f"{'=' * 32}", flush=True, ) diff --git a/packages/tracker/meshnet_tracker/server.py b/packages/tracker/meshnet_tracker/server.py index e15e773..0b9f6d5 100644 --- a/packages/tracker/meshnet_tracker/server.py +++ b/packages/tracker/meshnet_tracker/server.py @@ -429,24 +429,38 @@ def _node_quantization(node: _NodeEntry, preset: dict) -> str: return next(iter(bytes_per_layer)) +def _node_memory_budget_bytes(node: _NodeEntry) -> tuple[int, str]: + """Return the memory pool used for shard-capacity planning.""" + if node.vram_bytes > 0: + return node.vram_bytes, "vram" + if node.ram_bytes > 0: + return node.ram_bytes, "ram" + return DEFAULT_RAM_BYTES, "ram-default" + + def _node_layer_capacity(node: _NodeEntry, preset: dict) -> int: bytes_per_layer = _preset_bytes_per_layer(preset) quantization = _node_quantization(node, preset) layer_bytes = bytes_per_layer[quantization] if layer_bytes <= 0: return 0 - return int((node.vram_bytes * 0.8) // layer_bytes) + memory_budget_bytes, _ = _node_memory_budget_bytes(node) + return int((memory_budget_bytes * 0.8) // layer_bytes) def _node_capacity_summary(node: _NodeEntry, preset: dict | None = None) -> dict: """Operator-facing capacity fields for inspection endpoints.""" + memory_budget_bytes, memory_budget_source = _node_memory_budget_bytes(node) summary = { "vram_bytes": node.vram_bytes, "ram_bytes": node.ram_bytes, + "memory_budget_bytes": memory_budget_bytes, + "memory_budget_source": memory_budget_source, "max_loaded_shards": node.max_loaded_shards, "quantizations": list(node.quantizations), "quantization": node.quantization, "benchmark_tokens_per_sec": node.benchmark_tokens_per_sec, + "effective_throughput": round(_effective_throughput(node), 4), } if preset is not None: summary["max_assignable_layers"] = _node_layer_capacity(node, preset) @@ -1154,6 +1168,7 @@ class _TrackerHandler(http.server.BaseHTTPRequestHandler): "tracker_mode": node.tracker_mode, "last_heartbeat": node.last_heartbeat, "capacity": capacity_for(node), + "stats": _node_health(node, server.heartbeat_timeout), } for node in nodes ], @@ -1567,8 +1582,11 @@ 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 "" + budget_bytes, budget_source = _node_memory_budget_bytes(entry) + budget_gb = budget_bytes / (1024 ** 3) print( - f"[tracker] node registered: {node_id} {endpoint} {model}{repo_info} {shard_info}", + f"[tracker] node registered: {node_id} {endpoint} {model}{repo_info} {shard_info} " + f"capacity={budget_gb:.1f}GB {budget_source} slots={max_loaded_shards}", flush=True, ) @@ -1707,6 +1725,7 @@ class _TrackerHandler(http.server.BaseHTTPRequestHandler): model — model preset name (default: first preset) device — "cuda" | "cpu" vram_mb — integer VRAM in MB (0 for CPU) + ram_mb — integer system RAM in MB, used when vram_mb=0 The greedy strategy: find the first gap in current layer coverage and assign it. If no gap exists, assign the full model range so the @@ -1745,8 +1764,16 @@ class _TrackerHandler(http.server.BaseHTTPRequestHandler): vram_mb = int(params.get("vram_mb", ["0"])[0]) except ValueError: vram_mb = 0 + try: + ram_mb = int(params.get("ram_mb", ["0"])[0]) + except ValueError: + ram_mb = 0 max_layers = required_end - required_start + 1 - if device != "cuda" or vram_mb < 8192: + memory_mb = vram_mb if vram_mb > 0 else ram_mb + if memory_mb > 0: + layer_bytes = _preset_bytes_per_layer(preset).get("bfloat16", 30 * 1024 * 1024) + max_layers = min(max_layers, max(1, int(((memory_mb * 1024 * 1024) * 0.8) // layer_bytes))) + elif device != "cuda" or vram_mb < 8192: max_layers = min(max_layers, 16) # Collect covered intervals sorted by start layer. @@ -1798,6 +1825,7 @@ class _TrackerHandler(http.server.BaseHTTPRequestHandler): Query params: vram_mb — integer VRAM in MB (0 = CPU-only node) + ram_mb — integer system RAM in MB, used when vram_mb=0 device — "cuda" | "cpu" hf_repo — optional; if set, restrict search to this repo only @@ -1811,6 +1839,10 @@ class _TrackerHandler(http.server.BaseHTTPRequestHandler): vram_mb = int(params.get("vram_mb", ["0"])[0]) except ValueError: vram_mb = 0 + try: + ram_mb = int(params.get("ram_mb", ["0"])[0]) + except ValueError: + ram_mb = 0 device = params.get("device", ["cpu"])[0] filter_repo = params.get("hf_repo", [None])[0] # optional repo filter @@ -1894,9 +1926,15 @@ class _TrackerHandler(http.server.BaseHTTPRequestHandler): 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. + # Capacity: use the same 80%-of-memory rule as registered node planning. total_l = best_num_layers - if device == "cuda" and vram_mb >= 8192: + memory_mb = vram_mb if vram_mb > 0 else ram_mb + if memory_mb > 0: + max_layers = min( + total_l, + max(1, int(((memory_mb * 1024 * 1024) * 0.8) // (30 * 1024 * 1024))), + ) + elif device == "cuda" and vram_mb >= 8192: max_layers = total_l else: max_layers = max(1, total_l // 2) diff --git a/tests/conftest.py b/tests/conftest.py new file mode 100644 index 0000000..ee9c069 --- /dev/null +++ b/tests/conftest.py @@ -0,0 +1,19 @@ +"""Shared pytest fixtures for the meshnet test suite.""" + +import pytest + + +@pytest.fixture(autouse=True) +def _stub_benchmark_throughput(monkeypatch): + """Replace the GEMM benchmark with a fixed value in all tests. + + The benchmark runs 60 matmuls (warmup + measure) which adds ~100ms per test + on CPU. Tests verify registration flow, not hardware speed — stub it out. + Tests that specifically exercise benchmark_throughput import it directly from + meshnet_node.hardware and are not affected by this patch. + """ + try: + import meshnet_node.startup as startup_mod + monkeypatch.setattr(startup_mod, "benchmark_throughput", lambda _device: 999.0) + except ImportError: + pass diff --git a/tests/test_node_startup.py b/tests/test_node_startup.py index e77a554..4c64ab6 100644 --- a/tests/test_node_startup.py +++ b/tests/test_node_startup.py @@ -10,7 +10,7 @@ from pathlib import Path import pytest from meshnet_node.downloader import download_shard, write_shard_archive -from meshnet_node.hardware import detect_hardware +from meshnet_node.hardware import detect_hardware, benchmark_throughput from meshnet_node.startup import ( _infer_relay_url_from_tracker, _probationary_status_line, @@ -31,6 +31,8 @@ def test_detect_hardware_returns_valid_profile(): hw = detect_hardware() assert hw["device"] in {"cuda", "cpu"} assert isinstance(hw.get("vram_mb"), int) + assert isinstance(hw.get("ram_mb"), int) + assert hw["ram_mb"] > 0 if hw["device"] == "cpu": assert hw["gpu_name"] is None assert hw["vram_mb"] == 0 @@ -39,6 +41,62 @@ def test_detect_hardware_returns_valid_profile(): assert hw["vram_mb"] > 0 +def test_benchmark_throughput_cpu_returns_positive(): + """CPU benchmark returns a positive float greater than the 1.0 error fallback.""" + result = benchmark_throughput("cpu") + assert isinstance(result, float) + assert result > 1.0, f"expected benchmark > 1.0, got {result}" + + +def test_benchmark_throughput_fallback_on_bad_device(): + """benchmark_throughput returns 1.0 (not raises) when device is invalid.""" + result = benchmark_throughput("invalid_device_xyz") + assert result == 1.0 + + +def test_benchmark_throughput_is_registered_in_payload(monkeypatch, tmp_path): + """benchmark_tokens_per_sec from the benchmark is included in the tracker registration.""" + import meshnet_node.startup as startup_mod + + captured: dict = {} + + class FakeNode: + backend = None + tracker_node_id = None + + def start(self): + return 7099 + + def stop(self): + pass + + def apply_tracker_directives(self, directives): + return None + + monkeypatch.setattr(startup_mod, "detect_hardware", + lambda: {"device": "cpu", "gpu_name": None, "vram_mb": 0, "ram_mb": 16384}) + monkeypatch.setattr(startup_mod, "benchmark_throughput", lambda _device: 42.5) + monkeypatch.setattr(startup_mod, "TorchNodeServer", lambda **_kw: FakeNode()) + monkeypatch.setattr(startup_mod, "_detect_num_layers", lambda _model_id: 24) + monkeypatch.setattr(startup_mod, "RelayHttpBridge", None) + monkeypatch.setattr(startup_mod, "_get_json", + lambda _url, timeout=10.0: {"relay_url": None, "nodes": []}) + monkeypatch.setattr(startup_mod, "_post_json", + lambda _url, payload, timeout=10.0: (captured.update(payload) or {"node_id": "x"})) + monkeypatch.setattr(startup_mod, "_start_heartbeat", lambda *a, **kw: None) + + node = run_startup( + tracker_url="http://localhost:8080", + model_id="Qwen/Qwen2.5-0.5B-Instruct", + shard_start=0, + shard_end=23, + wallet_path=tmp_path / "wallet.json", + ) + node.stop() + + assert captured.get("benchmark_tokens_per_sec") == 42.5 + + def test_wallet_generates_new_keypair(tmp_path): """A new wallet is created when none exists, saved to disk.""" wallet_file = tmp_path / "wallet.json" @@ -490,6 +548,70 @@ def test_real_model_startup_summary_shows_total_layers(tmp_path, monkeypatch, ca assert "Node ID: node-test-123" in output +def test_real_model_startup_autodetects_cpu_memory_budget_and_logs_shard_budget( + tmp_path, + monkeypatch, + capsys, +): + """Without --memory, startup reports RAM-backed capacity to the tracker and operator.""" + import meshnet_node.startup as startup_mod + captured_registration = {} + + class FakeBackend: + total_layers = 24 + + class FakeTorchNodeServer: + def __init__(self, **kwargs): + self.kwargs = kwargs + self.backend = FakeBackend() + self.port = None + self.total_requests = 0 + self.failed_requests = 0 + self.queue_depth = 0 + + def start(self): + self.port = 8001 + return self.port + + def stop(self): + pass + + monkeypatch.setattr( + startup_mod, + "detect_hardware", + lambda: {"device": "cpu", "gpu_name": None, "vram_mb": 0, "ram_mb": 16384}, + ) + monkeypatch.setattr(startup_mod, "TorchNodeServer", FakeTorchNodeServer) + monkeypatch.setattr( + startup_mod, + "_post_json", + lambda _url, _payload, timeout=10.0: ( + captured_registration.update(_payload) or {"node_id": "node-auto-mem"} + ), + ) + + node = run_startup( + tracker_url="http://127.0.0.1:8080", + model_id="Qwen/Qwen2.5-0.5B-Instruct", + shard_start=0, + shard_end=23, + wallet_path=tmp_path / "wallet.json", + ) + try: + pass + finally: + node.stop() + + assert captured_registration["vram_bytes"] == 0 + assert captured_registration["ram_bytes"] == 16384 * 1024 * 1024 + assert captured_registration["max_loaded_shards"] == 1 + output = capsys.readouterr().out + assert "Memory budget: 16.0 GB RAM" in output + assert "Shard budget: up to 24/24 layers at bfloat16" in output + assert "GB remaining after full load" in output + assert "Node ID: node-auto-mem" in output + + def test_public_tracker_model_node_registers_relay_metadata_from_tracker_url_only( tmp_path, monkeypatch,