1use crate::accelerator::ExecutionBackend;
9use crate::cluster::ClusterState;
10use crate::cluster::NodeStatus;
11use crate::host::{AccelerationMode, RuntimeProfile};
12use crate::protocol::NodeResources;
13
14pub const DELIVERY_RATIO_INT8_THRESHOLD: f32 = 0.95;
16pub const DELIVERY_RATIO_INT4_THRESHOLD: f32 = 0.80;
17
18#[derive(Clone, Debug, PartialEq)]
20pub struct LayerSpec {
21 pub index: usize,
23 pub vram_gb: f32,
25 pub num_weights: u32,
27}
28
29impl Default for LayerSpec {
30 fn default() -> Self {
31 Self {
32 index: 0,
33 vram_gb: 1.0,
34 num_weights: 0,
35 }
36 }
37}
38
39#[derive(Clone, Debug, PartialEq)]
41pub struct LayerAssignment {
42 pub node_id: String,
44 pub start_layer: usize,
46 pub end_layer: usize,
48 pub used_vram_gb: f32,
50 pub num_layers: usize,
52}
53
54impl LayerAssignment {
55 pub fn new(node_id: String, start_layer: usize, end_layer: usize, vram_gb: f32) -> Self {
57 Self {
58 node_id,
59 start_layer,
60 end_layer,
61 used_vram_gb: vram_gb,
62 num_layers: end_layer - start_layer,
63 }
64 }
65
66 pub fn avg_vram_per_layer(&self) -> f32 {
68 if self.num_layers == 0 {
69 0.0
70 } else {
71 self.used_vram_gb / self.num_layers as f32
72 }
73 }
74}
75
76#[derive(Clone, Copy, Debug, Default, PartialEq, Eq)]
78pub enum QuantizationMode {
79 #[default]
81 None,
82 Int8,
84 Int4,
86}
87
88#[derive(Clone, Debug, Default)]
90pub struct PlacementPlan {
91 pub assignments: Vec<LayerAssignment>,
93 pub quantization_mode: QuantizationMode,
95 pub total_layers: usize,
97 pub participating_nodes: Vec<String>,
99}
100
101impl PlacementPlan {
102 pub fn new(assignments: Vec<LayerAssignment>, quantization_mode: QuantizationMode) -> Self {
104 let participating_nodes = assignments.iter().map(|a| a.node_id.clone()).collect();
105 let total_layers = assignments.iter().map(|a| a.num_layers).sum();
106
107 Self {
108 assignments,
109 quantization_mode,
110 total_layers,
111 participating_nodes,
112 }
113 }
114
115 pub fn summary(&self) -> String {
117 let mode_str = match self.quantization_mode {
118 QuantizationMode::None => "Full Precision",
119 QuantizationMode::Int8 => "8-bit Quantized",
120 QuantizationMode::Int4 => "4-bit Quantized",
121 };
122
123 format!(
124 "Placement Plan ({})\n\
125 =================\n\
126 Total layers: {}\n\
127 Quantization: {}\n\
128 Nodes: {}\n",
129 mode_str,
130 self.total_layers,
131 match self.quantization_mode {
132 QuantizationMode::None => "Full Precision".to_string(),
133 QuantizationMode::Int8 => "8-bit Quantized".to_string(),
134 QuantizationMode::Int4 => "4-bit Quantized".to_string(),
135 },
136 self.participating_nodes.join(", ")
137 )
138 }
139}
140
141#[derive(Clone, Copy, Debug, PartialEq, Eq)]
143pub struct PlanningTuning {
144 pub max_layers_per_assignment: usize,
146}
147
148impl PlanningTuning {
149 pub fn from_runtime_profile(profile: &RuntimeProfile, total_layers: usize) -> Self {
151 let backend = ExecutionBackend::from_runtime_profile(profile);
152 let worker_count = backend.worker_count.max(1);
153 let accelerator_bias = match profile.acceleration_mode {
154 AccelerationMode::Gpu => 2,
155 AccelerationMode::Avx512 => 1,
156 _ => 0,
157 };
158 let vector_bias = (backend.vector_width_bits / 256).max(1);
159 let target_chunks = (worker_count + accelerator_bias + vector_bias - 1).max(1);
160 let chunk_size = if total_layers == 0 {
161 1
162 } else {
163 total_layers.div_ceil(target_chunks).max(1)
164 };
165
166 Self {
167 max_layers_per_assignment: chunk_size,
168 }
169 }
170}
171
172pub fn select_quantization_mode(delivery_ratio: f32) -> QuantizationMode {
174 if delivery_ratio >= DELIVERY_RATIO_INT8_THRESHOLD {
175 QuantizationMode::None
176 } else if delivery_ratio >= DELIVERY_RATIO_INT4_THRESHOLD {
177 QuantizationMode::Int8
178 } else {
179 QuantizationMode::Int4
180 }
181}
182
183pub fn assign_layers_sequentially(
190 nodes: &[NodeResources],
191 layers: &[LayerSpec],
192) -> Result<Vec<LayerAssignment>, String> {
193 if nodes.is_empty() {
194 return Err("at least one node is required".into());
195 }
196 if layers.is_empty() {
197 return Ok(Vec::new());
198 }
199
200 let mut assignments = Vec::with_capacity(nodes.len());
201 let mut node_idx = 0usize;
202 let mut remaining_capacity = nodes[0].vram_gb;
203 let mut current_start = 0usize;
204 let mut current_vram = 0.0f32;
205
206 for (i, layer) in layers.iter().enumerate() {
207 while layer.vram_gb > remaining_capacity {
208 if i > current_start {
210 let start_layer = layers[current_start].index;
211 let end_layer = layers[i - 1].index + 1;
212 assignments.push(LayerAssignment::new(
213 nodes[node_idx].id.clone(),
214 start_layer,
215 end_layer,
216 current_vram,
217 ));
218 }
219
220 node_idx += 1;
221 if node_idx >= nodes.len() {
222 return Err(format!(
223 "insufficient cluster VRAM for layer {} (needs {:.2} GB)",
224 layer.index, layer.vram_gb
225 ));
226 }
227 remaining_capacity = nodes[node_idx].vram_gb;
228 current_start = i;
229 current_vram = 0.0;
230 }
231
232 remaining_capacity -= layer.vram_gb;
234 current_vram += layer.vram_gb;
235 }
236
237 if current_start < layers.len() {
239 let start_layer = layers[current_start].index;
240 let end_layer = layers[layers.len() - 1].index + 1;
241 assignments.push(LayerAssignment::new(
242 nodes[node_idx].id.clone(),
243 start_layer,
244 end_layer,
245 current_vram,
246 ));
247 }
248
249 Ok(assignments)
250}
251
252pub fn chunk_assignments_for_workers(
254 assignments: Vec<LayerAssignment>,
255 max_layers_per_assignment: usize,
256) -> Vec<LayerAssignment> {
257 let chunk_size = max_layers_per_assignment.max(1);
258
259 let needs_chunking = assignments.iter().any(|a| a.num_layers > chunk_size);
260 if !needs_chunking {
261 return assignments;
262 }
263
264 let mut chunked = Vec::with_capacity(assignments.len() * 2);
265
266 for assignment in assignments {
267 if assignment.num_layers <= chunk_size {
268 chunked.push(assignment);
269 continue;
270 }
271
272 let avg_vram = assignment.avg_vram_per_layer();
273 let mut start_layer = assignment.start_layer;
274
275 while start_layer < assignment.end_layer {
276 let end_layer = (start_layer + chunk_size).min(assignment.end_layer);
277 let num_layers = end_layer - start_layer;
278 chunked.push(LayerAssignment::new(
279 assignment.node_id.clone(),
280 start_layer,
281 end_layer,
282 avg_vram * num_layers as f32,
283 ));
284 start_layer = end_layer;
285 }
286 }
287
288 chunked
289}
290
291fn assign_layers_chunked(
295 nodes: &[NodeResources],
296 layers: &[LayerSpec],
297 max_layers_per_assignment: usize,
298) -> Result<Vec<LayerAssignment>, String> {
299 if nodes.is_empty() {
300 return Err("at least one node is required".into());
301 }
302 if layers.is_empty() {
303 return Ok(Vec::new());
304 }
305
306 let chunk_size = max_layers_per_assignment.max(1);
307 let est_assignments = (layers.len().div_ceil(chunk_size)).max(nodes.len());
308 let mut assignments = Vec::with_capacity(est_assignments);
309 let mut node_idx = 0usize;
310 let mut remaining_vram = nodes[0].vram_gb;
311 let mut chunk_start = 0usize;
313 let mut chunk_vram = 0.0f32;
314
315 for (i, layer) in layers.iter().enumerate() {
316 while layer.vram_gb > remaining_vram {
318 if i > chunk_start {
319 assignments.push(LayerAssignment::new(
320 nodes[node_idx].id.clone(),
321 chunk_start,
322 i,
323 chunk_vram,
324 ));
325 }
326 node_idx += 1;
327 if node_idx >= nodes.len() {
328 return Err(format!(
329 "insufficient cluster VRAM for layer {} (needs {:.2} GB)",
330 layer.index, layer.vram_gb
331 ));
332 }
333 remaining_vram = nodes[node_idx].vram_gb;
334 chunk_start = i;
335 chunk_vram = 0.0;
336 }
337
338 remaining_vram -= layer.vram_gb;
339 chunk_vram += layer.vram_gb;
340
341 let chunk_len = i + 1 - chunk_start;
343 if chunk_len >= chunk_size {
344 assignments.push(LayerAssignment::new(
345 nodes[node_idx].id.clone(),
346 chunk_start,
347 i + 1,
348 chunk_vram,
349 ));
350 chunk_start = i + 1;
351 chunk_vram = 0.0;
352 }
353 }
354
355 if chunk_start < layers.len() {
360 assignments.push(LayerAssignment::new(
361 nodes[node_idx].id.clone(),
362 chunk_start,
363 layers.len(),
364 chunk_vram,
365 ));
366 }
367
368 Ok(assignments)
369}
370
371pub fn assign_layers_with_runtime_profile(
373 nodes: &[NodeResources],
374 layers: &[LayerSpec],
375 profile: &RuntimeProfile,
376) -> Result<Vec<LayerAssignment>, String> {
377 let tuning = PlanningTuning::from_runtime_profile(profile, layers.len());
378 assign_layers_chunked(nodes, layers, tuning.max_layers_per_assignment)
379}
380
381pub fn assign_layers_with_fault_tolerance(
383 cluster: &ClusterState,
384 layers: &[LayerSpec],
385) -> Result<PlacementPlan, String> {
386 let nodes = cluster.nodes_snapshot();
387
388 if nodes.is_empty() {
389 return Err("no nodes available".into());
390 }
391
392 let assignments = assign_layers_sequentially(&nodes, layers)?;
394
395 let (sum_delivery, active_count) = {
403 let metrics = cluster
404 .metrics
405 .lock()
406 .unwrap_or_else(|poison| poison.into_inner());
407 let mut sum = 0.0f32;
408 let mut count = 0usize;
409 for metric in metrics.values() {
410 if metric.status == NodeStatus::Active {
411 sum += metric.delivery_ratio;
412 count += 1;
413 }
414 }
415 (sum, count)
416 };
417
418 let total_delivery_ratio = if active_count > 0 {
424 sum_delivery / active_count as f32
425 } else {
426 0.0
427 };
428
429 let quantization_mode = select_quantization_mode(total_delivery_ratio);
431
432 Ok(PlacementPlan::new(assignments, quantization_mode))
433}
434
435pub fn assign_layers_with_fault_tolerance_and_runtime(
440 cluster: &ClusterState,
441 layers: &[LayerSpec],
442 profile: &RuntimeProfile,
443) -> Result<PlacementPlan, String> {
444 let nodes = cluster.nodes_snapshot();
445
446 if nodes.is_empty() {
447 return Err("no nodes available".into());
448 }
449
450 let tuning = PlanningTuning::from_runtime_profile(profile, layers.len());
451 let assignments = assign_layers_chunked(&nodes, layers, tuning.max_layers_per_assignment)?;
453
454 let (sum_delivery, active_count) = {
456 let metrics = cluster
457 .metrics
458 .lock()
459 .unwrap_or_else(|poison| poison.into_inner());
460 let mut sum = 0.0f32;
461 let mut count = 0usize;
462 for metric in metrics.values() {
463 if metric.status == NodeStatus::Active {
464 sum += metric.delivery_ratio;
465 count += 1;
466 }
467 }
468 (sum, count)
469 };
470
471 let total_delivery_ratio = if active_count > 0 {
472 sum_delivery / active_count as f32
473 } else {
474 0.0
475 };
476
477 let quantization_mode = select_quantization_mode(total_delivery_ratio);
479
480 Ok(PlacementPlan::new(assignments, quantization_mode))
481}
482
483pub fn rebalance_assignments(cluster: &ClusterState, plan: &mut PlacementPlan) -> bool {
485 let mut needs_rebalance = false;
487
488 let metrics = cluster
489 .metrics
490 .lock()
491 .unwrap_or_else(|poison| poison.into_inner());
492
493 for assignment in &mut plan.assignments {
494 if let Some(metric) = metrics.get(&assignment.node_id) {
495 if metric.status != NodeStatus::Active {
496 continue;
498 }
499
500 let available = metric.available_vram_gb;
502 let avg_layer_size = assignment.avg_vram_per_layer();
503
504 if available > 0.0 && avg_layer_size > 0.0 {
505 let potential_layers = (available / avg_layer_size) as usize;
506
507 if potential_layers > 2 {
509 needs_rebalance = true;
510 break;
511 }
512 }
513 }
514 }
515
516 needs_rebalance
517}
518
519#[cfg(test)]
521pub fn simulate_layer_streaming(
522 node_id: &str,
523 cluster: &ClusterState,
524 start_layer: usize,
525 end_layer: usize,
526) -> Option<crate::cluster::NodeMetrics> {
527 let mut metrics = cluster.get_metrics(node_id)?;
528
529 let num_layers = end_layer - start_layer;
531 let avg_vram = (metrics.available_vram_gb + metrics.vram_gb) / 2.0;
532 let vram_per_layer = avg_vram / num_layers as f32;
533 metrics.record_vram_usage(vram_per_layer * num_layers as f32);
534
535 metrics.set_streaming_layers(start_layer, end_layer);
537
538 Some(metrics)
539}
540
541pub fn calculate_cluster_health(cluster: &ClusterState) -> (f32, usize, Vec<String>) {
543 let metrics = cluster
545 .metrics
546 .lock()
547 .unwrap_or_else(|poison| poison.into_inner());
548
549 if metrics.is_empty() {
550 return (0.0, 0, vec![]);
551 }
552
553 let mut sum_delivery_ratio = 0.0f32;
554 let mut active_count = 0usize;
555 let mut failed_nodes = Vec::new();
556
557 for (node_id, metric) in metrics.iter() {
559 match metric.status {
560 NodeStatus::Active => {
561 sum_delivery_ratio += metric.delivery_ratio;
562 active_count += 1;
563 }
564 NodeStatus::Failed => {
565 failed_nodes.push(node_id.clone());
566 }
567 _ => {}
568 }
569 }
570
571 let avg_delivery_ratio = if active_count > 0 {
572 sum_delivery_ratio / active_count as f32
573 } else {
574 0.0
575 };
576
577 let failed_count = failed_nodes.len();
578
579 (avg_delivery_ratio, failed_count, failed_nodes)
580}
581
582#[derive(Debug, Clone)]
584pub struct MigrationPlanner {
585 pub source_node: String,
586 pub target_node: String,
587 pub layers: Vec<usize>,
588 pub est_migration_time_ms: f32,
589}
590
591impl MigrationPlanner {
592 pub fn new(source: String, target: String, layers: Vec<usize>) -> Self {
593 let est_time = layers.len() as f32 * 45.0; Self {
595 source_node: source,
596 target_node: target,
597 layers,
598 est_migration_time_ms: est_time,
599 }
600 }
601
602 pub fn generate_handoff_plan(&self) -> Vec<String> {
604 let mut steps = Vec::new();
605 let target_node = &self.target_node;
606 let source_node = &self.source_node;
607 steps.push(format!(
608 "PREPARE: Target node {target_node} allocating VRAM"
609 ));
610 steps.push(format!(
611 "STREAM: Moving layers {:?} to {target_node}",
612 self.layers
613 ));
614 steps.push(format!("VERIFY: Integrity check on {target_node}"));
615 steps.push(format!("COMMIT: Switch routing to {target_node}"));
616 steps.push(format!("CLEANUP: Free VRAM on source node {source_node}"));
617 steps
618 }
619}
620
621pub struct RebalanceTrigger {
623 pub min_imbalance_ratio: f32,
624 pub max_p95_latency_ms: f32,
625}
626
627impl Default for RebalanceTrigger {
628 fn default() -> Self {
629 Self {
630 min_imbalance_ratio: 0.25, max_p95_latency_ms: 25.0,
632 }
633 }
634}
635
636impl RebalanceTrigger {
637 pub fn evaluate(
638 &self,
639 cluster: &ClusterState,
640 current_plan: &PlacementPlan,
641 ) -> Option<MigrationPlanner> {
642 let (overtaxed, available) = {
643 let metrics = cluster
644 .metrics
645 .lock()
646 .unwrap_or_else(|poison| poison.into_inner());
647
648 let mut active_count = 0usize;
649 for m in metrics.values() {
650 if m.status == NodeStatus::Active {
651 active_count += 1;
652 }
653 }
654 if active_count < 2 {
655 return None;
656 }
657
658 let mut overtaxed_name = None;
659 let mut available_name = None;
660
661 for m in metrics.values() {
662 if m.status == NodeStatus::Active {
663 if m.delivery_ratio < 0.90 || m.avg_latency_us > 15.0 {
664 overtaxed_name = Some(m.name.clone());
665 } else if m.available_vram_gb > 8.0 {
666 available_name = Some(m.name.clone());
667 }
668 }
669 }
670 (overtaxed_name, available_name)
671 };
672
673 if let (Some(source), Some(target)) = (overtaxed, available) {
674 if source == target {
675 return None;
676 }
677
678 let layers_to_move = current_plan
680 .assignments
681 .iter()
682 .find(|a| a.node_id == source)
683 .map(|a| (a.start_layer..a.end_layer).take(2).collect::<Vec<_>>())
684 .unwrap_or_default();
685
686 if !layers_to_move.is_empty() {
687 return Some(MigrationPlanner::new(source, target, layers_to_move));
688 }
689 }
690
691 None
692 }
693}
694
695#[cfg(test)]
696mod tests {
697 use super::*;
698 use crate::host::AccelerationMode;
699
700 fn sample_layers(count: usize, vram_gb: f32) -> Vec<LayerSpec> {
701 (0..count)
702 .map(|index| LayerSpec {
703 index,
704 vram_gb,
705 num_weights: 0,
706 })
707 .collect()
708 }
709
710 #[test]
711 fn greedily_places_layers_across_nodes() {
712 let nodes = vec![
713 NodeResources::new("node-a", 24.0, 64.0, "8.9", None),
714 NodeResources::new("node-b", 12.0, 32.0, "8.6", None),
715 ];
716
717 let assignments = assign_layers_sequentially(&nodes, &sample_layers(33, 1.0)).unwrap();
718
719 assert_eq!(
720 assignments,
721 vec![
722 LayerAssignment {
723 node_id: "node-a".into(),
724 start_layer: 0,
725 end_layer: 24,
726 used_vram_gb: 24.0,
727 num_layers: 24,
728 },
729 LayerAssignment {
730 node_id: "node-b".into(),
731 start_layer: 24,
732 end_layer: 33,
733 used_vram_gb: 9.0,
734 num_layers: 9,
735 }
736 ]
737 );
738 }
739
740 #[test]
741 fn reports_insufficient_capacity() {
742 let nodes = vec![NodeResources::new("node-a", 2.0, 64.0, "8.9", None)];
743 let error = assign_layers_sequentially(&nodes, &sample_layers(3, 1.0)).unwrap_err();
744
745 assert!(error.contains("insufficient cluster VRAM"));
746 }
747
748 #[test]
749 fn selects_quantization_mode_from_delivery_ratio() {
750 assert_eq!(select_quantization_mode(0.98), QuantizationMode::None);
751 assert_eq!(select_quantization_mode(0.90), QuantizationMode::Int8);
752 assert_eq!(select_quantization_mode(0.75), QuantizationMode::Int4);
753 }
754
755 #[test]
756 fn placement_plan_summary() {
757 let plan = PlacementPlan::new(
758 vec![
759 LayerAssignment::new("node-a".into(), 0, 24, 24.0),
760 LayerAssignment::new("node-b".into(), 24, 33, 9.0),
761 ],
762 QuantizationMode::None,
763 );
764
765 let summary = plan.summary();
766 assert!(summary.contains("Total layers: 33"));
767 assert!(summary.contains("Full Precision"));
768 }
769
770 #[test]
771 fn rebalance_assignments_detects_capacity() {
772 let mut plan = PlacementPlan::new(
773 vec![LayerAssignment::new("node-a".into(), 0, 24, 24.0)],
774 QuantizationMode::None,
775 );
776
777 assert!(!rebalance_assignments(&ClusterState::default(), &mut plan));
779 }
780
781 #[test]
782 fn runtime_profile_chunks_large_assignments() {
783 let assignments = vec![LayerAssignment::new("node-a".into(), 0, 12, 12.0)];
784 let chunked = chunk_assignments_for_workers(assignments, 5);
785
786 assert_eq!(chunked.len(), 3);
787 assert_eq!(chunked[0].start_layer, 0);
788 assert_eq!(chunked[0].end_layer, 5);
789 assert_eq!(chunked[2].start_layer, 10);
790 assert_eq!(chunked[2].end_layer, 12);
791 }
792
793 #[test]
794 fn runtime_profile_assignment_keeps_trailing_zero_vram_layers() {
795 let nodes = vec![NodeResources::new("node-a", 24.0, 64.0, "8.9", None)];
800 let layers = sample_layers(3, 0.0);
801 let profile = RuntimeProfile {
802 node_resources: NodeResources::new("node-a", 24.0, 64.0, "8.9", None),
803 logical_cores: 4,
804 recommended_workers: 1,
805 acceleration_mode: AccelerationMode::Generic,
806 gpu_backend: crate::host::GpuBackend::Cpu,
807 xdp_supported: false,
808 detection_source: String::from("test"),
809 probe_mode: crate::host::ProbeMode::Fast,
810 };
811
812 let assignments = assign_layers_with_runtime_profile(&nodes, &layers, &profile).unwrap();
813 let total: usize = assignments.iter().map(|a| a.num_layers).sum();
814 assert_eq!(total, 3, "all zero-VRAM layers must still be assigned");
815 }
816
817 #[test]
818 fn fault_tolerance_handles_all_failed_nodes_without_panicking() {
819 let cluster = ClusterState::new();
823 cluster.register(NodeResources::new("node-a", 24.0, 64.0, "8.9", None));
824 cluster.mark_failed("node-a");
825
826 let plan = assign_layers_with_fault_tolerance(&cluster, &sample_layers(4, 1.0)).unwrap();
827 assert_eq!(plan.quantization_mode, QuantizationMode::Int4);
828 }
829
830 #[test]
831 fn planning_tuning_scales_chunk_size_with_workers() {
832 let profile = RuntimeProfile {
833 node_resources: NodeResources::new("node-a", 24.0, 64.0, "8.9", None),
834 logical_cores: 16,
835 recommended_workers: 8,
836 acceleration_mode: AccelerationMode::Gpu,
837 gpu_backend: crate::host::GpuBackend::Cuda,
838 xdp_supported: true,
839 detection_source: String::from("test"),
840 probe_mode: crate::host::ProbeMode::Fast,
841 };
842
843 let tuning = PlanningTuning::from_runtime_profile(&profile, 40);
844 assert!(tuning.max_layers_per_assignment <= 4);
845 }
846}