perf: fix critical performance issues
- NexradCache: replace :ets.tab2list() with foldl to avoid ~1.3 GB heap allocation on full cache (OOM risk) - GridCache: replace :ets.select full-table copies with foldl in ets_latest_valid_time and ets_prune_keep_latest - Weather.backfill_hrrr_batch: batch N individual UPDATEs into single VALUES-based SQL statement eliminating ~500 round-trips per batch - Add 9 missing database indexes across 6 tables for hot-path queries (contacts, contact_edits, hrrr_profiles, hrrr_native_profiles, grid_tasks, rover_mission_paths, fixed_stations, beacons)
This commit is contained in:
parent
1899b3fb5a
commit
26be066f77
4 changed files with 105 additions and 31 deletions
|
|
@ -537,19 +537,45 @@ defmodule Microwaveprop.Weather do
|
||||||
end
|
end
|
||||||
end)
|
end)
|
||||||
|
|
||||||
Enum.each(updates, fn u ->
|
now = DateTime.truncate(DateTime.utc_now(), :second)
|
||||||
HrrrProfile
|
|
||||||
|> where([h], h.id == ^u.id)
|
if updates != [] do
|
||||||
|> Repo.update_all(
|
{values_sql, params} =
|
||||||
set: [
|
updates
|
||||||
surface_refractivity: u.surface_refractivity,
|
|> Enum.with_index()
|
||||||
min_refractivity_gradient: u.min_refractivity_gradient,
|
|> Enum.reduce({"", []}, fn u, {sql, params} ->
|
||||||
ducting_detected: u.ducting_detected,
|
base = Enum.count(params)
|
||||||
duct_characteristics: u.duct_characteristics,
|
frag = "($#{base + 1}::uuid, $#{base + 2}, $#{base + 3}, $#{base + 4}, $#{base + 5}, $#{base + 6})"
|
||||||
updated_at: DateTime.truncate(DateTime.utc_now(), :second)
|
sep = if sql == "", do: "", else: ", "
|
||||||
]
|
|
||||||
|
params =
|
||||||
|
params ++
|
||||||
|
[
|
||||||
|
Ecto.UUID.dump!(u.id),
|
||||||
|
u.surface_refractivity,
|
||||||
|
u.min_refractivity_gradient,
|
||||||
|
u.ducting_detected,
|
||||||
|
u.duct_characteristics || [],
|
||||||
|
now
|
||||||
|
]
|
||||||
|
|
||||||
|
{sql <> sep <> frag, params}
|
||||||
|
end)
|
||||||
|
|
||||||
|
Repo.query!(
|
||||||
|
"UPDATE hrrr_profiles AS h " <>
|
||||||
|
"SET surface_refractivity = v.surface_refractivity, " <>
|
||||||
|
"min_refractivity_gradient = v.min_refractivity_gradient, " <>
|
||||||
|
"ducting_detected = v.ducting_detected, " <>
|
||||||
|
"duct_characteristics = v.duct_characteristics, " <>
|
||||||
|
"updated_at = v.updated_at " <>
|
||||||
|
"FROM (VALUES #{values_sql}) " <>
|
||||||
|
"AS v(id, surface_refractivity, min_refractivity_gradient, " <>
|
||||||
|
"ducting_detected, duct_characteristics, updated_at) " <>
|
||||||
|
"WHERE h.id = v.id",
|
||||||
|
params
|
||||||
)
|
)
|
||||||
end)
|
end
|
||||||
|
|
||||||
length(updates)
|
length(updates)
|
||||||
end
|
end
|
||||||
|
|
|
||||||
|
|
@ -291,20 +291,21 @@ defmodule Microwaveprop.Weather.GridCache do
|
||||||
end
|
end
|
||||||
|
|
||||||
defp ets_latest_valid_time do
|
defp ets_latest_valid_time do
|
||||||
match_spec = [{{:"$1", :_}, [], [:"$1"]}]
|
:ets.foldl(fn
|
||||||
|
{vt, _}, nil -> vt
|
||||||
case :ets.select(@table, match_spec) do
|
{vt, _}, latest -> if DateTime.compare(vt, latest) == :gt, do: vt, else: latest
|
||||||
[] -> nil
|
end, nil, @table)
|
||||||
times -> Enum.max(times, DateTime)
|
|
||||||
end
|
|
||||||
end
|
end
|
||||||
|
|
||||||
defp ets_prune_keep_latest(keep) do
|
defp ets_prune_keep_latest(keep) do
|
||||||
case :ets.select(@table, [{{:"$1", :_}, [], [:"$1"]}]) do
|
times =
|
||||||
|
:ets.foldl(fn {vt, _}, acc -> [vt | acc] end, [], @table)
|
||||||
|
|
||||||
|
case times do
|
||||||
[] ->
|
[] ->
|
||||||
0
|
0
|
||||||
|
|
||||||
times ->
|
_ ->
|
||||||
sorted = Enum.sort(times, {:desc, DateTime})
|
sorted = Enum.sort(times, {:desc, DateTime})
|
||||||
to_drop = Enum.drop(sorted, keep)
|
to_drop = Enum.drop(sorted, keep)
|
||||||
Enum.each(to_drop, &:ets.delete(@table, &1))
|
Enum.each(to_drop, &:ets.delete(@table, &1))
|
||||||
|
|
|
||||||
|
|
@ -44,21 +44,25 @@ defmodule Microwaveprop.Weather.NexradCache do
|
||||||
end
|
end
|
||||||
|
|
||||||
defp enforce_size_cap do
|
defp enforce_size_cap do
|
||||||
size = :ets.info(@table, :size)
|
if :ets.info(@table, :size) > @max_entries do
|
||||||
|
case oldest_key() do
|
||||||
|
nil -> :ok
|
||||||
|
ts -> :ets.delete(@table, ts)
|
||||||
|
end
|
||||||
|
|
||||||
if size > @max_entries do
|
enforce_size_cap()
|
||||||
# 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
|
||||||
end
|
end
|
||||||
|
|
||||||
|
# Iterate ETS with foldl so frame payloads (~66 MB each) never land on the
|
||||||
|
# calling process's heap — :ets.tab2list() at capacity would allocate ~1.3 GB.
|
||||||
|
defp oldest_key do
|
||||||
|
:ets.foldl(fn
|
||||||
|
{ts, _, _}, nil -> ts
|
||||||
|
{ts, _, _}, oldest -> if DateTime.before?(ts, oldest), do: ts, else: oldest
|
||||||
|
end, nil, @table)
|
||||||
|
end
|
||||||
|
|
||||||
@spec prune_older_than(DateTime.t()) :: non_neg_integer()
|
@spec prune_older_than(DateTime.t()) :: non_neg_integer()
|
||||||
def prune_older_than(cutoff_ts) do
|
def prune_older_than(cutoff_ts) do
|
||||||
match_spec = [{{:"$1", :_, :_}, [{:<, :"$1", {:const, cutoff_ts}}], [true]}]
|
match_spec = [{{:"$1", :_, :_}, [{:<, :"$1", {:const, cutoff_ts}}], [true]}]
|
||||||
|
|
|
||||||
|
|
@ -0,0 +1,43 @@
|
||||||
|
defmodule Microwaveprop.Repo.Migrations.AddPerformanceIndexes do
|
||||||
|
use Ecto.Migration
|
||||||
|
|
||||||
|
def change do
|
||||||
|
# contacts.flagged_invalid — admin flagged-contacts page filters + sorts
|
||||||
|
create index(:contacts, [:flagged_invalid, :inserted_at, :id])
|
||||||
|
|
||||||
|
# contact_edits.inserted_at — ORDER BY DESC on list and per-contact queries
|
||||||
|
create index(:contact_edits, [:inserted_at])
|
||||||
|
|
||||||
|
# contact_edits(contact_id, user_id, status) — pending-edit lookup
|
||||||
|
create index(:contact_edits, [:contact_id, :user_id, :status])
|
||||||
|
|
||||||
|
# hrrr_profiles.surface_refractivity IS NULL — backfill query scans for nulls
|
||||||
|
create index(:hrrr_profiles, [:surface_refractivity],
|
||||||
|
where: "surface_refractivity IS NULL",
|
||||||
|
name: "hrrr_profiles_surface_refractivity_null_idx"
|
||||||
|
)
|
||||||
|
|
||||||
|
# hrrr_native_profiles.bulk_richardson IS NULL — derive task scans for nulls
|
||||||
|
create index(:hrrr_native_profiles, [:bulk_richardson],
|
||||||
|
where: "bulk_richardson IS NULL AND level_count > 2",
|
||||||
|
name: "hrrr_native_profiles_bulk_richardson_null_idx"
|
||||||
|
)
|
||||||
|
|
||||||
|
# grid_tasks.claimed_at for status='running' — reclaim-stale query
|
||||||
|
create index(:grid_tasks, [:claimed_at],
|
||||||
|
where: "status = 'running'",
|
||||||
|
name: "grid_tasks_running_claimed_at_idx"
|
||||||
|
)
|
||||||
|
|
||||||
|
# rover_mission_paths composite — worker lookup on all 4 columns
|
||||||
|
create index(:rover_mission_paths, [:mission_id, :rover_location_id, :station_id, :band_mhz],
|
||||||
|
name: "rover_mission_paths_lookup_idx"
|
||||||
|
)
|
||||||
|
|
||||||
|
# fixed_stations(user_id, position, inserted_at) — WHERE + ORDER BY
|
||||||
|
create index(:fixed_stations, [:user_id, :position, :inserted_at])
|
||||||
|
|
||||||
|
# beacons(user_id, inserted_at) — WHERE + ORDER BY
|
||||||
|
create index(:beacons, [:user_id, :inserted_at])
|
||||||
|
end
|
||||||
|
end
|
||||||
Loading…
Add table
Reference in a new issue