diff --git a/tools/fleet/box-prover.py b/tools/fleet/box-prover.py index 4ba6794d4..dd426becf 100644 --- a/tools/fleet/box-prover.py +++ b/tools/fleet/box-prover.py @@ -21,12 +21,14 @@ Env: LABEL (the key label, kept for the box's life), WALLET (payout), THRESHOLD MINER (keep|pause), RUN_HOURS (default 9). """ import json, os, sys, time, subprocess, datetime, binascii, urllib.request, signal -F = "/root/fleet"; OUT = f"{F}/out"; B = "/opt/igneum/pkg/bin"; FLOOR = "/opt/igneum-floor" +F = "/root/fleet"; CARD = os.environ.get("FLEET_CARD"); OUT = f"{F}/card{CARD}/out" if CARD else f"{F}/out" # a rig runs one loop per card under /root/fleet/card/; B = "/opt/igneum/pkg/bin"; FLOOR = "/opt/igneum-floor" HOST = f"{FLOOR}/bin/igneum-prove-host"; EXPORT = f"{FLOOR}/bin/igneum-prove-export" EVM = "http://127.0.0.1:26790"; GRPC = "grpc://127.0.0.1:26610"; CHAIN = "igneum-devnet" LABEL = os.environ.get("LABEL", "box"); WALLET = os.environ.get("WALLET", "0x" + "19" * 20) THRESHOLD = os.environ.get("THRESHOLD", ""); MINER = os.environ.get("MINER", "keep"); RUN_HOURS = float(os.environ.get("RUN_HOURS", "9")) -os.makedirs(f"{OUT}/segs", exist_ok=True); os.makedirs(f"{F}/mine/packs", exist_ok=True) +MINE = f"{F}/card{CARD}/mine" if CARD else f"{F}/mine" +os.makedirs(f"{OUT}/segs", exist_ok=True); os.makedirs(f"{MINE}/packs", exist_ok=True) +DEV = os.environ.get("IGNEUM_CUDA_DEVICE", "0") LOG = open(f"{OUT}/prover.log", "a") def stamp(): return datetime.datetime.now(datetime.timezone.utc).strftime("%Y-%m-%dT%H:%M:%SZ") def say(s): LOG.write(f"{s}\n"); LOG.flush(); print(s, flush=True) @@ -44,20 +46,22 @@ def fnv1a(s): h = 0xcbf29ce484222325 for c in s.encode(): h ^= c; h = (h * 0x100000001b3) & 0xffffffffffffffff return h -def kill_server(): subprocess.run("pkill -f sp1-gpu-server; sleep 1; rm -f /tmp/sp1-cuda-*.sock", shell=True) +def kill_server(): + if CARD: subprocess.run(f"rm -f /tmp/sp1-cuda-{DEV}.sock", shell=True) # a rig: never another card's server + else: subprocess.run("pkill -f sp1-gpu-server; sleep 1; rm -f /tmp/sp1-cuda-*.sock", shell=True) MPROC = None def miner_start(): global MPROC if MPROC and MPROC.poll() is None: return - subprocess.run(f"cd {F}/mine && rm -rf packs/devnet && {B}/igneum-miner export-pack {GRPC} packs/devnet > {OUT}/prover-export-pack.log 2>&1", shell=True) - MPROC = subprocess.Popen([f"{B}/igneum-miner", "mine", GRPC, "1", "100000000", LABEL, "--worker", f"{B}/igneum-worker-cuda", "--worker-args", "--device 0 --pack packs/devnet", + subprocess.run(f"cd {MINE} && rm -rf packs/devnet && {B}/igneum-miner export-pack {GRPC} packs/devnet > {OUT}/prover-export-pack.log 2>&1", shell=True) + MPROC = subprocess.Popen([f"{B}/igneum-miner", "mine", GRPC, "1", "100000000", LABEL, "--worker", f"{B}/igneum-worker-cuda", "--worker-args", f"--device {DEV} --pack packs/devnet", "--prepare-packs", "packs/prepare", "--exit-on-seed-change", "--evm-address", WALLET, "--payout-label", LABEL, "--status-secs", "30"], - cwd=f"{F}/mine", stdout=open(f"{OUT}/prover-miner.log", "a"), stderr=subprocess.STDOUT) + cwd=MINE, stdout=open(f"{OUT}/prover-miner.log", "a"), stderr=subprocess.STDOUT) say(f"RESULT miner_start {stamp()} pid={MPROC.pid}") def miner_stop(): global MPROC if MPROC: MPROC.terminate(); time.sleep(2); MPROC.kill(); MPROC = None - subprocess.run("pkill -f igneum-worker-cuda", shell=True) + subprocess.run(f"pkill -f 'igneum-worker-cuda --device {DEV} '", shell=True) def miner_rate(n=6): try: vals = [float(l.split(" now=")[1].split()[0]) for l in open(f"{OUT}/prover-miner.log").read().split("\n") if "STATUS" in l and " now=" in l][-n:] @@ -187,7 +191,7 @@ while (time.time() - t_run0) / 3600 < RUN_HOURS: if THRESHOLD: env["SP1_GPU_ELEMENT_THRESHOLD"] = THRESHOLD args = [HOST, "--mode", "chain", "--chain", ",".join(fixtures), "--prover", WALLET, "--save-shards", "--out", f"{d}/chain-results.json"] if prev_file: args += ["--prev", prev_file] - samp = subprocess.Popen("while :; do nvidia-smi --query-gpu=memory.used,utilization.gpu,power.draw --format=csv,noheader,nounits; sleep 1; done", shell=True, stdout=open(f"{d}/smi.csv", "w"), stderr=subprocess.DEVNULL, start_new_session=True) + samp = subprocess.Popen(f"while :; do nvidia-smi -i {DEV} --query-gpu=memory.used,utilization.gpu,power.draw --format=csv,noheader,nounits; sleep 1; done", shell=True, stdout=open(f"{d}/smi.csv", "w"), stderr=subprocess.DEVNULL, start_new_session=True) t = time.time() try: rr = subprocess.run(args, env=env, capture_output=True, text=True, timeout=3600) except subprocess.TimeoutExpired: rr = None diff --git a/tools/fleet/box-rig-prover.sh b/tools/fleet/box-rig-prover.sh new file mode 100755 index 000000000..30da88913 --- /dev/null +++ b/tools/fleet/box-rig-prover.sh @@ -0,0 +1,11 @@ +#!/usr/bin/env bash +# The rig as N provers for the fleet night: one box-prover.py per card with its own key label, payout, output dir and +# IGNEUM_CUDA_DEVICE (distinct sockets), the miners kept on every card (24 GB cards keep mining while proving). +set -uo pipefail +F=/root/fleet; N=$(nvidia-smi --query-gpu=name --format=csv,noheader | wc -l) +LABEL="${LABEL:-rig}"; WALLET="${WALLET:-0x1919191919191919191919191919191919191919}" +for d in $(seq 0 $((N-1))); do + mkdir -p $F/card$d/out $F/card$d/mine/packs + ( cd $F/card$d && LABEL="$LABEL-gpu$d" WALLET="$WALLET" THRESHOLD="${THRESHOLD:-}" MINER="${MINER:-keep}" RUN_HOURS="${RUN_HOURS:-9}" IGNEUM_CUDA_DEVICE=$d FLEET_CARD=$d setsid nohup python3 -u $F/in/box-prover.py $F/card$d/out/launch.log 2>&1 & ) +done +echo "RESULT rig_provers_started $(date -u +%Y-%m-%dT%H:%M:%SZ) cards=$N" diff --git a/tools/fleet/box-rig.sh b/tools/fleet/box-rig.sh index e49fe4077..f20fc45ba 100755 --- a/tools/fleet/box-rig.sh +++ b/tools/fleet/box-rig.sh @@ -21,11 +21,12 @@ N=$(nvidia-smi --query-gpu=name --format=csv,noheader | wc -l) echo "RESULT rig_start $(stamp) label=$LABEL cards=$N $(nvidia-smi --query-gpu=name,memory.total --format=csv,noheader | head -1) host_ram_gb=$(( $(awk '/MemTotal/{print $2}' /proc/meminfo) / 1048576 )) cores=$(nproc)" for b in igneum-prove-host igneum-prove-export; do for d in $FLOOR/target/release $FLOOR/prove/proving/igneum-prove/target/release; do [ -x $d/$b ] && ln -sfn $d/$b $FLOOR/bin/$b; done; done # A. the installer's real parts -if [ -d $F/in/linux ]; then +if [ -d $F/in/linux ] && [ -z "${PROVE_ONLY:-}" ]; then bash $F/in/linux/install-rig.sh --preflight-only > $OUT/rig-preflight.log 2>&1; echo "RESULT installer_preflight exit=$? fails=$(grep -c 'FAIL' $OUT/rig-preflight.log) warns=$(grep -c 'warn' $OUT/rig-preflight.log) cards=$(grep -c '^ card' $OUT/rig-preflight.log)" bash $F/in/linux/install-rig.sh --dry-run --yes --wallet "$WALLET" --name "$LABEL" --package-url https://dl.igneum.network/dl/public/igneum-hive-0.3.12.tar.gz --package-sha256 7972af92e7cd9a032303eca4d95b533f53e0e68d1b9cae5bfe406a5b7c30a454 --package-size 24506282 > $OUT/rig-dryrun.log 2>&1; echo "RESULT installer_dryrun exit=$? steps=$(grep -c '^\[dry-run\]' $OUT/rig-dryrun.log) manifest=$(grep -c 'signature verifies' $OUT/rig-dryrun.log) package=$(grep -c 'sha256 7972af92' $OUT/rig-dryrun.log) prover_decision=\"$(grep -o 'prover: .*' $OUT/rig-dryrun.log | head -1 | cut -c1-120)\"" fi # B. eight miners +if [ -z "${PROVE_ONLY:-}" ]; then cd $F/mine && rm -rf packs/devnet && $B/igneum-miner export-pack grpc://127.0.0.1:26610 packs/devnet > $OUT/rig-export-pack.log 2>&1 pids=() for d in $(seq 0 $((N-1))); do @@ -43,12 +44,13 @@ for d in $(seq 0 $((N-1))); do done echo "RESULT rig_miners $(stamp) cards=$N sum_mhs=$sum watts=$(awk -F', *' '{s+=$2; n++} END {printf "%.0f", s/n*'$N'}' $OUT/rig-samp-mine.csv)" for p in "${pids[@]}"; do kill $p 2>/dev/null; done; pkill -f '^/opt/igneum/pkg/bin/igneum-worker-cuda'; sleep 3 +fi # C + D. seven core-only provers and one compressing card, in parallel pkill -9 -x sp1-gpu-server; rm -f /tmp/sp1-cuda-*.sock prove_loop() { # local d=$1 mode=$2 thr=$3 secs=$4 t0=$(date +%s) n=0 tot=0 while [ $(( $(date +%s) - t0 )) -lt $secs ]; do - env HOME=$FLOOR/home CUDA_VISIBLE_DEVICES=$d IGNEUM_CUDA_DEVICE=$d SP1_PROVER=cuda RUST_LOG=off SP1_GPU_ELEMENT_THRESHOLD=$thr timeout 600 $HOST $V1 --mode $mode --shard 0 --prover "$WALLET" --out $OUT/rig-$mode-$d.json > $OUT/rig-$mode-$d.log 2>&1 + env HOME=$FLOOR/home IGNEUM_CUDA_DEVICE=$d SP1_PROVER=cuda RUST_LOG=off SP1_GPU_ELEMENT_THRESHOLD=$thr timeout 600 $HOST $V1 --mode $mode --shard 0 --prover "$WALLET" --out $OUT/rig-$mode-$d.json > $OUT/rig-$mode-$d.log 2>&1 s=$(grep -E "^RESULT ($mode) shard" $OUT/rig-$mode-$d.log | tail -1 | grep -o 'prove [0-9.]* s' | grep -o '[0-9.]*' | head -1) v=$(grep -E "^RESULT ($mode) shard" $OUT/rig-$mode-$d.log | tail -1 | grep -c VERIFIED) [ "$v" = 1 ] && { n=$((n+1)); tot=$(awk -v a=$tot -v b=${s:-0} 'BEGIN {print a+b}'); } @@ -70,10 +72,13 @@ if [ -n "$cf" ]; then sz=$(stat -c %s "$cf"); t0=$(date +%s%N); cp "$cf" /tmp/handoff.bin; sync; t1=$(date +%s%N); cat /tmp/handoff.bin > /dev/null; t2=$(date +%s%N) echo "RESULT handoff file=$(basename $cf) bytes=$sz copy_ms=$(( (t1-t0)/1000000 )) read_ms=$(( (t2-t1)/1000000 )) loopback_tcp_ms=$( (nc -l -p 29999 > /dev/null & sleep 0.3; t=$(date +%s%N); nc -q 0 127.0.0.1 29999 < /tmp/handoff.bin; echo $(( ($(date +%s%N)-t)/1000000 )) ) 2>/dev/null)" fi -# F. the 8-block chain on the big card -list=""; for f in block-72854-empty-block-first block-58927-empty-reward block-72803-skipped-copies block-56-transfers block-78-increment block-341-shards2 block-344-shards4 block-338-shard1; do [ -f $FIX/$f.json ] && list="$list${list:+,}$FIX/$f.json"; done +# F. the 8-block chain on the big card: --mode chain needs CONSECUTIVE live blocks (the chain rule checks number and parent hash), so this +# runs only when RIG_CHAIN_LIST names eight consecutive fixtures cut from the box's node (after the exec layer executes); else skipped +list="${RIG_CHAIN_LIST:-}" +if [ -n "$list" ]; then t0=$(date +%s) -env HOME=$FLOOR/home CUDA_VISIBLE_DEVICES=$((N-1)) IGNEUM_CUDA_DEVICE=$((N-1)) SP1_PROVER=cuda RUST_LOG=off timeout 1800 $HOST --mode chain --chain "$list" --prover "$WALLET" --save-shards --out $OUT/rig-chain.json > $OUT/rig-chain.log 2>&1; rc=$? +env HOME=$FLOOR/home IGNEUM_CUDA_DEVICE=$((N-1)) SP1_PROVER=cuda RUST_LOG=off timeout 1800 $HOST --mode chain --chain "$list" --prover "$WALLET" --save-shards --out $OUT/rig-chain.json > $OUT/rig-chain.log 2>&1; rc=$? echo "RESULT chain rc=$rc wall_s=$(( $(date +%s) - t0 )) blocks=$(echo "$list" | tr ',' '\n' | wc -l) $(grep -E '^RESULT chain' $OUT/rig-chain.log | tail -1 | cut -c1-200)" +else echo "RESULT chain skipped: no consecutive live fixtures yet (needs the exec layer)"; fi pkill -9 -x sp1-gpu-server; rm -f /tmp/sp1-cuda-*.sock echo "RESULT rig_done $(stamp)" diff --git a/tools/fleet/restage.py b/tools/fleet/restage.py new file mode 100644 index 000000000..a2a876706 --- /dev/null +++ b/tools/fleet/restage.py @@ -0,0 +1,18 @@ +#!/usr/bin/env python3 +"""Replaces /root/fleet/in/igneumd-0313 on every live phase-1/2 box and the hub with the given local file, checked by +sha256 on the box before the move (a wrong binary is never left under the swap's name).""" +import sys, os, subprocess, hashlib +sys.path.insert(0, os.path.dirname(os.path.abspath(__file__))); import fleet +from concurrent.futures import ThreadPoolExecutor +src, want = sys.argv[1], sys.argv[2] +assert hashlib.sha256(open(src, "rb").read()).hexdigest() == want, "local file does not match" +reg = fleet.load() +targets = [(iid, b) for iid, b in reg.items() if b.get("state") != "destroyed" and b.get("ssh_ok") and (b.get("phase") in ("1", "2") or b.get("hub"))] +def one(x): + iid, b = x + p = subprocess.run(["scp"] + fleet.SSH_OPTS + ["-P", str(b["ssh_port"]), src, f"root@{b['ssh_host']}:/root/fleet/in/igneumd-0313.new"], capture_output=True, text=True, timeout=900) + if p.returncode: return b["label"], "scp failed " + p.stderr.strip()[:60] + r, out, _ = fleet.ssh(b, f"cd /root/fleet/in && s=$(sha256sum igneumd-0313.new | cut -c1-64) && [ \"$s\" = {want} ] && mv igneumd-0313.new igneumd-0313 && chmod +x igneumd-0313 && echo replaced ${{s:0:16}} || echo BAD $s; rm -f /opt/igneum/pkg/bin/igneumd-0313", timeout=60) + return b["label"], out.strip() +with ThreadPoolExecutor(10) as ex: + for lab, out in sorted(ex.map(one, targets)): print(f"{lab:<12} {out}", flush=True) diff --git a/tools/fleet/swap.py b/tools/fleet/swap.py new file mode 100644 index 000000000..0a28443e7 --- /dev/null +++ b/tools/fleet/swap.py @@ -0,0 +1,39 @@ +#!/usr/bin/env python3 +"""The phase-2 node swap over the fleet, one call per shipper word: + swap.py binary on "hands on 0.3.13": every phase-2 box (the hub first) restarts on igneumd-0313 with the ten-field file + swap.py file on "hands on b18ed271": the thirteen-field file swapped in and the node restarted; the replay timed + swap.py status the swap log's last line and the exec status per box +The expected sha256 and digests are pinned here; a box whose binary or digest differs refuses and is reported.""" +import sys, os, time, json +sys.path.insert(0, os.path.dirname(os.path.abspath(__file__))); import fleet +from concurrent.futures import ThreadPoolExecutor +SHA = "d6350586fe837b1f71696546628ec071b6accce2531aef776cf2b1e5487a8cdc" +DIG = {"binary": "7bd98cc4118616455709d5e32a30b799e6e67caa42d2b5d09875cd49848a7ed7", "file": "b18ed271f75dd46406d230f4156c37472127415a4c32c558bac662f6f840e61c"} +def targets(): + reg = fleet.load() + t = [(iid, b) for iid, b in reg.items() if b.get("state") != "destroyed" and b.get("ssh_ok") and (b.get("phase") in ("1", "2") or b.get("hub"))] + return sorted(t, key=lambda x: (not x[1].get("hub"), x[1]["label"])) +def start(step): + hub = next((b for _, b in targets() if b.get("hub")), None) + env = f"STEP={step} NODE_SHA256={SHA} EXPECT_DIGEST={DIG[step]} HUB_PEER={hub['hub_peer']} HUB_SSH={hub['ssh_host']} HUB_PORT={hub['ssh_port']}" if hub else f"STEP={step} NODE_SHA256={SHA} EXPECT_DIGEST={DIG[step]}" + def one(x): + iid, b = x + r, out, err = fleet.ssh(b, f"cd /root/fleet && mv out/node-swap.log out/node-swap-{step}-prev.log 2>/dev/null; {env} setsid nohup in/box-node-swap.sh /dev/null 2>&1 & sleep 1; echo started", timeout=60) + fleet.patch(iid, stage="swap-" + step, doing=f"0.3.13 swap, step {step}: restart, digest check, the exec replay timed") + return b["label"], out.strip() or err[:60] + ts = targets() + if hub: # the hub first, then the rest in parallel 20 s later (it is every box's second peer) + print(one(next(x for x in ts if x[1].get("hub")))); time.sleep(20) + with ThreadPoolExecutor(12) as ex: + for lab, out in ex.map(one, [x for x in ts if not x[1].get("hub")]): print(lab, out) +def status(): + def one(x): + iid, b = x + r, out, err = fleet.ssh(b, "grep -E '^RESULT (node_started|swap_done|swap_failed|datadir_failed)' /root/fleet/out/node-swap.log 2>/dev/null | tail -2 | cut -c1-200; grep -E '^RESULT replay' /root/fleet/out/node-swap.log 2>/dev/null | tail -1 | sed -E 's/difficulty=[0-9.]* sink=[0-9a-f]* //' | cut -c1-120; grep -E '^RESULT exec_status' /root/fleet/out/node-swap.log 2>/dev/null | tail -1 | cut -c1-200", timeout=40) + return b["label"], out.strip().replace("\n", " | ") + with ThreadPoolExecutor(12) as ex: + for lab, out in ex.map(one, targets()): print(f"{lab:<12} {out}") +if __name__ == "__main__": + c = sys.argv[1] + if c in ("binary", "file"): start(c) + elif c == "status": status()