1use std::{
4 collections::{BTreeMap, BTreeSet},
5 sync::Weak,
6 time::Duration,
7};
8
9use eredu_checkpoint::{
10 recipe::{DerivedWeightRecipe, RecipeCatalog, RecipeError},
11 store::{TensorSelection, WeightStoreDiagnostics},
12};
13use eredu_core::residency::{
14 EvictedResidencyCopy, MemoryTier, OffloadPlan, OffloadReport, OffloadUnitId, PrefetchOutcome,
15 ResidencyLedger, ResidencyLedgerError, UnitResidencyReport,
16};
17
18#[derive(Debug, Clone, Default, Eq, PartialEq)]
20pub struct WeightMaterializationReport {
21 pub admitted_working_set_bytes: u64,
23 pub transformed_weights: usize,
25 pub source_tiles: usize,
27 pub peak_in_flight_tiles: usize,
29 pub source_bytes_read: u64,
31 pub output_bytes: u64,
33 pub peak_planned_working_set_bytes: u64,
35 pub largest_source_tile_bytes: u64,
37 pub largest_output_tile_bytes: u64,
39}
40
41impl WeightMaterializationReport {
42 pub fn merge(&mut self, next: Self) {
47 self.admitted_working_set_bytes = self
48 .admitted_working_set_bytes
49 .max(next.admitted_working_set_bytes);
50 self.transformed_weights += next.transformed_weights;
51 self.source_tiles += next.source_tiles;
52 self.peak_in_flight_tiles = self.peak_in_flight_tiles.max(next.peak_in_flight_tiles);
53 self.source_bytes_read += next.source_bytes_read;
54 self.output_bytes += next.output_bytes;
55 self.peak_planned_working_set_bytes = self
56 .peak_planned_working_set_bytes
57 .max(next.peak_planned_working_set_bytes);
58 self.largest_source_tile_bytes = self
59 .largest_source_tile_bytes
60 .max(next.largest_source_tile_bytes);
61 self.largest_output_tile_bytes = self
62 .largest_output_tile_bytes
63 .max(next.largest_output_tile_bytes);
64 }
65}
66
67#[derive(Debug, Clone, Eq, PartialEq)]
69pub struct ResidencyReport {
70 initialized: bool,
71 offload: OffloadReport,
72 units: Vec<UnitResidencyReport>,
73 active_window: Vec<OffloadUnitId>,
74 weight_store: WeightStoreDiagnostics,
75 materialization: Option<WeightMaterializationReport>,
76}
77
78impl ResidencyReport {
79 pub fn new(
81 initialized: bool,
82 offload: OffloadReport,
83 units: Vec<UnitResidencyReport>,
84 active_window: Vec<OffloadUnitId>,
85 weight_store: WeightStoreDiagnostics,
86 ) -> Self {
87 Self {
88 initialized,
89 offload,
90 units,
91 active_window,
92 weight_store,
93 materialization: None,
94 }
95 }
96
97 pub const fn initialized(&self) -> bool {
99 self.initialized
100 }
101
102 pub const fn offload(&self) -> &OffloadReport {
104 &self.offload
105 }
106
107 pub fn units(&self) -> &[UnitResidencyReport] {
109 &self.units
110 }
111
112 pub fn active_window(&self) -> &[OffloadUnitId] {
114 &self.active_window
115 }
116
117 pub const fn weight_store(&self) -> &WeightStoreDiagnostics {
119 &self.weight_store
120 }
121
122 pub const fn materialization(&self) -> Option<&WeightMaterializationReport> {
124 self.materialization.as_ref()
125 }
126
127 pub fn with_materialization(
129 mut self,
130 materialization: Option<WeightMaterializationReport>,
131 ) -> Self {
132 self.materialization = materialization;
133 self
134 }
135}
136
137#[derive(Debug, Clone, Eq, PartialEq)]
139pub struct WeightBinding {
140 name: String,
141 alias_of: Option<String>,
142 logical_target: Option<String>,
143 checkpoint_key: String,
144 selection: TensorSelection,
145 recipe: Option<DerivedWeightRecipe>,
146 quantization_companions: Option<QuantizationCompanionBindings>,
147 expected_bytes: u64,
148}
149
150#[derive(Debug, Clone, Eq, PartialEq)]
152pub struct QuantizationCompanionBindings {
153 scale: String,
154 affine_bias: Option<String>,
155}
156
157impl QuantizationCompanionBindings {
158 pub fn new(
160 scale: impl Into<String>,
161 affine_bias: Option<String>,
162 ) -> Result<Self, ResidencyDeclarationError> {
163 let scale = validate_name(scale.into())?;
164 let affine_bias = affine_bias.map(validate_name).transpose()?;
165 if affine_bias.as_ref() == Some(&scale) {
166 return Err(ResidencyDeclarationError::InvalidQuantizationCompanions {
167 name: "packed weight".into(),
168 scales: scale,
169 biases: affine_bias.unwrap(),
170 });
171 }
172 Ok(Self { scale, affine_bias })
173 }
174
175 pub fn scale(&self) -> &str {
177 &self.scale
178 }
179 pub fn affine_bias(&self) -> Option<&str> {
181 self.affine_bias.as_deref()
182 }
183}
184
185impl WeightBinding {
186 pub fn new(
188 name: impl Into<String>,
189 checkpoint_key: impl Into<String>,
190 selection: TensorSelection,
191 expected_bytes: u64,
192 ) -> Result<Self, ResidencyDeclarationError> {
193 let name = validate_name(name.into())?;
194 let checkpoint_key = checkpoint_key.into();
195 if checkpoint_key.trim().is_empty() {
196 return Err(ResidencyDeclarationError::InvalidCheckpointKey { name });
197 }
198 validate_size(&name, expected_bytes)?;
199 Ok(Self {
200 name,
201 alias_of: None,
202 logical_target: None,
203 checkpoint_key,
204 selection,
205 recipe: None,
206 quantization_companions: None,
207 expected_bytes,
208 })
209 }
210
211 pub fn from_recipe(
213 name: impl Into<String>,
214 recipe: DerivedWeightRecipe,
215 expected_bytes: u64,
216 ) -> Result<Self, ResidencyDeclarationError> {
217 let name = validate_name(name.into())?;
218 validate_size(&name, expected_bytes)?;
219 let checkpoint_key = first_source(&name, &recipe)?;
220 Ok(Self {
221 name,
222 alias_of: None,
223 logical_target: None,
224 checkpoint_key,
225 selection: TensorSelection::Full,
226 recipe: Some(recipe),
227 quantization_companions: None,
228 expected_bytes,
229 })
230 }
231
232 pub fn alias(
235 name: impl Into<String>,
236 owner: impl Into<String>,
237 expected_bytes: u64,
238 ) -> Result<Self, ResidencyDeclarationError> {
239 let name = validate_name(name.into())?;
240 let owner = validate_name(owner.into())?;
241 validate_size(&name, expected_bytes)?;
242 Ok(Self {
243 name,
244 alias_of: Some(owner),
245 logical_target: None,
246 checkpoint_key: String::new(),
247 selection: TensorSelection::Full,
248 recipe: None,
249 quantization_companions: None,
250 expected_bytes,
251 })
252 }
253
254 pub fn name(&self) -> &str {
256 &self.name
257 }
258
259 pub fn alias_of(&self) -> Option<&str> {
261 self.alias_of.as_deref()
262 }
263
264 pub const fn is_alias(&self) -> bool {
267 self.alias_of.is_some()
268 }
269
270 pub fn with_name(mut self, name: impl Into<String>) -> Result<Self, ResidencyDeclarationError> {
272 self.name = validate_name(name.into())?;
273 Ok(self)
274 }
275
276 pub fn logical_target(&self) -> Option<&str> {
278 self.logical_target.as_deref()
279 }
280
281 pub fn with_logical_target(
283 mut self,
284 target: impl Into<String>,
285 ) -> Result<Self, ResidencyDeclarationError> {
286 self.logical_target = Some(validate_name(target.into())?);
287 Ok(self)
288 }
289
290 pub fn with_quantization_companions(
292 mut self,
293 scales_binding: impl Into<String>,
294 biases_binding: Option<String>,
295 ) -> Result<Self, ResidencyDeclarationError> {
296 let scales_binding = validate_name(scales_binding.into())?;
297 let biases_binding = biases_binding.map(validate_name).transpose()?;
298 if scales_binding == self.name
299 || biases_binding.as_ref() == Some(&self.name)
300 || biases_binding.as_ref() == Some(&scales_binding)
301 {
302 return Err(ResidencyDeclarationError::InvalidQuantizationCompanions {
303 name: self.name,
304 scales: scales_binding,
305 biases: biases_binding.unwrap_or_default(),
306 });
307 }
308 self.quantization_companions = Some(QuantizationCompanionBindings {
309 scale: scales_binding,
310 affine_bias: biases_binding,
311 });
312 Ok(self)
313 }
314
315 pub const fn quantization_companions(&self) -> Option<&QuantizationCompanionBindings> {
317 self.quantization_companions.as_ref()
318 }
319
320 pub fn checkpoint_key(&self) -> &str {
322 &self.checkpoint_key
323 }
324
325 pub fn selection(&self) -> &TensorSelection {
327 &self.selection
328 }
329
330 pub const fn recipe(&self) -> Option<&DerivedWeightRecipe> {
332 self.recipe.as_ref()
333 }
334
335 pub fn source_recipe(&self) -> DerivedWeightRecipe {
337 assert!(
338 self.alias_of.is_none(),
339 "logical aliases have no physical recipe"
340 );
341 self.recipe.clone().unwrap_or_else(|| {
342 DerivedWeightRecipe::source(self.checkpoint_key.clone(), self.selection.clone())
343 })
344 }
345
346 pub fn checkpoint_keys(&self) -> Vec<&str> {
348 if self.alias_of.is_some() {
349 return Vec::new();
350 }
351 self.recipe.as_ref().map_or_else(
352 || vec![self.checkpoint_key.as_str()],
353 DerivedWeightRecipe::source_keys,
354 )
355 }
356
357 pub const fn expected_bytes(&self) -> u64 {
359 self.expected_bytes
360 }
361
362 pub fn with_source_recipe(
364 mut self,
365 recipe: DerivedWeightRecipe,
366 expected_bytes: u64,
367 ) -> Result<Self, ResidencyDeclarationError> {
368 if self.alias_of.is_some() {
369 return Err(ResidencyDeclarationError::AliasHasPhysicalSource { name: self.name });
370 }
371 validate_size(&self.name, expected_bytes)?;
372 self.checkpoint_key = first_source(&self.name, &recipe)?;
373 self.selection = TensorSelection::Full;
374 self.recipe = Some(recipe);
375 self.expected_bytes = expected_bytes;
376 Ok(self)
377 }
378
379 pub fn select_bounded_output<C: RecipeCatalog + ?Sized>(
381 self,
382 catalog: &C,
383 selection: TensorSelection,
384 ) -> Result<Self, WeightBindingSelectionError> {
385 let recipe = self.source_recipe().select_bounded(catalog, selection)?;
386 let bytes = recipe.infer(catalog)?.byte_len();
387 Ok(self.with_source_recipe(recipe, bytes)?)
388 }
389}
390
391#[derive(Debug)]
393pub struct WeightBindingPlan<'a> {
394 owners: Vec<&'a WeightBinding>,
395 aliases: Vec<(&'a WeightBinding, &'a WeightBinding)>,
396}
397
398impl<'a> WeightBindingPlan<'a> {
399 pub fn new(bindings: &'a [WeightBinding]) -> Result<Self, ResidencyDeclarationError> {
401 let by_name = bindings
402 .iter()
403 .map(|binding| (binding.name(), binding))
404 .collect::<BTreeMap<_, _>>();
405 if by_name.len() != bindings.len() {
406 let duplicate = bindings
407 .iter()
408 .map(WeightBinding::name)
409 .find(|name| bindings.iter().filter(|item| item.name() == *name).count() > 1)
410 .unwrap_or("<unknown>");
411 return Err(ResidencyDeclarationError::DuplicateLogicalBinding {
412 name: duplicate.to_owned(),
413 });
414 }
415
416 fn resolve<'a>(
417 binding: &'a WeightBinding,
418 by_name: &BTreeMap<&str, &'a WeightBinding>,
419 visiting: &mut BTreeSet<String>,
420 ) -> Result<&'a WeightBinding, ResidencyDeclarationError> {
421 let Some(owner_name) = binding.alias_of() else {
422 return Ok(binding);
423 };
424 if !visiting.insert(binding.name().to_owned()) {
425 return Err(ResidencyDeclarationError::BindingAliasCycle {
426 name: binding.name().to_owned(),
427 });
428 }
429 let owner = by_name.get(owner_name).copied().ok_or_else(|| {
430 ResidencyDeclarationError::UnknownBindingAliasOwner {
431 alias: binding.name().to_owned(),
432 owner: owner_name.to_owned(),
433 }
434 })?;
435 let resolved = resolve(owner, by_name, visiting)?;
436 visiting.remove(binding.name());
437 Ok(resolved)
438 }
439
440 let owners = bindings
441 .iter()
442 .filter(|binding| !binding.is_alias())
443 .collect::<Vec<_>>();
444 let mut aliases = Vec::new();
445 for alias in bindings.iter().filter(|binding| binding.is_alias()) {
446 let owner = resolve(alias, &by_name, &mut BTreeSet::new())?;
447 if alias.expected_bytes() != owner.expected_bytes() {
448 return Err(ResidencyDeclarationError::BindingAliasByteMismatch {
449 alias: alias.name().to_owned(),
450 owner: owner.name().to_owned(),
451 alias_bytes: alias.expected_bytes(),
452 owner_bytes: owner.expected_bytes(),
453 });
454 }
455 aliases.push((alias, owner));
456 }
457 Ok(Self { owners, aliases })
458 }
459
460 pub fn owners(&self) -> impl Iterator<Item = &'a WeightBinding> + '_ {
462 self.owners.iter().copied()
463 }
464
465 pub fn aliases(&self) -> impl Iterator<Item = (&'a WeightBinding, &'a WeightBinding)> + '_ {
467 self.aliases.iter().copied()
468 }
469}
470
471#[derive(Debug, thiserror::Error)]
473pub enum WeightBindingSelectionError {
474 #[error(transparent)]
476 Recipe(#[from] RecipeError),
477 #[error(transparent)]
479 Declaration(#[from] ResidencyDeclarationError),
480}
481
482#[derive(Debug, Clone, Eq, PartialEq)]
484pub struct OffloadUnit {
485 id: OffloadUnitId,
486 bindings: Vec<WeightBinding>,
487}
488
489#[derive(Debug)]
496pub struct ResidencyController {
497 ledger: ResidencyLedger,
498 units: BTreeMap<OffloadUnitId, OffloadUnit>,
499 alias_owners: BTreeMap<(OffloadUnitId, String), (OffloadUnitId, String)>,
500}
501
502#[derive(Debug, Clone, Eq, PartialEq)]
504pub struct ResidencyAcquisition {
505 ids: Vec<OffloadUnitId>,
506 missing: Vec<bool>,
507}
508
509pub trait ResidencyLeaseStorage {
511 type DeviceValue;
513 type HostValue;
515 type Error;
517 type BindingNames<'a>: Iterator<Item = &'a str>
519 where
520 Self: 'a;
521
522 fn device_value<'a>(
524 &'a self,
525 id: &OffloadUnitId,
526 name: &str,
527 ) -> Result<&'a Self::DeviceValue, Self::Error>;
528
529 fn host_value<'a>(
531 &'a self,
532 id: &OffloadUnitId,
533 name: &str,
534 ) -> Result<&'a Self::HostValue, Self::Error>;
535
536 fn binding_names(&self) -> Self::BindingNames<'_>;
538}
539
540pub trait ResidencyLeaseOwner: Sized {
542 fn release_residency_pin(&self, id: &OffloadUnitId, tier: MemoryTier);
544}
545
546pub trait ResidencyTransferOwner<C, R>: Sized {
548 type Executor: ?Sized;
550 type Error;
552
553 fn order_after(
555 completion: &C,
556 executor: &Self::Executor,
557 id: &OffloadUnitId,
558 ) -> Result<(), Self::Error>;
559
560 fn is_complete(completion: &C, id: &OffloadUnitId) -> Result<bool, Self::Error>;
562
563 fn wait(completion: &C, id: &OffloadUnitId) -> Result<(), Self::Error>;
565
566 fn finish_resources(resources: R, succeeded: bool);
568
569 fn resolve_transfer(
571 &self,
572 ids: &[OffloadUnitId],
573 tier: MemoryTier,
574 generation: u64,
575 succeeded: bool,
576 ) -> Result<(), Self::Error>;
577}
578
579pub struct ResidencyTransfer<L, C, R, O>
581where
582 O: ResidencyTransferOwner<C, R>,
583{
584 leases: Vec<L>,
585 completion: Option<C>,
586 resources: Option<R>,
587 owner: Weak<O>,
588 ids: Vec<OffloadUnitId>,
589 tier: MemoryTier,
590 generation: u64,
591}
592
593impl<L, C, R, O> ResidencyTransfer<L, C, R, O>
594where
595 O: ResidencyTransferOwner<C, R>,
596{
597 pub fn immediate(leases: Vec<L>, tier: MemoryTier) -> Self {
599 Self {
600 leases,
601 completion: None,
602 resources: None,
603 owner: Weak::new(),
604 ids: Vec::new(),
605 tier,
606 generation: 0,
607 }
608 }
609
610 pub fn submitted(
612 leases: Vec<L>,
613 completion: C,
614 resources: R,
615 owner: Weak<O>,
616 ids: Vec<OffloadUnitId>,
617 tier: MemoryTier,
618 generation: u64,
619 ) -> Self {
620 assert!(
621 !ids.is_empty(),
622 "an in-flight residency transfer must contain at least one unit"
623 );
624 Self {
625 leases,
626 completion: Some(completion),
627 resources: Some(resources),
628 owner,
629 ids,
630 tier,
631 generation,
632 }
633 }
634
635 pub fn leases(&self) -> &[L] {
637 &self.leases
638 }
639
640 pub fn is_empty(&self) -> bool {
642 self.leases.is_empty()
643 }
644
645 pub fn order_after(&self, executor: &O::Executor) -> Result<(), O::Error> {
647 match &self.completion {
648 Some(completion) => O::order_after(completion, executor, self.primary_id()),
649 None => Ok(()),
650 }
651 }
652
653 pub fn is_complete(&self) -> Result<bool, O::Error> {
655 match &self.completion {
656 Some(completion) => O::is_complete(completion, self.primary_id()),
657 None => Ok(true),
658 }
659 }
660
661 pub fn synchronize(&mut self) -> Result<(), O::Error> {
663 self.finish(true)
664 }
665
666 fn primary_id(&self) -> &OffloadUnitId {
667 self.ids
668 .first()
669 .expect("an in-flight residency transfer has at least one unit")
670 }
671
672 fn finish(&mut self, report_error: bool) -> Result<(), O::Error> {
673 let result = match &self.completion {
674 Some(completion) => O::wait(completion, self.primary_id()),
675 None => Ok(()),
676 };
677 let succeeded = result.is_ok();
678 if let Some(resources) = self.resources.take() {
679 O::finish_resources(resources, succeeded);
680 }
681 if succeeded {
682 self.completion = None;
683 }
684 self.publish(succeeded)?;
685 match result {
686 Ok(()) => Ok(()),
687 Err(error) if report_error => Err(error),
688 Err(_) => Ok(()),
689 }
690 }
691
692 fn publish(&mut self, succeeded: bool) -> Result<(), O::Error> {
693 if self.generation == 0 {
694 return Ok(());
695 }
696 let generation = std::mem::take(&mut self.generation);
697 if let Some(owner) = self.owner.upgrade() {
698 owner.resolve_transfer(&self.ids, self.tier, generation, succeeded)?;
699 }
700 Ok(())
701 }
702}
703
704impl<L, C, R, O> Drop for ResidencyTransfer<L, C, R, O>
705where
706 O: ResidencyTransferOwner<C, R>,
707{
708 fn drop(&mut self) {
709 let _ = self.finish(false);
710 }
711}
712
713pub struct ResidencyLease<S, O>
715where
716 S: ResidencyLeaseStorage,
717 O: ResidencyLeaseOwner,
718{
719 id: OffloadUnitId,
720 tier: MemoryTier,
721 storage: S,
722 owner: Weak<O>,
723}
724
725impl<S, O> ResidencyLease<S, O>
726where
727 S: ResidencyLeaseStorage,
728 O: ResidencyLeaseOwner,
729{
730 pub fn new(id: OffloadUnitId, tier: MemoryTier, storage: S, owner: Weak<O>) -> Self {
732 Self {
733 id,
734 tier,
735 storage,
736 owner,
737 }
738 }
739
740 pub fn id(&self) -> &OffloadUnitId {
742 &self.id
743 }
744
745 pub const fn tier(&self) -> MemoryTier {
747 self.tier
748 }
749
750 pub fn device_value(&self, name: &str) -> Result<&S::DeviceValue, S::Error> {
752 self.storage.device_value(&self.id, name)
753 }
754
755 pub fn host_value(&self, name: &str) -> Result<&S::HostValue, S::Error> {
757 self.storage.host_value(&self.id, name)
758 }
759
760 pub fn binding_names(&self) -> S::BindingNames<'_> {
762 self.storage.binding_names()
763 }
764}
765
766impl<S, O> Drop for ResidencyLease<S, O>
767where
768 S: ResidencyLeaseStorage,
769 O: ResidencyLeaseOwner,
770{
771 fn drop(&mut self) {
772 if let Some(owner) = self.owner.upgrade() {
773 owner.release_residency_pin(&self.id, self.tier);
774 }
775 }
776}
777
778impl ResidencyAcquisition {
779 pub fn ids(&self) -> &[OffloadUnitId] {
781 &self.ids
782 }
783
784 pub fn missing(&self) -> &[bool] {
786 &self.missing
787 }
788
789 pub fn is_hit(&self) -> bool {
791 self.missing.iter().all(|missing| !missing)
792 }
793
794 pub fn missing_ids(&self) -> impl Iterator<Item = &OffloadUnitId> {
796 self.ids
797 .iter()
798 .zip(&self.missing)
799 .filter_map(|(id, &missing)| missing.then_some(id))
800 }
801
802 fn temporary_protection(&self) -> BTreeSet<OffloadUnitId> {
803 self.ids.iter().cloned().collect()
804 }
805}
806
807impl ResidencyController {
808 pub fn new<C: RecipeCatalog + ?Sized>(
810 catalog: &C,
811 plan: OffloadPlan,
812 units: impl IntoIterator<Item = OffloadUnit>,
813 ) -> Result<Self, ResidencyControllerError> {
814 let mut definitions = BTreeMap::new();
815 for unit in units {
816 let id = unit.id().clone();
817 if definitions.insert(id.clone(), unit).is_some() {
818 return Err(ResidencyControllerError::DuplicateUnitDefinition { id });
819 }
820 }
821 for spec in plan.units() {
822 if !definitions.contains_key(spec.id()) {
823 return Err(ResidencyControllerError::MissingUnitDefinition {
824 id: spec.id().clone(),
825 });
826 }
827 }
828 if let Some(id) = definitions
829 .keys()
830 .find(|id| plan.unit(id).is_none())
831 .cloned()
832 {
833 return Err(ResidencyControllerError::UnexpectedUnitDefinition { id });
834 }
835
836 let alias_owners = validate_global_binding_aliases(&definitions)?;
837
838 for spec in plan.units() {
839 let unit = definitions
840 .get(spec.id())
841 .expect("definition identity validated above");
842 let mut total = 0u64;
843 for binding in unit.bindings().iter().filter(|binding| !binding.is_alias()) {
844 total = total.checked_add(binding.expected_bytes()).ok_or(
845 ResidencyControllerError::ArithmeticOverflow {
846 context: "unit binding byte total",
847 },
848 )?;
849 if !binding.is_alias() {
850 let actual = binding
851 .source_recipe()
852 .infer(catalog)
853 .map_err(|source| ResidencyControllerError::Recipe {
854 binding: binding.name().to_owned(),
855 source,
856 })?
857 .byte_len();
858 if actual != binding.expected_bytes() {
859 return Err(ResidencyControllerError::BindingByteMismatch {
860 id: unit.id().clone(),
861 binding: binding.name().to_owned(),
862 expected_bytes: binding.expected_bytes(),
863 actual_bytes: actual,
864 });
865 }
866 }
867 }
868 if total != spec.bytes() {
869 return Err(ResidencyControllerError::UnitByteMismatch {
870 id: unit.id().clone(),
871 planned_bytes: spec.bytes(),
872 actual_bytes: total,
873 });
874 }
875 }
876
877 Ok(Self {
878 ledger: ResidencyLedger::new(plan),
879 units: definitions,
880 alias_owners,
881 })
882 }
883
884 pub fn unit(&self, id: &OffloadUnitId) -> Option<&OffloadUnit> {
886 self.units.get(id)
887 }
888
889 pub fn units(&self) -> impl ExactSizeIterator<Item = &OffloadUnit> {
891 self.units.values()
892 }
893
894 pub fn binding_owner(
896 &self,
897 unit: &OffloadUnitId,
898 binding: &WeightBinding,
899 ) -> Option<(&OffloadUnitId, &WeightBinding)> {
900 let (owner_unit, owner_name) = self
901 .alias_owners
902 .get(&(unit.clone(), binding.name().to_owned()))?;
903 let owner = self.units.get(owner_unit)?;
904 let binding = owner
905 .bindings()
906 .iter()
907 .find(|binding| binding.name() == owner_name)?;
908 Some((owner_unit, binding))
909 }
910
911 pub fn shared_binding_owner(
914 &self,
915 unit: &OffloadUnitId,
916 binding: &WeightBinding,
917 ) -> Option<(&OffloadUnitId, &WeightBinding)> {
918 if binding.is_alias() {
919 return self.binding_owner(unit, binding);
920 }
921 let location = (unit.clone(), binding.name().to_owned());
922 if !self.alias_owners.values().any(|owner| owner == &location) {
923 return None;
924 }
925 let (owner_unit, owner) = self.units.get_key_value(unit)?;
926 let owner = owner
927 .bindings()
928 .iter()
929 .find(|candidate| candidate.name() == binding.name())?;
930 Some((owner_unit, owner))
931 }
932
933 pub const fn ledger(&self) -> &ResidencyLedger {
935 &self.ledger
936 }
937
938 pub fn ledger_mut(&mut self) -> &mut ResidencyLedger {
940 &mut self.ledger
941 }
942
943 pub fn plan_acquisition(
945 &mut self,
946 ids: &[OffloadUnitId],
947 tier: MemoryTier,
948 ) -> Result<ResidencyAcquisition, ResidencyLedgerError> {
949 self.ledger.require_initialized()?;
950 self.plan_acquisition_inner(ids, tier)
951 }
952
953 pub fn plan_initialization_acquisition(
955 &mut self,
956 ids: &[OffloadUnitId],
957 tier: MemoryTier,
958 ) -> Result<ResidencyAcquisition, ResidencyLedgerError> {
959 self.plan_acquisition_inner(ids, tier)
960 }
961
962 fn plan_acquisition_inner(
963 &mut self,
964 ids: &[OffloadUnitId],
965 tier: MemoryTier,
966 ) -> Result<ResidencyAcquisition, ResidencyLedgerError> {
967 self.ledger.validate_batch(ids, tier)?;
968 let missing = ids
969 .iter()
970 .map(|id| self.ledger.is_resident(id, tier).map(|resident| !resident))
971 .collect::<Result<Vec<_>, _>>()?;
972 Ok(ResidencyAcquisition {
973 ids: ids.to_vec(),
974 missing,
975 })
976 }
977
978 pub fn reserve_acquisition(
980 &mut self,
981 acquisition: &ResidencyAcquisition,
982 reservations: &[(OffloadUnitId, u64)],
983 tier: MemoryTier,
984 ) -> Result<Vec<EvictedResidencyCopy>, ResidencyLedgerError> {
985 self.ledger
986 .reserve_copies(reservations, tier, &acquisition.temporary_protection())
987 }
988
989 pub fn touch_acquisition_hits(
991 &mut self,
992 acquisition: &ResidencyAcquisition,
993 tier: MemoryTier,
994 ) -> Result<(), ResidencyLedgerError> {
995 for (id, &missing) in acquisition.ids.iter().zip(&acquisition.missing) {
996 if !missing {
997 self.ledger.touch(id, tier)?;
998 }
999 }
1000 Ok(())
1001 }
1002
1003 pub fn rollback_acquisition(
1005 &mut self,
1006 acquisition: &ResidencyAcquisition,
1007 tier: MemoryTier,
1008 ) -> Result<(), ResidencyLedgerError> {
1009 for id in acquisition.missing_ids() {
1010 self.ledger.rollback_reserved(id, tier)?;
1011 }
1012 Ok(())
1013 }
1014
1015 #[allow(clippy::too_many_arguments)]
1017 pub fn publish_acquisition_copy(
1018 &mut self,
1019 id: &OffloadUnitId,
1020 tier: MemoryTier,
1021 actual_bytes: u64,
1022 transferred_bytes: u64,
1023 transfer_generation: Option<u64>,
1024 direction: eredu_core::residency::TransferDirection,
1025 duration: Duration,
1026 ) -> Result<(), ResidencyLedgerError> {
1027 self.ledger
1028 .publish_reserved(id, tier, actual_bytes, transfer_generation)?;
1029 self.ledger
1030 .record_transfer(direction, transferred_bytes, duration);
1031 Ok(())
1032 }
1033
1034 pub fn begin_prefetch(
1036 &mut self,
1037 id: &OffloadUnitId,
1038 tier: MemoryTier,
1039 ) -> Result<PrefetchOutcome, ResidencyLedgerError> {
1040 self.ledger.require_initialized()?;
1041 let outcome = if self.ledger.is_resident(id, tier)? {
1042 PrefetchOutcome::Hit
1043 } else {
1044 PrefetchOutcome::Miss
1045 };
1046 self.ledger.record_prefetch(tier, outcome);
1047 Ok(outcome)
1048 }
1049
1050 pub fn resolve_transfer(
1052 &mut self,
1053 ids: &[OffloadUnitId],
1054 tier: MemoryTier,
1055 generation: u64,
1056 succeeded: bool,
1057 ) -> Result<Vec<EvictedResidencyCopy>, ResidencyLedgerError> {
1058 self.ledger
1059 .resolve_transfer(ids, tier, generation, succeeded)
1060 }
1061
1062 pub fn commit_group_window(
1067 &mut self,
1068 group: &str,
1069 active: &[OffloadUnitId],
1070 upcoming: &[OffloadUnitId],
1071 tier: MemoryTier,
1072 ) -> Result<Vec<OffloadUnitId>, eredu_core::residency::ResidencyLedgerError> {
1073 self.ledger.require_initialized()?;
1074 for id in active.iter().chain(upcoming) {
1075 self.ledger.spec(id)?;
1076 }
1077 self.ledger.set_group_window(group, active, tier)?;
1078 let depth = self.ledger.plan().config().prefetch_depth();
1079 let mut seen = BTreeSet::new();
1080 Ok(upcoming
1081 .iter()
1082 .filter(|id| seen.insert((*id).clone()))
1083 .take(depth)
1084 .cloned()
1085 .collect())
1086 }
1087
1088 pub fn protect_group_window(
1090 &mut self,
1091 group: &str,
1092 active: &[OffloadUnitId],
1093 tier: MemoryTier,
1094 ) -> Result<(), eredu_core::residency::ResidencyLedgerError> {
1095 self.commit_group_window(group, active, &[], tier)
1096 .map(|_| ())
1097 }
1098}
1099
1100type BindingLocation = (OffloadUnitId, String);
1101type BindingAliasMap = BTreeMap<BindingLocation, BindingLocation>;
1102
1103fn validate_global_binding_aliases(
1104 units: &BTreeMap<OffloadUnitId, OffloadUnit>,
1105) -> Result<BindingAliasMap, ResidencyControllerError> {
1106 let mut identities = BTreeMap::<String, Vec<BindingLocation>>::new();
1107 let mut aliases = BTreeMap::<BindingLocation, String>::new();
1108 let mut bytes = BTreeMap::<BindingLocation, u64>::new();
1109 for (unit_id, unit) in units {
1110 for binding in unit.bindings() {
1111 let location = (unit_id.clone(), binding.name().to_owned());
1112 let identity = binding
1113 .logical_target()
1114 .unwrap_or(binding.name())
1115 .to_owned();
1116 identities
1117 .entry(identity)
1118 .or_default()
1119 .push(location.clone());
1120 bytes.insert(location.clone(), binding.expected_bytes());
1121 if let Some(owner) = binding.alias_of() {
1122 aliases.insert(location, owner.to_owned());
1123 }
1124 }
1125 }
1126
1127 fn resolve(
1128 location: &BindingLocation,
1129 identities: &BTreeMap<String, Vec<BindingLocation>>,
1130 aliases: &BTreeMap<BindingLocation, String>,
1131 visiting: &mut BTreeSet<BindingLocation>,
1132 ) -> Result<BindingLocation, ResidencyControllerError> {
1133 let Some(destination) = aliases.get(location) else {
1134 return Ok(location.clone());
1135 };
1136 if !visiting.insert(location.clone()) {
1137 return Err(ResidencyControllerError::Declaration(
1138 ResidencyDeclarationError::BindingAliasCycle {
1139 name: location.1.clone(),
1140 },
1141 ));
1142 }
1143 let candidates = identities.get(destination).ok_or_else(|| {
1144 ResidencyControllerError::Declaration(
1145 ResidencyDeclarationError::UnknownBindingAliasOwner {
1146 alias: location.1.clone(),
1147 owner: destination.clone(),
1148 },
1149 )
1150 })?;
1151 if candidates.len() != 1 {
1152 return Err(ResidencyControllerError::Declaration(
1153 ResidencyDeclarationError::AmbiguousBindingAliasOwner {
1154 alias: location.1.clone(),
1155 owner: destination.clone(),
1156 },
1157 ));
1158 }
1159 let owner = resolve(&candidates[0], identities, aliases, visiting)?;
1160 visiting.remove(location);
1161 Ok(owner)
1162 }
1163
1164 let mut resolved = BTreeMap::new();
1165 for alias in aliases.keys() {
1166 let owner = resolve(alias, &identities, &aliases, &mut BTreeSet::new())?;
1167 let alias_bytes = bytes[alias];
1168 let owner_bytes = bytes[&owner];
1169 if alias_bytes != owner_bytes {
1170 return Err(ResidencyControllerError::Declaration(
1171 ResidencyDeclarationError::BindingAliasByteMismatch {
1172 alias: alias.1.clone(),
1173 owner: owner.1.clone(),
1174 alias_bytes,
1175 owner_bytes,
1176 },
1177 ));
1178 }
1179 resolved.insert(alias.clone(), owner);
1180 }
1181 Ok(resolved)
1182}
1183
1184#[derive(Debug, thiserror::Error)]
1186pub enum ResidencyControllerError {
1187 #[error(transparent)]
1189 Declaration(#[from] ResidencyDeclarationError),
1190 #[error("duplicate residency unit definition: {id}")]
1192 DuplicateUnitDefinition {
1193 id: OffloadUnitId,
1195 },
1196 #[error("offload plan unit {id} has no residency unit definition")]
1198 MissingUnitDefinition {
1199 id: OffloadUnitId,
1201 },
1202 #[error("residency unit {id} is absent from the offload plan")]
1204 UnexpectedUnitDefinition {
1205 id: OffloadUnitId,
1207 },
1208 #[error(
1210 "residency unit {id} defines {actual_bytes} bytes but its plan reserves {planned_bytes}"
1211 )]
1212 UnitByteMismatch {
1213 id: OffloadUnitId,
1215 planned_bytes: u64,
1217 actual_bytes: u64,
1219 },
1220 #[error(
1222 "binding {binding:?} in unit {id} selects {actual_bytes} bytes but declares {expected_bytes}"
1223 )]
1224 BindingByteMismatch {
1225 id: OffloadUnitId,
1227 binding: String,
1229 expected_bytes: u64,
1231 actual_bytes: u64,
1233 },
1234 #[error("derived-weight recipe for binding {binding:?} failed: {source}")]
1236 Recipe {
1237 binding: String,
1239 #[source]
1241 source: RecipeError,
1242 },
1243 #[error("residency arithmetic overflow: {context}")]
1245 ArithmeticOverflow {
1246 context: &'static str,
1248 },
1249}
1250
1251pub trait ResidencyWindowManager {
1253 type Error: std::error::Error + From<ResidencyWindowError>;
1255
1256 fn prepare_window(
1258 &self,
1259 active: &[OffloadUnitId],
1260 upcoming: &[OffloadUnitId],
1261 tier: MemoryTier,
1262 ) -> Result<Vec<(OffloadUnitId, PrefetchOutcome)>, Self::Error>;
1263
1264 fn prepare_group_window(
1266 &self,
1267 group: &str,
1268 active: &[OffloadUnitId],
1269 upcoming: &[OffloadUnitId],
1270 tier: MemoryTier,
1271 ) -> Result<Vec<(OffloadUnitId, PrefetchOutcome)>, Self::Error>;
1272
1273 fn evict(&self, id: &OffloadUnitId, tier: MemoryTier) -> Result<bool, Self::Error>;
1275
1276 fn unit_reports(&self) -> Result<Vec<UnitResidencyReport>, Self::Error>;
1278}
1279
1280#[derive(Debug, Clone)]
1282pub struct DeviceLayerWindow {
1283 units: Vec<OffloadUnitId>,
1284 depth: usize,
1285}
1286
1287impl DeviceLayerWindow {
1288 pub fn new(
1290 units: impl IntoIterator<Item = OffloadUnitId>,
1291 depth: usize,
1292 ) -> Result<Self, ResidencyWindowError> {
1293 let units = units.into_iter().collect::<Vec<_>>();
1294 if units.is_empty() {
1295 return Err(ResidencyWindowError::EmptyLayerWindow);
1296 }
1297 if depth == 0 || depth > units.len() {
1298 return Err(ResidencyWindowError::OversizedLayerWindow {
1299 depth,
1300 layer_count: units.len(),
1301 });
1302 }
1303 let unique = units.iter().collect::<BTreeSet<_>>();
1304 if unique.len() != units.len() {
1305 return Err(ResidencyWindowError::DuplicateLayerWindowUnit {
1306 id: units
1307 .iter()
1308 .find(|id| units.iter().filter(|candidate| *candidate == *id).count() > 1)
1309 .expect("duplicate exists")
1310 .clone(),
1311 });
1312 }
1313 Ok(Self { units, depth })
1314 }
1315
1316 pub const fn depth(&self) -> usize {
1318 self.depth
1319 }
1320
1321 pub fn units(&self) -> &[OffloadUnitId] {
1323 &self.units
1324 }
1325
1326 pub fn desired(&self, current: usize) -> Result<&[OffloadUnitId], ResidencyWindowError> {
1328 if current >= self.units.len() {
1329 return Err(ResidencyWindowError::InvalidLayerIndex {
1330 index: current,
1331 layer_count: self.units.len(),
1332 });
1333 }
1334 let end = current.saturating_add(self.depth).min(self.units.len());
1335 Ok(&self.units[current..end])
1336 }
1337
1338 pub fn prepare<M: ResidencyWindowManager>(
1340 &self,
1341 manager: &M,
1342 current: usize,
1343 ) -> Result<Vec<(OffloadUnitId, PrefetchOutcome)>, M::Error> {
1344 let desired = self.desired(current).map_err(M::Error::from)?;
1345 let outcomes = manager.prepare_window(desired, desired, MemoryTier::Device)?;
1346 self.trim_to(manager, desired)?;
1347 Ok(outcomes)
1348 }
1349
1350 pub fn trim_to<M: ResidencyWindowManager>(
1352 &self,
1353 manager: &M,
1354 desired: &[OffloadUnitId],
1355 ) -> Result<(), M::Error> {
1356 let desired = desired.iter().collect::<BTreeSet<_>>();
1357 for id in &self.units {
1358 if !desired.contains(id) {
1359 manager.evict(id, MemoryTier::Device)?;
1360 }
1361 }
1362 Ok(())
1363 }
1364
1365 pub fn clear<M: ResidencyWindowManager>(&self, manager: &M) -> Result<(), M::Error> {
1367 manager.prepare_window(&[], &[], MemoryTier::Device)?;
1368 self.trim_to(manager, &[])
1369 }
1370}
1371
1372#[derive(Debug, Clone)]
1374pub struct ResidentLayerGroup {
1375 id: String,
1376 window: DeviceLayerWindow,
1377}
1378
1379impl ResidentLayerGroup {
1380 pub fn new(
1382 id: impl Into<String>,
1383 units: impl IntoIterator<Item = OffloadUnitId>,
1384 depth: usize,
1385 ) -> Result<Self, ResidencyWindowError> {
1386 let id = id.into();
1387 if id.trim().is_empty() {
1388 return Err(ResidencyWindowError::InvalidGroupId);
1389 }
1390 Ok(Self {
1391 id,
1392 window: DeviceLayerWindow::new(units, depth)?,
1393 })
1394 }
1395
1396 pub fn id(&self) -> &str {
1398 &self.id
1399 }
1400
1401 pub fn units(&self) -> &[OffloadUnitId] {
1403 self.window.units()
1404 }
1405
1406 pub const fn depth(&self) -> usize {
1408 self.window.depth()
1409 }
1410
1411 pub fn prepare<M: ResidencyWindowManager>(
1413 &self,
1414 manager: &M,
1415 current: usize,
1416 ) -> Result<Vec<(OffloadUnitId, PrefetchOutcome)>, M::Error> {
1417 let desired = self.window.desired(current).map_err(M::Error::from)?;
1418 let outcomes =
1419 manager.prepare_group_window(&self.id, desired, desired, MemoryTier::Device)?;
1420 self.window.trim_to(manager, desired)?;
1421 Ok(outcomes)
1422 }
1423
1424 pub fn trim_to<M: ResidencyWindowManager>(
1426 &self,
1427 manager: &M,
1428 desired: &[OffloadUnitId],
1429 ) -> Result<(), M::Error> {
1430 self.window.trim_to(manager, desired)
1431 }
1432
1433 pub fn clear<M: ResidencyWindowManager>(&self, manager: &M) -> Result<(), M::Error> {
1435 manager.prepare_group_window(&self.id, &[], &[], MemoryTier::Device)?;
1436 self.window.trim_to(manager, &[])
1437 }
1438
1439 pub fn report<M: ResidencyWindowManager>(
1441 &self,
1442 manager: &M,
1443 ) -> Result<ResidentLayerGroupReport, M::Error> {
1444 let ids = self.units().iter().collect::<BTreeSet<_>>();
1445 let mut host_bytes = 0u64;
1446 let mut device_bytes = 0u64;
1447 let mut device_units = 0usize;
1448 for unit in manager
1449 .unit_reports()?
1450 .iter()
1451 .filter(|unit| ids.contains(unit.id()))
1452 {
1453 if unit.host_resident() {
1454 host_bytes = host_bytes
1455 .checked_add(unit.host_allocated_bytes())
1456 .ok_or(ResidencyWindowError::ArithmeticOverflow {
1457 context: "execution group host bytes",
1458 })
1459 .map_err(M::Error::from)?;
1460 }
1461 if unit.device_resident() {
1462 device_bytes = device_bytes
1463 .checked_add(unit.device_allocated_bytes())
1464 .ok_or(ResidencyWindowError::ArithmeticOverflow {
1465 context: "execution group device bytes",
1466 })
1467 .map_err(M::Error::from)?;
1468 device_units += 1;
1469 }
1470 }
1471 Ok(ResidentLayerGroupReport {
1472 id: self.id.clone(),
1473 ordered_units: self.units().len(),
1474 window_depth: self.depth(),
1475 host_bytes,
1476 device_bytes,
1477 device_units,
1478 })
1479 }
1480}
1481
1482#[derive(Debug, Clone, Eq, PartialEq)]
1484pub struct ResidentLayerGroupReport {
1485 id: String,
1486 ordered_units: usize,
1487 window_depth: usize,
1488 host_bytes: u64,
1489 device_bytes: u64,
1490 device_units: usize,
1491}
1492
1493impl ResidentLayerGroupReport {
1494 pub fn id(&self) -> &str {
1496 &self.id
1497 }
1498 pub const fn ordered_units(&self) -> usize {
1500 self.ordered_units
1501 }
1502 pub const fn window_depth(&self) -> usize {
1504 self.window_depth
1505 }
1506 pub const fn host_bytes(&self) -> u64 {
1508 self.host_bytes
1509 }
1510 pub const fn device_bytes(&self) -> u64 {
1512 self.device_bytes
1513 }
1514 pub const fn device_units(&self) -> usize {
1516 self.device_units
1517 }
1518}
1519
1520#[derive(Debug, Clone, Eq, PartialEq, thiserror::Error)]
1522pub enum ResidencyWindowError {
1523 #[error("device layer window requires at least one ordered unit")]
1525 EmptyLayerWindow,
1526 #[error("device layer window depth {depth} exceeds {layer_count} ordered units")]
1528 OversizedLayerWindow {
1529 depth: usize,
1531 layer_count: usize,
1533 },
1534 #[error("device layer index {index} is outside {layer_count} ordered units")]
1536 InvalidLayerIndex {
1537 index: usize,
1539 layer_count: usize,
1541 },
1542 #[error("device layer window contains duplicate unit {id}")]
1544 DuplicateLayerWindowUnit {
1545 id: OffloadUnitId,
1547 },
1548 #[error("residency window group identifiers must not be empty")]
1550 InvalidGroupId,
1551 #[error("residency window arithmetic overflow: {context}")]
1553 ArithmeticOverflow {
1554 context: &'static str,
1556 },
1557}
1558
1559impl OffloadUnit {
1560 pub fn new(
1562 id: OffloadUnitId,
1563 bindings: impl IntoIterator<Item = WeightBinding>,
1564 ) -> Result<Self, ResidencyDeclarationError> {
1565 let mut bindings = bindings.into_iter().collect::<Vec<_>>();
1566 if bindings.is_empty() {
1567 return Err(ResidencyDeclarationError::EmptyUnit { id });
1568 }
1569 bindings.sort_by(|left, right| left.name.cmp(&right.name));
1570 if let Some(pair) = bindings
1571 .windows(2)
1572 .find(|pair| pair[0].name == pair[1].name)
1573 {
1574 return Err(ResidencyDeclarationError::DuplicateBindingName {
1575 id,
1576 name: pair[0].name.clone(),
1577 });
1578 }
1579 Ok(Self { id, bindings })
1580 }
1581
1582 pub fn id(&self) -> &OffloadUnitId {
1584 &self.id
1585 }
1586
1587 pub fn bindings(&self) -> &[WeightBinding] {
1589 &self.bindings
1590 }
1591}
1592
1593fn validate_name(name: String) -> Result<String, ResidencyDeclarationError> {
1594 if name.trim().is_empty() {
1595 Err(ResidencyDeclarationError::InvalidBindingName)
1596 } else {
1597 Ok(name)
1598 }
1599}
1600
1601fn validate_size(name: &str, expected_bytes: u64) -> Result<(), ResidencyDeclarationError> {
1602 if expected_bytes == 0 {
1603 Err(ResidencyDeclarationError::ZeroSizedBinding {
1604 name: name.to_owned(),
1605 })
1606 } else {
1607 Ok(())
1608 }
1609}
1610
1611fn first_source(
1612 name: &str,
1613 recipe: &DerivedWeightRecipe,
1614) -> Result<String, ResidencyDeclarationError> {
1615 recipe
1616 .source_keys()
1617 .first()
1618 .map(|key| (*key).to_owned())
1619 .ok_or_else(|| ResidencyDeclarationError::EmptyRecipeSources {
1620 name: name.to_owned(),
1621 })
1622}
1623
1624#[derive(Debug, Clone, Eq, PartialEq, thiserror::Error)]
1626pub enum ResidencyDeclarationError {
1627 #[error("weight binding names must not be empty")]
1629 InvalidBindingName,
1630 #[error("weight binding {name:?} has an empty checkpoint key")]
1632 InvalidCheckpointKey {
1633 name: String,
1635 },
1636 #[error("weight binding {name:?} has no checkpoint recipe source")]
1638 EmptyRecipeSources {
1639 name: String,
1641 },
1642 #[error(
1644 "weight binding {name:?} has invalid quantization companions {scales:?} and {biases:?}"
1645 )]
1646 InvalidQuantizationCompanions {
1647 name: String,
1649 scales: String,
1651 biases: String,
1653 },
1654 #[error("weight binding {name:?} must contain at least one byte")]
1656 ZeroSizedBinding {
1657 name: String,
1659 },
1660 #[error("residency unit {id} must contain at least one binding")]
1662 EmptyUnit {
1663 id: OffloadUnitId,
1665 },
1666 #[error("residency unit {id} has duplicate binding name {name:?}")]
1668 DuplicateBindingName {
1669 id: OffloadUnitId,
1671 name: String,
1673 },
1674 #[error("duplicate logical weight binding {name:?}")]
1676 DuplicateLogicalBinding {
1677 name: String,
1679 },
1680 #[error("weight binding alias {alias:?} has unknown owner {owner:?}")]
1682 UnknownBindingAliasOwner {
1683 alias: String,
1685 owner: String,
1687 },
1688 #[error("weight binding alias {alias:?} has ambiguous owner {owner:?}")]
1690 AmbiguousBindingAliasOwner {
1691 alias: String,
1693 owner: String,
1695 },
1696 #[error("weight binding alias cycle contains {name:?}")]
1698 BindingAliasCycle {
1699 name: String,
1701 },
1702 #[error("weight binding alias {alias:?} declares {alias_bytes} bytes but owner {owner:?} declares {owner_bytes}")]
1704 BindingAliasByteMismatch {
1705 alias: String,
1707 owner: String,
1709 alias_bytes: u64,
1711 owner_bytes: u64,
1713 },
1714 #[error("weight binding alias {name:?} cannot own a physical checkpoint source")]
1716 AliasHasPhysicalSource {
1717 name: String,
1719 },
1720}
1721
1722#[cfg(test)]
1723mod tests {
1724 use std::{
1725 cell::RefCell,
1726 sync::{Arc, Mutex},
1727 };
1728
1729 use eredu_checkpoint::{store::TensorMetadata, StoredDtype};
1730 use eredu_core::residency::{MemoryTier, OffloadConfig, OffloadUnitSpec, ResidencyPolicy};
1731
1732 use super::*;
1733
1734 struct Catalog(BTreeMap<String, TensorMetadata>);
1735
1736 struct TestLeaseStorage(BTreeMap<String, u32>);
1737
1738 impl ResidencyLeaseStorage for TestLeaseStorage {
1739 type DeviceValue = u32;
1740 type HostValue = u32;
1741 type Error = &'static str;
1742 type BindingNames<'a> = std::iter::Map<
1743 std::collections::btree_map::Keys<'a, String, u32>,
1744 fn(&'a String) -> &'a str,
1745 >;
1746
1747 fn device_value<'a>(
1748 &'a self,
1749 _: &OffloadUnitId,
1750 name: &str,
1751 ) -> Result<&'a Self::DeviceValue, Self::Error> {
1752 self.0.get(name).ok_or("unknown binding")
1753 }
1754
1755 fn host_value<'a>(
1756 &'a self,
1757 _: &OffloadUnitId,
1758 name: &str,
1759 ) -> Result<&'a Self::HostValue, Self::Error> {
1760 self.0.get(name).ok_or("unknown binding")
1761 }
1762
1763 fn binding_names(&self) -> Self::BindingNames<'_> {
1764 self.0.keys().map(String::as_str)
1765 }
1766 }
1767
1768 #[derive(Default)]
1769 struct TestLeaseOwner(Mutex<Vec<(OffloadUnitId, MemoryTier)>>);
1770
1771 impl ResidencyLeaseOwner for TestLeaseOwner {
1772 fn release_residency_pin(&self, id: &OffloadUnitId, tier: MemoryTier) {
1773 self.0.lock().unwrap().push((id.clone(), tier));
1774 }
1775 }
1776
1777 struct TestTransferCompletion {
1778 succeeds: bool,
1779 waits: Arc<Mutex<usize>>,
1780 }
1781
1782 struct TestTransferResources(Arc<Mutex<Vec<bool>>>);
1783
1784 type TransferCompletionRecord = (Vec<OffloadUnitId>, MemoryTier, u64, bool);
1785
1786 #[derive(Default)]
1787 struct TestTransferOwner(Mutex<Vec<TransferCompletionRecord>>);
1788
1789 impl ResidencyTransferOwner<TestTransferCompletion, TestTransferResources> for TestTransferOwner {
1790 type Executor = Mutex<usize>;
1791 type Error = &'static str;
1792
1793 fn order_after(
1794 _: &TestTransferCompletion,
1795 executor: &Self::Executor,
1796 _: &OffloadUnitId,
1797 ) -> Result<(), Self::Error> {
1798 *executor.lock().unwrap() += 1;
1799 Ok(())
1800 }
1801
1802 fn is_complete(
1803 completion: &TestTransferCompletion,
1804 _: &OffloadUnitId,
1805 ) -> Result<bool, Self::Error> {
1806 Ok(completion.succeeds)
1807 }
1808
1809 fn wait(completion: &TestTransferCompletion, _: &OffloadUnitId) -> Result<(), Self::Error> {
1810 *completion.waits.lock().unwrap() += 1;
1811 completion.succeeds.then_some(()).ok_or("transfer failed")
1812 }
1813
1814 fn finish_resources(resources: TestTransferResources, succeeded: bool) {
1815 resources.0.lock().unwrap().push(succeeded);
1816 }
1817
1818 fn resolve_transfer(
1819 &self,
1820 ids: &[OffloadUnitId],
1821 tier: MemoryTier,
1822 generation: u64,
1823 succeeded: bool,
1824 ) -> Result<(), Self::Error> {
1825 self.0
1826 .lock()
1827 .unwrap()
1828 .push((ids.to_vec(), tier, generation, succeeded));
1829 Ok(())
1830 }
1831 }
1832
1833 #[derive(Default)]
1834 struct WindowManager {
1835 prepared: RefCell<Vec<(String, Vec<OffloadUnitId>)>>,
1836 evicted: RefCell<Vec<OffloadUnitId>>,
1837 }
1838
1839 impl ResidencyWindowManager for WindowManager {
1840 type Error = ResidencyWindowError;
1841
1842 fn prepare_window(
1843 &self,
1844 active: &[OffloadUnitId],
1845 _: &[OffloadUnitId],
1846 _: MemoryTier,
1847 ) -> Result<Vec<(OffloadUnitId, PrefetchOutcome)>, Self::Error> {
1848 self.prepared
1849 .borrow_mut()
1850 .push(("default".into(), active.to_vec()));
1851 Ok(active
1852 .iter()
1853 .cloned()
1854 .map(|id| (id, PrefetchOutcome::Hit))
1855 .collect())
1856 }
1857
1858 fn prepare_group_window(
1859 &self,
1860 group: &str,
1861 active: &[OffloadUnitId],
1862 _: &[OffloadUnitId],
1863 _: MemoryTier,
1864 ) -> Result<Vec<(OffloadUnitId, PrefetchOutcome)>, Self::Error> {
1865 self.prepared
1866 .borrow_mut()
1867 .push((group.to_owned(), active.to_vec()));
1868 Ok(active
1869 .iter()
1870 .cloned()
1871 .map(|id| (id, PrefetchOutcome::Miss))
1872 .collect())
1873 }
1874
1875 fn evict(&self, id: &OffloadUnitId, _: MemoryTier) -> Result<bool, Self::Error> {
1876 self.evicted.borrow_mut().push(id.clone());
1877 Ok(true)
1878 }
1879
1880 fn unit_reports(&self) -> Result<Vec<UnitResidencyReport>, Self::Error> {
1881 Ok(Vec::new())
1882 }
1883 }
1884
1885 impl RecipeCatalog for Catalog {
1886 fn tensor_metadata(
1887 &self,
1888 key: &str,
1889 ) -> Result<TensorMetadata, eredu_checkpoint::store::StoreError> {
1890 self.0.get(key).cloned().ok_or_else(|| {
1891 eredu_checkpoint::store::StoreError::UnknownTensor {
1892 key: key.to_owned(),
1893 }
1894 })
1895 }
1896 }
1897
1898 fn metadata(name: &str, shape: Vec<usize>) -> TensorMetadata {
1899 TensorMetadata {
1900 name: name.to_owned(),
1901 logical_shape: shape.clone(),
1902 physical_shape: shape,
1903 stored_dtype: StoredDtype::F32,
1904 encoded_byte_len: 0,
1905 backing_shard: None,
1906 }
1907 }
1908
1909 #[test]
1910 fn declarations_are_validated_and_deterministic() {
1911 let b = WeightBinding::new("b", "b.weight", TensorSelection::Full, 4).unwrap();
1912 let a = WeightBinding::new("a", "a.weight", TensorSelection::Full, 8).unwrap();
1913 let id = OffloadUnitId::new("layer.0").unwrap();
1914 let unit = OffloadUnit::new(id, [b, a]).unwrap();
1915 assert_eq!(unit.bindings()[0].name(), "a");
1916 assert_eq!(unit.bindings()[1].name(), "b");
1917 }
1918
1919 #[test]
1920 fn neutral_lease_exposes_native_storage_and_releases_exact_pin() {
1921 let owner = Arc::new(TestLeaseOwner::default());
1922 let id = OffloadUnitId::new("layer.0").unwrap();
1923 let lease = ResidencyLease::new(
1924 id.clone(),
1925 MemoryTier::Device,
1926 TestLeaseStorage(BTreeMap::from([("weight".into(), 7)])),
1927 Arc::downgrade(&owner),
1928 );
1929 assert_eq!(lease.device_value("weight"), Ok(&7));
1930 assert_eq!(lease.binding_names().collect::<Vec<_>>(), vec!["weight"]);
1931 drop(lease);
1932 assert_eq!(*owner.0.lock().unwrap(), vec![(id, MemoryTier::Device)]);
1933 }
1934
1935 #[test]
1936 fn neutral_transfer_orders_publishes_and_releases_resources() {
1937 let owner = Arc::new(TestTransferOwner::default());
1938 let waits = Arc::new(Mutex::new(0));
1939 let resources = Arc::new(Mutex::new(Vec::new()));
1940 let executor = Mutex::new(0);
1941 let id = OffloadUnitId::new("layer.0").unwrap();
1942 let mut transfer = ResidencyTransfer::submitted(
1943 vec![7],
1944 TestTransferCompletion {
1945 succeeds: true,
1946 waits: Arc::clone(&waits),
1947 },
1948 TestTransferResources(Arc::clone(&resources)),
1949 Arc::downgrade(&owner),
1950 vec![id.clone()],
1951 MemoryTier::Device,
1952 11,
1953 );
1954
1955 assert_eq!(transfer.leases(), &[7]);
1956 transfer.order_after(&executor).unwrap();
1957 assert_eq!(*executor.lock().unwrap(), 1);
1958 transfer.synchronize().unwrap();
1959 assert!(transfer.is_complete().unwrap());
1960 assert_eq!(*waits.lock().unwrap(), 1);
1961 assert_eq!(*resources.lock().unwrap(), vec![true]);
1962 assert_eq!(
1963 *owner.0.lock().unwrap(),
1964 vec![(vec![id], MemoryTier::Device, 11, true)]
1965 );
1966 }
1967
1968 #[test]
1969 fn neutral_transfer_failure_remains_observable_and_resolves_once() {
1970 let owner = Arc::new(TestTransferOwner::default());
1971 let waits = Arc::new(Mutex::new(0));
1972 let resources = Arc::new(Mutex::new(Vec::new()));
1973 let id = OffloadUnitId::new("layer.0").unwrap();
1974 let mut transfer = ResidencyTransfer::submitted(
1975 Vec::<u8>::new(),
1976 TestTransferCompletion {
1977 succeeds: false,
1978 waits: Arc::clone(&waits),
1979 },
1980 TestTransferResources(Arc::clone(&resources)),
1981 Arc::downgrade(&owner),
1982 vec![id.clone()],
1983 MemoryTier::Host,
1984 4,
1985 );
1986
1987 assert_eq!(transfer.synchronize(), Err("transfer failed"));
1988 assert_eq!(transfer.synchronize(), Err("transfer failed"));
1989 assert_eq!(*resources.lock().unwrap(), vec![false]);
1990 assert_eq!(
1991 *owner.0.lock().unwrap(),
1992 vec![(vec![id], MemoryTier::Host, 4, false)]
1993 );
1994 drop(transfer);
1995 assert_eq!(*waits.lock().unwrap(), 3);
1996 }
1997
1998 #[test]
1999 fn binding_selection_rewrites_sources_and_exact_bytes_neutrally() {
2000 let catalog = Catalog(BTreeMap::from([(
2001 "weight".into(),
2002 metadata("weight", vec![2, 2]),
2003 )]));
2004 let binding = WeightBinding::new("weight", "weight", TensorSelection::Full, 16)
2005 .unwrap()
2006 .select_bounded_output(
2007 &catalog,
2008 TensorSelection::Range {
2009 axis: 0,
2010 start: 1,
2011 end: 2,
2012 },
2013 )
2014 .unwrap();
2015
2016 assert_eq!(binding.expected_bytes(), 8);
2017 assert!(matches!(
2018 binding.source_recipe(),
2019 DerivedWeightRecipe::Source {
2020 selection: TensorSelection::Range {
2021 axis: 0,
2022 start: 1,
2023 end: 2,
2024 },
2025 ..
2026 }
2027 ));
2028 }
2029
2030 #[test]
2031 fn controller_validates_catalog_bytes_before_allocating_backend_storage() {
2032 let catalog = Catalog(BTreeMap::from([
2033 ("a.weight".into(), metadata("a.weight", vec![2])),
2034 ("b.weight".into(), metadata("b.weight", vec![1])),
2035 ]));
2036 let id = OffloadUnitId::new("layer.0").unwrap();
2037 let unit = OffloadUnit::new(
2038 id.clone(),
2039 [
2040 WeightBinding::new("a", "a.weight", TensorSelection::Full, 8).unwrap(),
2041 WeightBinding::new("b", "b.weight", TensorSelection::Full, 4).unwrap(),
2042 ],
2043 )
2044 .unwrap();
2045 let plan = OffloadPlan::new(
2046 OffloadConfig::default(),
2047 [
2048 OffloadUnitSpec::new(id.clone(), 12, ResidencyPolicy::Windowed, MemoryTier::Disk)
2049 .unwrap(),
2050 ],
2051 )
2052 .unwrap();
2053
2054 let controller = ResidencyController::new(&catalog, plan, [unit]).unwrap();
2055 assert_eq!(controller.units().len(), 1);
2056 assert_eq!(controller.unit(&id).unwrap().bindings().len(), 2);
2057 assert!(!controller.ledger().initialized());
2058 }
2059
2060 #[test]
2061 fn controller_resolves_aliases_across_independent_units() {
2062 let catalog = Catalog(BTreeMap::from([
2063 ("physical.owner".into(), metadata("physical.owner", vec![1])),
2064 ("slice.local".into(), metadata("slice.local", vec![1])),
2065 ]));
2066 let owner_id = OffloadUnitId::new("slice.0").unwrap();
2067 let alias_id = OffloadUnitId::new("slice.1").unwrap();
2068 let owner_binding =
2069 WeightBinding::new("weight", "physical.owner", TensorSelection::Full, 4)
2070 .unwrap()
2071 .with_logical_target("shared.owner")
2072 .unwrap();
2073 let alias_binding = WeightBinding::alias("weight", "shared.owner", 4)
2074 .unwrap()
2075 .with_logical_target("slice.1.weight")
2076 .unwrap();
2077 let local_binding =
2078 WeightBinding::new("local", "slice.local", TensorSelection::Full, 4).unwrap();
2079 let units = [
2080 OffloadUnit::new(owner_id.clone(), [owner_binding]).unwrap(),
2081 OffloadUnit::new(alias_id.clone(), [alias_binding, local_binding]).unwrap(),
2082 ];
2083 let plan = OffloadPlan::new(
2084 OffloadConfig::default(),
2085 [
2086 OffloadUnitSpec::new(
2087 owner_id.clone(),
2088 4,
2089 ResidencyPolicy::Windowed,
2090 MemoryTier::Disk,
2091 )
2092 .unwrap(),
2093 OffloadUnitSpec::new(
2094 alias_id.clone(),
2095 4,
2096 ResidencyPolicy::Windowed,
2097 MemoryTier::Disk,
2098 )
2099 .unwrap(),
2100 ],
2101 )
2102 .unwrap();
2103 let controller = ResidencyController::new(&catalog, plan, units).unwrap();
2104 let alias = controller
2105 .unit(&alias_id)
2106 .unwrap()
2107 .bindings()
2108 .iter()
2109 .find(|binding| binding.is_alias())
2110 .unwrap();
2111 let (resolved_unit, resolved) = controller.binding_owner(&alias_id, alias).unwrap();
2112 assert_eq!(resolved_unit, &owner_id);
2113 assert_eq!(resolved.logical_target(), Some("shared.owner"));
2114 }
2115
2116 #[test]
2117 fn controller_owns_named_window_and_unique_lookahead_selection() {
2118 let ids = ["a", "b", "c"].map(|name| OffloadUnitId::new(format!("layer.{name}")).unwrap());
2119 let catalog = Catalog(BTreeMap::from([
2120 ("a".into(), metadata("a", vec![1])),
2121 ("b".into(), metadata("b", vec![1])),
2122 ("c".into(), metadata("c", vec![1])),
2123 ]));
2124 let units = ids.iter().zip(["a", "b", "c"]).map(|(id, key)| {
2125 OffloadUnit::new(
2126 id.clone(),
2127 [WeightBinding::new("weight", key, TensorSelection::Full, 4).unwrap()],
2128 )
2129 .unwrap()
2130 });
2131 let plan = OffloadPlan::new(
2132 OffloadConfig::new(None, None, 2).unwrap(),
2133 ids.iter().map(|id| {
2134 OffloadUnitSpec::new(id.clone(), 4, ResidencyPolicy::Windowed, MemoryTier::Disk)
2135 .unwrap()
2136 }),
2137 )
2138 .unwrap();
2139 let mut controller = ResidencyController::new(&catalog, plan, units).unwrap();
2140 controller.ledger_mut().mark_initialized();
2141 let selected = controller
2142 .commit_group_window(
2143 "decoder",
2144 &[ids[0].clone()],
2145 &[ids[1].clone(), ids[1].clone(), ids[2].clone()],
2146 MemoryTier::Device,
2147 )
2148 .unwrap();
2149 assert_eq!(selected, vec![ids[1].clone(), ids[2].clone()]);
2150 assert_eq!(
2151 controller.ledger().active_window(),
2152 BTreeSet::from([ids[0].clone()])
2153 );
2154 }
2155
2156 #[test]
2157 fn controller_owns_acquisition_reservation_and_rollback() {
2158 let ids = ["a", "b"].map(|name| OffloadUnitId::new(format!("layer.{name}")).unwrap());
2159 let catalog = Catalog(BTreeMap::from([
2160 ("a".into(), metadata("a", vec![1])),
2161 ("b".into(), metadata("b", vec![1])),
2162 ]));
2163 let units = ids.iter().zip(["a", "b"]).map(|(id, key)| {
2164 OffloadUnit::new(
2165 id.clone(),
2166 [WeightBinding::new("weight", key, TensorSelection::Full, 4).unwrap()],
2167 )
2168 .unwrap()
2169 });
2170 let plan = OffloadPlan::new(
2171 OffloadConfig::new(Some(8), Some(8), 1).unwrap(),
2172 ids.iter().map(|id| {
2173 OffloadUnitSpec::new(id.clone(), 4, ResidencyPolicy::Cacheable, MemoryTier::Disk)
2174 .unwrap()
2175 }),
2176 )
2177 .unwrap();
2178 let mut controller = ResidencyController::new(&catalog, plan, units).unwrap();
2179 controller.ledger_mut().mark_initialized();
2180
2181 let acquisition = controller
2182 .plan_acquisition(&ids, MemoryTier::Device)
2183 .unwrap();
2184 assert_eq!(acquisition.missing(), &[true, true]);
2185 assert!(controller
2186 .reserve_acquisition(
2187 &acquisition,
2188 &[(ids[0].clone(), 4), (ids[1].clone(), 4)],
2189 MemoryTier::Device,
2190 )
2191 .unwrap()
2192 .is_empty());
2193 controller
2194 .rollback_acquisition(&acquisition, MemoryTier::Device)
2195 .unwrap();
2196 assert!(!controller
2197 .ledger()
2198 .is_resident(&ids[0], MemoryTier::Device)
2199 .unwrap());
2200 assert!(!controller
2201 .ledger()
2202 .is_resident(&ids[1], MemoryTier::Device)
2203 .unwrap());
2204
2205 let acquisition = controller
2206 .plan_acquisition(&[ids[0].clone()], MemoryTier::Device)
2207 .unwrap();
2208 controller
2209 .reserve_acquisition(&acquisition, &[(ids[0].clone(), 4)], MemoryTier::Device)
2210 .unwrap();
2211 controller
2212 .publish_acquisition_copy(
2213 &ids[0],
2214 MemoryTier::Device,
2215 4,
2216 4,
2217 None,
2218 eredu_core::residency::TransferDirection::DiskToDevice,
2219 Duration::from_millis(2),
2220 )
2221 .unwrap();
2222 assert!(controller
2223 .ledger()
2224 .is_resident(&ids[0], MemoryTier::Device)
2225 .unwrap());
2226 assert_eq!(
2227 controller
2228 .ledger()
2229 .telemetry()
2230 .transfer(eredu_core::residency::TransferDirection::DiskToDevice)
2231 .bytes(),
2232 4
2233 );
2234 }
2235
2236 #[test]
2237 fn controller_rejects_binding_and_plan_byte_mismatches() {
2238 let catalog = Catalog(BTreeMap::from([(
2239 "weight".into(),
2240 metadata("weight", vec![2]),
2241 )]));
2242 let id = OffloadUnitId::new("layer.0").unwrap();
2243 let plan = |bytes| {
2244 OffloadPlan::new(
2245 OffloadConfig::default(),
2246 [OffloadUnitSpec::new(
2247 id.clone(),
2248 bytes,
2249 ResidencyPolicy::Windowed,
2250 MemoryTier::Disk,
2251 )
2252 .unwrap()],
2253 )
2254 .unwrap()
2255 };
2256
2257 let wrong_binding = OffloadUnit::new(
2258 id.clone(),
2259 [WeightBinding::new("weight", "weight", TensorSelection::Full, 4).unwrap()],
2260 )
2261 .unwrap();
2262 assert!(matches!(
2263 ResidencyController::new(&catalog, plan(4), [wrong_binding]),
2264 Err(ResidencyControllerError::BindingByteMismatch { .. })
2265 ));
2266
2267 let wrong_plan = OffloadUnit::new(
2268 id.clone(),
2269 [WeightBinding::new("weight", "weight", TensorSelection::Full, 8).unwrap()],
2270 )
2271 .unwrap();
2272 assert!(matches!(
2273 ResidencyController::new(&catalog, plan(16), [wrong_plan]),
2274 Err(ResidencyControllerError::UnitByteMismatch { .. })
2275 ));
2276 }
2277
2278 #[test]
2279 fn named_windows_prepare_and_trim_without_backend_types() {
2280 let ids = ["layer.0", "layer.1", "layer.2"].map(|id| OffloadUnitId::new(id).unwrap());
2281 let group = ResidentLayerGroup::new("decoder", ids.clone(), 2).unwrap();
2282 let manager = WindowManager::default();
2283
2284 assert_eq!(group.prepare(&manager, 1).unwrap().len(), 2);
2285 assert_eq!(
2286 manager.prepared.borrow()[0],
2287 ("decoder".into(), vec![ids[1].clone(), ids[2].clone()])
2288 );
2289 assert_eq!(manager.evicted.borrow().as_slice(), &[ids[0].clone()]);
2290 assert_eq!(group.report(&manager).unwrap().device_units(), 0);
2291 }
2292}