#!/usr/bin/env python3
"""
DIFFUSION THROUGHPUT MATRIX — what actually makes tile generation cheaper.
hyper-games #46. Streaming terrain is bounded by two costs measured earlier:
~4.5 s fixed per request plus area, and 3.5 GB of resident VRAM with a 5 s
85-92% GPU burst per tile. This finds which knobs move those.
The levers, all read from source rather than guessed:
--dtype fp16 Turing (sm_75) has fp16 tensor cores; bf16 needs Ampere, so
it is not tested here.
onestep_latent world_pipeline.py:1253 -- the detail stage runs an initial
latent pass plus one intermediate (T_INTER). This drops the
intermediate. Roughly half the latent work.
--compile torch.compile, ON by default in the CLI and which I had
been disabling. Pays a large one-off cost, so the harness
warms up before timing.
--batch-size latent batch.
Each config runs in its OWN server on a scratch port, so the service the lookdev
app uses is never touched. Every tile is at fresh coordinates -- a cached tile
returns in 0.04 s and would silently fake any result here.
"""
import json, os, subprocess, sys, time, urllib.request, signal
PORT = 8819
MODEL = "xandergos/terrain-diffusion-30m"
VENV = "/storage/projects/code/godot/terrain-diffusion/.venv/bin/python"
CWD = "/storage/projects/code/godot/terrain-diffusion"
ORIGIN = 51_000_000
CHECK_AT = 55_555_000 # one fixed region every config is validated on # far from anything this session has touched
def wait_up(proc, timeout=600):
t0 = time.time()
while time.time() - t0 < timeout:
if proc.poll() is not None:
return False
try:
urllib.request.urlopen(f"http://127.0.0.1:{PORT}/health", timeout=2).read()
return True
except Exception:
time.sleep(2)
return False
REF = {} # reference tile from a config known to be correct
def tile_data(i1, j1, n):
"""Elevation + climate, for CORRECTNESS checking."""
import numpy as np
u = (f"http://127.0.0.1:{PORT}/terrain?i1={i1}&j1={j1}"
f"&i2={i1+n}&j2={j1+n}&scale=1")
with urllib.request.urlopen(u, timeout=3600) as r:
h = int(r.headers["X-Height"]); w = int(r.headers["X-Width"]); b = r.read()
e = np.frombuffer(b[:h*w*2], dtype="<i2").astype(float).reshape(h, w)
c = np.frombuffer(b[h*w*2:], dtype="<f4").reshape(h, w, 4) if len(b) > h*w*2 else None
return e, c
def correctness(i1, j1, n=256):
"""A config that returns ZEROS is infinitely fast. This is the gate that was
missing from the first three runs, and its absence shipped a 14x result that
was 70% blank tile."""
import numpy as np
e, c = tile_data(i1, j1, n)
zeros = float((e == 0).mean())
nans = int(np.isnan(c).sum()) if c is not None else 0
corr = float("nan")
if "e" in REF and REF["e"].shape == e.shape:
a, b_ = REF["e"].ravel(), e.ravel()
if a.std() > 0 and b_.std() > 0:
corr = float(np.corrcoef(a, b_)[0, 1])
ok = zeros < 0.02 and nans == 0 and (corr != corr or corr > 0.98)
return dict(zeros=zeros, nans=nans, corr=corr, ok=ok)
def tile(i1, j1, n):
u = (f"http://127.0.0.1:{PORT}/terrain?i1={i1}&j1={j1}"
f"&i2={i1+n}&j2={j1+n}&scale=1")
t0 = time.time()
with urllib.request.urlopen(u, timeout=3600) as r:
r.read()
return time.time() - t0
def gpu_apps():
"""{pid: MiB} for every process holding a CUDA context."""
try:
out = subprocess.run(["nvidia-smi", "--query-compute-apps=pid,used_memory",
"--format=csv,noheader,nounits"],
capture_output=True, text=True, timeout=20).stdout
d = {}
for line in out.strip().splitlines():
a, b = [x.strip() for x in line.split(",")[:2]]
if a.isdigit() and b.isdigit():
d[int(a)] = int(b)
return d
except Exception:
return {}
def gpu_mem_of(pid):
"""This server's OWN footprint. Taking the max across all apps is what made
every row of the first run report the dying previous server's fp32 3524 MiB."""
apps = gpu_apps()
if pid in apps:
return apps[pid]
# the server may be a child of the process group we launched
return max((m for p, m in apps.items() if abs(p - pid) < 8), default=0)
def wait_gpu_clear(exclude, timeout=90):
"""Do not start the next config until the previous context is really gone."""
t0 = time.time()
while time.time() - t0 < timeout:
if not {p: m for p, m in gpu_apps().items() if p not in exclude}:
return True
time.sleep(2)
return False
def run(label, extra, n=512, reps=3, offset=0):
cmd = [VENV, "-m", "terrain_diffusion.inference.api", MODEL,
"--seed", "1234", "--host", "127.0.0.1", "--port", str(PORT),
"--log-mode", "info"] + extra
log = open(f"/tmp/bench-{label.replace(' ','_')}.log", "w")
p = subprocess.Popen(cmd, cwd=CWD, stdout=log, stderr=subprocess.STDOUT,
preexec_fn=os.setsid)
try:
if not wait_up(p):
print(f" {label:<28} FAILED TO START (see /tmp/bench-{label.replace(' ','_')}.log)")
return None
base = ORIGIN + offset * 200_000
warm = tile(base, base, 256) # pay compile / lazy init here
ts = [tile(base + 20_000*(k+1), base + 20_000*(k+1), n) for k in range(reps)]
mem = gpu_mem_of(p.pid)
# ALWAYS the same coordinates, so every config is checked on one region.
ck = correctness(CHECK_AT, CHECK_AT)
if not REF and ck["zeros"] < 0.02 and ck["nans"] == 0:
REF["e"], _ = tile_data(CHECK_AT, CHECK_AT, 256)
best, avg = min(ts), sum(ts)/len(ts)
km2 = (n*30/1000)**2
flag = "ok " if ck["ok"] else "BAD "
print(f" {label:<26} {flag} tile {avg:6.2f}s {km2/avg:6.1f} km2/s "
f"{mem:5d} MiB zeros {100*ck['zeros']:5.1f}% nan {ck['nans']:6d} "
f"corr {ck['corr']:.4f}")
return dict(label=label, avg=avg, best=best, mem=mem, **ck)
finally:
try: os.killpg(os.getpgid(p.pid), signal.SIGTERM)
except Exception: pass
p.wait(timeout=60)
wait_gpu_clear(exclude=set(BASELINE_PIDS))
BASELINE_PIDS = set(gpu_apps().keys()) # the always-on service, not ours to wait on
if __name__ == "__main__":
print(f"512 tile = 15.36 km across = 236 km2. Fresh coordinates every time.\n")
print(f" {'config':<26} {'ok?':<4} {'per tile':>12} {'throughput':>11} "
f"{'VRAM':>9} correctness")
rows = []
os.environ.pop("TD_COMPILE_MODE", None)
F16 = ["--no-compile", "--dtype", "fp16", "--kwarg", "onestep_latent=True"]
for i, (label, extra) in enumerate([
("fp16+onestep (reference)", F16), # known good
(" + decoder_tile 256", F16 + ["--kwarg", "decoder_tile_size=256"]),
(" + decoder_tile 128", F16 + ["--kwarg", "decoder_tile_size=128"]),
(" + compile only", ["--compile", "--dtype", "fp16",
"--kwarg", "onestep_latent=True"]),
(" + compile + dtile 256", ["--compile", "--dtype", "fp16",
"--kwarg", "onestep_latent=True",
"--kwarg", "decoder_tile_size=256"]),
("fp32 plain (control)", ["--no-compile"]),
]):
r = run(label, extra, offset=i)
if r: rows.append(r)
if rows:
b = rows[0]["avg"]
print("\n speedup vs the reference config — ONLY meaningful where ok:")
for r in rows:
print(f" {r['label']:<26} {'ok ' if r['ok'] else 'BAD'} "
f"{b/r['avg']:5.2f}x VRAM {r['mem']:5d} MiB")
bad = [r["label"] for r in rows if not r["ok"]]
if bad:
print("\n DISCARD (wrong output, so the timing is meaningless): "
+ ", ".join(x.strip() for x in bad))