diff --git a/rqe-optimizer/examples/small_problem.rs b/rqe-optimizer/examples/small_problem.rs index b46403fc..c0fa6928 100644 --- a/rqe-optimizer/examples/small_problem.rs +++ b/rqe-optimizer/examples/small_problem.rs @@ -265,9 +265,9 @@ fn print_mapping( } } -fn print_candidate(candidate_number: usize, deployment: &rqe_optimizer::Deployment) { +fn print_deployment(label: &str, deployment: &rqe_optimizer::Deployment) { println!( - "candidate {candidate_number}: {} {} {} by {:?} (x={}ms, y={}ms)", + "{label}: {} {} {} by {:?} (x={}ms, y={}ms)", deployment.config.sketch, deployment.config.sketch_config, deployment.metric, @@ -417,7 +417,7 @@ fn main() { .take(usize::try_from(print_candidates).expect("u64 fits in usize")) .enumerate() { - print_candidate(index, deployment); + print_deployment(&format!("candidate {index}"), deployment); } return; } @@ -460,9 +460,23 @@ fn main() { objective.value(&solution.plan_cost) ); print_plan_cost(" totals", &solution.plan_cost); - print_mapping(1, &solution.mapping, &raqes, &deployments); - for (raqe, latency) in raqes.iter().zip(&solution.plan_cost.query_latency_ms) { - println!(" {}: query_latency={latency:.3e} ms", raqe.id); + for (index, planned) in solution.deployments.iter().enumerate() { + print_deployment(&format!("deployment {index}"), &planned.deployment); + println!( + " retained_instances={} key_tracker={}", + planned.retained_instance_count, + planned.deployment.key_tracker.is_some(), + ); + } + for ((raqe, planned), latency) in raqes + .iter() + .zip(&solution.raqes) + .zip(&solution.plan_cost.query_latency_ms) + { + println!( + " {} -> deployment {} merged_instances={} query_latency={latency:.3e} ms", + raqe.id, planned.deployment, planned.merged_instance_count, + ); } return; } diff --git a/rqe-optimizer/src/milp.rs b/rqe-optimizer/src/milp.rs index e107d5e3..de8f82a1 100644 --- a/rqe-optimizer/src/milp.rs +++ b/rqe-optimizer/src/milp.rs @@ -45,12 +45,32 @@ impl Objective { } } +/// The solved plan: the deployments to run and which one serves each RAQE. #[derive(Debug, Clone)] pub struct MilpSolution { - pub mapping: Mapping, + /// Active deployments only, in order of the first RAQE each serves. + pub deployments: Vec, + /// Index-aligned with the input RAQE slice. + pub raqes: Vec, pub plan_cost: PlanCost, } +#[derive(Debug, Clone)] +pub struct PlannedDeployment { + pub deployment: Deployment, + /// Windows kept per group (per sketch, not × `card(G)`): `x/y` open plus + /// `(L − x)/y + 1` closed, for the longest lookback `L` it serves. + pub retained_instance_count: u64, +} + +#[derive(Debug, Clone)] +pub struct PlannedRaqe { + /// Index into [`MilpSolution::deployments`]. + pub deployment: usize, + /// `n = L/x` windows merged per query; 1 means Direct, more means Merge. + pub merged_instance_count: u64, +} + /// Minimize `objective` subject to each RAQE's latency SLA. `facts` must pass /// [`crate::validate_facts`]. Missing eligible deployments are a caller input /// error; use `enumerate::unservable` to report them before calling this @@ -198,7 +218,54 @@ pub fn minimize( }) .collect(); let plan_cost = score(raqes, deployments, &mapping, facts); - Ok(MilpSolution { mapping, plan_cost }) + let (deployments, raqes) = plan(raqes, deployments, &mapping); + Ok(MilpSolution { + deployments, + raqes, + plan_cost, + }) +} + +/// Keeps the deployments `mapping` uses, renumbered, with their instance +/// counts. Every pair in `mapping` is eligible, so the counts exist. +fn plan( + raqes: &[Raqe], + candidates: &[Deployment], + mapping: &Mapping, +) -> (Vec, Vec) { + let mut planned_index: Vec> = vec![None; candidates.len()]; + let mut planned_deployments: Vec = Vec::new(); + let mut planned_raqes = Vec::with_capacity(raqes.len()); + // `retained_instance_count` holds the closed windows until the open ones + // are added below, once per deployment. + for (raqe, &candidate_index) in raqes.iter().zip(mapping) { + let deployment = &candidates[candidate_index]; + let index = *planned_index[candidate_index].get_or_insert_with(|| { + planned_deployments.push(PlannedDeployment { + deployment: deployment.clone(), + retained_instance_count: 0, + }); + planned_deployments.len() - 1 + }); + let closed = deployment + .closed_instance_count(raqe.lookback_ms) + .expect("eligible deployments have whole window counts"); + let planned = &mut planned_deployments[index]; + planned.retained_instance_count = planned.retained_instance_count.max(closed); + planned_raqes.push(PlannedRaqe { + deployment: index, + merged_instance_count: deployment + .query_instance_count(raqe.lookback_ms) + .expect("eligible deployments divide the lookback"), + }); + } + for planned in &mut planned_deployments { + planned.retained_instance_count += planned + .deployment + .active_instance_count() + .expect("eligible deployments have whole window counts"); + } + (planned_deployments, planned_raqes) } #[cfg(test)] @@ -208,7 +275,6 @@ mod tests { use crate::enumerate::brute_force; use crate::test_support::{self, facts}; use crate::{AtomicCostEntry, Capability}; - use std::collections::BTreeSet; /// Window and slide 60 s, merge 1 CPU-sec. fn deployment(insert_cpu_secs: f64, memory: f64, query_cpu_secs: f64) -> Deployment { @@ -222,6 +288,20 @@ mod tests { } } + /// The candidate index serving each RAQE. + fn chosen(milp: &MilpSolution, candidates: &[Deployment]) -> Mapping { + milp.raqes + .iter() + .map(|planned| { + let deployment = &milp.deployments[planned.deployment].deployment; + candidates + .iter() + .position(|candidate| candidate == deployment) + .expect("planned deployments come from the candidates") + }) + .collect() + } + fn weights(w_cpu: f64, w_mem: f64) -> Objective { Objective::AUCCost { w_cpu, w_mem } } @@ -260,8 +340,7 @@ mod tests { let candidates = build_all_candidates(&raqes, &costs, &facts(1, 1), false); let milp = minimize(&raqes, &candidates, &facts(1, 1), Objective::default()) .expect("feasible MILP"); - let used: BTreeSet = milp.mapping.iter().copied().collect(); - raqes.len() - used.len() + raqes.len() - milp.deployments.len() } #[test] @@ -287,9 +366,9 @@ mod tests { // CPU 1 + 3/60 with latency 3, or 2 + 1/60 with latency 1. let deployments = vec![deployment(1.0, 20.0, 3.0), deployment(2.0, 10.0, 1.0)]; let solve = |raqes: &[Raqe]| { - minimize(raqes, &deployments, &facts(1, 1), Objective::default()) - .expect("feasible MILP") - .mapping + let milp = minimize(raqes, &deployments, &facts(1, 1), Objective::default()) + .expect("feasible MILP"); + chosen(&milp, &deployments) }; assert_eq!(solve(&raqes), vec![0]); @@ -322,9 +401,9 @@ mod tests { deployment(0.05, 2.0 * BYTES_PER_GIB, 0.0), ]; let solve = |objective| { - minimize(&raqes, &deployments, &facts(1, 1), objective) - .expect("feasible MILP") - .mapping + let milp = + minimize(&raqes, &deployments, &facts(1, 1), objective).expect("feasible MILP"); + chosen(&milp, &deployments) }; assert_eq!(solve(weights(1.0, 0.0)), vec![1]); @@ -364,7 +443,33 @@ mod tests { ]; let milp = minimize(&raqes, &deployments, &facts(1, 1), Objective::default()) .expect("feasible MILP"); - assert_eq!(milp.mapping, vec![1]); + assert_eq!(chosen(&milp, &deployments), vec![1]); assert!(milp.plan_cost.query_latency_ms[0] <= 1e-4); } + + #[test] + fn plan_lists_active_deployments_with_their_instance_counts() { + // Window 60 s, slide 20 s; the unused candidate 0 is CPU-heavy. + let shared = test_support::deployment(1.0, 1.0, 1.0, 1.0, 60_000, 20_000); + let unused = test_support::deployment(1.0, 1e6, 1.0, 1.0, 60_000, 20_000); + let raqes = vec![raqe("short", 60_000), raqe("long", 600_000)]; + let milp = minimize( + &raqes, + &[unused, shared.clone()], + &facts(1, 1), + Objective::default(), + ) + .expect("feasible MILP"); + + assert_eq!(milp.deployments.len(), 1); + assert_eq!(milp.deployments[0].deployment, shared); + // 60/20 open + (600 − 60)/20 + 1 closed, for the longer lookback. + assert_eq!(milp.deployments[0].retained_instance_count, 3 + 28); + let merged: Vec<_> = milp + .raqes + .iter() + .map(|planned| (planned.deployment, planned.merged_instance_count)) + .collect(); + assert_eq!(merged, vec![(0, 1), (0, 10)]); + } }