diff --git a/.dockerignore b/.dockerignore index 8440f97..4611add 100644 --- a/.dockerignore +++ b/.dockerignore @@ -4,7 +4,7 @@ !target/release/swactor-diag-postproc !examples/single-gpu-inference/target/release/gpu-node !examples/single-gpu-inference/tinygrad_worker.py -!examples/pipeline-parallel-inference/target/release/pp-gpu-node -!examples/pipeline-parallel-inference/target/release/pp-smoke-run +!examples/pipeline-parallel-inference/target/release/pp-worker +!examples/pipeline-parallel-inference/target/release/pp-orchestrator !examples/pipeline-parallel-inference/pp_tinygrad_worker.py !examples/pipeline-parallel-inference/pp_entrypoint.sh diff --git a/crates/dashboard/src/html.rs b/crates/dashboard/src/html.rs index d4edf36..ac8f89b 100644 --- a/crates/dashboard/src/html.rs +++ b/crates/dashboard/src/html.rs @@ -125,6 +125,7 @@ pub const ACTOR_DETAIL_HTML: &str = r##" Actors Distribution Datastore + Fleet @@ -569,6 +570,7 @@ pub const ACTORS_HTML: &str = r##" Actors Distribution Datastore + Fleet
@@ -1377,6 +1379,7 @@ pub const DASHBOARD_HTML: &str = r##" Actors Distribution Datastore + Fleet
@@ -1934,6 +1937,7 @@ pub const TOPOLOGY_HTML: &str = r##" Topology Distribution Datastore + Fleet
diff --git a/crates/distribution/src/iroh_driver.rs b/crates/distribution/src/iroh_driver.rs index 554a76f..9628491 100644 --- a/crates/distribution/src/iroh_driver.rs +++ b/crates/distribution/src/iroh_driver.rs @@ -229,13 +229,9 @@ impl IrohDriver { match rt.block_on(start_embedded_relay(bind_addr, config.relay_public_ip)) { Ok((server, url)) => { let url_str = url.to_string(); - eprintln!("Relay: embedded relay started at {url}"); (Some(server), Some(url_str), RelayMode::Custom(url.into())) } - Err(e) => { - eprintln!("Relay: failed to start embedded relay: {e}, falling back"); - (None, None, config.relay_mode) - } + Err(_) => (None, None, config.relay_mode), } } None => (None, None, config.relay_mode), @@ -301,33 +297,18 @@ impl IrohDriver { Some(auth) => auth.lock().unwrap().is_allowed(&node_id), }; if !allowed { - eprintln!( - "iroh driver: rejected connection from unauthorized peer {}", - swactor::transport::hex_encode(&node_id.0[..4]) - ); conn.close(0u32.into(), b"unauthorized"); continue; } // Route by negotiated ALPN let negotiated_alpn = conn.alpn(); if negotiated_alpn == ALPN { - eprintln!( - "iroh driver: accepted SWIM connection from {}", - swactor::transport::hex_encode(&node_id.0[..4]) - ); swim_buf.lock().unwrap().push((node_id, conn)); } else { - eprintln!( - "iroh driver: accepted non-SWIM connection from {} (ALPN: {})", - swactor::transport::hex_encode(&node_id.0[..4]), - String::from_utf8_lossy(negotiated_alpn), - ); other_buf.lock().unwrap().push((node_id, conn)); } } - Err(e) => { - eprintln!("iroh driver: incoming connection error: {e}"); - } + Err(_) => {} }, None => break, // endpoint closed } @@ -659,7 +640,6 @@ impl IrohDriver { }); } - eprintln!("iroh driver: join attempt {attempt}/{max_attempts} connecting to {}...", seed_addr.id); diagnostics.emit_event(DiagEvent::DialStarted { peer: seed_node_id, attempt, @@ -726,7 +706,6 @@ impl IrohDriver { }); } - eprintln!("iroh driver: join attempt {attempt}/{max_attempts} connected to {}, sending...", seed_addr.id); let send_result: Result<(), String> = async { let mut send = conn.open_uni().await.map_err(|e| e.to_string())?; let tag_len = (tag.len() as u32).to_be_bytes(); @@ -740,7 +719,6 @@ impl IrohDriver { match send_result { Ok(()) => { - eprintln!("iroh driver: join attempt {attempt}/{max_attempts} sent to {}", seed_addr.id); diagnostics.emit_event(DiagEvent::MessageSent { peer: seed_node_id, kind: tag.to_string(), @@ -764,10 +742,6 @@ impl IrohDriver { return; } Err(e) => { - eprintln!( - "iroh driver: join attempt {attempt}/{max_attempts} send error to {}: {e}", - seed_addr.id - ); diagnostics.emit_event(DiagEvent::Error { component: "iroh_driver".into(), message: format!("join send error: {e}"), @@ -778,10 +752,6 @@ impl IrohDriver { } } Ok(Err(e)) => { - eprintln!( - "iroh driver: join attempt {attempt}/{max_attempts} connect error to {}: {e}", - seed_addr.id - ); let outcome = classify_dial_error_str(&e.to_string()); diagnostics.emit_event(DiagEvent::DialOutcome { peer: seed_node_id, @@ -792,10 +762,6 @@ impl IrohDriver { continue; } Err(_) => { - eprintln!( - "iroh driver: join attempt {attempt}/{max_attempts} connect timeout to {}", - seed_addr.id - ); diagnostics.emit_event(DiagEvent::DialOutcome { peer: seed_node_id, attempt, @@ -817,7 +783,6 @@ impl IrohDriver { updated_at: Instant::now(), }); } - eprintln!("iroh driver: join failed after {max_attempts} attempts to {}", seed_addr.id); }); } @@ -832,9 +797,6 @@ impl IrohDriver { // Collect completed background join connections { let mut pending = self.pending_joins.lock().unwrap(); - if !pending.is_empty() { - eprintln!("iroh driver: collecting {} pending join connection(s)", pending.len()); - } for result in pending.drain(..) { self.connection_cache_tracker .note_dial_success(result.node_id, wall_ms_now()); @@ -864,7 +826,6 @@ impl IrohDriver { let mut failure_targets: Vec = Vec::new(); for action in actions { if let Err(e) = self.send_action(action) { - eprintln!("iroh driver: send error: {e}"); let target = action_target(action); self.diagnostics.emit_event(DiagEvent::Error { component: "iroh_driver".into(), @@ -882,9 +843,7 @@ impl IrohDriver { let probe_actions = self.node.report_send_failure(target); // Best-effort send of probe actions — no recursion on failure for action in &probe_actions { - if let Err(e) = self.send_action(action) { - eprintln!("iroh driver: probe send error: {e}"); - } + let _ = self.send_action(action); } } } @@ -1230,10 +1189,6 @@ impl IrohDriver { self.read_streams(&conn, remote_id, &mut messages).await; } - if !messages.is_empty() { - eprintln!("iroh driver: received {} message(s)", messages.len()); - } - (messages, new_connections) } @@ -1250,8 +1205,7 @@ impl IrohDriver { Ok((tag, payload)) => { messages.push((tag, payload, remote_id)); } - Err(e) => { - eprintln!("iroh driver: read error: {e}"); + Err(_) => { break; } } @@ -1276,18 +1230,12 @@ impl IrohDriver { match tag { "swactor_dist::Ping" => match serde_json::from_slice::(payload) { Ok(msg) => self.node.handle_ping(msg.from, msg.sequence, &msg.piggyback), - Err(e) => { - eprintln!("iroh driver: decode Ping: {e}"); - Vec::new() - } + Err(_) => Vec::new(), }, "swactor_dist::Ack" => match serde_json::from_slice::(payload) { Ok(msg) => self.node.handle_ack(msg.from, msg.sequence, &msg.piggyback), - Err(e) => { - eprintln!("iroh driver: decode Ack: {e}"); - Vec::new() - } + Err(_) => Vec::new(), }, "swactor_dist::PingReq" => match serde_json::from_slice::(payload) { @@ -1295,44 +1243,29 @@ impl IrohDriver { self.node .handle_ping_req(msg.from, msg.target, msg.sequence, &msg.piggyback) } - Err(e) => { - eprintln!("iroh driver: decode PingReq: {e}"); - Vec::new() - } + Err(_) => Vec::new(), }, "swactor_dist::JoinRequest" => { match serde_json::from_slice::(payload) { Ok(msg) => self.node.handle_join_request(msg.from), - Err(e) => { - eprintln!("iroh driver: decode JoinRequest: {e}"); - Vec::new() - } + Err(_) => Vec::new(), } } "swactor_dist::JoinResponse" => { match serde_json::from_slice::(payload) { Ok(msg) => self.node.handle_join_response(msg.members), - Err(e) => { - eprintln!("iroh driver: decode JoinResponse: {e}"); - Vec::new() - } + Err(_) => Vec::new(), } } "swactor_dist::IndirectAck" => match serde_json::from_slice::(payload) { Ok(msg) => self.node.handle_indirect_ack(msg.target, msg.sequence, &msg.piggyback), - Err(e) => { - eprintln!("iroh driver: decode IndirectAck: {e}"); - Vec::new() - } + Err(_) => Vec::new(), }, - other => { - eprintln!("iroh driver: unknown message type: {other}"); - Vec::new() - } + _ => Vec::new(), } } diff --git a/examples/pipeline-parallel-inference/Cargo.lock b/examples/pipeline-parallel-inference/Cargo.lock index 90a2d8d..0e7ba02 100644 --- a/examples/pipeline-parallel-inference/Cargo.lock +++ b/examples/pipeline-parallel-inference/Cargo.lock @@ -634,6 +634,7 @@ dependencies = [ "axum", "crossbeam-queue", "ctrlc", + "distribution", "serde", "serde_json", "swactor", @@ -2655,6 +2656,7 @@ dependencies = [ "base64", "dashboard", "distribution", + "futures-util", "iroh", "libc", "reqwest 0.12.28", @@ -2935,6 +2937,7 @@ dependencies = [ "bytes", "encoding_rs", "futures-core", + "futures-util", "h2", "http", "http-body", @@ -2956,12 +2959,14 @@ dependencies = [ "sync_wrapper", "tokio", "tokio-native-tls", + "tokio-util", "tower", "tower-http", "tower-service", "url", "wasm-bindgen", "wasm-bindgen-futures", + "wasm-streams 0.4.2", "web-sys", ] @@ -2998,7 +3003,7 @@ dependencies = [ "url", "wasm-bindgen", "wasm-bindgen-futures", - "wasm-streams", + "wasm-streams 0.5.0", "web-sys", ] @@ -4227,6 +4232,19 @@ dependencies = [ "wasmparser", ] +[[package]] +name = "wasm-streams" +version = "0.4.2" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "15053d8d85c7eccdbefef60f06769760a563c7f0a9d6902a13d35c7800b0ad65" +dependencies = [ + "futures-util", + "js-sys", + "wasm-bindgen", + "wasm-bindgen-futures", + "web-sys", +] + [[package]] name = "wasm-streams" version = "0.5.0" diff --git a/examples/pipeline-parallel-inference/Cargo.toml b/examples/pipeline-parallel-inference/Cargo.toml index 745f0a3..e13627f 100644 --- a/examples/pipeline-parallel-inference/Cargo.toml +++ b/examples/pipeline-parallel-inference/Cargo.toml @@ -11,24 +11,27 @@ swactor = { path = "../..", features = ["transport", "serde", "std"] } swactor-process = { path = "../../crates/process" } serde = { version = "1", features = ["derive"] } serde_json = "1" -reqwest = { version = "0.12", features = ["json"] } +reqwest = { version = "0.12", features = ["json", "stream"] } +futures-util = "0.3" tokio = { version = "1", features = ["full"] } distribution = { path = "../../crates/distribution", features = ["iroh", "collector"] } -# Live runtime dashboard (default features = HTTP overview/actors/topology pages, -# served on localhost when PP_DASHBOARD is set). No tui/replay-viewer pulled in. -dashboard = { path = "../../crates/dashboard" } +# Live runtime dashboard (HTTP overview/actors/topology pages, served on +# localhost when PP_DASHBOARD is set). The `live-collector` feature pulls in the +# server-side vast.ai fold (`VastaiLivePlugin`/`fold`) so the orchestrator can +# fold the collector's record stream into the Fleet tab. No tui/replay pulled in. +dashboard = { path = "../../crates/dashboard", features = ["live-collector"] } iroh = "0.98" urlencoding = "2" base64 = "0.22" libc = "0.2" [[bin]] -name = "pp-gpu-node" -path = "src/bin/pp_gpu_node.rs" +name = "pp-worker" +path = "src/bin/pp_worker.rs" [[bin]] -name = "pp-smoke-run" -path = "src/bin/pp_smoke_run.rs" +name = "pp-orchestrator" +path = "src/bin/pp_orchestrator.rs" [dev-dependencies] wiremock = "0.6" diff --git a/examples/pipeline-parallel-inference/Dockerfile b/examples/pipeline-parallel-inference/Dockerfile index 705fda0..a736533 100644 --- a/examples/pipeline-parallel-inference/Dockerfile +++ b/examples/pipeline-parallel-inference/Dockerfile @@ -23,8 +23,8 @@ FROM ${BASE_IMAGE} # Pipeline binaries (this crate's target/) + diagnostics binaries (the # workspace-root target/). All are statically linked enough that the base # stage's libc is all they need; the worker is pure Python. -COPY examples/pipeline-parallel-inference/target/release/pp-gpu-node /usr/local/bin/pp-gpu-node -COPY examples/pipeline-parallel-inference/target/release/pp-smoke-run /usr/local/bin/pp-smoke-run +COPY examples/pipeline-parallel-inference/target/release/pp-worker /usr/local/bin/pp-worker +COPY examples/pipeline-parallel-inference/target/release/pp-orchestrator /usr/local/bin/pp-orchestrator COPY target/release/swactor-diag-collector /usr/local/bin/swactor-diag-collector COPY target/release/swactor-diag-postproc /usr/local/bin/swactor-diag-postproc COPY examples/pipeline-parallel-inference/pp_tinygrad_worker.py /usr/local/share/pp_tinygrad_worker.py diff --git a/examples/pipeline-parallel-inference/PP_N12_DEPLOY_SESSION_REPORT.md b/examples/pipeline-parallel-inference/PP_N12_DEPLOY_SESSION_REPORT.md deleted file mode 100644 index 79c02f2..0000000 --- a/examples/pipeline-parallel-inference/PP_N12_DEPLOY_SESSION_REPORT.md +++ /dev/null @@ -1,49 +0,0 @@ -# PP N=12 Deploy — Session Report - -## Problems encountered (trivial → blocking) -- **GPU filter too narrow** — exact `RTX 3060` match: stage 5 hit "no offers" mid-chain. Fixed by VRAM-based selection (`PP_GPU_MIN_RAM_MB`). -- **Offer churn** — `no_such_ask`: offers vanish between search and create. -- **API rate-limit (HTTP 429)** — `provision_stage` retried with zero backoff + no inter-stage pacing, burning the candidate pool. Fixed: 429 backoff + `PP_LEASE_PACE_MS`. -- **Worker crash `Code(2)` on all 12 stages** — two root causes, both invisible at first: - - `Tensor(str)` rejected by bundled tinygrad → needed `Tensor(Path(...))`. - - Image shipped only 4 CUDA headers; NVRTC needs the full set (`vector_types.h`). Dockerfile copied a hand-picked subset. -- **Worker stderr swallowed** — StageActor buffers it into a `worker_exit_detail` diag event that needs a collector; none configured → error vanished. -- **SSH auth** — `publickey` denied initially (propagation lag), then worked on 5/12 but **persistently failed on 7/12** (pp-gpu-node holds PID 1 via `exec`; vast key-injection never ran). `vastai attach`/reboot didn't fix it. -- **`vastai execute` unusable** — "Invalid command given" (restricted command set). -- **`--redeploy` blocked** — uses the same SSH/scp, so unusable on the 7 unreachable nodes. -- **Slow/stalled image pulls** — cheap Korea GTX-10-series hosts; one fully stalled (0 bytes), triggering Phase-2 **autoreplace churn**. -- **Autoreplace not disableable** in the running binary. -- **429 on relaunch** — teardown's 12 destroys consumed the budget; absorbed by the new backoff. - -## Where we spent the most time -1. **~20 min blind on the silent resolve loop** — connect-timeout SWIM noise looked like the problem but was a red herring; workers had actually crashed instantly. -2. **Getting on a node to see the real error** — SSH flakiness, restricted `execute`, local docker repro, then manual on-node run. -3. **Run #2 image-pull waiting** — many heartbeat ticks on slow/stalled pulls + the re-download after teardown. - -## Observability that was clunky / insufficient -- Worker stderr + Python traceback never reach the container log (no collector) — had to reproduce locally and SSH a node to see `Code(2)`'s cause. *(Fixed: pp-gpu-node now prints abnormal-exit stderr.)* -- Resolve loop emits **nothing per-stage** — orchestrator log is just SWIM gossip for up to 20 min; no per-stage worker-ready/download visibility. -- The rich SSE diag stream (`pp_download_progress`, `worker_exit_detail`) was dead — `SWACTOR_DIAG_COLLECTOR_URL` unset. -- vast exposes **no docker-pull %** — `status_msg` only says "Pulling from"; `disk_usage` = -1. -- Connect-timeout logs were prominent but cosmetic — actively misleading. -- Node-id→stage mapping had to be derived by hand from `stage_secrets`. - -## Where interaction with the live deployment was limited -- SSH worked on only 5/12 nodes; no reliable shell on the rest. -- `vastai execute` restricted; couldn't run arbitrary diagnostics via API. -- `--redeploy` (the intended fix-forward path) depends on the same broken SSH → fix-forward on live nodes was effectively impossible; had to rebuild the image + re-lease. -- Couldn't pause/disable autoreplace or see/intervene in image-pull progress. -- During the "loading" (pull) phase there's no container, so no SSH at all on the node that mattered most. - -## Other notes -- **Sharded fetch works** (~1.8 GB/stage, not 18 GB) — but a stale code comment claims the full GGUF is pulled, which misled diagnosis. -- Core bugs are fixed + validated on a real GPU (stage 0 → `ready`) and in the pushed image; the remaining blocker is purely **host quality** (slow-pull hosts), not code. -- Re-leasing fresh always re-pulls image + re-downloads model; the in-place cache advantage is lost on every teardown. -- Highest-leverage follow-ups: (1) configure a diagnostics collector, (2) emit per-stage resolve/download progress to the orchestrator log, (3) host-throughput preflight or stalled-pull fast-replace, (4) fix the onstart so vast SSH-key injection survives (don't `exec` over it). - -## Fixes shipped this session -- `vastai.rs`: 429 backoff in `provision_stage` (find + create paths) and inter-stage pacing (`PP_LEASE_PACE_MS`, default 600ms). -- `pp_tinygrad_worker.py`: `Tensor(gguf_path)` → `Tensor(Path(gguf_path))`. -- `Dockerfile`: copy the full CUDA include set (with a `test -f vector_types.h` build guard) instead of 4 hand-picked headers. -- `stage_actor.rs`: mirror an abnormal worker exit's stderr tail + Python traceback to pp-gpu-node's own stderr (→ container log, collector-independent). -- Image rebuilt + pushed (`zacheryasc/swactor-pp-gpu:latest`, digest `ca373d02…`); both bug fixes validated on a real GPU node (stage 0 reached `ready`). diff --git a/examples/pipeline-parallel-inference/docker-compose.diag.yml b/examples/pipeline-parallel-inference/docker-compose.diag.yml index 355df21..a4e356d 100644 --- a/examples/pipeline-parallel-inference/docker-compose.diag.yml +++ b/examples/pipeline-parallel-inference/docker-compose.diag.yml @@ -3,14 +3,14 @@ # The collector service runs in its own container with the bundles dir # bind-mounted from the host so the harness can read the finalized # tarball. Both the collector and the pp processes use the host network -# namespace, so the stage children (spawned by `pp-smoke-run` via +# namespace, so the stage children (spawned by `pp-orchestrator` via # `docker-gpu-node.sh`) share localhost reachability with the # collector — `SWACTOR_DIAG_COLLECTOR_URL=http://127.0.0.1:9080` works # uniformly from every actor in the run. # # The host network choice mirrors the existing `docker-e2e.sh` shape: -# pp-smoke-run runs on the host (orchestrator) and each -# `pp-gpu-node` runs in its own container under `--network host`. The +# pp-orchestrator runs on the host (orchestrator) and each +# `pp-worker` runs in its own container under `--network host`. The # collector container just adds one more service to that arrangement. # # Used by `scripts/docker-diag-e2e.sh`. Direct `docker compose up` @@ -24,7 +24,7 @@ services: container_name: ${PP_DIAG_COLLECTOR_NAME:-pp-diag-collector} network_mode: host # Override the image's default entrypoint (pp_entrypoint.sh, which runs - # pp-gpu-node) so this container runs the collector instead. The + # pp-worker) so this container runs the collector instead. The # diagnostics binaries ship in the same unified code image. entrypoint: /usr/local/bin/swactor-diag-collector command: diff --git a/examples/pipeline-parallel-inference/pp_entrypoint.sh b/examples/pipeline-parallel-inference/pp_entrypoint.sh index 28a2bfc..8ee2fbe 100644 --- a/examples/pipeline-parallel-inference/pp_entrypoint.sh +++ b/examples/pipeline-parallel-inference/pp_entrypoint.sh @@ -1,7 +1,7 @@ #!/usr/bin/env bash # PID-1 supervisor for the pipeline-parallel runtime container. # -# The worker used to run as PID 1 (`exec pp-gpu-node`), so SSH depended on +# The worker used to run as PID 1 (`exec pp-worker`), so SSH depended on # vast's host-side helper winning a race against our exec, and a worker crash # killed the whole container — no shell left to read the traceback. This script # instead owns PID 1: it brings up sshd deterministically, runs the worker as a @@ -41,9 +41,9 @@ echo "pp-entrypoint: sshd up on :22" >&2 # ── Worker: run as a child, tee output to a file readable over SSH ─────────── WORKER_LOG=/var/log/pp-worker.log -echo "pp-entrypoint: launching pp-gpu-node (log -> $WORKER_LOG)" >&2 +echo "pp-entrypoint: launching pp-worker (log -> $WORKER_LOG)" >&2 set -o pipefail -/usr/local/bin/pp-gpu-node 2>&1 | tee "$WORKER_LOG" +/usr/local/bin/pp-worker 2>&1 | tee "$WORKER_LOG" code=${PIPESTATUS[0]} # ── Operator runbooks (manual, over SSH) ───────────────────────────────────── @@ -51,23 +51,23 @@ code=${PIPESTATUS[0]} # # Worker hot-reload (no restart) — edit the Python in place, then SIGHUP: # scp -P pp_tinygrad_worker.py root@:/usr/local/share/pp_tinygrad_worker.py -# ssh 'kill -HUP $(pidof pp-gpu-node)' -# pp-gpu-node tears down its worker and re-execs the on-disk script; the +# ssh 'kill -HUP $(pidof pp-worker)' +# pp-worker tears down its worker and re-execs the on-disk script; the # swactor process (and SWIM membership) stays up across the swap. # # Swactor-binary swap — stop the binary, stage the new one, re-exec under # PID 1's env (preserves STAGE/SEED_ADDR/PP_STAGE_SECRET so the node id is # unchanged). `.new` staging avoids ETXTBSY on the mapped ELF: -# ssh 'pkill -x pp-gpu-node' # drops to the hold below -# scp -P pp-gpu-node root@:/usr/local/bin/pp-gpu-node.new -# ssh 'mv -f /usr/local/bin/pp-gpu-node.new /usr/local/bin/pp-gpu-node && \ -# chmod +x /usr/local/bin/pp-gpu-node && \ +# ssh 'pkill -x pp-worker' # drops to the hold below +# scp -P pp-worker root@:/usr/local/bin/pp-worker.new +# ssh 'mv -f /usr/local/bin/pp-worker.new /usr/local/bin/pp-worker && \ +# chmod +x /usr/local/bin/pp-worker && \ # setsid bash -c "while IFS= read -r -d \"\" kv; do export \"\$kv\"; done \ -# < /proc/1/environ; exec /usr/local/bin/pp-gpu-node" \ +# < /proc/1/environ; exec /usr/local/bin/pp-worker" \ # >/var/log/pp-restart.log 2>&1 &2 +echo "pp-entrypoint: pp-worker exited with code $code; NOT restarting (node held for postmortem)" >&2 echo "pp-entrypoint: --- last 40 lines of $WORKER_LOG ---" >&2 tail -n 40 "$WORKER_LOG" >&2 || true exec sleep infinity diff --git a/examples/pipeline-parallel-inference/profiles/example.env b/examples/pipeline-parallel-inference/profiles/example.env index 53a2f17..19a60e2 100644 --- a/examples/pipeline-parallel-inference/profiles/example.env +++ b/examples/pipeline-parallel-inference/profiles/example.env @@ -4,7 +4,7 @@ # across the binaries and one operator's shell. Copy it to set up a run: # # cp profiles/example.env profiles/local.env # untracked; put real secrets here -# PP_PROFILE=profiles/local.env cargo run --bin pp-smoke-run -- --seed ... +# PP_PROFILE=profiles/local.env cargo run --bin pp-orchestrator -- --seed ... # # When PP_PROFILE is unset, profiles/local.env is loaded automatically if it # exists. The run scripts pick this up too (the binaries load it at startup). @@ -25,27 +25,27 @@ #SWACTOR_IROH_RELAY_URL=https://relay.example.com # --- Topology --------------------------------------------------------------- -# Number of pipeline stages (>= 2). pp-smoke-run also accepts --num-stages. +# Number of pipeline stages (>= 2). pp-orchestrator also accepts --num-stages. #NUM_STAGES=2 # --- Compute target (vast.ai / docker) -------------------------------------- # Container image to run on each node. The run scripts already read PP_IMAGE; -# pp-smoke-run's --image default now reads it too. Set to your registry tag. +# pp-orchestrator's --image default now reads it too. Set to your registry tag. #PP_IMAGE=swactor-pp-gpu:latest -# GPU class requested when leasing on vast.ai (pp-smoke-run --gpu overrides). +# GPU class requested when leasing on vast.ai (pp-orchestrator --gpu overrides). #PP_GPU=RTX 3060 # --- Workload (what each stage computes) ------------------------------------ # Model identifier handed to the worker. #MODEL= # Python worker script + interpreter (per-stage compute). WORKER_SCRIPT is the -# pp-gpu-node default; pp-smoke-run --worker ships a script to remote nodes. +# pp-worker default; pp-orchestrator --worker ships a script to remote nodes. #WORKER_SCRIPT=./pp_tinygrad_worker.py #WORKER_CMD=python3 #PYTHON=python3 # Stub mode: skip the real worker, echo activations (fast local smoke runs). #PP_WORKER_STUB=1 -# Inference request prompt + token budget (pp-smoke-run --prompt/--max-tokens). +# Inference request prompt + token budget (pp-orchestrator --prompt/--max-tokens). #MAX_TOKENS=64 # --- Timeouts (seconds; sane defaults baked in — override only if needed) --- diff --git a/examples/pipeline-parallel-inference/profiles/local-seed.env b/examples/pipeline-parallel-inference/profiles/local-seed.env index c159d47..20e3b2f 100644 --- a/examples/pipeline-parallel-inference/profiles/local-seed.env +++ b/examples/pipeline-parallel-inference/profiles/local-seed.env @@ -1,7 +1,7 @@ # Local seed-mode smoke run: everything on loopback, stub workers, no relay. # Drives one request through a 2-stage chain without a GPU or external relay. # -# PP_PROFILE=profiles/local-seed.env cargo run --bin pp-smoke-run -- --seed +# PP_PROFILE=profiles/local-seed.env cargo run --bin pp-orchestrator -- --seed # # (No SWACTOR_IROH_RELAY_URL: seed mode wires nodes via direct loopback addrs.) diff --git a/examples/pipeline-parallel-inference/profiles/vastai.env b/examples/pipeline-parallel-inference/profiles/vastai.env index 735d840..306536a 100644 --- a/examples/pipeline-parallel-inference/profiles/vastai.env +++ b/examples/pipeline-parallel-inference/profiles/vastai.env @@ -1,7 +1,7 @@ # WAN run on leased vast.ai GPUs. Copy to profiles/local.env and fill in the # relay URL + image; keep that copy untracked. # -# PP_PROFILE=profiles/local.env cargo run --bin pp-smoke-run -- \ +# PP_PROFILE=profiles/local.env cargo run --bin pp-orchestrator -- \ # --vastai --api-key "$VASTAI_API_KEY" # # All nodes home onto one operator-controlled relay so they can hole-punch / @@ -13,6 +13,10 @@ SWACTOR_IROH_RELAY_URL=https://relay.example.com NUM_STAGES=2 PP_IMAGE=swactor-pp-gpu:latest -PP_GPU=RTX 3060 -MODEL= +# GPU filters are optional. Leave both unset to let any verified 1-GPU offer +# qualify (a 1B model fits anywhere). Set PP_GPU="RTX 3060" to pin a model, and/or +# PP_GPU_MIN_RAM_MB=8000 to require a VRAM floor. +# llama3.2:1b -> HF Llama-3.2-1B-Instruct-Q6_K.gguf (16 blocks, sharded across +# the N stages by compute_layer_range; validated key baked into the image). +MODEL=llama3.2:1b MAX_TOKENS=64 diff --git a/examples/pipeline-parallel-inference/scripts/demo-fleet.sh b/examples/pipeline-parallel-inference/scripts/demo-fleet.sh index 4b79e2f..43cbd84 100755 --- a/examples/pipeline-parallel-inference/scripts/demo-fleet.sh +++ b/examples/pipeline-parallel-inference/scripts/demo-fleet.sh @@ -1,17 +1,23 @@ #!/usr/bin/env bash # demo-fleet.sh — one-command local mock of a vast.ai fleet, watchable live. # -# Brings up, from a single command, a self-contained demo that streams the -# REAL host metrics each container measures (no synthetic data) into a live -# browser dashboard: +# Brings up, from a single command, a self-contained demo that mirrors the +# production topology: a diagnostics collector running "off-box" (in prod, a +# VPS) and the orchestrator running locally and hosting the FULL swactor +# dashboard. The orchestrator's dashboard shows its own live swactor process +# info (overview / actors / topology / distribution / netmap) and a Fleet tab +# that pulls the vast.ai + host metrics remotely from the collector: # -# - swactor-diag-collector on the HOST (HTTP 9080 + UDP 9081), serving the -# live fleet board at http://127.0.0.1:9080/dashboard -# - pp-smoke-run on the HOST in --seed mode, spawning N pp-gpu-node -# containers (one per stage) via docker-gpu-node.sh, each on --network host +# - swactor-diag-collector on the HOST (HTTP 9080 + UDP 9081) — the "VPS" +# sink. Each stage's in-VM monitor ships REAL host_sample + log records +# (no synthetic data) here; the orchestrator pushes its distribution +# snapshot here too. Its own fleet board stays at http://127.0.0.1:9080/dashboard +# - pp-orchestrator on the HOST in --seed mode (PP_DASHBOARD on), spawning N +# pp-worker containers (one per stage) via docker-gpu-node.sh, each on +# --network host, and serving the full dashboard at http://127.0.0.1:9095/ # - PP_HOLD=1 keeps the cluster up after the first drive, so every stage's -# in-VM monitor keeps shipping host_sample + log records (~every 5s) and -# the dashboard animates in real time. +# in-VM monitor keeps shipping records (~every 5s) and the dashboard +# animates in real time. # # Ctrl+C (or any exit) tears everything down: stage containers, collector, # orchestrator, and all temp files. @@ -31,6 +37,7 @@ # PP_MAX_TOKENS decode token cap (default: 4) # PP_BIND_HOST collector bind host (default: 127.0.0.1) # PP_PORT collector HTTP port (default: 9080) +# PP_DASHBOARD_PORT orchestrator dashboard HTTP port (default: 9095) # PP_NO_OPEN if set, don't try to open the dashboard in a browser # PP_DIAG_NETWORK docker network for stages (default: host) set -euo pipefail @@ -43,9 +50,13 @@ MAX_TOKENS="${PP_MAX_TOKENS:-4}" BIND_HOST="${PP_BIND_HOST:-127.0.0.1}" PORT="${PP_PORT:-9080}" UDP_PORT=$((PORT + 1)) +DASH_PORT="${PP_DASHBOARD_PORT:-9095}" CONTAINER_PREFIX="demo-fleet-stage" RUN_ID="demo-fleet-$(date +%s)" -DASH_URL="http://${BIND_HOST}:${PORT}/dashboard" +# The full swactor dashboard is served by the orchestrator at "/"; the collector +# keeps its own standalone fleet board at :PORT/dashboard. +DASH_URL="http://${BIND_HOST}:${DASH_PORT}/" +COLLECTOR_URL="http://${BIND_HOST}:${PORT}" if ! [[ "$NUM_STAGES" =~ ^[0-9]+$ ]] || [ "$NUM_STAGES" -lt 2 ]; then echo "demo-fleet: N must be an integer >= 2 (seed mode needs >=2 stages), got '$NUM_STAGES'" >&2 @@ -57,26 +68,36 @@ fi if ! docker info >/dev/null 2>&1; then echo "demo-fleet: docker daemon unreachable" >&2; exit 2 fi +# Fail loudly on a clash for the orchestrator dashboard port. Its HTTP server is +# spawned on the driver's tokio runtime and `.expect()`s its bind; a collision +# panics that task silently and the run carries on with no dashboard. Catch it +# here so the user can pick a free one. +if (exec 3<>"/dev/tcp/${BIND_HOST}/${DASH_PORT}") 2>/dev/null; then + exec 3>&- 3<&- + echo "demo-fleet: dashboard port ${DASH_PORT} is already in use." \ + "Pick a free one: PP_DASHBOARD_PORT=9096 $0 ${NUM_STAGES}" >&2 + exit 2 +fi SCRIPT_DIR="$(cd "$(dirname "$0")" && pwd)" CRATE_DIR="$(cd "$SCRIPT_DIR/.." && pwd)" WORKSPACE_DIR="$(cd "$CRATE_DIR/../.." && pwd)" -SMOKE_RUN_BIN="$CRATE_DIR/target/release/pp-smoke-run" -GPU_NODE_BIN="$CRATE_DIR/target/release/pp-gpu-node" +ORCHESTRATOR_BIN="$CRATE_DIR/target/release/pp-orchestrator" +WORKER_BIN="$CRATE_DIR/target/release/pp-worker" WORKER_PY="$CRATE_DIR/pp_tinygrad_worker.py" COLLECTOR_BIN="$WORKSPACE_DIR/target/release/swactor-diag-collector" # ── Step 1: build release artifacts ─────────────────────────────────────── if [ -z "${PP_SKIP_BUILD:-}" ]; then - echo "demo-fleet: cargo build pp-smoke-run + pp-gpu-node (release)" + echo "demo-fleet: cargo build pp-orchestrator + pp-worker (release)" cargo build --manifest-path "$CRATE_DIR/Cargo.toml" --release \ - --bin pp-gpu-node --bin pp-smoke-run + --bin pp-worker --bin pp-orchestrator echo "demo-fleet: cargo build swactor-diag-collector (release, --features collector)" cargo build --manifest-path "$WORKSPACE_DIR/Cargo.toml" --release \ -p distribution --features collector --bin swactor-diag-collector fi -for f in "$SMOKE_RUN_BIN" "$GPU_NODE_BIN" "$WORKER_PY" "$COLLECTOR_BIN"; do +for f in "$ORCHESTRATOR_BIN" "$WORKER_BIN" "$WORKER_PY" "$COLLECTOR_BIN"; do [ -f "$f" ] || { echo "demo-fleet: missing $f (run without PP_SKIP_BUILD)" >&2; exit 1; } done @@ -129,7 +150,7 @@ cleanup() { trap cleanup EXIT trap 'exit 130' INT TERM -# ── Step 3: collector on the host (serves the dashboard) ─────────────────── +# ── Step 3: collector on the host (the off-box "VPS" metrics sink) ───────── echo "demo-fleet: starting collector on ${BIND_HOST}:${PORT} (root=$COLLECTOR_ROOT)" "$COLLECTOR_BIN" --bind "${BIND_HOST}:${PORT}" --root "$COLLECTOR_ROOT" \ --udp "${BIND_HOST}:${UDP_PORT}" >"$COLLECTOR_LOG" 2>&1 & @@ -148,38 +169,31 @@ until (echo > "/dev/tcp/${BIND_HOST}/${PORT}") >/dev/null 2>&1; do done echo "demo-fleet: collector ready" -# ── Step 4: announce + open the dashboard ────────────────────────────────── -echo -echo " ┌─────────────────────────────────────────────────────────────┐" -echo " │ Live fleet dashboard: $DASH_URL" -echo " └─────────────────────────────────────────────────────────────┘" -echo -if [ -z "${PP_NO_OPEN:-}" ]; then - if command -v xdg-open >/dev/null 2>&1; then (xdg-open "$DASH_URL" >/dev/null 2>&1 &) || true - elif command -v open >/dev/null 2>&1; then (open "$DASH_URL" >/dev/null 2>&1 &) || true - fi -fi - -# ── Step 5: orchestrator (holds the cluster open, real metrics enabled) ───── +# ── Step 4: orchestrator (hosts the full dashboard, holds the cluster open) ─ # stdin is the FIFO; we hold its write end open on fd 3 so hold_open() never # sees EOF and the cluster stays up until we tear down. exec 3<>"$FIFO" echo "demo-fleet: launching orchestrator + ${NUM_STAGES} stage containers (run_id=$RUN_ID)" # The orchestrator (and, via inheritance, docker-gpu-node.sh) read these from -# the environment. PP_HOLD keeps the cluster up; the SWACTOR_DIAG_* vars enable -# each stage's in-VM monitor so real host metrics ship to the collector. +# the environment. PP_HOLD keeps the cluster up; PP_DASHBOARD makes the +# orchestrator host the full swactor dashboard locally; the SWACTOR_DIAG_* vars +# point each stage's in-VM monitor at the collector (the off-box sink) and give +# the orchestrator the same URL to push its distribution snapshot to and to pull +# the fleet model from for its Fleet tab. export PP_HOLD=1 export PP_WORKER_STUB=1 export PP_DEV=CPU export PP_IMAGE="$IMAGE" export PP_CONTAINER_PREFIX="$CONTAINER_PREFIX" -export SWACTOR_DIAG_COLLECTOR_URL="http://${BIND_HOST}:${PORT}" +export PP_DASHBOARD=1 +export PP_DASHBOARD_PORT="$DASH_PORT" +export SWACTOR_DIAG_COLLECTOR_URL="$COLLECTOR_URL" export SWACTOR_DIAG_RUN_ID="$RUN_ID" export SWACTOR_DIAG_SPOOL_DIR="$SPOOL_DIR" export SWACTOR_DIAG_UDP_ECHO="${BIND_HOST}:${UDP_PORT}" [ -n "${PP_GPUS:-}" ] && export PP_GPUS [ -n "${PP_DIAG_NETWORK:-}" ] && export PP_DIAG_NETWORK -"$SMOKE_RUN_BIN" \ +"$ORCHESTRATOR_BIN" \ --seed \ --num-stages "$NUM_STAGES" \ --gpu-node "$SCRIPT_DIR/docker-gpu-node.sh" \ @@ -189,6 +203,31 @@ export SWACTOR_DIAG_UDP_ECHO="${BIND_HOST}:${UDP_PORT}" <"$FIFO" >"$ORCH_LOG" 2>&1 & ORCH_PID=$! +# ── Step 5: wait for the orchestrator's dashboard to bind, then announce + open +WAITED=0 +until (echo > "/dev/tcp/${BIND_HOST}/${DASH_PORT}") >/dev/null 2>&1; do + if ! kill -0 "$ORCH_PID" >/dev/null 2>&1; then + echo "demo-fleet: orchestrator exited before its dashboard came up." >&2 + tail -n 40 "$ORCH_LOG" >&2 || true + exit 1 + fi + WAITED=$((WAITED + 1)) + [ "$WAITED" -ge 30 ] && { echo "demo-fleet: orchestrator dashboard did not bind :${DASH_PORT} in 30s" >&2; tail -n 40 "$ORCH_LOG" >&2; exit 1; } + sleep 1 +done +echo +echo " ┌─────────────────────────────────────────────────────────────┐" +echo " │ Full swactor dashboard: $DASH_URL" +echo " │ (overview / actors / topology / distribution / netmap / fleet)" +echo " │ Collector fleet board: ${COLLECTOR_URL}/dashboard" +echo " └─────────────────────────────────────────────────────────────┘" +echo +if [ -z "${PP_NO_OPEN:-}" ]; then + if command -v xdg-open >/dev/null 2>&1; then (xdg-open "$DASH_URL" >/dev/null 2>&1 &) || true + elif command -v open >/dev/null 2>&1; then (open "$DASH_URL" >/dev/null 2>&1 &) || true + fi +fi + # ── Step 6: wait until the cluster is converged + held open ──────────────── echo "demo-fleet: waiting for the cluster to converge (first inference drive)…" WAITED=0 diff --git a/examples/pipeline-parallel-inference/scripts/docker-dashboard-e2e.sh b/examples/pipeline-parallel-inference/scripts/docker-dashboard-e2e.sh index f56fe5c..d49654c 100755 --- a/examples/pipeline-parallel-inference/scripts/docker-dashboard-e2e.sh +++ b/examples/pipeline-parallel-inference/scripts/docker-dashboard-e2e.sh @@ -2,7 +2,7 @@ # docker-dashboard-e2e.sh — the docker-e2e run, held open under the live # swactor dashboard, one dashboard PER STAGE. # -# Brings up `N` stub-mode `pp-gpu-node` containers on localhost and drives +# Brings up `N` stub-mode `pp-worker` containers on localhost and drives # one InferenceRequest through them, exactly like `docker-e2e.sh` — but each # stage serves the live swactor dashboard (PP_STAGE_DASHBOARD) and the # orchestrator HOLDS after the drive (PP_HOLD). The stage containers run on @@ -78,8 +78,8 @@ done SCRIPT_DIR="$(cd "$(dirname "$0")" && pwd)" CRATE_DIR="$(cd "$SCRIPT_DIR/.." && pwd)" WORKSPACE_DIR="$(cd "$CRATE_DIR/../.." && pwd)" -SMOKE_RUN_BIN="$CRATE_DIR/target/release/pp-smoke-run" -GPU_NODE_BIN="$CRATE_DIR/target/release/pp-gpu-node" +ORCHESTRATOR_BIN="$CRATE_DIR/target/release/pp-orchestrator" +WORKER_BIN="$CRATE_DIR/target/release/pp-worker" WORKER_PY="$CRATE_DIR/pp_tinygrad_worker.py" COLLECTOR_BIN="$WORKSPACE_DIR/target/release/swactor-diag-collector" POSTPROC_BIN="$WORKSPACE_DIR/target/release/swactor-diag-postproc" @@ -87,20 +87,20 @@ POSTPROC_BIN="$WORKSPACE_DIR/target/release/swactor-diag-postproc" # Step 1: build the release artifacts the docker image packages (same set # docker-e2e.sh builds — the image's COPY needs the diag binaries present). if [ -z "${PP_SKIP_BUILD:-}" ]; then - echo "docker-dashboard-e2e: building pp-gpu-node + pp-smoke-run (release)" + echo "docker-dashboard-e2e: building pp-worker + pp-orchestrator (release)" cargo build --manifest-path "$CRATE_DIR/Cargo.toml" --release \ - --bin pp-gpu-node --bin pp-smoke-run + --bin pp-worker --bin pp-orchestrator echo "docker-dashboard-e2e: building swactor-diag-{collector,postproc} (release)" cargo build --manifest-path "$WORKSPACE_DIR/Cargo.toml" --release \ -p distribution --features collector \ --bin swactor-diag-collector --bin swactor-diag-postproc fi -for f in "$SMOKE_RUN_BIN" "$GPU_NODE_BIN" "$WORKER_PY" "$COLLECTOR_BIN" "$POSTPROC_BIN"; do +for f in "$ORCHESTRATOR_BIN" "$WORKER_BIN" "$WORKER_PY" "$COLLECTOR_BIN" "$POSTPROC_BIN"; do [ -f "$f" ] || { echo "docker-dashboard-e2e: missing $f" >&2; exit 1; } done # Step 2: build the layered image (heavy CUDA base, then thin code layer). -# The stage dashboard lives in the pp-gpu-node binary baked into this image, +# The stage dashboard lives in the pp-worker binary baked into this image, # so a stale image without it will show nothing — rebuild unless you know the # current image already carries the dashboard-enabled binary. if [ -z "${PP_SKIP_IMAGE_BUILD:-}" ]; then @@ -131,7 +131,7 @@ for ((k = 0; k < NUM_STAGES; k++)); do echo " stage ${k}: http://localhost:$((PORT_BASE + k))" done -# Step 4: drive pp-smoke-run with the docker shim. The orchestrator serves its +# Step 4: drive pp-orchestrator with the docker shim. The orchestrator serves its # own dashboard (PP_DASHBOARD) — including the live SWIM distribution graph and # message tallies — and each stage serves its own (PP_STAGE_DASHBOARD). PP_HOLD # makes the orchestrator block at the end, ticking the driver so the @@ -146,7 +146,7 @@ PP_DASHBOARD=1 \ PP_DASHBOARD_PORT="$ORCH_PORT" \ PP_STAGE_DASHBOARD=1 \ PP_STAGE_DASHBOARD_PORT_BASE="$PORT_BASE" \ -"$SMOKE_RUN_BIN" \ +"$ORCHESTRATOR_BIN" \ --seed \ --num-stages "$NUM_STAGES" \ --gpu-node "$SCRIPT_DIR/docker-gpu-node.sh" \ diff --git a/examples/pipeline-parallel-inference/scripts/docker-diag-e2e.sh b/examples/pipeline-parallel-inference/scripts/docker-diag-e2e.sh index 97f468d..5a7c588 100755 --- a/examples/pipeline-parallel-inference/scripts/docker-diag-e2e.sh +++ b/examples/pipeline-parallel-inference/scripts/docker-diag-e2e.sh @@ -3,7 +3,7 @@ # # Brings up: # - `swactor-diag-collector` in a container (HTTP 9080 + UDP 9081) -# - `pp-smoke-run` on the host, in seed mode with N stub-mode stage +# - `pp-orchestrator` on the host, in seed mode with N stub-mode stage # children spawned via `docker-gpu-node.sh` (each its own # container on `--network host`) # @@ -95,8 +95,8 @@ CRATE_DIR="$(cd "$SCRIPT_DIR/.." && pwd)" WORKSPACE_DIR="$(cd "$CRATE_DIR/../.." && pwd)" COMPOSE_FILE="$CRATE_DIR/docker-compose.diag.yml" -SMOKE_RUN_BIN="$CRATE_DIR/target/release/pp-smoke-run" -GPU_NODE_BIN="$CRATE_DIR/target/release/pp-gpu-node" +ORCHESTRATOR_BIN="$CRATE_DIR/target/release/pp-orchestrator" +WORKER_BIN="$CRATE_DIR/target/release/pp-worker" WORKER_PY="$CRATE_DIR/pp_tinygrad_worker.py" COLLECTOR_BIN="$WORKSPACE_DIR/target/release/swactor-diag-collector" POSTPROC_BIN="$WORKSPACE_DIR/target/release/swactor-diag-postproc" @@ -104,15 +104,15 @@ POSTPROC_BIN="$WORKSPACE_DIR/target/release/swactor-diag-postproc" # Step 1: build release artifacts the image will package. The pp binaries # live in their own workspace; the distribution binaries live at the top. if [ -z "${PP_SKIP_BUILD:-}" ]; then - echo "docker-diag-e2e: cargo build pp-smoke-run + pp-gpu-node (release)" + echo "docker-diag-e2e: cargo build pp-orchestrator + pp-worker (release)" cargo build --manifest-path "$CRATE_DIR/Cargo.toml" --release \ - --bin pp-gpu-node --bin pp-smoke-run + --bin pp-worker --bin pp-orchestrator echo "docker-diag-e2e: cargo build swactor-diag-{collector,postproc} (release, --features collector)" cargo build --manifest-path "$WORKSPACE_DIR/Cargo.toml" --release \ -p distribution --features collector \ --bin swactor-diag-collector --bin swactor-diag-postproc fi -for f in "$SMOKE_RUN_BIN" "$GPU_NODE_BIN" "$WORKER_PY" "$COLLECTOR_BIN" "$POSTPROC_BIN"; do +for f in "$ORCHESTRATOR_BIN" "$WORKER_BIN" "$WORKER_PY" "$COLLECTOR_BIN" "$POSTPROC_BIN"; do [ -f "$f" ] || { echo "docker-diag-e2e: missing $f" >&2; exit 1; } done @@ -190,7 +190,7 @@ if [ "$USE_COMPOSE" = 1 ]; then "${COMPOSE[@]}" -f "$COMPOSE_FILE" up -d --remove-orphans collector else # --entrypoint runs the collector directly; the default image entrypoint - # (pp_entrypoint.sh) would ignore these args and launch pp-gpu-node. + # (pp_entrypoint.sh) would ignore these args and launch pp-worker. docker run -d --rm \ --name "$COLLECTOR_NAME" \ --network "$DIAG_NETWORK" \ @@ -223,7 +223,7 @@ until (echo > /dev/tcp/127.0.0.1/9080) >/dev/null 2>&1; do done echo "docker-diag-e2e: collector ready" -# Step 5: drive pp-smoke-run with diagnostics env vars set. Stage children +# Step 5: drive pp-orchestrator with diagnostics env vars set. Stage children # pick up the same vars via docker-gpu-node.sh's `-e` forwarders. OUTPUT_DIR="$(mktemp -d)" STDOUT_LOG="$OUTPUT_DIR/stdout.log" @@ -244,7 +244,7 @@ SWACTOR_DIAG_COLLECTOR_URL="http://127.0.0.1:9080" \ SWACTOR_DIAG_RUN_ID="$RUN_ID" \ SWACTOR_DIAG_SPOOL_DIR="$OUTPUT_DIR/spool" \ SWACTOR_DIAG_UDP_ECHO="127.0.0.1:9081" \ -"$SMOKE_RUN_BIN" \ +"$ORCHESTRATOR_BIN" \ --seed \ --num-stages "$NUM_STAGES" \ --gpu-node "$SCRIPT_DIR/docker-gpu-node.sh" \ @@ -252,20 +252,20 @@ SWACTOR_DIAG_UDP_ECHO="127.0.0.1:9081" \ --prompt "$PROMPT" \ --max-tokens "$MAX_TOKENS" \ >"$STDOUT_LOG" 2>"$STDERR_LOG" -SMOKE_STATUS=$? +ORCH_STATUS=$? set -e -if [ "$SMOKE_STATUS" -ne 0 ]; then - echo "docker-diag-e2e: pp-smoke-run exited $SMOKE_STATUS" >&2 +if [ "$ORCH_STATUS" -ne 0 ]; then + echo "docker-diag-e2e: pp-orchestrator exited $ORCH_STATUS" >&2 echo "----- stdout -----" >&2 cat "$STDOUT_LOG" >&2 echo "----- stderr (last 60) -----" >&2 tail -n 60 "$STDERR_LOG" >&2 exit 1 fi -echo "docker-diag-e2e: pp-smoke-run exited 0" +echo "docker-diag-e2e: pp-orchestrator exited 0" if [ -n "${PP_DIAG_VERBOSE:-}" ]; then - echo "----- pp-smoke-run stderr (last 30) -----" + echo "----- pp-orchestrator stderr (last 30) -----" tail -n 30 "$STDERR_LOG" fi @@ -295,7 +295,7 @@ while true; do else docker logs "$COLLECTOR_NAME" 2>&1 | tail -n 50 >&2 || true fi - echo "----- pp-smoke-run stderr (last 60) -----" >&2 + echo "----- pp-orchestrator stderr (last 60) -----" >&2 tail -n 60 "$STDERR_LOG" >&2 || true exit 1 fi diff --git a/examples/pipeline-parallel-inference/scripts/docker-e2e.sh b/examples/pipeline-parallel-inference/scripts/docker-e2e.sh index fa5889b..84039a4 100755 --- a/examples/pipeline-parallel-inference/scripts/docker-e2e.sh +++ b/examples/pipeline-parallel-inference/scripts/docker-e2e.sh @@ -1,8 +1,8 @@ #!/usr/bin/env bash # docker-e2e.sh — the Stage 11 pre-deploy gate. # -# Brings up `N` stub-mode `pp-gpu-node` containers on localhost, drives -# one InferenceRequest through them via `pp-smoke-run --seed`, and tears +# Brings up `N` stub-mode `pp-worker` containers on localhost, drives +# one InferenceRequest through them via `pp-orchestrator --seed`, and tears # everything down. The image is built locally from the workspace's # release artifacts; no GPU, no tinygrad, no GGUF required. # @@ -63,8 +63,8 @@ fi SCRIPT_DIR="$(cd "$(dirname "$0")" && pwd)" CRATE_DIR="$(cd "$SCRIPT_DIR/.." && pwd)" WORKSPACE_DIR="$(cd "$CRATE_DIR/../.." && pwd)" -SMOKE_RUN_BIN="$CRATE_DIR/target/release/pp-smoke-run" -GPU_NODE_BIN="$CRATE_DIR/target/release/pp-gpu-node" +ORCHESTRATOR_BIN="$CRATE_DIR/target/release/pp-orchestrator" +WORKER_BIN="$CRATE_DIR/target/release/pp-worker" WORKER_PY="$CRATE_DIR/pp_tinygrad_worker.py" COLLECTOR_BIN="$WORKSPACE_DIR/target/release/swactor-diag-collector" POSTPROC_BIN="$WORKSPACE_DIR/target/release/swactor-diag-postproc" @@ -75,15 +75,15 @@ POSTPROC_BIN="$WORKSPACE_DIR/target/release/swactor-diag-postproc" # present even for this stub run. The pp binaries live in this crate's # workspace; the diagnostics binaries live at the repo root. if [ -z "${PP_SKIP_BUILD:-}" ]; then - echo "docker-e2e: building pp-gpu-node + pp-smoke-run (release)" + echo "docker-e2e: building pp-worker + pp-orchestrator (release)" cargo build --manifest-path "$CRATE_DIR/Cargo.toml" --release \ - --bin pp-gpu-node --bin pp-smoke-run + --bin pp-worker --bin pp-orchestrator echo "docker-e2e: building swactor-diag-{collector,postproc} (release, --features collector)" cargo build --manifest-path "$WORKSPACE_DIR/Cargo.toml" --release \ -p distribution --features collector \ --bin swactor-diag-collector --bin swactor-diag-postproc fi -for f in "$SMOKE_RUN_BIN" "$GPU_NODE_BIN" "$WORKER_PY" "$COLLECTOR_BIN" "$POSTPROC_BIN"; do +for f in "$ORCHESTRATOR_BIN" "$WORKER_BIN" "$WORKER_PY" "$COLLECTOR_BIN" "$POSTPROC_BIN"; do [ -f "$f" ] || { echo "docker-e2e: missing $f" >&2; exit 1; } done @@ -117,7 +117,7 @@ cleanup_containers() { } cleanup_containers -# Step 4: drive pp-smoke-run with the docker shim as its --gpu-node. +# Step 4: drive pp-orchestrator with the docker shim as its --gpu-node. # The shim consults PP_IMAGE / PP_CONTAINER_PREFIX / PP_DEV from its env. OUTPUT_DIR="$(mktemp -d)" STDOUT_LOG="$OUTPUT_DIR/stdout.log" @@ -135,7 +135,7 @@ if [ -n "${PP_REAL:-}" ]; then PP_DEV=CUDA \ PP_GPUS="${PP_GPUS:-all}" \ PP_MODEL_CACHE_DIR="$PP_MODEL_CACHE_DIR" \ - "$SMOKE_RUN_BIN" \ + "$ORCHESTRATOR_BIN" \ --seed \ --num-stages "$NUM_STAGES" \ --gpu-node "$SCRIPT_DIR/docker-gpu-node.sh" \ @@ -148,7 +148,7 @@ else PP_IMAGE="$IMAGE" \ PP_CONTAINER_PREFIX="$PREFIX" \ PP_DEV=CPU \ - "$SMOKE_RUN_BIN" \ + "$ORCHESTRATOR_BIN" \ --seed \ --num-stages "$NUM_STAGES" \ --gpu-node "$SCRIPT_DIR/docker-gpu-node.sh" \ @@ -157,11 +157,11 @@ else --max-tokens "$MAX_TOKENS" \ >"$STDOUT_LOG" 2>"$STDERR_LOG" fi -SMOKE_STATUS=$? +ORCH_STATUS=$? set -e -if [ $SMOKE_STATUS -ne 0 ]; then - echo "docker-e2e: pp-smoke-run exited $SMOKE_STATUS" >&2 +if [ $ORCH_STATUS -ne 0 ]; then + echo "docker-e2e: pp-orchestrator exited $ORCH_STATUS" >&2 echo "----- stdout -----" >&2 cat "$STDOUT_LOG" >&2 echo "----- stderr (last 60) -----" >&2 diff --git a/examples/pipeline-parallel-inference/scripts/docker-gpu-node.sh b/examples/pipeline-parallel-inference/scripts/docker-gpu-node.sh index b59953f..328e6a9 100755 --- a/examples/pipeline-parallel-inference/scripts/docker-gpu-node.sh +++ b/examples/pipeline-parallel-inference/scripts/docker-gpu-node.sh @@ -1,9 +1,9 @@ #!/usr/bin/env bash -# docker-gpu-node.sh — shim that pp-smoke-run can spawn instead of the -# pp-gpu-node binary directly. Boots one pp-gpu-node container per stage +# docker-gpu-node.sh — shim that pp-orchestrator can spawn instead of the +# pp-worker binary directly. Boots one pp-worker container per stage # on the host network so iroh can dial without NAT. # -# Required env (forwarded by pp-smoke-run): +# Required env (forwarded by pp-orchestrator): # STAGE, NUM_STAGES, SEED_ADDR, SEED_DIRECT, MAX_TOKENS # Optional env (forwarded if present): # MODEL, PP_WORKER_STUB, WORKER_CMD, @@ -80,10 +80,10 @@ mkdir -p "$CACHE_DIR" # # --entrypoint runs the binary directly, bypassing the image's default # pp_entrypoint.sh (sshd + postmortem hold). That supervisor is for remote -# vast.ai nodes; a local docker stage should exit cleanly when pp-gpu-node +# vast.ai nodes; a local docker stage should exit cleanly when pp-worker # does so --rm reaps it and the E2E's no-leftover-container check holds. exec docker run --rm --init \ - --entrypoint /usr/local/bin/pp-gpu-node \ + --entrypoint /usr/local/bin/pp-worker \ --name "$NAME" \ --network "${PP_DIAG_NETWORK:-host}" \ "${GPU_ARGS[@]}" \ diff --git a/examples/pipeline-parallel-inference/scripts/remote-collector-fleet.sh b/examples/pipeline-parallel-inference/scripts/remote-collector-fleet.sh new file mode 100755 index 0000000..b979571 --- /dev/null +++ b/examples/pipeline-parallel-inference/scripts/remote-collector-fleet.sh @@ -0,0 +1,179 @@ +#!/usr/bin/env bash +# remote-collector-fleet.sh — like demo-fleet.sh, but the diagnostics collector +# lives OFF-box on the real VPS (prod topology) instead of on localhost. +# +# Brings up, on THIS machine, the docker stage fleet you would normally deploy +# (N stub-mode pp-worker containers on --network host) plus pp-orchestrator +# hosting the FULL swactor dashboard locally. Every stage's in-VM monitor ships +# its VastaiSample/VastaiLogs records to the REMOTE collector, and the +# orchestrator's Fleet tab subscribes to that same remote collector's SSE +# stream (/diag/stream/) and folds the records — so the Fleet tab is +# exercised end-to-end against the production collector over the public WAN. +# +# The collector is NOT started here; it must already be running on the VPS and +# its port reachable (ufw). This is the "test the prod collector locally" path. +# +# Usage: +# examples/pipeline-parallel-inference/scripts/remote-collector-fleet.sh [N] +# +# Environment overrides: +# PP_COLLECTOR_HOST VPS host running swactor-diag-collector (default 139.59.195.69) +# PP_COLLECTOR_PORT collector HTTP port (default 9080) +# PP_COLLECTOR_UDP collector UDP echo port (default 9081) +# PP_RUN_ID diagnostics run id (default remote-fleet-) +# PP_DIAG_IMAGE code image tag (default swactor-pp-gpu:latest) +# PP_BASE_IMAGE base image tag (default swactor-pp-base:cuda12.6) +# PP_SKIP_BUILD skip cargo build (reuse target/) +# PP_SKIP_IMAGE_BUILD skip docker image build (reuse tag) +# PP_DASHBOARD_PORT orchestrator dashboard port (default 9095) +# PP_PROMPT inference prompt (default "remote fleet demo") +# PP_MAX_TOKENS decode token cap (default 4) +# PP_NO_OPEN if set, don't open a browser +set -euo pipefail + +NUM_STAGES="${1:-3}" +COLLECTOR_HOST="${PP_COLLECTOR_HOST:-139.59.195.69}" +COLLECTOR_PORT="${PP_COLLECTOR_PORT:-9080}" +COLLECTOR_UDP="${PP_COLLECTOR_UDP:-9081}" +RUN_ID="${PP_RUN_ID:-remote-fleet-$(date +%s)}" +IMAGE="${PP_DIAG_IMAGE:-swactor-pp-gpu:latest}" +BASE_IMAGE="${PP_BASE_IMAGE:-swactor-pp-base:cuda12.6}" +PROMPT="${PP_PROMPT:-remote fleet demo}" +MAX_TOKENS="${PP_MAX_TOKENS:-4}" +DASH_PORT="${PP_DASHBOARD_PORT:-9095}" +CONTAINER_PREFIX="remote-fleet-stage" +COLLECTOR_URL="http://${COLLECTOR_HOST}:${COLLECTOR_PORT}" +DASH_URL="http://127.0.0.1:${DASH_PORT}/" + +if ! [[ "$NUM_STAGES" =~ ^[0-9]+$ ]] || [ "$NUM_STAGES" -lt 2 ]; then + echo "remote-fleet: N must be an integer >= 2 (seed mode needs >=2 stages), got '$NUM_STAGES'" >&2 + exit 2 +fi +if ! command -v docker >/dev/null 2>&1; then echo "remote-fleet: docker not on PATH" >&2; exit 2; fi +if ! docker info >/dev/null 2>&1; then echo "remote-fleet: docker daemon unreachable" >&2; exit 2; fi +if (exec 3<>"/dev/tcp/127.0.0.1/${DASH_PORT}") 2>/dev/null; then + exec 3>&- 3<&- + echo "remote-fleet: dashboard port ${DASH_PORT} in use. Pick another: PP_DASHBOARD_PORT=9096 $0 ${NUM_STAGES}" >&2 + exit 2 +fi + +# Preflight: the remote collector must be reachable, else the Fleet tab and the +# stages' shipping will silently get nothing. Fail loudly here instead. +echo "remote-fleet: checking remote collector at ${COLLECTOR_URL} …" +if ! curl -fsS -m 8 -o /dev/null "${COLLECTOR_URL}/diag/runs"; then + echo "remote-fleet: cannot reach ${COLLECTOR_URL}/diag/runs — is the collector up and the port open (ufw)?" >&2 + exit 1 +fi +echo "remote-fleet: remote collector reachable." + +SCRIPT_DIR="$(cd "$(dirname "$0")" && pwd)" +CRATE_DIR="$(cd "$SCRIPT_DIR/.." && pwd)" +WORKSPACE_DIR="$(cd "$CRATE_DIR/../.." && pwd)" +ORCHESTRATOR_BIN="$CRATE_DIR/target/release/pp-orchestrator" +WORKER_BIN="$CRATE_DIR/target/release/pp-worker" +WORKER_PY="$CRATE_DIR/pp_tinygrad_worker.py" + +if [ -z "${PP_SKIP_BUILD:-}" ]; then + echo "remote-fleet: cargo build pp-orchestrator + pp-worker (release)" + cargo build --manifest-path "$CRATE_DIR/Cargo.toml" --release --bin pp-worker --bin pp-orchestrator +fi +for f in "$ORCHESTRATOR_BIN" "$WORKER_BIN" "$WORKER_PY"; do + [ -f "$f" ] || { echo "remote-fleet: missing $f" >&2; exit 1; } +done +if [ -z "${PP_SKIP_IMAGE_BUILD:-}" ]; then + echo "remote-fleet: docker build $BASE_IMAGE (base)" + docker build -f "$CRATE_DIR/Dockerfile.base" -t "$BASE_IMAGE" "$WORKSPACE_DIR" + echo "remote-fleet: docker build $IMAGE (code)" + docker build -f "$CRATE_DIR/Dockerfile" --build-arg "BASE_IMAGE=$BASE_IMAGE" -t "$IMAGE" "$WORKSPACE_DIR" +fi + +WORKDIR="$(mktemp -d -t remote-fleet.XXXXXX)" +SPOOL_DIR="$WORKDIR/spool" +ORCH_LOG="$WORKDIR/orch.log" +FIFO="$WORKDIR/orch.stdin" +mkdir -p "$SPOOL_DIR" +mkfifo "$FIFO" +ORCH_PID="" +CLEANED="" +cleanup() { + [ -n "$CLEANED" ] && return 0 + CLEANED=1 + set +e + echo; echo "remote-fleet: tearing down…" + local ids + ids=$(docker ps -aq --filter "name=^${CONTAINER_PREFIX}-[0-9]+$") + [ -n "$ids" ] && docker rm -f $ids >/dev/null 2>&1 + [ -n "$ORCH_PID" ] && kill "$ORCH_PID" >/dev/null 2>&1 + exec 3>&- 2>/dev/null + [ -d "$WORKDIR" ] && rm -rf "$WORKDIR" + set -e + echo "remote-fleet: done." +} +trap cleanup EXIT +trap 'exit 130' INT TERM + +exec 3<>"$FIFO" +echo "remote-fleet: launching orchestrator + ${NUM_STAGES} stage containers" +echo "remote-fleet: collector = ${COLLECTOR_URL} run_id = ${RUN_ID}" +export PP_HOLD=1 +export PP_WORKER_STUB=1 +export PP_DEV=CPU +export PP_IMAGE="$IMAGE" +export PP_CONTAINER_PREFIX="$CONTAINER_PREFIX" +export PP_DASHBOARD=1 +export PP_DASHBOARD_PORT="$DASH_PORT" +export SWACTOR_DIAG_COLLECTOR_URL="$COLLECTOR_URL" +export SWACTOR_DIAG_RUN_ID="$RUN_ID" +export SWACTOR_DIAG_SPOOL_DIR="$SPOOL_DIR" +export SWACTOR_DIAG_UDP_ECHO="${COLLECTOR_HOST}:${COLLECTOR_UDP}" +"$ORCHESTRATOR_BIN" \ + --seed --num-stages "$NUM_STAGES" \ + --gpu-node "$SCRIPT_DIR/docker-gpu-node.sh" \ + --worker "$WORKER_PY" \ + --prompt "$PROMPT" --max-tokens "$MAX_TOKENS" \ + <"$FIFO" >"$ORCH_LOG" 2>&1 & +ORCH_PID=$! + +WAITED=0 +until (echo > "/dev/tcp/127.0.0.1/${DASH_PORT}") >/dev/null 2>&1; do + if ! kill -0 "$ORCH_PID" >/dev/null 2>&1; then + echo "remote-fleet: orchestrator exited before its dashboard came up." >&2 + tail -n 40 "$ORCH_LOG" >&2 || true + exit 1 + fi + WAITED=$((WAITED + 1)) + [ "$WAITED" -ge 30 ] && { echo "remote-fleet: dashboard did not bind :${DASH_PORT} in 30s" >&2; tail -n 40 "$ORCH_LOG" >&2; exit 1; } + sleep 1 +done +echo +echo " ┌─────────────────────────────────────────────────────────────┐" +echo " │ Full swactor dashboard: $DASH_URL" +echo " │ Fleet tab pulls from remote collector: ${COLLECTOR_URL}/diag/stream/${RUN_ID}" +echo " │ Remote collector board: ${COLLECTOR_URL}/dashboard?run=${RUN_ID}" +echo " └─────────────────────────────────────────────────────────────┘" +echo +if [ -z "${PP_NO_OPEN:-}" ]; then + if command -v xdg-open >/dev/null 2>&1; then (xdg-open "$DASH_URL" >/dev/null 2>&1 &) || true + elif command -v open >/dev/null 2>&1; then (open "$DASH_URL" >/dev/null 2>&1 &) || true + fi +fi + +echo "remote-fleet: waiting for the cluster to converge (first inference drive)…" +WAITED=0 +until grep -q "holding cluster open" "$ORCH_LOG" 2>/dev/null; do + if ! kill -0 "$ORCH_PID" >/dev/null 2>&1; then + echo "remote-fleet: orchestrator exited before holding — drive failed." >&2 + tail -n 40 "$ORCH_LOG" >&2 || true + exit 1 + fi + WAITED=$((WAITED + 1)) + [ "$WAITED" -ge 180 ] && { echo "remote-fleet: cluster did not converge within 180s" >&2; tail -n 40 "$ORCH_LOG" >&2; exit 1; } + sleep 1 +done +RUNNING=$(docker ps -q --filter "name=^${CONTAINER_PREFIX}-[0-9]+$" | wc -l | tr -d ' ') +echo +echo "remote-fleet: ✅ fleet up — ${RUNNING}/${NUM_STAGES} stage containers shipping to ${COLLECTOR_URL}." +echo "remote-fleet: watch the Fleet tab live at $DASH_URL" +echo "remote-fleet: press Ctrl+C to tear everything down." +echo +wait "$ORCH_PID" diff --git a/examples/pipeline-parallel-inference/src/bin/pp_smoke_run.rs b/examples/pipeline-parallel-inference/src/bin/pp_orchestrator.rs similarity index 74% rename from examples/pipeline-parallel-inference/src/bin/pp_smoke_run.rs rename to examples/pipeline-parallel-inference/src/bin/pp_orchestrator.rs index e8778e8..c857acb 100644 --- a/examples/pipeline-parallel-inference/src/bin/pp_smoke_run.rs +++ b/examples/pipeline-parallel-inference/src/bin/pp_orchestrator.rs @@ -1,12 +1,12 @@ -//! pp-smoke-run — orchestrator for the pipeline-parallel smoke test. +//! pp-orchestrator — orchestrator for the pipeline-parallel smoke test. //! //! Two modes: //! -//! * `--seed` — fully local. Spawns `N` `pp-gpu-node` child processes +//! * `--seed` — fully local. Spawns `N` `pp-worker` child processes //! (`STAGE=0..N-1`) talking to a local iroh seed. Uses //! `RelayMode::Disabled` since direct addresses suffice on localhost. //! * `--vastai` — rents `N` GPU instances on vast.ai, deploys the -//! `pp-gpu-node` image to each, and drives the same orchestrator path +//! `pp-worker` image to each, and drives the same orchestrator path //! over WAN. The default one-shot destroys all rented instances before //! exit; `--hold` leaves the cluster running (tracked by a local handle //! file) so it can be iterated on, and `--teardown` destroys it. See the @@ -55,6 +55,7 @@ use distribution::diagnostics::Role as DiagRole; use pipeline_parallel_inference::diag; use pipeline_parallel_inference::dist_broadcast; +use pipeline_parallel_inference::fleet_plugin::RemoteVastaiPlugin; use pipeline_parallel_inference::dist_plugin::{ CountingEmitter, DistDashPlugin, MsgCounts, SharedSnapshot, }; @@ -94,8 +95,8 @@ fn node_config() -> DistributedNodeConfig { fn print_usage() { eprintln!("Usage:"); - eprintln!(" pp-smoke-run --seed [--num-stages N] [--prompt ] [--max-tokens ] [--gpu-node ] [--worker ]"); - eprintln!(" pp-smoke-run --vastai --api-key [--num-stages N] [--gpu \"RTX 3060\"] [--image ] [--prompt ] [--max-tokens ]"); + eprintln!(" pp-orchestrator --seed [--num-stages N] [--prompt ] [--max-tokens ] [--gpu-node ] [--worker ]"); + eprintln!(" pp-orchestrator --vastai --api-key [--num-stages N] [--gpu \"RTX 3060\"] [--image ] [--prompt ] [--max-tokens ]"); eprintln!("Cluster lifecycle (--vastai):"); eprintln!(" (default) lease N, drive one run, destroy."); eprintln!(" --hold lease N, drive, leave running; writes a cluster-handle file."); @@ -141,14 +142,14 @@ fn parse_args() -> Args { vastai: false, num_stages: 2, api_key: None, - // RTX 3060 (12GB) is our default deploy-test class: cheapest GPU class - // with deep, reliable supply on vast.ai (see fleet notes). Settable via - // PP_GPU in a profile, or --gpu for capacity tests. NOT sized for real - // model weights. + // GPU model pin: off by default (empty = no model filter). Set PP_GPU in + // a profile, or --gpu for capacity tests, to restrict to one model. Pair + // or replace with PP_GPU_MIN_RAM_MB to select by VRAM instead. gpu_name: std::env::var("PP_GPU") .ok() - .filter(|s| !s.trim().is_empty()) - .unwrap_or_else(|| "RTX 3060".into()), + .map(|s| s.trim().to_string()) + .filter(|s| !s.is_empty()) + .unwrap_or_default(), // Image is not baked to a personal registry: it defaults from the // PP_IMAGE env (the convention the run scripts already use, e.g. // scripts/docker-e2e.sh), falling back to a registry-less tag. Set it @@ -272,7 +273,7 @@ fn main() { // ─── Seed (localhost) mode ──────────────────────────────────────────── -/// Resolve the `pp-gpu-node` binary path. Defaults to a sibling of the +/// Resolve the `pp-worker` binary path. Defaults to a sibling of the /// current executable. fn resolve_gpu_node_path(args: &Args) -> PathBuf { if let Some(p) = &args.gpu_node_path { @@ -280,7 +281,7 @@ fn resolve_gpu_node_path(args: &Args) -> PathBuf { } let exe = std::env::current_exe().expect("current_exe failed"); let parent = exe.parent().expect("current exe has no parent"); - parent.join("pp-gpu-node") + parent.join("pp-worker") } /// Resolve the worker script path. Defaults to `pp_tinygrad_worker.py` @@ -292,7 +293,7 @@ fn resolve_worker_path(args: &Args) -> PathBuf { PathBuf::from(env!("CARGO_MANIFEST_DIR")).join("pp_tinygrad_worker.py") } -/// Build a `Command` for a single `pp-gpu-node` child. Captures all the +/// Build a `Command` for a single `pp-worker` child. Captures all the /// env-var bookkeeping in one place so the spawn-chain closure is short. fn build_gpu_node_command( gpu_node_bin: &PathBuf, @@ -303,7 +304,7 @@ fn build_gpu_node_command( ctx: &StageSpawnCtx, ) -> Command { eprintln!( - "pp-smoke-run: spawning {} STAGE={} NUM_STAGES={}", + "pp-orchestrator: spawning {} STAGE={} NUM_STAGES={}", gpu_node_bin.display(), ctx.stage, ctx.num_stages, @@ -421,7 +422,7 @@ fn run_seed(args: &Args) -> i32 { let gpu_node_bin = resolve_gpu_node_path(args); if !gpu_node_bin.exists() { eprintln!( - "pp-smoke-run: pp-gpu-node binary not found at {} (use --gpu-node to override)", + "pp-orchestrator: pp-worker binary not found at {} (use --gpu-node to override)", gpu_node_bin.display() ); return 1; @@ -429,7 +430,7 @@ fn run_seed(args: &Args) -> i32 { let worker_script = resolve_worker_path(args); if !worker_script.exists() { eprintln!( - "pp-smoke-run: worker script not found at {} (use --worker to override)", + "pp-orchestrator: worker script not found at {} (use --worker to override)", worker_script.display() ); return 1; @@ -444,7 +445,7 @@ fn run_seed(args: &Args) -> i32 { }) { Ok(d) => d, Err(e) => { - eprintln!("pp-smoke-run: failed to create iroh driver: {e}"); + eprintln!("pp-orchestrator: failed to create iroh driver: {e}"); return 1; } }; @@ -476,7 +477,7 @@ fn run_seed(args: &Args) -> i32 { .collect::>() .join(","); eprintln!( - "pp-smoke-run (--seed --num-stages {n}): orchestrator node {my_hex}, direct={direct:?}", + "pp-orchestrator (--seed --num-stages {n}): orchestrator node {my_hex}, direct={direct:?}", n = args.num_stages, ); @@ -543,10 +544,22 @@ fn run_seed(args: &Args) -> i32 { ))); let poll_stop = Arc::new(AtomicBool::new(false)); spawn_conn_poller(&driver, Arc::clone(cached), conn_tracker, poll_stop); + // Fleet plugin ("Fleet" nav tab): the vast.ai/host-metric fleet view. + // In production the collector lives on a VPS; this plugin *pulls* the + // collector's already-folded fleet model ~1/s and re-serves it locally, + // so the full dashboard shows the fleet beside the live actor views. + // Registered unconditionally so the tab exists; the poller only runs + // when a collector URL is known (the same one the dist snapshot is + // pushed to). Idle URL => the tab just waits for records. + let fleet = Arc::new(RemoteVastaiPlugin::new()); + if let Some(url) = broadcast_url.as_ref() { + fleet.spawn_stream(&driver.tokio_handle(), url.clone(), broadcast_run_id()); + } + handle.register_plugin(fleet); handle.start_http(driver.tokio_handle()); eprintln!( - "pp-smoke-run: live dashboard on http://localhost:{port} \ - (overview / actors / topology / distribution / netmap)" + "pp-orchestrator: live dashboard on http://localhost:{port} \ + (overview / actors / topology / distribution / netmap / fleet)" ); } @@ -560,13 +573,13 @@ fn run_seed(args: &Args) -> i32 { url.clone(), broadcast_run_id(), ); - eprintln!("pp-smoke-run: broadcasting distribution snapshot to {url}"); + eprintln!("pp-orchestrator: broadcasting distribution snapshot to {url}"); } let response_inbox = match rt.new_inbox::() { Ok(i) => i, Err(e) => { - eprintln!("pp-smoke-run: new_inbox failed: {e}"); + eprintln!("pp-orchestrator: new_inbox failed: {e}"); break 'run (1, "new_inbox_error"); } }; @@ -589,18 +602,18 @@ fn run_seed(args: &Args) -> i32 { }) { Ok(g) => g, Err(e) => { - eprintln!("pp-smoke-run: {e}"); + eprintln!("pp-orchestrator: {e}"); break 'run (1, "spawn_chain_error"); } }; - eprintln!("pp-smoke-run: spawned {} stage children", guard.len()); + eprintln!("pp-orchestrator: spawned {} stage children", guard.len()); // The chain spawner read the announcement line from each child's // stdout and kept the pipe draining in a background thread. No // further stdout pumping needed here. // Wait for the cluster (orchestrator + N stages) to converge. eprintln!( - "pp-smoke-run: waiting for cluster convergence ({} alive peers)...", + "pp-orchestrator: waiting for cluster convergence ({} alive peers)...", args.num_stages, ); let conv_res = await_convergence_or_child_death( @@ -620,17 +633,17 @@ fn run_seed(args: &Args) -> i32 { .node_mut() .register_name(ORCHESTRATOR_NAME.into(), inbox_addr); diag::emit_register_name(&driver, ORCHESTRATOR_NAME, inbox_addr, None); - eprintln!("pp-smoke-run: registered {ORCHESTRATOR_NAME} -> {inbox_addr:?}"); + eprintln!("pp-orchestrator: registered {ORCHESTRATOR_NAME} -> {inbox_addr:?}"); if let Err(e) = conv_res { - eprintln!("pp-smoke-run: {e}"); + eprintln!("pp-orchestrator: {e}"); break 'run (1, "convergence_error"); } - eprintln!("pp-smoke-run: cluster converged"); + eprintln!("pp-orchestrator: cluster converged"); // Spec §4.6 + §4.5: gate the request injection on (a) every // pp-stage-K resolvable and (b) pp-entry resolvable. The // per-index name is published by each stage only after its - // worker is ready (pp-gpu-node.rs), so resolution of every + // worker is ready (pp-worker.rs), so resolution of every // pp-stage-K is a faithful "all workers ready" signal. The // resolve loop polls children too so a stage that dies during // wiring fails fast instead of waiting out the timeout. @@ -638,7 +651,7 @@ fn run_seed(args: &Args) -> i32 { .ok() .and_then(|s| s.trim().parse().ok()) .unwrap_or(1800); - eprintln!("pp-smoke-run: waiting for pipeline-wired (all pp-stage-K + {ENTRY_NAME})..."); + eprintln!("pp-orchestrator: waiting for pipeline-wired (all pp-stage-K + {ENTRY_NAME})..."); let wire_deadline = Instant::now() + Duration::from_secs(roster_deadline_secs); let mut roster_hex: Vec> = vec![None; args.num_stages as usize]; let (stage0_addr, stage0_node_id) = loop { @@ -660,7 +673,7 @@ fn run_seed(args: &Args) -> i32 { break entry.unwrap(); } if let Err(e) = check_child_death(&mut guard) { - eprintln!("pp-smoke-run: {e}"); + eprintln!("pp-orchestrator: {e}"); break 'run (1, "stage_died_pre_resolve"); } if Instant::now() >= wire_deadline { @@ -670,7 +683,7 @@ fn run_seed(args: &Args) -> i32 { .filter_map(|(k, o)| o.is_none().then_some(k as u32)) .collect(); eprintln!( - "pp-smoke-run: pipeline did not wire within {roster_deadline_secs}s; \ + "pp-orchestrator: pipeline did not wire within {roster_deadline_secs}s; \ missing pp-stage-K for {missing:?} (entry resolved: {})", entry.is_some(), ); @@ -697,7 +710,7 @@ fn run_seed(args: &Args) -> i32 { .iter() .map(|b| format!("{:02x}", b)) .collect(); - eprintln!("pp-smoke-run: {ENTRY_NAME} -> {stage0_addr:?} on {stage0_hex}"); + eprintln!("pp-orchestrator: {ENTRY_NAME} -> {stage0_addr:?} on {stage0_hex}"); // Spec §4.5: emit one pp_stage_roster per drive, before request // injection, listing every stage. Seed mode runs a single drive @@ -719,7 +732,7 @@ fn run_seed(args: &Args) -> i32 { let key = match PublicKey::from_bytes(&stage0_node_id.0) { Ok(k) => k, Err(e) => { - eprintln!("pp-smoke-run: invalid stage-0 node key: {e}"); + eprintln!("pp-orchestrator: invalid stage-0 node key: {e}"); break 'run (1, "stage0_key_error"); } }; @@ -766,11 +779,11 @@ fn run_seed(args: &Args) -> i32 { }), }); eprintln!( - "pp-smoke-run: sending InferenceRequest (prompt={:?}, max_tokens={})", + "pp-orchestrator: sending InferenceRequest (prompt={:?}, max_tokens={})", request.prompt, request.max_tokens ); if let Err(e) = rt.send_to(stage0_addr, request) { - eprintln!("pp-smoke-run: send_to failed: {e}"); + eprintln!("pp-orchestrator: send_to failed: {e}"); break 'run (1, "send_to_error"); } @@ -815,7 +828,7 @@ fn run_seed(args: &Args) -> i32 { (0, "ok") } Err(e) => { - eprintln!("pp-smoke-run: {e}"); + eprintln!("pp-orchestrator: {e}"); (1, e.exit_reason()) } } @@ -884,7 +897,10 @@ enum AwaitError { Timeout(Duration), /// `InferenceResponse` arrived with an empty `text` field. EmptyResponse, - /// Some `pp-gpu-node` child exited locally (seed-mode child guard). + /// `send_to(stage0, request)` failed before the request left the + /// orchestrator, so the drive never started. + SendFailed(String), + /// Some `pp-worker` child exited locally (seed-mode child guard). ChildDied(String), /// Spec §4.4: a SWIM member on the forward path (orchestrator + /// every stage in the resolved roster) transitioned to `dead` while @@ -906,6 +922,7 @@ impl std::fmt::Display for AwaitError { d.as_secs_f32() ), AwaitError::EmptyResponse => write!(f, "received empty InferenceResponse"), + AwaitError::SendFailed(s) => write!(f, "send_to failed: {s}"), AwaitError::ChildDied(s) => write!(f, "{s}"), AwaitError::ForwardPathDead { stage_index, @@ -928,6 +945,7 @@ impl AwaitError { match self { AwaitError::Timeout(_) => "response_timeout", AwaitError::EmptyResponse => "response_empty", + AwaitError::SendFailed(_) => "send_to_error", AwaitError::ChildDied(_) => "stage_died_mid_drive", AwaitError::ForwardPathDead { .. } => "forward_path_dead", } @@ -959,11 +977,11 @@ fn hold_open( ) { match port { Some(p) => eprintln!( - "pp-smoke-run: holding cluster open — orchestrator dashboard at \ + "pp-orchestrator: holding cluster open — orchestrator dashboard at \ http://localhost:{p}. Press Enter (or Ctrl-D) to tear down." ), None => eprintln!( - "pp-smoke-run: holding cluster open. Press Enter (or Ctrl-D) to tear down." + "pp-orchestrator: holding cluster open. Press Enter (or Ctrl-D) to tear down." ), } // Read stdin on a side thread so the main thread can keep pumping the @@ -1060,7 +1078,7 @@ fn await_response( .map(|m| format!("{}={}", &m.node_id[..8.min(m.node_id.len())], m.state)) .collect(); eprintln!( - "pp-smoke-run: waiting for response ({:.0}s elapsed, members: {:?})", + "pp-orchestrator: waiting for response ({:.0}s elapsed, members: {:?})", start.elapsed().as_secs_f32(), members ); @@ -1082,6 +1100,196 @@ fn await_response_timeout_secs(default_secs: u64) -> u64 { .unwrap_or(default_secs) } +/// Drive a single inference request through the already-wired pipeline and +/// await its response (vast.ai mode). Self-contained per drive: emits this +/// drive's spec §4.5 `pp_stage_roster` + §4.6 `pp_pipeline_wired` markers and +/// the `pp_drive_start` / `pp_drive_end` boundaries, sends one +/// `InferenceRequest` to stage 0, waits for the response, and prints it on +/// success. The roster and the stage-0 route are resolved once by the caller +/// and reused across drives; the roster is re-emitted each drive so the bundle +/// reader can slice the interleaved event stream per `drive_seq`. Returns the +/// drive's result so the caller can derive an exit code. +#[allow(clippy::too_many_arguments)] +fn drive_once( + driver: &mut IrohDriver, + rt: &Runtime, + codecs: &Arc, + response_inbox: &Inbox, + stage0_addr: swactor::actor::ActorAddress, + roster: &[pipeline_parallel_inference::orchestrator::StageRosterEntry], + num_stages: u32, + label: &str, + prompt: &str, + max_tokens: u32, + drive_seq: u32, +) -> Result { + let drive_start = Instant::now(); + let inbox_addr = *response_inbox.addr(); + + // Spec §4.5: emit one pp_stage_roster per drive, before any request + // injection event. + driver.emit(distribution::diagnostics::event::Event::Custom { + kind: "pp_stage_roster".into(), + fields: stage_roster_event_fields(drive_seq, roster), + }); + // Spec §4.6: emit exactly one pp_pipeline_wired per drive (every stage + // ready, every neighbour wired — proxied by pp-stage-K registration being + // post-ready — and pp-entry resolved). + driver.emit(distribution::diagnostics::event::Event::Custom { + kind: "pp_pipeline_wired".into(), + fields: serde_json::json!({ "drive_seq": drive_seq }), + }); + + let request = InferenceRequest { + reply_to: inbox_addr, + prompt: prompt.to_string(), + max_tokens, + }; + // Mark this drive's slice of the event stream so a bundle reader can split + // events by drive. + driver.emit(distribution::diagnostics::event::Event::Custom { + kind: "pp_drive_start".into(), + fields: serde_json::json!({ + "drive_seq": drive_seq, + "prompt": prompt, + "max_tokens": max_tokens, + "num_stages": num_stages, + "label": label, + }), + }); + + let result = match rt.send_to(stage0_addr, request) { + Err(e) => Err(AwaitError::SendFailed(e.to_string())), + Ok(()) => { + let await_secs = await_response_timeout_secs(600); + await_response( + driver, + rt, + codecs, + response_inbox, + Duration::from_secs(await_secs), + None, + roster, + drive_seq, + None, + ) + } + }; + + let (code, reason): (i32, &'static str) = match &result { + Ok(_) => (0, "ok"), + Err(e) => (1, e.exit_reason()), + }; + // Close out this drive's slice of the event stream — emitted even on + // failure so the bundle reader can window events per drive. + driver.emit(distribution::diagnostics::event::Event::Custom { + kind: "pp_drive_end".into(), + fields: serde_json::json!({ + "drive_seq": drive_seq, + "exit_reason": reason, + "elapsed_ms": drive_start.elapsed().as_millis() as u64, + "code": code, + }), + }); + + match &result { + Ok(text) => { + println!("=== pipeline-parallel Inference Response ==="); + println!("{text}"); + println!("============================================"); + } + Err(e) => eprintln!("pp-orchestrator: drive {drive_seq} failed: {e}"), + } + result +} + +/// Live multi-prompt loop for vast.ai mode (dashboard or `--hold`). Keeps the +/// cluster converged and the dashboard SSE fed while the operator drives more +/// prompts. A stdin-reader side thread feeds prompt lines so the main thread +/// can keep pumping the driver; each non-empty line drives one more inference at +/// the next `drive_seq`. A blank line, `quit`, or EOF (Ctrl-D) ends the loop, +/// after which the caller's finalize + teardown tail runs. +#[allow(clippy::too_many_arguments)] +fn prompt_loop( + driver: &mut IrohDriver, + rt: &Runtime, + codecs: &Arc, + response_inbox: &Inbox, + stage0_addr: swactor::actor::ActorAddress, + roster: &[pipeline_parallel_inference::orchestrator::StageRosterEntry], + num_stages: u32, + label: &str, + max_tokens: u32, + first_drive_seq: u32, + dist_cached: Option<&SharedSnapshot>, + port: Option, +) { + match port { + Some(p) => eprintln!( + "pp-orchestrator: cluster live — dashboard at http://localhost:{p}. \ + Type a prompt + Enter to drive again; blank line / Ctrl-D / `quit` to tear down." + ), + None => eprintln!( + "pp-orchestrator: cluster live. Type a prompt + Enter to drive again; \ + blank line / Ctrl-D / `quit` to tear down." + ), + } + + // Read prompts on a side thread so the main thread keeps pumping the driver; + // a blocking stdin read here would freeze SWIM and the live snapshot. The + // thread forwards each prompt line over a channel and flips `stop` on a + // blank line, `quit`, or EOF. + let (tx, rx) = std::sync::mpsc::channel::(); + let stop = Arc::new(AtomicBool::new(false)); + { + let stop = Arc::clone(&stop); + std::thread::spawn(move || { + use std::io::BufRead; + let stdin = std::io::stdin(); + for line in stdin.lock().lines() { + let Ok(line) = line else { break }; + let trimmed = line.trim(); + if trimmed.is_empty() || trimmed.eq_ignore_ascii_case("quit") { + break; + } + if tx.send(trimmed.to_string()).is_err() { + break; + } + } + stop.store(true, Ordering::SeqCst); + }); + } + + let mut drive_seq = first_drive_seq; + while !stop.load(Ordering::SeqCst) { + driver.recv(); + driver.tick(); + if let Some(cached) = dist_cached { + *cached.lock().unwrap() = Some(driver.snapshot()); + } + // Drive any prompts that arrived since the last tick. drive_once pumps + // the driver itself while awaiting each response. + while let Ok(prompt) = rx.try_recv() { + drive_seq += 1; + eprintln!("pp-orchestrator: driving prompt #{drive_seq}: {prompt:?}"); + let _ = drive_once( + driver, + rt, + codecs, + response_inbox, + stage0_addr, + roster, + num_stages, + label, + &prompt, + max_tokens, + drive_seq, + ); + } + std::thread::sleep(Duration::from_millis(200)); + } +} + // ─── vast.ai mode ───────────────────────────────────────────────────── // ─── vast.ai cluster lifecycle (hold / teardown) ────────────────────── @@ -1336,14 +1544,14 @@ fn run_teardown( let st = match ClusterState::load(state_path) { Ok(s) => s, Err(e) => { - eprintln!("pp-smoke-run: {e}"); + eprintln!("pp-orchestrator: {e}"); eprintln!(" (nothing to tear down at that path)"); return 1; } }; let ids: Vec = st.contracts.iter().map(|c| c.id).collect(); eprintln!( - "pp-smoke-run: tearing down label={} contracts={ids:?}", + "pp-orchestrator: tearing down label={} contracts={ids:?}", st.label ); let results = tokio_rt.block_on(vastai::destroy_all_instances(http, base_url, api_key, &ids)); @@ -1351,13 +1559,13 @@ fn run_teardown( for (id, r) in ids.iter().zip(results.iter()) { if let Err(e) = r { ok = false; - eprintln!("pp-smoke-run: destroy {id} failed: {e}"); + eprintln!("pp-orchestrator: destroy {id} failed: {e}"); } } match tokio_rt.block_on(vastai::list_instances_by_label(http, base_url, api_key, &st.label)) { Ok(remaining) if remaining.is_empty() => { eprintln!( - "pp-smoke-run: confirmed 0 instances under label {}", + "pp-orchestrator: confirmed 0 instances under label {}", st.label ); // Seal the bundle now that the cluster is provably gone: @@ -1368,13 +1576,13 @@ fn run_teardown( match tokio_rt.block_on(post_teardown_finalize(http, &st)) { Ok(()) if st.collector_url.is_some() => { eprintln!( - "pp-smoke-run: posted finalize to collector for run_id={}", + "pp-orchestrator: posted finalize to collector for run_id={}", st.run_id.as_deref().unwrap_or(""), ); } Ok(()) => {} // no collector pinned — nothing to do Err(e) => { - eprintln!("pp-smoke-run: WARNING teardown finalize failed: {e}"); + eprintln!("pp-orchestrator: WARNING teardown finalize failed: {e}"); eprintln!( " (the bundle is still retrievable via GET {}/diag/bundle/{} — staging is intact)", st.collector_url.as_deref().unwrap_or(""), @@ -1384,7 +1592,7 @@ fn run_teardown( } if let Err(e) = std::fs::remove_file(state_path) { eprintln!( - "pp-smoke-run: note: could not remove {}: {e}", + "pp-orchestrator: note: could not remove {}: {e}", state_path.display() ); } @@ -1392,7 +1600,7 @@ fn run_teardown( Ok(remaining) => { ok = false; eprintln!( - "pp-smoke-run: WARNING {} instance(s) still under label {} — keeping handle file", + "pp-orchestrator: WARNING {} instance(s) still under label {} — keeping handle file", remaining.len(), st.label ); @@ -1402,7 +1610,7 @@ fn run_teardown( } Err(e) => { ok = false; - eprintln!("pp-smoke-run: could not verify teardown via API: {e}"); + eprintln!("pp-orchestrator: could not verify teardown via API: {e}"); } } if ok { @@ -1417,7 +1625,7 @@ fn run_vastai(args: &Args) -> i32 { let tokio_rt = match tokio::runtime::Runtime::new() { Ok(r) => r, Err(e) => { - eprintln!("pp-smoke-run: tokio runtime failed: {e}"); + eprintln!("pp-orchestrator: tokio runtime failed: {e}"); return 1; } }; @@ -1434,7 +1642,7 @@ fn run_vastai(args: &Args) -> i32 { let cluster = match resolve_cluster(args, &state_path) { Ok(c) => c, Err(e) => { - eprintln!("pp-smoke-run: {e}"); + eprintln!("pp-orchestrator: {e}"); return 1; } }; @@ -1450,7 +1658,7 @@ fn run_vastai(args: &Args) -> i32 { }) { Ok(d) => d, Err(e) => { - eprintln!("pp-smoke-run: failed to create iroh driver: {e}"); + eprintln!("pp-orchestrator: failed to create iroh driver: {e}"); return 1; } }; @@ -1466,6 +1674,23 @@ fn run_vastai(args: &Args) -> i32 { Some(cluster.run_id.as_str()), ); + // Distribution view is wanted when the in-process dashboard (PP_DASHBOARD) is + // on, or a dashboard/collector URL resolves for the remote broadcast. Same + // gating as run_seed. + let broadcast_url = resolve_broadcast_url(); + let want_dist = std::env::var_os("PP_DASHBOARD").is_some() || broadcast_url.is_some(); + + // Orchestrator dashboard message tallies. Decorate the driver's diagnostics + // emitter so every wire MessageSent/MessageReceived is counted for the + // distribution page; the decorator forwards to whatever emitter diag + // installed, so bundle shipping is unaffected. Installed only when the + // distribution view is wanted (local or remote). + let msg_counts = Arc::new(MsgCounts::default()); + if want_dist { + let inner = driver.diagnostics().clone(); + driver.set_diagnostics(Arc::new(CountingEmitter::new(inner, Arc::clone(&msg_counts)))); + } + // Per-cluster drive counter, emitted on pp_drive_start / pp_drive_end so // the bundle reader can slice the interleaved event stream by attempt. // One-shot and --hold each drive exactly once per process, so this is 1. @@ -1483,12 +1708,12 @@ fn run_vastai(args: &Args) -> i32 { let my_id = driver.node_id(); let my_hex: String = my_id.0.iter().map(|b| format!("{:02x}", b)).collect(); eprintln!( - "pp-smoke-run (--vastai --num-stages {n}): orchestrator node {my_hex}", + "pp-orchestrator (--vastai --num-stages {n}): orchestrator node {my_hex}", n = num_stages, ); if diag_env_for_stages.is_enabled() { eprintln!( - "pp-smoke-run: forwarding diagnostics to rented stages (collector={})", + "pp-orchestrator: forwarding diagnostics to rented stages (collector={})", diag_env_for_stages .collector_url .as_deref() @@ -1512,10 +1737,10 @@ fn run_vastai(args: &Args) -> i32 { url }; if let Some(ref u) = relay_url { - eprintln!("pp-smoke-run: home relay {u}"); + eprintln!("pp-orchestrator: home relay {u}"); // Gossip our own home relay through SWIM metadata so the rented // stages learn it without having to dial us back first. Mirrors - // what pp-gpu-node does on its side; together they ensure every + // what pp-worker does on its side; together they ensure every // pair of nodes can resolve each other's relay URL through // metadata gossip alone — the route enrichment in build_route() // depends on this. @@ -1523,7 +1748,7 @@ fn run_vastai(args: &Args) -> i32 { .node_mut() .set_relay_url(Some(u.clone())); } else { - eprintln!("pp-smoke-run: no relay URL after 20s — vastai mode usually requires one"); + eprintln!("pp-orchestrator: no relay URL after 20s — vastai mode usually requires one"); } // ── Vastai monitoring (independent layer) ──────────────────────── @@ -1546,11 +1771,11 @@ fn run_vastai(args: &Args) -> i32 { Some(num_stages), ) { Ok(p) => { - eprintln!("pp-smoke-run: vastai monitoring enabled (external poller)"); + eprintln!("pp-orchestrator: vastai monitoring enabled (external poller)"); Some(p) } Err(e) => { - eprintln!("pp-smoke-run: WARNING vastai monitoring disabled: {e}"); + eprintln!("pp-orchestrator: WARNING vastai monitoring disabled: {e}"); None } } @@ -1559,6 +1784,105 @@ fn run_vastai(args: &Args) -> i32 { }; let vastai_tracker = vastai_poller.as_ref().map(|p| p.tracker()); + // ── Live dashboard + runtime — started BEFORE the lease ────────────── + // The lease + image-load phase is the slow, failure-prone part the operator + // most needs to watch, so the HTTP server binds here (start_http) rather + // than after convergence. Plugins populate as the cluster comes up. Setup + // depends only on the driver + msg_counts created above, never on the lease. + let mut rt = Runtime::new(RuntimeConfig::default()); + let codecs = Arc::new(inference_codec_registry()); + let router = Arc::new(TransportRouter::new()); + rt.set_codec_registry(codecs.clone()); + rt.set_transport_router(router.clone()); + + // When PP_DASHBOARD is set we install a stats hook on the orchestrator + // runtime (must happen before `rt` is shared) and serve the HTTP dashboard + // on the iroh driver's tokio runtime. The handle is held for the whole run + // so the server stays up through the lease, drive, and multi-prompt loop. + let dashboard = if std::env::var_os("PP_DASHBOARD").is_some() { + let port: u16 = std::env::var("PP_DASHBOARD_PORT") + .ok() + .and_then(|s| s.trim().parse().ok()) + .unwrap_or(9090); + let collector = StatsCollector::new(RuntimeConfig::default().num_threads); + rt.set_stats_hook(collector.clone()); + let handle = start_dashboard(DashboardConfig { port, ..Default::default() }); + Some((handle, collector, port)) + } else { + None + }; + + let rt = Arc::new(rt); + + // Shared distribution snapshot cell, refreshed by the multi-prompt loop. + // The in-process plugins and the remote broadcaster both read this one + // cell, so it exists whenever the distribution view is wanted — even + // broadcast-only (no local dashboard). `None` when off so the loop skips + // the refresh. + let dist_cached: Option = if want_dist { + Some(Arc::new(Mutex::new(Some(driver.snapshot())))) + } else { + None + }; + + // When the in-process dashboard is on, register the distribution + net map + // plugins ("Distribution"/"Netmap" nav tabs) against the shared cell. + if let (Some((handle, collector, port)), Some(cached)) = (&dashboard, &dist_cached) { + handle.set_runtime(Arc::clone(&rt), Arc::clone(collector)); + handle.register_plugin(Arc::new(DistDashPlugin::new( + Arc::clone(cached), + Arc::clone(&msg_counts), + ))); + // Net map plugin: a live connection/bandwidth graph. Shares the + // cached snapshot and message tallies; a background poller keeps its + // transport map fresh by querying the iroh endpoint directly. The + // poller's stop flag rides the process lifetime (the driver's tokio + // runtime is torn down at end of run, aborting the task). + let conn_tracker = Arc::new(ConnTracker::default()); + handle.register_plugin(Arc::new(NetmapPlugin::new( + Arc::clone(cached), + Arc::clone(&msg_counts), + Arc::clone(&conn_tracker), + ))); + let poll_stop = Arc::new(AtomicBool::new(false)); + spawn_conn_poller(&driver, Arc::clone(cached), conn_tracker, poll_stop); + // Fleet plugin ("Fleet" nav tab): the vast.ai/host-metric fleet view. + // The collector lives on the VPS; this plugin *pulls* the collector's + // already-folded fleet model (the collector→orchestrator vast.ai + // stream) and re-serves it locally, so the full dashboard shows the + // fleet beside the live actor views. Registered unconditionally so the + // tab exists; the stream only runs when a collector URL is known. + let fleet = Arc::new(RemoteVastaiPlugin::new()); + if let Some(url) = broadcast_url.as_ref() { + // Subscribe under the SAME run_id the stages + vast.ai poller ship + // under (cluster.run_id), NOT broadcast_run_id() — otherwise the + // Fleet tab listens on the wrong slug and shows nothing while the + // poller publishes fine. cluster.run_id already honors + // SWACTOR_DIAG_RUN_ID, then falls back to pp-{label}. + fleet.spawn_stream(&driver.tokio_handle(), url.clone(), cluster.run_id.clone()); + } + handle.register_plugin(fleet); + handle.start_http(driver.tokio_handle()); + eprintln!( + "pp-orchestrator: live dashboard on http://localhost:{port} \ + (overview / actors / topology / distribution / netmap / fleet)" + ); + } + + // Default-on remote broadcast: ship the distribution snapshot to the + // dashboard collector ~1/s so a remote dashboard renders the same view. + // Use cluster.run_id so every stream for this run lands under one slug. + if let (Some(url), Some(cached)) = (broadcast_url.as_ref(), dist_cached.as_ref()) { + dist_broadcast::spawn_dist_broadcast( + driver.tokio_handle(), + Arc::clone(cached), + Arc::clone(&msg_counts), + url.clone(), + cluster.run_id.clone(), + ); + eprintln!("pp-orchestrator: broadcasting distribution snapshot to {url}"); + } + // ── Acquire the running cluster ────────────────────────────────── // Lease N fresh instances and (on --hold) persist the handle. let contract_ids: Vec = { @@ -1572,7 +1896,7 @@ fn run_vastai(args: &Args) -> i32 { match random_secret() { Ok(b) => v.push(to_hex(&b)), Err(e) => { - eprintln!("pp-smoke-run: {e}"); + eprintln!("pp-orchestrator: {e}"); return 1; } } @@ -1594,7 +1918,7 @@ fn run_vastai(args: &Args) -> i32 { // SAFETY: single-threaded here — no lease/diag worker threads have // been spawned yet, so there is no concurrent env access. unsafe { std::env::set_var(var, "20") }; - eprintln!("pp-smoke-run: defaulting {var}=20 (price image pull into offer ranking)"); + eprintln!("pp-orchestrator: defaulting {var}=20 (price image pull into offer ranking)"); } // One call into the lease helper handles select-the-pool, create-N, @@ -1602,16 +1926,28 @@ fn run_vastai(args: &Args) -> i32 { // Describe the selector accurately: VRAM-filter mode (PP_GPU_MIN_RAM_MB) // spans a heterogeneous set of cards, so naming a single model would // mislead. lease_chain logs the survivor pool and each stage's pick. - let selector = match std::env::var("PP_GPU_MIN_RAM_MB").ok().filter(|s| !s.trim().is_empty()) { - Some(mb) => format!("any 1-GPU offer with >={mb}MB VRAM"), - None => args.gpu_name.clone(), + let vram = std::env::var("PP_GPU_MIN_RAM_MB") + .ok() + .map(|s| s.trim().to_string()) + .filter(|s| !s.is_empty()); + let selector = match (&vram, args.gpu_name.is_empty()) { + (Some(mb), true) => format!("any 1-GPU offer with >={mb}MB VRAM"), + (Some(mb), false) => format!("{} with >={mb}MB VRAM", args.gpu_name), + (None, false) => args.gpu_name.clone(), + (None, true) => "any 1-GPU offer".to_string(), }; eprintln!( - "pp-smoke-run: leasing {} instances [{selector}] (label {label})...", + "pp-orchestrator: leasing {} instances [{selector}] (label {label})...", num_stages, ); - let created = match tokio_rt.block_on( - pipeline_parallel_inference::vastai::lease_chain( + // Run the lease in 200ms slices instead of one blocking call, so the + // main thread can pump SWIM and refresh the dashboard snapshot while + // instances come up. Otherwise the membership/topology/net-map panels + // freeze at the empty startup snapshot for the entire (multi-minute, + // CDI-retrying) lease — even though stages are already joining SWIM — + // which defeats the point of binding the dashboard before the lease. + let lease_result = { + let mut lease_fut = Box::pin(pipeline_parallel_inference::vastai::lease_chain( &http, base_url, &api_key, @@ -1633,11 +1969,26 @@ fn run_vastai(args: &Args) -> i32 { 30, Some(&diag_env_for_stages), vastai_tracker.as_ref(), - ), - ) { + )); + loop { + match tokio_rt.block_on(async { + tokio::time::timeout(Duration::from_millis(200), &mut lease_fut).await + }) { + Ok(res) => break res, + Err(_elapsed) => { + driver.recv(); + driver.tick(); + if let Some(cached) = dist_cached.as_ref() { + *cached.lock().unwrap() = Some(driver.snapshot()); + } + } + } + } + }; + let created = match lease_result { Ok(c) => c, Err(e) => { - eprintln!("pp-smoke-run: lease_chain failed: {e}"); + eprintln!("pp-orchestrator: lease_chain failed: {e}"); // One-shot lease failure (a --hold lease failure also // lands here) finalizes — there is no cluster left to // extend the window, so sealing the canonical bundle is @@ -1678,30 +2029,23 @@ fn run_vastai(args: &Args) -> i32 { orchestrator_node_id_hex: Some(my_hex.clone()), }; match st.save(&state_path) { - Ok(()) => eprintln!("pp-smoke-run: wrote cluster handle {}", state_path.display()), - Err(e) => eprintln!("pp-smoke-run: WARNING could not write cluster handle: {e}"), + Ok(()) => eprintln!("pp-orchestrator: wrote cluster handle {}", state_path.display()), + Err(e) => eprintln!("pp-orchestrator: WARNING could not write cluster handle: {e}"), } } ids }; - eprintln!("pp-smoke-run: cluster contracts {contract_ids:?}"); + eprintln!("pp-orchestrator: cluster contracts {contract_ids:?}"); // Drive the run inside a labelled block returning `(code, reason)` so // every failure point can name the reason it bailed; the orchestrator's // diagnostics finalize record then carries that reason into the bundle. // Mirrors the run_seed pattern. - let drive_start_instant = Instant::now(); let (code, exit_reason): (i32, &'static str) = 'run: { - // Set up runtime + inbox + orchestrator name, same as seed mode. - let mut rt = Runtime::new(RuntimeConfig::default()); - let codecs = Arc::new(inference_codec_registry()); - let router = Arc::new(TransportRouter::new()); - rt.set_codec_registry(codecs.clone()); - rt.set_transport_router(router.clone()); let response_inbox = match rt.new_inbox::() { Ok(i) => i, Err(e) => { - eprintln!("pp-smoke-run: new_inbox failed: {e}"); + eprintln!("pp-orchestrator: new_inbox failed: {e}"); break 'run (1, "new_inbox_error"); } }; @@ -1720,7 +2064,7 @@ fn run_vastai(args: &Args) -> i32 { .and_then(|s| s.trim().parse().ok()) .unwrap_or(180); eprintln!( - "pp-smoke-run: waiting for SWIM convergence ({} alive peers, {}s budget)...", + "pp-orchestrator: waiting for SWIM convergence ({} alive peers, {}s budget)...", num_stages, orch_converge_secs, ); let conv_res = await_convergence( @@ -1730,16 +2074,17 @@ fn run_vastai(args: &Args) -> i32 { || { driver.recv(); driver.tick(); - driver - .snapshot() - .members - .iter() - .filter(|m| m.state == "alive") - .count() + let snap = driver.snapshot(); + // Keep the dashboard membership/topology panels live as peers + // join during convergence. + if let Some(cached) = dist_cached.as_ref() { + *cached.lock().unwrap() = Some(snap.clone()); + } + snap.members.iter().filter(|m| m.state == "alive").count() }, ); if let Err(e) = conv_res { - eprintln!("pp-smoke-run: {e}"); + eprintln!("pp-orchestrator: {e}"); break 'run (1, "convergence_error"); } @@ -1747,12 +2092,12 @@ fn run_vastai(args: &Args) -> i32 { .node_mut() .register_name(ORCHESTRATOR_NAME.into(), inbox_addr); diag::emit_register_name(&driver, ORCHESTRATOR_NAME, inbox_addr, None); - eprintln!("pp-smoke-run: registered {ORCHESTRATOR_NAME} -> {inbox_addr:?}"); + eprintln!("pp-orchestrator: registered {ORCHESTRATOR_NAME} -> {inbox_addr:?}"); // Spec §4.5 + §4.6: gate the drive on (a) every pp-stage-K // resolvable and (b) pp-entry resolvable. Both are proxies for // "all stage workers ready and pipeline wired" because the - // per-index name is published post-worker-ready by pp-gpu-node. + // per-index name is published post-worker-ready by pp-worker. // 1200s covers an ~18 GB MoE GGUF (e.g. qwen3:30b-a3b) // downloading in parallel on N nodes even when some have slow // links; smaller models resolve in a fraction of this. @@ -1766,6 +2111,10 @@ fn run_vastai(args: &Args) -> i32 { let (stage0_addr, stage0_node_id) = loop { driver.recv(); driver.tick(); + // Keep the dashboard panels live while the pipeline wires up. + if let Some(cached) = dist_cached.as_ref() { + *cached.lock().unwrap() = Some(driver.snapshot()); + } for k in 0..num_stages { if roster_hex[k as usize].is_some() { continue; @@ -1788,7 +2137,7 @@ fn run_vastai(args: &Args) -> i32 { .filter_map(|(k, o)| o.is_none().then_some(k as u32)) .collect(); eprintln!( - "pp-smoke-run: pipeline did not wire within {resolve_secs}s; \ + "pp-orchestrator: pipeline did not wire within {resolve_secs}s; \ missing pp-stage-K for {missing:?} (entry resolved: {})", entry.is_some(), ); @@ -1813,14 +2162,14 @@ fn run_vastai(args: &Args) -> i32 { let key = match PublicKey::from_bytes(&stage0_node_id.0) { Ok(k) => k, Err(e) => { - eprintln!("pp-smoke-run: invalid stage-0 node key: {e}"); + eprintln!("pp-orchestrator: invalid stage-0 node key: {e}"); break 'run (1, "stage0_key_error"); } }; // Enrich the EndpointAddr with stage 0's relay URL (from SWIM // metadata gossip) or our own home relay as a fallback, so iroh has // routing info even if it has never dialed stage 0 directly. See - // pp-gpu-node::build_route for the same rationale on the worker side. + // pp-worker::build_route for the same rationale on the worker side. let mut stage0_endpoint = iroh::EndpointAddr::from(key); if let Some(url) = driver .node() @@ -1839,83 +2188,54 @@ fn run_vastai(args: &Args) -> i32 { )); router.add_route(stage0_addr, route); - // Spec §4.5: emit one pp_stage_roster per drive, before any - // request injection event. The roster was built above by - // resolving every pp-stage-K. - driver.emit(distribution::diagnostics::event::Event::Custom { - kind: "pp_stage_roster".into(), - fields: stage_roster_event_fields(drive_seq, &roster), - }); - // Spec §4.6: emit exactly one pp_pipeline_wired per drive once - // every stage is ready, every neighbour is wired (proxied by - // pp-stage-K registration being post-ready), and pp-entry is - // resolved. - driver.emit(distribution::diagnostics::event::Event::Custom { - kind: "pp_pipeline_wired".into(), - fields: serde_json::json!({ "drive_seq": drive_seq }), - }); - - let request = InferenceRequest { - reply_to: inbox_addr, - prompt: args.prompt.clone(), - max_tokens: args.max_tokens, - }; - // Mark this drive's slice of the event stream so a bundle - // reader can split events by drive. - driver.emit(distribution::diagnostics::event::Event::Custom { - kind: "pp_drive_start".into(), - fields: serde_json::json!({ - "drive_seq": drive_seq, - "prompt": args.prompt, - "max_tokens": args.max_tokens, - "num_stages": num_stages, - "label": label, - }), - }); - if let Err(e) = rt.send_to(stage0_addr, request) { - eprintln!("pp-smoke-run: send_to failed: {e}"); - break 'run (1, "send_to_error"); - } - - let await_secs = await_response_timeout_secs(600); - let result = await_response( + // First (canonical) drive: args.prompt at drive_seq 1. drive_once emits + // this drive's pp_stage_roster / pp_pipeline_wired / pp_drive_start / + // pp_drive_end markers, sends the request to stage 0, and prints the + // response. The roster + stage-0 route were resolved once above and are + // reused for every drive. + let (code, reason) = match drive_once( &mut driver, &rt, &codecs, &response_inbox, - Duration::from_secs(await_secs), - None, + stage0_addr, &roster, + num_stages, + &label, + &args.prompt, + args.max_tokens, drive_seq, - None, - ); + ) { + Ok(_) => (0, "ok"), + Err(e) => (1, e.exit_reason()), + }; - match result { - Ok(text) => { - println!("=== pipeline-parallel Inference Response ==="); - println!("{text}"); - println!("============================================"); - (0, "ok") - } - Err(e) => { - eprintln!("pp-smoke-run: {e}"); - (1, e.exit_reason()) - } + // Live multi-prompt loop, gated on the dashboard or --hold. A plain + // one-shot (neither set) keeps today's single-drive-then-exit behaviour. + // Modeled on hold_open: a stdin-reader side thread feeds prompt lines + // while the main thread pumps the driver (~200 ms) and refreshes the + // distribution snapshot so SWIM + the dashboard SSE stay live between + // prompts. EOF / blank line / `quit` leaves the loop, after which the + // existing finalize + teardown tail runs. + if std::env::var_os("PP_DASHBOARD").is_some() || args.hold { + prompt_loop( + &mut driver, + &rt, + &codecs, + &response_inbox, + stage0_addr, + &roster, + num_stages, + &label, + args.max_tokens, + drive_seq, + dist_cached.as_ref(), + dashboard.as_ref().map(|(_, _, p)| *p), + ); } - }; - // Close out this drive's slice of the event stream — emitted - // before finalize so the boundary marker lands in staging even - // when --hold intentionally skips finalize. - driver.emit(distribution::diagnostics::event::Event::Custom { - kind: "pp_drive_end".into(), - fields: serde_json::json!({ - "drive_seq": drive_seq, - "exit_reason": exit_reason, - "elapsed_ms": drive_start_instant.elapsed().as_millis() as u64, - "code": code, - }), - }); + (code, reason) + }; // Finalize policy: only one-shot runs seal a canonical bundle on // exit. --hold leaves the cluster running and the bundle window @@ -1936,12 +2256,12 @@ fn run_vastai(args: &Args) -> i32 { // Teardown policy: --hold leaves the cluster running so it can be // iterated on; only the default one-shot tears down on exit. if is_held { - eprintln!("pp-smoke-run: HOLDING cluster (label={label}, contracts={contract_ids:?})"); - eprintln!(" destroy when done: pp-smoke-run --vastai --api-key --teardown --state {}", state_path.display()); + eprintln!("pp-orchestrator: HOLDING cluster (label={label}, contracts={contract_ids:?})"); + eprintln!(" destroy when done: pp-orchestrator --vastai --api-key --teardown --state {}", state_path.display()); eprintln!(" inspect: vastai show instances (label {label})"); } else { // Default one-shot: always destroy rented instances, even on failure. - eprintln!("pp-smoke-run: destroying instances {contract_ids:?}"); + eprintln!("pp-orchestrator: destroying instances {contract_ids:?}"); let results = tokio_rt.block_on( pipeline_parallel_inference::vastai::destroy_all_instances( &http, @@ -1952,7 +2272,7 @@ fn run_vastai(args: &Args) -> i32 { ); for (id, r) in contract_ids.iter().zip(results.iter()) { if let Err(e) = r { - eprintln!("pp-smoke-run: destroy {id} failed: {e}"); + eprintln!("pp-orchestrator: destroy {id} failed: {e}"); } } } diff --git a/examples/pipeline-parallel-inference/src/bin/pp_gpu_node.rs b/examples/pipeline-parallel-inference/src/bin/pp_worker.rs similarity index 93% rename from examples/pipeline-parallel-inference/src/bin/pp_gpu_node.rs rename to examples/pipeline-parallel-inference/src/bin/pp_worker.rs index 4dc0969..ea28f1e 100644 --- a/examples/pipeline-parallel-inference/src/bin/pp_gpu_node.rs +++ b/examples/pipeline-parallel-inference/src/bin/pp_worker.rs @@ -1,4 +1,4 @@ -//! pp-gpu-node — pipeline-parallel GPU inference node. +//! pp-worker — pipeline-parallel GPU inference node. //! //! Boots one stage of a pipeline-parallel inference run. Reads its //! configuration from the environment (set at instance-create time on @@ -68,7 +68,7 @@ use swactor_process::{ProcessMode, ProcessSpec}; /// SWIM name the orchestrator uses to publish the address of its /// `InferenceResponse` inbox. The last stage resolves this name to learn /// where to send the final response. Defined here (and re-declared in -/// `pp-smoke-run`) so the topology module stays test-shaped; the binary +/// `pp-orchestrator`) so the topology module stays test-shaped; the binary /// is the only place that cares about this name. const ORCHESTRATOR_NAME: &str = "pp-orchestrator"; @@ -88,7 +88,7 @@ fn parse_hex_node_id(s: &str) -> [u8; 32] { fn require_env(name: &str) -> String { std::env::var(name) .unwrap_or_else(|_| { - eprintln!("pp-gpu-node: env {name} is required"); + eprintln!("pp-worker: env {name} is required"); std::process::exit(2); }) .trim() @@ -98,7 +98,7 @@ fn require_env(name: &str) -> String { fn require_u32(name: &str) -> u32 { let raw = require_env(name); raw.parse::().unwrap_or_else(|_| { - eprintln!("pp-gpu-node: env {name}={raw:?} must be a u32"); + eprintln!("pp-worker: env {name}={raw:?} must be a u32"); std::process::exit(2); }) } @@ -118,7 +118,7 @@ fn stage_secret_from_env() -> Option { } if hex.len() != 64 { eprintln!( - "pp-gpu-node: PP_STAGE_SECRET must be 64 hex chars, got {}", + "pp-worker: PP_STAGE_SECRET must be 64 hex chars, got {}", hex.len() ); std::process::exit(2); @@ -126,7 +126,7 @@ fn stage_secret_from_env() -> Option { let mut bytes = [0u8; 32]; for (i, b) in bytes.iter_mut().enumerate() { *b = u8::from_str_radix(&hex[i * 2..i * 2 + 2], 16).unwrap_or_else(|_| { - eprintln!("pp-gpu-node: PP_STAGE_SECRET is not valid hex"); + eprintln!("pp-worker: PP_STAGE_SECRET is not valid hex"); std::process::exit(2); }); } @@ -266,7 +266,7 @@ fn build_route( fn register_name(driver: &mut IrohDriver, name: &str, addr: ActorAddress, stage: u32) { driver.node_mut().register_name(name.into(), addr); diag::emit_register_name(driver, name, addr, Some(stage)); - eprintln!("pp-gpu-node: registered {name} -> {addr:?}"); + eprintln!("pp-worker: registered {name} -> {addr:?}"); } fn resolve_or_die( @@ -274,10 +274,10 @@ fn resolve_or_die( name: &str, timeout: Duration, ) -> (ActorAddress, String) { - eprintln!("pp-gpu-node: resolving {name}..."); + eprintln!("pp-worker: resolving {name}..."); resolve_name(driver, name, timeout).unwrap_or_else(|| { eprintln!( - "pp-gpu-node: failed to resolve {name} in {:.0}s", + "pp-worker: failed to resolve {name} in {:.0}s", timeout.as_secs_f32() ); std::process::exit(1); @@ -294,16 +294,16 @@ fn add_route_or_die( match build_route(driver, node_hex) { Ok(t) => router.add_route(addr, t), Err(e) => { - eprintln!("pp-gpu-node: route to {label} failed: {e}"); + eprintln!("pp-worker: route to {label} failed: {e}"); std::process::exit(1); } } } /// Ask the kernel to deliver `SIGTERM` to this process when its parent dies. -/// Without this, a `SIGKILL` to `pp-smoke-run` would orphan its children to +/// Without this, a `SIGKILL` to `pp-orchestrator` would orphan its children to /// pid 1 and leave them running — the orchestrator's `ChainGuard::drop` runs -/// only on graceful exit. With it, each `pp-gpu-node` dies seconds after its +/// only on graceful exit. With it, each `pp-worker` dies seconds after its /// orchestrator does, which is the `binary_e2e_orchestrator_sigkilled_*` /// contract from TEST_SPEC §13.2. Linux-only; other platforms are no-ops. #[cfg(target_os = "linux")] @@ -322,7 +322,7 @@ fn install_parent_death_signal() {} /// Set by the `SIGHUP` handler; polled by the pump loops to drive an /// in-place worker hot-reload (re-exec the on-disk worker script). An /// operator pushes a new `pp_tinygrad_worker.py` over the running one and -/// `kill -HUP $(pidof pp-gpu-node)` to pick it up without re-leasing. +/// `kill -HUP $(pidof pp-worker)` to pick it up without re-leasing. static RELOAD_REQUESTED: AtomicBool = AtomicBool::new(false); #[cfg(target_os = "linux")] @@ -352,7 +352,7 @@ fn install_sighup_handler() {} /// pump so both honour reloads with the same latency. fn drain_reload_request(rt: &Runtime, stage_actor_addr: ActorAddress) { if RELOAD_REQUESTED.swap(false, Ordering::SeqCst) { - eprintln!("pp-gpu-node: SIGHUP — reloading worker"); + eprintln!("pp-worker: SIGHUP — reloading worker"); let _ = rt.send_to(stage_actor_addr, StageMsg::ReloadWorker); } } @@ -366,7 +366,7 @@ fn maybe_simulate_boot_delay(stage: u32) { let secs = std::env::var("PP_BOOT_DELAY_SECS").ok().and_then(|s| s.trim().parse::().ok()); if let (Some(target), Some(secs)) = (target, secs) { if target == stage && secs > 0 { - eprintln!("pp-gpu-node: simulated boot delay of {secs}s on stage {stage}"); + eprintln!("pp-worker: simulated boot delay of {secs}s on stage {stage}"); std::thread::sleep(Duration::from_secs(secs)); } } @@ -378,7 +378,7 @@ fn main() { let num_stages = require_u32("NUM_STAGES"); if num_stages < 2 || stage >= num_stages { eprintln!( - "pp-gpu-node: invalid STAGE={stage} for NUM_STAGES={num_stages} \ + "pp-worker: invalid STAGE={stage} for NUM_STAGES={num_stages} \ (need NUM_STAGES >= 2 and STAGE < NUM_STAGES; N=1 is not supported)" ); std::process::exit(2); @@ -441,7 +441,7 @@ fn main() { let vastai_forwarder = _vastai_in_vm.as_ref().map(|m| m.forwarder()); // Stamp the bundle the moment this process announces itself, so a - // bundle reader can tell two pp-gpu-node incarnations of the same + // bundle reader can tell two pp-worker incarnations of the same // stage apart: a manual binary swap (the operator runbook) pkills the // old process and setsid's a new one under the same PID-1 env, which // means the same run_id + node_id, but the pid differs. The event carries that @@ -468,7 +468,7 @@ fn main() { .map(|sa| sa.to_string()) .collect(); eprintln!( - "pp-gpu-node: stage {stage}/{num_stages} ({role:?}) started (node_id: {my_hex})" + "pp-worker: stage {stage}/{num_stages} ({role:?}) started (node_id: {my_hex})" ); // PP_GPU_NODE_ADDR is printed to stdout (flushed) so a parent process // capturing this child's stdout can extract our addressing. The orchestrator @@ -485,7 +485,7 @@ fn main() { let mut seed_addr = iroh::EndpointAddr::from(seed_key); if let Some(relay) = seed_relay_env.as_deref() { if let Ok(relay_url) = relay.trim().parse::() { - eprintln!("pp-gpu-node: using seed relay {relay}"); + eprintln!("pp-worker: using seed relay {relay}"); seed_addr = seed_addr.with_relay_url(relay_url); } } @@ -525,7 +525,7 @@ fn main() { peer_addr = peer_addr.with_ip_addr(sa); } } - eprintln!("pp-gpu-node: also joining {label} {peer_hex}"); + eprintln!("pp-worker: also joining {label} {peer_hex}"); targets.push(peer_addr); } }; @@ -545,7 +545,7 @@ fn main() { "first-stage peer", ); - eprintln!("pp-gpu-node: joining seed {seed_hex}"); + eprintln!("pp-worker: joining seed {seed_hex}"); driver.join(&join_targets); // If we ended up with a relay (vast.ai / WAN), publish it via SWIM @@ -556,7 +556,7 @@ fn main() { // the autoregressive feedback edge (last → first) because SWIM has // not yet probed that specific pair. if let Some(home) = driver.home_relay_url() { - eprintln!("pp-gpu-node: publishing home relay {home} to SWIM gossip"); + eprintln!("pp-worker: publishing home relay {home} to SWIM gossip"); driver.node_mut().set_relay_url(Some(home.to_string())); } @@ -578,10 +578,10 @@ fn main() { .and_then(|s| s.trim().parse().ok()) .unwrap_or(1200); if !wait_for_cluster(&mut driver, Duration::from_secs(converge_secs)) { - eprintln!("pp-gpu-node: cluster did not converge in {converge_secs}s"); + eprintln!("pp-worker: cluster did not converge in {converge_secs}s"); std::process::exit(1); } - eprintln!("pp-gpu-node: cluster converged"); + eprintln!("pp-worker: cluster converged"); // Create the actor runtime, codec registry, and transport router. let mut rt = Runtime::new(RuntimeConfig::default()); @@ -617,7 +617,7 @@ fn main() { if let Some((handle, collector, port)) = &stage_dash { handle.set_runtime(Arc::clone(&rt), Arc::clone(collector)); handle.start_http(driver.tokio_handle()); - eprintln!("pp-gpu-node: stage {stage} dashboard on http://localhost:{port}"); + eprintln!("pp-worker: stage {stage} dashboard on http://localhost:{port}"); } run_stage( @@ -655,17 +655,17 @@ fn hold_until_worker_ready( if let Some(status) = status_inbox.try_recv() { match status { StageActorStatus::WorkerReady { pid } => { - eprintln!("pp-gpu-node: worker ready (pid: {pid:?})"); + eprintln!("pp-worker: worker ready (pid: {pid:?})"); return; } StageActorStatus::ProcessStarted => { - eprintln!("pp-gpu-node: worker process started"); + eprintln!("pp-worker: worker process started"); } StageActorStatus::ProcessExited { status } => { eprintln!( - "pp-gpu-node: stage-{stage} worker exited during startup: \ + "pp-worker: stage-{stage} worker exited during startup: \ {status:?}; holding (SWIM alive) — push a fixed worker.py \ - and `kill -HUP $(pidof pp-gpu-node)` to reload" + and `kill -HUP $(pidof pp-worker)` to reload" ); last_warn = Instant::now(); } @@ -674,7 +674,7 @@ fn hold_until_worker_ready( if last_warn.elapsed() >= warn_after { eprintln!( - "pp-gpu-node: stage-{stage} worker still not ready after {}s; \ + "pp-worker: stage-{stage} worker still not ready after {}s; \ holding — SIGHUP to reload the worker script", warn_after.as_secs() ); @@ -881,7 +881,7 @@ fn run_stage( let (next_addr, next_hex) = resolve_or_die(&mut driver, &next_name, neighbor_resolve_timeout); eprintln!( - "pp-gpu-node: resolved {next_name} -> {next_addr:?} on {next_hex}" + "pp-worker: resolved {next_name} -> {next_addr:?} on {next_hex}" ); add_route_or_die(&driver, &router, next_addr, &next_hex, &next_name); rt.send_to( @@ -900,7 +900,7 @@ fn run_stage( let (next_addr, next_hex) = resolve_or_die(&mut driver, &next_name, neighbor_resolve_timeout); eprintln!( - "pp-gpu-node: resolved {next_name} -> {next_addr:?} on {next_hex}" + "pp-worker: resolved {next_name} -> {next_addr:?} on {next_hex}" ); add_route_or_die(&driver, &router, next_addr, &next_hex, &next_name); rt.send_to( @@ -927,7 +927,7 @@ fn run_stage( let (orch_addr, orch_hex) = resolve_or_die(&mut driver, ORCHESTRATOR_NAME, neighbor_resolve_timeout); eprintln!( - "pp-gpu-node: resolved {feedback_name}={feedback_addr:?} on \ + "pp-worker: resolved {feedback_name}={feedback_addr:?} on \ {feedback_hex}, orch={orch_addr:?} on {orch_hex}" ); add_route_or_die( @@ -978,7 +978,7 @@ fn main_pump( msg_pump: ActorMessagePump, stage_actor_addr: ActorAddress, ) { - eprintln!("pp-gpu-node: entering main pump loop"); + eprintln!("pp-worker: entering main pump loop"); loop { driver.recv(); driver.tick(); @@ -989,10 +989,10 @@ fn main_pump( if let Some(status) = status_inbox.try_recv() { match status { StageActorStatus::ProcessExited { status } => { - eprintln!("pp-gpu-node: worker exited: {status:?}"); - eprintln!("pp-gpu-node: keeping SWIM alive for diagnostics"); + eprintln!("pp-worker: worker exited: {status:?}"); + eprintln!("pp-worker: keeping SWIM alive for diagnostics"); } - other => eprintln!("pp-gpu-node: status: {other:?}"), + other => eprintln!("pp-worker: status: {other:?}"), } } std::thread::sleep(Duration::from_millis(20)); diff --git a/examples/pipeline-parallel-inference/src/diag.rs b/examples/pipeline-parallel-inference/src/diag.rs index b8fad6d..f0c65a6 100644 --- a/examples/pipeline-parallel-inference/src/diag.rs +++ b/examples/pipeline-parallel-inference/src/diag.rs @@ -1,5 +1,5 @@ -//! Wire `crates/distribution` diagnostics into `pp-smoke-run` and -//! `pp-gpu-node` from environment variables. +//! Wire `crates/distribution` diagnostics into `pp-orchestrator` and +//! `pp-worker` from environment variables. //! //! Reading `SWACTOR_DIAG_COLLECTOR_URL` is the opt-in switch. When it is //! unset (or empty) `install_from_env` returns `None` and the binary diff --git a/examples/pipeline-parallel-inference/src/dist_page.html b/examples/pipeline-parallel-inference/src/dist_page.html index 31170f3..3e120d1 100644 --- a/examples/pipeline-parallel-inference/src/dist_page.html +++ b/examples/pipeline-parallel-inference/src/dist_page.html @@ -7,10 +7,14 @@ + + +
+
+

