feat: playground simulador de rendimiento NATS (webapp WebSocket + canvas)
Reemplaza el widget ipywidgets del notebook 04 (fragil: 'model not found' sin re-ejecutar) por una webapp standalone. server.py corre el benchmark NATS y transmite muestras por WebSocket; index.html dibuja la grafica en movimiento en un canvas sin dependencias front. Reutiliza el venv del analisis. Verificado: 100k msgs -> 4 subs = 400k entregas (~367k/s) en ~1.1s.
This commit is contained in:
Egutierrez
@@ -0,0 +1,2 @@
|
||||
playground/playground.log
|
||||
playground/__pycache__/
|
||||
+17
-1
@@ -25,7 +25,23 @@ Analisis didactico de **NATS** como sistema de mensajeria pub/sub entre procesos
|
||||
|
||||
Los scripts `notebooks/procs/publisher.py` y `notebooks/procs/subscriber.py` son los programas que el notebook 03 lanza como procesos reales.
|
||||
|
||||
El notebook 04 requiere `ipywidgets` (incluido en el `.venv` del análisis). El simulador es interactivo: al abrir el notebook en JupyterLab, ejecuta sus celdas hasta el widget y pulsa **▶ Ejecutar benchmark** (los sliders ajustan número de mensajes y de subscribers). La gráfica se anima mientras corre.
|
||||
El notebook 04 requiere `ipywidgets` (incluido en el `.venv` del análisis). El simulador del notebook es interactivo: al abrir el notebook en JupyterLab, ejecuta sus celdas hasta el widget y pulsa **▶ Ejecutar benchmark** (los sliders ajustan número de mensajes y de subscribers). La gráfica se anima mientras corre.
|
||||
|
||||
> Si al abrir el notebook el widget muestra `Error displaying widget: model not found`, **re-ejecuta la celda del widget** en tu kernel (los modelos de `ipywidgets` no se rehidratan desde un kernel anterior). Para una versión interactiva más robusta y sin depender de Jupyter, usa el **playground** (ver abajo).
|
||||
|
||||
## Playground: simulador de rendimiento (webapp)
|
||||
|
||||
`playground/` es una webapp standalone equivalente al simulador del notebook 04, pero sin `ipywidgets`: sirve una página con un botón y unos sliders, y al pulsarlo lanza el benchmark en el servidor y transmite las muestras por WebSocket a un canvas que dibuja la gráfica en movimiento en el navegador. Reutiliza el `.venv` del análisis (con `nats-py` y `websockets`); no tiene dependencias ni repo propios.
|
||||
|
||||
```bash
|
||||
cd analysis/nats/playground
|
||||
../.venv/bin/python server.py
|
||||
# abrir http://127.0.0.1:7788 (WebSocket en 7879)
|
||||
```
|
||||
|
||||
Pulsa **▶ Ejecutar benchmark**: un publisher envía N mensajes (slider, hasta 200.000) a M subscribers (slider, hasta 12) y la gráfica muestra en vivo los acumulados de enviados vs recibidos. Verificado: 100.000 msgs → 4 subs = 400.000 entregas en ~1,1 s (fan-out ×4 exacto, ~367.000 entregas/s).
|
||||
|
||||
Archivos: `playground/server.py` (servidor WebSocket + HTTP estático + benchmark NATS) y `playground/index.html` (UI con canvas, sin librerías externas).
|
||||
|
||||
### Como usar
|
||||
|
||||
|
||||
@@ -0,0 +1,221 @@
|
||||
<!DOCTYPE html>
|
||||
<html lang="es">
|
||||
<head>
|
||||
<meta charset="utf-8">
|
||||
<meta name="viewport" content="width=device-width, initial-scale=1">
|
||||
<title>NATS · Simulador de rendimiento pub/sub</title>
|
||||
<style>
|
||||
:root {
|
||||
--bg: #0b1020; --panel: #151b30; --line: #243049;
|
||||
--blue: #3b82f6; --green: #22c55e; --pink: #ec4899; --txt: #e5e9f0; --muted: #8b95ad;
|
||||
}
|
||||
* { box-sizing: border-box; }
|
||||
body {
|
||||
margin: 0; background: var(--bg); color: var(--txt);
|
||||
font-family: ui-sans-serif, system-ui, -apple-system, "Segoe UI", Roboto, sans-serif;
|
||||
}
|
||||
header { padding: 20px 28px 8px; }
|
||||
h1 { margin: 0; font-size: 20px; font-weight: 650; }
|
||||
.sub { color: var(--muted); font-size: 13px; margin-top: 4px; }
|
||||
.wrap { max-width: 1000px; margin: 0 auto; padding: 0 28px 40px; }
|
||||
.panel { background: var(--panel); border: 1px solid var(--line); border-radius: 12px; padding: 18px 20px; }
|
||||
.controls { display: flex; gap: 28px; flex-wrap: wrap; align-items: flex-end; margin-bottom: 16px; }
|
||||
.ctl { display: flex; flex-direction: column; gap: 6px; }
|
||||
.ctl label { font-size: 12px; color: var(--muted); }
|
||||
.ctl .val { font-weight: 650; color: var(--txt); font-variant-numeric: tabular-nums; }
|
||||
input[type=range] { width: 240px; accent-color: var(--blue); }
|
||||
button {
|
||||
background: var(--green); color: #06210f; border: 0; border-radius: 9px;
|
||||
padding: 11px 20px; font-size: 14px; font-weight: 700; cursor: pointer;
|
||||
}
|
||||
button:disabled { opacity: .5; cursor: not-allowed; }
|
||||
.stats { display: flex; gap: 26px; flex-wrap: wrap; margin: 16px 2px 4px; }
|
||||
.stat { min-width: 120px; }
|
||||
.stat .k { font-size: 11px; color: var(--muted); text-transform: uppercase; letter-spacing: .04em; }
|
||||
.stat .v { font-size: 24px; font-weight: 700; font-variant-numeric: tabular-nums; }
|
||||
canvas { width: 100%; height: 340px; display: block; margin-top: 8px; border-radius: 8px; }
|
||||
.legend { display: flex; gap: 18px; font-size: 12px; color: var(--muted); margin-top: 6px; }
|
||||
.dot { display: inline-block; width: 10px; height: 10px; border-radius: 50%; margin-right: 6px; vertical-align: middle; }
|
||||
.state { font-size: 13px; color: var(--muted); margin-left: 14px; }
|
||||
.persub { font-size: 12px; color: var(--muted); margin-top: 8px; font-variant-numeric: tabular-nums; }
|
||||
</style>
|
||||
</head>
|
||||
<body>
|
||||
<header>
|
||||
<h1>NATS · Simulador de rendimiento pub/sub</h1>
|
||||
<div class="sub">Un publisher inunda el broker con miles de mensajes que varios subscribers reciben (fan-out). La gráfica se mueve en tiempo real.</div>
|
||||
</header>
|
||||
|
||||
<div class="wrap">
|
||||
<div class="panel">
|
||||
<div class="controls">
|
||||
<div class="ctl">
|
||||
<label>Mensajes a publicar</label>
|
||||
<input id="msgs" type="range" min="1000" max="200000" step="1000" value="60000">
|
||||
<span class="val" id="msgsVal">60 000</span>
|
||||
</div>
|
||||
<div class="ctl">
|
||||
<label>Subscribers</label>
|
||||
<input id="subs" type="range" min="1" max="12" step="1" value="3">
|
||||
<span class="val" id="subsVal">3</span>
|
||||
</div>
|
||||
<div>
|
||||
<button id="run">▶ Ejecutar benchmark</button>
|
||||
<span class="state" id="state">listo</span>
|
||||
</div>
|
||||
</div>
|
||||
|
||||
<div class="stats">
|
||||
<div class="stat"><div class="k">Enviados</div><div class="v" id="sent" style="color:var(--blue)">0</div></div>
|
||||
<div class="stat"><div class="k">Recibidos (Σ subs)</div><div class="v" id="recv" style="color:var(--green)">0</div></div>
|
||||
<div class="stat"><div class="k">Throughput recv</div><div class="v" id="tps" style="color:var(--pink)">0</div></div>
|
||||
<div class="stat"><div class="k">Tiempo</div><div class="v" id="time">0.00 s</div></div>
|
||||
</div>
|
||||
|
||||
<canvas id="chart"></canvas>
|
||||
<div class="legend">
|
||||
<span><span class="dot" style="background:var(--blue)"></span>enviados (publisher)</span>
|
||||
<span><span class="dot" style="background:var(--green)"></span>recibidos (suma de subscribers)</span>
|
||||
</div>
|
||||
<div class="persub" id="persub"></div>
|
||||
</div>
|
||||
</div>
|
||||
|
||||
<script>
|
||||
const WS_URL = `ws://${location.hostname}:7879`;
|
||||
const $ = (id) => document.getElementById(id);
|
||||
const fmt = (n) => n.toLocaleString("es-ES");
|
||||
|
||||
const msgs = $("msgs"), subs = $("subs");
|
||||
const msgsVal = $("msgsVal"), subsVal = $("subsVal");
|
||||
msgs.oninput = () => msgsVal.textContent = fmt(+msgs.value);
|
||||
subs.oninput = () => subsVal.textContent = subs.value;
|
||||
|
||||
// --- estado del benchmark ---
|
||||
let samples = []; // {t, sent, recv}
|
||||
let running = false;
|
||||
|
||||
// --- canvas con soporte HiDPI ---
|
||||
const canvas = $("chart"), ctx = canvas.getContext("2d");
|
||||
function resize() {
|
||||
const dpr = window.devicePixelRatio || 1;
|
||||
const r = canvas.getBoundingClientRect();
|
||||
canvas.width = r.width * dpr;
|
||||
canvas.height = r.height * dpr;
|
||||
ctx.setTransform(dpr, 0, 0, dpr, 0, 0);
|
||||
draw();
|
||||
}
|
||||
window.addEventListener("resize", resize);
|
||||
|
||||
function draw() {
|
||||
const r = canvas.getBoundingClientRect();
|
||||
const W = r.width, H = r.height;
|
||||
const padL = 64, padR = 16, padT = 14, padB = 28;
|
||||
ctx.clearRect(0, 0, W, H);
|
||||
|
||||
// fondo de la zona de plot
|
||||
ctx.fillStyle = "#0e1426";
|
||||
ctx.fillRect(padL, padT, W - padL - padR, H - padT - padB);
|
||||
|
||||
if (samples.length < 2) { drawAxes(W, H, padL, padR, padT, padB, 1, 1); return; }
|
||||
|
||||
const tMax = Math.max(samples[samples.length - 1].t, 0.001);
|
||||
const yMax = Math.max(...samples.map(s => Math.max(s.sent, s.recv)), 1);
|
||||
|
||||
drawAxes(W, H, padL, padR, padT, padB, tMax, yMax);
|
||||
|
||||
const x = (t) => padL + (t / tMax) * (W - padL - padR);
|
||||
const y = (v) => (H - padB) - (v / yMax) * (H - padT - padB);
|
||||
|
||||
const series = (key, color) => {
|
||||
ctx.beginPath();
|
||||
ctx.lineWidth = 2.4;
|
||||
ctx.strokeStyle = color;
|
||||
samples.forEach((s, i) => {
|
||||
const px = x(s.t), py = y(s[key]);
|
||||
i === 0 ? ctx.moveTo(px, py) : ctx.lineTo(px, py);
|
||||
});
|
||||
ctx.stroke();
|
||||
};
|
||||
series("sent", getCSS("--blue"));
|
||||
series("recv", getCSS("--green"));
|
||||
}
|
||||
|
||||
function drawAxes(W, H, padL, padR, padT, padB, tMax, yMax) {
|
||||
ctx.strokeStyle = "#243049"; ctx.fillStyle = "#8b95ad"; ctx.lineWidth = 1;
|
||||
ctx.font = "11px ui-sans-serif, system-ui";
|
||||
// Y gridlines (5)
|
||||
for (let i = 0; i <= 5; i++) {
|
||||
const yy = (H - padB) - (i / 5) * (H - padT - padB);
|
||||
ctx.beginPath(); ctx.moveTo(padL, yy); ctx.lineTo(W - padR, yy); ctx.stroke();
|
||||
const val = Math.round((i / 5) * yMax);
|
||||
ctx.textAlign = "right"; ctx.fillText(fmt(val), padL - 8, yy + 3);
|
||||
}
|
||||
// X labels (inicio / fin)
|
||||
ctx.textAlign = "center";
|
||||
ctx.fillText("0 s", padL, H - padB + 16);
|
||||
ctx.fillText(tMax.toFixed(2) + " s", W - padR, H - padB + 16);
|
||||
}
|
||||
|
||||
function getCSS(name) {
|
||||
return getComputedStyle(document.documentElement).getPropertyValue(name).trim();
|
||||
}
|
||||
|
||||
// --- WebSocket / control ---
|
||||
$("run").onclick = () => {
|
||||
if (running) return;
|
||||
samples = [];
|
||||
setRunning(true);
|
||||
$("state").textContent = "conectando…";
|
||||
$("persub").textContent = "";
|
||||
|
||||
const ws = new WebSocket(WS_URL);
|
||||
ws.onopen = () => {
|
||||
$("state").textContent = "corriendo…";
|
||||
ws.send(JSON.stringify({ action: "start", n_msgs: +msgs.value, n_subs: +subs.value }));
|
||||
};
|
||||
ws.onmessage = (ev) => {
|
||||
const m = JSON.parse(ev.data);
|
||||
if (m.type === "sample" || m.type === "start") {
|
||||
if (m.type === "sample") {
|
||||
samples.push({ t: m.t, sent: m.sent, recv: m.recv });
|
||||
$("sent").textContent = fmt(m.sent);
|
||||
$("recv").textContent = fmt(m.recv);
|
||||
$("time").textContent = m.t.toFixed(2) + " s";
|
||||
if (samples.length >= 2) {
|
||||
const a = samples[samples.length - 2], b = samples[samples.length - 1];
|
||||
const dt = b.t - a.t;
|
||||
if (dt > 0) $("tps").textContent = fmt(Math.round((b.recv - a.recv) / dt));
|
||||
}
|
||||
draw();
|
||||
}
|
||||
} else if (m.type === "done") {
|
||||
samples.push({ t: m.t, sent: m.sent, recv: m.recv });
|
||||
$("sent").textContent = fmt(m.sent);
|
||||
$("recv").textContent = fmt(m.recv);
|
||||
$("time").textContent = m.t.toFixed(2) + " s";
|
||||
$("tps").textContent = fmt(m.recv_tps);
|
||||
$("state").textContent = `✓ listo en ${m.t.toFixed(2)} s · pub ${fmt(m.pub_tps)} msgs/s · recv ${fmt(m.recv_tps)} msgs/s`;
|
||||
$("persub").textContent = `por subscriber: [${m.per_sub.map(fmt).join(", ")}] · fan-out ×${m.n_subs ?? m.per_sub.length} = ${fmt(m.recv)} entregas totales`;
|
||||
draw();
|
||||
ws.close();
|
||||
setRunning(false);
|
||||
} else if (m.type === "error") {
|
||||
$("state").textContent = "error: " + m.msg;
|
||||
ws.close();
|
||||
setRunning(false);
|
||||
}
|
||||
};
|
||||
ws.onerror = () => { $("state").textContent = "error de conexión WebSocket"; setRunning(false); };
|
||||
ws.onclose = () => { if (running) setRunning(false); };
|
||||
};
|
||||
|
||||
function setRunning(v) {
|
||||
running = v;
|
||||
$("run").disabled = v;
|
||||
}
|
||||
|
||||
resize();
|
||||
</script>
|
||||
</body>
|
||||
</html>
|
||||
@@ -0,0 +1,156 @@
|
||||
#!/usr/bin/env python3
|
||||
"""Playground: simulador de rendimiento NATS pub/sub con grafica en vivo.
|
||||
|
||||
Sirve una pagina (index.html) con un boton y unos sliders. Al pulsar el boton,
|
||||
el navegador abre un WebSocket y pide un benchmark; el servidor lanza 1 publisher
|
||||
que envia miles de mensajes a N subscribers (fan-out) y va emitiendo muestras
|
||||
(enviados, recibidos, throughput) por el WebSocket. El navegador las pinta en un
|
||||
canvas en tiempo real, de modo que la grafica se mueve mientras corre el test.
|
||||
|
||||
Reutiliza el venv del analisis padre (analysis/nats/.venv, con nats-py). No tiene
|
||||
dependencias propias ni repo propio: vive dentro del sub-repo del analisis.
|
||||
|
||||
Lanzar:
|
||||
cd analysis/nats/playground
|
||||
../.venv/bin/python server.py
|
||||
# abrir http://127.0.0.1:7788
|
||||
"""
|
||||
import asyncio
|
||||
import functools
|
||||
import http.server
|
||||
import json
|
||||
import os
|
||||
import subprocess
|
||||
import threading
|
||||
import time
|
||||
|
||||
import nats
|
||||
from websockets.asyncio.server import serve
|
||||
|
||||
HERE = os.path.dirname(os.path.abspath(__file__))
|
||||
NATS_URL = "nats://127.0.0.1:4222"
|
||||
HTTP_PORT = 7788
|
||||
WS_PORT = 7879
|
||||
SUBJECT = "bench.load"
|
||||
PAYLOAD = b"x" * 128 # 128 bytes por mensaje
|
||||
|
||||
# Limites de seguridad para los parametros que llegan del navegador
|
||||
MAX_MSGS = 500_000
|
||||
MAX_SUBS = 12
|
||||
|
||||
|
||||
def ensure_nats() -> str:
|
||||
"""Arranca el broker NATS en Docker de forma idempotente."""
|
||||
def docker(*args):
|
||||
return subprocess.run(["docker", *args], capture_output=True, text=True)
|
||||
|
||||
state = docker("ps", "-a", "--filter", "name=^nats_demo$", "--format", "{{.State}}").stdout.strip()
|
||||
if state == "running":
|
||||
return "already-running"
|
||||
if state in ("exited", "created", "paused"):
|
||||
docker("start", "nats_demo")
|
||||
time.sleep(1.0)
|
||||
return "started"
|
||||
docker("run", "-d", "--name", "nats_demo", "-p", "4222:4222", "-p", "8222:8222",
|
||||
"nats:latest", "-js", "-m", "8222")
|
||||
time.sleep(1.5)
|
||||
return "created"
|
||||
|
||||
|
||||
def start_http_server() -> None:
|
||||
"""Sirve los archivos estaticos del playground (index.html) en un thread."""
|
||||
handler = functools.partial(http.server.SimpleHTTPRequestHandler, directory=HERE)
|
||||
http.server.ThreadingHTTPServer.allow_reuse_address = True
|
||||
httpd = http.server.ThreadingHTTPServer(("127.0.0.1", HTTP_PORT), handler)
|
||||
httpd.serve_forever()
|
||||
|
||||
|
||||
async def run_benchmark(ws, n_msgs: int, n_subs: int) -> None:
|
||||
"""Un publisher envia n_msgs a n_subs subscribers. Emite muestras por el WebSocket."""
|
||||
nc = await nats.connect(NATS_URL, name="playground-benchmark")
|
||||
counters = [0] * n_subs
|
||||
|
||||
def make_cb(i):
|
||||
async def cb(_msg):
|
||||
counters[i] += 1
|
||||
return cb
|
||||
|
||||
subs = [await nc.subscribe(SUBJECT, cb=make_cb(i)) for i in range(n_subs)]
|
||||
|
||||
sent = 0
|
||||
t0 = time.monotonic()
|
||||
|
||||
async def publish_all():
|
||||
nonlocal sent
|
||||
for k in range(n_msgs):
|
||||
await nc.publish(SUBJECT, PAYLOAD)
|
||||
sent += 1
|
||||
if k % 500 == 0:
|
||||
await nc.flush()
|
||||
await asyncio.sleep(0) # ceder al event loop para que corran los callbacks
|
||||
await nc.flush()
|
||||
|
||||
await ws.send(json.dumps({"type": "start", "n_msgs": n_msgs, "n_subs": n_subs}))
|
||||
task = asyncio.create_task(publish_all())
|
||||
|
||||
# Muestreo periodico para la grafica en movimiento
|
||||
while not task.done() or sum(counters) < sent:
|
||||
await asyncio.sleep(0.04)
|
||||
t = time.monotonic() - t0
|
||||
await ws.send(json.dumps({"type": "sample", "t": round(t, 3), "sent": sent, "recv": sum(counters)}))
|
||||
if t > 60: # tope de seguridad
|
||||
break
|
||||
await task
|
||||
|
||||
# Drenaje final: dar tiempo a que los callbacks alcancen al publisher
|
||||
for _ in range(60):
|
||||
if sum(counters) >= sent:
|
||||
break
|
||||
await asyncio.sleep(0.05)
|
||||
|
||||
dur = time.monotonic() - t0
|
||||
recv = sum(counters)
|
||||
await ws.send(json.dumps({
|
||||
"type": "done",
|
||||
"t": round(dur, 3),
|
||||
"sent": sent,
|
||||
"recv": recv,
|
||||
"per_sub": counters,
|
||||
"pub_tps": round(n_msgs / dur) if dur else 0,
|
||||
"recv_tps": round(recv / dur) if dur else 0,
|
||||
}))
|
||||
|
||||
for s in subs:
|
||||
await s.unsubscribe()
|
||||
await nc.drain()
|
||||
|
||||
|
||||
async def ws_handler(ws) -> None:
|
||||
async for raw in ws:
|
||||
try:
|
||||
msg = json.loads(raw)
|
||||
except (json.JSONDecodeError, TypeError):
|
||||
continue
|
||||
if msg.get("action") == "start":
|
||||
n_msgs = max(1000, min(int(msg.get("n_msgs", 20000)), MAX_MSGS))
|
||||
n_subs = max(1, min(int(msg.get("n_subs", 3)), MAX_SUBS))
|
||||
try:
|
||||
await run_benchmark(ws, n_msgs, n_subs)
|
||||
except Exception as exc: # noqa: BLE001 - reportar cualquier fallo al navegador
|
||||
await ws.send(json.dumps({"type": "error", "msg": f"{type(exc).__name__}: {exc}"}))
|
||||
|
||||
|
||||
async def main() -> None:
|
||||
print("broker NATS:", ensure_nats())
|
||||
threading.Thread(target=start_http_server, daemon=True).start()
|
||||
print(f"HTTP -> http://127.0.0.1:{HTTP_PORT}")
|
||||
print(f"WS -> ws://127.0.0.1:{WS_PORT}")
|
||||
async with serve(ws_handler, "127.0.0.1", WS_PORT):
|
||||
await asyncio.Future() # corre indefinidamente
|
||||
|
||||
|
||||
if __name__ == "__main__":
|
||||
try:
|
||||
asyncio.run(main())
|
||||
except KeyboardInterrupt:
|
||||
print("\nplayground detenido")
|
||||
Reference in New Issue
Block a user