From 1d72f52dad20be7f620a0518c5ced3b7c62e3425 Mon Sep 17 00:00:00 2001 From: Graham McIntire Date: Tue, 28 Apr 2026 17:24:12 -0500 Subject: [PATCH] perf(weather/grid_cache): chunk-keyed ETS layout for viewport reads MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Mirror ScalarFile's 5°×5° chunk layout inside GridCache. Each ETS entry is now `%{{lat_band, lon_band} => %{{lat, lon} => row}}` instead of a flat `%{{lat,lon} => row}`. fetch_bounds/2 walks only the chunks that intersect the requested viewport, so latest-hour pans become proportional to visible chunks (~2-4) rather than O(92k cells). fetch/1, fetch_point/3, put/2 and broadcast_put/2 keep their pre-existing contracts. Also closes the open items in docs/weather.md: Phase 3 is now done end to end. The only remaining future-work item is moving WeatherLayers.derive itself into the Rust pipeline so the BEAM doesn't pay the one-time materialization pass per new valid_time — small runtime savings vs. significant port effort, so left as documented future work. --- docs/weather.md | 27 ++++---- lib/microwaveprop/weather/grid_cache.ex | 89 +++++++++++++++++-------- 2 files changed, 74 insertions(+), 42 deletions(-) diff --git a/docs/weather.md b/docs/weather.md index df63f56a..6f5f4c64 100644 --- a/docs/weather.md +++ b/docs/weather.md @@ -16,8 +16,7 @@ Completed so far: Still pending: -- Have the propagation pipeline write the scalar artifact directly on the Rust side, alongside `.mp.gz`. Today, scalar materialization is triggered from Elixir's `NotifyListener.handle_propagation_ready/1` as soon as the Rust pipeline fires `NOTIFY propagation_ready`, which closes the gap for steady-state forecast updates but still costs an extra per-`valid_time` derivation pass on the BEAM. -- Optional: replace `GridCache`'s in-memory `%{{lat,lon} => row}` map with the same chunk-keyed layout as `ScalarFile`. Low priority — `GridCache` only holds the latest valid_time, so its full-map filter is a small fraction of the surrounding work. +- Have the propagation pipeline write the scalar artifact directly on the Rust side, alongside `.mp.gz`. Today, scalar materialization is triggered from Elixir's `NotifyListener.handle_propagation_ready/1` as soon as the Rust pipeline fires `NOTIFY propagation_ready`, so the gap between profile-write and scalar-availability is just one BEAM derivation pass per new `valid_time`. Closing it fully would require porting `WeatherLayers.derive` (lapse rates, inversion strength, 850/700 mb interpolation, duct summarization) to Rust, plus a compatible writer Elixir's `ScalarFile` reader can decode. Tracked as future work — the runtime cost it would save is small relative to the port effort. Available infrastructure that should shape the solution: @@ -99,18 +98,16 @@ What's still open: - Same as #1 / #2: the very first scrub to a never-materialized hour pays the cold-derive cost once before the background task fills in the scalar file. -### 4. `GridCache.fetch_bounds/2` still scans the whole cached grid on every pan — partially RESOLVED +### 4. `GridCache.fetch_bounds/2` still scans the whole cached grid on every pan — RESOLVED Severity: medium (was). What shipped: -- The expensive code path (`weather_grid_at/2` cold reads) now uses chunked spatial reads via `ScalarFile`. The latest hour also benefits when its grid hasn't been ETS-cached yet. -- The latest-hour `GridCache` ETS path still does an `Enum.filter/2` over the full cached map — we kept it as-is because the in-memory map is small (one valid_time at most) and the filter is the cheap leg compared to the rest of the read path. - -What's still open: - -- Optionally swap `GridCache` storage to the same chunk-keyed layout used by `ScalarFile` so latest-hour pans become O(visible chunks). Low priority since `ScalarFile` already covers the dominant cost. +- `GridCache` now stores each `valid_time` as `%{{lat_band, lon_band} => %{{lat, lon} => row}}` using the same 5°×5° chunk layout as `ScalarFile` ([`grid_cache.ex`](/Users/graham/dev/ntms/microwaveprop/lib/microwaveprop/weather/grid_cache.ex:30)). +- `fetch_bounds/2` only walks the chunks that intersect the requested viewport, so a state-sized viewport reads ~2-4 chunks instead of filtering ~92k cells. +- `fetch_point/3` reads exactly one chunk. +- All public APIs (`fetch/1`, `fetch_bounds/2`, `fetch_point/3`, `put/2`, `broadcast_put/2`) preserve their pre-rewrite contracts. ### 5. The frontend overlay renderer is doing too much work on the main thread — RESOLVED @@ -219,11 +216,11 @@ Status: - tile generation still derives from raw profiles on the server - tiles are generated on request rather than read from a weather-specific derived artifact -### Phase 3: improve cache shape for latest-hour panning +### Phase 3: improve cache shape for latest-hour panning — DONE -1. Replace full-map `Enum.filter/2` in `GridCache.fetch_bounds/2` with chunked spatial storage. (Optional now — `ScalarFile` covers the dominant viewport-read cost. `GridCache` is a thin in-memory hot path with one entry, so the filter is cheap relative to the surrounding work.) +1. Replace full-map `Enum.filter/2` in `GridCache.fetch_bounds/2` with chunked spatial storage. ✅ shipped — `GridCache` now mirrors `ScalarFile`'s 5°×5° chunk layout in ETS, so latest-hour pans walk only the chunks that intersect the viewport. 2. Persist chunks on NFS and read only the visible chunk set. ✅ done in Phase 1. -3. Pre-generate the latest analysis hour and adjacent forecast hours as part of the pipeline so the app serves ready-made artifacts. Pending — the propagation pipeline (Rust side) should write the scalar artifact alongside `.mp.gz`. Until then, the first read on each new hour does a one-time materialization on the BEAM. +3. Pre-generate the latest analysis hour and adjacent forecast hours as part of the pipeline so the app serves ready-made artifacts. ✅ shipped — `NotifyListener.handle_propagation_ready/1` fires `Weather.materialize_scalar_file/1` in a detached Task immediately on `NOTIFY propagation_ready`. Moving the derivation into Rust itself remains future work; see Still Pending. ## Suggested Target Architecture @@ -396,13 +393,13 @@ That is the most balanced scaled-down design here. It is materially faster than ## Bottom Line -The structural fixes are now in place: +All listed phases are shipped: - `/weather` reads from pre-derived scalar chunks on NFS instead of re-deriving from raw HRRR data on every request. -- Viewport reads only touch the chunks that overlap the current bounds. +- Viewport reads only touch the chunks that overlap the current bounds — for both NFS-backed `ScalarFile` reads and the in-memory `GridCache`. - Point clicks read a single chunk instead of the whole grid file. - The browser displays server-rendered tiles; layer/time changes only swap tile URLs. - New forecast hours are pre-warmed: `NotifyListener.handle_propagation_ready/1` fires `Weather.materialize_scalar_file/1` in a detached Task as soon as the Rust pipeline fires `propagation_ready`, so the first interactive `/weather` reader of that hour finds the scalar artifact already on disk. - Legend and timeline DOM churn is gone — the legend is mounted once and the timeline only rebuilds its full HTML on data changes, otherwise just restyling existing buttons. -Remaining work is one optional optimization (move scalar-file production into the Rust pipeline so we save the one-time BEAM derive on each new forecast hour) and a low-priority `GridCache` chunk rewrite. +Remaining work is a single non-runtime-critical port — move `WeatherLayers.derive` into the Rust pipeline so the BEAM doesn't pay even the one-time derivation pass per new `valid_time`. Documented as future work; the runtime savings are small relative to the port effort. diff --git a/lib/microwaveprop/weather/grid_cache.ex b/lib/microwaveprop/weather/grid_cache.ex index 305d0865..62ad9df7 100644 --- a/lib/microwaveprop/weather/grid_cache.ex +++ b/lib/microwaveprop/weather/grid_cache.ex @@ -8,11 +8,18 @@ defmodule Microwaveprop.Weather.GridCache do table, runs per-row `derive_and_clean` transforms, and returns 3-10k rows. With this cache those calls become in-memory map iterations. - Each cache entry stores `%{{lat, lon} => derived_row}` so per-point lookups - (used by `weather_point_detail/3`) are O(1). Populated by - `Microwaveprop.Weather.warm_grid_cache/1` after the hourly worker upserts - new HRRR data, fanned out across the cluster via the `"weather:cache"` - PubSub topic so every node stays in sync. + ## Storage layout + + Each cache entry is bucketed into 5°×5° spatial chunks matching + `Microwaveprop.Weather.ScalarFile`'s on-disk layout. The ETS value for + `valid_time` is `%{{lat_band, lon_band} => %{{lat, lon} => row}}`. That + way `fetch_bounds/2` only walks the chunks that intersect the requested + viewport instead of the full 92k-cell CONUS map. `fetch_point/3` reads + exactly one chunk. + + Populated by `Microwaveprop.Weather.warm_grid_cache/1` after the hourly + worker upserts new HRRR data, fanned out across the cluster via the + `"weather:cache"` PubSub topic so every node stays in sync. """ use GenServer @@ -22,6 +29,7 @@ defmodule Microwaveprop.Weather.GridCache do @lock_table :weather_grid_fill_locks @topic "weather:cache" @pubsub Microwaveprop.PubSub + @chunk_step 5 @type row :: %{required(:lat) => float(), required(:lon) => float(), optional(atom()) => any()} @type bounds :: %{optional(String.t()) => float()} @@ -32,9 +40,9 @@ defmodule Microwaveprop.Weather.GridCache do @spec fetch(DateTime.t()) :: {:ok, [row()]} | :miss def fetch(valid_time) do case :ets.lookup(@table, valid_time) do - [{_, grid}] -> + [{_, chunked}] -> emit_lookup(true) - {:ok, grid_to_list(grid)} + {:ok, chunks_to_list(chunked)} [] -> emit_lookup(false) @@ -45,9 +53,9 @@ defmodule Microwaveprop.Weather.GridCache do @spec fetch_bounds(DateTime.t(), bounds() | nil) :: {:ok, [row()]} | :miss def fetch_bounds(valid_time, bounds) do case :ets.lookup(@table, valid_time) do - [{_, grid}] -> + [{_, chunked}] -> emit_lookup(true) - {:ok, grid_to_filtered_list(grid, bounds)} + {:ok, chunks_filtered_to_list(chunked, bounds)} [] -> emit_lookup(false) @@ -58,8 +66,10 @@ defmodule Microwaveprop.Weather.GridCache do @spec fetch_point(DateTime.t(), float(), float()) :: {:ok, row()} | :miss def fetch_point(valid_time, lat, lon) do case :ets.lookup(@table, valid_time) do - [{_, grid}] -> - case Map.get(grid, {lat, lon}) do + [{_, chunked}] -> + chunk_key = chunk_key_for(lat, lon) + + case chunked |> Map.get(chunk_key, %{}) |> Map.get({lat, lon}) do nil -> emit_lookup(false) :miss @@ -81,8 +91,8 @@ defmodule Microwaveprop.Weather.GridCache do @spec put(DateTime.t(), [row()]) :: :ok def put(valid_time, rows) do - grid = list_to_grid(rows) - :ets.insert(@table, {valid_time, grid}) + chunked = list_to_chunks(rows) + :ets.insert(@table, {valid_time, chunked}) :ok end @@ -124,13 +134,13 @@ defmodule Microwaveprop.Weather.GridCache do The GenServer `Process.monitor`s the caller: if the caller crashes before calling `release_fill/1`, the lock is released automatically. """ - @spec claim_fill(DateTime.t()) :: boolean() + @spec claim_fill(term()) :: boolean() def claim_fill(valid_time) do GenServer.call(__MODULE__, {:claim_fill, valid_time, self()}) end @doc "Release a fill lock claimed via `claim_fill/1`." - @spec release_fill(DateTime.t()) :: :ok + @spec release_fill(term()) :: :ok def release_fill(valid_time) do GenServer.call(__MODULE__, {:release_fill, valid_time}) end @@ -206,21 +216,46 @@ defmodule Microwaveprop.Weather.GridCache do end end - # ---------- Internal ---------- + # ---------- Chunk helpers ---------- - defp list_to_grid(rows) do - Map.new(rows, fn %{lat: lat, lon: lon} = row -> {{lat, lon}, row} end) + defp list_to_chunks(rows) do + Enum.reduce(rows, %{}, fn %{lat: lat, lon: lon} = row, acc -> + key = chunk_key_for(lat, lon) + chunk = Map.get(acc, key, %{}) + Map.put(acc, key, Map.put(chunk, {lat, lon}, row)) + end) end - defp grid_to_list(grid), do: Enum.map(grid, fn {_, row} -> row end) - - defp grid_to_filtered_list(grid, nil), do: grid_to_list(grid) - - defp grid_to_filtered_list(grid, %{"south" => s, "north" => n, "west" => w, "east" => e}) do - grid - |> Enum.filter(fn {{lat, lon}, _} -> - lat >= s and lat <= n and lon >= w and lon <= e + defp chunks_to_list(chunked) do + Enum.flat_map(chunked, fn {_chunk_key, cells} -> + Enum.map(cells, fn {_, row} -> row end) end) - |> Enum.map(fn {_, row} -> row end) + end + + defp chunks_filtered_to_list(chunked, nil), do: chunks_to_list(chunked) + + defp chunks_filtered_to_list(chunked, %{"south" => s, "north" => n, "west" => w, "east" => e} = bounds) do + chunked + |> Enum.filter(fn {chunk_key, _} -> chunk_intersects_bounds?(chunk_key, bounds) end) + |> Enum.flat_map(fn {_, cells} -> + for {{lat, lon}, row} <- cells, lat >= s, lat <= n, lon >= w, lon <= e, do: row + end) + end + + defp chunk_key_for(lat, lon) do + {chunk_band(lat * 1.0), chunk_band(lon * 1.0)} + end + + defp chunk_band(value) when is_float(value) do + (value / @chunk_step) |> Float.floor() |> trunc() + end + + defp chunk_intersects_bounds?({lat_band, lon_band}, %{"south" => s, "north" => n, "west" => w, "east" => e}) do + chunk_south = lat_band * @chunk_step + chunk_north = chunk_south + @chunk_step + chunk_west = lon_band * @chunk_step + chunk_east = chunk_west + @chunk_step + + chunk_north >= s and chunk_south <= n and chunk_east >= w and chunk_west <= e end end