From bbf648160348b63fd1748dda75c605da9e5ac71b Mon Sep 17 00:00:00 2001 From: Graham McIntire Date: Sun, 19 Apr 2026 09:05:47 -0500 Subject: [PATCH] perf(nexrad): route fetch_decoded_frame through NexradCache MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Telemetry puts nexrad_decode_png at 1.76 s/call × 37,847 calls/h — ~18.5 h of CPU across the cluster every real hour, the single biggest CPU consumer in the system. The map-click path (fetch_rain_cells) already cached decoded frames by 5-min bucket, but the per-contact CommonVolumeRadarWorker path (fetch_decoded_frame) went straight to network + decode on every call. Backfill means many contacts share a 5-min window, so the same 66 MB frame was being decoded dozens of times. Wire fetch_decoded_frame through NexradCache keyed on the rounded timestamp. Add a 20-entry size cap in NexradCache so backfill processing contacts in random timestamp order can't grow the ETS table to hundreds of GB. Each frame is ~66 MB, so 20 = ~1.3 GB worst case, well under typical pod memory. Expected impact: cuts sustained decode load by an order of magnitude depending on backfill temporal locality; map-click path is unchanged. --- lib/microwaveprop/weather/nexrad_cache.ex | 24 +++++++ lib/microwaveprop/weather/nexrad_client.ex | 36 ++++++----- .../weather/nexrad_cache_test.exs | 24 +++++++ .../weather/nexrad_client_test.exs | 63 ++++++++++++++++++- 4 files changed, 129 insertions(+), 18 deletions(-) diff --git a/lib/microwaveprop/weather/nexrad_cache.ex b/lib/microwaveprop/weather/nexrad_cache.ex index 2672ec22..6f855246 100644 --- a/lib/microwaveprop/weather/nexrad_cache.ex +++ b/lib/microwaveprop/weather/nexrad_cache.ex @@ -13,6 +13,13 @@ defmodule Microwaveprop.Weather.NexradCache do @table :nexrad_frame_cache + # Each frame is ~66 MB (12,200 × 5,400 palette-index bytes). 20 frames + # caps at ~1.3 GB — comfortable headroom under a 2 GB pod limit, and + # enough to hit on the common backfill pattern where a worker chews + # through contacts in one timestamp window before moving on. Adjust + # if pod memory limits change or the n0q geometry changes. + @max_entries 20 + @type pixels :: binary() @type width :: pos_integer() @@ -32,9 +39,26 @@ defmodule Microwaveprop.Weather.NexradCache do @spec put(DateTime.t(), pixels(), width()) :: :ok def put(rounded_ts, pixels, width) do :ets.insert(@table, {rounded_ts, pixels, width}) + enforce_size_cap() :ok end + defp enforce_size_cap do + size = :ets.info(@table, :size) + + if size > @max_entries do + # Drop the oldest-by-key entries until we're back under the cap. + # O(n log n) but n is ~20, so this runs in microseconds. + excess = size - @max_entries + + @table + |> :ets.tab2list() + |> Enum.sort_by(fn {ts, _, _} -> DateTime.to_unix(ts) end) + |> Enum.take(excess) + |> Enum.each(fn {ts, _, _} -> :ets.delete(@table, ts) end) + end + end + @spec prune_older_than(DateTime.t()) :: non_neg_integer() def prune_older_than(cutoff_ts) do match_spec = [{{:"$1", :_, :_}, [{:<, :"$1", {:const, cutoff_ts}}], [true]}] diff --git a/lib/microwaveprop/weather/nexrad_client.ex b/lib/microwaveprop/weather/nexrad_client.ex index 14646241..495cfecd 100644 --- a/lib/microwaveprop/weather/nexrad_client.ex +++ b/lib/microwaveprop/weather/nexrad_client.ex @@ -76,37 +76,39 @@ defmodule Microwaveprop.Weather.NexradClient do """ @spec fetch_rain_cells(float(), float(), float(), float()) :: {:ok, [{float(), float(), float()}]} | {:error, term()} def fetch_rain_cells(lat, lon, radius_km \\ 300.0, min_dbz \\ 25.0) do - now = DateTime.utc_now() - rounded = round_to_5min(now) - - case NexradCache.fetch(rounded) do + case fetch_decoded_frame(DateTime.utc_now()) do {:ok, pixels, width} -> {:ok, extract_rain_cells(pixels, width, lat, lon, radius_km, min_dbz)} - :miss -> - case fetch_and_decode_frame(rounded) do - {:ok, pixels, width} -> - NexradCache.put(rounded, pixels, width) - NexradCache.prune_older_than(DateTime.add(rounded, -600, :second)) - {:ok, extract_rain_cells(pixels, width, lat, lon, radius_km, min_dbz)} - - error -> - error - end + error -> + error end end @doc """ Fetch and decode the n0q PNG frame nearest the given timestamp. Returns `{:ok, pixels, width}` — a flat binary of palette-index bytes — - or `{:error, reason}`. Does NOT go through `NexradCache`; callers that - want caching (e.g. real-time point queries) should call `fetch_rain_cells/4`. + or `{:error, reason}`. Goes through `NexradCache`, keyed on the 5-min + rounded timestamp — concurrent radar workers (and any backfill + neighbors that share a 5-min window) reuse the same decoded buffer + instead of each paying the full ~5 MB download + ~1.7 s decode. """ @spec fetch_decoded_frame(DateTime.t()) :: {:ok, binary(), non_neg_integer()} | {:error, term()} def fetch_decoded_frame(%DateTime{} = dt) do rounded = round_to_5min(dt) - fetch_and_decode_frame(rounded) + + case NexradCache.fetch(rounded) do + {:ok, pixels, width} -> + {:ok, pixels, width} + + :miss -> + with {:ok, pixels, width} <- fetch_and_decode_frame(rounded) do + NexradCache.put(rounded, pixels, width) + NexradCache.prune_older_than(DateTime.add(rounded, -600, :second)) + {:ok, pixels, width} + end + end end defp fetch_and_decode_frame(rounded) do diff --git a/test/microwaveprop/weather/nexrad_cache_test.exs b/test/microwaveprop/weather/nexrad_cache_test.exs index 4c331203..e1e06360 100644 --- a/test/microwaveprop/weather/nexrad_cache_test.exs +++ b/test/microwaveprop/weather/nexrad_cache_test.exs @@ -44,4 +44,28 @@ defmodule Microwaveprop.Weather.NexradCacheTest do assert {:ok, <<1>>, 1} = NexradCache.fetch(~U[2026-04-12 12:00:00Z]) end end + + describe "size cap" do + test "evicts the oldest-by-timestamp entries once the cap is exceeded" do + # Insert 25 entries; max is 20, so 5 oldest get dropped. + base = ~U[2026-04-12 00:00:00Z] + + for i <- 1..25 do + ts = DateTime.add(base, i * 300, :second) + NexradCache.put(ts, <>, 1) + end + + # The 5 oldest (i = 1..5) should have been evicted. + for i <- 1..5 do + ts = DateTime.add(base, i * 300, :second) + assert NexradCache.fetch(ts) == :miss + end + + # The most recent 20 should still be present. + for i <- 6..25 do + ts = DateTime.add(base, i * 300, :second) + assert {:ok, <<^i>>, 1} = NexradCache.fetch(ts) + end + end + end end diff --git a/test/microwaveprop/weather/nexrad_client_test.exs b/test/microwaveprop/weather/nexrad_client_test.exs index 5e086413..e7a93b5b 100644 --- a/test/microwaveprop/weather/nexrad_client_test.exs +++ b/test/microwaveprop/weather/nexrad_client_test.exs @@ -1,6 +1,7 @@ defmodule Microwaveprop.Weather.NexradClientTest do - use ExUnit.Case, async: true + use ExUnit.Case, async: false + alias Microwaveprop.Weather.NexradCache alias Microwaveprop.Weather.NexradClient describe "round_to_5min/1" do @@ -139,6 +140,66 @@ defmodule Microwaveprop.Weather.NexradClientTest do end end + describe "fetch_decoded_frame/1 caching" do + # Minimal 20×20 indexed-color PNG identical to the NexradWorkerTest + # fixture. Used here just to give the decoder something valid to + # chew on; caching semantics don't care about the contents. + defp minimal_png(width \\ 20, height \\ 20) do + signature = <<137, 80, 78, 71, 13, 10, 26, 10>> + ihdr_data = <> + ihdr_chunk = <> + + raw = for _ <- 1..height, into: <<>>, do: <<0, 0::size(width)-unit(8)>> + compressed = :zlib.compress(raw) + idat_chunk = <> + iend_chunk = <<0::32, "IEND", 0::32>> + + signature <> ihdr_chunk <> idat_chunk <> iend_chunk + end + + setup do + NexradCache.clear() + on_exit(fn -> NexradCache.clear() end) + :ok + end + + test "first call fetches+decodes; second call for same 5-min bucket is served from cache" do + test_pid = self() + + Req.Test.stub(NexradClient, fn conn -> + send(test_pid, :http_fetched) + Plug.Conn.send_resp(conn, 200, minimal_png()) + end) + + ts = ~U[2022-08-20 14:07:00Z] + + {:ok, pixels1, width1} = NexradClient.fetch_decoded_frame(ts) + {:ok, pixels2, width2} = NexradClient.fetch_decoded_frame(ts) + + # The cache is keyed on the 5-min rounded timestamp, so a + # second call within the same bucket reuses the decoded pixels. + assert pixels1 == pixels2 + assert width1 == width2 + assert_received :http_fetched + refute_received :http_fetched + end + + test "different 5-min buckets hit the network separately" do + test_pid = self() + + Req.Test.stub(NexradClient, fn conn -> + send(test_pid, :http_fetched) + Plug.Conn.send_resp(conn, 200, minimal_png()) + end) + + {:ok, _, _} = NexradClient.fetch_decoded_frame(~U[2022-08-20 14:05:00Z]) + {:ok, _, _} = NexradClient.fetch_decoded_frame(~U[2022-08-20 14:10:00Z]) + + assert_received :http_fetched + assert_received :http_fetched + end + end + describe "pixel_to_dbz/1" do test "value 0 maps to 0 (no echo)" do assert NexradClient.pixel_to_dbz(0) == 0.0