diff --git a/QUICKSTART.md b/QUICKSTART.md index b4bccde..61a3607 100644 --- a/QUICKSTART.md +++ b/QUICKSTART.md @@ -349,18 +349,26 @@ curl -s http://192.168.0.179:8080/v1/network/map | python3 -m json.tool # from ### Public tracker + relay (NAT / WSL2 / internet nodes) -Nodes behind NAT cannot receive inbound connections. Run **both** services on the -tracker host; the tracker advertises the relay URL in `/v1/network/map`. +Nodes behind NAT cannot receive inbound connections. Keep the relay as its own +component, but for alpha deployments you can run it **embedded in the tracker +process**. This still uses the same `meshnet_relay.RelayServer` class as a +relay-only node, so relay-only hosts remain a clean scaling path later. -**Terminal 1 — relay:** +**Recommended alpha: one process on the tracker host, relay bound locally and +published through the same public hostname:** + +```bash +.venv/bin/meshnet-tracker start \ + --host 0.0.0.0 --port 8081 \ + --self-url https://ai.neuron.d-popov.com \ + --embedded-relay --relay-host 127.0.0.1 --relay-port 8765 +``` + +The tracker advertises `wss://ai.neuron.d-popov.com/ws` in `/v1/network/map`. +If you want a relay-only process instead, keep running: ```bash .venv/bin/meshnet-relay --host 0.0.0.0 --port 8765 -``` - -**Terminal 2 — tracker:** - -```bash .venv/bin/meshnet-tracker start --host 0.0.0.0 --port 8081 --relay-url wss://ai.neuron.d-popov.com/ws ``` diff --git a/_e2e_kv_check.py b/_e2e_kv_check.py new file mode 100644 index 0000000..983b637 --- /dev/null +++ b/_e2e_kv_check.py @@ -0,0 +1,77 @@ +"""E2E check: two real TorchNodeServers over HTTP, user's live topology. + +Head owns layers 0-20, tail owns 10-23 (overlapping shards, clamped to @21), +exactly like the reported 2-node run. Compares cached vs stateless tps and +verifies identical output text. +""" +import json +import sys +import time +import urllib.request + +sys.path.insert(0, "packages/node") + +from meshnet_node.model_backend import TorchModelShard +from meshnet_node.torch_server import TorchNodeServer + +MODEL = "Qwen/Qwen2.5-0.5B-Instruct" +MAX_TOKENS = 60 + +print("loading shards...", flush=True) +head_backend = TorchModelShard(MODEL, 0, 20) +tail_backend = TorchModelShard(MODEL, 10, 23) + +head = TorchNodeServer(backend=head_backend, tracker_mode=True) +tail = TorchNodeServer(backend=tail_backend) +head_port = head.start() +tail_port = tail.start() +print(f"head :{head_port} layers 0-20 | tail :{tail_port} layers 10-23", flush=True) + + +def chat(label): + body = json.dumps({ + "model": MODEL, + "messages": [{"role": "user", "content": "Count from 1 to 10, then say done."}], + "max_tokens": MAX_TOKENS, + "stream": False, + }).encode() + req = urllib.request.Request( + f"http://127.0.0.1:{head_port}/v1/chat/completions", + data=body, + headers={ + "Content-Type": "application/json", + # tracker-injected route: tail continues at layer 21 (clamped) + "X-Meshnet-Route": json.dumps( + [{"endpoint": f"http://127.0.0.1:{tail_port}", "start_layer": 21}] + ), + }, + method="POST", + ) + started = time.monotonic() + with urllib.request.urlopen(req, timeout=600) as r: + resp = json.loads(r.read()) + elapsed = time.monotonic() - started + text = resp["choices"][0]["message"]["content"] + tokens = resp.get("usage", {}).get("completion_tokens", 0) + print(f"\n[{label}] tokens={tokens} elapsed={elapsed:.2f}s " + f"tps={tokens / max(elapsed, 1e-6):.2f}", flush=True) + print(f"[{label}] text={text!r}", flush=True) + return text, tokens, elapsed + + +try: + cached_text, ct, ce = chat("cached-warmup") # first run pays model warmup + cached_text, ct, ce = chat("cached") + forwards_cached = tail.forward_chunk_count + + head_backend.supports_kv_cache = False # force legacy stateless path + stateless_text, st, se = chat("stateless") + forwards_stateless = tail.forward_chunk_count - forwards_cached + + print(f"\ntail /forward calls: cached-run total={forwards_cached} " + f"stateless-run={forwards_stateless}", flush=True) + print(f"speedup: {(ct / ce) / max(st / se, 1e-9):.2f}x", flush=True) + print(f"outputs identical: {cached_text == stateless_text}", flush=True) +finally: + head.stop() + tail.stop() diff --git a/docs/dev/_NOTES.md b/docs/dev/_NOTES.md new file mode 100644 index 0000000..409da7d --- /dev/null +++ b/docs/dev/_NOTES.md @@ -0,0 +1,11 @@ +# IDEAS + + - use real torrenting library/ infrastructure + + + + + let's work on the ability to have the tracker on the interet. i want to realease the alpha version to see the first feedbacks. we have to check if we need a relay + node, and is it working. or it will be more practical if the the tracker integrates the relay +functionality as well. I'd say keep it separate, as we may have relay only nodes, but a tracker also is a relay -it uses the same code/class for the relay to expose it on it's api as well - +a good architecture \ No newline at end of file diff --git a/packages/node/meshnet_node/torch_server.py b/packages/node/meshnet_node/torch_server.py index 03193da..10d52d3 100644 --- a/packages/node/meshnet_node/torch_server.py +++ b/packages/node/meshnet_node/torch_server.py @@ -169,6 +169,35 @@ class _TorchHTTPServer(http.server.HTTPServer): self.queue_depth: int = 0 self._stats_lock = threading.Lock() self._active_requests: dict[str, dict[str, Any]] = {} + self._decode_log: dict[str, dict[str, float]] = {} + + def note_decode_step( + self, session: str, now: float | None = None, + ) -> int | None: + """Count one decode forward; return the cumulative step count when a + log line is due (first step of a session, then every 5s), else None.""" + if now is None: + now = time.monotonic() + with self._stats_lock: + rec = self._decode_log.get(session) + if rec is None: + if len(self._decode_log) >= 64: + stale = [ + sid for sid, r in self._decode_log.items() + if now - r["seen"] > 600.0 + ] + for sid in stale: + del self._decode_log[sid] + while len(self._decode_log) >= 64: + self._decode_log.pop(next(iter(self._decode_log))) + self._decode_log[session] = {"steps": 1.0, "logged": now, "seen": now} + return 1 + rec["steps"] += 1 + rec["seen"] = now + if now - rec["logged"] >= 5.0: + rec["logged"] = now + return int(rec["steps"]) + return None def snapshot_current_requests(self) -> list[dict[str, Any]]: """In-flight request snapshots for tracker heartbeats.""" @@ -346,23 +375,34 @@ class _TorchHandler(http.server.BaseHTTPRequestHandler): 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 - # Session KV-cache protocol: prefill establishes per-session state on # this node's layer range; decode reuses it. Absent header = legacy # stateless call (also the signature fake backends implement). cache_mode = self.headers.get("X-Meshnet-Cache") + if chunk_index_value == 0: + shard_start = getattr(server.backend, "shard_start", "?") + shard_end = getattr(server.backend, "shard_end", "?") + if cache_mode == "decode": + # One decode forward arrives per generated token — log a + # periodic per-session summary instead of one line per token. + steps = server.note_decode_step(session) + if steps is not None: + print( + f" [node] decoding layers={shard_start}-{shard_end} " + f"session={session[:8]} steps={steps}" + f"{self._request_log_suffix()}", + flush=True, + ) + else: + 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 forward_kwargs: dict[str, object] = {} if cache_mode in ("prefill", "decode"): past_len_header = self.headers.get("X-Meshnet-Past-Len") @@ -585,12 +625,14 @@ class _TorchHandler(http.server.BaseHTTPRequestHandler): session_id = str(uuid.uuid4()) use_kv = bool(getattr(backend, "supports_kv_cache", False)) + # EOS detection by id must work on the stateless path too: the tail + # returns token_id regardless of caching, and EOS usually decodes to + # "" (skip_special_tokens), so the text comparison never fires. eos_ids: set[int] = set() - if use_kv: - try: - eos_ids = set(backend.eos_token_ids()) - except Exception: - eos_ids = set() + try: + eos_ids = set(backend.eos_token_ids()) + except Exception: + eos_ids = set() stream_emit = None if stream: diff --git a/packages/tracker/meshnet_tracker/cli.py b/packages/tracker/meshnet_tracker/cli.py index 0b40b55..25147d8 100644 --- a/packages/tracker/meshnet_tracker/cli.py +++ b/packages/tracker/meshnet_tracker/cli.py @@ -103,6 +103,28 @@ def main() -> None: default=None, help="Public ws(s):// relay URL advertised to nodes, for example wss://ai.neuron.d-popov.com/ws", ) + common.add_argument( + "--embedded-relay", + action="store_true", + help="Run the relay WebSocket server in this tracker process (still uses meshnet_relay.RelayServer)", + ) + common.add_argument( + "--relay-host", + default="127.0.0.1", + help="Bind address for --embedded-relay (default: 127.0.0.1; use 0.0.0.0 only when exposing the relay port directly)", + ) + common.add_argument( + "--relay-port", + type=int, + default=8765, + help="Bind port for --embedded-relay (default: 8765)", + ) + common.add_argument( + "--relay-max-peers", + type=int, + default=500, + help="Maximum WebSocket peers accepted by --embedded-relay", + ) common.add_argument( "--billing-db", default=DEFAULT_BILLING_DB_PATH, @@ -378,6 +400,10 @@ def main() -> None: cluster_self_url=args.self_url, stats_db=getattr(args, "stats_db", None), relay_url=relay_url, + embedded_relay=args.embedded_relay, + embedded_relay_host=args.relay_host, + embedded_relay_port=args.relay_port, + embedded_relay_max_peers=args.relay_max_peers, enable_billing=not args.no_billing, billing_db=None if args.no_billing else args.billing_db, max_charge_per_request=args.max_charge_per_request, diff --git a/packages/tracker/meshnet_tracker/server.py b/packages/tracker/meshnet_tracker/server.py index 27ff134..50d01f6 100644 --- a/packages/tracker/meshnet_tracker/server.py +++ b/packages/tracker/meshnet_tracker/server.py @@ -101,6 +101,12 @@ def derive_relay_url_from_public_tracker_url(url: str | None) -> str | None: return f"wss://{parsed.netloc}/ws" +def _local_relay_url(host: str, port: int) -> str: + """Return a node-consumable ws:// URL for an embedded relay bind address.""" + advertise_host = "127.0.0.1" if host in {"", "0.0.0.0", "::"} else host + return f"ws://{advertise_host}:{port}/ws" + + def _load_model_presets_from_data() -> dict[str, dict]: """Load recommended model presets from package JSON data.""" try: @@ -2459,6 +2465,7 @@ class _TrackerHTTPServer(socketserver.ThreadingMixIn, http.server.HTTPServer): hf_pricing_log: "HfPricingLog | None" = None, models_dir: Path | None = None, route_stats: "RouteStatsStore | None" = None, + relay_status: dict | None = None, ) -> None: super().__init__(addr, handler) self.registry = registry @@ -2496,6 +2503,9 @@ class _TrackerHTTPServer(socketserver.ThreadingMixIn, http.server.HTTPServer): self.active_proxies_lock = threading.Lock() self.route_stats: RouteStatsStore = route_stats or RouteStatsStore() self.route_rng = random.Random() + self.relay_status = dict( + relay_status or {"mode": "external" if self.relay_url else "off"} + ) class _TrackerHandler(http.server.BaseHTTPRequestHandler): @@ -2930,6 +2940,7 @@ class _TrackerHandler(http.server.BaseHTTPRequestHandler): self._send_json(200, { "relay_url": server.relay_url, + "relay": dict(server.relay_status), "pool": _pool_summary(nodes), "memory_pool": memory_pool, "recommended_models": [ @@ -5779,6 +5790,10 @@ class TrackerServer: cluster_self_url: str | None = None, stats_db: str | None = None, relay_url: str | None = None, + embedded_relay: bool = False, + embedded_relay_host: str = "127.0.0.1", + embedded_relay_port: int = 8765, + embedded_relay_max_peers: int = 500, billing: BillingLedger | None = None, enable_billing: bool = False, billing_db: str | None = None, @@ -5821,6 +5836,12 @@ class TrackerServer: self._cluster_peers: list[str] = list(cluster_peers) if cluster_peers else [] self._cluster_self_url: str | None = cluster_self_url self._relay_url = relay_url + self._embedded_relay_enabled = bool(embedded_relay) + self._embedded_relay_host = embedded_relay_host + self._embedded_relay_port = int(embedded_relay_port) + self._embedded_relay_max_peers = int(embedded_relay_max_peers) + self._embedded_relay: Any | None = None + self._embedded_relay_actual_port: int | None = None self._registry: dict[str, _NodeEntry] = {} self._lock = threading.Lock() self._server: _TrackerHTTPServer | None = None @@ -5924,13 +5945,59 @@ class TrackerServer: self._route_stats = RouteStatsStore(routing_config) self.port: int | None = None + def _start_embedded_relay(self) -> dict: + """Start the shared RelayServer class in-process for tracker+relay deployments.""" + if not self._embedded_relay_enabled: + return {"mode": "external" if self._relay_url else "off"} + if self._embedded_relay is not None: + return self._embedded_relay_status() + try: + from meshnet_relay.server import RelayServer # type: ignore[import-not-found] + except Exception as exc: # pragma: no cover - only hit in broken installs + raise RuntimeError( + "embedded relay requires the meshnet-relay package to be installed" + ) from exc + relay = RelayServer( + host=self._embedded_relay_host, + port=self._embedded_relay_port, + max_peers=self._embedded_relay_max_peers, + ) + actual_port = relay.start() + self._embedded_relay = relay + self._embedded_relay_actual_port = actual_port + return self._embedded_relay_status() + + def _embedded_relay_status(self) -> dict: + advertised_url = ( + self._relay_url + or derive_relay_url_from_public_tracker_url(self._cluster_self_url) + or _local_relay_url(self._embedded_relay_host, self._embedded_relay_actual_port or self._embedded_relay_port) + ) + return { + "mode": "embedded", + "url": advertised_url, + "bind_host": self._embedded_relay_host, + "bind_port": self._embedded_relay_actual_port or self._embedded_relay_port, + "max_peers": self._embedded_relay_max_peers, + } + def start(self) -> int: if self._server is not None: raise RuntimeError("TrackerServer is already running") + relay_status = self._start_embedded_relay() effective_relay_url = ( self._relay_url or derive_relay_url_from_public_tracker_url(self._cluster_self_url) + or (relay_status.get("url") if relay_status.get("mode") == "embedded" else None) + ) + + http_relay_status = ( + {**relay_status, "url": effective_relay_url} + if relay_status.get("mode") == "embedded" + else {"mode": "external", "url": effective_relay_url} + if effective_relay_url + else {"mode": "off"} ) # Start HTTP server first so we know our port @@ -5960,6 +6027,7 @@ class TrackerServer: hf_pricing_log=self._hf_pricing_log, models_dir=self._models_dir, route_stats=self._route_stats, + relay_status=http_relay_status, ) self.port = self._server.server_address[1] @@ -6275,6 +6343,11 @@ class TrackerServer: self._raft.stop() if self._gossip is not None: self._gossip.stop() + embedded_relay = self._embedded_relay + self._embedded_relay = None + self._embedded_relay_actual_port = None + if embedded_relay is not None: + embedded_relay.stop() if self._server is None: return self._rebalance_stop.set() diff --git a/tests/test_gossip_and_relay.py b/tests/test_gossip_and_relay.py index 4dc62a9..9c272ce 100644 --- a/tests/test_gossip_and_relay.py +++ b/tests/test_gossip_and_relay.py @@ -451,6 +451,8 @@ def test_tracker_derives_relay_url_from_public_self_url(): with urllib.request.urlopen(f"http://127.0.0.1:{port}/v1/network/map", timeout=5) as resp: body = _json.loads(resp.read()) assert body["relay_url"] == "wss://ai.neuron.d-popov.com/ws" + assert body["relay"]["mode"] == "external" + assert body["relay"]["url"] == "wss://ai.neuron.d-popov.com/ws" finally: tracker.stop() @@ -501,10 +503,46 @@ def test_tracker_network_map_exposes_relay_and_registered_peer(): tracker.stop() assert body["relay_url"] == "wss://ai.neuron.d-popov.com/ws" + assert body["relay"]["mode"] == "external" assert body["nodes"][0]["relay_addr"] == "wss://ai.neuron.d-popov.com/rpc/peer123" assert body["nodes"][0]["peer_id"] == "peer123" +def test_tracker_can_embed_relay_server_and_advertise_it(): + import json as _json + import urllib.request + + import websockets.sync.client as wsc # type: ignore[import] + from meshnet_tracker.server import TrackerServer + + tracker = TrackerServer( + host="127.0.0.1", + port=0, + embedded_relay=True, + embedded_relay_host="127.0.0.1", + embedded_relay_port=0, + ) + port = tracker.start() + try: + with urllib.request.urlopen(f"http://127.0.0.1:{port}/v1/network/map", timeout=5) as resp: + body = _json.loads(resp.read()) + assert body["relay"]["mode"] == "embedded" + assert body["relay_url"].startswith("ws://127.0.0.1:") + assert body["relay_url"].endswith("/ws") + with wsc.connect(body["relay_url"], open_timeout=5) as ws: + ws.send(_json.dumps({ + "topic": "peer-register", + "version": 1, + "from_peer": "embedded-peer", + "msg_id": "embedded-reg-1", + "payload": {"peer_id": "embedded-peer", "addr": ""}, + })) + peer_list = _json.loads(ws.recv(timeout=5)) + assert peer_list["topic"] == "peer-list" + finally: + tracker.stop() + + # --------------------------------------------------------------------------- # mDNS (no-op without zeroconf installed) # --------------------------------------------------------------------------- diff --git a/tests/test_kv_cache_distributed.py b/tests/test_kv_cache_distributed.py index d08fa80..b4dd860 100644 --- a/tests/test_kv_cache_distributed.py +++ b/tests/test_kv_cache_distributed.py @@ -260,6 +260,49 @@ def test_eos_token_id_stops_generation(): assert len(tail_backend.calls) == 2 +def test_stateless_fallback_stops_at_eos_token_id(): + """When kv caching is off, EOS must still stop generation by token id — + EOS decodes to "" (skip_special_tokens) so the text check never fires.""" + + class _StatelessHead(_CachedHeadBackend): + supports_kv_cache = False + + head_backend = _StatelessHead() + tail_backend = _CachedTailBackend([(" a", 1), ("", 99), (" never", 3)]) + head = TorchNodeServer(backend=head_backend, tracker_mode=True) + tail = TorchNodeServer(backend=tail_backend) + head_port = head.start() + tail_port = tail.start() + try: + content = _chat_once(head_port, tail_port, max_tokens=8) + finally: + head.stop() + tail.stop() + + assert content == " a" + # Stops after the EOS step instead of burning steps until max_tokens. + assert len(tail_backend.calls) == 2 + assert head_backend.decode_calls == [] + + +def test_decode_forward_logging_is_rate_limited(): + """Shard nodes log a per-session decode summary, not one line per token.""" + tail_backend = _CachedTailBackend([]) + tail = TorchNodeServer(backend=tail_backend) + tail.start() + try: + srv = tail._server + assert srv.note_decode_step("s1", now=0.0) == 1 + assert srv.note_decode_step("s1", now=1.0) is None + assert srv.note_decode_step("s1", now=4.9) is None + assert srv.note_decode_step("s1", now=5.5) == 4 + assert srv.note_decode_step("s1", now=6.0) is None + # Sessions are throttled independently. + assert srv.note_decode_step("s2", now=6.0) == 1 + finally: + tail.stop() + + def test_downstream_cache_miss_falls_back_to_full_reprefill(): head_backend = _CachedHeadBackend() # Call 1 (the first decode) raises KVCacheMiss → node answers 409 →