datum-core 0.10.4

use std::{
    alloc::{GlobalAlloc, Layout, System},
    hint::black_box,
    sync::{
        OnceLock,
        atomic::{AtomicU64, Ordering},
    },
    time::Instant,
};

use datum::{
    AsyncBoundary, AsyncBoundaryExecutionConfig, Attributes, Balance, BidiFlow, Broadcast, Buffer,
    Concat, Flow, FusedExecutionConfig, Fusion, GraphBlueprint, GraphDsl, GraphFlowShape, Identity,
    Interleave, KernelBackend, MapStage, Merge, MergeLatest, MergePreferred, MergePrioritized,
    MergeSequence, MergeSorted, MetricsLevel, OverflowStrategy, Partition, Sink, Source, TakeWhile,
    Unzip, UnzipWith, Zip,
};

struct CountingAllocator;

static ALLOCATED_BYTES: AtomicU64 = AtomicU64::new(0);

unsafe impl GlobalAlloc for CountingAllocator {
    unsafe fn alloc(&self, layout: Layout) -> *mut u8 {
        let ptr = unsafe { System.alloc(layout) };
        if !ptr.is_null() {
            ALLOCATED_BYTES.fetch_add(layout.size() as u64, Ordering::Relaxed);
        }
        ptr
    }

    unsafe fn alloc_zeroed(&self, layout: Layout) -> *mut u8 {
        let ptr = unsafe { System.alloc_zeroed(layout) };
        if !ptr.is_null() {
            ALLOCATED_BYTES.fetch_add(layout.size() as u64, Ordering::Relaxed);
        }
        ptr
    }

    unsafe fn dealloc(&self, ptr: *mut u8, layout: Layout) {
        unsafe { System.dealloc(ptr, layout) };
    }

    unsafe fn realloc(&self, ptr: *mut u8, layout: Layout, new_size: usize) -> *mut u8 {
        let new_ptr = unsafe { System.realloc(ptr, layout, new_size) };
        if !new_ptr.is_null() {
            if new_ptr == ptr {
                // Grown in place: only the additional bytes are newly allocated.
                if new_size > layout.size() {
                    ALLOCATED_BYTES.fetch_add((new_size - layout.size()) as u64, Ordering::Relaxed);
                }
            } else {
                ALLOCATED_BYTES.fetch_add(new_size as u64, Ordering::Relaxed);
            }
        }
        new_ptr
    }
}

#[global_allocator]
static GLOBAL: CountingAllocator = CountingAllocator;

/// Process-wide CPU time (user + system, all threads) in nanoseconds from
/// `/proc/self/stat` utime+stime (USER_HZ = 100 on Linux, 10 ms/tick). Coarse
/// per op but meaningful over a measured loop; captures CPU used by stream
/// worker threads (including any busy-spin) that wall-clock latency omits.
fn process_cpu_ns() -> u128 {
    let Ok(stat) = std::fs::read_to_string("/proc/self/stat") else {
        return 0;
    };
    let Some(close) = stat.rfind(')') else {
        return 0;
    };
    let fields: Vec<&str> = stat[close + 1..].split_whitespace().collect();
    if fields.len() <= 12 {
        return 0;
    }
    let utime: u64 = fields[11].parse().unwrap_or(0);
    let stime: u64 = fields[12].parse().unwrap_or(0);
    (utime as u128 + stime as u128) * 10_000_000
}

struct Scenario {
    name: &'static str,
    iterations: u64,
    run: fn() -> u64,
}

type PairMergeLatestGraph = GraphBlueprint<GraphFlowShape<(u64, u64), Vec<u64>>>;

