perf: reduce per-pod RSS and HRRR chain wall time

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.
This commit is contained in:
Graham McIntire 2026-04-19 14:56:48 -05:00
parent dfc7895197
commit f1846c0a53
No known key found for this signature in database
GPG key ID: F4ABF488E6029E59
8 changed files with 87 additions and 23 deletions

View file

@ -43,7 +43,12 @@ defmodule Microwaveprop.Application do
# Warm GridCache from the latest persisted ProfilesFile so /weather # Warm GridCache from the latest persisted ProfilesFile so /weather
# has data immediately after a pod restart. Runs after the GridCache # has data immediately after a pod restart. Runs after the GridCache
# GenServer is up (it's in `children` above). # GenServer is up (it's in `children` above).
Microwaveprop.Weather.warm_grid_cache_from_latest_profile() #
# Must run in a short-lived process, not inline in the application
# master: when executed here the loaded rows live on the app master's
# heap and never GC (it's idle after boot), pinning ~800 MiB for the
# lifetime of the pod.
Task.start(fn -> Microwaveprop.Weather.warm_grid_cache_from_latest_profile() end)
# Load ML model in dev/test only (Nx/Axon not compiled for prod) # Load ML model in dev/test only (Nx/Axon not compiled for prod)
if Application.get_env(:microwaveprop, :load_ml_model, false) do if Application.get_env(:microwaveprop, :load_ml_model, false) do

View file

@ -122,7 +122,7 @@ defmodule Microwaveprop.Propagation.ScoreCache do
@impl true @impl true
def init(_opts) do def init(_opts) do
:ets.new(@table, [:set, :named_table, :public, read_concurrency: true]) :ets.new(@table, [:set, :named_table, :public, :compressed, read_concurrency: true])
PubSub.subscribe(@pubsub, @topic) PubSub.subscribe(@pubsub, @topic)
{:ok, %{}} {:ok, %{}}
end end

View file

@ -121,7 +121,7 @@ defmodule Microwaveprop.Weather.GridCache do
@impl true @impl true
def init(_opts) do def init(_opts) do
:ets.new(@table, [:set, :named_table, :public, read_concurrency: true]) :ets.new(@table, [:set, :named_table, :public, :compressed, read_concurrency: true])
:ets.new(@lock_table, [:set, :named_table, :public]) :ets.new(@lock_table, [:set, :named_table, :public])
PubSub.subscribe(@pubsub, @topic) PubSub.subscribe(@pubsub, @topic)
{:ok, %{}} {:ok, %{}}

View file

@ -413,21 +413,27 @@ defmodule Microwaveprop.Weather.HrrrClient do
end end
end end
# A published HRRR idx file is immutable for the lifetime of its
# model run (60 min), but the hourly propagation chain fetches the
# same `.idx` URL dozens of times across forecast-hour steps and
# products. Memoising in `Microwaveprop.Cache` removes 100+ seconds
# per chain from the `hrrr_fetch_idx` span.
@idx_cache_ttl_ms to_timeout(hour: 1)
defp fetch_idx(url) do defp fetch_idx(url) do
Microwaveprop.Instrument.span([:hrrr, :fetch_idx], %{url: url}, fn -> Microwaveprop.Cache.fetch_or_store({:hrrr_idx, url}, @idx_cache_ttl_ms, fn ->
case Req.get(url, req_options()) do Microwaveprop.Instrument.span([:hrrr, :fetch_idx], %{url: url}, fn -> do_fetch_idx(url) end)
{:ok, %{status: 200, body: body}} ->
{:ok, body}
{:ok, %{status: status}} ->
{:error, "HRRR idx HTTP #{status}"}
{:error, reason} ->
{:error, reason}
end
end) end)
end end
defp do_fetch_idx(url) do
case Req.get(url, req_options()) do
{:ok, %{status: 200, body: body}} -> {:ok, body}
{:ok, %{status: status}} -> {:error, "HRRR idx HTTP #{status}"}
{:error, reason} -> {:error, reason}
end
end
@doc """ @doc """
Downloads the given byte ranges from a GRIB2 URL and concatenates Downloads the given byte ranges from a GRIB2 URL and concatenates
them in offset order. Merges adjacent/overlapping ranges and runs them in offset order. Merges adjacent/overlapping ranges and runs

View file

@ -65,7 +65,7 @@ defmodule Microwaveprop.Weather.MrmsCache do
@impl true @impl true
def init(_opts) do def init(_opts) do
:ets.new(@table, [:set, :named_table, :public, read_concurrency: true]) :ets.new(@table, [:set, :named_table, :public, :compressed, read_concurrency: true])
PubSub.subscribe(@pubsub, @topic) PubSub.subscribe(@pubsub, @topic)
{:ok, %{}} {:ok, %{}}
end end

