Graham McIntire
f1846c0a53
Telemetry showed the application-master process holding ~830 MiB of terms from warm_grid_cache_from_latest_profile — the data lives in the app master's heap and never GCs because the process is idle. Running it in a Task.start lets the terms die with the task. Mark GridCache, MrmsCache, NexradCache, and ScoreCache ETS tables :compressed. The scored-band-map and HRRR grid data are map-heavy; compression trims hundreds of MiB at a few percent CPU cost. Memoise HRRR .idx responses in Microwaveprop.Cache. Published idx files are immutable for a model run, but the hourly chain re-fetches the same URL dozens of times across forecast hours. Cuts ~10s per repeat out of hrrr_fetch_idx. Force a garbage collect at the end of HrrrFetchWorker.perform to reclaim the refc binary heap held from GRIB2 ranges before the Oban producer hands the process its next job.
80 lines
2.3 KiB
Elixir
80 lines
2.3 KiB
Elixir
defmodule Microwaveprop.Weather.MrmsCache do
|
|
@moduledoc """
|
|
Node-local ETS cache of the latest MRMS PrecipRate grid, regridded onto
|
|
the 0.125° propagation grid. Mirrors the pattern used by
|
|
`Microwaveprop.Propagation.ScoreCache` and
|
|
`Microwaveprop.Weather.GridCache` — a single GenServer owns the table,
|
|
callers read directly from ETS, and `broadcast_put/2` fans updates out
|
|
to peer nodes via PubSub.
|
|
|
|
Only one node runs `Microwaveprop.Workers.MrmsFetchWorker` per cron
|
|
tick (Oban Pro Smart engine picks a leader), so the other nodes get
|
|
the grid via the `"mrms:cache"` topic and stay in sync without each
|
|
one paying the 1 MB fetch + wgrib2 regrid cost.
|
|
"""
|
|
use GenServer
|
|
|
|
alias Phoenix.PubSub
|
|
|
|
@table :mrms_cache
|
|
@topic "mrms:cache"
|
|
@pubsub Microwaveprop.PubSub
|
|
|
|
@type rain_grid :: %{{float(), float()} => float()}
|
|
|
|
@spec start_link(keyword()) :: GenServer.on_start()
|
|
def start_link(opts), do: GenServer.start_link(__MODULE__, opts, name: __MODULE__)
|
|
|
|
@doc "Return the cached `{valid_time, rain_grid}` or `:miss`."
|
|
@spec fetch() :: {:ok, DateTime.t(), rain_grid()} | :miss
|
|
def fetch do
|
|
case :ets.lookup(@table, :current) do
|
|
[{_, valid_time, grid}] -> {:ok, valid_time, grid}
|
|
[] -> :miss
|
|
end
|
|
end
|
|
|
|
@doc "Current MRMS valid_time if anything is cached."
|
|
@spec valid_time() :: DateTime.t() | nil
|
|
def valid_time do
|
|
case fetch() do
|
|
{:ok, vt, _} -> vt
|
|
:miss -> nil
|
|
end
|
|
end
|
|
|
|
@spec put(DateTime.t(), rain_grid()) :: :ok
|
|
def put(valid_time, grid) do
|
|
:ets.insert(@table, {:current, valid_time, grid})
|
|
:ok
|
|
end
|
|
|
|
@doc "Insert locally and broadcast to peer nodes."
|
|
@spec broadcast_put(DateTime.t(), rain_grid()) :: :ok
|
|
def broadcast_put(valid_time, grid) do
|
|
put(valid_time, grid)
|
|
PubSub.broadcast(@pubsub, @topic, {:mrms_cache_refresh, valid_time, grid})
|
|
:ok
|
|
end
|
|
|
|
@spec clear() :: :ok
|
|
def clear do
|
|
:ets.delete_all_objects(@table)
|
|
:ok
|
|
end
|
|
|
|
@impl true
|
|
def init(_opts) do
|
|
:ets.new(@table, [:set, :named_table, :public, :compressed, read_concurrency: true])
|
|
PubSub.subscribe(@pubsub, @topic)
|
|
{:ok, %{}}
|
|
end
|
|
|
|
@impl true
|
|
def handle_info({:mrms_cache_refresh, valid_time, grid}, state) do
|
|
put(valid_time, grid)
|
|
{:noreply, state}
|
|
end
|
|
|
|
def handle_info(_msg, state), do: {:noreply, state}
|
|
end
|