1use super::checkpoint_capacity::{validate_checkpoint_pool_ceiling, CheckpointCapacityPolicy};
2use super::{
3 invalid_plan, minimum_for_lifetime, node_completion_precedes, quantize_storage_bytes,
4 static_contiguous_storage_profile, validate_active_sequence_ceiling,
5 validate_pool_liveness_rows, workspace_storage_layout_fingerprint, AllocationKind,
6 AllocationLifetime, BTreeMap, BTreeSet, BufferRequest, BufferUsage, CanonicalU128, Deserialize,
7 Deserializer, DynamicBackingPoolId, DynamicBackingPoolSpec, DynamicResourceDemand,
8 DynamicResourceDescriptor, DynamicResourceShape, ElementType, InvocationLivenessMode,
9 InvocationResourceLiveness, NodeId, PlanNode, PoolAggregateEvidence, PoolCompatibilityKey,
10 ResolvedReusableExecutionBucket, ResourceAllocation, ResourceId, ReusableExecutionMemoryPlan,
11 ReusableExecutionPolicy, ReusablePoolWorkspaceBudget, Serialize, StepResourceSlot,
12 StepResourceSlotKind, VNextError, MAX_EXECUTION_PLAN_RESOURCE_ROWS,
13};
14
15mod startup;
16
17#[derive(Debug, Clone, PartialEq, Eq, Serialize)]
18pub struct MemoryPlan {
19 pub(super) device_capacity_bytes: u64,
20 pub(super) policy_capacity_bytes: u64,
21 pub(super) reserve_bytes: u64,
22 pub(super) usable_capacity_bytes: u64,
23 pub(super) maximum_active_sequences: u32,
24 pub(super) static_bytes: u64,
25 pub(super) minimum_request_bytes: u64,
26 pub(super) minimum_sequence_bytes: u64,
27 pub(super) minimum_step_bytes: u64,
28 pub(super) minimum_invocation_peak_bytes: u64,
29 pub(super) minimum_runnable_request_bytes: u64,
30 pub(super) theoretical_ceiling_bytes: CanonicalU128,
31 pub(super) static_allocations: Vec<ResourceAllocation>,
32 pub(super) dynamic_descriptors: Vec<DynamicResourceDescriptor>,
33 pub(super) dynamic_pools: Vec<DynamicBackingPoolSpec>,
34 pub(super) reusable_execution: Option<ReusableExecutionMemoryPlan>,
35 #[serde(skip_serializing_if = "Option::is_none")]
36 pub(super) checkpoint_capacity: Option<CheckpointCapacityPolicy>,
37 pub(super) invocation_liveness_mode: InvocationLivenessMode,
38 pub(super) invocation_liveness: Vec<InvocationResourceLiveness>,
39}
40
41impl MemoryPlan {
42 #[cfg(test)]
43 pub(super) fn from_core(
44 device_capacity_bytes: u64,
45 policy_capacity_bytes: u64,
46 reserve_bytes: u64,
47 maximum_active_sequences: u32,
48 static_allocations: Vec<ResourceAllocation>,
49 dynamic_descriptors: Vec<DynamicResourceDescriptor>,
50 nodes: &[PlanNode],
51 reusable_execution_policy: Option<&ReusableExecutionPolicy>,
52 ) -> Result<Self, VNextError> {
53 Self::from_core_with_completion_retention(
54 device_capacity_bytes,
55 policy_capacity_bytes,
56 reserve_bytes,
57 maximum_active_sequences,
58 static_allocations,
59 dynamic_descriptors,
60 nodes,
61 reusable_execution_policy,
62 &BTreeSet::new(),
63 )
64 }
65
66 #[allow(clippy::too_many_arguments)]
67 pub(super) fn from_core_with_completion_retention(
68 device_capacity_bytes: u64,
69 policy_capacity_bytes: u64,
70 reserve_bytes: u64,
71 maximum_active_sequences: u32,
72 mut static_allocations: Vec<ResourceAllocation>,
73 mut dynamic_descriptors: Vec<DynamicResourceDescriptor>,
74 nodes: &[PlanNode],
75 reusable_execution_policy: Option<&ReusableExecutionPolicy>,
76 retained_completion_resources: &BTreeSet<ResourceId>,
77 ) -> Result<Self, VNextError> {
78 if static_allocations.len() + dynamic_descriptors.len() > MAX_EXECUTION_PLAN_RESOURCE_ROWS {
79 return Err(invalid_plan(format!(
80 "execution plan resource rows exceed {MAX_EXECUTION_PLAN_RESOURCE_ROWS}"
81 )));
82 }
83 let usable_capacity_bytes = policy_capacity_bytes
84 .checked_sub(reserve_bytes)
85 .ok_or_else(|| invalid_plan("memory reserve exceeds the policy capacity"))?;
86 static_allocations.sort_by(|left, right| left.resource_id.cmp(&right.resource_id));
87 dynamic_descriptors
88 .sort_by(|left, right| left.base_resource_id.cmp(&right.base_resource_id));
89 for resource_id in retained_completion_resources {
90 let descriptor = dynamic_descriptors
91 .iter()
92 .find(|descriptor| &descriptor.base_resource_id == resource_id)
93 .ok_or_else(|| {
94 invalid_plan(format!(
95 "retained completion resource `{resource_id}` has no dynamic descriptor"
96 ))
97 })?;
98 if descriptor.usage != BufferUsage::Activations
99 || !matches!(
100 descriptor.lifetime,
101 AllocationLifetime::Step | AllocationLifetime::Request
102 )
103 {
104 return Err(invalid_plan(format!(
105 "retained completion resource `{resource_id}` is not a readable activation"
106 )));
107 }
108 }
109 let static_bytes = static_allocations
110 .iter()
111 .try_fold(0_u64, |total, allocation| {
112 total
113 .checked_add(allocation.size_bytes)
114 .ok_or_else(|| invalid_plan("static memory total overflows u64"))
115 })?;
116 let dynamic_capacity_bytes = usable_capacity_bytes
117 .checked_sub(static_bytes)
118 .ok_or_else(|| invalid_plan("static memory exceeds usable capacity"))?;
119 let base_dynamic_pools = Self::derive_dynamic_pools_with_completion_retention(
120 &dynamic_descriptors,
121 nodes,
122 dynamic_capacity_bytes,
123 retained_completion_resources,
124 )?;
125 let reusable_execution = reusable_execution_policy
126 .map(|policy| {
127 Self::derive_reusable_execution(
128 policy,
129 nodes.len(),
130 &dynamic_descriptors,
131 &base_dynamic_pools,
132 )
133 })
134 .transpose()?;
135 let reusable_workspace_ceilings = reusable_execution
136 .as_ref()
137 .map(ReusableExecutionMemoryPlan::pool_workspace_ceilings)
138 .transpose()?
139 .unwrap_or_default();
140 let dynamic_pools = Self::derive_dynamic_pools_with_reusable(
141 &dynamic_descriptors,
142 nodes,
143 dynamic_capacity_bytes,
144 &reusable_workspace_ceilings,
145 retained_completion_resources,
146 )?;
147 let (invocation_liveness_mode, invocation_liveness) =
148 Self::summarize_pool_invocation_liveness(&dynamic_pools)?;
149 let minimum_request_bytes = dynamic_pools.iter().try_fold(0_u64, |total, pool| {
150 total
151 .checked_add(pool.minimum_request_bytes)
152 .ok_or_else(|| invalid_plan("minimum request bytes overflow u64"))
153 })?;
154 let minimum_sequence_bytes = dynamic_pools.iter().try_fold(0_u64, |total, pool| {
155 total
156 .checked_add(pool.minimum_sequence_bytes)
157 .ok_or_else(|| invalid_plan("minimum sequence bytes overflow u64"))
158 })?;
159 let minimum_step_bytes = dynamic_pools.iter().try_fold(0_u64, |total, pool| {
160 total
161 .checked_add(pool.minimum_step_bytes)
162 .ok_or_else(|| invalid_plan("minimum step bytes overflow u64"))
163 })?;
164 let minimum_invocation_peak_bytes =
165 dynamic_pools.iter().try_fold(0_u64, |total, pool| {
166 total
167 .checked_add(pool.minimum_invocation_peak_bytes)
168 .ok_or_else(|| invalid_plan("invocation pool minimum bytes overflow u64"))
169 })?;
170 let minimum_runnable_request_bytes = minimum_request_bytes
171 .checked_add(minimum_sequence_bytes)
172 .and_then(|bytes| bytes.checked_add(minimum_step_bytes))
173 .and_then(|bytes| bytes.checked_add(minimum_invocation_peak_bytes))
174 .ok_or_else(|| invalid_plan("minimum runnable request bytes overflow u64"))?;
175 let theoretical_dynamic_ceiling =
176 dynamic_pools.iter().try_fold(0_u128, |total, pool| {
177 total
178 .checked_add(pool.theoretical_ceiling_bytes.get())
179 .and_then(|bytes| {
180 bytes.checked_add(u128::from(pool.reusable_workspace_ceiling_bytes))
181 })
182 .ok_or_else(|| invalid_plan("dynamic theoretical total overflows u128"))
183 })?;
184 let theoretical_ceiling_bytes = u128::from(static_bytes)
185 .checked_add(theoretical_dynamic_ceiling)
186 .ok_or_else(|| invalid_plan("plan theoretical ceiling overflows u128"))?;
187 let minimum_runnable_bytes = static_bytes
188 .checked_add(minimum_runnable_request_bytes)
189 .ok_or_else(|| invalid_plan("minimum runnable plan bytes overflow u64"))?;
190 if minimum_runnable_bytes > usable_capacity_bytes {
191 return Err(invalid_plan(format!(
192 "minimum runnable plan requires {minimum_runnable_bytes} bytes, exceeding usable capacity {usable_capacity_bytes}"
193 )));
194 }
195 let plan = Self {
196 device_capacity_bytes,
197 policy_capacity_bytes,
198 reserve_bytes,
199 usable_capacity_bytes,
200 maximum_active_sequences,
201 static_bytes,
202 minimum_request_bytes,
203 minimum_sequence_bytes,
204 minimum_step_bytes,
205 minimum_invocation_peak_bytes,
206 minimum_runnable_request_bytes,
207 theoretical_ceiling_bytes: CanonicalU128::new(theoretical_ceiling_bytes),
208 static_allocations,
209 dynamic_descriptors,
210 dynamic_pools,
211 reusable_execution,
212 checkpoint_capacity: None,
213 invocation_liveness_mode,
214 invocation_liveness,
215 };
216 plan.validate()?;
217 Ok(plan)
218 }
219
220 pub(super) fn validate(&self) -> Result<(), VNextError> {
221 validate_active_sequence_ceiling(self.maximum_active_sequences)?;
222 if self.static_allocations.len() + self.dynamic_descriptors.len()
223 > MAX_EXECUTION_PLAN_RESOURCE_ROWS
224 {
225 return Err(invalid_plan(format!(
226 "execution plan resource rows exceed {MAX_EXECUTION_PLAN_RESOURCE_ROWS}"
227 )));
228 }
229 if self
230 .static_allocations
231 .windows(2)
232 .any(|pair| pair[0].resource_id >= pair[1].resource_id)
233 || self
234 .dynamic_descriptors
235 .windows(2)
236 .any(|pair| pair[0].base_resource_id >= pair[1].base_resource_id)
237 || self
238 .dynamic_pools
239 .windows(2)
240 .any(|pair| pair[0].pool_id >= pair[1].pool_id)
241 {
242 return Err(invalid_plan("memory plan resource rows are not canonical"));
243 }
244 if self.device_capacity_bytes == 0
245 || self.policy_capacity_bytes == 0
246 || self.policy_capacity_bytes > self.device_capacity_bytes
247 || self.reserve_bytes >= self.policy_capacity_bytes
248 || self.usable_capacity_bytes
249 != self
250 .policy_capacity_bytes
251 .checked_sub(self.reserve_bytes)
252 .ok_or_else(|| invalid_plan("memory reserve underflows policy capacity"))?
253 {
254 return Err(invalid_plan("memory plan capacity or reserve is invalid"));
255 }
256 let mut resources = BTreeSet::new();
257 let workspace_layout_fingerprint = workspace_storage_layout_fingerprint()?;
258 let static_contiguous_profile = static_contiguous_storage_profile()?;
259 let actual_static =
260 self.static_allocations
261 .iter()
262 .try_fold(0_u64, |total, allocation| {
263 if !resources.insert(allocation.resource_id.clone())
264 || allocation.per_instance_bytes == 0
265 || allocation.instance_stride_bytes
266 != quantize_storage_bytes(
267 allocation.per_instance_bytes,
268 allocation.alignment_bytes,
269 allocation.storage.profile(),
270 )?
271 || allocation.instance_count != 1
272 || allocation.size_bytes != allocation.instance_stride_bytes
273 || allocation.lifetime != AllocationLifetime::Plan
274 || match &allocation.kind {
275 AllocationKind::Value => {
276 allocation.storage.profile() != static_contiguous_profile
277 }
278 AllocationKind::InitializationScratch => {
279 allocation.usage != BufferUsage::Scratch
280 || allocation.element_type != ElementType::U8
281 || allocation.storage.logical_layout_fingerprint()
282 != workspace_layout_fingerprint
283 }
284 AllocationKind::Persistent { .. } => {
285 allocation.usage != BufferUsage::Persistent
286 || allocation.element_type != ElementType::U8
287 || allocation.storage.logical_layout_fingerprint()
288 != workspace_layout_fingerprint
289 }
290 AllocationKind::Scratch { .. } | AllocationKind::Binding { .. } => true,
291 }
292 {
293 return Err(invalid_plan(format!(
294 "static resource `{}` is duplicate or invalid",
295 allocation.resource_id
296 )));
297 }
298 total
299 .checked_add(allocation.size_bytes)
300 .ok_or_else(|| invalid_plan("static allocation total overflows u64"))
301 })?;
302 if actual_static != self.static_bytes {
303 return Err(invalid_plan("static byte total is not core-derived"));
304 }
305 let mut actual_theoretical_dynamic = 0_u128;
306 let dynamic_by_id = self
307 .dynamic_descriptors
308 .iter()
309 .map(|descriptor| (descriptor.base_resource_id.clone(), descriptor))
310 .collect::<BTreeMap<_, _>>();
311 for descriptor in &self.dynamic_descriptors {
312 if !resources.insert(descriptor.base_resource_id.clone()) {
313 return Err(invalid_plan(format!(
314 "dynamic resource `{}` is duplicated",
315 descriptor.base_resource_id
316 )));
317 }
318 validate_active_sequence_ceiling(descriptor.theoretical_maximum_instances)?;
319 descriptor.demand.validate()?;
320 if descriptor.lifetime == AllocationLifetime::Plan {
321 return Err(invalid_plan(
322 "dynamic descriptor cannot have plan-static lifetime",
323 ));
324 }
325 actual_theoretical_dynamic = actual_theoretical_dynamic
326 .checked_add(descriptor.theoretical_maximum_resident_bytes()?)
327 .ok_or_else(|| invalid_plan("dynamic ceiling total overflows u128"))?;
328 }
329 let reusable_workspace_ceilings = self
330 .reusable_execution
331 .as_ref()
332 .map(|plan| {
333 plan.validate_local()?;
334 plan.pool_workspace_ceilings()
335 })
336 .transpose()?
337 .unwrap_or_default();
338 let actual_reusable_workspace =
339 reusable_workspace_ceilings
340 .values()
341 .try_fold(0_u128, |total, bytes| {
342 total
343 .checked_add(u128::from(*bytes))
344 .ok_or_else(|| invalid_plan("reusable workspace total overflows u128"))
345 })?;
346 let actual_theoretical = u128::from(actual_static)
347 .checked_add(actual_theoretical_dynamic)
348 .and_then(|bytes| bytes.checked_add(actual_reusable_workspace))
349 .ok_or_else(|| invalid_plan("theoretical plan ceiling overflows u128"))?;
350 let dynamic_capacity_bytes = self
351 .usable_capacity_bytes
352 .checked_sub(actual_static)
353 .ok_or_else(|| invalid_plan("static memory exceeds usable capacity"))?;
354 let aggregate = Self::validate_dynamic_pools_structure(
355 &self.dynamic_pools,
356 &dynamic_by_id,
357 dynamic_capacity_bytes,
358 &reusable_workspace_ceilings,
359 self.checkpoint_capacity.as_ref(),
360 )?;
361 let (actual_liveness_mode, actual_liveness) =
362 Self::summarize_pool_invocation_liveness(&self.dynamic_pools)?;
363 if actual_liveness_mode != self.invocation_liveness_mode
364 || actual_liveness != self.invocation_liveness
365 {
366 return Err(invalid_plan(
367 "global invocation liveness is not derived from per-pool evidence",
368 ));
369 }
370 let actual_request_minimum = aggregate.minimum_request_bytes;
371 let actual_sequence_minimum = aggregate.minimum_sequence_bytes;
372 let actual_step_minimum = aggregate.minimum_step_bytes;
373 let actual_invocation_peak = aggregate.minimum_invocation_peak_bytes;
374 let actual_minimum = actual_request_minimum
375 .checked_add(actual_sequence_minimum)
376 .and_then(|bytes| bytes.checked_add(actual_step_minimum))
377 .and_then(|bytes| bytes.checked_add(actual_invocation_peak))
378 .ok_or_else(|| invalid_plan("minimum runnable request bytes overflow u64"))?;
379 if aggregate.theoretical_ceiling_bytes != actual_theoretical_dynamic
380 || aggregate.reusable_workspace_ceiling_bytes != actual_reusable_workspace
381 || actual_request_minimum != self.minimum_request_bytes
382 || actual_sequence_minimum != self.minimum_sequence_bytes
383 || actual_step_minimum != self.minimum_step_bytes
384 || actual_invocation_peak != self.minimum_invocation_peak_bytes
385 || actual_minimum != self.minimum_runnable_request_bytes
386 || actual_theoretical != self.theoretical_ceiling_bytes.get()
387 || actual_static
388 .checked_add(actual_minimum)
389 .is_none_or(|minimum| minimum > self.usable_capacity_bytes)
390 {
391 return Err(invalid_plan(
392 "dynamic minimum or theoretical ceiling is not core-derived",
393 ));
394 }
395 Ok(())
396 }
397
398 #[cfg(test)]
399 pub(crate) fn derive_dynamic_pools(
400 dynamic_descriptors: &[DynamicResourceDescriptor],
401 nodes: &[PlanNode],
402 dynamic_capacity_bytes: u64,
403 ) -> Result<Vec<DynamicBackingPoolSpec>, VNextError> {
404 Self::derive_dynamic_pools_with_reusable(
405 dynamic_descriptors,
406 nodes,
407 dynamic_capacity_bytes,
408 &BTreeMap::new(),
409 &BTreeSet::new(),
410 )
411 }
412
413 pub(super) fn derive_dynamic_pools_with_completion_retention(
414 dynamic_descriptors: &[DynamicResourceDescriptor],
415 nodes: &[PlanNode],
416 dynamic_capacity_bytes: u64,
417 retained_completion_resources: &BTreeSet<ResourceId>,
418 ) -> Result<Vec<DynamicBackingPoolSpec>, VNextError> {
419 Self::derive_dynamic_pools_with_reusable(
420 dynamic_descriptors,
421 nodes,
422 dynamic_capacity_bytes,
423 &BTreeMap::new(),
424 retained_completion_resources,
425 )
426 }
427
428 pub(super) fn derive_dynamic_pools_with_reusable(
429 dynamic_descriptors: &[DynamicResourceDescriptor],
430 nodes: &[PlanNode],
431 dynamic_capacity_bytes: u64,
432 reusable_workspace_ceilings: &BTreeMap<DynamicBackingPoolId, u64>,
433 retained_completion_resources: &BTreeSet<ResourceId>,
434 ) -> Result<Vec<DynamicBackingPoolSpec>, VNextError> {
435 Self::derive_dynamic_pools_with_checkpoint(
436 dynamic_descriptors,
437 nodes,
438 dynamic_capacity_bytes,
439 reusable_workspace_ceilings,
440 retained_completion_resources,
441 &BTreeMap::new(),
442 )
443 }
444
445 pub(super) fn derive_dynamic_pools_with_checkpoint(
446 dynamic_descriptors: &[DynamicResourceDescriptor],
447 nodes: &[PlanNode],
448 dynamic_capacity_bytes: u64,
449 reusable_workspace_ceilings: &BTreeMap<DynamicBackingPoolId, u64>,
450 retained_completion_resources: &BTreeSet<ResourceId>,
451 checkpoint_growth_ceilings: &BTreeMap<DynamicBackingPoolId, u64>,
452 ) -> Result<Vec<DynamicBackingPoolSpec>, VNextError> {
453 let mut groups = BTreeMap::<
454 DynamicBackingPoolId,
455 (PoolCompatibilityKey, Vec<&DynamicResourceDescriptor>),
456 >::new();
457 for descriptor in dynamic_descriptors {
458 let compatibility = PoolCompatibilityKey::new(
459 &descriptor.storage,
460 descriptor.usage,
461 descriptor.element_type,
462 descriptor.alignment_bytes,
463 )?;
464 let expected_pool_id = DynamicBackingPoolId::from_compatibility(&compatibility)?;
465 if descriptor.pool_id != expected_pool_id {
466 return Err(invalid_plan(format!(
467 "dynamic resource `{}` has a non-derived backing pool id",
468 descriptor.base_resource_id
469 )));
470 }
471 match groups.entry(expected_pool_id) {
472 std::collections::btree_map::Entry::Vacant(entry) => {
473 entry.insert((compatibility, vec![descriptor]));
474 }
475 std::collections::btree_map::Entry::Occupied(mut entry) => {
476 if entry.get().0 != compatibility {
477 return Err(invalid_plan(
478 "dynamic backing pool hash collision has incompatible contracts",
479 ));
480 }
481 entry.get_mut().1.push(descriptor);
482 }
483 }
484 }
485
486 let pools = groups
487 .into_values()
488 .map(|(compatibility, mut descriptors)| {
489 descriptors
490 .sort_by(|left, right| left.base_resource_id.cmp(&right.base_resource_id));
491 let resource_ids = descriptors
492 .iter()
493 .map(|descriptor| descriptor.base_resource_id.clone())
494 .collect::<Vec<_>>();
495 let minimum_request_bytes = minimum_for_lifetime(
496 &descriptors,
497 AllocationLifetime::Request,
498 "pool request minimum",
499 )?;
500 let minimum_sequence_bytes = minimum_for_lifetime(
501 &descriptors,
502 AllocationLifetime::Sequence,
503 "pool sequence minimum",
504 )?;
505 let step_resource_slots = Self::derive_pool_step_slots(
506 nodes,
507 &descriptors,
508 retained_completion_resources,
509 )?;
510 let minimum_step_bytes = Self::step_slot_bytes(
511 &step_resource_slots,
512 &descriptors,
513 false,
514 "pool step minimum",
515 )?;
516 let theoretical_ceiling_bytes =
517 descriptors.iter().try_fold(0_u128, |total, descriptor| {
518 total
519 .checked_add(descriptor.theoretical_maximum_resident_bytes()?)
520 .ok_or_else(|| {
521 invalid_plan("dynamic pool theoretical ceiling overflows u128")
522 })
523 })?;
524 let (invocation_liveness_mode, invocation_liveness, invocation_peak) =
525 Self::derive_pool_invocation_liveness(nodes, &descriptors)?;
526 let pool_id = DynamicBackingPoolId::from_compatibility(&compatibility)?;
527 let reusable_workspace_ceiling_bytes = reusable_workspace_ceilings
528 .get(&pool_id)
529 .copied()
530 .unwrap_or(0);
531 let checkpoint_growth_ceiling_bytes = checkpoint_growth_ceilings
532 .get(&pool_id)
533 .copied()
534 .unwrap_or(0);
535 DynamicBackingPoolSpec::from_core(
536 compatibility,
537 resource_ids,
538 minimum_request_bytes,
539 minimum_sequence_bytes,
540 minimum_step_bytes,
541 invocation_peak,
542 step_resource_slots,
543 theoretical_ceiling_bytes,
544 reusable_workspace_ceiling_bytes,
545 checkpoint_growth_ceiling_bytes,
546 dynamic_capacity_bytes,
547 invocation_liveness_mode,
548 invocation_liveness,
549 )
550 })
551 .collect::<Result<Vec<_>, VNextError>>()?;
552 if reusable_workspace_ceilings
553 .keys()
554 .any(|pool_id| !pools.iter().any(|pool| &pool.pool_id == pool_id))
555 {
556 return Err(invalid_plan(
557 "reusable workspace ceiling references an unknown dynamic pool",
558 ));
559 }
560 if checkpoint_growth_ceilings
561 .keys()
562 .any(|pool_id| !pools.iter().any(|pool| &pool.pool_id == pool_id))
563 {
564 return Err(invalid_plan(
565 "checkpoint growth ceiling references an unknown dynamic pool",
566 ));
567 }
568 Ok(pools)
569 }
570
571 pub(super) fn derive_reusable_execution(
572 policy: &ReusableExecutionPolicy,
573 node_count: usize,
574 dynamic_descriptors: &[DynamicResourceDescriptor],
575 base_pools: &[DynamicBackingPoolSpec],
576 ) -> Result<ReusableExecutionMemoryPlan, VNextError> {
577 policy.validate()?;
578 let node_count = u64::try_from(node_count)
579 .ok()
580 .filter(|count| *count > 0)
581 .ok_or_else(|| invalid_plan("reusable execution requires at least one plan node"))?;
582 let startup_capture_case_count = u64::try_from(policy.startup_capture_case_count())
583 .map_err(|_| {
584 invalid_plan("reusable execution startup capture case count exceeds u64")
585 })?;
586 let maximum_device_executables = node_count
587 .checked_mul(startup_capture_case_count)
588 .and_then(|count| count.checked_mul(u64::from(policy.maximum_reusable_lanes())))
589 .ok_or_else(|| invalid_plan("reusable device executable count overflows u64"))?;
590 let descriptors = dynamic_descriptors
591 .iter()
592 .map(|descriptor| (descriptor.base_resource_id.clone(), descriptor))
593 .collect::<BTreeMap<_, _>>();
594 let mut buckets = Vec::with_capacity(policy.buckets().len());
595 for bucket in policy.buckets() {
596 let capacity = bucket.capacity();
597 let shape = DynamicResourceShape::from_validated(
598 capacity.maximum_sequences(),
599 capacity.maximum_tokens(),
600 capacity.maximum_pages(),
601 );
602 let mut pool_budgets = Vec::new();
603 for pool in base_pools {
604 let step_bytes = Self::reusable_step_bytes_for_shape(pool, &descriptors, shape)?;
605 let invocation_bytes =
606 Self::reusable_invocation_bytes_for_shape(pool, &descriptors, shape)?;
607 if step_bytes != 0 || invocation_bytes != 0 {
608 pool_budgets.push(ReusablePoolWorkspaceBudget::new(
609 pool.pool_id.clone(),
610 step_bytes,
611 invocation_bytes,
612 )?);
613 }
614 }
615 buckets.push(ResolvedReusableExecutionBucket::new(
616 bucket.clone(),
617 pool_budgets,
618 )?);
619 }
620 ReusableExecutionMemoryPlan::new_with_program_policy(
621 policy.maximum_reusable_lanes(),
622 maximum_device_executables,
623 buckets,
624 policy.program_policy().cloned(),
625 )
626 }
627
628 fn reusable_step_bytes_for_shape(
629 pool: &DynamicBackingPoolSpec,
630 descriptors: &BTreeMap<ResourceId, &DynamicResourceDescriptor>,
631 shape: DynamicResourceShape,
632 ) -> Result<u64, VNextError> {
633 pool.step_resource_slots
634 .iter()
635 .try_fold(0_u64, |total, slot| {
636 let slot_bytes =
637 slot.resource_ids
638 .iter()
639 .try_fold(0_u64, |maximum, resource_id| {
640 let descriptor = descriptors.get(resource_id).ok_or_else(|| {
641 invalid_plan(
642 "reusable Step slot references a missing dynamic descriptor",
643 )
644 })?;
645 Ok::<u64, VNextError>(
646 maximum.max(descriptor.evaluate_request_bytes_for_shape(shape)?),
647 )
648 })?;
649 total
650 .checked_add(slot_bytes)
651 .ok_or_else(|| invalid_plan("reusable Step workspace budget overflows u64"))
652 })
653 }
654
655 fn reusable_invocation_bytes_for_shape(
656 pool: &DynamicBackingPoolSpec,
657 descriptors: &BTreeMap<ResourceId, &DynamicResourceDescriptor>,
658 shape: DynamicResourceShape,
659 ) -> Result<u64, VNextError> {
660 let row_bytes = |row: &InvocationResourceLiveness| {
661 row.resource_ids
662 .iter()
663 .try_fold(0_u64, |total, resource_id| {
664 total
665 .checked_add(
666 descriptors
667 .get(resource_id)
668 .ok_or_else(|| {
669 invalid_plan(
670 "reusable invocation row references a missing descriptor",
671 )
672 })?
673 .evaluate_request_bytes_for_shape(shape)?,
674 )
675 .ok_or_else(|| invalid_plan("reusable invocation row budget overflows u64"))
676 })
677 };
678 match pool.invocation_liveness_mode {
679 InvocationLivenessMode::NoInvocationResources => Ok(0),
680 InvocationLivenessMode::TotalOrderReuse => pool
681 .invocation_liveness
682 .iter()
683 .try_fold(0_u64, |maximum, row| {
684 Ok::<u64, VNextError>(maximum.max(row_bytes(row)?))
685 }),
686 InvocationLivenessMode::ConservativeConcurrent => pool
687 .invocation_liveness
688 .iter()
689 .try_fold(0_u64, |total, row| {
690 total.checked_add(row_bytes(row)?).ok_or_else(|| {
691 invalid_plan("reusable concurrent invocation budget overflows u64")
692 })
693 }),
694 }
695 }
696
697 pub(super) fn derive_pool_step_slots(
698 nodes: &[PlanNode],
699 descriptors: &[&DynamicResourceDescriptor],
700 retained_completion_resources: &BTreeSet<ResourceId>,
701 ) -> Result<Vec<StepResourceSlot>, VNextError> {
702 struct Interval<'a> {
703 resource_id: &'a ResourceId,
704 first_user: usize,
705 last_user: usize,
706 reusable: bool,
707 }
708
709 let nodes_by_id = nodes
710 .iter()
711 .map(|node| (node.id.clone(), node))
712 .collect::<BTreeMap<_, _>>();
713 let mut intervals = descriptors
714 .iter()
715 .filter(|descriptor| descriptor.lifetime == AllocationLifetime::Step)
716 .map(|descriptor| {
717 let users = nodes
718 .iter()
719 .enumerate()
720 .filter_map(|(index, node)| {
721 node.resources
722 .contains(&descriptor.base_resource_id)
723 .then_some(index)
724 })
725 .collect::<Vec<_>>();
726 let first_user = users.first().copied().ok_or_else(|| {
727 invalid_plan(format!(
728 "step resource `{}` is not referenced by a plan node",
729 descriptor.base_resource_id
730 ))
731 })?;
732 let last_user = users.last().copied().ok_or_else(|| {
733 invalid_plan(format!(
734 "step resource `{}` is not referenced by a plan node",
735 descriptor.base_resource_id
736 ))
737 })?;
738 Ok(Interval {
739 resource_id: &descriptor.base_resource_id,
740 first_user,
741 last_user,
742 reusable: Self::is_reusable_step_activation(descriptor)
745 && !retained_completion_resources.contains(&descriptor.base_resource_id),
746 })
747 })
748 .collect::<Result<Vec<_>, VNextError>>()?;
749 intervals.sort_by(|left, right| {
750 (left.first_user, left.last_user, left.resource_id).cmp(&(
751 right.first_user,
752 right.last_user,
753 right.resource_id,
754 ))
755 });
756
757 let ordered_without_overlap = |left: &Interval<'_>, right: &Interval<'_>| {
758 let left_before = left.last_user < right.first_user
759 && node_completion_precedes(
760 &nodes_by_id,
761 &nodes[left.last_user].id,
762 &nodes[right.first_user].id,
763 )?;
764 let right_before = right.last_user < left.first_user
765 && node_completion_precedes(
766 &nodes_by_id,
767 &nodes[right.last_user].id,
768 &nodes[left.first_user].id,
769 )?;
770 Ok::<bool, VNextError>(left_before || right_before)
771 };
772
773 let mut slots = Vec::<Vec<Interval<'_>>>::new();
774 for interval in intervals {
775 if !interval.reusable {
776 slots.push(vec![interval]);
777 continue;
778 }
779 let mut reusable_slot = None;
780 for (index, slot) in slots.iter().enumerate() {
781 let mut reusable = slot.iter().all(|member| member.reusable);
782 if reusable {
783 for member in slot {
784 reusable = ordered_without_overlap(member, &interval)?;
785 if !reusable {
786 break;
787 }
788 }
789 }
790 if reusable {
791 reusable_slot = Some(index);
792 break;
793 }
794 }
795 if let Some(index) = reusable_slot {
796 slots[index].push(interval);
797 } else {
798 slots.push(vec![interval]);
799 }
800 }
801 let mut slots = slots
802 .into_iter()
803 .map(|slot| {
804 let resource_ids = slot
805 .into_iter()
806 .map(|interval| interval.resource_id.clone())
807 .collect::<Vec<_>>();
808 if resource_ids.len() == 1 {
809 Ok(StepResourceSlot::dedicated(
810 resource_ids
811 .into_iter()
812 .next()
813 .expect("single resource slot"),
814 ))
815 } else {
816 StepResourceSlot::ordered_single_fence_wave(resource_ids)
817 }
818 })
819 .collect::<Result<Vec<_>, VNextError>>()?;
820 slots.sort_by(|left, right| left.resource_ids.cmp(&right.resource_ids));
821 Ok(slots)
822 }
823
824 fn is_reusable_step_activation(descriptor: &DynamicResourceDescriptor) -> bool {
825 descriptor.lifetime == AllocationLifetime::Step
826 && descriptor.usage == BufferUsage::Activations
827 && matches!(descriptor.kind, AllocationKind::Value)
828 && matches!(descriptor.demand, DynamicResourceDemand::Tokens { .. })
829 }
830
831 fn step_slot_bytes(
832 slots: &[StepResourceSlot],
833 descriptors: &[&DynamicResourceDescriptor],
834 theoretical: bool,
835 overflow_context: &str,
836 ) -> Result<u64, VNextError> {
837 let descriptors = descriptors
838 .iter()
839 .map(|descriptor| (descriptor.base_resource_id.clone(), *descriptor))
840 .collect::<BTreeMap<_, _>>();
841 slots.iter().try_fold(0_u64, |total, slot| {
842 let slot_bytes = slot
843 .resource_ids
844 .iter()
845 .try_fold(0_u64, |maximum, resource_id| {
846 let descriptor = descriptors.get(resource_id).ok_or_else(|| {
847 invalid_plan("step slot references a missing dynamic descriptor")
848 })?;
849 let bytes = if theoretical {
850 descriptor.theoretical_maximum_request_bytes()?
851 } else {
852 descriptor.minimum_request_bytes()?
853 };
854 Ok::<u64, VNextError>(maximum.max(bytes))
855 })?;
856 total
857 .checked_add(slot_bytes)
858 .ok_or_else(|| invalid_plan(format!("{overflow_context} overflows u64")))
859 })
860 }
861
862 pub(super) fn derive_pool_invocation_liveness(
863 nodes: &[PlanNode],
864 descriptors: &[&DynamicResourceDescriptor],
865 ) -> Result<(InvocationLivenessMode, Vec<InvocationResourceLiveness>, u64), VNextError> {
866 let invocation_ids = descriptors
867 .iter()
868 .filter(|descriptor| descriptor.lifetime == AllocationLifetime::Invocation)
869 .map(|descriptor| descriptor.base_resource_id.clone())
870 .collect::<BTreeSet<_>>();
871 if invocation_ids.is_empty() {
872 return Ok((InvocationLivenessMode::NoInvocationResources, Vec::new(), 0));
873 }
874 let descriptor_by_id = descriptors
875 .iter()
876 .map(|descriptor| (descriptor.base_resource_id.clone(), *descriptor))
877 .collect::<BTreeMap<_, _>>();
878 let mut covered = BTreeSet::new();
879 let liveness_in_execution_order = nodes
880 .iter()
881 .filter_map(|node| {
882 let resource_ids = node
883 .resources
884 .iter()
885 .filter(|resource_id| invocation_ids.contains(*resource_id))
886 .cloned()
887 .collect::<Vec<_>>();
888 covered.extend(resource_ids.iter().cloned());
889 (!resource_ids.is_empty()).then(|| InvocationResourceLiveness {
890 node_id: node.id.clone(),
891 resource_ids,
892 })
893 })
894 .collect::<Vec<_>>();
895 if covered != invocation_ids {
896 return Err(invalid_plan(
897 "pool node liveness does not cover every invocation resource",
898 ));
899 }
900 let nodes_by_id = nodes
901 .iter()
902 .map(|node| (node.id.clone(), node))
903 .collect::<BTreeMap<_, _>>();
904 let contains_binding = descriptors.iter().any(|descriptor| {
905 descriptor.lifetime == AllocationLifetime::Invocation
906 && matches!(descriptor.kind, AllocationKind::Binding { .. })
907 });
908 let total_ordered = !contains_binding
909 && liveness_in_execution_order
910 .windows(2)
911 .try_fold(true, |ordered, pair| {
912 Ok::<bool, VNextError>(
913 ordered
914 && node_completion_precedes(
915 &nodes_by_id,
916 &pair[0].node_id,
917 &pair[1].node_id,
918 )?,
919 )
920 })?;
921 let row_bytes = |row: &InvocationResourceLiveness| {
922 row.resource_ids
923 .iter()
924 .try_fold(0_u64, |total, resource_id| {
925 total
926 .checked_add(
927 descriptor_by_id
928 .get(resource_id)
929 .ok_or_else(|| {
930 invalid_plan("pool liveness references an unknown descriptor")
931 })?
932 .minimum_request_bytes()?,
933 )
934 .ok_or_else(|| invalid_plan("pool invocation row bytes overflow u64"))
935 })
936 };
937 let invocation_peak = if total_ordered {
938 liveness_in_execution_order
939 .iter()
940 .try_fold(0_u64, |peak, row| {
941 Ok::<u64, VNextError>(peak.max(row_bytes(row)?))
942 })?
943 } else {
944 liveness_in_execution_order
945 .iter()
946 .try_fold(0_u64, |total, row| {
947 total.checked_add(row_bytes(row)?).ok_or_else(|| {
948 invalid_plan("pool concurrent invocation bytes overflow u64")
949 })
950 })?
951 };
952 let mut invocation_liveness = liveness_in_execution_order;
953 invocation_liveness.sort_by(|left, right| left.node_id.cmp(&right.node_id));
954 Ok((
955 if total_ordered {
956 InvocationLivenessMode::TotalOrderReuse
957 } else {
958 InvocationLivenessMode::ConservativeConcurrent
959 },
960 invocation_liveness,
961 invocation_peak,
962 ))
963 }
964
965 pub(super) fn validate_dynamic_pools_structure(
966 pools: &[DynamicBackingPoolSpec],
967 descriptors: &BTreeMap<ResourceId, &DynamicResourceDescriptor>,
968 dynamic_capacity_bytes: u64,
969 reusable_workspace_ceilings: &BTreeMap<DynamicBackingPoolId, u64>,
970 checkpoint_capacity: Option<&CheckpointCapacityPolicy>,
971 ) -> Result<PoolAggregateEvidence, VNextError> {
972 let mut expected_members =
973 BTreeMap::<DynamicBackingPoolId, (PoolCompatibilityKey, Vec<ResourceId>)>::new();
974 for descriptor in descriptors.values() {
975 let compatibility = PoolCompatibilityKey::new(
976 &descriptor.storage,
977 descriptor.usage,
978 descriptor.element_type,
979 descriptor.alignment_bytes,
980 )?;
981 let expected_id = DynamicBackingPoolId::from_compatibility(&compatibility)?;
982 if descriptor.pool_id != expected_id {
983 return Err(invalid_plan(
984 "dynamic descriptor has a non-derived pool identity",
985 ));
986 }
987 match expected_members.entry(expected_id) {
988 std::collections::btree_map::Entry::Vacant(entry) => {
989 entry.insert((compatibility, vec![descriptor.base_resource_id.clone()]));
990 }
991 std::collections::btree_map::Entry::Occupied(mut entry) => {
992 if entry.get().0 != compatibility {
993 return Err(invalid_plan(
994 "dynamic pool hash collision has incompatible descriptors",
995 ));
996 }
997 entry.get_mut().1.push(descriptor.base_resource_id.clone());
998 }
999 }
1000 }
1001 if pools.len() != expected_members.len() {
1002 return Err(invalid_plan(
1003 "dynamic backing pool count is not derived from descriptors",
1004 ));
1005 }
1006 if reusable_workspace_ceilings
1007 .keys()
1008 .any(|pool_id| !expected_members.contains_key(pool_id))
1009 {
1010 return Err(invalid_plan(
1011 "reusable workspace ceiling references an unknown dynamic pool",
1012 ));
1013 }
1014 let mut aggregate = PoolAggregateEvidence::default();
1015 for pool in pools {
1016 pool.validate_local()?;
1017 let (compatibility, resource_ids) = expected_members
1018 .remove(&pool.pool_id)
1019 .ok_or_else(|| invalid_plan("dynamic backing pool has no descriptor members"))?;
1020 if pool.compatibility != compatibility || pool.resource_ids != resource_ids {
1021 return Err(invalid_plan(
1022 "dynamic backing pool compatibility or membership is not core-derived",
1023 ));
1024 }
1025 let members =
1026 pool.resource_ids
1027 .iter()
1028 .map(|resource_id| {
1029 descriptors.get(resource_id).copied().ok_or_else(|| {
1030 invalid_plan("dynamic pool member descriptor is missing")
1031 })
1032 })
1033 .collect::<Result<Vec<_>, VNextError>>()?;
1034 let request = minimum_for_lifetime(
1035 &members,
1036 AllocationLifetime::Request,
1037 "pool request minimum",
1038 )?;
1039 let sequence = minimum_for_lifetime(
1040 &members,
1041 AllocationLifetime::Sequence,
1042 "pool sequence minimum",
1043 )?;
1044 let expected_step_resources = members
1045 .iter()
1046 .filter(|descriptor| descriptor.lifetime == AllocationLifetime::Step)
1047 .map(|descriptor| descriptor.base_resource_id.clone())
1048 .collect::<BTreeSet<_>>();
1049 let actual_step_resources = pool
1050 .step_resource_slots
1051 .iter()
1052 .flat_map(|slot| slot.resource_ids.iter().cloned())
1053 .collect::<BTreeSet<_>>();
1054 if actual_step_resources != expected_step_resources {
1055 return Err(invalid_plan(
1056 "step resource slots do not cover exactly the pool's Step descriptors",
1057 ));
1058 }
1059 for slot in &pool.step_resource_slots {
1060 slot.validate()?;
1061 if slot.kind == StepResourceSlotKind::OrderedSingleFenceStepWave
1062 && slot.resource_ids.iter().any(|resource_id| {
1063 descriptors
1064 .get(resource_id)
1065 .is_none_or(|descriptor| !Self::is_reusable_step_activation(descriptor))
1066 })
1067 {
1068 return Err(invalid_plan(
1069 "shared Step slot contains a non-transient activation resource",
1070 ));
1071 }
1072 }
1073 let step = Self::step_slot_bytes(
1074 &pool.step_resource_slots,
1075 &members,
1076 false,
1077 "pool step minimum",
1078 )?;
1079 let theoretical = members.iter().try_fold(0_u128, |total, descriptor| {
1080 total
1081 .checked_add(descriptor.theoretical_maximum_resident_bytes()?)
1082 .ok_or_else(|| invalid_plan("pool theoretical ceiling overflows u128"))
1083 })?;
1084 let invocation_ids = members
1085 .iter()
1086 .filter(|descriptor| descriptor.lifetime == AllocationLifetime::Invocation)
1087 .map(|descriptor| descriptor.base_resource_id.clone())
1088 .collect::<BTreeSet<_>>();
1089 let invocation_peak = validate_pool_liveness_rows(
1090 pool.invocation_liveness_mode,
1091 &pool.invocation_liveness,
1092 &invocation_ids,
1093 descriptors,
1094 )?;
1095 let reusable_workspace_ceiling_bytes = reusable_workspace_ceilings
1096 .get(&pool.pool_id)
1097 .copied()
1098 .unwrap_or(0);
1099 let combined_ceiling = theoretical
1100 .checked_add(u128::from(reusable_workspace_ceiling_bytes))
1101 .and_then(|bytes| {
1102 bytes.checked_add(u128::from(pool.checkpoint_growth_ceiling_bytes))
1103 })
1104 .ok_or_else(|| invalid_plan("pool combined ceiling overflows u128"))?;
1105 validate_checkpoint_pool_ceiling(
1106 checkpoint_capacity,
1107 &members,
1108 pool.checkpoint_growth_ceiling_bytes,
1109 )?;
1110 let maximum_resident =
1111 u64::try_from(combined_ceiling.min(u128::from(dynamic_capacity_bytes)))
1112 .map_err(|_| invalid_plan("pool resident ceiling exceeds u64"))?;
1113 if pool.minimum_request_bytes != request
1114 || pool.minimum_sequence_bytes != sequence
1115 || pool.minimum_step_bytes != step
1116 || pool.minimum_invocation_peak_bytes != invocation_peak
1117 || pool.theoretical_ceiling_bytes.get() != theoretical
1118 || pool.reusable_workspace_ceiling_bytes != reusable_workspace_ceiling_bytes
1119 || pool.provisioning.maximum_resident_bytes != maximum_resident
1120 {
1121 return Err(invalid_plan(
1122 "dynamic backing pool bounds or liveness are not core-derived",
1123 ));
1124 }
1125 aggregate.add(pool)?;
1126 }
1127 if !expected_members.is_empty() {
1128 return Err(invalid_plan(
1129 "dynamic descriptors are missing canonical backing pools",
1130 ));
1131 }
1132 Ok(aggregate)
1133 }
1134
1135 pub(super) fn summarize_pool_invocation_liveness(
1136 pools: &[DynamicBackingPoolSpec],
1137 ) -> Result<(InvocationLivenessMode, Vec<InvocationResourceLiveness>), VNextError> {
1138 let mut by_node = BTreeMap::<NodeId, BTreeSet<ResourceId>>::new();
1139 let mut has_invocation = false;
1140 let mut all_total_ordered = true;
1141 for pool in pools {
1142 match pool.invocation_liveness_mode {
1143 InvocationLivenessMode::NoInvocationResources => {}
1144 InvocationLivenessMode::TotalOrderReuse => has_invocation = true,
1145 InvocationLivenessMode::ConservativeConcurrent => {
1146 has_invocation = true;
1147 all_total_ordered = false;
1148 }
1149 }
1150 for row in &pool.invocation_liveness {
1151 by_node
1152 .entry(row.node_id.clone())
1153 .or_default()
1154 .extend(row.resource_ids.iter().cloned());
1155 }
1156 }
1157 let liveness = by_node
1158 .into_iter()
1159 .map(|(node_id, resource_ids)| InvocationResourceLiveness {
1160 node_id,
1161 resource_ids: resource_ids.into_iter().collect(),
1162 })
1163 .collect::<Vec<_>>();
1164 let reference_count = liveness.iter().try_fold(0_usize, |total, row| {
1165 total.checked_add(row.resource_ids.len())
1166 });
1167 if reference_count.is_none_or(|count| count > MAX_EXECUTION_PLAN_RESOURCE_ROWS) {
1168 return Err(invalid_plan(
1169 "invocation liveness reference count is invalid",
1170 ));
1171 }
1172 Ok((
1173 if !has_invocation {
1174 InvocationLivenessMode::NoInvocationResources
1175 } else if all_total_ordered {
1176 InvocationLivenessMode::TotalOrderReuse
1177 } else {
1178 InvocationLivenessMode::ConservativeConcurrent
1179 },
1180 liveness,
1181 ))
1182 }
1183
1184 pub const fn device_capacity_bytes(&self) -> u64 {
1185 self.device_capacity_bytes
1186 }
1187
1188 pub const fn policy_capacity_bytes(&self) -> u64 {
1189 self.policy_capacity_bytes
1190 }
1191
1192 pub const fn reserve_bytes(&self) -> u64 {
1193 self.reserve_bytes
1194 }
1195
1196 pub const fn usable_capacity_bytes(&self) -> u64 {
1197 self.usable_capacity_bytes
1198 }
1199
1200 pub const fn maximum_active_sequences(&self) -> u32 {
1201 self.maximum_active_sequences
1202 }
1203
1204 pub const fn checkpoint_capacity(&self) -> Option<&CheckpointCapacityPolicy> {
1207 self.checkpoint_capacity.as_ref()
1208 }
1209
1210 pub const fn capacity_bytes(&self) -> u64 {
1211 self.usable_capacity_bytes
1212 }
1213
1214 pub const fn static_bytes(&self) -> u64 {
1215 self.static_bytes
1216 }
1217
1218 pub const fn minimum_request_bytes(&self) -> u64 {
1219 self.minimum_request_bytes
1220 }
1221
1222 pub const fn minimum_sequence_bytes(&self) -> u64 {
1223 self.minimum_sequence_bytes
1224 }
1225
1226 pub const fn minimum_step_bytes(&self) -> u64 {
1227 self.minimum_step_bytes
1228 }
1229
1230 pub const fn minimum_invocation_peak_bytes(&self) -> u64 {
1231 self.minimum_invocation_peak_bytes
1232 }
1233
1234 pub const fn minimum_runnable_request_bytes(&self) -> u64 {
1235 self.minimum_runnable_request_bytes
1236 }
1237
1238 pub const fn invocation_liveness_mode(&self) -> InvocationLivenessMode {
1239 self.invocation_liveness_mode
1240 }
1241
1242 pub fn invocation_liveness(&self) -> &[InvocationResourceLiveness] {
1243 &self.invocation_liveness
1244 }
1245
1246 pub fn theoretical_ceiling_bytes(&self) -> u128 {
1252 self.theoretical_ceiling_bytes.get()
1253 }
1254
1255 pub fn static_allocations(&self) -> &[ResourceAllocation] {
1256 &self.static_allocations
1257 }
1258
1259 pub fn dynamic_descriptors(&self) -> &[DynamicResourceDescriptor] {
1260 &self.dynamic_descriptors
1261 }
1262
1263 pub fn dynamic_pools(&self) -> &[DynamicBackingPoolSpec] {
1264 &self.dynamic_pools
1265 }
1266
1267 pub fn reusable_execution(&self) -> Option<&ReusableExecutionMemoryPlan> {
1268 self.reusable_execution.as_ref()
1269 }
1270
1271 pub fn static_buffer_requests(&self) -> Result<Vec<BufferRequest>, VNextError> {
1272 self.static_allocations
1273 .iter()
1274 .map(ResourceAllocation::buffer_request)
1275 .collect()
1276 }
1277}
1278
1279#[derive(Deserialize)]
1280#[serde(deny_unknown_fields)]
1281pub(super) struct MemoryPlanWire {
1282 pub(super) device_capacity_bytes: u64,
1283 pub(super) policy_capacity_bytes: u64,
1284 pub(super) reserve_bytes: u64,
1285 pub(super) usable_capacity_bytes: u64,
1286 pub(super) maximum_active_sequences: u32,
1287 pub(super) static_bytes: u64,
1288 pub(super) minimum_request_bytes: u64,
1289 pub(super) minimum_sequence_bytes: u64,
1290 pub(super) minimum_step_bytes: u64,
1291 pub(super) minimum_invocation_peak_bytes: u64,
1292 pub(super) minimum_runnable_request_bytes: u64,
1293 pub(super) theoretical_ceiling_bytes: CanonicalU128,
1294 pub(super) static_allocations: Vec<ResourceAllocation>,
1295 pub(super) dynamic_descriptors: Vec<DynamicResourceDescriptor>,
1296 pub(super) dynamic_pools: Vec<DynamicBackingPoolSpec>,
1297 pub(super) reusable_execution: Option<ReusableExecutionMemoryPlan>,
1298 #[serde(default)]
1299 pub(super) checkpoint_capacity: Option<CheckpointCapacityPolicy>,
1300 pub(super) invocation_liveness_mode: InvocationLivenessMode,
1301 pub(super) invocation_liveness: Vec<InvocationResourceLiveness>,
1302}
1303
1304impl<'de> Deserialize<'de> for MemoryPlan {
1305 fn deserialize<D>(deserializer: D) -> Result<Self, D::Error>
1306 where
1307 D: Deserializer<'de>,
1308 {
1309 let wire = MemoryPlanWire::deserialize(deserializer)?;
1310 let plan = Self {
1311 device_capacity_bytes: wire.device_capacity_bytes,
1312 policy_capacity_bytes: wire.policy_capacity_bytes,
1313 reserve_bytes: wire.reserve_bytes,
1314 usable_capacity_bytes: wire.usable_capacity_bytes,
1315 maximum_active_sequences: wire.maximum_active_sequences,
1316 static_bytes: wire.static_bytes,
1317 minimum_request_bytes: wire.minimum_request_bytes,
1318 minimum_sequence_bytes: wire.minimum_sequence_bytes,
1319 minimum_step_bytes: wire.minimum_step_bytes,
1320 minimum_invocation_peak_bytes: wire.minimum_invocation_peak_bytes,
1321 minimum_runnable_request_bytes: wire.minimum_runnable_request_bytes,
1322 theoretical_ceiling_bytes: wire.theoretical_ceiling_bytes,
1323 static_allocations: wire.static_allocations,
1324 dynamic_descriptors: wire.dynamic_descriptors,
1325 dynamic_pools: wire.dynamic_pools,
1326 reusable_execution: wire.reusable_execution,
1327 checkpoint_capacity: wire.checkpoint_capacity,
1328 invocation_liveness_mode: wire.invocation_liveness_mode,
1329 invocation_liveness: wire.invocation_liveness,
1330 };
1331 plan.validate().map_err(serde::de::Error::custom)?;
1332 Ok(plan)
1333 }
1334}