fn main() {
    let scenarios = [
        Scenario {
            name: "fused_identity_10_100k",
            iterations: 25,
            run: fused_identity_10_100k,
        },
        Scenario {
            name: "fused_map_10_100k",
            iterations: 20,
            run: fused_map_10_100k,
        },
        Scenario {
            name: "fused_identity_10_100k_count_sink",
            iterations: 100,
            run: fused_identity_10_100k_count_sink,
        },
        Scenario {
            name: "fused_map_10_100k_fold_sink",
            iterations: 80,
            run: fused_map_10_100k_fold_sink,
        },
        Scenario {
            name: "fused_identity_10_100k_typed_count_sink",
            iterations: 1_000,
            run: fused_identity_10_100k_typed_count_sink,
        },
        Scenario {
            name: "fused_map_10_100k_typed_fold_sink",
            iterations: 1_000,
            run: fused_map_10_100k_typed_fold_sink,
        },
        Scenario {
            name: "scalar_graph_add_sub_10_100k_closure",
            iterations: 100,
            run: scalar_graph_add_sub_10_100k_closure,
        },
        Scenario {
            name: "scalar_graph_add_sub_10_100k_chunked",
            iterations: 100,
            run: scalar_graph_add_sub_10_100k_chunked,
        },
        Scenario {
            name: "scalar_graph_add_sub_10_1m_closure",
            iterations: 30,
            run: scalar_graph_add_sub_10_1m_closure,
        },
        Scenario {
            name: "scalar_graph_add_sub_10_1m_chunked",
            iterations: 30,
            run: scalar_graph_add_sub_10_1m_chunked,
        },
        Scenario {
            name: "fused_identity_10_100k_typed_count_sink_consuming_builder",
            iterations: 1_000,
            run: fused_identity_10_100k_typed_count_sink_consuming_builder,
        },
        Scenario {
            name: "fused_map_10_100k_typed_fold_sink_consuming_builder",
            iterations: 1_000,
            run: fused_map_10_100k_typed_fold_sink_consuming_builder,
        },
        Scenario {
            name: "fused_identity_10_100k_typed_count_sink_named_attr",
            iterations: 1_000,
            run: fused_identity_10_100k_typed_count_sink_named_attr,
        },
        Scenario {
            name: "fused_identity_10_100k_typed_count_sink_buffer_attr",
            iterations: 1_000,
            run: fused_identity_10_100k_typed_count_sink_buffer_attr,
        },
        Scenario {
            name: "fused_identity_10_100k_count_sink_report_default_attrs",
            iterations: 10_000,
            run: fused_identity_10_100k_count_sink_report_default_attrs,
        },
        Scenario {
            name: "fused_identity_10_100k_count_sink_full_noop_attrs",
            iterations: 10_000,
            run: fused_identity_10_100k_count_sink_full_noop_attrs,
        },
        Scenario {
            name: "metrics_linear_500k_off",
            iterations: 500,
            run: metrics_linear_500k_off,
        },
        Scenario {
            name: "metrics_linear_500k_counts",
            iterations: 500,
            run: metrics_linear_500k_counts,
        },
        Scenario {
            name: "metrics_linear_500k_timing",
            iterations: 200,
            run: metrics_linear_500k_timing,
        },
        Scenario {
            name: "metrics_linear_500k_stalls",
            iterations: 200,
            run: metrics_linear_500k_stalls,
        },
        Scenario {
            name: "fused_identity_10_100k_count_sink_forced_erased",
            iterations: 100,
            run: fused_identity_10_100k_count_sink_forced_erased,
        },
        Scenario {
            name: "async_boundary_1_10k_count_sink",
            iterations: 100,
            run: async_boundary_1_10k_count_sink,
        },
        Scenario {
            name: "async_boundary_9_10k_count_sink",
            iterations: 100,
            run: async_boundary_9_10k_count_sink,
        },
        Scenario {
            name: "graph_build_identity_1",
            iterations: 200_000,
            run: graph_build_identity_1,
        },
        Scenario {
            name: "graph_build_identity_10",
            iterations: 80_000,
            run: graph_build_identity_10,
        },
        Scenario {
            name: "graph_build_identity_10_consuming_builder",
            iterations: 80_000,
            run: graph_build_identity_10_consuming_builder,
        },
        Scenario {
            name: "graph_build_identity_100",
            iterations: 10_000,
            run: graph_build_identity_100,
        },
        Scenario {
            name: "graph_build_broadcast_zip_current_builder",
            iterations: 120_000,
            run: graph_build_broadcast_zip_current_builder,
        },
        Scenario {
            name: "graph_build_broadcast_zip_consuming_builder",
            iterations: 120_000,
            run: graph_build_broadcast_zip_consuming_builder,
        },
        Scenario {
            name: "graph_build_balance_merge_current_builder",
            iterations: 120_000,
            run: graph_build_balance_merge_current_builder,
        },
        Scenario {
            name: "graph_build_balance_merge_consuming_builder",
            iterations: 120_000,
            run: graph_build_balance_merge_consuming_builder,
        },
        Scenario {
            name: "graph_build_junction_1",
            iterations: 120_000,
            run: graph_build_junction_1,
        },
        Scenario {
            name: "graph_build_junction_10",
            iterations: 30_000,
            run: graph_build_junction_10,
        },
        Scenario {
            name: "graph_build_junction_50",
            iterations: 5_000,
            run: graph_build_junction_50,
        },
        Scenario {
            name: "junction_broadcast_zip_10k",
            iterations: 2_000,
            run: junction_broadcast_zip_10k,
        },
        Scenario {
            name: "junction_broadcast_zip_10k_consuming_builder",
            iterations: 2_000,
            run: junction_broadcast_zip_10k_consuming_builder,
        },
        Scenario {
            name: "junction_broadcast_zip_10k_report_default_attrs",
            iterations: 100_000,
            run: junction_broadcast_zip_10k_report_default_attrs,
        },
        Scenario {
            name: "junction_broadcast_zip_10k_full_noop_attrs",
            iterations: 100_000,
            run: junction_broadcast_zip_10k_full_noop_attrs,
        },
        Scenario {
            name: "metrics_broadcast_zip_32mib_off",
            iterations: 100,
            run: metrics_broadcast_zip_32mib_off,
        },
        Scenario {
            name: "metrics_broadcast_zip_32mib_counts",
            iterations: 100,
            run: metrics_broadcast_zip_32mib_counts,
        },
        Scenario {
            name: "metrics_broadcast_zip_32mib_timing",
            iterations: 100,
            run: metrics_broadcast_zip_32mib_timing,
        },
        Scenario {
            name: "metrics_broadcast_zip_32mib_stalls",
            iterations: 100,
            run: metrics_broadcast_zip_32mib_stalls,
        },
        Scenario {
            name: "junction_broadcast_zip_10k_forced_erased",
            iterations: 500,
            run: junction_broadcast_zip_10k_forced_erased,
        },
        Scenario {
            name: "junction_balance_merge_10k",
            iterations: 2_000,
            run: junction_balance_merge_10k,
        },
        Scenario {
            name: "junction_balance_merge_10k_consuming_builder",
            iterations: 2_000,
            run: junction_balance_merge_10k_consuming_builder,
        },
        Scenario {
            name: "junction_balance_merge_10k_report_default_attrs",
            iterations: 100_000,
            run: junction_balance_merge_10k_report_default_attrs,
        },
        Scenario {
            name: "junction_balance_merge_10k_full_noop_attrs",
            iterations: 100_000,
            run: junction_balance_merge_10k_full_noop_attrs,
        },
        Scenario {
            name: "junction_prioritized_merge_10k",
            iterations: 1_000,
            run: junction_prioritized_merge_10k,
        },
        Scenario {
            name: "junction_merge_preferred_10k",
            iterations: 1_000,
            run: junction_merge_preferred_10k,
        },
        Scenario {
            name: "junction_concat_10k",
            iterations: 1_000,
            run: junction_concat_10k,
        },
        Scenario {
            name: "junction_interleave_10k",
            iterations: 1_000,
            run: junction_interleave_10k,
        },
        Scenario {
            name: "junction_merge_sorted_10k",
            iterations: 1_000,
            run: junction_merge_sorted_10k,
        },
        Scenario {
            name: "junction_merge_sequence_10k",
            iterations: 500,
            run: junction_merge_sequence_10k,
        },
        Scenario {
            name: "junction_merge_latest_10k",
            iterations: 1_000,
            run: junction_merge_latest_10k,
        },
        Scenario {
            name: "junction_partition_10k",
            iterations: 1_000,
            run: junction_partition_10k,
        },
        Scenario {
            name: "junction_unzip_with_10k",
            iterations: 1_000,
            run: junction_unzip_with_10k,
        },
        Scenario {
            name: "cycle_merge_preferred_feedback_10k",
            iterations: 500,
            run: cycle_merge_preferred_feedback_10k,
        },
        Scenario {
            name: "custom_identity_stage_10k",
            iterations: 1_000,
            run: custom_identity_stage_10k,
        },
        Scenario {
            name: "custom_map_stage_10k",
            iterations: 1_000,
            run: custom_map_stage_10k,
        },
        Scenario {
            name: "bidi_join_10k",
            iterations: 500,
            run: bidi_join_10k,
        },
        Scenario {
            name: "bidi_atop_10k",
            iterations: 500,
            run: bidi_atop_10k,
        },
    ];

    println!("scenario\titerations\tns_per_op\tallocated_bytes_per_op\tcpu_ns_per_op");
    let scenario_filter = std::env::var("DATUM_GRAPH_SCENARIO_FILTER").ok();
    for scenario in scenarios {
        if scenario_filter.as_deref().is_some_and(|filter| {
            filter.strip_prefix('=').map_or_else(
                || !scenario.name.contains(filter),
                |exact| scenario.name != exact,
            )
        }) {
            continue;
        }
        for _ in 0..5 {
            black_box((scenario.run)());
        }

        let cpu_start = process_cpu_ns();
        ALLOCATED_BYTES.store(0, Ordering::Relaxed);
        let started = Instant::now();
        let mut checksum = 0_u64;
        for _ in 0..scenario.iterations {
            checksum = checksum.wrapping_add(black_box((scenario.run)()));
        }
        let elapsed = started.elapsed();
        // process_cpu_ns allocates, so read the allocation counter before the
        // final CPU sample to keep CPU sampling out of the measured window.
        let allocated_bytes = ALLOCATED_BYTES.load(Ordering::Relaxed);
        let cpu_delta = process_cpu_ns().saturating_sub(cpu_start);
        black_box(checksum);

        let ns_per_op = elapsed.as_nanos() as f64 / scenario.iterations as f64;
        let allocated_bytes_per_op = allocated_bytes as f64 / scenario.iterations as f64;
        let cpu_ns_per_op = cpu_delta as f64 / scenario.iterations as f64;
        println!(
            "{}\t{}\t{ns_per_op:.2}\t{allocated_bytes_per_op:.2}\t{cpu_ns_per_op:.2}",
            scenario.name, scenario.iterations
        );
    }
}

