20260802 delta wall优化：输出段全并行+归约尾段分片并行与打点（#139）

src/validation/run_tree_from_tal.rs

@@ -328,15 +328,18 @@ fn persist_transport_request_prefetch_snapshot(
     }
     let snapshot = runtime.transport_prefetch_snapshot();
     let recorded = snapshot.requests.len() as u64;
+    let persist_started = std::time::Instant::now();
     store
         .put_transport_prefetch_snapshot(&snapshot)
         .map_err(|err| TreeRunError::Runner(err.to_string()))?;
+    let persist_ms = persist_started.elapsed().as_millis() as u64;
     record_transport_prefetch_count(timing, "transport_prefetch_recorded_requests", recorded);
     crate::progress_log::emit(
         "phase1_repo_prefetch_persisted",
         serde_json::json!({
             "recorded_requests": recorded,
             "schema_version": snapshot.schema_version,
+            "persist_ms": persist_ms,
         }),
     );
     Ok(())

src/validation/tree_parallel.rs

@@ -2720,19 +2720,45 @@ fn is_complete(
         && staging_inflight == 0
 }
 
-fn build_tree_output(mut finished: Vec<FinishedPublicationPoint>) -> TreeRunAuditOutput {
-    finished.sort_by_key(|item| item.node.id);
-    let mut instances_processed = 0usize;
-    let mut instances_failed = 0usize;
-    let mut warnings = Vec::new();
-    let mut vrps = Vec::new();
-    let mut aspas = Vec::new();
-    let mut router_keys = Vec::new();
-    let mut publication_points = Vec::new();
-    let mut roa_cache_stats = crate::validation::objects::RoaValidationCacheStats::default();
-    let mut cir_input = CirInputAccumulator::default();
+/// Minimum publication points per reduction shard; smaller runs stay
+/// single-threaded to avoid thread-spawn overhead dominating the reduction.
+const TREE_OUTPUT_MIN_SHARD_LEN: usize = 4096;
+/// Hard cap on reduction shards regardless of core count.
+const TREE_OUTPUT_MAX_SHARDS: usize = 8;
 
-    for item in finished {
+fn tree_output_shard_count(len: usize) -> usize {
+    if len < TREE_OUTPUT_MIN_SHARD_LEN * 2 {
+        return 1;
+    }
+    let parallel = std::thread::available_parallelism()
+        .map(|n| n.get())
+        .unwrap_or(1);
+    parallel
+        .min(TREE_OUTPUT_MAX_SHARDS)
+        .min(len / TREE_OUTPUT_MIN_SHARD_LEN)
+        .max(1)
+}
+
+/// Per-shard reduction result of the phase2 output merge. Shards cover
+/// contiguous ranges of the id-sorted finished list, so concatenating shard
+/// outputs in shard order reproduces the sequential single-threaded order
+/// exactly.
+#[derive(Default)]
+struct TreeOutputShardReduction {
+    instances_processed: usize,
+    instances_failed: usize,
+    warnings: Vec<Warning>,
+    vrps: Vec<crate::validation::objects::Vrp>,
+    aspas: Vec<crate::validation::objects::AspaAttestation>,
+    router_keys: Vec<crate::validation::objects::RouterKeyPayload>,
+    publication_points: Vec<PublicationPointAudit>,
+    roa_cache_stats: crate::validation::objects::RoaValidationCacheStats,
+    cir_input: CirInputAccumulator,
+}
+
+fn reduce_finished_shard(items: Vec<FinishedPublicationPoint>) -> TreeOutputShardReduction {
+    let mut reduction = TreeOutputShardReduction::default();
+    for item in items {
         match item.result {
             FinishedPublicationPointResult::Ok {
                 source,
@@ -2742,51 +2768,136 @@ fn build_tree_output(mut finished: Vec<FinishedPublicationPoint>) -> TreeRunAudi
                 cir_fresh_objects,
                 cir_cached_objects,
             } => {
-                instances_processed += 1;
-                warnings.extend(result_warnings);
-                warnings.extend(objects.warnings);
-                roa_cache_stats.add_assign(&objects.roa_cache_stats);
-                vrps.extend(objects.vrps);
-                aspas.extend(objects.aspas);
-                router_keys.extend(objects.router_keys);
+                reduction.instances_processed += 1;
+                reduction.warnings.extend(result_warnings);
+                reduction.warnings.extend(objects.warnings);
+                reduction
+                    .roa_cache_stats
+                    .add_assign(&objects.roa_cache_stats);
+                reduction.vrps.extend(objects.vrps);
+                reduction.aspas.extend(objects.aspas);
+                reduction.router_keys.extend(objects.router_keys);
 
                 let mut audit: PublicationPointAudit = audit;
                 audit.node_id = Some(item.node.id);
                 audit.parent_node_id = item.node.parent_id;
                 audit.discovered_from = item.node.discovered_from;
                 crate::validation::tree::submit_publication_point_cir_input(
-                    &mut cir_input,
+                    &mut reduction.cir_input,
                     source,
                     &audit,
                     &cir_fresh_objects,
                     &cir_cached_objects,
                 )
                 .expect("CIR input collection from validated audit must not fail");
-                publication_points.push(audit);
+                reduction.publication_points.push(audit);
             }
             FinishedPublicationPointResult::Err(err) => {
-                instances_failed += 1;
-                warnings.push(
+                reduction.instances_failed += 1;
+                reduction.warnings.push(
                     Warning::new(format!("publication point failed: {err}"))
                         .with_context(&item.node.manifest_rsync_uri),
                 );
             }
         }
     }
+    reduction
+}
 
-    TreeRunAuditOutput {
-        tree: TreeRunOutput {
-            instances_processed,
-            instances_failed,
-            warnings,
-            vrps,
-            aspas,
-            router_keys,
-        },
-        publication_points,
-        roa_cache_stats,
-        cir_input: cir_input.finalize(),
+fn merge_shard_reductions(
+    reductions: Vec<TreeOutputShardReduction>,
+) -> TreeOutputShardReduction {
+    let mut iter = reductions.into_iter();
+    let mut merged = iter.next().unwrap_or_default();
+    for mut other in iter {
+        merged.instances_processed += other.instances_processed;
+        merged.instances_failed += other.instances_failed;
+        merged.warnings.append(&mut other.warnings);
+        merged.vrps.append(&mut other.vrps);
+        merged.aspas.append(&mut other.aspas);
+        merged.router_keys.append(&mut other.router_keys);
+        merged
+            .publication_points
+            .append(&mut other.publication_points);
+        merged.roa_cache_stats.add_assign(&other.roa_cache_stats);
+        merged
+            .cir_input
+            .merge(std::mem::take(&mut other.cir_input))
+            .expect("CIR input merge from validated audits must not fail");
     }
+    merged
+}
+
+fn build_tree_output(mut finished: Vec<FinishedPublicationPoint>) -> TreeRunAuditOutput {
+    let total_started = Instant::now();
+    finished.sort_by_key(|item| item.node.id);
+    let sort_ms = total_started.elapsed().as_millis() as u64;
+    let shard_count = tree_output_shard_count(finished.len());
+
+    let reduce_started = Instant::now();
+    let reductions = if shard_count <= 1 {
+        vec![reduce_finished_shard(finished)]
+    } else {
+        let chunk_len = finished.len().div_ceil(shard_count);
+        let mut shards: Vec<Vec<FinishedPublicationPoint>> = Vec::new();
+        let mut rest = finished;
+        while !rest.is_empty() {
+            let split_at = chunk_len.min(rest.len());
+            let tail = rest.split_off(split_at);
+            shards.push(std::mem::replace(&mut rest, tail));
+        }
+        std::thread::scope(|scope| {
+            let handles: Vec<_> = shards
+                .into_iter()
+                .map(|shard| scope.spawn(move || reduce_finished_shard(shard)))
+                .collect();
+            handles
+                .into_iter()
+                .map(|handle| {
+                    handle
+                        .join()
+                        .expect("tree output reduction shard panicked")
+                })
+                .collect()
+        })
+    };
+    let reduce_ms = reduce_started.elapsed().as_millis() as u64;
+
+    let merge_started = Instant::now();
+    let merged = merge_shard_reductions(reductions);
+    let merge_ms = merge_started.elapsed().as_millis() as u64;
+
+    let finalize_started = Instant::now();
+    let output = TreeRunAuditOutput {
+        tree: TreeRunOutput {
+            instances_processed: merged.instances_processed,
+            instances_failed: merged.instances_failed,
+            warnings: merged.warnings,
+            vrps: merged.vrps,
+            aspas: merged.aspas,
+            router_keys: merged.router_keys,
+        },
+        publication_points: merged.publication_points,
+        roa_cache_stats: merged.roa_cache_stats,
+        cir_input: merged.cir_input.finalize(),
+    };
+    let finalize_ms = finalize_started.elapsed().as_millis() as u64;
+
+    crate::progress_log::emit(
+        "phase2_build_tree_output",
+        serde_json::json!({
+            "sort_ms": sort_ms,
+            "shard_count": shard_count,
+            "reduce_ms": reduce_ms,
+            "merge_ms": merge_ms,
+            "finalize_ms": finalize_ms,
+            "total_ms": total_started.elapsed().as_millis() as u64,
+            "publication_points": output.publication_points.len(),
+            "instances_processed": output.tree.instances_processed,
+            "instances_failed": output.tree.instances_failed,
+        }),
+    );
+    output
 }
 
 pub fn run_tree_parallel_phase2_audit(
@@ -2800,12 +2911,14 @@ pub fn run_tree_parallel_phase2_audit(
 #[cfg(test)]
 mod tests {
     use super::{
-        CacheHitOutcome, CompleteOutcome, FinishedPublicationPointResult, InflightPublicationPoint,
-        QueuedCaInstance, ReadyCaInstance, ReadyStageMetrics, ReadyStageSubmitter, StageOutcome,
-        apply_ready_publication_point_stage, compact_phase2_finished_result,
-        compact_phase2_finished_result_result, compute_ready_publication_point_stage,
-        event_poll_timeout, finalize_metrics_from_output, finalize_publication_point_state,
-        is_complete, submit_ready_batch_to_stage_pool,
+        CacheHitOutcome, CompleteOutcome, FinishedPublicationPoint, FinishedPublicationPointNode,
+        FinishedPublicationPointResult, InflightPublicationPoint, QueuedCaInstance,
+        ReadyCaInstance, ReadyStageMetrics, ReadyStageSubmitter, StageOutcome,
+        TREE_OUTPUT_MAX_SHARDS, TREE_OUTPUT_MIN_SHARD_LEN, apply_ready_publication_point_stage,
+        build_tree_output, compact_phase2_finished_result, compact_phase2_finished_result_result,
+        compute_ready_publication_point_stage, event_poll_timeout, finalize_metrics_from_output,
+        finalize_publication_point_state, is_complete, merge_shard_reductions,
+        reduce_finished_shard, submit_ready_batch_to_stage_pool, tree_output_shard_count,
     };
     use crate::audit::{
         AuditObjectKind, AuditObjectResult, DiscoveredFrom, ObjectAuditEntry,
@@ -2935,6 +3048,135 @@ mod tests {
         }
     }
 
+    fn finished_ok_item(id: u64) -> FinishedPublicationPoint {
+        let mut result = sample_result();
+        result
+            .warnings
+            .push(crate::report::Warning::new(format!("result-warning-{id:05}")));
+        result
+            .objects
+            .warnings
+            .push(crate::report::Warning::new(format!("objects-warning-{id:05}")));
+        result.objects.vrps.push(crate::validation::objects::Vrp {
+            asn: 64496 + id as u32,
+            prefix: crate::data_model::roa::IpPrefix {
+                afi: crate::data_model::roa::RoaAfi::Ipv4,
+                addr: [192, 0, 2, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0],
+                prefix_len: 24,
+            },
+            max_length: 24,
+        });
+        result.audit.objects.push(crate::audit::ObjectAuditEntry {
+            rsync_uri: format!("rsync://example.test/repo/{id:05}.roa"),
+            sha256_hex: format!("{id:064x}"),
+            kind: crate::audit::AuditObjectKind::Roa,
+            result: crate::audit::AuditObjectResult::Ok,
+            detail: None,
+        });
+        FinishedPublicationPoint {
+            node: FinishedPublicationPointNode {
+                id,
+                parent_id: None,
+                discovered_from: None,
+                manifest_rsync_uri: format!("rsync://example.test/repo/{id:05}.mft"),
+            },
+            result: compact_phase2_finished_result(result, false),
+        }
+    }
+
+    fn finished_err_item(id: u64) -> FinishedPublicationPoint {
+        FinishedPublicationPoint {
+            node: FinishedPublicationPointNode {
+                id,
+                parent_id: None,
+                discovered_from: None,
+                manifest_rsync_uri: format!("rsync://example.test/repo/{id:05}.mft"),
+            },
+            result: FinishedPublicationPointResult::Err(format!("boom-{id:05}")),
+        }
+    }
+
+    fn mixed_finished_items(n: u64) -> Vec<FinishedPublicationPoint> {
+        (0..n)
+            .map(|id| {
+                if id % 97 == 0 {
+                    finished_err_item(id)
+                } else {
+                    finished_ok_item(id)
+                }
+            })
+            .collect()
+    }
+
+    #[test]
+    fn tree_output_shard_count_respects_thresholds() {
+        assert_eq!(tree_output_shard_count(0), 1);
+        assert_eq!(tree_output_shard_count(TREE_OUTPUT_MIN_SHARD_LEN * 2 - 1), 1);
+        assert!(tree_output_shard_count(TREE_OUTPUT_MIN_SHARD_LEN * 2) >= 1);
+        let parallel = std::thread::available_parallelism()
+            .map(|n| n.get())
+            .unwrap_or(1);
+        assert_eq!(
+            tree_output_shard_count(1_000_000),
+            parallel.min(TREE_OUTPUT_MAX_SHARDS)
+        );
+    }
+
+    #[test]
+    fn tree_output_sharded_merge_matches_single_reduction() {
+        let single = merge_shard_reductions(vec![reduce_finished_shard(mixed_finished_items(300))]);
+
+        let mut chunks: Vec<Vec<FinishedPublicationPoint>> = Vec::new();
+        let mut rest = mixed_finished_items(300);
+        while !rest.is_empty() {
+            let split_at = 100.min(rest.len());
+            let tail = rest.split_off(split_at);
+            chunks.push(std::mem::replace(&mut rest, tail));
+        }
+        let merged = merge_shard_reductions(
+            chunks.into_iter().map(reduce_finished_shard).collect(),
+        );
+
+        assert_eq!(single.instances_processed, merged.instances_processed);
+        assert_eq!(single.instances_failed, merged.instances_failed);
+        assert_eq!(format!("{:?}", single.warnings), format!("{:?}", merged.warnings));
+        assert_eq!(format!("{:?}", single.vrps), format!("{:?}", merged.vrps));
+        assert_eq!(
+            format!("{:?}", single.publication_points),
+            format!("{:?}", merged.publication_points)
+        );
+        assert_eq!(
+            format!("{:?}", single.roa_cache_stats),
+            format!("{:?}", merged.roa_cache_stats)
+        );
+        assert_eq!(single.cir_input.finalize(), merged.cir_input.finalize());
+    }
+
+    #[test]
+    fn build_tree_output_orders_results_by_node_id() {
+        let mut items = mixed_finished_items(50);
+        items.reverse();
+        let output = build_tree_output(items);
+        let got: Vec<u64> = output
+            .publication_points
+            .iter()
+            .map(|pp| pp.node_id.expect("node id set"))
+            .collect();
+        let want: Vec<u64> = (1..50).filter(|id| id % 97 != 0).collect();
+        assert_eq!(got, want);
+        assert_eq!(output.tree.instances_failed, 1);
+        let first_warning = format!("{:?}", output.tree.warnings[0]);
+        assert!(
+            first_warning.contains("publication point failed: boom-00000"),
+            "failed publication point warning keeps id order: {first_warning}"
+        );
+        assert_eq!(output.cir_input.fresh_validated_objects.len(), 49);
+        assert_eq!(
+            output.cir_input.fresh_validated_objects[0].rsync_uri,
+            "rsync://example.test/repo/00001.roa"
+        );
+    }
+
     #[test]
     fn finalize_metrics_from_output_captures_breakdown_and_counts() {
         let mut result = sample_result();