#!/usr/bin/env python3
"""
TILE STREAMING — moving around an infinite diffusion heightmap.
Everything here rests on one property, measured earlier in #46 and not assumed:
tiles are DECOMPOSITION-INDEPENDENT and SEAM-FREE. Fetching a 512 region as one
request or as four 256 quadrants with a cold cache differs by max 1 m (int16
quantisation), and the step across a join equals the step inside a tile. So
tiles can be generated in ANY order, at ANY time, with no knowledge of their
neighbours -- which is what makes background streaming possible at all, and is
exactly the property erosion simulation lacks.
DESIGN, and the reasons are measured:
ONE worker, one generation in flight, ever.
A tile is a multi-second 85-92% GPU burst. More than one in flight and the
renderer starves. This is the entire contention mitigation.
LARGE tiles, not small.
Cost floor is ~2.4 s regardless of size. A 16 px tile costs 2.41 s and is
385x less efficient per km2 than a 1024. Small tiles do not spread the
cost, they multiply it. Default 256 (7.68 km) is a compromise with
latency; 512 is more efficient if you can wait.
DISK cache, not just memory.
The upstream 'direct' strategy is a 100 MB in-memory LRU, so a revisited
region regenerates. Elevation is cached here as int16 npy: a 256 tile is
128 KB, so an entire 500 x 500 km area is about 340 MB.
PRIORITY BY DISTANCE, EVALUATED AT DEQUEUE TIME.
This was a PriorityQueue keyed on distance-to-centre AT INSERTION. Pan,
and stale entries kept their old high priority, so the worker carried on
filling tiles near where you used to be while the tiles you were actually
looking at waited behind them -- observed as "the tiling focuses on
filling black tiles vs refining those blurry tiles". A frozen priority is
the wrong model for a queue whose ordering depends on a moving camera.
The worker now picks the nearest to the CURRENT centre each iteration.
"""
import os, queue, re, threading, time
import numpy as np
TILE_M = 30.0 # model native metres per pixel
class TileStore:
def __init__(self, fetch_fn, cache_dir, tile_px=256, log=print,
hblk_dir=None):
self.fetch = fetch_fn # (i1, j1, n) -> (elev, climate)
self.dir = cache_dir
self.tile_px = tile_px
self.log = log
# PUBLISH AS WE GENERATE, because the engine reads a DIFFERENT
# directory in a DIFFERENT format, and a manual conversion step between
# the two is a step that gets forgotten. Browsing then looks like it did
# nothing. The .npy cache stays authoritative -- .hblk is a derived
# view, regenerable from it at any time by stitch_block --per-tile.
self.hblk_dir = hblk_dir
os.makedirs(self.dir, exist_ok=True)
if self.hblk_dir:
os.makedirs(self.hblk_dir, exist_ok=True)
self._mem = {} # (tx,ty) -> np.int16 array
self._state = {} # (tx,ty) -> 'cached'|'queued'|'busy'|'error'
self._want = {} # (tx,ty) -> True, the pending set
self._wake = threading.Event()
self._lock = threading.Lock()
self._centre = (0, 0)
self.stats = dict(generated=0, from_disk=0, from_mem=0,
gen_seconds=0.0, km2=0.0, errors=0)
self._worker = threading.Thread(target=self._run, daemon=True)
self._worker.start()
# ---- addressing --------------------------------------------------------
# Tiles are stored with a 1 px APRON on every side.
#
# np.gradient falls back to ONE-SIDED differences at an array edge, so a
# tile shaded on its own has its outer row and column lit differently from
# its interior -- and adjacent tiles then mismatch exactly at the join. The
# elevation has no seam (measured: join step 6.25 m vs inner 6.14 m, ratio
# 1.02x); the seam was entirely in the shading. Shade the apron, crop it off.
#
# v2 filename, so the old apron-less cache is ignored rather than deleted.
APRON = 1
def path(self, tx, ty):
return os.path.join(self.dir, f"t2_{self.tile_px}_{tx}_{ty}.npy")
def state(self, tx, ty):
k = (tx, ty)
if k in self._mem:
return "cached"
if os.path.exists(self.path(tx, ty)):
return "disk"
return self._state.get(k, "missing")
# ---- read --------------------------------------------------------------
def get(self, tx, ty):
"""Return the tile if it is already available. NEVER generates -- the
request path must not block on the GPU, or panning stalls the UI."""
k = (tx, ty)
if k in self._mem:
self.stats["from_mem"] += 1
return self._mem[k]
p = self.path(tx, ty)
if os.path.exists(p):
a = np.load(p)
self._mem[k] = a
self.stats["from_disk"] += 1
if len(self._mem) > 400: # bounded; disk is the real cache
self._mem.pop(next(iter(self._mem)))
return a
return None
# ---- queue -------------------------------------------------------------
def request(self, tiles, centre):
"""Add to the pending set and update the centre. No ordering is decided
here -- see the note at the top of the file."""
self._centre = centre
with self._lock:
for tx, ty in tiles:
k = (tx, ty)
if k in self._want or self.state(tx, ty) in ("cached", "disk"):
continue
self._want[k] = True
self._state[k] = "queued"
self._wake.set()
def queue_depth(self):
return len(self._want)
def _take_nearest(self):
"""Pop the pending tile closest to the CURRENT centre."""
with self._lock:
if not self._want:
return None
cx, cy = self._centre
k = min(self._want, key=lambda t: (t[0] - cx) ** 2 + (t[1] - cy) ** 2)
self._want.pop(k, None)
return k
# ---- the single worker -------------------------------------------------
def hblk_path(self, tx, ty):
return os.path.join(self.hblk_dir, f"tile_{tx}_{ty}.hblk")
def _publish_hblk(self, tx, ty, arr):
"""Write the engine-format twin. Never fatal: the .npy is the cache."""
if not self.hblk_dir:
return
try:
from stitch_block import write_hblk, tile_origin
ox, oz = tile_origin(tx, ty, self.tile_px, TILE_M, self.APRON)
write_hblk(self.hblk_path(tx, ty), arr, ox, oz, TILE_M)
except Exception as ex:
self.log(f" tile ({tx},{ty}) hblk FAILED: {ex}")
def backfill_hblk(self):
"""Publish every cached .npy that has no .hblk yet.
The cache predates this path by 841 tiles, and a legacy prefix sits
alongside the current one -- so walk the SAME pattern the converter
does rather than every file in the directory.
"""
if not self.hblk_dir:
return 0
pat = re.compile(rf"^t2_{self.tile_px}_(-?\d+)_(-?\d+)\.npy$")
n = 0
for f in os.listdir(self.dir):
m = pat.match(f)
if not m:
continue
tx, ty = int(m.group(1)), int(m.group(2))
if os.path.exists(self.hblk_path(tx, ty)):
continue
try:
self._publish_hblk(tx, ty, np.load(os.path.join(self.dir, f)))
n += 1
except Exception as ex:
self.log(f" backfill ({tx},{ty}) FAILED: {ex}")
return n
def _run(self):
while True:
k = self._take_nearest()
if k is None:
self._wake.wait(0.25); self._wake.clear(); continue
tx, ty = k
if self.state(tx, ty) in ("cached", "disk"):
self._state.pop((tx, ty), None)
continue
self._state[(tx, ty)] = "busy"
t0 = time.time()
try:
n = self.tile_px
a_ = self.APRON
e, _c = self.fetch(tx * n - a_, ty * n - a_, n + 2 * a_)
a = np.clip(e, -32768, 32767).astype(np.int16)
np.save(self.path(tx, ty), a)
self._publish_hblk(tx, ty, a)
self._mem[(tx, ty)] = a
dt = time.time() - t0
self.stats["generated"] += 1
self.stats["gen_seconds"] += dt
self.stats["km2"] += (n * TILE_M / 1000.0) ** 2
self._state.pop((tx, ty), None)
self.log(f" tile ({tx},{ty}) in {dt:.2f}s "
f"queue {len(self._want)}")
except Exception as ex:
self.stats["errors"] += 1
self._state[(tx, ty)] = "error"
self.stats["last_error"] = str(ex)[:120]
self.log(f" tile ({tx},{ty}) FAILED: {ex}")
# BACK OFF. Without this a downed backend fails instantly, the
# worker takes the next tile and fails instantly too, and the
# whole queue burns in seconds -- observed as 962 errors from
# one stopped service. Failures are almost always the backend,
# and hammering it helps nobody.
self._fail_streak = getattr(self, "_fail_streak", 0) + 1
time.sleep(min(30.0, 0.5 * (2 ** min(self._fail_streak, 6))))
else:
self._fail_streak = 0
def drop_pending(self):
"""Forget everything not yet started. The in-flight tile finishes."""
with self._lock:
for k in list(self._want):
self._state.pop(k, None)
self._want.clear()
def summary(self):
s = dict(self.stats)
s["queue"] = self.queue_depth()
s["mem_tiles"] = len(self._mem)
try:
s["disk_tiles"] = len([f for f in os.listdir(self.dir)
if f.startswith(f"t2_{self.tile_px}_")])
except OSError:
s["disk_tiles"] = 0
s["avg_gen_s"] = (s["gen_seconds"] / s["generated"]) if s["generated"] else 0.0
s["km2_per_s"] = (s["km2"] / s["gen_seconds"]) if s["gen_seconds"] else 0.0
s["busy"] = [f"{k[0]},{k[1]}" for k, v in self._state.items() if v == "busy"]
return s