perf(weather/grid_cache): chunk-keyed ETS layout for viewport reads
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.
This commit is contained in:
parent
a14f14485e
commit
1d72f52dad
2 changed files with 74 additions and 42 deletions
|
|
@ -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.
|
||||
|
|
|
|||
|
|
@ -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
|
||||
|
|
|
|||
Loading…
Add table
Reference in a new issue