From 95976b6b9065be5e9bc2f9ed72df1a9190de2ffb Mon Sep 17 00:00:00 2001 From: Graham McIntire Date: Sat, 1 Aug 2026 12:48:01 -0500 Subject: [PATCH] perf(grid-rs): drop legacy .mp.gz write, parallelize decodes, overlap fetch+decode, cap retries Four optimisations targeting the 11.4s chain-step critical path: 1. Drop legacy .mp.gz dual-write (~2.5-2.8s, 24% of step) The Elixir reader prefers .sgrid; the chunked gzip+msgpack write was pure overhead on the Rust write path. Removed scalar_future from both run_chain_step and run_analysis_step. 2. Parallelize wgrib2 decodes (~0.5-1s) sfc and prs decode ran serially (await sfc, then await prs). Both now spawn concurrently on the blocking pool. 3. Overlap sfc decode with prs fetch (~1-2s) sfc (9 messages) resolves before prs (39 messages). Start the sfc wgrib2 subprocess as soon as the sfc blob lands instead of waiting for both blobs. The sfc decode hides behind the remaining prs fetch. 4. Cap grib range retries from 5 to 3 2^n exponential backoff meant 1+2+4+8+16 = 31s worst-case per failed range. Three retries caps at 1+2+4 = 7s. 404s already fail fast. Also: instrument .sgrid write (observe_stage) so it appears in the prop_grid_rs_stage_duration_seconds histogram alongside fetch/decode/derive. Expected mean chain step: ~6-7s (down from 11.4s). --- rust/prop_grid_rs/src/fetcher.rs | 5 +- rust/prop_grid_rs/src/pipeline.rs | 120 +++++++++++------------------- 2 files changed, 48 insertions(+), 77 deletions(-) diff --git a/rust/prop_grid_rs/src/fetcher.rs b/rust/prop_grid_rs/src/fetcher.rs index 5b725bd8..1af50ab0 100644 --- a/rust/prop_grid_rs/src/fetcher.rs +++ b/rust/prop_grid_rs/src/fetcher.rs @@ -600,7 +600,7 @@ async fn retry_get_range( let mut headers = HeaderMap::new(); headers.insert(RANGE, value); - for attempt in 0..5 { + for attempt in 0..3 { match client.get(url).headers(headers.clone()).send().await { Ok(resp) => { let status = resp.status().as_u16(); @@ -622,6 +622,9 @@ async fn retry_get_range( } fn is_retryable_status(status: u16) -> bool { + // 404 means the forecast hour isn't published yet — a permanent + // failure that retrying won't fix. Fail fast so the caller can + // move on instead of burning 31 s in exponential backoff. matches!(status, 429 | 500 | 502 | 503 | 504) } diff --git a/rust/prop_grid_rs/src/pipeline.rs b/rust/prop_grid_rs/src/pipeline.rs index aa470394..11e90fab 100644 --- a/rust/prop_grid_rs/src/pipeline.rs +++ b/rust/prop_grid_rs/src/pipeline.rs @@ -104,47 +104,54 @@ pub async fn run_chain_step( let sfc_wanted = fetcher::surface_messages_owned(); let prs_wanted = fetcher::pressure_messages_grid(); - let sfc_fut = client.fetch_product_blob( + + // Pin so we can drive them sequentially without losing the prs + // future when sfc resolves first. + let mut sfc_fut = Box::pin(client.fetch_product_blob( date, hour, Product::Surface, step.forecast_hour, &sfc_wanted, - ); - let prs_fut = client.fetch_product_blob( + )); + let mut prs_fut = Box::pin(client.fetch_product_blob( date, hour, Product::Pressure, step.forecast_hour, &prs_wanted, - ); - let fetch_started = std::time::Instant::now(); - let (sfc_blob, prs_blob) = tokio::try_join!(sfc_fut, prs_fut)?; - metrics::record_stage("fetch", fetch_started.elapsed()); - if sfc_blob.is_empty() { - return Err(PipelineError::EmptyGrib); - } + )); let grid_spec = wgrib2_grid_spec(); let sfc_pattern = match_pattern(&sfc_wanted); let prs_pattern = match_pattern(&prs_wanted); - // Decoder runs wgrib2 subprocess — block so we don't hog the tokio - // reactor. `spawn_blocking` lifts it onto the dedicated pool. - let decode_started = std::time::Instant::now(); - let grid_spec_sfc = grid_spec; - let sfc_grid = tokio::task::spawn_blocking(move || { - decoder::extract_grid(&sfc_blob, &sfc_pattern, grid_spec_sfc) - }) - .await - .expect("blocking join")?; + let fetch_started = std::time::Instant::now(); + // sfc (9 messages) resolves before prs (39 messages). Start sfc + // decode as soon as it lands so the wgrib2 subprocess overlaps + // with the remaining prs byte-range downloads. + let sfc_blob = sfc_fut.as_mut().await?; + if sfc_blob.is_empty() { + return Err(PipelineError::EmptyGrib); + } + let grid_spec_sfc = grid_spec; + let sfc_decode = tokio::task::spawn_blocking(move || { + decoder::extract_grid(&sfc_blob, &sfc_pattern, grid_spec_sfc) + }); + + // Now await prs while sfc decodes in the background. + let prs_blob = prs_fut.as_mut().await?; + metrics::record_stage("fetch", fetch_started.elapsed()); + + // Both decodes run concurrently on the blocking pool. + let decode_started = std::time::Instant::now(); let grid_spec_prs = grid_spec; - let prs_grid = tokio::task::spawn_blocking(move || { + let prs_decode = tokio::task::spawn_blocking(move || { decoder::extract_grid(&prs_blob, &prs_pattern, grid_spec_prs) - }) - .await - .expect("blocking join")?; + }); + let sfc_grid = sfc_decode.await.expect("blocking join")?; + let prs_grid = prs_decode.await.expect("blocking join")?; metrics::record_stage("decode", decode_started.elapsed()); let mut merged = sfc_grid; @@ -179,36 +186,19 @@ pub async fn run_chain_step( }) }; - // Dense `.sgrid` body built from the scalar rows while they're - // still available. The old chunked write below moves the rows; this - // must come first. + // Dense `.sgrid` body built from the scalar rows. This is the + // primary weather-scalar artifact; the Elixir reader prefers it + // over the legacy chunked `.mp.gz` format, which is only written + // as a cold-miss fallback by the Elixir side. let sgrid_body = sgrid::build_body(&spec, &fused.scalar_rows); - // Persist the derived scalar artifact alongside the raw profile so - // Elixir's `/weather` reads land on a pre-derived chunk file - // instead of falling back to the cold ProfilesFile decode + per-cell - // SoundingParams + WeatherLayers derivation. Identical wire format - // to what the Elixir writer produces — the reader doesn't care which - // side wrote the bytes. - let scores_dir_for_scalars = scores_dir_owned.clone(); - let scalar_future = { - let rows = fused.scalar_rows; - tokio::task::spawn_blocking(move || { - metrics::observe_stage("write_scalar", || { - weather_scalar_file::write_atomic(&scores_dir_for_scalars, valid_time, &rows) - .map(|_| 0u32) - }) - }) - }; - - // Write the `.sgrid` in parallel with the chunked write. During - // the transition window both formats coexist; the Elixir reader - // prefers `.sgrid` when it exists. let scores_dir_for_sgrid = scores_dir_owned.clone(); let sgrid_future = { tokio::task::spawn_blocking(move || { - sgrid::write_atomic(&scores_dir_for_sgrid, valid_time, &spec, &sgrid_body, false) - .map(|_| 0u32) + metrics::observe_stage("write_sgrid", || { + sgrid::write_atomic(&scores_dir_for_sgrid, valid_time, &spec, &sgrid_body, false) + .map(|_| 0u32) + }) }) }; @@ -226,10 +216,6 @@ pub async fn run_chain_step( .await .expect("blocking join") .map_err(PipelineError::ProfileWrite)?; - scalar_future - .await - .expect("blocking join") - .map_err(PipelineError::ScalarWrite)?; sgrid_future .await .expect("blocking join") @@ -1433,31 +1419,17 @@ pub async fn run_analysis_step( }) }; - // Dense `.sgrid` body built from the scalar rows while they're - // still available (the chunked write below moves them). + // Dense `.sgrid` body built from the scalar rows. Primary + // weather-scalar artifact; see run_chain_step for rationale. let sgrid_body = sgrid::build_body(&spec, &fused.scalar_rows); - // Same wire format as Elixir's `ScalarFile`. See run_chain_step - // (forecast hours) for the full rationale — both code paths go - // through the same writer so f00 and f01..f48 land identical - // artifacts on NFS. - let scores_dir_for_scalars = scores_dir_owned.clone(); - let scalar_future = { - let rows = fused.scalar_rows; - tokio::task::spawn_blocking(move || { - metrics::observe_stage("write_scalar", || { - weather_scalar_file::write_atomic(&scores_dir_for_scalars, valid_time, &rows) - .map(|_| 0u32) - }) - }) - }; - - // Write the `.sgrid` alongside the chunked scalar artifact. let scores_dir_for_sgrid = scores_dir_owned.clone(); let sgrid_future = { tokio::task::spawn_blocking(move || { - sgrid::write_atomic(&scores_dir_for_sgrid, valid_time, &spec, &sgrid_body, false) - .map(|_| 0u32) + metrics::observe_stage("write_sgrid", || { + sgrid::write_atomic(&scores_dir_for_sgrid, valid_time, &spec, &sgrid_body, false) + .map(|_| 0u32) + }) }) }; @@ -1475,10 +1447,6 @@ pub async fn run_analysis_step( .await .expect("blocking join") .map_err(PipelineError::ProfileWrite)?; - scalar_future - .await - .expect("blocking join") - .map_err(PipelineError::ScalarWrite)?; sgrid_future .await .expect("blocking join")