fn identity_chain(stages: usize) -> GraphBlueprint<GraphFlowShape<u64, u64>> {
    assert!(stages > 0);
    GraphDsl::try_create(|builder| {
        let first = builder.add(Identity::<u64>::new());
        let inlet = first.inlet();
        let mut outlet = first.outlet();

        for _ in 1..stages {
            let next = builder.add(Identity::<u64>::new());
            builder.connect(outlet, next.inlet())?;
            outlet = next.outlet();
        }

        Ok(GraphFlowShape::new(inlet, outlet))
    })
    .unwrap()
}

fn map_chain(stages: usize) -> GraphBlueprint<GraphFlowShape<u64, u64>> {
    assert!(stages > 0);
    GraphDsl::try_create(|builder| {
        let first = builder.add(MapStage::new(|item: u64| item.wrapping_add(1)));
        let inlet = first.inlet();
        let mut outlet = first.outlet();

        for _ in 1..stages {
            let next = builder.add(MapStage::new(|item: u64| item.wrapping_add(1)));
            builder.connect(outlet, next.inlet())?;
            outlet = next.outlet();
        }

        Ok(GraphFlowShape::new(inlet, outlet))
    })
    .unwrap()
}

fn typed_identity_chain(stages: usize) -> GraphBlueprint<GraphFlowShape<u64, u64>> {
    assert!(stages > 0);
    GraphDsl::typed_try_create(|builder| {
        let (mut builder, inlet, mut outlet) = builder.add_flow(Identity::<u64>::new());

        for _ in 1..stages {
            let (next_builder, next_inlet, next_outlet) = builder.add_flow(Identity::<u64>::new());
            builder = next_builder.connect(outlet, next_inlet)?;
            outlet = next_outlet;
        }

        Ok((
            builder,
            GraphFlowShape::new(inlet.into_port(), outlet.into_port()),
        ))
    })
    .unwrap()
}

fn typed_map_chain(stages: usize) -> GraphBlueprint<GraphFlowShape<u64, u64>> {
    assert!(stages > 0);
    GraphDsl::typed_try_create(|builder| {
        let (mut builder, inlet, mut outlet) =
            builder.add_flow(MapStage::new(|item: u64| item.wrapping_add(1)));

        for _ in 1..stages {
            let (next_builder, next_inlet, next_outlet) =
                builder.add_flow(MapStage::new(|item: u64| item.wrapping_add(1)));
            builder = next_builder.connect(outlet, next_inlet)?;
            outlet = next_outlet;
        }

        Ok((
            builder,
            GraphFlowShape::new(inlet.into_port(), outlet.into_port()),
        ))
    })
    .unwrap()
}

fn scalar_closure_chain(stages: usize) -> GraphBlueprint<GraphFlowShape<i64, i64>> {
    assert!(stages > 0);
    GraphDsl::try_create(|builder| {
        let first = builder.add(MapStage::new(|value: i64| value.wrapping_add(1)));
        let inlet = first.inlet();
        let mut outlet = first.outlet();
        for stage in 1..stages {
            let next = if stage % 2 == 0 {
                builder.add(MapStage::new(|value: i64| value.wrapping_add(1)))
            } else {
                builder.add(MapStage::new(|value: i64| value.wrapping_sub(1)))
            };
            builder.connect(outlet, next.inlet())?;
            outlet = next.outlet();
        }
        Ok(GraphFlowShape::new(inlet, outlet))
    })
    .unwrap()
}

fn scalar_builtin_chain(stages: usize) -> GraphBlueprint<GraphFlowShape<i64, i64>> {
    assert!(stages > 0);
    GraphDsl::try_create(|builder| {
        let first = builder.add(MapStage::map_add(1));
        let inlet = first.inlet();
        let mut outlet = first.outlet();
        for stage in 1..stages {
            let next = if stage % 2 == 0 {
                builder.add(MapStage::map_add(1))
            } else {
                builder.add(MapStage::map_subtract(1))
            };
            builder.connect(outlet, next.inlet())?;
            outlet = next.outlet();
        }
        Ok(GraphFlowShape::new(inlet, outlet))
    })
    .unwrap()
}

fn named_identity_chain(stages: usize) -> GraphBlueprint<GraphFlowShape<u64, u64>> {
    identity_chain(stages).named("bench.identity")
}

fn buffered_identity_chain(stages: usize) -> GraphBlueprint<GraphFlowShape<u64, u64>> {
    identity_chain(stages).add_attributes(datum::Attributes::input_buffer(16, 16))
}

fn full_noop_attributes() -> Attributes {
    Attributes::named("bench.full-noop")
        .and(Attributes::input_buffer(16, 16))
        .and(Attributes::dispatcher("bench-dispatcher"))
        .and(Attributes::batch_size(256))
        .and(Attributes::parallelism(1))
        .and(Attributes::fusion(Fusion::Auto))
        .and(Attributes::metrics_level(MetricsLevel::Off))
        .and(Attributes::kernel_backend(KernelBackend::ArrowRust))
}

fn full_noop_attr_identity_chain(stages: usize) -> GraphBlueprint<GraphFlowShape<u64, u64>> {
    identity_chain(stages).add_attributes(full_noop_attributes())
}

fn forced_erased_identity_chain(stages: usize) -> GraphBlueprint<GraphFlowShape<u64, u64>> {
    identity_chain(stages).add_attributes(Attributes::fusion(Fusion::ErasedOnly))
}

fn identity_chain_with_async_boundaries(
    stages: usize,
    boundary_positions: &[usize],
) -> GraphBlueprint<GraphFlowShape<u64, u64>> {
    assert!(stages > 0);
    GraphDsl::try_create(|builder| {
        let first = builder.add(Identity::<u64>::new());
        let inlet = first.inlet();
        let mut outlet = first.outlet();

        for stage_index in 1..stages {
            if boundary_positions.contains(&stage_index) {
                let boundary = builder.add(AsyncBoundary::<u64>::new());
                builder.connect(outlet, boundary.inlet())?;
                outlet = boundary.outlet();
            }

            let next = builder.add(Identity::<u64>::new());
            builder.connect(outlet, next.inlet())?;
            outlet = next.outlet();
        }

        Ok(GraphFlowShape::new(inlet, outlet))
    })
    .unwrap()
}

