| … | … | @@ -27,9 +27,14 @@ DESIGN, and the reasons are measured: |
| 27 | 27 | region regenerates. Elevation is cached here as int16 npy: a 256 tile is |
| 28 | 28 | 128 KB, so an entire 500 x 500 km area is about 340 MB. |
| 29 | 29 | |
| 30 | | PRIORITY BY DISTANCE. |
| 31 | | The queue is re-sorted on every view change, so walking somewhere new |
| 32 | | abandons the old prefetch order rather than finishing it first. |
| 30 | PRIORITY BY DISTANCE, EVALUATED AT DEQUEUE TIME. |
| 31 | This was a PriorityQueue keyed on distance-to-centre AT INSERTION. Pan, |
| 32 | and stale entries kept their old high priority, so the worker carried on |
| 33 | filling tiles near where you used to be while the tiles you were actually |
| 34 | looking at waited behind them -- observed as "the tiling focuses on |
| 35 | filling black tiles vs refining those blurry tiles". A frozen priority is |
| 36 | the wrong model for a queue whose ordering depends on a moving camera. |
| 37 | The worker now picks the nearest to the CURRENT centre each iteration. |
| 33 | 38 | """ |
| 34 | 39 | import os, queue, threading, time |
| 35 | 40 | import numpy as np |
| … | … | @@ -46,11 +51,10 @@ class TileStore: |
| 46 | 51 | os.makedirs(self.dir, exist_ok=True) |
| 47 | 52 | self._mem = {} # (tx,ty) -> np.int16 array |
| 48 | 53 | self._state = {} # (tx,ty) -> 'cached'|'queued'|'busy'|'error' |
| 49 | | self._q = queue.PriorityQueue() |
| 50 | | self._queued = set() |
| 54 | self._want = {} # (tx,ty) -> True, the pending set |
| 55 | self._wake = threading.Event() |
| 51 | 56 | self._lock = threading.Lock() |
| 52 | 57 | self._centre = (0, 0) |
| 53 | | self._seq = 0 |
| 54 | 58 | self.stats = dict(generated=0, from_disk=0, from_mem=0, |
| 55 | 59 | gen_seconds=0.0, km2=0.0, errors=0) |
| 56 | 60 | self._worker = threading.Thread(target=self._run, daemon=True) |
| … | … | @@ -99,29 +103,38 @@ class TileStore: |
| 99 | 103 | |
| 100 | 104 | # ---- queue ------------------------------------------------------------- |
| 101 | 105 | def request(self, tiles, centre): |
| 102 | | """Ask for these tiles. Priority is distance from centre, recomputed |
| 103 | | every call, so a new view supersedes the old prefetch order.""" |
| 106 | """Add to the pending set and update the centre. No ordering is decided |
| 107 | here -- see the note at the top of the file.""" |
| 104 | 108 | self._centre = centre |
| 105 | 109 | with self._lock: |
| 106 | 110 | for tx, ty in tiles: |
| 107 | 111 | k = (tx, ty) |
| 108 | | if k in self._queued or self.state(tx, ty) in ("cached", "disk"): |
| 112 | if k in self._want or self.state(tx, ty) in ("cached", "disk"): |
| 109 | 113 | continue |
| 110 | | d = (tx - centre[0]) ** 2 + (ty - centre[1]) ** 2 |
| 111 | | self._seq += 1 |
| 112 | | self._q.put((d, self._seq, k)) |
| 113 | | self._queued.add(k) |
| 114 | self._want[k] = True |
| 114 | 115 | self._state[k] = "queued" |
| 116 | self._wake.set() |
| 115 | 117 | |
| 116 | 118 | def queue_depth(self): |
| 117 | | return self._q.qsize() |
| 119 | return len(self._want) |
| 120 | |
| 121 | def _take_nearest(self): |
| 122 | """Pop the pending tile closest to the CURRENT centre.""" |
| 123 | with self._lock: |
| 124 | if not self._want: |
| 125 | return None |
| 126 | cx, cy = self._centre |
| 127 | k = min(self._want, key=lambda t: (t[0] - cx) ** 2 + (t[1] - cy) ** 2) |
| 128 | self._want.pop(k, None) |
| 129 | return k |
| 118 | 130 | |
| 119 | 131 | # ---- the single worker ------------------------------------------------- |
| 120 | 132 | def _run(self): |
| 121 | 133 | while True: |
| 122 | | _, _, (tx, ty) = self._q.get() |
| 123 | | with self._lock: |
| 124 | | self._queued.discard((tx, ty)) |
| 134 | k = self._take_nearest() |
| 135 | if k is None: |
| 136 | self._wake.wait(0.25); self._wake.clear(); continue |
| 137 | tx, ty = k |
| 125 | 138 | if self.state(tx, ty) in ("cached", "disk"): |
| 126 | 139 | self._state.pop((tx, ty), None) |
| 127 | 140 | continue |
| … | … | @@ -140,12 +153,19 @@ class TileStore: |
| 140 | 153 | self.stats["km2"] += (n * TILE_M / 1000.0) ** 2 |
| 141 | 154 | self._state.pop((tx, ty), None) |
| 142 | 155 | self.log(f" tile ({tx},{ty}) in {dt:.2f}s " |
| 143 | | f"queue {self._q.qsize()}") |
| 156 | f"queue {len(self._want)}") |
| 144 | 157 | except Exception as ex: |
| 145 | 158 | self.stats["errors"] += 1 |
| 146 | 159 | self._state[(tx, ty)] = "error" |
| 147 | 160 | self.log(f" tile ({tx},{ty}) FAILED: {ex}") |
| 148 | 161 | |
| 162 | def drop_pending(self): |
| 163 | """Forget everything not yet started. The in-flight tile finishes.""" |
| 164 | with self._lock: |
| 165 | for k in list(self._want): |
| 166 | self._state.pop(k, None) |
| 167 | self._want.clear() |
| 168 | |
| 149 | 169 | def summary(self): |
| 150 | 170 | s = dict(self.stats) |
| 151 | 171 | s["queue"] = self.queue_depth() |