popov/neuron-tai
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

node metrics

Browse Source
This commit is contained in:
Dobromir Popov
2026-07-14 20:33:02 +02:00
parent b661590ac7
commit ba7c656364
6 changed files with 109 additions and 17 deletions
Download Patch File Download Diff File Expand all files Collapse all files

26
packages/tracker/meshnet_tracker/server.py
View File

@@ -1494,7 +1494,9 @@ def _scale_demanded_models_locked(server: "_TrackerHTTPServer") -> None:
break
def _request_model_load_locked(server: "_TrackerHTTPServer", model_key: str) -> dict | None:
def _request_model_load_locked(
server: "_TrackerHTTPServer", model_key: str, node_id: str | None = None,
) -> dict | None:
"""Queue an explicitly requested model on the best available joined node."""
resolved_name, preset = _resolve_model_preset(server.model_presets, model_key)
if preset is None or not preset.get("hf_repo"):
@@ -1510,6 +1512,8 @@ def _request_model_load_locked(server: "_TrackerHTTPServer", model_key: str) ->
continue
host_nodes = [server.registry[item["node_id"]] for item in host["loaded"] if item["node_id"] in server.registry]
placeable = [node for node in host_nodes if _has_usable_quantization(node)]
if node_id is not None:
placeable = [node for node in placeable if node.node_id == node_id]
if not placeable:
continue
anchor = max(placeable, key=lambda node: node.benchmark_tokens_per_sec)
@@ -1556,13 +1560,17 @@ def _force_model_load_locked(server: "_TrackerHTTPServer", model_key: str) -> di
"shard_start": start, "shard_end": shard_end, "replaced_model": replaced}
def _release_model_locked(server: "_TrackerHTTPServer", model_key: str) -> int:
def _release_model_locked(
server: "_TrackerHTTPServer", model_key: str, node_id: str | None = None,
) -> int:
"""Queue DROP_SHARD for every served shard and remove it from routing immediately."""
resolved_name, preset = _resolve_model_preset(server.model_presets, model_key)
if preset is None:
return 0
released = 0
for node in server.registry.values():
if node_id is not None and node.node_id != node_id:
continue
if not _node_matches_preset(node, resolved_name, preset) or node.shard_start is None or node.shard_end is None:
continue
node.pending_directives.append(_drop_directive(node, str(preset.get("hf_repo") or resolved_name), node.shard_start, node.shard_end, node.quantization or "bfloat16"))
@@ -4880,10 +4888,14 @@ class _TrackerHandler(http.server.BaseHTTPRequestHandler):
if not isinstance(model, str) or not model.strip():
self._send_json(400, {"error": "model is required"})
return
node_id = body.get("node_id")
if node_id is not None and (not isinstance(node_id, str) or not node_id):
self._send_json(400, {"error": "node_id must be a non-empty string"})
return
with server.lock:
self._purge_expired_nodes()
assignment = _request_model_load_locked(server, model)
if assignment is None and body.get("force") is True:
assignment = _request_model_load_locked(server, model, node_id)
if assignment is None and node_id is None and body.get("force") is True:
assignment = _force_model_load_locked(server, model)
if assignment is None:
self._send_json(409, {"error": "no ready joined node has an available model slot and sufficient capacity"})
@@ -4901,9 +4913,13 @@ class _TrackerHandler(http.server.BaseHTTPRequestHandler):
if not isinstance(model, str) or not model.strip():
self._send_json(400, {"error": "model is required"})
return
node_id = body.get("node_id")
if node_id is not None and (not isinstance(node_id, str) or not node_id):
self._send_json(400, {"error": "node_id must be a non-empty string"})
return
with server.lock:
self._purge_expired_nodes()
released = _release_model_locked(server, model)
released = _release_model_locked(server, model, node_id)
if not released:
self._send_json(404, {"error": "no served shards found for model"})
return