fn junction_chain(pairs: usize) -> GraphBlueprint<GraphFlowShape<u64, u64>> {
    assert!(pairs > 0);
    GraphDsl::try_create(|builder| {
        let first = builder.add(Identity::<u64>::new());
        let inlet = first.inlet();
        let mut outlet = first.outlet();

        for _ in 0..pairs {
            let balance = builder.add(Balance::<u64>::new(2));
            let merge = builder.add(Merge::<u64>::new(2));

            builder.connect(outlet, balance.inlet())?;
            builder.connect(balance.outlet(0)?, merge.inlet(0)?)?;
            builder.connect(balance.outlet(1)?, merge.inlet(1)?)?;
            outlet = merge.outlet();
        }

        Ok(GraphFlowShape::new(inlet, outlet))
    })
    .unwrap()
}

fn broadcast_zip_graph() -> GraphBlueprint<GraphFlowShape<u64, (u64, u64)>> {
    GraphDsl::try_create(|builder| {
        let broadcast = builder.add(Broadcast::<u64>::new(2));
        let zip = builder.add(Zip::<u64, u64>::new());

        builder.connect(broadcast.outlet(0)?, zip.in0())?;
        builder.connect(broadcast.outlet(1)?, zip.in1())?;

        Ok(GraphFlowShape::new(broadcast.inlet(), zip.outlet()))
    })
    .unwrap()
}

fn broadcast_zip_graph_full_noop_attrs() -> GraphBlueprint<GraphFlowShape<u64, (u64, u64)>> {
    broadcast_zip_graph().add_attributes(full_noop_attributes())
}

fn broadcast_zip_graph_forced_erased() -> GraphBlueprint<GraphFlowShape<u64, (u64, u64)>> {
    broadcast_zip_graph().add_attributes(Attributes::fusion(Fusion::ErasedOnly))
}

fn balance_merge_graph() -> GraphBlueprint<GraphFlowShape<u64, u64>> {
    GraphDsl::try_create(|builder| {
        let balance = builder.add(Balance::<u64>::new(2));
        let merge = builder.add(Merge::<u64>::new(2));

        builder.connect(balance.outlet(0)?, merge.inlet(0)?)?;
        builder.connect(balance.outlet(1)?, merge.inlet(1)?)?;

        Ok(GraphFlowShape::new(balance.inlet(), merge.outlet()))
    })
    .unwrap()
}

fn balance_merge_graph_full_noop_attrs() -> GraphBlueprint<GraphFlowShape<u64, u64>> {
    balance_merge_graph().add_attributes(full_noop_attributes())
}

fn typed_broadcast_zip_graph() -> GraphBlueprint<GraphFlowShape<u64, (u64, u64)>> {
    GraphDsl::typed_try_create(|builder| {
        let (builder, inlet, [out0, out1]) = builder.add_broadcast2::<u64>();
        let (builder, in0, in1, outlet) = builder.add_zip::<u64, u64>();

        let builder = builder.connect(out0, in0)?;
        let builder = builder.connect(out1, in1)?;

        Ok((
            builder,
            GraphFlowShape::new(inlet.into_port(), outlet.into_port()),
        ))
    })
    .unwrap()
}

fn typed_balance_merge_graph() -> GraphBlueprint<GraphFlowShape<u64, u64>> {
    GraphDsl::typed_try_create(|builder| {
        let (builder, inlet, [out0, out1]) = builder.add_balance2::<u64>();
        let (builder, [in0, in1], outlet) = builder.add_merge2::<u64>();

        let builder = builder.connect(out0, in0)?;
        let builder = builder.connect(out1, in1)?;

        Ok((
            builder,
            GraphFlowShape::new(inlet.into_port(), outlet.into_port()),
        ))
    })
    .unwrap()
}

fn identity_chain_10() -> &'static GraphBlueprint<GraphFlowShape<u64, u64>> {
    static GRAPH: OnceLock<GraphBlueprint<GraphFlowShape<u64, u64>>> = OnceLock::new();
    GRAPH.get_or_init(|| identity_chain(10))
}

fn map_chain_10() -> &'static GraphBlueprint<GraphFlowShape<u64, u64>> {
    static GRAPH: OnceLock<GraphBlueprint<GraphFlowShape<u64, u64>>> = OnceLock::new();
    GRAPH.get_or_init(|| map_chain(10))
}

fn typed_identity_chain_10() -> &'static GraphBlueprint<GraphFlowShape<u64, u64>> {
    static GRAPH: OnceLock<GraphBlueprint<GraphFlowShape<u64, u64>>> = OnceLock::new();
    GRAPH.get_or_init(|| typed_identity_chain(10))
}

fn typed_map_chain_10() -> &'static GraphBlueprint<GraphFlowShape<u64, u64>> {
    static GRAPH: OnceLock<GraphBlueprint<GraphFlowShape<u64, u64>>> = OnceLock::new();
    GRAPH.get_or_init(|| typed_map_chain(10))
}

fn scalar_closure_chain_10() -> &'static GraphBlueprint<GraphFlowShape<i64, i64>> {
    static GRAPH: OnceLock<GraphBlueprint<GraphFlowShape<i64, i64>>> = OnceLock::new();
    GRAPH.get_or_init(|| scalar_closure_chain(10))
}

fn scalar_builtin_chain_10() -> &'static GraphBlueprint<GraphFlowShape<i64, i64>> {
    static GRAPH: OnceLock<GraphBlueprint<GraphFlowShape<i64, i64>>> = OnceLock::new();
    GRAPH.get_or_init(|| scalar_builtin_chain(10))
}

fn named_identity_chain_10() -> &'static GraphBlueprint<GraphFlowShape<u64, u64>> {
    static GRAPH: OnceLock<GraphBlueprint<GraphFlowShape<u64, u64>>> = OnceLock::new();
    GRAPH.get_or_init(|| named_identity_chain(10))
}

fn buffered_identity_chain_10() -> &'static GraphBlueprint<GraphFlowShape<u64, u64>> {
    static GRAPH: OnceLock<GraphBlueprint<GraphFlowShape<u64, u64>>> = OnceLock::new();
    GRAPH.get_or_init(|| buffered_identity_chain(10))
}

fn full_noop_attr_identity_chain_10() -> &'static GraphBlueprint<GraphFlowShape<u64, u64>> {
    static GRAPH: OnceLock<GraphBlueprint<GraphFlowShape<u64, u64>>> = OnceLock::new();
    GRAPH.get_or_init(|| full_noop_attr_identity_chain(10))
}

