From 15ce50365f3a3514b7f82d29f23bb7f03c6f1b5a Mon Sep 17 00:00:00 2001 From: Graham McIntire Date: Tue, 21 Apr 2026 14:20:29 -0500 Subject: [PATCH] perf(concurrency): partition Task.Supervisor to remove bottleneck on heavy async_stream paths MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Wrap a Task.Supervisor in a PartitionSupervisor (one partition per scheduler) and route the four sustained async_stream callers through `{:via, PartitionSupervisor, {Microwaveprop.TaskSupervisor, self()}}`. Concurrent work — the hourly PropagationGridWorker pulling f00-f18 HRRR range downloads, the GEFS worker scoring its grid, and the Recalibrator's 20-way positive/negative fan-out — no longer funnels through a single supervisor PID. Light/one-shot Task.async + Task.start sites (hrrr_client surface/ pressure fan-out, application boot warmup, weather fire-and-forget) are left alone; partitioning only helps under sustained concurrency. --- lib/microwaveprop/application.ex | 6 +++++ lib/microwaveprop/propagation/recalibrator.ex | 13 +++++++---- lib/microwaveprop/weather/gefs_client.ex | 7 ++++-- lib/microwaveprop/weather/hrrr_client.ex | 20 ++++++++++++----- .../workers/gefs_fetch_worker.ex | 6 +++-- test/microwaveprop/application_test.exs | 22 +++++++++++++++++++ 6 files changed, 60 insertions(+), 14 deletions(-) diff --git a/lib/microwaveprop/application.ex b/lib/microwaveprop/application.ex index e87771e2..61f2d4f1 100644 --- a/lib/microwaveprop/application.ex +++ b/lib/microwaveprop/application.ex @@ -25,6 +25,12 @@ defmodule Microwaveprop.Application do Microwaveprop.Repo, {Cluster.Supervisor, [topologies, [name: Microwaveprop.ClusterSupervisor]]}, {Phoenix.PubSub, name: Microwaveprop.PubSub}, + # Partitioned Task.Supervisor — callers route heavy `async_stream` + # work through `{:via, PartitionSupervisor, {Microwaveprop.TaskSupervisor, + # self()}}` so concurrent workers (recalibrator, HRRR/GEFS range + # downloads, GEFS scoring) do not serialize on a single supervisor PID. + {PartitionSupervisor, + child_spec: Task.Supervisor, name: Microwaveprop.TaskSupervisor, partitions: System.schedulers_online()}, Microwaveprop.Cache, Microwaveprop.Propagation.ScoreCache, Microwaveprop.Propagation.ScoreCacheReconciler, diff --git a/lib/microwaveprop/propagation/recalibrator.ex b/lib/microwaveprop/propagation/recalibrator.ex index a540dbd0..036557f6 100644 --- a/lib/microwaveprop/propagation/recalibrator.ex +++ b/lib/microwaveprop/propagation/recalibrator.ex @@ -180,9 +180,11 @@ defmodule Microwaveprop.Propagation.Recalibrator do # query round-trips. Parallelize at 20 concurrent queries, well # under the :db pool size (30 in prod), so the recalibration # finishes in seconds instead of minutes. + supervisor = {:via, PartitionSupervisor, {Microwaveprop.TaskSupervisor, self()}} + factor_vectors = - contacts - |> Task.async_stream(&contact_to_factors(&1, factor_band), + supervisor + |> Task.Supervisor.async_stream_nolink(contacts, &contact_to_factors(&1, factor_band), max_concurrency: 20, timeout: 30_000, on_timeout: :kill_task @@ -213,8 +215,11 @@ defmodule Microwaveprop.Propagation.Recalibrator do factor_band = band_mhz || 10_000 - baselines - |> Task.async_stream( + supervisor = {:via, PartitionSupervisor, {Microwaveprop.TaskSupervisor, self()}} + + supervisor + |> Task.Supervisor.async_stream_nolink( + baselines, fn {lat, lon, time} -> case Weather.find_nearest_hrrr(lat, lon, time) do nil -> nil diff --git a/lib/microwaveprop/weather/gefs_client.ex b/lib/microwaveprop/weather/gefs_client.ex index e2479520..87264054 100644 --- a/lib/microwaveprop/weather/gefs_client.ex +++ b/lib/microwaveprop/weather/gefs_client.ex @@ -256,9 +256,12 @@ defmodule Microwaveprop.Weather.GefsClient do end) |> Enum.reverse() + supervisor = {:via, PartitionSupervisor, {Microwaveprop.TaskSupervisor, self()}} + results = - merged - |> Task.async_stream( + supervisor + |> Task.Supervisor.async_stream_nolink( + merged, fn {start, stop} -> case Req.get(url, [{:headers, [{"Range", "bytes=#{start}-#{stop}"}]} | req_options()]) do {:ok, %{status: 206, body: body}} -> {:ok, start, body} diff --git a/lib/microwaveprop/weather/hrrr_client.ex b/lib/microwaveprop/weather/hrrr_client.ex index 75fbd02e..f9b93c83 100644 --- a/lib/microwaveprop/weather/hrrr_client.ex +++ b/lib/microwaveprop/weather/hrrr_client.ex @@ -496,12 +496,16 @@ defmodule Microwaveprop.Weather.HrrrClient do # when sequential. With 40+ ranges per native-duct fetch, serial # download dominated the per-forecast-hour wall time. 8-way # parallelism drops a 20s download to ~3s while still respecting - # the mirror's modest connection concurrency. + # the mirror's modest connection concurrency. Routed through the + # partitioned Task.Supervisor to avoid contention when the hourly + # PropagationGridWorker fetches f00-f18 alongside recalibrator/GEFS. merged = merge_ranges(ranges) + supervisor = {:via, PartitionSupervisor, {Microwaveprop.TaskSupervisor, self()}} results = - merged - |> Task.async_stream( + supervisor + |> Task.Supervisor.async_stream_nolink( + merged, fn {start, stop} -> case Req.get(url, [{:headers, [{"Range", "bytes=#{start}-#{stop}"}]} | req_options()]) do {:ok, %{status: 206, body: body}} -> {:ok, start, body} @@ -539,11 +543,15 @@ defmodule Microwaveprop.Weather.HrrrClient do defp download_grib_ranges_remote(url, ranges) do # Merge adjacent/overlapping ranges to reduce HTTP requests merged = merge_ranges(ranges) + supervisor = {:via, PartitionSupervisor, {Microwaveprop.TaskSupervisor, self()}} - # Download ranges in parallel + # Download ranges in parallel, routed through the partitioned + # Task.Supervisor so concurrent f00-f18 fetches do not serialize on + # a single supervisor PID. results = - merged - |> Task.async_stream( + supervisor + |> Task.Supervisor.async_stream_nolink( + merged, fn {start, stop} -> case Req.get(url, [{:headers, [{"Range", "bytes=#{start}-#{stop}"}]} | req_options()]) do {:ok, %{status: 206, body: body}} -> {:ok, start, body} diff --git a/lib/microwaveprop/workers/gefs_fetch_worker.ex b/lib/microwaveprop/workers/gefs_fetch_worker.ex index 75484059..78f20f9c 100644 --- a/lib/microwaveprop/workers/gefs_fetch_worker.ex +++ b/lib/microwaveprop/workers/gefs_fetch_worker.ex @@ -148,10 +148,12 @@ defmodule Microwaveprop.Workers.GefsFetchWorker do # `gefs_profiles` for future per-contact extended-horizon lookups. defp score_and_persist(profiles, run_time, fh, valid_time) do grid_data = profiles_to_grid(profiles) + supervisor = {:via, PartitionSupervisor, {Microwaveprop.TaskSupervisor, self()}} scores = - grid_data - |> Task.async_stream( + supervisor + |> Task.Supervisor.async_stream_nolink( + grid_data, fn {{lat, lon}, profile} -> profile |> Propagation.score_grid_point(valid_time, lat, lon) diff --git a/test/microwaveprop/application_test.exs b/test/microwaveprop/application_test.exs index c4a57e9f..d46c573a 100644 --- a/test/microwaveprop/application_test.exs +++ b/test/microwaveprop/application_test.exs @@ -3,6 +3,28 @@ defmodule Microwaveprop.ApplicationTest do alias Microwaveprop.Application, as: App + describe "Microwaveprop.TaskSupervisor partition supervisor" do + test "is started and exposes schedulers_online partitions" do + # A partitioned Task.Supervisor removes contention on a single + # supervisor PID when many callers run Task.async_stream in parallel + # (recalibrator, hrrr_client range downloads, gefs worker, etc.). + %{active: active, specs: specs} = PartitionSupervisor.count_children(Microwaveprop.TaskSupervisor) + + expected = System.schedulers_online() + assert active == expected + assert specs == expected + end + + test "routes async_stream work through a partition keyed on the caller" do + results = + {:via, PartitionSupervisor, {Microwaveprop.TaskSupervisor, self()}} + |> Task.Supervisor.async_stream(1..5, fn n -> n * 2 end, max_concurrency: 2) + |> Enum.map(fn {:ok, v} -> v end) + + assert results == [2, 4, 6, 8, 10] + end + end + describe "build_timestamp/0" do setup do original_env = System.get_env("DEPLOY_TIMESTAMP")