#![allow(unused)]
fn main() {
// benches/pipeline.rs
// use criterion::{black_box, criterion_group, criterion_main, BenchmarkId, Criterion};
//
// fn bench_pipeline(c: &mut Criterion) {
//     let mut group = c.benchmark_group("pipeline");
//
//     for batch_size in [100, 500, 1000, 5000].iter() {
//         let headers = build_test_headers(*batch_size);
//
//         group.bench_with_input(
//             BenchmarkId::new("full", batch_size),
//             batch_size,
//             |b, _| {
//                 b.iter(|| {
//                     black_box(run_pipeline(black_box(&headers)))
//                 })
//             },
//         );
//     }
//     group.finish();
// }
//
// criterion_group!(benches, bench_pipeline);
// criterion_main!(benches);
}

#![allow(unused)]
fn main() {
// Reset counter, run stage, snapshot:
ALLOC_COUNT.store(0, Ordering::Relaxed);
let result = run_validator(black_box(&frames));
let validator_allocs = ALLOC_COUNT.load(Ordering::Relaxed);

ALLOC_COUNT.store(0, Ordering::Relaxed);
let deduped = run_deduplicator(black_box(&result));
let dedup_allocs = ALLOC_COUNT.load(Ordering::Relaxed);

println!("validator: {validator_allocs} allocs/batch");
println!("deduplicator: {dedup_allocs} allocs/batch");
}

If the deduplicator creates a new HashSet each batch, convert it to a persistent struct:

#![allow(unused)]
fn main() {
pub struct Deduplicator {
    seen: std::collections::HashSet<(u32, u64)>,
    unique_indices: Vec<usize>,
}

impl Deduplicator {
    pub fn new(expected_batch: usize) -> Self {
        Self {
            seen: std::collections::HashSet::with_capacity(expected_batch),
            unique_indices: Vec::with_capacity(expected_batch),
        }
    }

    pub fn process(&mut self, headers: &[(u32, u64)]) -> &[usize] {
        self.seen.clear();           // Retains allocation.
        self.unique_indices.clear(); // Retains allocation.
        for (i, &key) in headers.iter().enumerate() {
            if self.seen.insert(key) {
                self.unique_indices.push(i);
            }
        }
        &self.unique_indices
    }
}
}

Add to Cargo.toml:

[profile.release]
debug = true

[profile.profiling]
inherits = "release"
debug = true

Build and profile:

cargo build --profile profiling
cargo flamegraph --profile profiling --bin meridian-pipeline -- \
    --duration 30 --batch-size 1000

If cargo flamegraph is not installed: cargo install flamegraph. Requires perf on Linux or Xcode instruments on macOS.

// src/main.rs — pipeline implementation for profiling
use std::alloc::{GlobalAlloc, Layout, System};
use std::sync::atomic::{AtomicU64, Ordering::Relaxed};
use std::hint::black_box;
use std::time::Instant;

// --- Counting allocator ---

struct CountingAllocator;
static ALLOC_COUNT: AtomicU64 = AtomicU64::new(0);

unsafe impl GlobalAlloc for CountingAllocator {
    unsafe fn alloc(&self, layout: Layout) -> *mut u8 {
        ALLOC_COUNT.fetch_add(1, Relaxed);
        System.alloc(layout)
    }
    unsafe fn dealloc(&self, ptr: *mut u8, layout: Layout) {
        System.dealloc(ptr, layout)
    }
}
#[global_allocator]
static ALLOCATOR: CountingAllocator = CountingAllocator;

// --- Pipeline stages ---

#[inline(never)]
fn validate(headers: &[(u32, u64, u8)]) -> Vec<(u32, u64)> {
    headers.iter()
        .filter(|&&(_, _, flags)| flags & 0x80 == 0)
        .map(|&(sat, seq, _)| (sat, seq))
        .collect()
}

pub struct Deduplicator {
    seen:    std::collections::HashSet<(u32, u64)>,
    indices: Vec<usize>,
}

impl Deduplicator {
    pub fn new(cap: usize) -> Self {
        Self {
            seen:    std::collections::HashSet::with_capacity(cap),
            indices: Vec::with_capacity(cap),
        }
    }

    #[inline(never)]
    pub fn process(&mut self, valid: &[(u32, u64)]) -> &[usize] {
        self.seen.clear();
        self.indices.clear();
        for (i, &key) in valid.iter().enumerate() {
            if self.seen.insert(key) { self.indices.push(i); }
        }
        &self.indices
    }
}

#[inline(never)]
fn forward(valid: &[(u32, u64)], unique: &[usize]) -> usize {
    unique.iter().map(|&i| valid[i].0 as usize).sum()
}

fn run_pipeline(
    headers: &[(u32, u64, u8)],
    dedup: &mut Deduplicator,
) -> usize {
    let valid = validate(headers);
    let unique = dedup.process(&valid).to_vec();
    forward(&valid, &unique)
}

fn main() {
    let batch_size = 1_000usize;
    let headers: Vec<(u32, u64, u8)> = (0..batch_size)
        .map(|i| ((i % 48) as u32, (i / 3) as u64, 0u8))
        .collect();

    let mut dedup = Deduplicator::new(batch_size);

    // Warm up.
    for _ in 0..10 { run_pipeline(&headers, &mut dedup); }

    // Measure allocations per batch.
    ALLOC_COUNT.store(0, Relaxed);
    for _ in 0..1000 {
        black_box(run_pipeline(black_box(&headers), &mut dedup));
    }
    let allocs = ALLOC_COUNT.load(Relaxed);
    println!("allocs across 1000 batches: {allocs}");
    println!("allocs per batch: {:.1}", allocs as f64 / 1000.0);

    // Throughput measurement.
    let batches = 100_000u32;
    let start = Instant::now();
    for _ in 0..batches {
        black_box(run_pipeline(black_box(&headers), &mut dedup));
    }
    let elapsed = start.elapsed();
    let fps = (batches as usize * batch_size) as f64 / elapsed.as_secs_f64();
    println!("throughput: {:.0} frames/sec", fps);
    println!("elapsed: {:.2?}", elapsed);
}