fn forced_erased_identity_chain_10() -> &'static GraphBlueprint<GraphFlowShape<u64, u64>> {
    static GRAPH: OnceLock<GraphBlueprint<GraphFlowShape<u64, u64>>> = OnceLock::new();
    GRAPH.get_or_init(|| forced_erased_identity_chain(10))
}

fn async_boundary_1() -> &'static GraphBlueprint<GraphFlowShape<u64, u64>> {
    static GRAPH: OnceLock<GraphBlueprint<GraphFlowShape<u64, u64>>> = OnceLock::new();
    GRAPH.get_or_init(|| identity_chain_with_async_boundaries(10, &[5]))
}

fn async_boundary_9() -> &'static GraphBlueprint<GraphFlowShape<u64, u64>> {
    static GRAPH: OnceLock<GraphBlueprint<GraphFlowShape<u64, u64>>> = OnceLock::new();
    GRAPH.get_or_init(|| {
        let boundary_positions = (1_usize..10).collect::<Vec<_>>();
        identity_chain_with_async_boundaries(10, &boundary_positions)
    })
}

fn broadcast_zip() -> &'static GraphBlueprint<GraphFlowShape<u64, (u64, u64)>> {
    static GRAPH: OnceLock<GraphBlueprint<GraphFlowShape<u64, (u64, u64)>>> = OnceLock::new();
    GRAPH.get_or_init(broadcast_zip_graph)
}

fn balance_merge() -> &'static GraphBlueprint<GraphFlowShape<u64, u64>> {
    static GRAPH: OnceLock<GraphBlueprint<GraphFlowShape<u64, u64>>> = OnceLock::new();
    GRAPH.get_or_init(balance_merge_graph)
}

fn typed_broadcast_zip() -> &'static GraphBlueprint<GraphFlowShape<u64, (u64, u64)>> {
    static GRAPH: OnceLock<GraphBlueprint<GraphFlowShape<u64, (u64, u64)>>> = OnceLock::new();
    GRAPH.get_or_init(typed_broadcast_zip_graph)
}

fn broadcast_zip_full_noop_attrs() -> &'static GraphBlueprint<GraphFlowShape<u64, (u64, u64)>> {
    static GRAPH: OnceLock<GraphBlueprint<GraphFlowShape<u64, (u64, u64)>>> = OnceLock::new();
    GRAPH.get_or_init(broadcast_zip_graph_full_noop_attrs)
}

fn broadcast_zip_forced_erased() -> &'static GraphBlueprint<GraphFlowShape<u64, (u64, u64)>> {
    static GRAPH: OnceLock<GraphBlueprint<GraphFlowShape<u64, (u64, u64)>>> = OnceLock::new();
    GRAPH.get_or_init(broadcast_zip_graph_forced_erased)
}

fn typed_balance_merge() -> &'static GraphBlueprint<GraphFlowShape<u64, u64>> {
    static GRAPH: OnceLock<GraphBlueprint<GraphFlowShape<u64, u64>>> = OnceLock::new();
    GRAPH.get_or_init(typed_balance_merge_graph)
}

fn balance_merge_full_noop_attrs() -> &'static GraphBlueprint<GraphFlowShape<u64, u64>> {
    static GRAPH: OnceLock<GraphBlueprint<GraphFlowShape<u64, u64>>> = OnceLock::new();
    GRAPH.get_or_init(balance_merge_graph_full_noop_attrs)
}

fn prioritized_merge() -> &'static GraphBlueprint<datum::FanInShape<u64, u64>> {
    static GRAPH: OnceLock<GraphBlueprint<datum::FanInShape<u64, u64>>> = OnceLock::new();
    GRAPH.get_or_init(|| {
        GraphDsl::create(|builder| builder.add(MergePrioritized::<u64>::new(vec![4, 1]))).unwrap()
    })
}

fn merge_preferred() -> &'static GraphBlueprint<datum::MergePreferredShape<u64>> {
    static GRAPH: OnceLock<GraphBlueprint<datum::MergePreferredShape<u64>>> = OnceLock::new();
    GRAPH.get_or_init(|| {
        GraphDsl::create(|builder| builder.add(MergePreferred::<u64>::new(2))).unwrap()
    })
}

fn concat() -> &'static GraphBlueprint<datum::FanInShape<u64, u64>> {
    static GRAPH: OnceLock<GraphBlueprint<datum::FanInShape<u64, u64>>> = OnceLock::new();
    GRAPH.get_or_init(|| GraphDsl::create(|builder| builder.add(Concat::<u64>::new(2))).unwrap())
}

fn interleave() -> &'static GraphBlueprint<datum::FanInShape<u64, u64>> {
    static GRAPH: OnceLock<GraphBlueprint<datum::FanInShape<u64, u64>>> = OnceLock::new();
    GRAPH.get_or_init(|| {
        GraphDsl::create(|builder| builder.add(Interleave::<u64>::new(2, 2))).unwrap()
    })
}

fn merge_sorted() -> &'static GraphBlueprint<GraphFlowShape<(u64, u64), u64>> {
    static GRAPH: OnceLock<GraphBlueprint<GraphFlowShape<(u64, u64), u64>>> = OnceLock::new();
    GRAPH.get_or_init(|| {
        GraphDsl::try_create(|builder| {
            let unzip = builder.add(Unzip::<u64, u64>::new());
            let merge = builder.add(MergeSorted::<u64>::new());

            builder.connect(unzip.out0(), merge.inlet(0)?)?;
            builder.connect(unzip.out1(), merge.inlet(1)?)?;

            Ok(GraphFlowShape::new(unzip.inlet(), merge.outlet()))
        })
        .unwrap()
    })
}

fn merge_sequence() -> &'static GraphBlueprint<GraphFlowShape<(u64, u64), u64>> {
    static GRAPH: OnceLock<GraphBlueprint<GraphFlowShape<(u64, u64), u64>>> = OnceLock::new();
    GRAPH.get_or_init(|| {
        GraphDsl::try_create(|builder| {
            let unzip = builder.add(Unzip::<u64, u64>::new());
            let merge = builder.add(MergeSequence::<u64>::new(2, |item| *item));

            builder.connect(unzip.out0(), merge.inlet(0)?)?;
            builder.connect(unzip.out1(), merge.inlet(1)?)?;

            Ok(GraphFlowShape::new(unzip.inlet(), merge.outlet()))
        })
        .unwrap()
    })
}

fn merge_latest() -> &'static PairMergeLatestGraph {
    static GRAPH: OnceLock<PairMergeLatestGraph> = OnceLock::new();
    GRAPH.get_or_init(|| {
        GraphDsl::try_create(|builder| {
            let unzip = builder.add(Unzip::<u64, u64>::new());
            let merge = builder.add(MergeLatest::<u64>::new(2, false));

            builder.connect(unzip.out0(), merge.inlet(0)?)?;
            builder.connect(unzip.out1(), merge.inlet(1)?)?;

            Ok(GraphFlowShape::new(unzip.inlet(), merge.outlet()))
        })
        .unwrap()
    })
}

