fleet: swap.py (one call per shipper word over the fleet, the hub first), the rig as N provers (per-card loops, distinct sockets), the rig script's device and chain fixes, restage.py
This commit is contained in:
igneum-labs
parent
490f0e41d4
commit
2b45606a52
5 changed files with 90 additions and 13 deletions
|
|
@ -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<n>/; 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
|
||||
|
|
|
|||
11
tools/fleet/box-rig-prover.sh
Executable file
11
tools/fleet/box-rig-prover.sh
Executable file
|
|
@ -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 </dev/null > $F/card$d/out/launch.log 2>&1 & )
|
||||
done
|
||||
echo "RESULT rig_provers_started $(date -u +%Y-%m-%dT%H:%M:%SZ) cards=$N"
|
||||
|
|
@ -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() { # <device> <mode> <threshold> <seconds>
|
||||
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)"
|
||||
|
|
|
|||
18
tools/fleet/restage.py
Normal file
18
tools/fleet/restage.py
Normal file
|
|
@ -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)
|
||||
39
tools/fleet/swap.py
Normal file
39
tools/fleet/swap.py
Normal file
|
|
@ -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 >/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()
|
||||
Loading…
Reference in a new issue