Swactor Runtime Dashboard

+ +
+
+
+ +
+
+

Fleet

+
+
—
Running
+
—
Fleet GPU
+
—
Spend
+
—
Stages
+
+
+ +
+

Stages

+
waiting for vast.ai records…
+
+ +
+

Logs

+
+
+
+ + + + diff --git a/examples/pipeline-parallel-inference/src/fleet_plugin.rs b/examples/pipeline-parallel-inference/src/fleet_plugin.rs index 8dc7769..fe62d15 100644 --- a/examples/pipeline-parallel-inference/src/fleet_plugin.rs +++ b/examples/pipeline-parallel-inference/src/fleet_plugin.rs @@ -1,71 +1,135 @@ //! Orchestrator "fleet" dashboard plugin — a remote-sourced vast.ai view. //! //! In production the diagnostics collector runs on a VPS: the stage containers -//! ship their vast.ai/host-metric records to it, and it folds them into a fleet -//! board model. The orchestrator runs locally and hosts the *full* swactor -//! dashboard (overview / actors / topology / distribution / netmap). To surface -//! the fleet alongside those live views, this plugin **pulls** the collector's -//! already-folded model (`GET {collector}/api/plugin/vastai/model`) ~1/s and -//! re-serves it verbatim under the same `"vastai"` name — so the Fleet page -//! renders identically to the standalone collector board, with no extra ingest. +//! ship their vast.ai/host-metric records to it. The orchestrator runs locally +//! and hosts the *full* swactor dashboard (overview / actors / topology / +//! distribution / netmap). To surface the fleet alongside those live views, this +//! plugin **subscribes** to the collector's raw record stream +//! (`GET {collector}/diag/stream/{run_id}`, an SSE feed of `LiveRecord`s) and +//! folds the vast.ai records locally — reusing the dashboard's own server-side +//! [`VastaiLivePlugin`] fold — then serves the result same-origin at +//! `/api/plugin/vastai/model` and over `/events` (the `vastai` event), so the +//! Fleet page renders without any cross-origin calls to the collector. //! //! It is the read mirror of the orchestrator's distribution broadcaster //! ([`crate::dist_broadcast`]), which *pushes* its snapshot to the same -//! collector. The poll loop tolerates an unreachable collector: a failed fetch -//! leaves the last good model in place, so a transient blip never blanks the UI. +//! collector. The stream loop reconnects on drops and tolerates an unreachable +//! collector: until records arrive, the Fleet tab simply waits. use std::collections::HashMap; -use std::sync::{Arc, Mutex}; +use std::sync::Arc; use std::time::Duration; +use dashboard::live_collector::VastaiLivePlugin; use dashboard::plugin::{DashboardPlugin, PluginResponse}; +use distribution::diagnostics::collector::protocol::{LiveRecord, RecordKind}; +use futures_util::StreamExt; -/// The Fleet page. Reuses the standalone collector board's Fleet renderer but -/// with the orchestrator dashboard's nav, so it sits beside the live tabs. +/// Rebuild a [`LiveRecord`] from one stream frame's `data:` JSON. `LiveRecord` is +/// a serialize-only wire type, so we parse the fields by hand. Returns `None` for +/// non-vast.ai kinds (e.g. the pushed dist snapshot) so we skip them cheaply. +fn parse_vastai_record(data: &str) -> Option { + let v: serde_json::Value = serde_json::from_str(data).ok()?; + let kind = RecordKind::parse(v.get("kind")?.as_str()?)?; + if !kind.is_vastai() { + return None; + } + Some(LiveRecord { + run_id: v.get("run_id").and_then(|x| x.as_str()).unwrap_or("").to_string(), + node_id: v.get("node_id").and_then(|x| x.as_str()).unwrap_or("").to_string(), + kind, + recv_ms: v.get("recv_ms").and_then(|x| x.as_u64()).unwrap_or(0), + seq: v.get("seq").and_then(|x| x.as_u64()).unwrap_or(0), + body: v.get("body").cloned().unwrap_or(serde_json::Value::Null), + }) +} + +/// The Fleet page. Reuses the Swactor Runtime Dashboard layout/CSS so it sits +/// beside the live tabs (no scrubber / Live control — this is a live-only view). const FLEET_HTML: &str = include_str!("fleet_page.html"); -/// Read-only dashboard plugin backing `/plugin/vastai`. Holds the last good fleet -/// model JSON pulled from the remote collector; serves it to the SSE stream -/// (event `vastai`) and the seed API (`/api/plugin/vastai/model`). +/// Dashboard plugin (name `"vastai"`) backing `/plugin/vastai`. Wraps the +/// dashboard's [`VastaiLivePlugin`] (which buffers records and serves the fold) +/// and feeds it from the remote collector's record stream; overrides only the +/// HTML page so the Fleet tab wears the dashboard chrome. pub struct RemoteVastaiPlugin { - /// Last successfully fetched fleet model JSON (the collector's folded board). - cached: Arc>>, + inner: Arc, } impl RemoteVastaiPlugin { pub fn new() -> Self { Self { - cached: Arc::new(Mutex::new(None)), + inner: Arc::new(VastaiLivePlugin::new()), } } - /// Spawn the poll loop on `rt`. Every ~1s it GETs - /// `{collector_url}/api/plugin/vastai/model`; a successful response with a - /// non-`null` body replaces the cache, anything else (error, non-2xx, empty, - /// `null`) leaves the last good model untouched. The task ends when `rt`'s - /// runtime is dropped at end of run. Cadence matches [`crate::dist_broadcast`]. - pub fn spawn_poller(&self, rt: &tokio::runtime::Handle, collector_url: String) { + /// Subscribe to the collector's `/diag/stream/{run_id}` SSE feed and fold each + /// vast.ai `LiveRecord` into the inner plugin. Reconnects every ~2s on drop or + /// while the collector is unreachable. Ends when `rt`'s runtime is dropped at + /// end of run. `run_id` must match what the stages ship under + /// (`SWACTOR_DIAG_RUN_ID`). + pub fn spawn_stream(&self, rt: &tokio::runtime::Handle, collector_url: String, run_id: String) { let url = format!( - "{}/api/plugin/vastai/model", - collector_url.trim_end_matches('/') + "{}/diag/stream/{}", + collector_url.trim_end_matches('/'), + run_id ); - let cached = Arc::clone(&self.cached); + let inner = Arc::clone(&self.inner); + eprintln!("pp-orchestrator: fleet tab streaming vast.ai records from {url}"); rt.spawn(async move { let http = reqwest::Client::new(); - let mut ticker = tokio::time::interval(Duration::from_secs(1)); - ticker.set_missed_tick_behavior(tokio::time::MissedTickBehavior::Skip); + // Log the first record and the first error once, so a blank Fleet tab + // is easy to localize (collector empty vs. stream unreachable) without + // spamming the orchestrator log. + let mut logged_first = false; + let mut logged_err = false; loop { - ticker.tick().await; - let body = match http.get(&url).send().await { - Ok(resp) if resp.status().is_success() => resp.text().await.ok(), - _ => None, - }; - if let Some(body) = body { - let trimmed = body.trim(); - if !trimmed.is_empty() && trimmed != "null" { - *cached.lock().unwrap() = Some(body); + match http.get(&url).send().await { + Ok(resp) if resp.status().is_success() => { + let mut stream = resp.bytes_stream(); + let mut buf: Vec = Vec::new(); + while let Some(chunk) = stream.next().await { + let Ok(chunk) = chunk else { break }; + buf.extend_from_slice(&chunk); + // SSE frames are separated by a blank line. Parse on + // byte boundaries so a chunk split mid-frame is safe. + while let Some(idx) = buf.windows(2).position(|w| w == b"\n\n") { + let frame: Vec = buf.drain(..idx + 2).collect(); + let Ok(text) = std::str::from_utf8(&frame) else { + continue; + }; + for line in text.lines() { + let Some(data) = line.strip_prefix("data:") else { + continue; + }; + let data = data.trim_start(); + if let Some(rec) = parse_vastai_record(data) { + inner.ingest(&rec); + if !logged_first { + logged_first = true; + eprintln!( + "pp-orchestrator: fleet tab is receiving vast.ai records from the collector" + ); + } + } + } + } + } + } + Ok(resp) => { + if !logged_err { + logged_err = true; + eprintln!("pp-orchestrator: fleet stream got HTTP {}", resp.status()); + } + } + Err(e) => { + if !logged_err { + logged_err = true; + eprintln!("pp-orchestrator: fleet stream error (collector reachable?): {e}"); + } } } + tokio::time::sleep(Duration::from_secs(2)).await; } }); } @@ -83,28 +147,17 @@ impl DashboardPlugin for RemoteVastaiPlugin { } fn snapshot_json(&self) -> Option { - self.cached.lock().unwrap().clone() + self.inner.snapshot_json() } fn handle_request( &self, method: &str, path: &str, - _query: &HashMap, - _body: &[u8], + query: &HashMap, + body: &[u8], ) -> PluginResponse { - // The page seeds from `/api/plugin/vastai/model` (the route requires a - // non-empty trailing segment), matching the standalone collector board. - match (method, path) { - ("GET", "" | "model") => PluginResponse::json( - self.cached - .lock() - .unwrap() - .clone() - .unwrap_or_else(|| "null".into()), - ), - _ => PluginResponse::not_found(), - } + self.inner.handle_request(method, path, query, body) } fn html_page(&self) -> Option<&str> { diff --git a/examples/pipeline-parallel-inference/src/lib.rs b/examples/pipeline-parallel-inference/src/lib.rs index c3b1c4a..e914445 100644 --- a/examples/pipeline-parallel-inference/src/lib.rs +++ b/examples/pipeline-parallel-inference/src/lib.rs @@ -1,6 +1,7 @@ pub mod diag; pub mod dist_broadcast; pub mod dist_plugin; +pub mod fleet_plugin; pub mod messages; pub mod netmap_plugin; pub mod profile; diff --git a/examples/pipeline-parallel-inference/src/netmap_page.html b/examples/pipeline-parallel-inference/src/netmap_page.html index f65ef14..e0ea5c8 100644 --- a/examples/pipeline-parallel-inference/src/netmap_page.html +++ b/examples/pipeline-parallel-inference/src/netmap_page.html @@ -7,10 +7,14 @@