fn partition() -> &'static GraphBlueprint<GraphFlowShape<u64, u64>> {
    static GRAPH: OnceLock<GraphBlueprint<GraphFlowShape<u64, u64>>> = OnceLock::new();
    GRAPH.get_or_init(|| {
        GraphDsl::try_create(|builder| {
            let partition = builder.add(Partition::<u64>::new(2, |item| (*item % 2) as usize));
            let merge = builder.add(Merge::<u64>::new(2));

            builder.connect(partition.outlet(0)?, merge.inlet(0)?)?;
            builder.connect(partition.outlet(1)?, merge.inlet(1)?)?;

            Ok(GraphFlowShape::new(partition.inlet(), merge.outlet()))
        })
        .unwrap()
    })
}

fn unzip_with() -> &'static GraphBlueprint<GraphFlowShape<u64, (u64, u64)>> {
    static GRAPH: OnceLock<GraphBlueprint<GraphFlowShape<u64, (u64, u64)>>> = OnceLock::new();
    GRAPH.get_or_init(|| {
        GraphDsl::try_create(|builder| {
            let unzip = builder.add(UnzipWith::<u64, u64, u64>::new(|item| (item, item + 1)));
            let zip = builder.add(Zip::<u64, u64>::new());

            builder.connect(unzip.out0(), zip.in0())?;
            builder.connect(unzip.out1(), zip.in1())?;

            Ok(GraphFlowShape::new(unzip.inlet(), zip.outlet()))
        })
        .unwrap()
    })
}

fn cyclic_merge_preferred() -> &'static GraphBlueprint<GraphFlowShape<u64, u64>> {
    static GRAPH: OnceLock<GraphBlueprint<GraphFlowShape<u64, u64>>> = OnceLock::new();
    GRAPH.get_or_init(|| {
        GraphDsl::try_create(|builder| {
            let merge = builder.add(MergePreferred::<u64>::new(1));
            let broadcast = builder.add(Broadcast::<u64>::new(2));
            let buffer = builder.add(Buffer::<u64>::new(16, OverflowStrategy::Backpressure));
            let positive = builder.add(TakeWhile::<u64>::new(|item| *item > 0));
            let decrement = builder.add(MapStage::new(|item: u64| item - 1));

            builder.connect(merge.outlet(), broadcast.inlet())?;
            builder.connect(broadcast.outlet(1)?, buffer.inlet())?;
            builder.connect(buffer.outlet(), positive.inlet())?;
            builder.connect(positive.outlet(), decrement.inlet())?;
            builder.connect(decrement.outlet(), merge.preferred())?;

            Ok(GraphFlowShape::new(
                merge.secondary(0)?,
                broadcast.outlet(0)?,
            ))
        })
        .unwrap()
    })
}

fn fused_identity_10_100k() -> u64 {
    identity_chain_10()
        .run_with_input_report(
            0_u64..100_000,
            FusedExecutionConfig {
                event_limit: 5_000_000,
            },
        )
        .unwrap()
        .output
        .len() as u64
}

fn fused_map_10_100k() -> u64 {
    map_chain_10()
        .run_with_input_report(
            0_u64..100_000,
            FusedExecutionConfig {
                event_limit: 5_000_000,
            },
        )
        .unwrap()
        .output
        .last()
        .copied()
        .unwrap_or_default()
}

fn fused_identity_10_100k_count_sink() -> u64 {
    identity_chain_10()
        .run_count_with_input_report(
            0_u64..100_000,
            FusedExecutionConfig {
                event_limit: 5_000_000,
            },
        )
        .unwrap()
        .result as u64
}

fn fused_map_10_100k_fold_sink() -> u64 {
    map_chain_10()
        .run_fold_with_input_report(
            0_u64..100_000,
            0_u64,
            |acc, item| acc.wrapping_add(item),
            FusedExecutionConfig {
                event_limit: 5_000_000,
            },
        )
        .unwrap()
        .result
}

fn fused_identity_10_100k_typed_count_sink() -> u64 {
    identity_chain_10()
        .run_typed_linear_count_with_input_report(
            0_u64..100_000,
            FusedExecutionConfig {
                event_limit: 5_000_000,
            },
        )
        .unwrap()
        .result as u64
}

fn fused_map_10_100k_typed_fold_sink() -> u64 {
    map_chain_10()
        .run_typed_linear_fold_with_input_report(
            0_u64..100_000,
            0_u64,
            |acc, item| acc.wrapping_add(item),
            FusedExecutionConfig {
                event_limit: 5_000_000,
            },
        )
        .unwrap()
        .result
}

fn scalar_graph_add_sub_closure(elements: i64) -> u64 {
    scalar_closure_chain_10()
        .run_fold_with_input_report(
            0_i64..elements,
            0_i64,
            i64::wrapping_add,
            FusedExecutionConfig {
                event_limit: 25_000_000,
            },
        )
        .unwrap()
        .result as u64
}

fn scalar_graph_add_sub_chunked(elements: i64) -> u64 {
    scalar_builtin_chain_10()
        .run_fold_with_input_report(
            0_i64..elements,
            0_i64,
            i64::wrapping_add,
            FusedExecutionConfig {
                event_limit: 25_000_000,
            },
        )
        .unwrap()
        .result as u64
}

fn scalar_graph_add_sub_10_100k_closure() -> u64 {
    scalar_graph_add_sub_closure(100_000)
}

fn scalar_graph_add_sub_10_100k_chunked() -> u64 {
    scalar_graph_add_sub_chunked(100_000)
}

fn scalar_graph_add_sub_10_1m_closure() -> u64 {
    scalar_graph_add_sub_closure(1_000_000)
}

fn scalar_graph_add_sub_10_1m_chunked() -> u64 {
    scalar_graph_add_sub_chunked(1_000_000)
}

fn fused_identity_10_100k_typed_count_sink_consuming_builder() -> u64 {
    typed_identity_chain_10()
        .run_typed_linear_count_with_input_report(
            0_u64..100_000,
            FusedExecutionConfig {
                event_limit: 5_000_000,
            },
        )
        .unwrap()
        .result as u64
}

fn fused_map_10_100k_typed_fold_sink_consuming_builder() -> u64 {
    typed_map_chain_10()
        .run_typed_linear_fold_with_input_report(
            0_u64..100_000,
            0_u64,
            |acc, item| acc.wrapping_add(item),
            FusedExecutionConfig {
                event_limit: 5_000_000,
            },
        )
        .unwrap()
        .result
}

fn fused_identity_10_100k_typed_count_sink_named_attr() -> u64 {
    named_identity_chain_10()
        .run_typed_linear_count_with_input_report(
            0_u64..100_000,
            FusedExecutionConfig {
                event_limit: 5_000_000,
            },
        )
        .unwrap()
        .result as u64
}

