Graham McIntire
ec1529fa6b
Two regressions from the recent performance commits: 1. ScoresController.encode_binary/1 (9cce257d): Enum.reduce produces iolists, but <<lats::binary>> requires actual binaries. Caused HTTP 500 on every /scores/cells request when score files exist on disk, making the entire propagation heatmap overlay invisible. 2. ScoreCache.valid_times/1 (15f4175c): Pipe syntax feeds @table as the first argument to :ets.foldl/3, creating a 4-argument call that crashes at runtime.
271 lines
8.8 KiB
Elixir
271 lines
8.8 KiB
Elixir
defmodule Microwaveprop.Propagation.ScoreCache do
|
||
@moduledoc """
|
||
Node-local ETS cache of propagation scores keyed by `{band_mhz, valid_time}`.
|
||
|
||
Populated lazily by `Microwaveprop.Propagation.scores_at/3` on the first
|
||
request for a given `{band_mhz, valid_time}`; subsequent requests hit ETS
|
||
until the entry is evicted. The cache is bounded to `max_entries/0` total
|
||
entries via LRU-by-`valid_time` eviction; on overflow the oldest
|
||
`valid_time` is dropped first so the freshly-arrived hour stays warm.
|
||
|
||
`broadcast_put/3` and the matching `"propagation:cache"` PubSub topic are
|
||
retained for the rare cluster-wide pre-warm path; the hot pipeline (Rust
|
||
Phase 2 cutover) does not call them — `NotifyListener` only emits the
|
||
lighter-weight `"propagation:updated"` PubSub so LiveView clients refresh
|
||
while the actual scores load on demand from `/data/scores/.../*.prop`.
|
||
|
||
## Storage layout
|
||
|
||
Each cache entry is bucketed into 5°×5° spatial chunks matching the layout
|
||
used by `Microwaveprop.Weather.GridCache`. The ETS value for
|
||
`{band_mhz, valid_time}` is `%{{lat_band, lon_band} => %{{lat, lon} => score}}`.
|
||
That way `fetch_bounds/3` only walks the chunks that intersect the
|
||
requested viewport instead of the full 92k-cell CONUS map; `fetch_point/4`
|
||
reads exactly one chunk.
|
||
"""
|
||
use GenServer
|
||
|
||
alias Phoenix.PubSub
|
||
|
||
@table :propagation_score_cache
|
||
@topic "propagation:cache"
|
||
@pubsub Microwaveprop.PubSub
|
||
@chunk_step 5
|
||
|
||
# Hard cap on cached `{band_mhz, valid_time}` entries. Each value is a
|
||
# 92k-cell map (now bucketed into 5°×5° chunks) on the order of ~2 MiB;
|
||
# without a cap the eager warm path materialises 23 bands × 19 forecast
|
||
# hours = 437 entries per pod (~850 MiB) and OOMs the BEAM. The cap
|
||
# keeps the LiveView hot path warm (one band × the hour or two the
|
||
# user is actively scrubbing) and lets the rest fall back to direct
|
||
# `.prop` reads via `read_from_disk_and_cache/3`, which is sub-100 ms
|
||
# per file.
|
||
@max_entries 32
|
||
|
||
@type score :: %{lat: float(), lon: float(), score: non_neg_integer()}
|
||
@type bounds :: %{optional(String.t()) => float()}
|
||
|
||
# ---------- Public API ----------
|
||
|
||
@spec start_link(keyword()) :: GenServer.on_start()
|
||
def start_link(opts) do
|
||
GenServer.start_link(__MODULE__, opts, name: __MODULE__)
|
||
end
|
||
|
||
@spec fetch(non_neg_integer(), DateTime.t()) :: {:ok, [score()]} | :miss
|
||
def fetch(band_mhz, valid_time) do
|
||
case :ets.lookup(@table, {band_mhz, valid_time}) do
|
||
[{_, chunked}] -> {:ok, chunks_to_list(chunked)}
|
||
[] -> :miss
|
||
end
|
||
end
|
||
|
||
@spec fetch_bounds(non_neg_integer(), DateTime.t(), bounds() | nil) ::
|
||
{:ok, [score()]} | :miss
|
||
def fetch_bounds(band_mhz, valid_time, bounds) do
|
||
case :ets.lookup(@table, {band_mhz, valid_time}) do
|
||
[{_, chunked}] -> {:ok, chunks_filtered_to_list(chunked, bounds)}
|
||
[] -> :miss
|
||
end
|
||
end
|
||
|
||
@doc """
|
||
Look up the score for a single `{lat, lon}` point at `{band_mhz, valid_time}`.
|
||
Returns `{:ok, score}` on hit or `:miss` on cache miss. The coordinates are
|
||
used as-is (no snap-to-grid) — callers that need the nearest grid cell should
|
||
snap first.
|
||
"""
|
||
@spec fetch_point(non_neg_integer(), DateTime.t(), float(), float()) ::
|
||
{:ok, non_neg_integer()} | :miss
|
||
def fetch_point(band_mhz, valid_time, lat, lon) do
|
||
case :ets.lookup(@table, {band_mhz, valid_time}) do
|
||
[{_, chunked}] ->
|
||
chunk_key = chunk_key_for(lat, lon)
|
||
|
||
case chunked |> Map.get(chunk_key, %{}) |> Map.get({lat, lon}) do
|
||
nil -> :miss
|
||
score -> {:ok, score}
|
||
end
|
||
|
||
[] ->
|
||
:miss
|
||
end
|
||
end
|
||
|
||
@doc "Maximum number of `{band_mhz, valid_time}` entries kept in ETS."
|
||
@spec max_entries() :: pos_integer()
|
||
def max_entries, do: @max_entries
|
||
|
||
@spec put(non_neg_integer(), DateTime.t(), [score()]) :: :ok
|
||
def put(band_mhz, valid_time, scores) do
|
||
chunked = list_to_chunks(scores)
|
||
:ets.insert(@table, {{band_mhz, valid_time}, chunked})
|
||
enforce_capacity()
|
||
:ok
|
||
end
|
||
|
||
# When the table is over the cap, evict whichever entry has the oldest
|
||
# `valid_time`. Picking the oldest matches the LiveView access pattern:
|
||
# the timeline scrubs forward, the hourly chain replaces the trailing
|
||
# edge, and the freshest hours are by far the most-requested. Loops
|
||
# until the table is back at or below the cap so that any historical
|
||
# over-shoot self-heals after this commit deploys.
|
||
defp enforce_capacity do
|
||
if :ets.info(@table, :size) > @max_entries do
|
||
case oldest_key() do
|
||
nil -> :ok
|
||
key -> :ets.delete(@table, key)
|
||
end
|
||
|
||
enforce_capacity()
|
||
end
|
||
end
|
||
|
||
defp oldest_key do
|
||
fun = fn
|
||
{{band, vt}, _}, nil ->
|
||
{band, vt}
|
||
|
||
{{band, vt}, _}, {_, oldest} = acc ->
|
||
if DateTime.before?(vt, oldest), do: {band, vt}, else: acc
|
||
end
|
||
|
||
:ets.foldl(fun, nil, @table)
|
||
end
|
||
|
||
@doc """
|
||
Insert locally AND broadcast to peer nodes via PubSub.
|
||
|
||
Call this from the single pod that computed the scores; every cluster member
|
||
(including the caller) will receive the PubSub message and populate its local
|
||
ETS. Use `put/3` directly if you already have a cluster-wide fan-out.
|
||
"""
|
||
@spec broadcast_put(non_neg_integer(), DateTime.t(), [score()]) :: :ok
|
||
def broadcast_put(band_mhz, valid_time, scores) do
|
||
_ = PubSub.broadcast(@pubsub, @topic, {:cache_refresh, band_mhz, valid_time, scores})
|
||
:ok
|
||
end
|
||
|
||
@doc "Returns the sorted list of valid_times cached for `band_mhz`."
|
||
@spec valid_times(non_neg_integer()) :: [DateTime.t()]
|
||
def valid_times(band_mhz) do
|
||
:ets.foldl(
|
||
fn
|
||
{{^band_mhz, vt}, _}, acc -> [vt | acc]
|
||
_, acc -> acc
|
||
end,
|
||
[],
|
||
@table
|
||
)
|
||
|> Enum.sort({:asc, DateTime})
|
||
end
|
||
|
||
@spec prune_older_than(DateTime.t()) :: non_neg_integer()
|
||
def prune_older_than(cutoff) do
|
||
match_spec = [
|
||
{{{:_, :"$1"}, :_}, [{:<, :"$1", {:const, cutoff}}], [true]}
|
||
]
|
||
|
||
:ets.select_delete(@table, match_spec)
|
||
end
|
||
|
||
@doc """
|
||
Delete every entry whose `valid_time` falls outside the inclusive
|
||
window `[past_cutoff, future_cutoff]`. Used to bound the cache to
|
||
the active forecast horizon (past ~1 h + HRRR's 18 h forward) so
|
||
long-horizon GEFS valid_times never accumulate in ETS.
|
||
"""
|
||
@spec prune_outside_window(DateTime.t(), DateTime.t()) :: non_neg_integer()
|
||
def prune_outside_window(past_cutoff, future_cutoff) do
|
||
match_spec = [
|
||
{{{:_, :"$1"}, :_},
|
||
[
|
||
{:orelse, {:<, :"$1", {:const, past_cutoff}}, {:>, :"$1", {:const, future_cutoff}}}
|
||
], [true]}
|
||
]
|
||
|
||
:ets.select_delete(@table, match_spec)
|
||
end
|
||
|
||
@spec clear() :: :ok
|
||
def clear do
|
||
:ets.delete_all_objects(@table)
|
||
:ok
|
||
end
|
||
|
||
@doc "Flush the GenServer mailbox so any in-flight cache_refresh messages are applied."
|
||
@spec sync() :: :ok
|
||
def sync do
|
||
GenServer.call(__MODULE__, :sync)
|
||
end
|
||
|
||
# ---------- GenServer callbacks ----------
|
||
|
||
@impl true
|
||
def init(_opts) do
|
||
_ = :ets.new(@table, [:set, :named_table, :public, :compressed, read_concurrency: true])
|
||
:ok = PubSub.subscribe(@pubsub, @topic)
|
||
{:ok, %{}}
|
||
end
|
||
|
||
@impl true
|
||
def handle_call(:sync, _from, state), do: {:reply, :ok, state}
|
||
|
||
@impl true
|
||
def handle_info({:cache_refresh, band_mhz, valid_time, scores}, state) do
|
||
put(band_mhz, valid_time, scores)
|
||
{:noreply, state}
|
||
end
|
||
|
||
def handle_info(_msg, state), do: {:noreply, state}
|
||
|
||
# ---------- Internal ----------
|
||
|
||
defp list_to_chunks(scores) do
|
||
Enum.reduce(scores, %{}, fn %{lat: lat, lon: lon, score: score}, acc ->
|
||
key = chunk_key_for(lat, lon)
|
||
chunk = Map.get(acc, key, %{})
|
||
Map.put(acc, key, Map.put(chunk, {lat, lon}, score))
|
||
end)
|
||
end
|
||
|
||
defp chunks_to_list(chunked) do
|
||
Enum.flat_map(chunked, fn {_chunk_key, cells} ->
|
||
Enum.map(cells, fn {{lat, lon}, score} ->
|
||
%{lat: lat, lon: lon, score: score}
|
||
end)
|
||
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}, score} <- cells,
|
||
lat >= s,
|
||
lat <= n,
|
||
lon >= w,
|
||
lon <= e,
|
||
do: %{lat: lat, lon: lon, score: score}
|
||
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
|