graham/prop
1
0
Fork 0
Code Issues Pull requests Projects Releases Packages Wiki Activity Actions

perf(grid-rs): stream bands to disk, drop merged grid early

Browse source 
Previous pipeline held three overlapping large structures in RAM:
1. `merged: HashMap<Key, HashMap<String, f32>>` — 92k outer × ~60 String
   keys inner = ~200 MB of fragmented heap
2. `per_band: HashMap<u32, Vec<ScorePoint>>` — all 23 bands × 92k
   ScorePoints before the first write
3. A `scores.clone()` per band during the write loop (doubled that
   band's footprint transiently)

Consolidate into:
- Consume `merged` into a compact `Vec<(lat, lon, Conditions,
  BandInvariants)>`; the huge String-keyed HashMaps drop as soon as
  collection finishes.
- Loop over bands: score → write → vector drops at end of iteration.
  One band's ~1.8 MB score vector lives at a time.

Peak heap goes from ~1.5 Gi to ~250 Mi on the chain step. 2 Gi limit
stays, but we now have 8× headroom instead of tight-fit.
This commit is contained in:
Graham McIntire 2026-04-19 16:44:29 -05:00
parent ae4b0742fb
commit 6d914614ac
No known key found for this signature in database
GPG key ID: F4ABF488E6029E59
1 changed files with 25 additions and 24 deletions
Show stats Download patch file Download diff file Expand all files Collapse all files

49
rust/prop_grid_rs/src/pipeline.rs
View file

@ -6,7 +6,6 @@
//! minus the f00-only enrichment (native duct, NEXRAD, commercial link
//! boost, ProfilesFile write) — Elixir retains f00.
use std::collections::HashMap;
use std::path::Path;
use chrono::{DateTime, Datelike, Timelike, Utc};
@ -108,33 +107,35 @@ pub async fn run_chain_step(
let merged = merge_grids(sfc_grid, prs_grid);
let point_count = merged.len() as u32;
// Score every point × every band; one ScorePoint per (cell, band).
// Step 1: consume `merged` into a compact `Vec<(lat, lon, Conditions,
// invariants)>`. This is the biggest win — `merged`'s inner HashMaps
// of `String -> f32` account for ~200 MB of heap (92k × ~60 string
// keys) and are dropped as soon as this loop finishes.
let prepared: Vec<(f64, f64, Conditions, scorer::BandInvariants)> = merged
.into_iter()
.filter_map(|(key, cell)| {
let (lat, lon) = decoder::key_to_latlon(key);
let conditions = cell_to_conditions(&cell, lat, lon, &valid_time)?;
let invariants = scorer::precompute_band_invariants(&conditions);
Some((lat, lon, conditions, invariants))
})
.collect();
// Step 2: for each band, score → write → drop. One band's score
// vector (~92k × 20 B ≈ 1.8 MB) lives at a time instead of holding
// all 23 bands at once. The write runs on the blocking pool so the
// reactor stays responsive.
let bands = band_config::all_bands();
let mut per_band: HashMap<u32, Vec<ScorePoint>> =
HashMap::with_capacity(bands.len());
for (key, cell) in &merged {
let (lat, lon) = decoder::key_to_latlon(*key);
let conditions = match cell_to_conditions(cell, lat, lon, &valid_time) {
Some(c) => c,
None => continue,
};
let invariants = scorer::precompute_band_invariants(&conditions);
for band in bands {
let r = scorer::composite_score_with(&conditions, band, Some(invariants));
per_band
.entry(band.freq_mhz)
.or_default()
.push(ScorePoint { lat, lon, score: r.score });
}
}
let scores_dir = scores_dir.to_path_buf();
let mut files_written = 0;
for (band_mhz, scores) in &per_band {
for band in bands {
let mut scores: Vec<ScorePoint> = Vec::with_capacity(prepared.len());
for (lat, lon, conditions, invariants) in &prepared {
let r = scorer::composite_score_with(conditions, band, Some(*invariants));
scores.push(ScorePoint { lat: *lat, lon: *lon, score: r.score });
}
let dir = scores_dir.clone();
let band_mhz = *band_mhz;
let scores = scores.clone();
let band_mhz = band.freq_mhz;
tokio::task::spawn_blocking(move || {
scores_file::write_atomic(&dir, band_mhz, valid_time, &scores)
})