fn fused_identity_10_100k_typed_count_sink_buffer_attr() -> u64 {
    buffered_identity_chain_10()
        .run_typed_linear_count_with_input_report(
            0_u64..100_000,
            FusedExecutionConfig {
                event_limit: 5_000_000,
            },
        )
        .unwrap()
        .result as u64
}

fn fused_identity_10_100k_count_sink_report_default_attrs() -> u64 {
    let report = identity_chain_10()
        .run_count_with_input_report(
            0_u64..100_000,
            FusedExecutionConfig {
                event_limit: 5_000_000,
            },
        )
        .unwrap();
    report.result as u64 + report.plan.stage_attributes.len() as u64
}

fn fused_identity_10_100k_count_sink_full_noop_attrs() -> u64 {
    let report = full_noop_attr_identity_chain_10()
        .run_count_with_input_report(
            0_u64..100_000,
            FusedExecutionConfig {
                event_limit: 5_000_000,
            },
        )
        .unwrap();
    report.result as u64 + report.plan.stage_attributes.len() as u64
}

fn metrics_linear_500k(level: MetricsLevel) -> u64 {
    let report = metrics_identity_chain_10(level)
        .run_count_with_input_report(
            0_u64..500_000,
            FusedExecutionConfig {
                event_limit: 20_000_000,
            },
        )
        .unwrap();
    report.result as u64
        + report
            .node_metrics
            .iter()
            .map(|node| node.elements_out)
            .sum::<u64>()
}

fn metrics_identity_chain_10(
    level: MetricsLevel,
) -> &'static GraphBlueprint<GraphFlowShape<u64, u64>> {
    static COUNTS: OnceLock<GraphBlueprint<GraphFlowShape<u64, u64>>> = OnceLock::new();
    static TIMING: OnceLock<GraphBlueprint<GraphFlowShape<u64, u64>>> = OnceLock::new();
    static STALLS: OnceLock<GraphBlueprint<GraphFlowShape<u64, u64>>> = OnceLock::new();
    let slot = match level {
        MetricsLevel::Counts => &COUNTS,
        MetricsLevel::Timing => &TIMING,
        MetricsLevel::Stalls => &STALLS,
        MetricsLevel::Off => unreachable!("metrics benchmark requires an enabled level"),
    };
    slot.get_or_init(|| {
        identity_chain_10()
            .clone()
            .add_attributes(Attributes::metrics_level(level))
    })
}

fn metrics_linear_500k_off() -> u64 {
    identity_chain_10()
        .run_count_with_input_report(
            0_u64..500_000,
            FusedExecutionConfig {
                event_limit: 20_000_000,
            },
        )
        .unwrap()
        .result as u64
}

fn metrics_linear_500k_counts() -> u64 {
    metrics_linear_500k(MetricsLevel::Counts)
}

fn metrics_linear_500k_timing() -> u64 {
    metrics_linear_500k(MetricsLevel::Timing)
}

fn metrics_linear_500k_stalls() -> u64 {
    metrics_linear_500k(MetricsLevel::Stalls)
}

fn fused_identity_10_100k_count_sink_forced_erased() -> u64 {
    let report = forced_erased_identity_chain_10()
        .run_count_with_input_report(
            0_u64..100_000,
            FusedExecutionConfig {
                event_limit: 5_000_000,
            },
        )
        .unwrap();
    report.result as u64 + report.plan.tier_changes.len() as u64
}

fn async_boundary_1_10k_count_sink() -> u64 {
    let ractor_boundary_config = AsyncBoundaryExecutionConfig {
        fused: FusedExecutionConfig {
            event_limit: 5_000_000,
        },
        buffer_size: 16,
    };
    let report = async_boundary_1()
        .run_async_boundary_count_with_input_report(0_u64..10_000, ractor_boundary_config)
        .unwrap();
    report.result as u64 + report.async_boundary_crossings as u64
}

fn async_boundary_9_10k_count_sink() -> u64 {
    let ractor_boundary_config = AsyncBoundaryExecutionConfig {
        fused: FusedExecutionConfig {
            event_limit: 5_000_000,
        },
        buffer_size: 16,
    };
    let report = async_boundary_9()
        .run_async_boundary_count_with_input_report(0_u64..10_000, ractor_boundary_config)
        .unwrap();
    report.result as u64 + report.async_boundary_crossings as u64
}

fn graph_build_identity_1() -> u64 {
    identity_chain(1).stage_count() as u64
}

fn graph_build_identity_10() -> u64 {
    identity_chain(10).stage_count() as u64
}

fn graph_build_identity_10_consuming_builder() -> u64 {
    typed_identity_chain(10).stage_count() as u64
}

fn graph_build_identity_100() -> u64 {
    identity_chain(100).stage_count() as u64
}

fn graph_build_broadcast_zip_current_builder() -> u64 {
    broadcast_zip_graph().stage_count() as u64
}

fn graph_build_broadcast_zip_consuming_builder() -> u64 {
    typed_broadcast_zip_graph().stage_count() as u64
}

fn graph_build_balance_merge_current_builder() -> u64 {
    balance_merge_graph().stage_count() as u64
}

fn graph_build_balance_merge_consuming_builder() -> u64 {
    typed_balance_merge_graph().stage_count() as u64
}

fn graph_build_junction_1() -> u64 {
    junction_chain(1).stage_count() as u64
}

fn graph_build_junction_10() -> u64 {
    junction_chain(10).stage_count() as u64
}

fn graph_build_junction_50() -> u64 {
    junction_chain(50).stage_count() as u64
}

fn junction_broadcast_zip_10k() -> u64 {
    broadcast_zip().run_with_input(0_u64..10_000).unwrap().len() as u64
}

fn junction_broadcast_zip_10k_consuming_builder() -> u64 {
    typed_broadcast_zip()
        .run_with_input(0_u64..10_000)
        .unwrap()
        .len() as u64
}

fn junction_broadcast_zip_10k_report_default_attrs() -> u64 {
    let report = broadcast_zip()
        .run_with_input_report(0_u64..10_000, FusedExecutionConfig::default())
        .unwrap();
    report.output.len() as u64 + report.plan.stage_attributes.len() as u64
}

fn junction_broadcast_zip_10k_full_noop_attrs() -> u64 {
    let report = broadcast_zip_full_noop_attrs()
        .run_with_input_report(0_u64..10_000, FusedExecutionConfig::default())
        .unwrap();
    report.output.len() as u64 + report.plan.stage_attributes.len() as u64
}

fn metrics_broadcast_zip_32mib(level: MetricsLevel) -> u64 {
    let report = metrics_broadcast_zip(level)
        .run_with_input_report(0_u64..4_194_304, FusedExecutionConfig::default())
        .unwrap();
    black_box(&report.output);
    report.output.len() as u64
        + report
            .node_metrics
            .iter()
            .map(|node| node.elements_out)
            .sum::<u64>()
}

