diff --git a/CHANGELOG.md b/CHANGELOG.md index 1bdd20bd..fc1c6c29 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -54,6 +54,13 @@ before 1.0). ### Changed +- **Load in-flight outs: iterate pins, not parents×layers.** Post-milestone + unique spent parents (~160k/window) were probing every in-flight layer + via `get_out` in `fill_missing_parent_ranges` and pin adopt. Union + `out_ids` / `for_each_out` instead. `ibd: perf` `stamp_sub` prints + **`fill=`** / **`idx=`** for that wall (it was inside `batch=` with no + token). + - **Pin `merge_outs` no-op Arc:** empty / already-covered `checked`+`live` keep the outs Arc (assemble sticky `ptr_eq`). RCU compose borrows `live` instead of cloning script bytes on every retry. diff --git a/OPERATOR.md b/OPERATOR.md index 581a2614..81902df5 100644 --- a/OPERATOR.md +++ b/OPERATOR.md @@ -211,7 +211,7 @@ Requires **tip mode** (`node: catch-up complete … tip tracking`). During IBD u | Line | Level | Use | |------|-------|-----| | `ibd: progress` | INFO | Tip rate, `loadq`/`scriptq`/`writeq`, `txs=` (Class A / `tx.idx` count), horizon, tip ETA, **`bq soft=n/win RAM=`** (in-RAM body queue; soft densify: under ~100 MiB free ahead, over that only ~1 min confirm window, at/over 1 GiB assign-stop fill holes in the fetched range only) | -| `ibd: perf` | DEBUG | Inflight + **`bq soft= RAM=`**; **`load=`** is pin+assemble only. **`load_thr pack/stamp/pin/asm/prune`** is the load OS thread. **`stamp=`** nests **`pack=`** (plan HashMap) vs **`head=`** (leftover TipOnly). After a published wave `head=` is ~0. **`script=`** is verify ns (`jobs=` / `skip=`); recv/send are wait. **`lookup_thr keep=`** is live-union splice. **`pin_txid=`** vs leftover `tx.head` | +| `ibd: perf` | DEBUG | Inflight + **`bq soft= RAM=`**; **`load=`** is pin+assemble only. **`load_thr pack/stamp/pin/asm/prune`** is the load OS thread. **`stamp=`** nests **`pack=`** (plan HashMap) vs **`head=`** (leftover TipOnly) vs **`fill=`** (`fill_missing`; **`idx=`** spent/body ranges). After a published wave `head=` is ~0. **`script=`** is verify ns (`jobs=` / `skip=`); recv/send are wait. **`lookup_thr keep=`** is live-union splice. **`pin_txid=`** vs leftover `tx.head` | | `ibd: sizes` | DEBUG | RSS + work path + **`bq soft=` / `RAM=`** + **conf_plans** + confirm pipe | | `ibd: perf_dbg` | DEBUG | µs/blk load/write, pin/edge detail, **plan_batch** (`us/pin_txid` vs `probe/idx/body us/key`) + **class_a commit** | diff --git a/crates/rbitcoin-consensus/src/confirm_run/pin.rs b/crates/rbitcoin-consensus/src/confirm_run/pin.rs index 843c3b95..904b79d0 100644 --- a/crates/rbitcoin-consensus/src/confirm_run/pin.rs +++ b/crates/rbitcoin-consensus/src/confirm_run/pin.rs @@ -129,15 +129,12 @@ pub(super) fn pin_for_wire_batch( } if let Some(ifo) = in_flight { - for (id, need) in &parent_vouts { - if plan_by_id.contains_key(id) { - continue; + ifo.for_each_out(|id, pin| { + if plan_by_id.contains_key(&id) || !parent_vouts.contains_key(&id) { + return; } - if let Some(pin) = ifo.get_out(*id) { - let _ = need; - plan_by_id.insert(*id, std::sync::Arc::clone(pin)); - } - } + plan_by_id.insert(id, std::sync::Arc::clone(pin)); + }); } for (id, _need) in &parent_vouts { if plan_by_id.contains_key(id) { diff --git a/crates/rbitcoin-net/src/ibd/perf_log.rs b/crates/rbitcoin-net/src/ibd/perf_log.rs index ab21d4bd..388d87dd 100644 --- a/crates/rbitcoin-net/src/ibd/perf_log.rs +++ b/crates/rbitcoin-net/src/ibd/perf_log.rs @@ -41,7 +41,8 @@ //! `scriptq` can stay empty. Prefer `lookup_thr busy=` / `thr load=busy/wait=` / //! `ready=` + `scriptq_hwm=` (OS-thread occupancy + queue high-water). High //! `load_thr stamp=` nests `pack=` (plan HashMap) vs `head=` (leftover TipOnly -//! `prep_head_fk_ns`). After a published wave `head=` is ~0. High `head=` + +//! `prep_head_fk_ns`) vs `fill=` (`fill_missing_parent_ranges`; `idx=` is +//! spent/body range batches). After a published wave `head=` is ~0. High `head=` + //! `ready>0` + `scriptq=1` ⇒ leftover TipOnly on load, not “scripts hungry.” //! High load_recv_wait + ready=0 ⇒ lookup is the pole. //! @@ -346,6 +347,9 @@ pub(crate) struct IbdPerfSample { pub stamp_batch_head_fk_ms: u64, pub stamp_batch_stamp_ms: u64, pub stamp_batch_finish_ms: u64, + /// `fill_missing_parent_ranges` (skip walk + idx). Nested `idx=` is spent/body range batches. + pub stamp_batch_fill_ms: u64, + pub stamp_batch_fill_idx_ms: u64, pub thr_load_recv_wait_ms: u64, pub thr_load_pack_ms: u64, pub thr_load_clone_ms: u64, @@ -436,6 +440,8 @@ pub(crate) struct IbdPerfSample { pub arch_prep_body_lookups: u64, pub arch_prep_stamp_ms: u64, pub arch_prep_finish_ms: u64, + pub arch_prep_fill_ms: u64, + pub arch_prep_fill_idx_ms: u64, pub arch_write_total_ms: u64, pub arch_write_reserve_ms: u64, pub arch_write_body_ms: u64, @@ -609,6 +615,8 @@ impl Default for IbdPerfSample { stamp_batch_head_fk_ms: 0, stamp_batch_stamp_ms: 0, stamp_batch_finish_ms: 0, + stamp_batch_fill_ms: 0, + stamp_batch_fill_idx_ms: 0, thr_load_recv_wait_ms: 0, thr_load_pack_ms: 0, thr_load_clone_ms: 0, @@ -686,6 +694,8 @@ impl Default for IbdPerfSample { arch_prep_body_lookups: 0, arch_prep_stamp_ms: 0, arch_prep_finish_ms: 0, + arch_prep_fill_ms: 0, + arch_prep_fill_idx_ms: 0, arch_write_total_ms: 0, arch_write_reserve_ms: 0, arch_write_body_ms: 0, @@ -1072,6 +1082,8 @@ pub(crate) fn sample( stamp_batch_head_fk_ms: ns_ms(arch_res.prep_head_fk_ns), stamp_batch_stamp_ms: ns_ms(arch_res.prep_stamp_ns), stamp_batch_finish_ms: ns_ms(arch_res.prep_finish_ns), + stamp_batch_fill_ms: ns_ms(arch_res.prep_fill_ns), + stamp_batch_fill_idx_ms: ns_ms(arch_res.prep_fill_idx_ns), thr_load_recv_wait_ms: ns_ms(thr.load_recv_wait_ns), thr_load_pack_ms: ns_ms(thr.load_pack_ns), thr_load_clone_ms: ns_ms(thr.load_clone_ns), @@ -1149,6 +1161,8 @@ pub(crate) fn sample( arch_prep_body_lookups: head_res.body_lookups, arch_prep_stamp_ms: ns_ms(arch_res.prep_stamp_ns), arch_prep_finish_ms: ns_ms(arch_res.prep_finish_ns), + arch_prep_fill_ms: ns_ms(arch_res.prep_fill_ns), + arch_prep_fill_idx_ms: ns_ms(arch_res.prep_fill_idx_ns), arch_write_total_ms: ns_ms(arch_res.write_total_ns), arch_write_reserve_ms: ns_ms(arch_res.write_reserve_ns), arch_write_body_ms: ns_ms(arch_res.write_body_ns), @@ -1209,6 +1223,7 @@ fn plan_batch_ms(s: &IbdPerfSample) -> u64 { .saturating_add(s.arch_prep_inflight_ms) .saturating_add(s.arch_prep_head_ms) .saturating_add(s.arch_prep_stamp_ms) + .saturating_add(s.arch_prep_fill_ms) .saturating_add(s.arch_prep_finish_ms) } @@ -1317,7 +1332,7 @@ pub(crate) fn format_info(s: &IbdPerfSample) -> String { batch_assign={}ms collect={}ms pin_txid={} pin_txid%={} pin_txid_ms={} \ leftover_n={} leftover_hit={} leftover_ms={} leftover_pend={} leftover_cdf0={} leftover_cdf3={} leftover_age_n={} \ recent={} recent_ms={} \ - head={}ms stamp={}ms finish={}ms)", + head={}ms stamp={}ms fill={}ms(idx={}ms) finish={}ms)", s.stamp_struct_ms, s.stamp_struct_txid_ms, s.stamp_struct_walk_ms, @@ -1340,6 +1355,8 @@ pub(crate) fn format_info(s: &IbdPerfSample) -> String { s.arch_recent_ms, s.stamp_batch_head_ms, s.stamp_batch_stamp_ms, + s.stamp_batch_fill_ms, + s.stamp_batch_fill_idx_ms, s.stamp_batch_finish_ms, )); } @@ -1701,7 +1718,7 @@ pub(crate) fn format_debug(s: &IbdPerfSample) -> String { out.push_str(&format!( " | plan_batch assign={} collect={} inflight={} pin_txid={}/{} pin_txid_ms={} \ us/pin_txid={} recent={} recent_ms={} head_fk={} head={} \ - stamp={} finish={} resolve_us/blk={} ext={} head_hit={}/{} \ + stamp={} finish={} fill={} fill_idx={} resolve_us/blk={} ext={} head_hit={}/{} \ stamp_n batch={}", s.arch_prep_assign_ms, s.arch_prep_collect_ms, @@ -1716,6 +1733,8 @@ pub(crate) fn format_debug(s: &IbdPerfSample) -> String { s.arch_prep_head_ms, s.arch_prep_stamp_ms, s.arch_prep_finish_ms, + s.arch_prep_fill_ms, + s.arch_prep_fill_idx_ms, resolve_us_blk, s.arch_ext_need, s.arch_head_hit, @@ -2380,6 +2399,8 @@ mod tests { s.stamp_batch_head_ms = 2; s.stamp_batch_stamp_ms = 1; s.stamp_batch_finish_ms = 1; + s.stamp_batch_fill_ms = 8; + s.stamp_batch_fill_idx_ms = 7; s.arch_prep_age_hit_n = 50; s.arch_prep_age_cdf0_pct = 10; s.arch_prep_age_cdf3_pct = 40; @@ -2394,6 +2415,8 @@ mod tests { s.arch_prep_head_ms = 3; s.arch_prep_stamp_ms = 1; s.arch_prep_finish_ms = 1; + s.arch_prep_fill_ms = 8; + s.arch_prep_fill_idx_ms = 7; s.arch_resolve_ns = 8_000_000; s.arch_resolve_blocks = 4; s.arch_head_hit = 20; @@ -2430,6 +2453,7 @@ mod tests { s.stamp_struct_walk_ms = 2; let info = format_info(&s); assert!(info.contains("stamp_sub("), "{info}"); + assert!(info.contains("fill=8ms(idx=7ms)"), "{info}"); assert!(info.contains("struct_txid=6ms"), "{info}"); assert!(info.contains("struct_walk=2ms"), "{info}"); assert!(info.contains("pin_txid=15"), "{info}"); diff --git a/crates/rbitcoin-query/src/archive.rs b/crates/rbitcoin-query/src/archive.rs index 7f6808e1..a1d6db04 100644 --- a/crates/rbitcoin-query/src/archive.rs +++ b/crates/rbitcoin-query/src/archive.rs @@ -1852,6 +1852,40 @@ mod tests { let _ = std::fs::remove_dir_all(&dir); } + /// In-flight CreatePin on the oldest layer still skips idx (no store row). + #[test] + fn fill_missing_skips_idx_when_oldest_layer_has_outs() { + let (dir, q) = temp_query("fill-skip-oldest-outs"); + let pin = std::sync::Arc::new(( + TxRecord { + txid: [1u8; 32], + version: 1, + locktime: 0, + input_start_fk: Fk::NULL, + input_count: 1, + output_start_fk: Fk::NULL, + output_count: 1, + }, + vec![OutputRecord::unspent(1, vec![0x51])], + )); + let mut log = crate::InFlightLog::new(); + log.note_layer(crate::InFlightLayer::from_plan_pins([(Fk(1), &pin)])); + for i in 10u8..18 { + let mut tid = [0u8; 32]; + tid[0] = i; + log.note_layer(crate::InFlightLayer::from_txid_fks([(tid, Fk(i as u64))])); + } + let mut idents = crate::U64Map::default(); + idents.insert(1, crate::ParentIdent::new([1u8; 32])); + crate::fill_missing_parent_ranges(q.store(), &log.snapshot(), &mut idents) + .expect("in-flight outs skip idx even on the oldest layer"); + assert!( + idents.get(&1).and_then(|p| p.body).is_none(), + "skip must not invent a body_range" + ); + let _ = std::fs::remove_dir_all(&dir); + } + #[test] fn archive_filter_need_header_fks_drops_archived() { use rbitcoin_store::HeaderRecord; diff --git a/crates/rbitcoin-query/src/in_flight.rs b/crates/rbitcoin-query/src/in_flight.rs index 64ec9a90..37247359 100644 --- a/crates/rbitcoin-query/src/in_flight.rs +++ b/crates/rbitcoin-query/src/in_flight.rs @@ -12,8 +12,9 @@ //! layer (Class C ∥ drain/seal). Call after pin — stamp skips `body_range` //! when this map still has CreatePin outs. //! -//! Lookup is newest→oldest scan over layers (O(L)); pack counts are small and -//! L is bounded by pipeline queue depth. +//! Per-key [`InFlightView::get_out`] / `get_create_fk` is newest→oldest (O(L)). +//! Bulk skip/adopt over many parents must iterate outs (O(pins)), not +//! parents×L — post-milestone unique parents dwarf pipeline CreatePins. use crate::archive::CreatePin; use crate::U64Map; @@ -255,6 +256,30 @@ impl InFlightView { None } + /// Union of CreatePin keys across layers. + pub fn out_ids(&self) -> crate::U64Set { + let n: usize = self.layers.iter().map(|l| l.outs.len()).sum(); + let mut ids = crate::U64Set::with_capacity_and_hasher(n, Default::default()); + for layer in self.layers.iter() { + for id in layer.outs.keys() { + ids.insert(*id); + } + } + ids + } + + /// Newest layer first; each id once (same winner as [`Self::get_out`]). + pub fn for_each_out(&self, mut f: impl FnMut(u64, &CreatePin)) { + let mut seen = crate::U64Set::default(); + for layer in self.layers.iter().rev() { + for (id, pin) in &layer.outs { + if seen.insert(*id) { + f(*id, pin); + } + } + } + } + /// Create fk for txid from prior uncommitted packs. #[inline] pub fn get_create_fk(&self, txid: &[u8; 32]) -> Option { @@ -327,6 +352,30 @@ mod tests { ); } + #[test] + fn out_ids_unions_layers_and_matches_get_out() { + let mut log = InFlightLog::new(); + let a = pin(1); + let b = pin(2); + log.note_layer(InFlightLayer::from_plan_pins([(Fk(1), &a)])); + log.note_layer(InFlightLayer::from_plan_pins([(Fk(2), &b)])); + let v = log.snapshot(); + let ids = v.out_ids(); + assert!(ids.contains(&1)); + assert!(ids.contains(&2)); + assert_eq!(ids.len(), 2); + assert!(v.get_out(1).is_some()); + assert!(v.get_out(99).is_none()); + assert!(!ids.contains(&99)); + let mut newest = None; + v.for_each_out(|id, _| { + if newest.is_none() { + newest = Some(id); + } + }); + assert_eq!(newest, Some(2), "newest layer first"); + } + #[test] fn note_does_not_mutate_prior_layer_arcs() { let mut log = InFlightLog::new(); diff --git a/crates/rbitcoin-query/src/lib.rs b/crates/rbitcoin-query/src/lib.rs index 782358fc..4868a7f2 100644 --- a/crates/rbitcoin-query/src/lib.rs +++ b/crates/rbitcoin-query/src/lib.rs @@ -474,6 +474,10 @@ pub mod archive_phase_stats { pub static RESOLVED_STAMP: AtomicU64 = AtomicU64::new(0); /// `fill_missing_parent_ranges` entries (stamp + optional prestamp). pub static FILL_MISSING_N: AtomicU64 = AtomicU64::new(0); + /// Wall of `fill_missing_parent_ranges` (skip walk + idx). + pub static FILL_MISSING_NS: AtomicU64 = AtomicU64::new(0); + /// `tx_body_range_batch` + `tx_spent_range_batch` inside fill_missing. + pub static FILL_MISSING_IDX_NS: AtomicU64 = AtomicU64::new(0); /// Full load batch wall (struct → lookup → enqueue wait). pub static PREP_TOTAL_NS: AtomicU64 = AtomicU64::new(0); @@ -534,6 +538,8 @@ pub mod archive_phase_stats { pub prep_head_fk_ns: u64, pub prep_stamp_ns: u64, pub prep_finish_ns: u64, + pub prep_fill_ns: u64, + pub prep_fill_idx_ns: u64, pub prep_publish_ns: u64, pub prep_qwait_ns: u64, pub prep_blocks: u64, @@ -558,6 +564,7 @@ pub mod archive_phase_stats { .saturating_add(self.prep_head_ns) .saturating_add(self.prep_stamp_ns) .saturating_add(self.prep_finish_ns) + .saturating_add(self.prep_fill_ns) .saturating_add(self.prep_publish_ns) .saturating_add(self.prep_qwait_ns) } @@ -623,6 +630,8 @@ pub mod archive_phase_stats { prep_head_fk_ns: prep_head_fk, prep_stamp_ns: PREP_STAMP_NS.swap(0, Ordering::Relaxed), prep_finish_ns: PREP_FINISH_NS.swap(0, Ordering::Relaxed), + prep_fill_ns: FILL_MISSING_NS.swap(0, Ordering::Relaxed), + prep_fill_idx_ns: FILL_MISSING_IDX_NS.swap(0, Ordering::Relaxed), prep_publish_ns: PREP_PUBLISH_NS.swap(0, Ordering::Relaxed), prep_qwait_ns: PREP_QWAIT_NS.swap(0, Ordering::Relaxed), prep_blocks: PREP_BLOCKS.swap(0, Ordering::Relaxed), @@ -677,6 +686,14 @@ pub mod archive_phase_stats { }); } + #[inline] + pub fn note_fill_missing_ns(all_ns: u64, idx_ns: u64) { + exclusive::with(|| { + add(&FILL_MISSING_NS, all_ns); + add(&FILL_MISSING_IDX_NS, idx_ns); + }); + } + /// Leftover TipOnly pending hits + winner age buckets (load stamp). #[inline] pub fn note_leftover_mix(pend: u64, age0: u64, age3: u64, age_n: u64) { diff --git a/crates/rbitcoin-query/src/stamp.rs b/crates/rbitcoin-query/src/stamp.rs index c1832456..f9cc0f30 100644 --- a/crates/rbitcoin-query/src/stamp.rs +++ b/crates/rbitcoin-query/src/stamp.rs @@ -215,10 +215,12 @@ pub fn fill_missing_parent_ranges( idents: &mut U64Map, ) -> Result<(), QueryError> { crate::archive_phase_stats::note_fill_missing(); + let t_all = Instant::now(); + let inflight_outs = in_flight.out_ids(); let mut need_body: Vec = Vec::new(); let mut need_spent: Vec = Vec::new(); for (&id, ident) in idents.iter() { - if in_flight.get_out(id).is_some() { + if inflight_outs.contains(&id) { continue; } let fk = Fk(id); @@ -229,6 +231,7 @@ pub fn fill_missing_parent_ranges( need_spent.push(fk); } } + let t_idx = Instant::now(); let mut body_filled = U64Set::default(); if !need_body.is_empty() { let filled = store.tx_body_range_batch(&need_body)?; @@ -270,5 +273,7 @@ pub fn fill_missing_parent_ranges( } } } + let idx_ns = t_idx.elapsed().as_nanos() as u64; + crate::archive_phase_stats::note_fill_missing_ns(t_all.elapsed().as_nanos() as u64, idx_ns); Ok(()) }