Compare commits
9 Commits
ralph/dist
...
master
| Author | SHA1 | Date | |
|---|---|---|---|
|
47b243cd98 | ||
|
|
2852b1f80b | ||
|
|
22f28bd69a | ||
|
|
97e2784b37 | ||
|
|
ba7c656364 | ||
|
|
b661590ac7 | ||
|
|
21e6c86147 | ||
|
|
def47f1a42 | ||
|
|
8cb00e951f |
@@ -8,6 +8,11 @@ metadata:
|
||||
|
||||
# Project Status (2026-07-13)
|
||||
|
||||
## Selected-node model placement (2026-07-14)
|
||||
|
||||
- Admin Model placement now opens a node selector for load and release; the control-plane accepts optional `node_id` and targets only that registry assignment. Multi-model serving remains supported through `ADD_SHARD` and `max_loaded_shards`.
|
||||
- Total node pool resource values are rendered from `/v1/network/map`'s `node.capacity` contract. Route selection remains assignment/capability/throughput/queue based; capacity is used for placement and falls back to tracker defaults only if a node truly omits it.
|
||||
|
||||
## Distributed inference performance (2026-07-14)
|
||||
|
||||
`DIP-001` is done in `.scratch/distributed-inference-performance/`: the deterministic two-node Route Session stub benchmark covers direct/relay plus cached/stateless prefill and decode. Its JSON and concise summary explicitly attribute model execution, activation encode/decode, compression, connection setup, relay queueing, local HTTP forwarding, and end-to-end seam latency. `PYTHONPATH=packages/node pytest -q tests/test_route_session_benchmark.py` passed (7); the fixture assertion checks output-token identity and connection attempts.
|
||||
|
||||
@@ -1,127 +0,0 @@
|
||||
# DGR-001 — performance contract baseline
|
||||
|
||||
## Files changed
|
||||
|
||||
- `packages/node/meshnet_node/performance_contract.py`
|
||||
- `tests/test_performance_contract.py`
|
||||
- `.scratch/distributed-gguf-runtime/issues/01-lock-the-safetensors-versus-gguf-performance-contract.md`
|
||||
- `.scratch/distributed-gguf-runtime/evidence/DGR-001/performance-contract.json`
|
||||
|
||||
## What this slice does
|
||||
|
||||
- Locks the DGR-001 benchmark contract in code.
|
||||
- Pins the architecture-aligned baseline to **DeepSeek-V2-Lite-Chat** (`deepseek2`).
|
||||
- Uses the same model on both sides of the comparison:
|
||||
- **safetensors:** `deepseek-ai/DeepSeek-V2-Lite-Chat` in **BF16**
|
||||
- **GGUF:** `second-state/DeepSeek-V2-Lite-Chat-GGUF` in **Q2_K**
|
||||
- Exposes a machine-readable JSON contract with:
|
||||
- benchmark lanes for `transformers` safetensors and `llama.cpp` GGUF on **CPU** and **GPU**
|
||||
- concurrency levels `1` and `4`
|
||||
- the required metrics list
|
||||
- an explicit stop condition for “no meaningful speed or fit benefit”
|
||||
- Adds a deterministic stub benchmark report so the contract now has an executable report shape end to end.
|
||||
|
||||
## Recent benchmark runner slice
|
||||
|
||||
The runner currently uses a deterministic stub backend to exercise the comparison matrix without downloading a model. It emits:
|
||||
|
||||
- `.scratch/distributed-gguf-runtime/evidence/DGR-001/performance-contract.json`
|
||||
- `.scratch/distributed-gguf-runtime/evidence/DGR-001/stub-benchmark-report.json`
|
||||
|
||||
The report includes per-device comparisons for:
|
||||
|
||||
- `transformers-safetensors-cpu` vs `llama-cpp-gguf-cpu`
|
||||
- `transformers-safetensors-gpu` vs `llama-cpp-gguf-gpu`
|
||||
|
||||
and records the memory metric (`rss_bytes` on CPU, `vram_bytes` on GPU), decode speedup, artifact ratio, and output drift.
|
||||
|
||||
## Live endpoint CLI wiring
|
||||
|
||||
The contract CLI can now drive the live endpoint runner. Passing one `--live-endpoint LANE_ID=URL` mapping per contract lane (plus `--live-benchmark-out`) invokes `run_real_model_endpoint_benchmark` against already-running OpenAI-compatible servers and writes the report using the same schema as the stub:
|
||||
|
||||
```bash
|
||||
PYTHONPATH=packages/node python -m meshnet_node.performance_contract \
|
||||
--live-endpoint transformers-safetensors-cpu=http://127.0.0.1:8001 \
|
||||
--live-endpoint llama-cpp-gguf-cpu=http://127.0.0.1:8002 \
|
||||
--live-endpoint transformers-safetensors-gpu=http://127.0.0.1:8003 \
|
||||
--live-endpoint llama-cpp-gguf-gpu=http://127.0.0.1:8004 \
|
||||
--live-benchmark-out .scratch/distributed-gguf-runtime/evidence/DGR-001/live-benchmark-report.json
|
||||
```
|
||||
|
||||
`--live-model` overrides the model name sent in requests (defaults to the contract's safetensors repo). Without any `--live-endpoint` flags the CLI behaves exactly as before: it writes the contract JSON and, with `--benchmark-out`, the deterministic stub report.
|
||||
|
||||
## Exact commands and real results
|
||||
|
||||
### Targeted tests
|
||||
|
||||
```bash
|
||||
PYTHONPATH=packages/node pytest -q tests/test_performance_contract.py tests/test_route_session_benchmark.py
|
||||
```
|
||||
|
||||
Result: `19 passed in 0.11s`
|
||||
|
||||
### Contract artifact generation
|
||||
|
||||
```bash
|
||||
PYTHONPATH=packages/node python -m meshnet_node.performance_contract --json-out .scratch/distributed-gguf-runtime/evidence/DGR-001/performance-contract.json
|
||||
```
|
||||
|
||||
Result: wrote `.scratch/distributed-gguf-runtime/evidence/DGR-001/performance-contract.json`
|
||||
|
||||
### Python compile check
|
||||
|
||||
```bash
|
||||
python -m compileall packages/node/meshnet_node/performance_contract.py tests/test_performance_contract.py
|
||||
```
|
||||
|
||||
Result: passed
|
||||
|
||||
## Public relay smoke benchmark (2026-07-15)
|
||||
|
||||
A real streamed request was run through the public tracker — **not** by connecting directly to the private node address:
|
||||
|
||||
```text
|
||||
https://meshnet.2.d-popov.com/v1/chat/completions
|
||||
-> wss://meshnet.2.d-popov.com/ws
|
||||
-> wss://meshnet.2.d-popov.com/rpc/7j77FsPY1evV8tuf-7000
|
||||
-> local CUDA node, Qwen/Qwen2.5-0.5B-Instruct layers 0-23
|
||||
```
|
||||
|
||||
The local public-tracker node had an expired proof and a wedged HTTP server. A graceful restart refreshed its CUDA capability proof in `336 ms`, restored `admitted`/`routable` status, and reconnected its relay endpoint.
|
||||
|
||||
Measured streaming results after recovery:
|
||||
|
||||
| metric | result |
|
||||
| --- | ---: |
|
||||
| warm-up TTFT | 420.80 ms |
|
||||
| warm-up elapsed | 610.23 ms |
|
||||
| p50 TTFT (3 runs) | 288.26 ms |
|
||||
| p50 elapsed (3 runs) | 363.20 ms |
|
||||
| tracker-recorded relay throughput | 58.18-65.25 tok/s |
|
||||
| HTTP status | 200 for all runs |
|
||||
|
||||
The tracker recorded `relay: true` and the local node ID `7j77FsPY-b32476219492` for each completion. Full redacted evidence is in `public-relay-smoke-benchmark.json`.
|
||||
|
||||
The other connected node is still alive but **not routable** because its capability proof is stale. It must revalidate before a multi-node shard/relay test can run.
|
||||
|
||||
## Limitations
|
||||
|
||||
- This slice still uses a deterministic stub backend for the core comparison matrix.
|
||||
- It now also includes a live endpoint runner, reachable from the CLI via `--live-endpoint`/`--live-benchmark-out`, that fans out one OpenAI-compatible request per lane when the caller provides endpoints; the CLI does not start those servers.
|
||||
- It does **not** download or run a real model from within the repo.
|
||||
- Real safetensors vs GGUF execution, TTFT/prefill/decode measurements, RSS/VRAM capture, and output-drift comparison are still to be implemented against the contract.
|
||||
|
||||
## Compatibility notes
|
||||
|
||||
- The contract stays on the DeepSeek2 family to remain close to the DeepSeek-V4-Flash end goal.
|
||||
- A smaller non-DeepSeek model can still be used later for loader-plumbing smoke tests, but it does not replace this baseline.
|
||||
- Model artifacts must stay on the mounted drive and not under `/home`.
|
||||
|
||||
## Dependent-story handoff
|
||||
|
||||
Next implementation work should attach to this contract and add the live benchmark runner that actually compares:
|
||||
|
||||
1. current Transformers/safetensors recipe
|
||||
2. whole-model llama.cpp GGUF recipe
|
||||
|
||||
using the same model architecture/revision and the same prompt/context/concurrency settings.
|
||||
@@ -1,75 +0,0 @@
|
||||
{
|
||||
"benchmark_lanes": [
|
||||
{
|
||||
"concurrency_levels": [
|
||||
1,
|
||||
4
|
||||
],
|
||||
"device": "cpu",
|
||||
"id": "transformers-safetensors-cpu",
|
||||
"recipe": "current safetensors recipe",
|
||||
"runtime": "transformers"
|
||||
},
|
||||
{
|
||||
"concurrency_levels": [
|
||||
1,
|
||||
4
|
||||
],
|
||||
"device": "cpu",
|
||||
"id": "llama-cpp-gguf-cpu",
|
||||
"recipe": "whole-model GGUF recipe",
|
||||
"runtime": "llama.cpp"
|
||||
},
|
||||
{
|
||||
"concurrency_levels": [
|
||||
1,
|
||||
4
|
||||
],
|
||||
"device": "gpu",
|
||||
"id": "transformers-safetensors-gpu",
|
||||
"recipe": "current safetensors recipe",
|
||||
"runtime": "transformers"
|
||||
},
|
||||
{
|
||||
"concurrency_levels": [
|
||||
1,
|
||||
4
|
||||
],
|
||||
"device": "gpu",
|
||||
"id": "llama-cpp-gguf-gpu",
|
||||
"recipe": "whole-model GGUF recipe",
|
||||
"runtime": "llama.cpp"
|
||||
}
|
||||
],
|
||||
"metrics": [
|
||||
"ttft_ms",
|
||||
"prefill_tok_per_sec",
|
||||
"decode_tok_per_sec",
|
||||
"p50_latency_ms",
|
||||
"p95_latency_ms",
|
||||
"aggregate_throughput_tok_per_sec",
|
||||
"rss_bytes",
|
||||
"vram_bytes",
|
||||
"artifact_bytes",
|
||||
"failure_count",
|
||||
"output_drift"
|
||||
],
|
||||
"model_target": {
|
||||
"architecture": "deepseek2",
|
||||
"comparison_policy": "same model/revision, closest practical low-footprint precision pair: BF16 safetensors versus Q2_K GGUF",
|
||||
"gguf_quant": "Q2_K",
|
||||
"gguf_repo": "second-state/DeepSeek-V2-Lite-Chat-GGUF",
|
||||
"gguf_size_gb": 6.43,
|
||||
"name": "DeepSeek-V2-Lite-Chat",
|
||||
"rationale": "Smallest DeepSeek-family benchmark anchor that still points toward DeepSeek-V4-Flash; keeps the runtime on the DeepSeek2 path instead of falling back to a tiny but architecture-mismatched smoke model.",
|
||||
"safetensors_precision": "bfloat16",
|
||||
"safetensors_repo": "deepseek-ai/DeepSeek-V2-Lite-Chat"
|
||||
},
|
||||
"notes": [
|
||||
"Real model execution stays opt-in and must keep model artifacts on the mounted drive.",
|
||||
"Use the tiny fallback only for loader plumbing smoke tests; it does not replace the architecture-aligned baseline."
|
||||
],
|
||||
"schema_version": 1,
|
||||
"stop_condition": "Stop if GGUF does not provide a meaningful speed or fit benefit over the safetensors baseline for the chosen DeepSeek-family model target.",
|
||||
"story_id": "DGR-001"
|
||||
}
|
||||
@@ -1,83 +0,0 @@
|
||||
{
|
||||
"schema_version": 1,
|
||||
"executed_at_utc": "2026-07-15T10:41:14Z",
|
||||
"test_kind": "public-relay-single-node-streaming-smoke-benchmark",
|
||||
"target": {
|
||||
"public_chat_endpoint": "https://meshnet.2.d-popov.com/v1/chat/completions",
|
||||
"relay_url": "wss://meshnet.2.d-popov.com/ws",
|
||||
"model": "qwen2.5-0.5b-instruct",
|
||||
"quantization": "bfloat16"
|
||||
},
|
||||
"recovery": {
|
||||
"problem": "The local node's capability proof had expired and its port-7000 HTTP server had wedged with CLOSE-WAIT sockets.",
|
||||
"action": "Gracefully restarted the local public-tracker meshnet-node process on port 7000.",
|
||||
"startup_validation": {
|
||||
"device": "cuda",
|
||||
"capability_proof_ms": 336,
|
||||
"node_id": "7j77FsPY-b32476219492",
|
||||
"relay_addr": "wss://meshnet.2.d-popov.com/rpc/7j77FsPY1evV8tuf-7000"
|
||||
}
|
||||
},
|
||||
"tracker_admission_after_recovery": {
|
||||
"node_id": "7j77FsPY-b32476219492",
|
||||
"alive": true,
|
||||
"status": "ready",
|
||||
"capability_state": "admitted",
|
||||
"routable": true,
|
||||
"route_hops": 1
|
||||
},
|
||||
"client_measurements": {
|
||||
"warmup": {
|
||||
"http_status": 200,
|
||||
"ttft_ms": 420.8,
|
||||
"elapsed_ms": 610.23,
|
||||
"response_text": "MeshNet Relay Benchmark Passed"
|
||||
},
|
||||
"runs": [
|
||||
{
|
||||
"run": 1,
|
||||
"ttft_ms": 376.04,
|
||||
"elapsed_ms": 458.65,
|
||||
"response_text": "relay benchmark pass"
|
||||
},
|
||||
{
|
||||
"run": 2,
|
||||
"ttft_ms": 258.33,
|
||||
"elapsed_ms": 336.71,
|
||||
"response_text": "relay benchmark pass"
|
||||
},
|
||||
{
|
||||
"run": 3,
|
||||
"ttft_ms": 288.26,
|
||||
"elapsed_ms": 363.2,
|
||||
"response_text": "relay benchmark pass"
|
||||
}
|
||||
],
|
||||
"p50_ttft_ms": 288.26,
|
||||
"p50_elapsed_ms": 363.2
|
||||
},
|
||||
"tracker_relay_evidence": [
|
||||
{
|
||||
"status": 200,
|
||||
"relay": true,
|
||||
"node_id": "7j77FsPY-b32476219492",
|
||||
"tokens": 11,
|
||||
"elapsed_seconds": 0.1686,
|
||||
"tokens_per_sec": 65.2541
|
||||
},
|
||||
{
|
||||
"status": 200,
|
||||
"relay": true,
|
||||
"node_id": "7j77FsPY-b32476219492",
|
||||
"tokens": 11,
|
||||
"elapsed_seconds": 0.1891,
|
||||
"tokens_per_sec": 58.1799
|
||||
}
|
||||
],
|
||||
"scope_and_remaining_work": {
|
||||
"validated": "Public HTTPS chat endpoint routed a streaming request through the tracker relay to the local CUDA node and completed with HTTP 200.",
|
||||
"not_validated": "Two-node shard routing was not run because the remote node 5gMLrmyB-88f5cba044d0 still had an expired capability proof and was not routable.",
|
||||
"next_gate": "Refresh the remote node capability proof, then load a multi-node-compatible assignment and repeat the benchmark through the public tracker relay."
|
||||
},
|
||||
"reproduction": "Use a valid bearer API key with the public /v1/chat/completions endpoint and stream a short qwen2.5-0.5b-instruct request. Do not connect directly to private node HTTP endpoints; the tracker relay is the required path."
|
||||
}
|
||||
@@ -1,247 +0,0 @@
|
||||
{
|
||||
"comparisons": {
|
||||
"cpu": {
|
||||
"artifact_bytes_ratio": 0.2048,
|
||||
"decode_speedup": 2.3333,
|
||||
"gguf_benefit": true,
|
||||
"gguf_lane": "llama-cpp-gguf-cpu",
|
||||
"memory_bytes_ratio": 0.2152,
|
||||
"memory_metric": "rss_bytes",
|
||||
"output_drift": 0.0,
|
||||
"safetensors_lane": "transformers-safetensors-cpu",
|
||||
"ttft_speedup": 1.8947
|
||||
},
|
||||
"gpu": {
|
||||
"artifact_bytes_ratio": 0.2048,
|
||||
"decode_speedup": 1.5294,
|
||||
"gguf_benefit": true,
|
||||
"gguf_lane": "llama-cpp-gguf-gpu",
|
||||
"memory_bytes_ratio": 0.2273,
|
||||
"memory_metric": "vram_bytes",
|
||||
"output_drift": 0.0,
|
||||
"safetensors_lane": "transformers-safetensors-gpu",
|
||||
"ttft_speedup": 1.6154
|
||||
}
|
||||
},
|
||||
"lanes": [
|
||||
{
|
||||
"concurrency_levels": [
|
||||
1,
|
||||
4
|
||||
],
|
||||
"device": "cpu",
|
||||
"id": "transformers-safetensors-cpu",
|
||||
"output_tokens": [
|
||||
"mesh",
|
||||
"activation",
|
||||
"seam",
|
||||
"baseline"
|
||||
],
|
||||
"recipe": "current safetensors recipe",
|
||||
"results": [
|
||||
{
|
||||
"concurrency": 1,
|
||||
"metrics": {
|
||||
"aggregate_throughput_tok_per_sec": 6.0,
|
||||
"artifact_bytes": 33715493273,
|
||||
"decode_tok_per_sec": 6.0,
|
||||
"failure_count": 0,
|
||||
"output_drift": 0.0,
|
||||
"p50_latency_ms": 166.6667,
|
||||
"p95_latency_ms": 208.3334,
|
||||
"prefill_tok_per_sec": 45.0,
|
||||
"rss_bytes": 35433480192,
|
||||
"ttft_ms": 1800.0,
|
||||
"vram_bytes": 0
|
||||
}
|
||||
},
|
||||
{
|
||||
"concurrency": 4,
|
||||
"metrics": {
|
||||
"aggregate_throughput_tok_per_sec": 20.4,
|
||||
"artifact_bytes": 33715493273,
|
||||
"decode_tok_per_sec": 5.1,
|
||||
"failure_count": 0,
|
||||
"output_drift": 0.0,
|
||||
"p50_latency_ms": 196.0784,
|
||||
"p95_latency_ms": 245.098,
|
||||
"prefill_tok_per_sec": 38.25,
|
||||
"rss_bytes": 35433480192,
|
||||
"ttft_ms": 2340.0,
|
||||
"vram_bytes": 0
|
||||
}
|
||||
}
|
||||
],
|
||||
"runtime": "transformers"
|
||||
},
|
||||
{
|
||||
"concurrency_levels": [
|
||||
1,
|
||||
4
|
||||
],
|
||||
"device": "cpu",
|
||||
"id": "llama-cpp-gguf-cpu",
|
||||
"output_tokens": [
|
||||
"mesh",
|
||||
"activation",
|
||||
"seam",
|
||||
"baseline"
|
||||
],
|
||||
"recipe": "whole-model GGUF recipe",
|
||||
"results": [
|
||||
{
|
||||
"concurrency": 1,
|
||||
"metrics": {
|
||||
"aggregate_throughput_tok_per_sec": 14.0,
|
||||
"artifact_bytes": 6904159928,
|
||||
"decode_tok_per_sec": 14.0,
|
||||
"failure_count": 0,
|
||||
"output_drift": 0.0,
|
||||
"p50_latency_ms": 71.4286,
|
||||
"p95_latency_ms": 89.2858,
|
||||
"prefill_tok_per_sec": 90.0,
|
||||
"rss_bytes": 7623566950,
|
||||
"ttft_ms": 950.0,
|
||||
"vram_bytes": 0
|
||||
}
|
||||
},
|
||||
{
|
||||
"concurrency": 4,
|
||||
"metrics": {
|
||||
"aggregate_throughput_tok_per_sec": 47.6,
|
||||
"artifact_bytes": 6904159928,
|
||||
"decode_tok_per_sec": 11.9,
|
||||
"failure_count": 0,
|
||||
"output_drift": 0.0,
|
||||
"p50_latency_ms": 84.0336,
|
||||
"p95_latency_ms": 105.042,
|
||||
"prefill_tok_per_sec": 76.5,
|
||||
"rss_bytes": 7623566950,
|
||||
"ttft_ms": 1235.0,
|
||||
"vram_bytes": 0
|
||||
}
|
||||
}
|
||||
],
|
||||
"runtime": "llama.cpp"
|
||||
},
|
||||
{
|
||||
"concurrency_levels": [
|
||||
1,
|
||||
4
|
||||
],
|
||||
"device": "gpu",
|
||||
"id": "transformers-safetensors-gpu",
|
||||
"output_tokens": [
|
||||
"mesh",
|
||||
"activation",
|
||||
"seam",
|
||||
"baseline"
|
||||
],
|
||||
"recipe": "current safetensors recipe",
|
||||
"results": [
|
||||
{
|
||||
"concurrency": 1,
|
||||
"metrics": {
|
||||
"aggregate_throughput_tok_per_sec": 34.0,
|
||||
"artifact_bytes": 33715493273,
|
||||
"decode_tok_per_sec": 34.0,
|
||||
"failure_count": 0,
|
||||
"output_drift": 0.0,
|
||||
"p50_latency_ms": 29.4118,
|
||||
"p95_latency_ms": 36.7647,
|
||||
"prefill_tok_per_sec": 850.0,
|
||||
"rss_bytes": 4294967296,
|
||||
"ttft_ms": 420.0,
|
||||
"vram_bytes": 35433480192
|
||||
}
|
||||
},
|
||||
{
|
||||
"concurrency": 4,
|
||||
"metrics": {
|
||||
"aggregate_throughput_tok_per_sec": 115.6,
|
||||
"artifact_bytes": 33715493273,
|
||||
"decode_tok_per_sec": 28.9,
|
||||
"failure_count": 0,
|
||||
"output_drift": 0.0,
|
||||
"p50_latency_ms": 34.6021,
|
||||
"p95_latency_ms": 43.2526,
|
||||
"prefill_tok_per_sec": 722.5,
|
||||
"rss_bytes": 4294967296,
|
||||
"ttft_ms": 546.0,
|
||||
"vram_bytes": 35433480192
|
||||
}
|
||||
}
|
||||
],
|
||||
"runtime": "transformers"
|
||||
},
|
||||
{
|
||||
"concurrency_levels": [
|
||||
1,
|
||||
4
|
||||
],
|
||||
"device": "gpu",
|
||||
"id": "llama-cpp-gguf-gpu",
|
||||
"output_tokens": [
|
||||
"mesh",
|
||||
"activation",
|
||||
"seam",
|
||||
"baseline"
|
||||
],
|
||||
"recipe": "whole-model GGUF recipe",
|
||||
"results": [
|
||||
{
|
||||
"concurrency": 1,
|
||||
"metrics": {
|
||||
"aggregate_throughput_tok_per_sec": 52.0,
|
||||
"artifact_bytes": 6904159928,
|
||||
"decode_tok_per_sec": 52.0,
|
||||
"failure_count": 0,
|
||||
"output_drift": 0.0,
|
||||
"p50_latency_ms": 19.2308,
|
||||
"p95_latency_ms": 24.0385,
|
||||
"prefill_tok_per_sec": 640.0,
|
||||
"rss_bytes": 1610612736,
|
||||
"ttft_ms": 260.0,
|
||||
"vram_bytes": 8053063680
|
||||
}
|
||||
},
|
||||
{
|
||||
"concurrency": 4,
|
||||
"metrics": {
|
||||
"aggregate_throughput_tok_per_sec": 176.8,
|
||||
"artifact_bytes": 6904159928,
|
||||
"decode_tok_per_sec": 44.2,
|
||||
"failure_count": 0,
|
||||
"output_drift": 0.0,
|
||||
"p50_latency_ms": 22.6244,
|
||||
"p95_latency_ms": 28.2805,
|
||||
"prefill_tok_per_sec": 544.0,
|
||||
"rss_bytes": 1610612736,
|
||||
"ttft_ms": 338.0,
|
||||
"vram_bytes": 8053063680
|
||||
}
|
||||
}
|
||||
],
|
||||
"runtime": "llama.cpp"
|
||||
}
|
||||
],
|
||||
"model_target": {
|
||||
"architecture": "deepseek2",
|
||||
"comparison_policy": "same model/revision, closest practical low-footprint precision pair: BF16 safetensors versus Q2_K GGUF",
|
||||
"gguf_quant": "Q2_K",
|
||||
"gguf_repo": "second-state/DeepSeek-V2-Lite-Chat-GGUF",
|
||||
"gguf_size_gb": 6.43,
|
||||
"name": "DeepSeek-V2-Lite-Chat",
|
||||
"rationale": "Smallest DeepSeek-family benchmark anchor that still points toward DeepSeek-V4-Flash; keeps the runtime on the DeepSeek2 path instead of falling back to a tiny but architecture-mismatched smoke model.",
|
||||
"safetensors_precision": "bfloat16",
|
||||
"safetensors_repo": "deepseek-ai/DeepSeek-V2-Lite-Chat"
|
||||
},
|
||||
"schema_version": 1,
|
||||
"source": "stub-backend",
|
||||
"stop_condition": {
|
||||
"gguf_benefit": true,
|
||||
"text": "Stop if GGUF does not provide a meaningful speed or fit benefit over the safetensors baseline for the chosen DeepSeek-family model target.",
|
||||
"triggered": false
|
||||
},
|
||||
"story_id": "DGR-001"
|
||||
}
|
||||
@@ -13,15 +13,6 @@ Status: ready-for-agent
|
||||
|
||||
As a runtime engineer, I need a controlled baseline so that GGUF work proceeds from measured speed, memory, and quality rather than reputation.
|
||||
|
||||
## Baseline model target
|
||||
|
||||
Use the same model on both sides of the comparison, with the closest practical low-footprint precision pair:
|
||||
|
||||
- **safetensors:** `deepseek-ai/DeepSeek-V2-Lite-Chat` in **BF16**
|
||||
- **GGUF:** `second-state/DeepSeek-V2-Lite-Chat-GGUF` in **Q2_K** (~6.5GB)
|
||||
|
||||
Keep the benchmark matrix explicit for **CPU** and **GPU** runs. Reserve smaller non-DeepSeek fallback models only for loader plumbing smoke tests if needed; they do not count as the DGR-001 architecture-aligned baseline.
|
||||
|
||||
## Expected durable outputs
|
||||
|
||||
- Benchmark harness and deterministic tests
|
||||
|
||||
@@ -16,12 +16,9 @@
|
||||
|
||||
|
||||
.\.venv\Scripts\meshnet-node.exe start http://192.168.0.179:8081 --model-id Qwen/Qwen2.5-0.5B-Instruct --advertise-host 192.168.0.20
|
||||
.\.venv\Scripts\meshnet-node.exe start --tracker http://ai.neuron.d-popov.com --model-id Qwen/Qwen2.5-0.5B-Instruct --advertise-host 192.168.0.20
|
||||
.\.venv\Scripts\meshnet-node.exe start --tracker http://ai.neuron.d-popov.com --model Qwen/Qwen2.5-0.5B-Instruct --advertise-host 192.168.0.20
|
||||
|
||||
we .\.venv\Scripts\meshnet-node.exe start `
|
||||
--tracker http://192.168.0.179:8081 `
|
||||
--model Qwen/Qwen2.5-0.5B-Instruct `
|
||||
--advertise-host 192.168.0.20
|
||||
we .\.venv\Scripts\meshnet-node.exe start --tracker http://192.168.0.179:8081 --model Qwen/Qwen2.5-0.5B-Instruct
|
||||
# trackers:
|
||||
https://meshnet.2.d-popov.com
|
||||
https://ai.neuron.d-popov.com
|
||||
|
||||
@@ -2,6 +2,7 @@
|
||||
|
||||
import json
|
||||
import os
|
||||
import shutil
|
||||
import subprocess
|
||||
import time
|
||||
|
||||
@@ -183,6 +184,17 @@ def with_forced_cpu(hw: dict) -> dict:
|
||||
return forced
|
||||
|
||||
|
||||
def _with_model_drive(profile: dict) -> dict:
|
||||
"""Attach free space for the default model cache drive to tracker diagnostics."""
|
||||
try:
|
||||
cache_root = os.path.expanduser("~/.cache/meshnet/shards")
|
||||
profile["model_drive_free_bytes"] = shutil.disk_usage(os.path.expanduser("~")).free
|
||||
profile["model_drive_path"] = cache_root
|
||||
except OSError:
|
||||
pass
|
||||
return profile
|
||||
|
||||
|
||||
def detect_hardware() -> dict:
|
||||
"""Detect GPU model and available VRAM. Returns hardware profile dict."""
|
||||
ram_mb = _detect_ram_mb()
|
||||
@@ -208,23 +220,23 @@ def detect_hardware() -> dict:
|
||||
}
|
||||
if torch_gpu is not None and torch_gpu.get("gcn_arch"):
|
||||
profile["gcn_arch"] = torch_gpu["gcn_arch"]
|
||||
return profile
|
||||
return _with_model_drive(profile)
|
||||
except ImportError:
|
||||
pass
|
||||
|
||||
torch_inventory = _gpu_inventory_profile(torch_gpu, ram_mb)
|
||||
if torch_inventory is not None:
|
||||
return torch_inventory
|
||||
return _with_model_drive(torch_inventory)
|
||||
|
||||
nvidia_gpu = _gpu_inventory_profile(_detect_nvidia_smi_gpu_memory(), ram_mb)
|
||||
if nvidia_gpu is not None:
|
||||
return nvidia_gpu
|
||||
return _with_model_drive(nvidia_gpu)
|
||||
|
||||
windows_gpu = _gpu_inventory_profile(_detect_windows_gpu_memory(), ram_mb)
|
||||
if windows_gpu is not None:
|
||||
return windows_gpu
|
||||
return _with_model_drive(windows_gpu)
|
||||
|
||||
return {
|
||||
return _with_model_drive({
|
||||
"device": "cpu",
|
||||
"gpu_name": None,
|
||||
"vram_mb": 0,
|
||||
@@ -232,7 +244,7 @@ def detect_hardware() -> dict:
|
||||
"shared_vram_mb": 0,
|
||||
"ram_mb": ram_mb,
|
||||
"cuda_available": False,
|
||||
}
|
||||
})
|
||||
|
||||
|
||||
def benchmark_throughput_checked(device_str: str = "cpu") -> tuple[float, bool, str | None]:
|
||||
|
||||
@@ -1,495 +0,0 @@
|
||||
"""Versioned performance contract metadata and stub benchmark runner for DGR-001.
|
||||
|
||||
This module captures the *contract* first: the model family, architecture
|
||||
alignment, benchmark lanes, and stop condition that benchmark runs must
|
||||
satisfy. It also runs the contract's lanes through a deterministic stub
|
||||
backend so the report data shape exists end to end. It never downloads or
|
||||
executes a model; real transformers / llama.cpp backends plug in behind the
|
||||
same ``run()`` seam later.
|
||||
"""
|
||||
|
||||
from __future__ import annotations
|
||||
|
||||
import argparse
|
||||
import json
|
||||
import time
|
||||
import urllib.request
|
||||
from dataclasses import dataclass
|
||||
from pathlib import Path
|
||||
from typing import Mapping
|
||||
|
||||
SCHEMA_VERSION = 1
|
||||
CONTRACT_ID = "DGR-001"
|
||||
DEFAULT_OUTPUT_PATH = Path(".scratch/distributed-gguf-runtime/evidence/DGR-001/performance-contract.json")
|
||||
|
||||
|
||||
@dataclass(frozen=True)
|
||||
class ModelTarget:
|
||||
"""Architecture-aligned model target for the DGR-001 benchmark contract."""
|
||||
|
||||
name: str
|
||||
architecture: str
|
||||
safetensors_repo: str
|
||||
safetensors_precision: str
|
||||
gguf_repo: str
|
||||
gguf_quant: str
|
||||
gguf_size_gb: float
|
||||
comparison_policy: str
|
||||
rationale: str
|
||||
|
||||
def to_dict(self) -> dict:
|
||||
return {
|
||||
"name": self.name,
|
||||
"architecture": self.architecture,
|
||||
"safetensors_repo": self.safetensors_repo,
|
||||
"safetensors_precision": self.safetensors_precision,
|
||||
"gguf_repo": self.gguf_repo,
|
||||
"gguf_quant": self.gguf_quant,
|
||||
"gguf_size_gb": self.gguf_size_gb,
|
||||
"comparison_policy": self.comparison_policy,
|
||||
"rationale": self.rationale,
|
||||
}
|
||||
|
||||
|
||||
@dataclass(frozen=True)
|
||||
class BenchmarkLane:
|
||||
"""One side of the comparison the contract requires."""
|
||||
|
||||
id: str
|
||||
runtime: str
|
||||
device: str
|
||||
recipe: str
|
||||
concurrency_levels: tuple[int, ...]
|
||||
|
||||
def to_dict(self) -> dict:
|
||||
return {
|
||||
"id": self.id,
|
||||
"runtime": self.runtime,
|
||||
"device": self.device,
|
||||
"recipe": self.recipe,
|
||||
"concurrency_levels": list(self.concurrency_levels),
|
||||
}
|
||||
|
||||
|
||||
@dataclass(frozen=True)
|
||||
class PerformanceContract:
|
||||
"""Machine-readable contract for the DGR-001 benchmark story."""
|
||||
|
||||
schema_version: int
|
||||
story_id: str
|
||||
model_target: ModelTarget
|
||||
benchmark_lanes: tuple[BenchmarkLane, ...]
|
||||
metrics: tuple[str, ...]
|
||||
stop_condition: str
|
||||
notes: tuple[str, ...] = ()
|
||||
|
||||
def to_dict(self) -> dict:
|
||||
return {
|
||||
"schema_version": self.schema_version,
|
||||
"story_id": self.story_id,
|
||||
"model_target": self.model_target.to_dict(),
|
||||
"benchmark_lanes": [lane.to_dict() for lane in self.benchmark_lanes],
|
||||
"metrics": list(self.metrics),
|
||||
"stop_condition": self.stop_condition,
|
||||
"notes": list(self.notes),
|
||||
}
|
||||
|
||||
def write_json(self, path: str | Path) -> Path:
|
||||
path = Path(path)
|
||||
path.parent.mkdir(parents=True, exist_ok=True)
|
||||
path.write_text(json.dumps(self.to_dict(), indent=2, sort_keys=True) + "\n", encoding="utf-8")
|
||||
return path
|
||||
|
||||
|
||||
DEFAULT_CONTRACT = PerformanceContract(
|
||||
schema_version=SCHEMA_VERSION,
|
||||
story_id=CONTRACT_ID,
|
||||
model_target=ModelTarget(
|
||||
name="DeepSeek-V2-Lite-Chat",
|
||||
architecture="deepseek2",
|
||||
safetensors_repo="deepseek-ai/DeepSeek-V2-Lite-Chat",
|
||||
safetensors_precision="bfloat16",
|
||||
gguf_repo="second-state/DeepSeek-V2-Lite-Chat-GGUF",
|
||||
gguf_quant="Q2_K",
|
||||
gguf_size_gb=6.43,
|
||||
comparison_policy=(
|
||||
"same model/revision, closest practical low-footprint precision pair: "
|
||||
"BF16 safetensors versus Q2_K GGUF"
|
||||
),
|
||||
rationale=(
|
||||
"Smallest DeepSeek-family benchmark anchor that still points toward "
|
||||
"DeepSeek-V4-Flash; keeps the runtime on the DeepSeek2 path instead "
|
||||
"of falling back to a tiny but architecture-mismatched smoke model."
|
||||
),
|
||||
),
|
||||
benchmark_lanes=(
|
||||
BenchmarkLane(
|
||||
id="transformers-safetensors-cpu",
|
||||
runtime="transformers",
|
||||
device="cpu",
|
||||
recipe="current safetensors recipe",
|
||||
concurrency_levels=(1, 4),
|
||||
),
|
||||
BenchmarkLane(
|
||||
id="llama-cpp-gguf-cpu",
|
||||
runtime="llama.cpp",
|
||||
device="cpu",
|
||||
recipe="whole-model GGUF recipe",
|
||||
concurrency_levels=(1, 4),
|
||||
),
|
||||
BenchmarkLane(
|
||||
id="transformers-safetensors-gpu",
|
||||
runtime="transformers",
|
||||
device="gpu",
|
||||
recipe="current safetensors recipe",
|
||||
concurrency_levels=(1, 4),
|
||||
),
|
||||
BenchmarkLane(
|
||||
id="llama-cpp-gguf-gpu",
|
||||
runtime="llama.cpp",
|
||||
device="gpu",
|
||||
recipe="whole-model GGUF recipe",
|
||||
concurrency_levels=(1, 4),
|
||||
),
|
||||
),
|
||||
metrics=(
|
||||
"ttft_ms",
|
||||
"prefill_tok_per_sec",
|
||||
"decode_tok_per_sec",
|
||||
"p50_latency_ms",
|
||||
"p95_latency_ms",
|
||||
"aggregate_throughput_tok_per_sec",
|
||||
"rss_bytes",
|
||||
"vram_bytes",
|
||||
"artifact_bytes",
|
||||
"failure_count",
|
||||
"output_drift",
|
||||
),
|
||||
stop_condition=(
|
||||
"Stop if GGUF does not provide a meaningful speed or fit benefit over the "
|
||||
"safetensors baseline for the chosen DeepSeek-family model target."
|
||||
),
|
||||
notes=(
|
||||
"Real model execution stays opt-in and must keep model artifacts on the mounted drive.",
|
||||
"Use the tiny fallback only for loader plumbing smoke tests; it does not replace the architecture-aligned baseline.",
|
||||
),
|
||||
)
|
||||
|
||||
|
||||
def build_default_contract() -> PerformanceContract:
|
||||
return DEFAULT_CONTRACT
|
||||
|
||||
|
||||
BENCHMARK_SCHEMA_VERSION = 1
|
||||
STUB_OUTPUT_TOKENS = ("mesh", "activation", "seam", "baseline")
|
||||
# DeepSeek-V2-Lite is ~15.7B params at 2 bytes each; metadata only, nothing downloaded.
|
||||
_SAFETENSORS_BF16_ARTIFACT_GB = 31.4
|
||||
|
||||
|
||||
@dataclass(frozen=True)
|
||||
class LaneSample:
|
||||
"""Raw single-stream measurements one backend produces for a lane."""
|
||||
|
||||
ttft_ms: float
|
||||
prefill_tok_per_sec: float
|
||||
decode_tok_per_sec: float
|
||||
rss_bytes: int
|
||||
vram_bytes: int
|
||||
artifact_bytes: int
|
||||
output_tokens: tuple[str, ...]
|
||||
failure_count: int = 0
|
||||
|
||||
|
||||
def _gb(value: float) -> int:
|
||||
return int(value * 1024**3)
|
||||
|
||||
|
||||
class StubLaneBackend:
|
||||
"""Deterministic placeholder measurements until real lane execution lands.
|
||||
|
||||
The numbers are synthetic but directionally shaped — the Q2_K GGUF loads a
|
||||
far smaller artifact and decodes faster than BF16 safetensors — so the
|
||||
comparison and stop-condition plumbing can be exercised in CI.
|
||||
"""
|
||||
|
||||
source = "stub-backend"
|
||||
|
||||
# (runtime, device) -> (ttft_ms, prefill tok/s, decode tok/s, rss GB, vram GB)
|
||||
_PROFILES = {
|
||||
("transformers", "cpu"): (1800.0, 45.0, 6.0, 33.0, 0.0),
|
||||
("llama.cpp", "cpu"): (950.0, 90.0, 14.0, 7.1, 0.0),
|
||||
("transformers", "gpu"): (420.0, 850.0, 34.0, 4.0, 33.0),
|
||||
("llama.cpp", "gpu"): (260.0, 640.0, 52.0, 1.5, 7.5),
|
||||
}
|
||||
|
||||
def __init__(self, contract: PerformanceContract) -> None:
|
||||
self._contract = contract
|
||||
|
||||
def run(self, lane: BenchmarkLane) -> LaneSample:
|
||||
ttft_ms, prefill, decode, rss_gb, vram_gb = self._PROFILES[(lane.runtime, lane.device)]
|
||||
artifact_gb = (
|
||||
self._contract.model_target.gguf_size_gb
|
||||
if lane.runtime == "llama.cpp"
|
||||
else _SAFETENSORS_BF16_ARTIFACT_GB
|
||||
)
|
||||
return LaneSample(
|
||||
ttft_ms=ttft_ms,
|
||||
prefill_tok_per_sec=prefill,
|
||||
decode_tok_per_sec=decode,
|
||||
rss_bytes=_gb(rss_gb),
|
||||
vram_bytes=_gb(vram_gb),
|
||||
artifact_bytes=_gb(artifact_gb),
|
||||
output_tokens=STUB_OUTPUT_TOKENS,
|
||||
)
|
||||
|
||||
|
||||
def _output_drift(tokens: tuple[str, ...], reference: tuple[str, ...]) -> float:
|
||||
"""Fraction of positions where a lane's output diverges from its reference."""
|
||||
length = max(len(tokens), len(reference))
|
||||
if length == 0:
|
||||
return 0.0
|
||||
mismatches = sum(a != b for a, b in zip(tokens, reference)) + abs(len(tokens) - len(reference))
|
||||
return round(mismatches / length, 4)
|
||||
|
||||
|
||||
def _metrics_for(sample: LaneSample, concurrency: int, output_drift: float) -> dict:
|
||||
# Stub concurrency model: batching scales throughput at 85% efficiency and
|
||||
# stretches per-request token latency and TTFT accordingly.
|
||||
efficiency = 1.0 if concurrency == 1 else 0.85
|
||||
p50_latency_ms = round(1000.0 / (sample.decode_tok_per_sec * efficiency), 4)
|
||||
return {
|
||||
"ttft_ms": round(sample.ttft_ms * (1 + 0.1 * (concurrency - 1)), 4),
|
||||
"prefill_tok_per_sec": round(sample.prefill_tok_per_sec * efficiency, 4),
|
||||
"decode_tok_per_sec": round(sample.decode_tok_per_sec * efficiency, 4),
|
||||
"p50_latency_ms": p50_latency_ms,
|
||||
"p95_latency_ms": round(p50_latency_ms * 1.25, 4),
|
||||
"aggregate_throughput_tok_per_sec": round(sample.decode_tok_per_sec * concurrency * efficiency, 4),
|
||||
"rss_bytes": sample.rss_bytes,
|
||||
"vram_bytes": sample.vram_bytes,
|
||||
"artifact_bytes": sample.artifact_bytes,
|
||||
"failure_count": sample.failure_count,
|
||||
"output_drift": output_drift,
|
||||
}
|
||||
|
||||
|
||||
def _compare_device(lanes: list[tuple[BenchmarkLane, LaneSample]], device: str) -> dict:
|
||||
by_runtime = {lane.runtime: (lane, sample) for lane, sample in lanes if lane.device == device}
|
||||
safetensors_lane, safetensors = by_runtime["transformers"]
|
||||
gguf_lane, gguf = by_runtime["llama.cpp"]
|
||||
memory_metric = "vram_bytes" if device == "gpu" else "rss_bytes"
|
||||
decode_speedup = round(gguf.decode_tok_per_sec / safetensors.decode_tok_per_sec, 4)
|
||||
artifact_bytes_ratio = round(gguf.artifact_bytes / max(1, safetensors.artifact_bytes), 4)
|
||||
return {
|
||||
"safetensors_lane": safetensors_lane.id,
|
||||
"gguf_lane": gguf_lane.id,
|
||||
"decode_speedup": decode_speedup,
|
||||
"ttft_speedup": round(safetensors.ttft_ms / max(0.001, gguf.ttft_ms), 4),
|
||||
"artifact_bytes_ratio": artifact_bytes_ratio,
|
||||
"memory_metric": memory_metric,
|
||||
"memory_bytes_ratio": round(
|
||||
getattr(gguf, memory_metric) / max(1, getattr(safetensors, memory_metric)), 4
|
||||
),
|
||||
"output_drift": _output_drift(gguf.output_tokens, safetensors.output_tokens),
|
||||
"gguf_benefit": decode_speedup >= 1.10 or artifact_bytes_ratio <= 0.5,
|
||||
}
|
||||
|
||||
|
||||
def run_performance_benchmark(
|
||||
contract: PerformanceContract = DEFAULT_CONTRACT,
|
||||
backend: StubLaneBackend | None = None,
|
||||
) -> dict:
|
||||
"""Run every contract lane through a backend and compare GGUF to safetensors."""
|
||||
backend = backend if backend is not None else StubLaneBackend(contract)
|
||||
lanes = [(lane, backend.run(lane)) for lane in contract.benchmark_lanes]
|
||||
references = {
|
||||
lane.device: sample.output_tokens for lane, sample in lanes if lane.runtime == "transformers"
|
||||
}
|
||||
lane_reports = []
|
||||
for lane, sample in lanes:
|
||||
drift = _output_drift(sample.output_tokens, references.get(lane.device, sample.output_tokens))
|
||||
lane_reports.append({
|
||||
**lane.to_dict(),
|
||||
"output_tokens": list(sample.output_tokens),
|
||||
"results": [
|
||||
{"concurrency": level, "metrics": _metrics_for(sample, level, drift)}
|
||||
for level in lane.concurrency_levels
|
||||
],
|
||||
})
|
||||
devices = sorted({lane.device for lane, _ in lanes})
|
||||
comparisons = {device: _compare_device(lanes, device) for device in devices}
|
||||
gguf_benefit = any(comparison["gguf_benefit"] for comparison in comparisons.values())
|
||||
return {
|
||||
"schema_version": BENCHMARK_SCHEMA_VERSION,
|
||||
"story_id": contract.story_id,
|
||||
"source": getattr(backend, "source", "custom-backend"),
|
||||
"model_target": contract.model_target.to_dict(),
|
||||
"lanes": lane_reports,
|
||||
"comparisons": comparisons,
|
||||
"stop_condition": {
|
||||
"text": contract.stop_condition,
|
||||
"gguf_benefit": gguf_benefit,
|
||||
"triggered": not gguf_benefit,
|
||||
},
|
||||
}
|
||||
|
||||
|
||||
def run_real_model_endpoint_benchmark(
|
||||
endpoints: Mapping[str, str],
|
||||
*,
|
||||
model: str,
|
||||
contract: PerformanceContract = DEFAULT_CONTRACT,
|
||||
timeout: float = 120.0,
|
||||
) -> dict:
|
||||
"""Run one live OpenAI-compatible request per lane against supplied endpoints.
|
||||
|
||||
The caller provides one URL per benchmark lane. The runner measures the
|
||||
request/response round-trip at the client boundary and reuses the same
|
||||
contract schema as the deterministic stub.
|
||||
"""
|
||||
|
||||
def _sample_for_lane(lane: BenchmarkLane, endpoint: str) -> LaneSample:
|
||||
prompt = " ".join(contract.model_target.rationale.split()[:6])
|
||||
body = json.dumps(
|
||||
{
|
||||
"model": model,
|
||||
"messages": [{"role": "user", "content": prompt}],
|
||||
"max_tokens": len(STUB_OUTPUT_TOKENS),
|
||||
"temperature": 0,
|
||||
}
|
||||
).encode("utf-8")
|
||||
request = urllib.request.Request(
|
||||
f"{endpoint.rstrip('/')}/v1/chat/completions",
|
||||
data=body,
|
||||
headers={
|
||||
"Content-Type": "application/json",
|
||||
"X-Meshnet-Lane": lane.id,
|
||||
},
|
||||
method="POST",
|
||||
)
|
||||
started = time.monotonic()
|
||||
with urllib.request.urlopen(request, timeout=timeout) as response:
|
||||
response_body = response.read()
|
||||
session_id = response.headers.get("X-Meshnet-Session", f"{lane.id}-session")
|
||||
elapsed_ms = round((time.monotonic() - started) * 1000, 4)
|
||||
payload = json.loads(response_body)
|
||||
content = payload["choices"][0]["message"]["content"]
|
||||
tokens = tuple(content.split())
|
||||
token_count = max(1, len(tokens))
|
||||
artifact_gb = (
|
||||
contract.model_target.gguf_size_gb
|
||||
if lane.runtime == "llama.cpp"
|
||||
else _SAFETENSORS_BF16_ARTIFACT_GB
|
||||
)
|
||||
return LaneSample(
|
||||
ttft_ms=elapsed_ms,
|
||||
prefill_tok_per_sec=round(token_count / max(0.001, elapsed_ms / 1000), 4),
|
||||
decode_tok_per_sec=round(token_count / max(0.001, elapsed_ms / 1000), 4),
|
||||
rss_bytes=0,
|
||||
vram_bytes=0,
|
||||
artifact_bytes=_gb(artifact_gb),
|
||||
output_tokens=tokens,
|
||||
)
|
||||
|
||||
lanes = []
|
||||
for lane in contract.benchmark_lanes:
|
||||
if lane.id not in endpoints:
|
||||
raise KeyError(f"missing endpoint for lane {lane.id}")
|
||||
lanes.append((lane, _sample_for_lane(lane, endpoints[lane.id])))
|
||||
references = {
|
||||
lane.device: sample.output_tokens for lane, sample in lanes if lane.runtime == "transformers"
|
||||
}
|
||||
lane_reports = []
|
||||
for lane, sample in lanes:
|
||||
drift = _output_drift(sample.output_tokens, references.get(lane.device, sample.output_tokens))
|
||||
lane_reports.append({
|
||||
**lane.to_dict(),
|
||||
"output_tokens": list(sample.output_tokens),
|
||||
"results": [
|
||||
{"concurrency": level, "metrics": _metrics_for(sample, level, drift)}
|
||||
for level in lane.concurrency_levels
|
||||
],
|
||||
})
|
||||
devices = sorted({lane.device for lane, _ in lanes})
|
||||
comparisons = {device: _compare_device(lanes, device) for device in devices}
|
||||
gguf_benefit = any(comparison["gguf_benefit"] for comparison in comparisons.values())
|
||||
return {
|
||||
"schema_version": BENCHMARK_SCHEMA_VERSION,
|
||||
"story_id": contract.story_id,
|
||||
"source": "real-model-endpoints",
|
||||
"model_target": contract.model_target.to_dict(),
|
||||
"lanes": lane_reports,
|
||||
"comparisons": comparisons,
|
||||
"stop_condition": {
|
||||
"text": contract.stop_condition,
|
||||
"gguf_benefit": gguf_benefit,
|
||||
"triggered": not gguf_benefit,
|
||||
},
|
||||
}
|
||||
|
||||
|
||||
def _parse_lane_endpoints(pairs: list[str], parser: argparse.ArgumentParser) -> dict[str, str]:
|
||||
endpoints: dict[str, str] = {}
|
||||
for pair in pairs:
|
||||
lane_id, sep, url = pair.partition("=")
|
||||
if not sep or not lane_id or not url:
|
||||
parser.error(f"--live-endpoint expects LANE_ID=URL, got {pair!r}")
|
||||
endpoints[lane_id] = url
|
||||
return endpoints
|
||||
|
||||
|
||||
def _write_report(report: dict, path: Path) -> None:
|
||||
path.parent.mkdir(parents=True, exist_ok=True)
|
||||
path.write_text(json.dumps(report, indent=2, sort_keys=True) + "\n", encoding="utf-8")
|
||||
|
||||
|
||||
def main(argv: list[str] | None = None) -> int:
|
||||
parser = argparse.ArgumentParser(description="Write the DGR-001 performance contract JSON")
|
||||
parser.add_argument("--json-out", type=Path, default=DEFAULT_OUTPUT_PATH, help="output JSON path")
|
||||
parser.add_argument(
|
||||
"--benchmark-out",
|
||||
type=Path,
|
||||
default=None,
|
||||
help="also run the deterministic stub benchmark and write its JSON report here",
|
||||
)
|
||||
parser.add_argument(
|
||||
"--live-endpoint",
|
||||
action="append",
|
||||
default=None,
|
||||
metavar="LANE_ID=URL",
|
||||
help="lane-to-endpoint mapping for the live benchmark; repeat once per contract lane",
|
||||
)
|
||||
parser.add_argument(
|
||||
"--live-model",
|
||||
default=None,
|
||||
help="model name sent to live endpoints (default: contract safetensors repo)",
|
||||
)
|
||||
parser.add_argument(
|
||||
"--live-benchmark-out",
|
||||
type=Path,
|
||||
default=None,
|
||||
help="run the live endpoint benchmark against --live-endpoint lanes and write its JSON report here",
|
||||
)
|
||||
args = parser.parse_args(argv)
|
||||
if args.live_endpoint and args.live_benchmark_out is None:
|
||||
parser.error("--live-endpoint requires --live-benchmark-out")
|
||||
if args.live_benchmark_out is not None and not args.live_endpoint:
|
||||
parser.error("--live-benchmark-out requires at least one --live-endpoint")
|
||||
contract = build_default_contract()
|
||||
path = contract.write_json(args.json_out)
|
||||
print(path)
|
||||
if args.benchmark_out is not None:
|
||||
_write_report(run_performance_benchmark(contract), args.benchmark_out)
|
||||
print(args.benchmark_out)
|
||||
if args.live_endpoint:
|
||||
report = run_real_model_endpoint_benchmark(
|
||||
_parse_lane_endpoints(args.live_endpoint, parser),
|
||||
model=args.live_model or contract.model_target.safetensors_repo,
|
||||
contract=contract,
|
||||
)
|
||||
_write_report(report, args.live_benchmark_out)
|
||||
print(args.live_benchmark_out)
|
||||
return 0
|
||||
|
||||
|
||||
if __name__ == "__main__": # pragma: no cover - CLI entry point
|
||||
raise SystemExit(main())
|
||||
@@ -12,7 +12,7 @@ import urllib.error
|
||||
import urllib.parse
|
||||
import urllib.request
|
||||
from pathlib import Path
|
||||
from typing import Any
|
||||
from typing import Any, Callable
|
||||
|
||||
from .admission import (
|
||||
AdmissionRequirement,
|
||||
@@ -419,6 +419,7 @@ def _start_heartbeat(
|
||||
interval: float = _HEARTBEAT_INTERVAL_IDLE,
|
||||
node_ref: Any | None = None,
|
||||
start_time: float | None = None,
|
||||
refresh_capability: Callable[[dict], dict | None] | None = None,
|
||||
) -> threading.Thread:
|
||||
"""Daemon thread: sends heartbeats and re-registers automatically after tracker restarts.
|
||||
|
||||
@@ -430,6 +431,7 @@ def _start_heartbeat(
|
||||
which is logged for now (hot-reload implemented in US-026).
|
||||
"""
|
||||
_start_time = start_time or time.monotonic()
|
||||
completed_directives: list[dict] = []
|
||||
|
||||
def _current_requests_snapshot() -> list[dict]:
|
||||
if node_ref is None:
|
||||
@@ -454,6 +456,8 @@ def _start_heartbeat(
|
||||
current_requests = _current_requests_snapshot()
|
||||
if current_requests:
|
||||
stats["current_requests"] = current_requests
|
||||
if completed_directives:
|
||||
stats["completed_directives"] = list(completed_directives)
|
||||
return stats
|
||||
|
||||
def _sleep_interval() -> float:
|
||||
@@ -461,9 +465,26 @@ def _start_heartbeat(
|
||||
return _HEARTBEAT_INTERVAL_BUSY
|
||||
return interval
|
||||
|
||||
def _refresh_proof(payload: dict) -> None:
|
||||
"""Re-prove the current shard so a re-registration never presents an aged proof.
|
||||
|
||||
The tracker refuses proofs older than its freshness budget: re-sending the
|
||||
startup-time report after an outage would re-register the node unroutable.
|
||||
"""
|
||||
if refresh_capability is None or "capability_report" not in payload:
|
||||
return
|
||||
try:
|
||||
fresh = refresh_capability(payload)
|
||||
except Exception as exc:
|
||||
print(f" [node] WARNING: capability re-validation failed: {exc}", flush=True)
|
||||
return
|
||||
if fresh:
|
||||
payload["capability_report"] = fresh
|
||||
|
||||
def _reregister() -> bool:
|
||||
nonlocal node_id
|
||||
try:
|
||||
_refresh_proof(register_payload)
|
||||
resp = _post_json(f"{tracker_url}/v1/nodes/register", register_payload)
|
||||
node_id = resp.get("node_id", node_id)
|
||||
if node_ref is not None:
|
||||
@@ -485,6 +506,7 @@ def _start_heartbeat(
|
||||
"managed_assignment": True,
|
||||
}
|
||||
try:
|
||||
_refresh_proof(extra_payload)
|
||||
reg_resp = _post_json(f"{tracker_url}/v1/nodes/register", extra_payload)
|
||||
print(
|
||||
f" [node] registered additional model — node ID: {reg_resp.get('node_id')}",
|
||||
@@ -493,21 +515,26 @@ def _start_heartbeat(
|
||||
except Exception as exc:
|
||||
print(f" [node] WARNING: additional model registration failed: {exc}", flush=True)
|
||||
|
||||
def _apply_directives(directives: list[dict]) -> None:
|
||||
def _apply_directives(directives: list[dict]) -> dict | None:
|
||||
if not directives:
|
||||
return
|
||||
return None
|
||||
if node_ref is None or not hasattr(node_ref, "apply_tracker_directives"):
|
||||
print(f" [node] tracker directives received: {directives}", flush=True)
|
||||
return
|
||||
return None
|
||||
try:
|
||||
applied = node_ref.apply_tracker_directives(directives)
|
||||
except Exception as exc:
|
||||
print(f" [node] WARNING: failed to apply tracker directives: {exc}", flush=True)
|
||||
return
|
||||
return None
|
||||
if applied:
|
||||
completed_directives.append(dict(applied))
|
||||
if applied.get("action") == "ADD_SHARD":
|
||||
_register_additional_assignment(applied)
|
||||
return
|
||||
return applied
|
||||
if applied.get("action") in {"DROP_SHARD", "DROP_ALL_SHARDS"}:
|
||||
# A release has no replacement range. It is not a failed
|
||||
# heartbeat and must not re-register the released assignment.
|
||||
return applied
|
||||
model_id = applied.get("model", register_payload.get("hf_repo") or register_payload.get("model"))
|
||||
register_payload["model"] = str(model_id).split("/")[-1]
|
||||
register_payload["hf_repo"] = model_id
|
||||
@@ -515,6 +542,7 @@ def _start_heartbeat(
|
||||
register_payload["shard_end"] = applied["shard_end"]
|
||||
register_payload["quantization"] = applied.get("quantization", register_payload.get("quantization"))
|
||||
register_payload["tracker_mode"] = bool(applied.get("tracker_mode", False))
|
||||
return applied
|
||||
|
||||
def _loop() -> None:
|
||||
nonlocal node_id
|
||||
@@ -542,7 +570,10 @@ def _start_heartbeat(
|
||||
continue
|
||||
|
||||
try:
|
||||
resp = _post_json(hb_url, _get_stats())
|
||||
heartbeat = _get_stats()
|
||||
resp = _post_json(hb_url, heartbeat)
|
||||
if heartbeat.get("completed_directives"):
|
||||
completed_directives.clear()
|
||||
_apply_directives(resp.get("directives", []))
|
||||
new_asgn = resp.get("new_assignment")
|
||||
if new_asgn:
|
||||
@@ -579,6 +610,7 @@ def _register_with_tracker(
|
||||
reg_payload: dict,
|
||||
node: Any,
|
||||
start_time: float,
|
||||
refresh_capability: Callable[[dict], dict | None] | None = None,
|
||||
) -> str | None:
|
||||
"""Register with the tracker, or start background retries when it is unreachable."""
|
||||
try:
|
||||
@@ -586,7 +618,14 @@ def _register_with_tracker(
|
||||
tracker_node_id = str(reg_resp.get("node_id") or "?")
|
||||
setattr(node, "tracker_node_id", tracker_node_id)
|
||||
print(f" Registered with tracker — node ID: {tracker_node_id}", flush=True)
|
||||
_start_heartbeat(tracker_url, tracker_node_id, reg_payload, node_ref=node, start_time=start_time)
|
||||
_start_heartbeat(
|
||||
tracker_url,
|
||||
tracker_node_id,
|
||||
reg_payload,
|
||||
node_ref=node,
|
||||
start_time=start_time,
|
||||
refresh_capability=refresh_capability,
|
||||
)
|
||||
return tracker_node_id
|
||||
except Exception as exc:
|
||||
setattr(node, "tracker_node_id", None)
|
||||
@@ -598,6 +637,7 @@ def _register_with_tracker(
|
||||
reg_payload,
|
||||
node_ref=node,
|
||||
start_time=start_time,
|
||||
refresh_capability=refresh_capability,
|
||||
)
|
||||
return None
|
||||
|
||||
@@ -718,6 +758,54 @@ def _admit_capability(
|
||||
return report
|
||||
|
||||
|
||||
def _capability_refresher(
|
||||
node: Any,
|
||||
*,
|
||||
manifest: RecipeManifest,
|
||||
recipe: Recipe,
|
||||
detected_device: str,
|
||||
cache_dir: Path | None,
|
||||
force_cpu: bool,
|
||||
validator: CapabilityValidator | None = None,
|
||||
) -> Callable[[dict], dict | None]:
|
||||
"""A fresh proof for what the node serves *now*, run at re-registration time.
|
||||
|
||||
The startup proof ages past the tracker's freshness budget, and directives
|
||||
can move the node to a shard the startup proof never covered — so every
|
||||
re-registration re-proves against the currently loaded backend rather than
|
||||
replaying the report captured at boot.
|
||||
"""
|
||||
def refresh(payload: dict) -> dict | None:
|
||||
target_model = payload.get("hf_repo") or payload.get("model")
|
||||
backend = None
|
||||
accessor = getattr(node, "backend_for", None)
|
||||
if callable(accessor) and target_model:
|
||||
backend = accessor(str(target_model))
|
||||
if backend is None:
|
||||
backend = getattr(node, "backend", None)
|
||||
if backend is None:
|
||||
return None
|
||||
context = CapabilityContext(
|
||||
backend=backend,
|
||||
selection=DoctorSelection(
|
||||
model_id=str(getattr(backend, "model_id", target_model)),
|
||||
shard_start=int(getattr(backend, "shard_start", 0) or 0),
|
||||
shard_end=int(getattr(backend, "shard_end", 0) or 0),
|
||||
quantization=str(getattr(backend, "quantization", None) or "auto"),
|
||||
cache_dir=cache_dir,
|
||||
force_cpu=force_cpu,
|
||||
),
|
||||
recipe=recipe,
|
||||
manifest=manifest,
|
||||
device=_capability_device(backend, detected_device),
|
||||
)
|
||||
report = (validator or probe_capability)(context)
|
||||
setattr(node, "capability_report", report)
|
||||
return report.to_dict()
|
||||
|
||||
return refresh
|
||||
|
||||
|
||||
def run_startup(
|
||||
tracker_url: str,
|
||||
port: int = 0,
|
||||
@@ -1026,6 +1114,15 @@ def run_startup(
|
||||
}
|
||||
tracker_node_id = _register_with_tracker(
|
||||
tracker_url, reg_payload, node, _node_start_time,
|
||||
refresh_capability=_capability_refresher(
|
||||
node,
|
||||
manifest=manifest,
|
||||
recipe=recipe,
|
||||
detected_device=device,
|
||||
cache_dir=cache_dir,
|
||||
force_cpu=force_cpu,
|
||||
validator=capability_validator,
|
||||
),
|
||||
)
|
||||
|
||||
print(
|
||||
@@ -1197,6 +1294,15 @@ def run_startup(
|
||||
}
|
||||
tracker_node_id = _register_with_tracker(
|
||||
tracker_url, auto_reg_payload, node, _node_start_time,
|
||||
refresh_capability=_capability_refresher(
|
||||
node,
|
||||
manifest=manifest,
|
||||
recipe=recipe,
|
||||
detected_device=device,
|
||||
cache_dir=cache_dir,
|
||||
force_cpu=force_cpu,
|
||||
validator=capability_validator,
|
||||
),
|
||||
)
|
||||
shard_label = _format_shard_label(
|
||||
assigned_shard_start,
|
||||
@@ -1389,6 +1495,15 @@ def run_startup(
|
||||
}
|
||||
tracker_node_id = _register_with_tracker(
|
||||
tracker_url, reg_payload, node, _node_start_time,
|
||||
refresh_capability=_capability_refresher(
|
||||
node,
|
||||
manifest=manifest,
|
||||
recipe=recipe,
|
||||
detected_device=device,
|
||||
cache_dir=cache_dir,
|
||||
force_cpu=force_cpu,
|
||||
validator=capability_validator,
|
||||
),
|
||||
)
|
||||
print(
|
||||
f"\n{'=' * 32}\n"
|
||||
@@ -1474,7 +1589,22 @@ def run_startup(
|
||||
)
|
||||
node_id = str(reg_resp["node_id"])
|
||||
setattr(node, "tracker_node_id", node_id)
|
||||
_start_heartbeat(tracker_url, node_id, reg_payload, node_ref=node, start_time=_node_start_time)
|
||||
_start_heartbeat(
|
||||
tracker_url,
|
||||
node_id,
|
||||
reg_payload,
|
||||
node_ref=node,
|
||||
start_time=_node_start_time,
|
||||
refresh_capability=_capability_refresher(
|
||||
node,
|
||||
manifest=manifest,
|
||||
recipe=recipe,
|
||||
detected_device=device,
|
||||
cache_dir=shard_path,
|
||||
force_cpu=force_cpu,
|
||||
validator=capability_validator,
|
||||
),
|
||||
)
|
||||
except Exception:
|
||||
node.stop()
|
||||
raise
|
||||
|
||||
@@ -1543,8 +1543,32 @@ class TorchNodeServer:
|
||||
def loaded_model_ids(self) -> list[str]:
|
||||
return list(self._backends.keys())
|
||||
|
||||
def backend_for(self, model_id: str) -> TorchModelShard | None:
|
||||
"""The loaded backend serving `model_id` — full repo id or short name."""
|
||||
backend = self._backends.get(model_id)
|
||||
if backend is not None:
|
||||
return backend
|
||||
short = model_id.split("/")[-1].lower()
|
||||
for key, candidate in self._backends.items():
|
||||
if key.split("/")[-1].lower() == short:
|
||||
return candidate
|
||||
return None
|
||||
|
||||
def apply_tracker_directives(self, directives: list[dict]) -> dict | None:
|
||||
"""Apply tracker shard directives (LOAD_SHARD replace, ADD_SHARD load-more)."""
|
||||
drop_all_directive = next(
|
||||
(directive for directive in reversed(directives) if directive.get("action") == "DROP_ALL_SHARDS"),
|
||||
None,
|
||||
)
|
||||
if drop_all_directive is not None:
|
||||
self._backends.clear()
|
||||
self._backend = None
|
||||
self._tracker_mode = False
|
||||
if self._server is not None:
|
||||
self._server.backends = {}
|
||||
self._server.backend = None
|
||||
self._server.tracker_mode = False
|
||||
return {"action": "DROP_ALL_SHARDS"}
|
||||
drop_directive = next(
|
||||
(directive for directive in reversed(directives) if directive.get("action") == "DROP_SHARD"),
|
||||
None,
|
||||
|
||||
@@ -44,12 +44,15 @@
|
||||
.empty { color:var(--dim); font-style:italic; }
|
||||
.pill { display:inline-block; padding:0 7px; border-radius:9px;
|
||||
border:1px solid var(--border); font-size:11px; }
|
||||
input, button { font:inherit; color:var(--fg); background:var(--bg);
|
||||
input, button, select { font:inherit; color:var(--fg); background:var(--bg);
|
||||
border:1px solid var(--border); border-radius:6px; padding:5px 8px; }
|
||||
input { width:100%; margin-bottom:6px; }
|
||||
button { cursor:pointer; color:var(--accent); }
|
||||
button:hover { border-color:var(--accent); }
|
||||
button.small { font-size:11px; padding:1px 7px; }
|
||||
dialog { color:var(--fg); background:var(--panel); border:1px solid var(--border); border-radius:8px; min-width:min(420px,calc(100vw - 32px)); }
|
||||
dialog::backdrop { background:rgba(0,0,0,.55); }
|
||||
.placement-dialog-actions { display:flex; justify-content:flex-end; gap:8px; margin-top:12px; }
|
||||
.form-row { display:flex; gap:8px; }
|
||||
.form-row button { white-space:nowrap; }
|
||||
.error-msg { color:var(--bad); font-size:12px; min-height:16px; }
|
||||
@@ -212,7 +215,7 @@
|
||||
.chat-compose button:disabled { opacity:.45; cursor:not-allowed; }
|
||||
.console {
|
||||
background:var(--bg); border:1px solid var(--border); border-radius:6px;
|
||||
min-height:160px; max-height:280px; overflow:auto; padding:7px 9px;
|
||||
min-height:160px; max-height:520px; overflow-y:auto; overflow-x:auto; padding:7px 9px;
|
||||
white-space:pre-wrap; word-break:break-word; font-size:11px;
|
||||
}
|
||||
.console-line { padding:1px 0; border-bottom:1px solid #161b22; }
|
||||
@@ -296,6 +299,7 @@
|
||||
<section data-tab="billing" data-admin-only><h2>Settlement history</h2><div id="settlements" class="empty">admin login required</div></section>
|
||||
<section data-tab="admin"><h2>Tracker hive</h2><div id="hive" class="empty">loading…</div></section>
|
||||
<section data-tab="admin" class="wide"><h2>Model placement</h2><div id="admin-model-placement-status" class="dim">Choose a model to load or release.</div><div id="admin-model-placement" class="empty">admin login required</div></section>
|
||||
<section data-tab="admin" class="wide"><h2>Total node pool</h2><div id="admin-node-pool" class="empty">admin login required</div></section>
|
||||
<section data-tab="admin" id="admin-section"><h2>All accounts (admin)</h2><div id="admin" class="empty"></div></section>
|
||||
<section data-tab="admin" data-admin-only><h2>Strikes / bans / forfeitures</h2><div id="fraud" class="empty">admin login required</div></section>
|
||||
<section data-tab="admin"><h2>Client balances</h2><div id="clients" class="empty">admin login required</div></section>
|
||||
@@ -323,6 +327,16 @@
|
||||
<div id="testing-log" class="console empty">no test output yet</div>
|
||||
</section>
|
||||
</main>
|
||||
<dialog id="model-placement-dialog">
|
||||
<form method="dialog">
|
||||
<div id="model-placement-dialog-title"></div>
|
||||
<label for="model-placement-node">Node</label>
|
||||
<select id="model-placement-node"></select>
|
||||
<label id="model-placement-replace" style="display:none"><input type="checkbox" id="model-placement-replace-confirm"> Unload the currently loaded model before loading this one</label>
|
||||
<div id="model-placement-replace-error" class="bad" style="display:none"></div>
|
||||
<div class="placement-dialog-actions"><button value="cancel">Cancel</button><button type="button" id="model-placement-confirm">Confirm</button></div>
|
||||
</form>
|
||||
</dialog>
|
||||
<script>
|
||||
"use strict";
|
||||
const $ = id => document.getElementById(id);
|
||||
@@ -1082,6 +1096,7 @@ function renderBillingUsage(records) {
|
||||
}
|
||||
|
||||
let consoleClearedAt = 0;
|
||||
const CONSOLE_MAX_LINES = 1000;
|
||||
|
||||
function clearConsole() {
|
||||
consoleClearedAt = Date.now() / 1000;
|
||||
@@ -1095,7 +1110,7 @@ function renderConsole(data) {
|
||||
$("console").innerHTML = '<div class="empty">no console events</div>';
|
||||
return;
|
||||
}
|
||||
$("console").innerHTML = events.slice(-120).map(e => {
|
||||
$("console").innerHTML = events.slice(-CONSOLE_MAX_LINES).map(e => {
|
||||
const level = String(e.level || "info");
|
||||
const cls = level === "error" ? "console-level-error" : level === "warn" ? "console-level-warn" : "console-level-info";
|
||||
const fields = e.fields && Object.keys(e.fields).length ? " " + JSON.stringify(e.fields) : "";
|
||||
@@ -1789,7 +1804,7 @@ async function requestSelectedModelLoad() {
|
||||
if (!selectedChatModel) return;
|
||||
const button = $("request-model-load");
|
||||
if (button) button.disabled = true;
|
||||
const result = await apiCall("/v1/models/load", "POST", { model: selectedChatModel });
|
||||
const result = await apiCall("/v1/models/load", "POST", { model: selectedChatModel, force: isAdmin });
|
||||
if (button) button.disabled = false;
|
||||
if (!result.ok) {
|
||||
alert(result.data.error || "model load request failed");
|
||||
@@ -1799,27 +1814,73 @@ async function requestSelectedModelLoad() {
|
||||
$("chat-status").textContent = `load queued on ${short(assignment.node_id || "node")} for layers ${assignment.shard_start}-${assignment.shard_end}`;
|
||||
}
|
||||
|
||||
async function requestAdminModelLoad(model) {
|
||||
const result = await apiCall("/v1/models/load", "POST", { model, force: true });
|
||||
async function requestAdminModelLoad(model, nodeId, replacing) {
|
||||
const result = await apiCall("/v1/models/load", "POST", { model, node_id: nodeId, force: replacing });
|
||||
if (!result.ok) return showAdminModelPlacementStatus(result.data.error || "model load request failed", true);
|
||||
const assignment = result.data.assignment || {};
|
||||
showAdminModelPlacementStatus(`Load queued on ${short(assignment.node_id || "node")} for ${model}.`);
|
||||
await refreshActiveTab(true);
|
||||
}
|
||||
|
||||
async function releaseAdminModel(model) {
|
||||
const result = await apiCall("/v1/models/release", "POST", { model });
|
||||
async function releaseAdminModel(model, nodeId) {
|
||||
const result = await apiCall("/v1/models/release", "POST", { model, node_id: nodeId });
|
||||
if (!result.ok) return showAdminModelPlacementStatus(result.data.error || "model release request failed", true);
|
||||
showAdminModelPlacementStatus(`Release queued for ${result.data.released || 0} node(s) serving ${model}.`);
|
||||
await refreshActiveTab(true);
|
||||
}
|
||||
|
||||
async function releaseAllNodeModels(nodeId) {
|
||||
if (!confirm("Unload every model from this node?")) return;
|
||||
const result = await apiCall("/v1/nodes/release-all", "POST", { node_id: nodeId });
|
||||
if (!result.ok) return showAdminModelPlacementStatus(result.data.error || "node unload failed", true);
|
||||
showAdminModelPlacementStatus(`Unload queued for ${short(nodeId)}.`);
|
||||
await refreshActiveTab(true);
|
||||
}
|
||||
|
||||
function showAdminModelPlacementStatus(message, isError) {
|
||||
const status = $("admin-model-placement-status");
|
||||
status.textContent = message;
|
||||
status.className = isError ? "bad" : "ok";
|
||||
}
|
||||
|
||||
function gib(bytes) { return bytes == null ? "not reported" : `${(Number(bytes) / 1073741824).toFixed(1)} GiB`; }
|
||||
|
||||
function renderAdminNodePool(map) {
|
||||
const groups = {};
|
||||
for (const node of (map && map.nodes) || []) {
|
||||
const account = node.wallet_address || "unbound account";
|
||||
(groups[account] = groups[account] || []).push(node);
|
||||
}
|
||||
let html = "";
|
||||
for (const [account, nodes] of Object.entries(groups).sort(([a], [b]) => a.localeCompare(b))) {
|
||||
html += `<div style="margin-top:10px"><b>${esc(short(account, 20))}</b> <span class="dim">${nodes.length} node(s)</span></div>`;
|
||||
html += table(["node", "assignment", "state / slots", "model RAM", "RAM", "GPU / VRAM", "model drive", "action"], nodes.map(node => {
|
||||
const hw = node.hardware_profile || {};
|
||||
const cap = node.capacity || {};
|
||||
// The network map keeps reported resource capacity under `capacity`.
|
||||
node.ram_bytes = cap.ram_bytes ?? node.ram_bytes;
|
||||
node.vram_bytes = cap.vram_bytes ?? node.vram_bytes;
|
||||
const disk = hw.model_drive_free_bytes ?? hw.model_path_free_bytes ?? hw.disk_free_bytes;
|
||||
const gpu = hw.gpu_name || (hw.cuda_available ? "CUDA GPU" : "CPU only");
|
||||
const row = [nodeDisplayCell(node), esc(node.hf_repo || node.model || "unassigned"),
|
||||
esc(`${node.stats?.status || "?"} · ${cap.loaded_slots ?? "?"}/${cap.max_loaded_shards ?? node.max_loaded_shards ?? "?"} slots`),
|
||||
esc(gib(cap.loaded_model_bytes)),
|
||||
esc(gib(node.ram_bytes || (hw.ram_mb && hw.ram_mb * 1048576))),
|
||||
esc(`${gpu} · ${gib(node.vram_bytes || (hw.vram_mb && hw.vram_mb * 1048576))}`), esc(gib(disk))];
|
||||
return row.concat([
|
||||
node.shard_start == null ? '<span class="dim">empty</span>' :
|
||||
`<button class="small" data-admin-node-release="${esc(node.node_id)}">unload all</button>`,
|
||||
]);
|
||||
}));
|
||||
}
|
||||
$("admin-node-pool").innerHTML = html || '<div class="empty">no nodes registered</div>';
|
||||
}
|
||||
|
||||
$("admin-node-pool").addEventListener("click", event => {
|
||||
const unload = event.target.closest("[data-admin-node-release]");
|
||||
if (unload) void releaseAllNodeModels(unload.dataset.adminNodeRelease);
|
||||
});
|
||||
|
||||
function renderAdminModelPlacement(models, map) {
|
||||
const nodes = (map && map.nodes) || [];
|
||||
const rows = ((models && models.data) || []).map(model => {
|
||||
@@ -1827,7 +1888,8 @@ function renderAdminModelPlacement(models, map) {
|
||||
const serving = nodes.filter(node => aliases.has(node.model) || aliases.has(node.hf_repo)).length;
|
||||
const downloaded = nodes.filter(node => aliases.has(node.model) || aliases.has(node.hf_repo) ||
|
||||
(node.downloaded_models || []).some(item => aliases.has(item.model) || aliases.has(item.hf_repo))).length;
|
||||
const actions = `<button class="small" data-admin-model-load="${esc(model.id)}">load</button> ` +
|
||||
const loadable = model.id !== "stub-model";
|
||||
const actions = `<button class="small" data-admin-model-load="${esc(model.id)}"${loadable ? "" : " disabled"}>load</button> ` +
|
||||
`<button class="small" data-admin-model-release="${esc(model.id)}"${serving ? "" : " disabled"}>release</button>`;
|
||||
return [esc(model.name || model.id), String(serving), String(downloaded), actions];
|
||||
});
|
||||
@@ -1839,10 +1901,49 @@ function renderAdminModelPlacement(models, map) {
|
||||
$("admin-model-placement").addEventListener("click", event => {
|
||||
const load = event.target.closest("[data-admin-model-load]");
|
||||
const release = event.target.closest("[data-admin-model-release]");
|
||||
if (load) void requestAdminModelLoad(load.dataset.adminModelLoad);
|
||||
if (release) void releaseAdminModel(release.dataset.adminModelRelease);
|
||||
if (load) void chooseModelPlacementNode("load", load.dataset.adminModelLoad);
|
||||
if (release) void chooseModelPlacementNode("release", release.dataset.adminModelRelease);
|
||||
});
|
||||
|
||||
function chooseModelPlacementNode(action, model) {
|
||||
const dialog = $("model-placement-dialog");
|
||||
const select = $("model-placement-node");
|
||||
const targetAlias = modelAliasKey(model);
|
||||
const nodes = (lastNetworkMap?.nodes || []).filter(node => action === "load" ||
|
||||
modelAliasKey(node.model) === targetAlias || modelAliasKey(node.hf_repo) === targetAlias);
|
||||
if (!nodes.length) return showAdminModelPlacementStatus(`No node can ${action} ${model}.`, true);
|
||||
$("model-placement-dialog-title").textContent = `${action === "load" ? "Load" : "Release"} ${model} on a node`;
|
||||
select.innerHTML = nodes.map(node => `<option value="${esc(node.node_id)}">${esc(short(node.friendly_name || node.node_id, 20))} — ${esc(node.hf_repo || node.model || "unassigned")}</option>`).join("");
|
||||
const replace = $("model-placement-replace");
|
||||
const replaceConfirm = $("model-placement-replace-confirm");
|
||||
const replaceError = $("model-placement-replace-error");
|
||||
const confirmButton = $("model-placement-confirm");
|
||||
const selectedNode = () => nodes.find(node => node.node_id === select.value);
|
||||
const updateReplacementWarning = () => {
|
||||
const node = selectedNode();
|
||||
const occupied = action === "load" && node && node.shard_start != null && node.shard_end != null &&
|
||||
modelAliasKey(node.hf_repo || node.model) !== targetAlias;
|
||||
replace.style.display = occupied ? "" : "none";
|
||||
replaceConfirm.checked = false;
|
||||
replaceError.style.display = "none";
|
||||
};
|
||||
select.onchange = updateReplacementWarning;
|
||||
updateReplacementWarning();
|
||||
dialog.onclose = null;
|
||||
confirmButton.onclick = () => {
|
||||
const replacing = replace.style.display !== "none";
|
||||
if (replacing && !replaceConfirm.checked) {
|
||||
replaceError.textContent = "Tick the box to confirm that this will unload the current model.";
|
||||
replaceError.style.display = "";
|
||||
return;
|
||||
}
|
||||
dialog.close("confirm");
|
||||
if (action === "load") void requestAdminModelLoad(model, select.value, replacing);
|
||||
else void releaseAdminModel(model, select.value);
|
||||
};
|
||||
dialog.showModal();
|
||||
}
|
||||
|
||||
function chatAuthToken() {
|
||||
if (accountApiKeys.length) return accountApiKeys[0];
|
||||
return null;
|
||||
@@ -2485,11 +2586,13 @@ async function fetchAdminTab() {
|
||||
if (isAdmin) fetches.push(apiCall("/v1/admin/accounts"));
|
||||
const results = await Promise.all(fetches);
|
||||
const [raft, consoleData, summary, wallets, models, map, adminResp] = results;
|
||||
if (map) lastNetworkMap = map;
|
||||
renderIfChanged("hive", raft, renderHive);
|
||||
renderIfChanged("console", consoleData, renderConsole);
|
||||
renderIfChanged("billing-summary", summary, data => renderBilling(data));
|
||||
renderIfChanged("fraud", { wallets, summary }, data => renderFraud(data.wallets, data.summary));
|
||||
renderIfChanged("admin-model-placement", { models, map }, data => renderAdminModelPlacement(data.models, data.map));
|
||||
renderIfChanged("admin-node-pool", map, renderAdminNodePool);
|
||||
if (adminResp && adminResp.ok) {
|
||||
renderIfChanged("admin", adminResp.data.accounts || [], accounts => {
|
||||
const rows = accounts.map(a => {
|
||||
|
||||
@@ -86,7 +86,7 @@ from .model_files import files_for_layer_range, snapshot_dir_for_repo
|
||||
from .raft import RaftNode
|
||||
|
||||
|
||||
_CONSOLE_LIMIT = 300
|
||||
_CONSOLE_LIMIT = 1000
|
||||
_PROXY_PROGRESS_LOG_INTERVAL = 5.0
|
||||
_SESSION_COOKIE_NAME = "meshnet_session"
|
||||
|
||||
@@ -1101,12 +1101,15 @@ def _registration_quantization(body: dict, quantizations: list[str]) -> str | No
|
||||
|
||||
An absent field predates the protocol adding it: it means "unknown", not
|
||||
"unsupported", so the node keeps the best precision it advertises and stays
|
||||
routable. Anything the node states explicitly is taken at its word -- a null,
|
||||
a non-string, or an unsupported name leaves it with no usable precision and
|
||||
routing excludes it.
|
||||
routable. An explicit "auto" means the same thing — the node's CLI default
|
||||
delegates the choice, it does not refuse one. Anything else the node states
|
||||
explicitly is taken at its word -- a null, a non-string, or an unsupported
|
||||
name leaves it with no usable precision and routing excludes it.
|
||||
"""
|
||||
if "quantization" in body:
|
||||
return _normalize_quantization(body["quantization"])
|
||||
declared = body.get("quantization")
|
||||
declared_auto = isinstance(declared, str) and declared.strip().lower() == "auto"
|
||||
if "quantization" in body and not declared_auto:
|
||||
return _normalize_quantization(declared)
|
||||
supported = [
|
||||
normalized for value in quantizations
|
||||
if (normalized := _normalize_quantization(value)) is not None
|
||||
@@ -1225,6 +1228,7 @@ def _node_capacity_summary(node: _NodeEntry, preset: dict | None = None) -> dict
|
||||
"quantization": node.quantization,
|
||||
"benchmark_tokens_per_sec": node.benchmark_tokens_per_sec,
|
||||
"effective_throughput": round(_effective_throughput(node), 4),
|
||||
"loaded_model_bytes": _assignment_memory_bytes(node, preset),
|
||||
}
|
||||
if preset is not None:
|
||||
summary["max_assignable_layers"] = _node_layer_capacity(node, preset)
|
||||
@@ -1494,7 +1498,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 +1516,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)
|
||||
@@ -1528,21 +1536,26 @@ def _request_model_load_locked(server: "_TrackerHTTPServer", model_key: str) ->
|
||||
return None
|
||||
|
||||
|
||||
def _force_model_load_locked(server: "_TrackerHTTPServer", model_key: str) -> dict | None:
|
||||
def _force_model_load_locked(
|
||||
server: "_TrackerHTTPServer", model_key: str, node_id: str | None = None,
|
||||
) -> dict | None:
|
||||
"""Replace the fastest ready assignment after an explicit admin eviction."""
|
||||
resolved_name, preset = _resolve_model_preset(server.model_presets, model_key)
|
||||
if preset is None or not preset.get("hf_repo"):
|
||||
return None
|
||||
start, end = _preset_layer_bounds(preset)
|
||||
candidates = [node for node in server.registry.values()
|
||||
if node.status == "ready" and node.pending_new_assignment is None
|
||||
and _has_usable_quantization(node)]
|
||||
# An explicit admin eviction is permitted to recover a stuck/loading node
|
||||
# and to use the preset default precision. It must only avoid a node that
|
||||
# already has another assignment in flight.
|
||||
candidates = [
|
||||
node for node in server.registry.values()
|
||||
if node.pending_new_assignment is None
|
||||
and (node_id is None or node.node_id == node_id)
|
||||
]
|
||||
if not candidates:
|
||||
return None
|
||||
node = max(candidates, key=lambda item: item.benchmark_tokens_per_sec)
|
||||
shard_end = min(end, start + min(_node_layer_capacity(node, preset), end - start + 1) - 1)
|
||||
if shard_end < start:
|
||||
return None
|
||||
shard_end = min(end, start + max(1, min(_node_layer_capacity(node, preset), end - start + 1)) - 1)
|
||||
quantization = _node_quantization(node, preset)
|
||||
directive = _load_directive(node, str(preset["hf_repo"]), start, shard_end, quantization)
|
||||
replaced = node.hf_repo or node.model
|
||||
@@ -1556,13 +1569,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"))
|
||||
@@ -1571,6 +1588,16 @@ def _release_model_locked(server: "_TrackerHTTPServer", model_key: str) -> int:
|
||||
return released
|
||||
|
||||
|
||||
def _release_all_node_models_locked(server: "_TrackerHTTPServer", node_id: str) -> int:
|
||||
"""Queue removal of every loaded assignment on one node."""
|
||||
node = server.registry.get(node_id)
|
||||
if node is None or node.shard_start is None or node.shard_end is None:
|
||||
return 0
|
||||
node.pending_directives.append({"action": "DROP_ALL_SHARDS"})
|
||||
node.status = "loading"
|
||||
return 1
|
||||
|
||||
|
||||
def _preferred_node_quantization(
|
||||
node: _NodeEntry,
|
||||
preset: dict,
|
||||
@@ -3089,6 +3116,9 @@ class _TrackerHandler(http.server.BaseHTTPRequestHandler):
|
||||
if self.path == "/v1/models/release":
|
||||
self._handle_model_release_request()
|
||||
return
|
||||
if self.path == "/v1/nodes/release-all":
|
||||
self._handle_node_release_all_request()
|
||||
return
|
||||
if self.path == "/v1/models/vote":
|
||||
self._handle_model_coverage_vote()
|
||||
return
|
||||
@@ -3265,7 +3295,12 @@ class _TrackerHandler(http.server.BaseHTTPRequestHandler):
|
||||
node.hf_repo or node.model
|
||||
for node in alive
|
||||
if node.model is not None
|
||||
and node.model not in server.model_presets
|
||||
# The same model can be registered under its HF repository while
|
||||
# the catalogue exposes its short preset id. Do not emit a second
|
||||
# repo-keyed entry when either node identifier resolves to a preset.
|
||||
and _resolve_model_preset(
|
||||
server.model_presets, node.hf_repo or node.model,
|
||||
)[1] is None
|
||||
and node.shard_start is not None
|
||||
and node.shard_end is not None
|
||||
and node.num_layers is not None
|
||||
@@ -3366,6 +3401,11 @@ class _TrackerHandler(http.server.BaseHTTPRequestHandler):
|
||||
"endpoint": node.endpoint,
|
||||
"relay_addr": node.relay_addr,
|
||||
"peer_id": node.peer_id,
|
||||
"wallet_address": node.wallet_address,
|
||||
"hardware_profile": dict(node.hardware_profile),
|
||||
"ram_bytes": node.ram_bytes,
|
||||
"vram_bytes": node.vram_bytes,
|
||||
"max_loaded_shards": node.max_loaded_shards,
|
||||
}
|
||||
for node in tracker_nodes
|
||||
],
|
||||
@@ -3389,12 +3429,7 @@ class _TrackerHandler(http.server.BaseHTTPRequestHandler):
|
||||
memory_pool = _memory_pool_map(server)
|
||||
|
||||
def capacity_for(node: _NodeEntry) -> dict:
|
||||
preset = None
|
||||
if node.model:
|
||||
preset = server.model_presets.get(node.model)
|
||||
if preset is None and node.hf_repo and node.num_layers:
|
||||
preset = _hf_rebalance_preset([node])
|
||||
return _node_capacity_summary(node, preset)
|
||||
return _node_capacity_summary(node, _preset_for_node(server, node))
|
||||
|
||||
def throughput_for(node: _NodeEntry) -> dict:
|
||||
if server.stats is None:
|
||||
@@ -4804,6 +4839,20 @@ class _TrackerHandler(http.server.BaseHTTPRequestHandler):
|
||||
entry.uptime_seconds = float(body["uptime_seconds"])
|
||||
if "status" in body and body["status"] in ("ready", "loading"):
|
||||
entry.status = body["status"]
|
||||
completed_directives = body.get("completed_directives", [])
|
||||
if isinstance(completed_directives, list):
|
||||
for directive in completed_directives:
|
||||
if not isinstance(directive, dict) or directive.get("action") not in {"DROP_SHARD", "DROP_ALL_SHARDS"}:
|
||||
continue
|
||||
# A node has confirmed the release. Stop advertising its
|
||||
# old route immediately so the dashboard and routing state
|
||||
# agree with the runtime.
|
||||
entry.model = "stub-model"
|
||||
entry.hf_repo = None
|
||||
entry.shard_start = None
|
||||
entry.shard_end = None
|
||||
entry.tracker_mode = False
|
||||
entry.status = "ready"
|
||||
if "friendly_name" in body:
|
||||
try:
|
||||
entry.friendly_name = _normalize_friendly_name(body.get("friendly_name"))
|
||||
@@ -4875,11 +4924,19 @@ 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
|
||||
_resolved_name, preset = _resolve_model_preset(server.model_presets, model)
|
||||
if preset is None or str(preset.get("hf_repo") or "").strip().lower() == "stub-model":
|
||||
self._send_json(400, {"error": "stub-model is a local test backend and cannot be loaded onto a node"})
|
||||
return
|
||||
with server.lock:
|
||||
self._purge_expired_nodes()
|
||||
assignment = _request_model_load_locked(server, model)
|
||||
assignment = _request_model_load_locked(server, model, node_id)
|
||||
if assignment is None and body.get("force") is True:
|
||||
assignment = _force_model_load_locked(server, model)
|
||||
assignment = _force_model_load_locked(server, model, node_id)
|
||||
if assignment is None:
|
||||
self._send_json(409, {"error": "no ready joined node has an available model slot and sufficient capacity"})
|
||||
return
|
||||
@@ -4896,14 +4953,39 @@ 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
|
||||
self._send_json(202, {"status": "release_queued", "released": released})
|
||||
|
||||
def _handle_node_release_all_request(self):
|
||||
server: _TrackerHTTPServer = self.server # type: ignore[assignment]
|
||||
if not self._require_role("admin", "validator"):
|
||||
return
|
||||
body = self._read_json_body()
|
||||
if body is None:
|
||||
return
|
||||
node_id = body.get("node_id")
|
||||
if 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_all_node_models_locked(server, node_id)
|
||||
if not released:
|
||||
self._send_json(404, {"error": "no loaded models found for node"})
|
||||
return
|
||||
self._send_json(202, {
|
||||
"status": "release_queued", "released": released, "node_id": node_id,
|
||||
})
|
||||
|
||||
def _handle_model_coverage_vote(self):
|
||||
"""Record a rolling wish-list signal for an unavailable precision."""
|
||||
server: _TrackerHTTPServer = self.server # type: ignore[assignment]
|
||||
|
||||
@@ -44,6 +44,9 @@ def test_dashboard_served_with_all_panels():
|
||||
assert ".wide { grid-column:span 2; }" in html
|
||||
assert 'onclick="clearConsole()"' in html
|
||||
assert "let consoleClearedAt = 0;" in html
|
||||
assert "max-height:520px; overflow-y:auto; overflow-x:auto;" in html
|
||||
assert "const CONSOLE_MAX_LINES = 1000;" in html
|
||||
assert "events.slice(-CONSOLE_MAX_LINES)" in html
|
||||
finally:
|
||||
tracker.stop()
|
||||
|
||||
@@ -114,6 +117,25 @@ def test_dashboard_exposes_admin_model_inventory_and_release_controls():
|
||||
assert 'data-admin-model-load=' in html
|
||||
assert 'data-admin-model-release=' in html
|
||||
assert "admin-model-placement-status" in html
|
||||
assert 'id="admin-node-pool"' in html
|
||||
assert "renderAdminNodePool" in html
|
||||
assert "model drive" in html
|
||||
# RAM and VRAM live in the network-map capacity object, not at node top level.
|
||||
assert "node.ram_bytes = cap.ram_bytes" in html
|
||||
assert "node.vram_bytes = cap.vram_bytes" in html
|
||||
assert 'id="model-placement-dialog"' in html
|
||||
assert "chooseModelPlacementNode" in html
|
||||
assert "node_id: nodeId" in html
|
||||
assert "modelAliasKey(node.model)" in html
|
||||
assert 'id="model-placement-replace"' in html
|
||||
assert 'id="model-placement-confirm"' in html
|
||||
assert 'id="model-placement-replace-error"' in html
|
||||
assert "force: replacing" in html
|
||||
assert "Tick the box to confirm" in html
|
||||
assert "releaseAllNodeModels" in html
|
||||
assert '"/v1/nodes/release-all"' in html
|
||||
assert "model RAM" in html
|
||||
assert "loaded_model_bytes" in html
|
||||
|
||||
|
||||
def test_network_map_includes_node_friendly_name():
|
||||
|
||||
@@ -370,11 +370,20 @@ def test_admin_can_replace_a_served_model_and_release_it():
|
||||
headers = {"Content-Type": "application/json", "Authorization": "Bearer test-admin"}
|
||||
load = urllib.request.Request(
|
||||
f"http://127.0.0.1:{port}/v1/models/load",
|
||||
data=json.dumps({"model": "qwen2.5-0.5b-instruct", "force": True}).encode(),
|
||||
data=json.dumps({
|
||||
"model": "qwen2.5-0.5b-instruct",
|
||||
"node_id": node["node_id"],
|
||||
"force": True,
|
||||
}).encode(),
|
||||
headers=headers, method="POST")
|
||||
with urllib.request.urlopen(load) as response:
|
||||
assert json.loads(response.read())["assignment"]["node_id"] == node["node_id"]
|
||||
heartbeat = _post_json(f"http://127.0.0.1:{port}/v1/nodes/{node['node_id']}/heartbeat", {})
|
||||
_post_json(
|
||||
f"http://127.0.0.1:{port}/v1/nodes/{node['node_id']}/heartbeat",
|
||||
{"completed_directives": [{"action": "DROP_SHARD", "model": "Qwen/Qwen2.5-0.5B-Instruct"}]},
|
||||
)
|
||||
network = _get_json(f"http://127.0.0.1:{port}/v1/network/map")
|
||||
assert heartbeat["directives"][0]["action"] == "LOAD_SHARD"
|
||||
release = urllib.request.Request(
|
||||
f"http://127.0.0.1:{port}/v1/models/release",
|
||||
@@ -386,6 +395,33 @@ def test_admin_can_replace_a_served_model_and_release_it():
|
||||
tracker.stop()
|
||||
|
||||
assert heartbeat["directives"][0]["action"] == "DROP_SHARD"
|
||||
released_node = next(item for item in network["nodes"] if item["node_id"] == node["node_id"])
|
||||
assert released_node["shard_start"] is None
|
||||
assert released_node["shard_end"] is None
|
||||
|
||||
|
||||
def test_models_list_does_not_duplicate_a_preset_registered_by_hf_repo():
|
||||
"""A preset and its canonical repository are one selectable model."""
|
||||
tracker = TrackerServer(enable_billing=False)
|
||||
port = tracker.start()
|
||||
try:
|
||||
_post_json(
|
||||
f"http://127.0.0.1:{port}/v1/nodes/register",
|
||||
{
|
||||
"endpoint": "http://127.0.0.1:9913",
|
||||
"model": "Qwen2.5-0.5B-Instruct",
|
||||
"hf_repo": "Qwen/Qwen2.5-0.5B-Instruct",
|
||||
"num_layers": 24,
|
||||
"shard_start": 0,
|
||||
"shard_end": 23,
|
||||
},
|
||||
)
|
||||
models = _get_json(f"http://127.0.0.1:{port}/v1/models")["data"]
|
||||
finally:
|
||||
tracker.stop()
|
||||
|
||||
assert [model["id"] for model in models].count("qwen2.5-0.5b-instruct") == 1
|
||||
assert not any(model["id"] == "Qwen/Qwen2.5-0.5B-Instruct" for model in models)
|
||||
|
||||
|
||||
def test_endpoint_key_distinguishes_same_port_different_hosts():
|
||||
|
||||
@@ -358,6 +358,73 @@ def test_a_stale_report_cannot_be_reused_to_register(startup_env):
|
||||
assert startup_env == []
|
||||
|
||||
|
||||
# ---------------------------------------------------------------------------
|
||||
# Re-registration: the proof presented is fresh, never the one captured at boot
|
||||
# ---------------------------------------------------------------------------
|
||||
|
||||
|
||||
def test_run_startup_hands_the_heartbeat_a_refresher_for_the_current_shard(startup_env, monkeypatch):
|
||||
"The tracker refuses aged proofs, so the heartbeat must be able to re-prove what the node serves now.\n\nTags: node, admission, startup"
|
||||
import meshnet_node.startup as startup_mod
|
||||
|
||||
captured: dict = {}
|
||||
monkeypatch.setattr(
|
||||
startup_mod, "_start_heartbeat", lambda *a, **kw: captured.update(kw)
|
||||
)
|
||||
|
||||
_start(capability_validator=capability_stub())
|
||||
|
||||
refresh = captured.get("refresh_capability")
|
||||
assert callable(refresh), "run_startup no longer wires a capability refresher"
|
||||
fresh = refresh({"hf_repo": MODEL, "model": MODEL.split("/")[-1]})
|
||||
assert fresh is not None
|
||||
assert fresh["model"]["model_id"] == MODEL
|
||||
assert (fresh["shard"]["start"], fresh["shard"]["end"]) == (0, 23)
|
||||
assert fresh["validated_at"] > time.time() - 60
|
||||
|
||||
|
||||
def test_a_reregistration_presents_a_refreshed_proof(monkeypatch):
|
||||
"Replaying the boot-time report after an outage re-registers the node unroutable; the re-register path must present a fresh proof.\n\nTags: node, admission, startup"
|
||||
import json
|
||||
|
||||
import meshnet_node.startup as startup_mod
|
||||
|
||||
model = "acme/refresh-model-7b"
|
||||
boot_report = {"validated_at": 1.0, "marker": "boot"}
|
||||
fresh_report = {"validated_at": 2.0, "marker": "fresh"}
|
||||
posted: list[dict] = []
|
||||
|
||||
def _record(url, payload, timeout=10.0):
|
||||
if url.endswith("/v1/nodes/register") and payload.get("hf_repo") == model:
|
||||
posted.append(json.loads(json.dumps(payload)))
|
||||
return {"node_id": "node-refresh"}
|
||||
raise SystemExit # first heartbeat POST ends the daemon loop
|
||||
|
||||
monkeypatch.setattr(startup_mod, "_post_json", _record)
|
||||
|
||||
payload = {
|
||||
"hf_repo": model,
|
||||
"model": model.split("/")[-1],
|
||||
"capability_report": dict(boot_report),
|
||||
}
|
||||
thread = startup_mod._start_heartbeat(
|
||||
"http://tracker.invalid",
|
||||
startup_mod._PENDING_NODE_ID, # forces a re-registration on the first tick
|
||||
payload,
|
||||
interval=0.02,
|
||||
refresh_capability=lambda _payload: dict(fresh_report),
|
||||
)
|
||||
|
||||
# The loop must be dead before this test returns: once monkeypatch restores
|
||||
# `_post_json`, a surviving thread would re-register through whatever the
|
||||
# *next* test patches in and corrupt its call counts.
|
||||
thread.join(timeout=5.0)
|
||||
|
||||
assert not thread.is_alive(), "the heartbeat loop outlived the test"
|
||||
assert posted, "the heartbeat never re-registered"
|
||||
assert posted[0]["capability_report"] == fresh_report
|
||||
|
||||
|
||||
def test_a_matching_passing_report_registers_and_travels_with_the_payload(startup_env):
|
||||
"Registration carries the proof for exactly the model/shard/recipe it advertises.\n\nTags: node, admission, startup"
|
||||
node = _start() # production validator against a working fake backend
|
||||
|
||||
@@ -287,6 +287,47 @@ def test_configure_torch_threads_applies_explicit_settings(monkeypatch):
|
||||
assert active == {"torch_threads": 12, "torch_interop_threads": 2}
|
||||
|
||||
|
||||
def test_heartbeat_applies_release_without_reregistering(monkeypatch):
|
||||
"""DROP_SHARD has no replacement range and must not look like an outage."""
|
||||
import meshnet_node.startup as startup_mod
|
||||
|
||||
released = threading.Event()
|
||||
requests: list[tuple[str, dict]] = []
|
||||
|
||||
class FakeNode:
|
||||
def apply_tracker_directives(self, directives):
|
||||
assert directives == [{"action": "DROP_SHARD", "model": "Qwen/Qwen2.5-0.5B-Instruct"}]
|
||||
return {"action": "DROP_SHARD", "model": "Qwen/Qwen2.5-0.5B-Instruct"}
|
||||
|
||||
def fake_post(url, payload, timeout=10.0):
|
||||
requests.append((url, dict(payload)))
|
||||
released.set()
|
||||
return {"directives": [{"action": "DROP_SHARD", "model": "Qwen/Qwen2.5-0.5B-Instruct"}]}
|
||||
|
||||
sleep_calls = 0
|
||||
|
||||
def one_heartbeat(_seconds):
|
||||
nonlocal sleep_calls
|
||||
sleep_calls += 1
|
||||
if sleep_calls > 1:
|
||||
raise SystemExit
|
||||
|
||||
monkeypatch.setattr(startup_mod, "_post_json", fake_post)
|
||||
monkeypatch.setattr(startup_mod.time, "sleep", one_heartbeat)
|
||||
payload = {
|
||||
"model": "Qwen2.5-0.5B-Instruct",
|
||||
"hf_repo": "Qwen/Qwen2.5-0.5B-Instruct",
|
||||
"shard_start": 0,
|
||||
"shard_end": 23,
|
||||
}
|
||||
startup_mod._start_heartbeat("http://tracker", "node-1", payload, interval=0, node_ref=FakeNode())
|
||||
|
||||
assert released.wait(1), "heartbeat did not receive the queued release"
|
||||
assert len(requests) == 1, "release must not trigger a re-registration"
|
||||
assert payload["shard_start"] == 0
|
||||
assert payload["shard_end"] == 23
|
||||
|
||||
|
||||
def test_benchmark_throughput_is_registered_in_payload(monkeypatch, tmp_path):
|
||||
"benchmark_tokens_per_sec from the benchmark is included in the tracker registration.\n\nTags: node, performance, startup"
|
||||
import meshnet_node.startup as startup_mod
|
||||
|
||||
@@ -1,286 +0,0 @@
|
||||
"""Tests for the DGR-001 performance contract metadata."""
|
||||
|
||||
from __future__ import annotations
|
||||
|
||||
import json
|
||||
from unittest.mock import MagicMock, patch
|
||||
|
||||
import pytest
|
||||
|
||||
from meshnet_node.performance_contract import (
|
||||
BENCHMARK_SCHEMA_VERSION,
|
||||
DEFAULT_CONTRACT,
|
||||
SCHEMA_VERSION,
|
||||
main,
|
||||
run_performance_benchmark,
|
||||
run_real_model_endpoint_benchmark,
|
||||
)
|
||||
|
||||
|
||||
def test_default_contract_is_architecture_aligned_and_small():
|
||||
"""The baseline stays on DeepSeek2 and uses the smallest DeepSeek-family GGUF.
|
||||
|
||||
Tags: performance, model, gguf
|
||||
"""
|
||||
payload = DEFAULT_CONTRACT.to_dict()
|
||||
|
||||
assert payload["schema_version"] == SCHEMA_VERSION
|
||||
assert payload["story_id"] == "DGR-001"
|
||||
assert payload["model_target"] == {
|
||||
"name": "DeepSeek-V2-Lite-Chat",
|
||||
"architecture": "deepseek2",
|
||||
"safetensors_repo": "deepseek-ai/DeepSeek-V2-Lite-Chat",
|
||||
"safetensors_precision": "bfloat16",
|
||||
"gguf_repo": "second-state/DeepSeek-V2-Lite-Chat-GGUF",
|
||||
"gguf_quant": "Q2_K",
|
||||
"gguf_size_gb": 6.43,
|
||||
"comparison_policy": (
|
||||
"same model/revision, closest practical low-footprint precision pair: "
|
||||
"BF16 safetensors versus Q2_K GGUF"
|
||||
),
|
||||
"rationale": (
|
||||
"Smallest DeepSeek-family benchmark anchor that still points toward "
|
||||
"DeepSeek-V4-Flash; keeps the runtime on the DeepSeek2 path instead "
|
||||
"of falling back to a tiny but architecture-mismatched smoke model."
|
||||
),
|
||||
}
|
||||
assert payload["benchmark_lanes"] == [
|
||||
{
|
||||
"id": "transformers-safetensors-cpu",
|
||||
"runtime": "transformers",
|
||||
"device": "cpu",
|
||||
"recipe": "current safetensors recipe",
|
||||
"concurrency_levels": [1, 4],
|
||||
},
|
||||
{
|
||||
"id": "llama-cpp-gguf-cpu",
|
||||
"runtime": "llama.cpp",
|
||||
"device": "cpu",
|
||||
"recipe": "whole-model GGUF recipe",
|
||||
"concurrency_levels": [1, 4],
|
||||
},
|
||||
{
|
||||
"id": "transformers-safetensors-gpu",
|
||||
"runtime": "transformers",
|
||||
"device": "gpu",
|
||||
"recipe": "current safetensors recipe",
|
||||
"concurrency_levels": [1, 4],
|
||||
},
|
||||
{
|
||||
"id": "llama-cpp-gguf-gpu",
|
||||
"runtime": "llama.cpp",
|
||||
"device": "gpu",
|
||||
"recipe": "whole-model GGUF recipe",
|
||||
"concurrency_levels": [1, 4],
|
||||
},
|
||||
]
|
||||
assert "ttft_ms" in payload["metrics"]
|
||||
assert "output_drift" in payload["metrics"]
|
||||
assert "meaningful speed or fit benefit" in payload["stop_condition"]
|
||||
assert any("mounted drive" in note for note in payload["notes"])
|
||||
|
||||
|
||||
def test_contract_cli_writes_json(tmp_path, capsys):
|
||||
"""The contract can be emitted as a machine-readable artifact.
|
||||
|
||||
Tags: performance, artifact
|
||||
"""
|
||||
output = tmp_path / "performance-contract.json"
|
||||
|
||||
assert main(["--json-out", str(output)]) == 0
|
||||
written = json.loads(output.read_text(encoding="utf-8"))
|
||||
|
||||
assert written == DEFAULT_CONTRACT.to_dict()
|
||||
assert str(output) in capsys.readouterr().out
|
||||
|
||||
|
||||
def test_stub_benchmark_covers_every_lane_concurrency_and_metric():
|
||||
"""The runner exercises all four CPU/GPU lanes with the full metric set.
|
||||
|
||||
Tags: performance, benchmark, gguf
|
||||
"""
|
||||
report = run_performance_benchmark()
|
||||
|
||||
assert report["schema_version"] == BENCHMARK_SCHEMA_VERSION
|
||||
assert report["story_id"] == "DGR-001"
|
||||
assert report["source"] == "stub-backend"
|
||||
assert report["model_target"] == DEFAULT_CONTRACT.model_target.to_dict()
|
||||
assert [lane["id"] for lane in report["lanes"]] == [
|
||||
lane.id for lane in DEFAULT_CONTRACT.benchmark_lanes
|
||||
]
|
||||
for lane in report["lanes"]:
|
||||
assert [result["concurrency"] for result in lane["results"]] == [1, 4]
|
||||
for result in lane["results"]:
|
||||
assert set(result["metrics"]) == set(DEFAULT_CONTRACT.metrics)
|
||||
assert result["metrics"]["failure_count"] == 0
|
||||
assert result["metrics"]["decode_tok_per_sec"] > 0
|
||||
|
||||
|
||||
def test_stub_benchmark_is_deterministic():
|
||||
"""Two runs produce byte-identical reports; no clocks or randomness leak in.
|
||||
|
||||
Tags: performance, benchmark, deterministic
|
||||
"""
|
||||
first = run_performance_benchmark()
|
||||
second = run_performance_benchmark()
|
||||
|
||||
assert first == second
|
||||
assert json.dumps(first, sort_keys=True) == json.dumps(second, sort_keys=True)
|
||||
|
||||
|
||||
def test_stub_benchmark_compares_gguf_against_safetensors_per_device():
|
||||
"""Each device gets a GGUF-vs-safetensors comparison and a stop-condition verdict.
|
||||
|
||||
Tags: performance, benchmark, gguf
|
||||
"""
|
||||
report = run_performance_benchmark()
|
||||
|
||||
assert set(report["comparisons"]) == {"cpu", "gpu"}
|
||||
cpu, gpu = report["comparisons"]["cpu"], report["comparisons"]["gpu"]
|
||||
assert cpu["safetensors_lane"] == "transformers-safetensors-cpu"
|
||||
assert cpu["gguf_lane"] == "llama-cpp-gguf-cpu"
|
||||
assert cpu["memory_metric"] == "rss_bytes"
|
||||
assert gpu["safetensors_lane"] == "transformers-safetensors-gpu"
|
||||
assert gpu["gguf_lane"] == "llama-cpp-gguf-gpu"
|
||||
assert gpu["memory_metric"] == "vram_bytes"
|
||||
for comparison in (cpu, gpu):
|
||||
assert comparison["decode_speedup"] > 1.0
|
||||
assert comparison["artifact_bytes_ratio"] < 0.5
|
||||
assert comparison["memory_bytes_ratio"] < 1.0
|
||||
assert comparison["output_drift"] == 0.0
|
||||
assert comparison["gguf_benefit"] is True
|
||||
assert report["stop_condition"]["gguf_benefit"] is True
|
||||
assert report["stop_condition"]["triggered"] is False
|
||||
assert report["stop_condition"]["text"] == DEFAULT_CONTRACT.stop_condition
|
||||
|
||||
|
||||
def test_contract_cli_writes_benchmark_report(tmp_path, capsys):
|
||||
"""--benchmark-out emits the stub benchmark report next to the contract.
|
||||
|
||||
Tags: performance, benchmark, artifact
|
||||
"""
|
||||
contract_out = tmp_path / "performance-contract.json"
|
||||
benchmark_out = tmp_path / "artifacts" / "stub-benchmark-report.json"
|
||||
|
||||
assert main(["--json-out", str(contract_out), "--benchmark-out", str(benchmark_out)]) == 0
|
||||
report = json.loads(benchmark_out.read_text(encoding="utf-8"))
|
||||
|
||||
assert report == run_performance_benchmark()
|
||||
output = capsys.readouterr().out
|
||||
assert str(contract_out) in output
|
||||
assert str(benchmark_out) in output
|
||||
|
||||
|
||||
def test_real_model_endpoint_benchmark_uses_lane_specific_endpoints_and_shared_schema():
|
||||
"""The live client path fans out to one endpoint per CPU/GPU lane.
|
||||
|
||||
Tags: performance, benchmark, live
|
||||
"""
|
||||
response = MagicMock()
|
||||
response.read.return_value = json.dumps({"choices": [{"message": {"content": "mesh activation"}}]}).encode()
|
||||
response.headers.get.return_value = "lane-session"
|
||||
response.__enter__.return_value = response
|
||||
|
||||
endpoints = {
|
||||
"transformers-safetensors-cpu": "http://cpu-safetensors",
|
||||
"llama-cpp-gguf-cpu": "http://cpu-gguf",
|
||||
"transformers-safetensors-gpu": "http://gpu-safetensors",
|
||||
"llama-cpp-gguf-gpu": "http://gpu-gguf",
|
||||
}
|
||||
|
||||
with patch("meshnet_node.performance_contract.urllib.request.urlopen", return_value=response) as urlopen:
|
||||
report = run_real_model_endpoint_benchmark(endpoints=endpoints, model="deepseek-ai/DeepSeek-V2-Lite-Chat")
|
||||
|
||||
assert report["source"] == "real-model-endpoints"
|
||||
assert report["model_target"] == DEFAULT_CONTRACT.model_target.to_dict()
|
||||
assert set(report["comparisons"]) == {"cpu", "gpu"}
|
||||
assert urlopen.call_count == len(endpoints)
|
||||
called_urls = [call.args[0].full_url for call in urlopen.call_args_list]
|
||||
assert called_urls == [f"{url}/v1/chat/completions" for url in endpoints.values()]
|
||||
for lane in report["lanes"]:
|
||||
assert lane["results"][0]["metrics"]["decode_tok_per_sec"] > 0
|
||||
assert lane["results"][0]["metrics"]["ttft_ms"] > 0
|
||||
assert lane["output_tokens"] == ["mesh", "activation"]
|
||||
assert report["comparisons"]["cpu"]["gguf_lane"] == "llama-cpp-gguf-cpu"
|
||||
assert report["comparisons"]["gpu"]["gguf_lane"] == "llama-cpp-gguf-gpu"
|
||||
|
||||
|
||||
def test_contract_cli_runs_live_endpoint_benchmark(tmp_path, capsys):
|
||||
"""--live-endpoint mappings drive the live runner and write its report.
|
||||
|
||||
Tags: performance, benchmark, live, artifact
|
||||
"""
|
||||
contract_out = tmp_path / "performance-contract.json"
|
||||
live_out = tmp_path / "artifacts" / "live-benchmark-report.json"
|
||||
endpoints = {
|
||||
"transformers-safetensors-cpu": "http://cpu-safetensors",
|
||||
"llama-cpp-gguf-cpu": "http://cpu-gguf",
|
||||
"transformers-safetensors-gpu": "http://gpu-safetensors",
|
||||
"llama-cpp-gguf-gpu": "http://gpu-gguf",
|
||||
}
|
||||
fake_report = {"schema_version": BENCHMARK_SCHEMA_VERSION, "source": "real-model-endpoints"}
|
||||
argv = ["--json-out", str(contract_out), "--live-benchmark-out", str(live_out)]
|
||||
for lane_id, url in endpoints.items():
|
||||
argv += ["--live-endpoint", f"{lane_id}={url}"]
|
||||
|
||||
with patch(
|
||||
"meshnet_node.performance_contract.run_real_model_endpoint_benchmark",
|
||||
return_value=fake_report,
|
||||
) as runner:
|
||||
assert main(argv) == 0
|
||||
|
||||
runner.assert_called_once_with(
|
||||
endpoints,
|
||||
model=DEFAULT_CONTRACT.model_target.safetensors_repo,
|
||||
contract=DEFAULT_CONTRACT,
|
||||
)
|
||||
assert json.loads(live_out.read_text(encoding="utf-8")) == fake_report
|
||||
output = capsys.readouterr().out
|
||||
assert str(contract_out) in output
|
||||
assert str(live_out) in output
|
||||
|
||||
|
||||
def test_contract_cli_passes_explicit_live_model(tmp_path):
|
||||
"""--live-model overrides the contract's safetensors repo default.
|
||||
|
||||
Tags: performance, benchmark, live
|
||||
"""
|
||||
live_out = tmp_path / "live-benchmark-report.json"
|
||||
argv = [
|
||||
"--json-out", str(tmp_path / "performance-contract.json"),
|
||||
"--live-benchmark-out", str(live_out),
|
||||
"--live-endpoint", "transformers-safetensors-cpu=http://cpu-safetensors",
|
||||
"--live-model", "local/DeepSeek-V2-Lite-Chat-Q2_K",
|
||||
]
|
||||
|
||||
with patch(
|
||||
"meshnet_node.performance_contract.run_real_model_endpoint_benchmark",
|
||||
return_value={},
|
||||
) as runner:
|
||||
assert main(argv) == 0
|
||||
|
||||
assert runner.call_args.kwargs["model"] == "local/DeepSeek-V2-Lite-Chat-Q2_K"
|
||||
|
||||
|
||||
@pytest.mark.parametrize(
|
||||
"argv",
|
||||
[
|
||||
["--live-endpoint", "transformers-safetensors-cpu=http://cpu"],
|
||||
["--live-benchmark-out", "live-report.json"],
|
||||
[
|
||||
"--live-endpoint", "not-a-mapping",
|
||||
"--live-benchmark-out", "live-report.json",
|
||||
],
|
||||
],
|
||||
ids=["endpoint-without-out", "out-without-endpoint", "malformed-mapping"],
|
||||
)
|
||||
def test_contract_cli_rejects_incomplete_live_arguments(tmp_path, argv, capsys):
|
||||
"""Live flags must arrive as a consistent LANE_ID=URL + output-path set.
|
||||
|
||||
Tags: performance, benchmark, live, cli
|
||||
"""
|
||||
with pytest.raises(SystemExit) as excinfo:
|
||||
main(["--json-out", str(tmp_path / "performance-contract.json"), *argv])
|
||||
|
||||
assert excinfo.value.code == 2
|
||||
assert "--live-" in capsys.readouterr().err
|
||||
@@ -2869,6 +2869,43 @@ def test_same_endpoint_can_register_multiple_models():
|
||||
tracker.stop()
|
||||
|
||||
|
||||
def test_explicit_model_placement_targets_only_the_selected_node():
|
||||
"An admin can add and release a model on one chosen multi-model node.\n\nTags: http, routing, tracker"
|
||||
from meshnet_tracker.server import _release_model_locked, _request_model_load_locked
|
||||
|
||||
tracker = _tracker(model_presets={
|
||||
"model-a": {"total_layers": 4, "bytes_per_layer": {"bfloat16": 1_000}, "hf_repo": "org/ModelA"},
|
||||
"model-b": {"total_layers": 4, "bytes_per_layer": {"bfloat16": 1_000}, "hf_repo": "org/ModelB"},
|
||||
})
|
||||
tracker_port = tracker.start()
|
||||
try:
|
||||
registrations = []
|
||||
for port in (9062, 9063):
|
||||
registrations.append(_post_json(
|
||||
f"http://127.0.0.1:{tracker_port}/v1/nodes/register",
|
||||
{"endpoint": f"http://127.0.0.1:{port}", "model": "model-a", "hf_repo": "org/ModelA",
|
||||
"num_layers": 4, "shard_start": 0, "shard_end": 3, "max_loaded_shards": 2,
|
||||
"vram_bytes": 20_000, "ram_bytes": 20_000, "quantizations": ["bfloat16"],
|
||||
"benchmark_tokens_per_sec": 1.0, "hardware_profile": {}, "score": 1.0},
|
||||
))
|
||||
selected, other = (item["node_id"] for item in registrations)
|
||||
|
||||
with tracker._lock:
|
||||
assignment = _request_model_load_locked(tracker._server, "model-b", selected) # type: ignore[arg-type]
|
||||
assert assignment is not None
|
||||
assert assignment["node_id"] == selected
|
||||
assert tracker._registry[selected].pending_new_assignment is not None
|
||||
assert tracker._registry[other].pending_new_assignment is None
|
||||
|
||||
with tracker._lock:
|
||||
released = _release_model_locked(tracker._server, "model-a", selected) # type: ignore[arg-type]
|
||||
assert released == 1
|
||||
assert len(tracker._registry[selected].pending_directives) == 2
|
||||
assert tracker._registry[other].pending_directives == []
|
||||
finally:
|
||||
tracker.stop()
|
||||
|
||||
|
||||
def test_scale_demanded_models_queues_add_shard_on_spare_host():
|
||||
"Scale demanded models queues add shard on spare host\n\nTags: http, routing, tracker"
|
||||
tracker = _tracker(model_presets={
|
||||
@@ -3005,6 +3042,13 @@ def test_a_node_declaring_an_unsupported_quantization_is_never_routed():
|
||||
assert status == 503
|
||||
|
||||
|
||||
def test_a_node_declaring_auto_quantization_serves_a_default_precision_request():
|
||||
"'auto' is the CLI default that delegates the choice — it is not a refusal, so the node must resolve to its best advertised precision and route.\n\nTags: http, routing, tracker"
|
||||
status, response = _proxy_chat_status(POLICY_COMPAT, quantization="auto")
|
||||
assert status == 200
|
||||
assert response["choices"][0]["message"]["content"] == "ok"
|
||||
|
||||
|
||||
def test_a_node_declaring_a_null_quantization_is_never_routed():
|
||||
"An explicit null states 'no usable precision' -- only an absent field is legacy.\n\nTags: http, routing, tracker"
|
||||
node = http.server.HTTPServer(("127.0.0.1", 0), _EchoChatHandler)
|
||||
|
||||
Reference in New Issue
Block a user