fn metrics_broadcast_zip(
    level: MetricsLevel,
) -> &'static GraphBlueprint<GraphFlowShape<u64, (u64, u64)>> {
    static COUNTS: OnceLock<GraphBlueprint<GraphFlowShape<u64, (u64, u64)>>> = OnceLock::new();
    static TIMING: OnceLock<GraphBlueprint<GraphFlowShape<u64, (u64, u64)>>> = OnceLock::new();
    static STALLS: OnceLock<GraphBlueprint<GraphFlowShape<u64, (u64, u64)>>> = OnceLock::new();
    let slot = match level {
        MetricsLevel::Counts => &COUNTS,
        MetricsLevel::Timing => &TIMING,
        MetricsLevel::Stalls => &STALLS,
        MetricsLevel::Off => unreachable!("metrics benchmark requires an enabled level"),
    };
    slot.get_or_init(|| {
        broadcast_zip()
            .clone()
            .add_attributes(Attributes::metrics_level(level))
    })
}

fn metrics_broadcast_zip_32mib_off() -> u64 {
    let report = broadcast_zip()
        .run_with_input_report(0_u64..4_194_304, FusedExecutionConfig::default())
        .unwrap();
    black_box(&report.output);
    report.output.len() as u64
}

fn metrics_broadcast_zip_32mib_counts() -> u64 {
    metrics_broadcast_zip_32mib(MetricsLevel::Counts)
}

fn metrics_broadcast_zip_32mib_timing() -> u64 {
    metrics_broadcast_zip_32mib(MetricsLevel::Timing)
}

fn metrics_broadcast_zip_32mib_stalls() -> u64 {
    metrics_broadcast_zip_32mib(MetricsLevel::Stalls)
}

fn junction_broadcast_zip_10k_forced_erased() -> u64 {
    let report = broadcast_zip_forced_erased()
        .run_with_input_report(0_u64..10_000, FusedExecutionConfig::default())
        .unwrap();
    report.output.len() as u64 + report.plan.tier_changes.len() as u64
}

fn junction_balance_merge_10k() -> u64 {
    balance_merge().run_with_input(0_u64..10_000).unwrap().len() as u64
}

fn junction_balance_merge_10k_consuming_builder() -> u64 {
    typed_balance_merge()
        .run_with_input(0_u64..10_000)
        .unwrap()
        .len() as u64
}

fn junction_balance_merge_10k_report_default_attrs() -> u64 {
    let report = balance_merge()
        .run_with_input_report(0_u64..10_000, FusedExecutionConfig::default())
        .unwrap();
    report.output.len() as u64 + report.plan.stage_attributes.len() as u64
}

fn junction_balance_merge_10k_full_noop_attrs() -> u64 {
    let report = balance_merge_full_noop_attrs()
        .run_with_input_report(0_u64..10_000, FusedExecutionConfig::default())
        .unwrap();
    report.output.len() as u64 + report.plan.stage_attributes.len() as u64
}

fn junction_prioritized_merge_10k() -> u64 {
    let high = (0_u64..8_000).collect::<Vec<_>>();
    let low = (10_000_u64..12_000).collect::<Vec<_>>();
    prioritized_merge()
        .run_fan_in(vec![high, low])
        .unwrap()
        .len() as u64
}

fn junction_merge_preferred_10k() -> u64 {
    let preferred = (0_u64..8_000).collect::<Vec<_>>();
    let secondary_one = (10_000_u64..11_000).collect::<Vec<_>>();
    let secondary_two = (20_000_u64..21_000).collect::<Vec<_>>();
    merge_preferred()
        .run_merge_preferred(preferred, vec![secondary_one, secondary_two])
        .unwrap()
        .len() as u64
}

fn junction_concat_10k() -> u64 {
    concat()
        .run_concat(vec![
            (0_u64..5_000).collect(),
            (5_000_u64..10_000).collect(),
        ])
        .unwrap()
        .len() as u64
}

fn junction_interleave_10k() -> u64 {
    interleave()
        .run_interleave(
            vec![(0_u64..5_000).collect(), (5_000_u64..10_000).collect()],
            2,
            false,
        )
        .unwrap()
        .len() as u64
}

fn junction_merge_sorted_10k() -> u64 {
    merge_sorted()
        .run_with_input((0_u64..10_000).step_by(2).map(|item| (item, item + 1)))
        .unwrap()
        .len() as u64
}

fn junction_merge_sequence_10k() -> u64 {
    merge_sequence()
        .run_with_input((0_u64..20_000).step_by(2).map(|item| (item, item + 1)))
        .unwrap()
        .len() as u64
}

fn junction_merge_latest_10k() -> u64 {
    merge_latest()
        .run_with_input((0_u64..10_000).map(|item| (item, item + 10_000)))
        .unwrap()
        .len() as u64
}

fn junction_partition_10k() -> u64 {
    partition().run_with_input(0_u64..10_000).unwrap().len() as u64
}

fn junction_unzip_with_10k() -> u64 {
    unzip_with().run_with_input(0_u64..10_000).unwrap().len() as u64
}

fn cycle_merge_preferred_feedback_10k() -> u64 {
    cyclic_merge_preferred()
        .run_with_input_report(
            [10_000_u64],
            FusedExecutionConfig {
                event_limit: 5_000_000,
            },
        )
        .unwrap()
        .output
        .len() as u64
}

fn custom_identity_stage_10k() -> u64 {
    static GRAPH: OnceLock<GraphBlueprint<GraphFlowShape<u64, u64>>> = OnceLock::new();
    GRAPH
        .get_or_init(|| identity_chain(1))
        .run_with_input(0_u64..10_000)
        .unwrap()
        .len() as u64
}

fn custom_map_stage_10k() -> u64 {
    static GRAPH: OnceLock<GraphBlueprint<GraphFlowShape<u64, u64>>> = OnceLock::new();
    GRAPH
        .get_or_init(|| map_chain(1))
        .run_with_input(0_u64..10_000)
        .unwrap()
        .len() as u64
}

fn bidi_join_10k() -> u64 {
    let flow = BidiFlow::from_flows(
        Flow::identity().map(|item: u64| item + 1),
        Flow::identity().map(|item: u64| item * 2),
    )
    .join(Flow::identity().map(|item: u64| item - 1));
    Source::from_iter(0_u64..10_000)
        .via(flow)
        .run_with(Sink::fold(0_u64, |acc, item| acc.wrapping_add(item)))
        .unwrap()
        .wait()
        .unwrap()
}

fn bidi_atop_10k() -> u64 {
    let first = BidiFlow::from_flows(
        Flow::identity().map(|item: u64| item + 1),
        Flow::identity().map(|item: u64| item * 2),
    );
    let second = BidiFlow::from_flows(
        Flow::identity().map(|item: u64| item * 3),
        Flow::identity().map(|item: u64| item - 4),
    );
    Source::from_iter(0_u64..10_000)
        .via(first.atop(second).join(Flow::identity()))
        .run_with(Sink::fold(0_u64, |acc, item| acc.wrapping_add(item)))
        .unwrap()
        .wait()
        .unwrap()
}