lix 0.17.1

Embeddable version control for apps and AI agents.
Documentation
use std::hint::black_box;
use std::time::{Duration, Instant};

use super::{ObserveQueryEvaluation, ObserveQueryState, ObserveSharedContent};
use crate::{ExecuteResult, Value};

fn sized_blob_result(size: usize, changed: bool) -> ExecuteResult {
    let mut bytes = vec![0; size];
    if changed {
        bytes[size / 2] = 1;
    }
    ExecuteResult::from_rows(
        vec!["data".to_string()],
        vec![vec![Value::Blob(bytes.into())]],
    )
}

#[test]
#[ignore = "manual performance probe"]
fn observe_result_equivalence_baseline_performance_probe() {
    run_observe_result_equivalence_performance_probe(false);
}

#[test]
#[ignore = "manual performance probe"]
fn observe_result_equivalence_shared_performance_probe() {
    run_observe_result_equivalence_performance_probe(true);
}

fn run_observe_result_equivalence_performance_probe(share_comparison: bool) {
    let runtime = tokio::runtime::Builder::new_current_thread()
        .enable_all()
        .build()
        .expect("build performance probe runtime");
    let mode = if share_comparison {
        "candidate"
    } else {
        "baseline"
    };
    println!("size_mib\tobservers\tworkload\tmode\tmedian_ms\tp10_ms\tp90_ms\titerations");
    for size_mib in [1_usize, 10] {
        for observers in [1_usize, 2, 4, 16] {
            for changed in [false, true] {
                let state = ObserveQueryState::new();
                let initial = runtime
                    .block_on(
                        state.evaluate(1, share_comparison && observers > 1, || async {
                            Ok(sized_blob_result(size_mib * 1024 * 1024, false))
                        }),
                    )
                    .expect("seed performance probe result");
                let mut last_rows = vec![initial.rows; observers];
                let mut last_content = vec![initial.shared_content; observers];
                let mut generation = 1_u64;

                for _ in 0..3 {
                    run_probe_iteration(
                        &runtime,
                        &state,
                        &mut generation,
                        size_mib * 1024 * 1024,
                        changed,
                        share_comparison,
                        &mut last_rows,
                        &mut last_content,
                    );
                }
                let calibration_start = Instant::now();
                for _ in 0..3 {
                    run_probe_iteration(
                        &runtime,
                        &state,
                        &mut generation,
                        size_mib * 1024 * 1024,
                        changed,
                        share_comparison,
                        &mut last_rows,
                        &mut last_content,
                    );
                }
                let per_iteration = calibration_start.elapsed() / 3;
                let iterations = u32::try_from(
                    (Duration::from_millis(30).as_nanos() / per_iteration.as_nanos().max(1))
                        .clamp(2, 100),
                )
                .expect("bounded performance probe iteration count should fit u32");

                let mut samples = Vec::with_capacity(21);
                for _ in 0..21 {
                    let start = Instant::now();
                    for _ in 0..iterations {
                        run_probe_iteration(
                            &runtime,
                            &state,
                            &mut generation,
                            size_mib * 1024 * 1024,
                            changed,
                            share_comparison,
                            &mut last_rows,
                            &mut last_content,
                        );
                    }
                    samples.push(start.elapsed().as_secs_f64() * 1000.0 / f64::from(iterations));
                }
                samples.sort_by(f64::total_cmp);
                println!(
                    "{size_mib}\t{observers}\t{}\t{mode}\t{:.3}\t{:.3}\t{:.3}\t{iterations}",
                    if changed { "changed" } else { "unchanged" },
                    samples[10],
                    samples[2],
                    samples[18]
                );
            }
        }
    }
}

fn run_probe_iteration(
    runtime: &tokio::runtime::Runtime,
    state: &ObserveQueryState,
    generation: &mut u64,
    size: usize,
    changed: bool,
    share_comparison: bool,
    last_rows: &mut [ExecuteResult],
    last_content: &mut [Option<ObserveSharedContent>],
) {
    *generation += 1;
    let variant = changed && (*generation).is_multiple_of(2);
    let evaluation = runtime
        .block_on(state.evaluate(
            *generation,
            share_comparison && last_rows.len() > 1,
            || async move { Ok(sized_blob_result(size, variant)) },
        ))
        .expect("performance probe evaluation");
    for (previous_rows, previous_content) in last_rows.iter_mut().zip(last_content.iter_mut()) {
        let rows_changed = observed_rows_changed(previous_rows, *previous_content, &evaluation);
        *previous_content = evaluation.shared_content;
        if rows_changed {
            *previous_rows = evaluation.rows.clone();
        }
        black_box(rows_changed);
    }
}

fn observed_rows_changed(
    previous_rows: &ExecuteResult,
    previous_content: Option<ObserveSharedContent>,
    evaluation: &ObserveQueryEvaluation,
) -> bool {
    evaluation.rows_changed_since(Some(previous_rows), previous_content)
}