From 3edd9393682dee9113e2392ddd3c8865a4b29d23 Mon Sep 17 00:00:00 2001 From: Prakhar Verma Date: Fri, 8 May 2026 00:28:39 -0400 Subject: [PATCH 1/5] Add live worker controls --- gateway/app/dashboard.py | 46 +++++++++++++++++++++++++++++++ gateway/app/main.py | 39 +++++++++++++++++++++++++-- gateway/app/schemas.py | 5 ++++ tests/test_worker_control.py | 28 +++++++++++++++++++ worker/app/main.py | 52 ++++++++++++++++++++++++++++++------ 5 files changed, 160 insertions(+), 10 deletions(-) create mode 100644 tests/test_worker_control.py diff --git a/gateway/app/dashboard.py b/gateway/app/dashboard.py index 47db5d1..aa3c2fa 100644 --- a/gateway/app/dashboard.py +++ b/gateway/app/dashboard.py @@ -114,6 +114,29 @@ .bad { color: #fca5a5; } + input { + width: 76px; + border: 1px solid #334155; + border-radius: 6px; + background: #020617; + color: #e2e8f0; + padding: 7px 8px; + font: 12px ui-monospace, SFMono-Regular, Menlo, monospace; + } + button { + border: 1px solid #38bdf8; + border-radius: 6px; + background: #082f49; + color: #bae6fd; + padding: 8px 10px; + font: 12px ui-monospace, SFMono-Regular, Menlo, monospace; + cursor: pointer; + } + button.secondary { + border-color: #475569; + background: #111827; + color: #cbd5e1; + } @media (max-width: 760px) { header { display: block; @@ -174,6 +197,9 @@ Requests Failures Latency EMA + Delay + Fail rate + @@ -195,6 +221,20 @@ return `${Math.round(value * 100)}%`; } + async function controlWorker(index, delayMs, failRate) { + await fetch(`/workers/${index}/control`, { + method: "POST", + headers: { "content-type": "application/json" }, + body: JSON.stringify({ + delay_ms: Number(delayMs), + fail_rate: Number(failRate), + }), + }); + await refresh(); + } + + window.controlWorker = controlWorker; + async function refresh() { try { const [health, metrics, workers] = await Promise.all([ @@ -216,6 +256,12 @@ ${worker.requests} ${worker.failures} ${worker.latency_ema_ms} ms + + + + + + `).join(""); } catch (error) { diff --git a/gateway/app/main.py b/gateway/app/main.py index 60f1d0d..53d98f8 100644 --- a/gateway/app/main.py +++ b/gateway/app/main.py @@ -13,7 +13,7 @@ from .dashboard import DASHBOARD_HTML from .metrics import MetricsStore from .router import WorkerPool -from .schemas import GatewayPrediction, PredictRequest +from .schemas import GatewayPrediction, PredictRequest, WorkerControlRequest def env_flag(name: str, default: bool = True) -> bool: @@ -58,7 +58,42 @@ async def dashboard() -> str: @app.get("/workers") async def workers() -> list[dict[str, str | int | float | None]]: - return router.snapshot() + client: httpx.AsyncClient = app.state.client + rows = router.snapshot() + for index, row in enumerate(rows): + row["index"] = index + try: + response = await client.get(f"{row['url']}/config", timeout=0.4) + response.raise_for_status() + config = response.json() + row["worker_id"] = config.get("worker_id") + row["model_version"] = config.get("model_version") + row["delay_ms"] = config.get("delay_ms") + row["fail_rate"] = config.get("fail_rate") + row["control_status"] = "ok" + except httpx.HTTPError: + row["control_status"] = "offline" + return rows + + +@app.post("/workers/{worker_index}/control") +async def control_worker(worker_index: int, update: WorkerControlRequest) -> dict[str, object]: + if worker_index < 0 or worker_index >= len(router.workers): + raise HTTPException(status_code=404, detail="worker not found") + + worker = router.workers[worker_index] + client: httpx.AsyncClient = app.state.client + try: + response = await client.post( + f"{worker.url}/control", + json=update.model_dump(exclude_none=True), + timeout=request_timeout, + ) + response.raise_for_status() + except httpx.HTTPError as exc: + raise HTTPException(status_code=502, detail=f"worker control failed: {exc.__class__.__name__}") from exc + + return {"worker_index": worker_index, "worker_url": worker.url, "config": response.json()} @app.get("/metrics") diff --git a/gateway/app/schemas.py b/gateway/app/schemas.py index c5c8cd2..6d41ce0 100644 --- a/gateway/app/schemas.py +++ b/gateway/app/schemas.py @@ -21,3 +21,8 @@ class GatewayPrediction(BaseModel): worker_url: str attempts: int latency_ms: float + + +class WorkerControlRequest(BaseModel): + delay_ms: int | None = Field(default=None, ge=0, le=5000) + fail_rate: float | None = Field(default=None, ge=0, le=1) diff --git a/tests/test_worker_control.py b/tests/test_worker_control.py new file mode 100644 index 0000000..734bd69 --- /dev/null +++ b/tests/test_worker_control.py @@ -0,0 +1,28 @@ +from fastapi.testclient import TestClient + +from worker.app.main import app, runtime + + +def test_worker_control_updates_runtime() -> None: + previous_delay = runtime.delay_ms + previous_fail_rate = runtime.fail_rate + + try: + with TestClient(app) as client: + response = client.post("/control", json={"delay_ms": 123, "fail_rate": 0.25}) + + assert response.status_code == 200 + assert response.json()["delay_ms"] == 123 + assert response.json()["fail_rate"] == 0.25 + assert runtime.delay_ms == 123 + assert runtime.fail_rate == 0.25 + finally: + runtime.delay_ms = previous_delay + runtime.fail_rate = previous_fail_rate + + +def test_worker_control_rejects_invalid_fail_rate() -> None: + with TestClient(app) as client: + response = client.post("/control", json={"fail_rate": 2}) + + assert response.status_code == 422 diff --git a/worker/app/main.py b/worker/app/main.py index f56a6a6..a95e7b2 100644 --- a/worker/app/main.py +++ b/worker/app/main.py @@ -2,7 +2,8 @@ import os import random -import time +import asyncio +from dataclasses import dataclass from fastapi import FastAPI, HTTPException from pydantic import BaseModel, Field @@ -15,24 +16,59 @@ class PredictRequest(BaseModel): request_id: str | None = None +class ControlUpdate(BaseModel): + delay_ms: int | None = Field(default=None, ge=0, le=5000) + fail_rate: float | None = Field(default=None, ge=0, le=1) + + +@dataclass +class WorkerRuntime: + delay_ms: int + fail_rate: float + + def snapshot(self) -> dict[str, int | float | str]: + return { + "worker_id": worker_id, + "model_version": model.version, + "delay_ms": self.delay_ms, + "fail_rate": self.fail_rate, + } + + app = FastAPI(title="ModelMesh Worker", version="0.1.0") model = load_model(os.getenv("MODEL_ARTIFACT")) worker_id = os.getenv("WORKER_ID", "worker-local") -delay_ms = int(os.getenv("SIMULATE_DELAY_MS", "0")) -fail_rate = float(os.getenv("FAIL_RATE", "0")) +runtime = WorkerRuntime( + delay_ms=int(os.getenv("SIMULATE_DELAY_MS", "0")), + fail_rate=float(os.getenv("FAIL_RATE", "0")), +) @app.get("/health") -async def health() -> dict[str, str]: - return {"status": "ok", "worker_id": worker_id, "model_version": model.version} +async def health() -> dict[str, str | int | float]: + return {"status": "ok", **runtime.snapshot()} + + +@app.get("/config") +async def config() -> dict[str, str | int | float]: + return runtime.snapshot() + + +@app.post("/control") +async def control(update: ControlUpdate) -> dict[str, str | int | float]: + if update.delay_ms is not None: + runtime.delay_ms = update.delay_ms + if update.fail_rate is not None: + runtime.fail_rate = update.fail_rate + return runtime.snapshot() @app.post("/predict") async def predict(payload: PredictRequest) -> dict[str, str | float]: - if delay_ms > 0: - time.sleep(delay_ms / 1000) + if runtime.delay_ms > 0: + await asyncio.sleep(runtime.delay_ms / 1000) - if fail_rate > 0 and random.random() < fail_rate: + if runtime.fail_rate > 0 and random.random() < runtime.fail_rate: raise HTTPException(status_code=503, detail="simulated worker failure") result = model.predict(payload.text) From bcec4c877dea7cd5b32c8ab3bdd284720f2cb9f2 Mon Sep 17 00:00:00 2001 From: Prakhar Verma Date: Fri, 8 May 2026 00:28:43 -0400 Subject: [PATCH 2/5] Record MPS training stats --- training/runs/latest.json | 16 ++ training/runs/latest.md | 15 ++ training/train_text_model.py | 187 ++++++++++++++++++---- worker/app/artifacts/tiny_text_model.json | 88 ++++++---- 4 files changed, 236 insertions(+), 70 deletions(-) create mode 100644 training/runs/latest.json create mode 100644 training/runs/latest.md diff --git a/training/runs/latest.json b/training/runs/latest.json new file mode 100644 index 0000000..a70acc2 --- /dev/null +++ b/training/runs/latest.json @@ -0,0 +1,16 @@ +{ + "model_path": "worker/app/artifacts/tiny_text_model.json", + "device": "mps", + "epochs": 320, + "learning_rate": 0.05, + "samples_per_class": 2048, + "train_examples": 3276, + "test_examples": 820, + "vocab_size": 37, + "parameter_count": 38, + "train_seconds": 0.4351, + "examples_per_second": 2409512.37, + "final_loss": 0.002556, + "train_accuracy": 1.0, + "test_accuracy": 1.0 +} diff --git a/training/runs/latest.md b/training/runs/latest.md new file mode 100644 index 0000000..028266c --- /dev/null +++ b/training/runs/latest.md @@ -0,0 +1,15 @@ +# Training run + +| Metric | Value | +| --- | ---: | +| Device | mps | +| Epochs | 320 | +| Train examples | 3276 | +| Test examples | 820 | +| Vocab size | 37 | +| Parameters | 38 | +| Train time | 0.4351 s | +| Examples/sec | 2409512.37 | +| Final loss | 0.002556 | +| Train accuracy | 1.0 | +| Test accuracy | 1.0 | diff --git a/training/train_text_model.py b/training/train_text_model.py index 2cdbffc..3b08443 100644 --- a/training/train_text_model.py +++ b/training/train_text_model.py @@ -2,7 +2,9 @@ import argparse import json +import random import re +import time from pathlib import Path import torch @@ -11,32 +13,47 @@ ROOT = Path(__file__).resolve().parents[1] OUTPUT_PATH = ROOT / "worker" / "app" / "artifacts" / "tiny_text_model.json" +RUN_PATH = ROOT / "training" / "runs" / "latest.json" +RUN_MD_PATH = ROOT / "training" / "runs" / "latest.md" TOKEN_RE = re.compile(r"[a-z']+") -HEALTHY = [ - "fast stable reliable inference", - "worker is healthy and responsive", - "great latency with stable service", - "cache hit keeps the service fast", - "model route is reliable", - "request completed with healthy worker", +HEALTHY_PATTERNS = [ + ("fast", "stable", "reliable", "inference"), + ("healthy", "responsive", "worker", "service"), + ("great", "latency", "stable", "route"), + ("cache", "hit", "keeps", "service", "fast"), + ("request", "completed", "healthy", "worker"), + ("reliable", "model", "route", "responsive"), ] -RISKY = [ - "slow broken worker timeout", - "request failed with error", - "worker crash caused retry", - "service timeout and broken response", - "model route is slow", - "unhealthy worker returned error", +RISKY_PATTERNS = [ + ("slow", "broken", "worker", "timeout"), + ("request", "failed", "with", "error"), + ("worker", "crash", "caused", "retry"), + ("service", "timeout", "broken", "response"), + ("model", "route", "slow"), + ("unhealthy", "worker", "returned", "error"), +] + +FILLER = [ + "batch", + "gateway", + "prediction", + "queue", + "client", + "traffic", + "burst", + "mesh", ] def parse_args() -> argparse.Namespace: parser = argparse.ArgumentParser() - parser.add_argument("--epochs", type=int, default=250) - parser.add_argument("--lr", type=float, default=0.08) + parser.add_argument("--epochs", type=int, default=320) + parser.add_argument("--lr", type=float, default=0.05) + parser.add_argument("--samples-per-class", type=int, default=2048) parser.add_argument("--output", type=Path, default=OUTPUT_PATH) + parser.add_argument("--run-output", type=Path, default=RUN_PATH) parser.add_argument("--allow-cpu", action="store_true") return parser.parse_args() @@ -53,9 +70,29 @@ def tokens(text: str) -> list[str]: return TOKEN_RE.findall(text.lower()) -def build_dataset() -> tuple[list[str], torch.Tensor, torch.Tensor]: - texts = HEALTHY + RISKY - labels = [1.0] * len(HEALTHY) + [0.0] * len(RISKY) +def build_texts(samples_per_class: int, seed: int = 17) -> tuple[list[str], list[float]]: + rng = random.Random(seed) + texts: list[str] = [] + labels: list[float] = [] + + for label, patterns in [(1.0, HEALTHY_PATTERNS), (0.0, RISKY_PATTERNS)]: + for index in range(samples_per_class): + pattern = list(patterns[index % len(patterns)]) + rng.shuffle(pattern) + if index % 3 == 0: + pattern.append(rng.choice(FILLER)) + if index % 5 == 0: + pattern.insert(0, rng.choice(FILLER)) + texts.append(" ".join(pattern)) + labels.append(label) + + order = list(range(len(texts))) + rng.shuffle(order) + return [texts[index] for index in order], [labels[index] for index in order] + + +def build_dataset(samples_per_class: int) -> tuple[list[str], torch.Tensor, torch.Tensor]: + texts, labels = build_texts(samples_per_class) vocab = sorted({token for text in texts for token in tokens(text)}) index = {token: pos for pos, token in enumerate(vocab)} rows: list[list[float]] = [] @@ -69,34 +106,75 @@ def build_dataset() -> tuple[list[str], torch.Tensor, torch.Tensor]: return vocab, torch.tensor(rows, dtype=torch.float32), torch.tensor(labels, dtype=torch.float32).view(-1, 1) -def train(epochs: int, lr: float, device: torch.device) -> tuple[list[str], nn.Linear, float]: +def split_dataset(x: torch.Tensor, y: torch.Tensor, train_fraction: float = 0.8) -> tuple[torch.Tensor, torch.Tensor, torch.Tensor, torch.Tensor]: + train_count = int(len(x) * train_fraction) + return x[:train_count], y[:train_count], x[train_count:], y[train_count:] + + +def accuracy(model: nn.Linear, x: torch.Tensor, y: torch.Tensor) -> float: + with torch.no_grad(): + predictions = (torch.sigmoid(model(x)) >= 0.5).float() + return float((predictions == y).float().mean().cpu()) + + +def sync_if_needed(device: torch.device) -> None: + if device.type == "mps": + torch.mps.synchronize() + + +def train( + *, + epochs: int, + lr: float, + samples_per_class: int, + device: torch.device, +) -> tuple[list[str], nn.Linear, dict[str, float | int | str]]: torch.manual_seed(17) - vocab, x_cpu, y_cpu = build_dataset() - x = x_cpu.to(device) - y = y_cpu.to(device) + vocab, x_cpu, y_cpu = build_dataset(samples_per_class) + x_train_cpu, y_train_cpu, x_test_cpu, y_test_cpu = split_dataset(x_cpu, y_cpu) + x_train = x_train_cpu.to(device) + y_train = y_train_cpu.to(device) + x_test = x_test_cpu.to(device) + y_test = y_test_cpu.to(device) model = nn.Linear(len(vocab), 1).to(device) optimizer = torch.optim.Adam(model.parameters(), lr=lr) loss_fn = nn.BCEWithLogitsLoss() + sync_if_needed(device) + started = time.perf_counter() + final_loss = 0.0 for _ in range(epochs): optimizer.zero_grad(set_to_none=True) - loss = loss_fn(model(x), y) + loss = loss_fn(model(x_train), y_train) loss.backward() optimizer.step() + final_loss = float(loss.detach().cpu()) + sync_if_needed(device) + train_seconds = time.perf_counter() - started - with torch.no_grad(): - predictions = (torch.sigmoid(model(x)) >= 0.5).float() - accuracy = float((predictions == y).float().mean().cpu()) - - return vocab, model.cpu(), accuracy + stats = { + "device": str(device), + "epochs": epochs, + "learning_rate": lr, + "samples_per_class": samples_per_class, + "train_examples": int(len(x_train_cpu)), + "test_examples": int(len(x_test_cpu)), + "vocab_size": len(vocab), + "parameter_count": sum(parameter.numel() for parameter in model.parameters()), + "train_seconds": round(train_seconds, 4), + "examples_per_second": round((len(x_train_cpu) * epochs) / train_seconds, 2), + "final_loss": round(final_loss, 6), + "train_accuracy": round(accuracy(model, x_train, y_train), 4), + "test_accuracy": round(accuracy(model, x_test, y_test), 4), + } + return vocab, model.cpu(), stats -def write_artifact(path: Path, vocab: list[str], model: nn.Linear, accuracy: float, device: torch.device) -> None: +def write_artifact(path: Path, vocab: list[str], model: nn.Linear, stats: dict[str, float | int | str]) -> None: weights = model.weight.detach().reshape(-1).tolist() artifact = { "model_version": "tiny-text-mps-v1", - "device": str(device), - "train_accuracy": round(accuracy, 4), + "training": stats, "weights": {token: round(float(weights[index]), 6) for index, token in enumerate(vocab)}, "bias": round(float(model.bias.detach()[0]), 6), } @@ -104,12 +182,51 @@ def write_artifact(path: Path, vocab: list[str], model: nn.Linear, accuracy: flo path.write_text(json.dumps(artifact, indent=2) + "\n") +def write_run_files(json_path: Path, md_path: Path, stats: dict[str, float | int | str], model_path: Path) -> None: + try: + display_model_path = str(model_path.relative_to(ROOT)) + except ValueError: + display_model_path = str(model_path) + payload = { + "model_path": display_model_path, + **stats, + } + json_path.parent.mkdir(parents=True, exist_ok=True) + json_path.write_text(json.dumps(payload, indent=2) + "\n") + md_path.write_text( + "\n".join([ + "# Training run", + "", + "| Metric | Value |", + "| --- | ---: |", + f"| Device | {stats['device']} |", + f"| Epochs | {stats['epochs']} |", + f"| Train examples | {stats['train_examples']} |", + f"| Test examples | {stats['test_examples']} |", + f"| Vocab size | {stats['vocab_size']} |", + f"| Parameters | {stats['parameter_count']} |", + f"| Train time | {stats['train_seconds']} s |", + f"| Examples/sec | {stats['examples_per_second']} |", + f"| Final loss | {stats['final_loss']} |", + f"| Train accuracy | {stats['train_accuracy']} |", + f"| Test accuracy | {stats['test_accuracy']} |", + "", + ]) + ) + + def main() -> None: args = parse_args() device = choose_device(args.allow_cpu) - vocab, model, accuracy = train(args.epochs, args.lr, device) - write_artifact(args.output, vocab, model, accuracy, device) - print({"device": str(device), "train_accuracy": round(accuracy, 4), "output": str(args.output)}) + vocab, model, stats = train( + epochs=args.epochs, + lr=args.lr, + samples_per_class=args.samples_per_class, + device=device, + ) + write_artifact(args.output, vocab, model, stats) + write_run_files(args.run_output, args.run_output.with_suffix(".md"), stats, args.output) + print(json.dumps({"output": str(args.output), **stats}, indent=2)) if __name__ == "__main__": diff --git a/worker/app/artifacts/tiny_text_model.json b/worker/app/artifacts/tiny_text_model.json index 98beb48..8091376 100644 --- a/worker/app/artifacts/tiny_text_model.json +++ b/worker/app/artifacts/tiny_text_model.json @@ -1,40 +1,58 @@ { "model_version": "tiny-text-mps-v1", - "device": "mps", - "train_accuracy": 1.0, + "training": { + "device": "mps", + "epochs": 320, + "learning_rate": 0.05, + "samples_per_class": 2048, + "train_examples": 3276, + "test_examples": 820, + "vocab_size": 37, + "parameter_count": 38, + "train_seconds": 0.4351, + "examples_per_second": 2409512.37, + "final_loss": 0.002556, + "train_accuracy": 1.0, + "test_accuracy": 1.0 + }, "weights": { - "and": 0.127082, - "broken": -2.073759, - "cache": 1.618408, - "caused": -2.428945, - "completed": 2.755103, - "crash": -2.278496, - "error": -2.767643, - "failed": -2.994795, - "fast": 1.548141, - "great": 1.78853, - "healthy": 2.40558, - "hit": 1.454436, - "inference": 1.495255, - "is": 1.747386, - "keeps": 1.450163, - "latency": 1.859522, - "model": -0.495443, - "reliable": 4.507694, - "request": -0.840839, - "response": -2.345497, - "responsive": 2.046548, - "retry": -2.43846, - "returned": -2.233388, - "route": -0.48599, - "service": 0.636005, - "slow": -4.552207, - "stable": 1.818228, - "the": 1.385478, - "timeout": -2.225621, - "unhealthy": -2.075621, - "with": 1.362204, - "worker": 1.133086 + "batch": -0.070073, + "broken": -1.928928, + "burst": -0.248643, + "cache": 1.540021, + "caused": -2.06341, + "client": -0.141606, + "completed": 2.38669, + "crash": -1.893442, + "error": -2.172957, + "failed": -2.588806, + "fast": 1.522728, + "gateway": -0.207467, + "great": 1.918606, + "healthy": 2.235116, + "hit": 1.615801, + "inference": 1.602651, + "keeps": 1.49933, + "latency": 1.897305, + "mesh": -0.172917, + "model": -2.128533, + "prediction": 0.031068, + "queue": -0.110654, + "reliable": 3.177283, + "request": 1.983225, + "response": -2.478622, + "responsive": 3.435494, + "retry": -1.880252, + "returned": -1.960394, + "route": 1.494559, + "service": 1.518566, + "slow": -3.264679, + "stable": 1.973113, + "timeout": -2.023214, + "traffic": -0.111536, + "unhealthy": -1.699983, + "with": -2.362207, + "worker": 0.302216 }, - "bias": -0.740176 + "bias": -0.955532 } From faaa5dc4cfd5841a7900aa61d9c7f7bb23b29863 Mon Sep 17 00:00:00 2001 From: Prakhar Verma Date: Fri, 8 May 2026 00:28:47 -0400 Subject: [PATCH 3/5] Add traffic replay demo --- bench/replay_traffic.py | 133 +++++++++++++++++++++ bench/results/replay-latest.json | 191 +++++++++++++++++++++++++++++++ docs/demo.md | 32 ++++++ 3 files changed, 356 insertions(+) create mode 100644 bench/replay_traffic.py create mode 100644 bench/results/replay-latest.json create mode 100644 docs/demo.md diff --git a/bench/replay_traffic.py b/bench/replay_traffic.py new file mode 100644 index 0000000..e51f3b3 --- /dev/null +++ b/bench/replay_traffic.py @@ -0,0 +1,133 @@ +from __future__ import annotations + +import argparse +import asyncio +import json +import time +from pathlib import Path +from typing import Any + +import httpx + + +ROOT = Path(__file__).resolve().parents[1] +RESULT_PATH = ROOT / "bench" / "results" / "replay-latest.json" +TEXTS = [ + "fast stable reliable service", + "cache hit keeps service fast", + "great latency stable route", + "slow broken worker timeout", + "worker crash caused retry", + "request completed healthy worker", +] + + +def parse_args() -> argparse.Namespace: + parser = argparse.ArgumentParser() + parser.add_argument("--base-url", default="http://localhost:8000") + parser.add_argument("--seconds-per-stage", type=float, default=8.0) + parser.add_argument("--rps", type=float, default=18.0) + parser.add_argument("--output", type=Path, default=RESULT_PATH) + return parser.parse_args() + + +async def control_worker(client: httpx.AsyncClient, base_url: str, index: int, *, delay_ms: int, fail_rate: float) -> bool: + try: + response = await client.post( + f"{base_url}/workers/{index}/control", + json={"delay_ms": delay_ms, "fail_rate": fail_rate}, + timeout=1.5, + ) + return response.status_code == 200 + except httpx.HTTPError: + return False + + +async def reset_workers(client: httpx.AsyncClient, base_url: str) -> None: + try: + workers = (await client.get(f"{base_url}/workers", timeout=1.5)).json() + except httpx.HTTPError: + return + + for index, _ in enumerate(workers): + await control_worker(client, base_url, index, delay_ms=0, fail_rate=0) + + +async def snapshot(client: httpx.AsyncClient, base_url: str, stage: str) -> dict[str, Any]: + metrics = (await client.get(f"{base_url}/metrics")).json() + workers = (await client.get(f"{base_url}/workers")).json() + return { + "stage": stage, + "metrics": metrics, + "workers": workers, + } + + +async def send_stage(client: httpx.AsyncClient, base_url: str, stage: str, seconds: float, rps: float) -> dict[str, Any]: + deadline = time.perf_counter() + seconds + interval = 1 / rps + sent = 0 + ok = 0 + started = time.perf_counter() + + while time.perf_counter() < deadline: + text = TEXTS[sent % len(TEXTS)] + if stage in {"slow-worker", "flaky-worker"}: + text = f"{text} {stage} live {sent}" + + try: + response = await client.post( + f"{base_url}/predict", + json={"text": text, "request_id": f"{stage}-{sent}"}, + timeout=3.0, + ) + ok += int(response.status_code == 200) + except httpx.HTTPError: + pass + + sent += 1 + elapsed = time.perf_counter() - started + sleep_for = max(0.0, sent * interval - elapsed) + await asyncio.sleep(sleep_for) + + snap = await snapshot(client, base_url, stage) + return { + "stage": stage, + "sent": sent, + "ok": ok, + **snap, + } + + +async def main() -> None: + args = parse_args() + base_url = args.base_url.rstrip("/") + stages: list[dict[str, Any]] = [] + + async with httpx.AsyncClient(timeout=5.0) as client: + await reset_workers(client, base_url) + stages.append(await send_stage(client, base_url, "baseline", args.seconds_per_stage, args.rps)) + + await control_worker(client, base_url, 1, delay_ms=350, fail_rate=0) + stages.append(await send_stage(client, base_url, "slow-worker", args.seconds_per_stage, args.rps)) + + await reset_workers(client, base_url) + await control_worker(client, base_url, 0, delay_ms=0, fail_rate=0.45) + stages.append(await send_stage(client, base_url, "flaky-worker", args.seconds_per_stage, args.rps)) + + await reset_workers(client, base_url) + stages.append(await send_stage(client, base_url, "recovery", args.seconds_per_stage, args.rps)) + + payload = { + "base_url": base_url, + "seconds_per_stage": args.seconds_per_stage, + "target_rps": args.rps, + "stages": stages, + } + args.output.parent.mkdir(parents=True, exist_ok=True) + args.output.write_text(json.dumps(payload, indent=2) + "\n") + print(json.dumps(payload, indent=2)) + + +if __name__ == "__main__": + asyncio.run(main()) diff --git a/bench/results/replay-latest.json b/bench/results/replay-latest.json new file mode 100644 index 0000000..01c5064 --- /dev/null +++ b/bench/results/replay-latest.json @@ -0,0 +1,191 @@ +{ + "base_url": "http://localhost:8500", + "seconds_per_stage": 8.0, + "target_rps": 18.0, + "stages": [ + { + "stage": "baseline", + "sent": 144, + "ok": 144, + "metrics": { + "requests": 144, + "cache_hits": 138, + "worker_errors": 0, + "cache_hit_rate": 0.9583, + "requests_per_second": 11.92, + "p50_latency_ms": 0.05, + "p95_latency_ms": 0.09 + }, + "workers": [ + { + "url": "http://localhost:8511", + "in_flight": 0, + "requests": 4, + "failures": 0, + "latency_ema_ms": 1.71, + "last_error": null, + "strategy": "load_aware", + "index": 0, + "worker_id": "worker-fast", + "model_version": "tiny-text-mps-v1", + "delay_ms": 0, + "fail_rate": 0.0, + "control_status": "ok" + }, + { + "url": "http://localhost:8512", + "in_flight": 0, + "requests": 2, + "failures": 0, + "latency_ema_ms": 1.31, + "last_error": null, + "strategy": "load_aware", + "index": 1, + "worker_id": "worker-control", + "model_version": "tiny-text-mps-v1", + "delay_ms": 0, + "fail_rate": 0.0, + "control_status": "ok" + } + ] + }, + { + "stage": "slow-worker", + "sent": 144, + "ok": 144, + "metrics": { + "requests": 288, + "cache_hits": 138, + "worker_errors": 0, + "cache_hit_rate": 0.4792, + "requests_per_second": 14.32, + "p50_latency_ms": 1.32, + "p95_latency_ms": 5.11 + }, + "workers": [ + { + "url": "http://localhost:8511", + "in_flight": 0, + "requests": 147, + "failures": 0, + "latency_ema_ms": 4.59, + "last_error": null, + "strategy": "load_aware", + "index": 0, + "worker_id": "worker-fast", + "model_version": "tiny-text-mps-v1", + "delay_ms": 0, + "fail_rate": 0.0, + "control_status": "ok" + }, + { + "url": "http://localhost:8512", + "in_flight": 0, + "requests": 3, + "failures": 0, + "latency_ema_ms": 72.27, + "last_error": null, + "strategy": "load_aware", + "index": 1, + "worker_id": "worker-control", + "model_version": "tiny-text-mps-v1", + "delay_ms": 350, + "fail_rate": 0.0, + "control_status": "ok" + } + ] + }, + { + "stage": "flaky-worker", + "sent": 144, + "ok": 144, + "metrics": { + "requests": 432, + "cache_hits": 138, + "worker_errors": 0, + "cache_hit_rate": 0.3194, + "requests_per_second": 15.35, + "p50_latency_ms": 3.96, + "p95_latency_ms": 5.34 + }, + "workers": [ + { + "url": "http://localhost:8511", + "in_flight": 0, + "requests": 151, + "failures": 1, + "latency_ema_ms": 3.33, + "last_error": "HTTPStatusError", + "strategy": "load_aware", + "index": 0, + "worker_id": "worker-fast", + "model_version": "tiny-text-mps-v1", + "delay_ms": 0, + "fail_rate": 0.45, + "control_status": "ok" + }, + { + "url": "http://localhost:8512", + "in_flight": 0, + "requests": 144, + "failures": 0, + "latency_ema_ms": 4.55, + "last_error": null, + "strategy": "load_aware", + "index": 1, + "worker_id": "worker-control", + "model_version": "tiny-text-mps-v1", + "delay_ms": 0, + "fail_rate": 0.0, + "control_status": "ok" + } + ] + }, + { + "stage": "recovery", + "sent": 144, + "ok": 144, + "metrics": { + "requests": 576, + "cache_hits": 282, + "worker_errors": 0, + "cache_hit_rate": 0.4896, + "requests_per_second": 15.92, + "p50_latency_ms": 3.45, + "p95_latency_ms": 5.25 + }, + "workers": [ + { + "url": "http://localhost:8511", + "in_flight": 0, + "requests": 151, + "failures": 1, + "latency_ema_ms": 3.33, + "last_error": "HTTPStatusError", + "strategy": "load_aware", + "index": 0, + "worker_id": "worker-fast", + "model_version": "tiny-text-mps-v1", + "delay_ms": 0, + "fail_rate": 0.0, + "control_status": "ok" + }, + { + "url": "http://localhost:8512", + "in_flight": 0, + "requests": 144, + "failures": 0, + "latency_ema_ms": 4.55, + "last_error": null, + "strategy": "load_aware", + "index": 1, + "worker_id": "worker-control", + "model_version": "tiny-text-mps-v1", + "delay_ms": 0, + "fail_rate": 0.0, + "control_status": "ok" + } + ] + } + ] +} diff --git a/docs/demo.md b/docs/demo.md new file mode 100644 index 0000000..59747e7 --- /dev/null +++ b/docs/demo.md @@ -0,0 +1,32 @@ +# Demo script + +This is the local demo flow I use to show the system moving. + +Terminal 1: + +```bash +docker compose up --build +``` + +Open: + +```text +http://localhost:8000/dashboard +``` + +Terminal 2: + +```bash +python bench/replay_traffic.py --seconds-per-stage 8 --rps 18 +``` + +What to watch: + +- Baseline traffic warms up the gateway. +- The slow-worker stage adds delay to worker 1 through the gateway control endpoint. +- The flaky-worker stage sets worker 0 to a 45 percent failure rate. +- Recovery resets both workers. + +The dashboard should show request count, p95 latency, worker failure counts, delay, and fail rate moving as the replay runs. + +The replay writes `bench/results/replay-latest.json` for later inspection. From 6004cd72cd4d144495c41e2388ddf1c5668961f4 Mon Sep 17 00:00:00 2001 From: Prakhar Verma Date: Fri, 8 May 2026 00:28:52 -0400 Subject: [PATCH 4/5] Refresh benchmark results --- bench/results/latest.json | 164 ++++++++++++++++++++++++-------------- bench/results/latest.md | 10 +-- 2 files changed, 111 insertions(+), 63 deletions(-) diff --git a/bench/results/latest.json b/bench/results/latest.json index 900d514..b709c26 100644 --- a/bench/results/latest.json +++ b/bench/results/latest.json @@ -10,30 +10,36 @@ "concurrency": 8, "successes": 80, "failures": 0, - "cache_hits": 71, - "cache_hit_rate": 0.8875, - "total_ms": 378.48, - "requests_per_second": 211.37, - "p50_latency_ms": 5.54, - "p95_latency_ms": 312.52, + "cache_hits": 72, + "cache_hit_rate": 0.9, + "total_ms": 109.75, + "requests_per_second": 728.95, + "p50_latency_ms": 5.01, + "p95_latency_ms": 52.04, "gateway_metrics": { "requests": 80, - "cache_hits": 71, + "cache_hits": 72, "worker_errors": 0, - "cache_hit_rate": 0.8875, - "requests_per_second": 159.35, - "p50_latency_ms": 0.01, - "p95_latency_ms": 294.25 + "cache_hit_rate": 0.9, + "requests_per_second": 356.94, + "p50_latency_ms": 0.0, + "p95_latency_ms": 41.44 }, "workers": [ { "url": "http://localhost:8311", "in_flight": 0, - "requests": 9, + "requests": 8, "failures": 0, - "latency_ema_ms": 268.22, + "latency_ema_ms": 41.54, "last_error": null, - "strategy": "load_aware" + "strategy": "load_aware", + "index": 0, + "worker_id": "worker-a", + "model_version": "tiny-text-mps-v1", + "delay_ms": 35, + "fail_rate": 0.0, + "control_status": "ok" } ] }, @@ -48,18 +54,18 @@ "failures": 0, "cache_hits": 0, "cache_hit_rate": 0.0, - "total_ms": 3210.55, - "requests_per_second": 24.92, - "p50_latency_ms": 318.85, - "p95_latency_ms": 326.38, + "total_ms": 430.79, + "requests_per_second": 185.7, + "p50_latency_ms": 41.1, + "p95_latency_ms": 53.09, "gateway_metrics": { "requests": 80, "cache_hits": 0, "worker_errors": 0, "cache_hit_rate": 0.0, - "requests_per_second": 23.93, - "p50_latency_ms": 309.16, - "p95_latency_ms": 316.54 + "requests_per_second": 143.14, + "p50_latency_ms": 38.43, + "p95_latency_ms": 42.38 }, "workers": [ { @@ -67,9 +73,15 @@ "in_flight": 0, "requests": 80, "failures": 0, - "latency_ema_ms": 294.34, + "latency_ema_ms": 38.69, "last_error": null, - "strategy": "load_aware" + "strategy": "load_aware", + "index": 0, + "worker_id": "worker-a", + "model_version": "tiny-text-mps-v1", + "delay_ms": 35, + "fail_rate": 0.0, + "control_status": "ok" } ] }, @@ -84,37 +96,49 @@ "failures": 0, "cache_hits": 0, "cache_hit_rate": 0.0, - "total_ms": 3115.31, - "requests_per_second": 25.68, - "p50_latency_ms": 67.3, - "p95_latency_ms": 1033.24, + "total_ms": 426.99, + "requests_per_second": 187.36, + "p50_latency_ms": 57.79, + "p95_latency_ms": 111.91, "gateway_metrics": { "requests": 80, "cache_hits": 0, "worker_errors": 0, "cache_hit_rate": 0.0, - "requests_per_second": 24.67, - "p50_latency_ms": 108.0, - "p95_latency_ms": 1029.4 + "requests_per_second": 143.63, + "p50_latency_ms": 95.87, + "p95_latency_ms": 103.0 }, "workers": [ { "url": "http://localhost:8311", "in_flight": 0, - "requests": 57, + "requests": 40, "failures": 0, - "latency_ema_ms": 11.48, + "latency_ema_ms": 1.7, "last_error": null, - "strategy": "round_robin" + "strategy": "round_robin", + "index": 0, + "worker_id": "worker-fast", + "model_version": "tiny-text-mps-v1", + "delay_ms": 0, + "fail_rate": 0.0, + "control_status": "ok" }, { "url": "http://localhost:8312", "in_flight": 0, "requests": 40, - "failures": 17, - "latency_ema_ms": 777.5, - "last_error": "ReadTimeout", - "strategy": "round_robin" + "failures": 0, + "latency_ema_ms": 97.45, + "last_error": null, + "strategy": "round_robin", + "index": 1, + "worker_id": "worker-slow", + "model_version": "tiny-text-mps-v1", + "delay_ms": 95, + "fail_rate": 0.0, + "control_status": "ok" } ] }, @@ -129,18 +153,18 @@ "failures": 0, "cache_hits": 0, "cache_hit_rate": 0.0, - "total_ms": 622.53, - "requests_per_second": 128.51, - "p50_latency_ms": 4.94, - "p95_latency_ms": 425.03, + "total_ms": 116.2, + "requests_per_second": 688.48, + "p50_latency_ms": 5.21, + "p95_latency_ms": 111.9, "gateway_metrics": { "requests": 80, "cache_hits": 0, "worker_errors": 0, "cache_hit_rate": 0.0, - "requests_per_second": 109.34, - "p50_latency_ms": 3.61, - "p95_latency_ms": 408.65 + "requests_per_second": 333.04, + "p50_latency_ms": 3.75, + "p95_latency_ms": 101.86 }, "workers": [ { @@ -148,18 +172,30 @@ "in_flight": 0, "requests": 74, "failures": 0, - "latency_ema_ms": 3.82, + "latency_ema_ms": 3.51, "last_error": null, - "strategy": "load_aware" + "strategy": "load_aware", + "index": 0, + "worker_id": "worker-fast", + "model_version": "tiny-text-mps-v1", + "delay_ms": 0, + "fail_rate": 0.0, + "control_status": "ok" }, { "url": "http://localhost:8312", "in_flight": 0, "requests": 6, "failures": 0, - "latency_ema_ms": 365.08, + "latency_ema_ms": 102.25, "last_error": null, - "strategy": "load_aware" + "strategy": "load_aware", + "index": 1, + "worker_id": "worker-slow", + "model_version": "tiny-text-mps-v1", + "delay_ms": 95, + "fail_rate": 0.0, + "control_status": "ok" } ] }, @@ -174,18 +210,18 @@ "failures": 0, "cache_hits": 0, "cache_hit_rate": 0.0, - "total_ms": 85.84, - "requests_per_second": 698.97, - "p50_latency_ms": 11.15, - "p95_latency_ms": 28.1, + "total_ms": 84.75, + "requests_per_second": 707.98, + "p50_latency_ms": 11.37, + "p95_latency_ms": 28.93, "gateway_metrics": { "requests": 60, "cache_hits": 0, "worker_errors": 0, "cache_hit_rate": 0.0, - "requests_per_second": 275.53, - "p50_latency_ms": 8.53, - "p95_latency_ms": 19.94 + "requests_per_second": 281.19, + "p50_latency_ms": 8.46, + "p95_latency_ms": 22.61 }, "workers": [ { @@ -195,16 +231,28 @@ "failures": 5, "latency_ema_ms": 0.0, "last_error": "HTTPStatusError", - "strategy": "load_aware" + "strategy": "load_aware", + "index": 0, + "worker_id": "worker-dead", + "model_version": "tiny-text-mps-v1", + "delay_ms": 0, + "fail_rate": 1.0, + "control_status": "ok" }, { "url": "http://localhost:8312", "in_flight": 0, "requests": 60, "failures": 0, - "latency_ema_ms": 8.83, + "latency_ema_ms": 10.88, "last_error": null, - "strategy": "load_aware" + "strategy": "load_aware", + "index": 1, + "worker_id": "worker-healthy", + "model_version": "tiny-text-mps-v1", + "delay_ms": 0, + "fail_rate": 0.0, + "control_status": "ok" } ] } diff --git a/bench/results/latest.md b/bench/results/latest.md index bf46122..32f641a 100644 --- a/bench/results/latest.md +++ b/bench/results/latest.md @@ -4,11 +4,11 @@ Generated with `python bench/run_scenarios.py` on the local machine. | Scenario | Router | Cache | Requests | RPS | p50 ms | p95 ms | Cache hits | Failures | | --- | --- | --- | ---: | ---: | ---: | ---: | ---: | ---: | -| cache_enabled | load_aware | True | 80 | 211.37 | 5.54 | 312.52 | 71 | 0 | -| cache_disabled | load_aware | False | 80 | 24.92 | 318.85 | 326.38 | 0 | 0 | -| slow_worker_round_robin | round_robin | False | 80 | 25.68 | 67.3 | 1033.24 | 0 | 0 | -| slow_worker_load_aware | load_aware | False | 80 | 128.51 | 4.94 | 425.03 | 0 | 0 | -| dead_worker_recovery | load_aware | False | 60 | 698.97 | 11.15 | 28.1 | 0 | 0 | +| cache_enabled | load_aware | True | 80 | 728.95 | 5.01 | 52.04 | 72 | 0 | +| cache_disabled | load_aware | False | 80 | 185.7 | 41.1 | 53.09 | 0 | 0 | +| slow_worker_round_robin | round_robin | False | 80 | 187.36 | 57.79 | 111.91 | 0 | 0 | +| slow_worker_load_aware | load_aware | False | 80 | 688.48 | 5.21 | 111.9 | 0 | 0 | +| dead_worker_recovery | load_aware | False | 60 | 707.98 | 11.37 | 28.93 | 0 | 0 | Notes: From 3c3211bc3b5c5a3b1e544f80533d9c4e64ebab74 Mon Sep 17 00:00:00 2001 From: Prakhar Verma Date: Fri, 8 May 2026 00:28:55 -0400 Subject: [PATCH 5/5] Explain the live demo path --- README.md | 34 ++++++++++++++++++++++++++++------ docs/architecture.md | 11 +++++++++++ 2 files changed, 39 insertions(+), 6 deletions(-) diff --git a/README.md b/README.md index 041cda7..a6e6556 100644 --- a/README.md +++ b/README.md @@ -36,6 +36,12 @@ Open the local dashboard: http://localhost:8000/dashboard ``` +Run live traffic against the dashboard: + +```bash +python bench/replay_traffic.py --seconds-per-stage 8 --rps 18 +``` + ## Docker ```bash @@ -56,17 +62,17 @@ Latest local run: | Scenario | RPS | p50 ms | p95 ms | Cache hits | Failures | | --- | ---: | ---: | ---: | ---: | ---: | -| cache enabled | 211.37 | 5.54 | 312.52 | 71 | 0 | -| cache disabled | 24.92 | 318.85 | 326.38 | 0 | 0 | -| slow worker, round-robin | 25.68 | 67.3 | 1033.24 | 0 | 0 | -| slow worker, load-aware | 128.51 | 4.94 | 425.03 | 0 | 0 | -| dead worker recovery | 698.97 | 11.15 | 28.1 | 0 | 0 | +| cache enabled | 728.95 | 5.01 | 52.04 | 72 | 0 | +| cache disabled | 185.7 | 41.1 | 53.09 | 0 | 0 | +| slow worker, round-robin | 187.36 | 57.79 | 111.91 | 0 | 0 | +| slow worker, load-aware | 688.48 | 5.21 | 111.9 | 0 | 0 | +| dead worker recovery | 707.98 | 11.37 | 28.93 | 0 | 0 | Full output is in `bench/results/latest.md` and `bench/results/latest.json`. ## Training -The worker can load a small trained text model artifact from `worker/app/artifacts/tiny_text_model.json`. +The worker can load a small trained text model artifact from `worker/app/artifacts/tiny_text_model.json`. The current model trains on generated service-status examples, so the point is the local training and serving path, not benchmark-grade NLP accuracy. On Apple Silicon, the training script uses PyTorch MPS. It refuses to run on CPU by default so a slow accidental CPU run fails fast. @@ -77,6 +83,21 @@ python training/train_text_model.py If it is using the Mac GPU, the output includes `'device': 'mps'`. +Latest training run on an M3 Pro: + +| Metric | Value | +| --- | ---: | +| Device | mps | +| Epochs | 320 | +| Train examples | 3,276 | +| Test examples | 820 | +| Train time | 0.4351 s | +| Examples/sec | 2,409,512.37 | +| Train accuracy | 1.0 | +| Test accuracy | 1.0 | + +The full run output is in `training/runs/latest.md` and `training/runs/latest.json`. + ## What is in here now - FastAPI gateway with `/predict`, `/health`, `/workers`, and `/metrics` @@ -85,6 +106,7 @@ If it is using the Mac GPU, the output includes `'device': 'mps'`. - Retry on failed worker calls and basic load-aware routing - Scenario benchmark runner with saved results - Local dashboard for gateway and worker state +- Worker delay/failure controls for live demos - Unit tests for routing, caching, metrics, and the model This is still intentionally small. The useful part is the system behavior around the model, not the model itself. diff --git a/docs/architecture.md b/docs/architecture.md index 2552661..e2b1ea2 100644 --- a/docs/architecture.md +++ b/docs/architecture.md @@ -35,3 +35,14 @@ The benchmark runner uses both modes against the same slow-worker scenario so th `training/train_text_model.py` trains the tiny text model with PyTorch. On Apple Silicon it uses `mps`, and it exits instead of silently falling back to CPU unless `--allow-cpu` is passed. The trained artifact is plain JSON so the worker can load it without depending on PyTorch at inference time. + +The latest checked-in run trained on `mps` in under a second. The run file records device, epochs, examples/sec, train accuracy, and test accuracy so the performance claim is reproducible. + +## Live failure demo + +Workers expose a small local control endpoint for demo runs: + +- `GET /config` returns delay and failure settings. +- `POST /control` updates artificial delay and failure rate. + +The gateway wraps those controls at `POST /workers/{index}/control`, and the dashboard exposes inputs for delay and fail rate. `bench/replay_traffic.py` uses the same endpoint to move the system through baseline, slow-worker, flaky-worker, and recovery stages.