View file

@ -73,7 +73,7 @@ defmodule Microwaveprop.Weather.NexradCache do
@impl true @impl true
def init(_opts) do def init(_opts) do
:ets.new(@table, [:set, :named_table, :public, read_concurrency: true]) :ets.new(@table, [:set, :named_table, :public, :compressed, read_concurrency: true])
{:ok, %{}} {:ok, %{}}
end end
end end

View file

@ -27,13 +27,20 @@ defmodule Microwaveprop.Workers.HrrrFetchWorker do
|> Enum.map(fn %{"lat" => lat, "lon" => lon} -> {lat, lon} end) |> Enum.map(fn %{"lat" => lat, "lon" => lon} -> {lat, lon} end)
|> Enum.reject(fn {lat, lon} -> Weather.has_hrrr_profile?(lat, lon, valid_time) end) |> Enum.reject(fn {lat, lon} -> Weather.has_hrrr_profile?(lat, lon, valid_time) end)
if point_tuples == [] do result =
Logger.info("HRRR batch: all #{length(points)} points already exist for #{valid_time_str}") if point_tuples == [] do
:ok Logger.info("HRRR batch: all #{length(points)} points already exist for #{valid_time_str}")
else :ok
Logger.info("HRRR batch: fetching #{length(point_tuples)} points for #{valid_time_str}") else
fetch_and_store_batch(point_tuples, valid_time, valid_time_str) Logger.info("HRRR batch: fetching #{length(point_tuples)} points for #{valid_time_str}")
end fetch_and_store_batch(point_tuples, valid_time, valid_time_str)
end
# Reclaim the refc binary heap held from GRIB2 ranges before the
# Oban producer hands this process its next job. Without this each
# worker steady-states at 100+ MiB of retained HRRR payload.
:erlang.garbage_collect(self())
result
end end
def perform(%Oban.Job{args: %{"lat" => raw_lat, "lon" => raw_lon, "valid_time" => valid_time_str}}) do def perform(%Oban.Job{args: %{"lat" => raw_lat, "lon" => raw_lon, "valid_time" => valid_time_str}}) do

View file

@ -171,6 +171,52 @@ defmodule Microwaveprop.Weather.HrrrClientTest do
assert is_float(profile.surface_temp_c) assert is_float(profile.surface_temp_c)
assert is_struct(profile.run_time, DateTime) assert is_struct(profile.run_time, DateTime)
end end
test "caches idx responses across calls to the same HRRR run" do
# `.idx` files are static once a model run is published. The 114s
# `hrrr_fetch_grid` span in prod spends ~10s re-fetching the same idx
# URL across forecast hours / products. Cache eliminates that per-run.
grib_data = File.read!("test/fixtures/grib2/hrrr_tmp_2m.grib2")
grib_size = byte_size(grib_data)
idx_text = """
1:0:d=2026032818:TMP:2 m above ground:anl:
2:#{grib_size}:d=2026032818:DPT:2 m above ground:anl:
3:#{grib_size * 2}:d=2026032818:PRES:surface:anl:
"""
Microwaveprop.Cache.invalidate(
{:hrrr_idx, "https://noaa-hrrr-bdp-pds.s3.amazonaws.com/hrrr.20260328/conus/hrrr.t18z.wrfsfcf00.grib2.idx"}
)
{:ok, counter} = Agent.start_link(fn -> %{idx: 0, grib: 0} end)
Req.Test.stub(HrrrClient, fn conn ->
if String.ends_with?(conn.request_path, ".idx") do
Agent.update(counter, &Map.update!(&1, :idx, fn n -> n + 1 end))
Plug.Conn.send_resp(conn, 200, idx_text)
else
Agent.update(counter, &Map.update!(&1, :grib, fn n -> n + 1 end))
[range] = Plug.Conn.get_req_header(conn, "range")
if String.contains?(range, ",") do
Plug.Conn.send_resp(conn, 200, "full file")
else
Plug.Conn.send_resp(conn, 206, grib_data)
end
end
end)
valid_time = ~U[2026-03-28 18:00:00Z]
assert {:ok, _} = HrrrClient.fetch_profile(32.90, -97.04, valid_time)
assert {:ok, _} = HrrrClient.fetch_profile(32.90, -97.04, valid_time)
%{idx: idx_count, grib: grib_count} = Agent.get(counter, & &1)
assert idx_count == 1, "idx should be cached — expected 1 fetch, got #{idx_count}"
assert grib_count > 0, "grib fetches should still happen"
end
end end
describe "build_profile/1" do describe "build_profile/1" do