From 1d3fb060aeef6cfdbecda15f1d44fdb986fa336c Mon Sep 17 00:00:00 2001 From: Dobromir Popov Date: Thu, 9 Jul 2026 10:48:32 +0200 Subject: [PATCH] relay working with qwen2.5; relay anounced on node ready --- _DEV_NOTES.md | 8 +++++ packages/node/meshnet_node/model_backend.py | 24 ++++++++++++-- packages/node/meshnet_node/startup.py | 11 +++++++ packages/node/meshnet_node/torch_server.py | 28 ++++++++++++++--- tests/test_kv_cache_distributed.py | 35 +++++++++++++++++++++ tests/test_node_startup.py | 1 + 6 files changed, 100 insertions(+), 7 deletions(-) diff --git a/_DEV_NOTES.md b/_DEV_NOTES.md index 5e6b4b5..661bfed 100644 --- a/_DEV_NOTES.md +++ b/_DEV_NOTES.md @@ -30,6 +30,14 @@ HF_HOME=/run/media/popov/d/DEV/models .venv/bin/meshnet-node start --m meshnet-node.exe start --tracker http://192.168.0.179:8080 --model Qwen/Qwen2.5-0.5B-Instruct --shard-start 0 --shard-end 20 meshnet-node.exe start --tracker https://meshnet.2.d-popov.com --model Qwen/Qwen2.5-0.5B-Instruct meshnet-node.exe start --tracker https://meshnet.2.d-popov.com --model qwen3.6-35b-a3b --cpu + +meshnet-node.exe start --tracker https://meshnet.2.d-popov.com --model qwen3.6-35b-a3b --shard-start 0 --shard-end 21 --node-name gpu-head +meshnet-node.exe start --tracker https://meshnet.2.d-popov.com --model qwen3.6-35b-a3b --shard-start 22 --shard-end 39 --cpu --node-name cpu-tail + + +meshnet-node.exe start --tracker https://meshnet.2.d-popov.com --model Qwen/Qwen2.5-0.5B-Instruct --shard-end 20 --node-name gpu-head +meshnet-node.exe start --tracker https://meshnet.2.d-popov.com --model Qwen/Qwen2.5-0.5B-Instruct --shard-start 12 --cpu --node-name cpu-tail + # win meshnet-node start --tracker http://ai.neuron.d-popov.com --model Qwen/Qwen2.5-0.5B-Instruct --shard-start 10 meshnet-node start --tracker http://192.168.0.179:8081 --model Qwen/Qwen2.5-0.5B-Instruct --shard-start 10 diff --git a/packages/node/meshnet_node/model_backend.py b/packages/node/meshnet_node/model_backend.py index c0e8e2f..b38ea14 100644 --- a/packages/node/meshnet_node/model_backend.py +++ b/packages/node/meshnet_node/model_backend.py @@ -308,7 +308,11 @@ class TorchModelShard: self._norm = _final_norm(self.model) if self.is_tail else None self._lm_head = getattr(self.model, "lm_head", None) if self.is_tail else None # Per-session KV/recurrent-state cache for this shard's layer range. - self.supports_kv_cache = True + # Hybrid/linear-attention models such as Qwen3.6 can dispatch Triton + # recurrent-cache kernels when use_cache=True. Those kernels cannot + # consume CPU tensors ("Pointer argument cannot be accessed from Triton"), + # so CPU shards intentionally stay on the stateless prefill path. + self.supports_kv_cache = self.device.type != "cpu" self.kv_sessions = SessionCacheStore( max_sessions=int(os.environ.get("MESHNET_KV_MAX_SESSIONS", "8")), ttl_seconds=float(os.environ.get("MESHNET_KV_TTL_SECONDS", "600")), @@ -612,9 +616,13 @@ class TorchModelShard: hidden_states, attention_mask, position_ids, start_layer=start_layer, cache=cache, past_len=0, ) - except TypeError as exc: + except Exception as exc: + if not _cache_unsupported_for_shard(exc): + raise # Layers reject cache kwargs (exotic architecture) — disable caching - # for this backend and stay on the stateless path. + # for this backend and stay on the stateless path. Some hybrid + # CPU paths also accept cache kwargs but fail at runtime inside + # Triton-only kernels; treat those as cache-unsupported too. self.supports_kv_cache = False print(f" [node] kv cache unsupported by {self.model_id}: {exc}", flush=True) return self._run_layers( @@ -1146,3 +1154,13 @@ def _looks_like_oom(exc: BaseException) -> bool: return True current = current.__cause__ or current.__context__ return False + + +def _cache_unsupported_for_shard(exc: BaseException) -> bool: + """True when a layer failure means session cache is unsupported, not fatal.""" + text = str(exc).lower() + return ( + isinstance(exc, TypeError) + or "pointer argument cannot be accessed from triton" in text + or ("triton" in text and "cpu tensor" in text) + ) diff --git a/packages/node/meshnet_node/startup.py b/packages/node/meshnet_node/startup.py index 31ebf6c..6d97b42 100644 --- a/packages/node/meshnet_node/startup.py +++ b/packages/node/meshnet_node/startup.py @@ -140,6 +140,13 @@ def _hardware_label(device: str, gpu_name: str | None = None) -> str: return "CPU" +def _relay_ready_line(relay_fields: dict) -> str: + relay_addr = relay_fields.get("relay_addr") + if not relay_addr: + return "" + return f" Relay: {relay_addr}\n" + + def _positive_int(value: int | str | None, name: str) -> int | None: if value is None or value == "": return None @@ -917,6 +924,7 @@ def run_startup( f" {_shard_budget_line(memory_budget_mb, memory_budget_source, total_layers, quantization)}\n" f" Quantization: {quantization}\n" f" Endpoint: {endpoint}\n" + f"{_relay_ready_line(relay_fields)}" f" Node ID: {tracker_node_id or 'unregistered'}\n" f" Hardware: {_hardware_label(device, gpu_name)}\n" f" Benchmark: {bench_tps:,.0f} (throughput index)\n" @@ -1072,6 +1080,7 @@ def run_startup( f" {_shard_budget_line(memory_budget_mb, memory_budget_source, assigned_num_layers, quantization, bytes_per_layer=assigned_bytes_per_layer, safety_fraction=_runtime_shard_safety_fraction(device))}\n" f" Quantization: {quantization}\n" f" Endpoint: {endpoint}\n" + f"{_relay_ready_line(relay_fields)}" f" Node ID: {tracker_node_id or 'unregistered'}\n" f" Hardware: {_hardware_label(device, gpu_name)}\n" f" Benchmark: {bench_tps:,.0f} (throughput index)\n" @@ -1237,6 +1246,7 @@ def run_startup( f" {_shard_budget_line(memory_budget_mb, memory_budget_source, total_layers, quantization, bytes_per_layer=assignment_bytes_per_layer, safety_fraction=_runtime_shard_safety_fraction(device))}\n" f" Quantization: {quantization}\n" f" Endpoint: {endpoint}\n" + f"{_relay_ready_line(relay_fields)}" f" Node ID: {tracker_node_id or 'unregistered'}\n" f" Hardware: {_hardware_label(device, gpu_name)}\n" f" Benchmark: {bench_tps:,.0f} (throughput index)\n" @@ -1315,6 +1325,7 @@ def run_startup( f" Shard: {shard_label}\n" f" {_shard_budget_line(memory_budget_mb, memory_budget_source, assigned_total_layers, quantization, bytes_per_layer=assignment_bytes_per_layer, safety_fraction=_runtime_shard_safety_fraction(device))}\n" f" Endpoint: {endpoint}\n" + f"{_relay_ready_line(relay_fields)}" f" Node ID: {node_id}\n" f" Hardware: {hw_str}\n" f" Benchmark: {bench_tps:,.0f} (throughput index)\n" diff --git a/packages/node/meshnet_node/torch_server.py b/packages/node/meshnet_node/torch_server.py index 10d52d3..d03315e 100644 --- a/packages/node/meshnet_node/torch_server.py +++ b/packages/node/meshnet_node/torch_server.py @@ -141,6 +141,18 @@ def _is_cache_miss_body(body: bytes) -> bool: return False +def _response_error_snippet(body: bytes, limit: int = 500) -> str: + """Return a compact error string from a downstream JSON/text response body.""" + try: + payload = json.loads(body) + if isinstance(payload, dict): + message = payload.get("error") or payload.get("detail") or payload + return str(message)[:limit] + except (json.JSONDecodeError, TypeError, UnicodeDecodeError): + pass + return body.decode("utf-8", errors="replace")[:limit] + + class _TorchHTTPServer(http.server.HTTPServer): def __init__( self, @@ -425,6 +437,12 @@ class _TorchHandler(http.server.BaseHTTPRequestHandler): self._send_json(409, {"error": "cache_miss", "detail": str(exc)}) return except Exception as exc: + print( + f" [node] forward failed layers={getattr(server.backend, 'shard_start', '?')}-" + f"{getattr(server.backend, 'shard_end', '?')} session={session[:8]}: {exc}" + f"{self._request_log_suffix()}", + flush=True, + ) self._send_json(500, {"error": str(exc)}) return @@ -900,11 +918,12 @@ class _TorchHandler(http.server.BaseHTTPRequestHandler): if status == 409 and _is_cache_miss_body(resp_body): raise _PipelineCacheMiss(node_url) if status >= 400: + detail = _response_error_snippet(resp_body) print( - f" [node] relay hop {hop_index} returned {status} from {relay_addr}", + f" [node] relay hop {hop_index} returned {status} from {relay_addr}: {detail}", flush=True, ) - return f"pipeline error at {node_url} via relay: status {status}", None + return f"pipeline error at {node_url} via relay: status {status}: {detail}", None except _PipelineCacheMiss: raise except Exception as exc: @@ -929,8 +948,9 @@ class _TorchHandler(http.server.BaseHTTPRequestHandler): body = exc.read() if exc.code == 409 and _is_cache_miss_body(body): raise _PipelineCacheMiss(node_url) from exc - print(f" [node] pipeline hop {hop_index} failed at {node_url}: {exc}", flush=True) - return f"pipeline error at {node_url}: {exc}", None + detail = _response_error_snippet(body) + print(f" [node] pipeline hop {hop_index} failed at {node_url}: {exc}: {detail}", flush=True) + return f"pipeline error at {node_url}: {exc}: {detail}", None 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}", None diff --git a/tests/test_kv_cache_distributed.py b/tests/test_kv_cache_distributed.py index b4dd860..66b25ca 100644 --- a/tests/test_kv_cache_distributed.py +++ b/tests/test_kv_cache_distributed.py @@ -18,6 +18,7 @@ from meshnet_node.model_backend import ( SessionCacheStore, TailTokenResult, TensorPayload, + TorchModelShard, ) from meshnet_node.torch_server import TorchNodeServer @@ -98,6 +99,40 @@ def test_drop_removes_session(): store.lookup("s1") +def test_prefill_cache_triton_cpu_failure_disables_cache_and_retries_stateless(): + """CPU shards must recover when hybrid model cache path dispatches Triton.""" + shard = object.__new__(TorchModelShard) + shard.model_id = "fake-hybrid" + shard.supports_kv_cache = True + shard._effective_start = lambda start_layer=None: 22 + shard._new_session_cache = lambda: object() + + calls = [] + + def fake_run_layers(hidden_states, attention_mask, position_ids, *, start_layer=None, cache=None, past_len=0): + calls.append({"cache": cache, "past_len": past_len}) + if cache is not None: + raise RuntimeError("Pointer argument cannot be accessed from Triton (cpu tensor?)") + return "stateless-ok" + + shard._run_layers = fake_run_layers + + result = TorchModelShard._run_layers_session( + shard, + hidden_states=object(), + attention_mask=None, + position_ids=None, + session_id="session-1", + cache_mode="prefill", + ) + + assert result == "stateless-ok" + assert shard.supports_kv_cache is False + assert len(calls) == 2 + assert calls[0]["cache"] is not None + assert calls[1]["cache"] is None + + # --------------------------------------------------------------------------- # HTTP session protocol with fake cached backends # --------------------------------------------------------------------------- diff --git a/tests/test_node_startup.py b/tests/test_node_startup.py index cece94d..70b9cee 100644 --- a/tests/test_node_startup.py +++ b/tests/test_node_startup.py @@ -1347,6 +1347,7 @@ def test_public_tracker_model_node_registers_relay_metadata_from_tracker_url_onl output = capsys.readouterr().out assert "Relay advertised by tracker" in output assert "Cross-host pipeline hops WILL time out" not in output + assert f" Relay: {registered['relay_addr']}" in output def test_public_tracker_relay_suppresses_virtual_ip_warning(