1use super::{
2 invalid_resource, AllocationKind, AllocationLifetime, Arc, AtomicU64, AtomicU8, BTreeMap,
3 BackingChunkIdentity, BackingSegment, BufferDescriptor, BufferUsage, CapacityDomainId,
4 CapacityEntry, CapacityEpochs, CapacityUnits, CapacityVector, CapacityWaitCondition,
5 DeviceBufferRetention, DeviceCapacityAvailabilitySnapshot, DeviceCapacityGrant, DeviceRuntime,
6 DynamicBackingPoolId, DynamicBackingPoolSpec, DynamicResourceDescriptor, DynamicResourceShape,
7 DynamicStorageAllocator, DynamicStorageProfile, ElementType, ExecutionLaneId, FreeExtentIndex,
8 InvocationLivenessMode, LogicalAdmissionCoordinator, LogicalAdmissionCoordinatorId, Mutex,
9 Ordering, PlanNode, ResourceId, Serialize, StateInitialization, VNextError,
10};
11use crate::vnext::{
12 DeviceCapacityPressure, DeviceReusableAddressScope, DynamicPoolProvisioningPolicy,
13 DynamicResourceDemand, PoolCompatibilityKey, ReusableExecutionBucketId,
14 ReusableExecutionMemoryPlan,
15};
16use sha2::{Digest, Sha256};
17
18pub(super) static NEXT_DYNAMIC_POOL_INSTANCE_ID: AtomicU64 = AtomicU64::new(1);
19
20pub(super) fn align_up_resource(value: u64, alignment: u64) -> Result<u64, VNextError> {
21 if alignment == 0 || !alignment.is_power_of_two() {
22 return Err(invalid_resource(
23 "dynamic pool alignment is not a non-zero power of two",
24 ));
25 }
26 value
27 .checked_add(alignment - 1)
28 .map(|rounded| rounded & !(alignment - 1))
29 .ok_or_else(|| invalid_resource("dynamic pool aligned bytes overflow u64"))
30}
31
32pub(super) fn free_extent_layout_fingerprint(allocator: &FreeExtentIndex) -> String {
33 let mut hasher = Sha256::new();
34 for (&(chunk_ordinal, offset_bytes), extent) in &allocator.by_offset {
35 hasher.update(chunk_ordinal.to_be_bytes());
36 hasher.update(extent.chunk_generation.to_be_bytes());
37 hasher.update(offset_bytes.to_be_bytes());
38 hasher.update(extent.length_bytes.to_be_bytes());
39 }
40 format!("sha256/{:x}", hasher.finalize())
41}
42
43fn unused_simulation_chunk_ordinal(allocator: &FreeExtentIndex) -> Result<u32, VNextError> {
44 let mut candidate = u32::MAX;
45 loop {
46 if !allocator
47 .by_offset
48 .keys()
49 .any(|(chunk_ordinal, _)| *chunk_ordinal == candidate)
50 {
51 return Ok(candidate);
52 }
53 candidate = candidate.checked_sub(1).ok_or_else(|| {
54 invalid_resource("contiguous packing simulation exhausted chunk identities")
55 })?;
56 }
57}
58
59pub(super) fn contiguous_packing_growth_bytes(
63 allocator: &FreeExtentIndex,
64 pool_id: &DynamicBackingPoolId,
65 claim_bytes_descending: &[u64],
66) -> Result<u64, VNextError> {
67 if claim_bytes_descending.is_empty()
68 || claim_bytes_descending.iter().any(|bytes| *bytes == 0)
69 || claim_bytes_descending
70 .windows(2)
71 .any(|pair| pair[0] < pair[1])
72 {
73 return Err(invalid_resource(
74 "contiguous packing demand is empty, zero-sized, or non-canonical",
75 ));
76 }
77 let maximum_growth = claim_bytes_descending
78 .iter()
79 .try_fold(0_u64, |total, bytes| total.checked_add(*bytes))
80 .ok_or_else(|| invalid_resource("contiguous packing demand overflows u64"))?;
81 let synthetic_chunk = unused_simulation_chunk_ordinal(allocator)?;
82 let mut growth_bytes = 0_u64;
83 loop {
84 let mut simulation = allocator.clone();
85 if growth_bytes != 0 {
86 simulation.insert_extent(synthetic_chunk, u64::MAX, 0, growth_bytes)?;
87 }
88 let mut failed_claim = None;
89 for &claim_bytes in claim_bytes_descending {
90 if simulation
91 .allocate_contiguous(pool_id, claim_bytes)?
92 .is_none()
93 {
94 failed_claim = Some(claim_bytes);
95 break;
96 }
97 }
98 let Some(failed_claim) = failed_claim else {
99 return Ok(growth_bytes);
100 };
101 growth_bytes = growth_bytes
102 .checked_add(failed_claim)
103 .ok_or_else(|| invalid_resource("contiguous packing growth overflows u64"))?;
104 if growth_bytes > maximum_growth {
105 return Err(invalid_resource(
106 "contiguous packing planner exceeded its guaranteed growth bound",
107 ));
108 }
109 }
110}
111
112#[derive(Debug, Clone, PartialEq, Eq, Serialize)]
113pub(super) struct DynamicPoolDomainSpec {
114 pub(super) domain_id: CapacityDomainId,
115 pub(super) pool: DynamicBackingPoolSpec,
116 pub(super) descriptors: Vec<DynamicResourceDescriptor>,
117}
118
119impl DynamicPoolDomainSpec {
120 pub const fn domain_id(&self) -> CapacityDomainId {
121 self.domain_id
122 }
123
124 pub(super) fn pool_id(&self) -> &DynamicBackingPoolId {
125 self.pool.pool_id()
126 }
127}
128
129#[derive(Debug)]
130pub(super) struct SubmissionWaveProjectionLayout {
131 pub(super) descriptor_index: usize,
132 pub(super) projection_index: usize,
133}
134
135#[derive(Debug)]
136pub(super) struct SubmissionWaveRowLayout {
137 pub(super) projections: Vec<SubmissionWaveProjectionLayout>,
138}
139
140#[derive(Debug)]
141pub(super) struct SubmissionWaveDomainLayout {
142 pub(super) rows: Vec<SubmissionWaveRowLayout>,
143 pub(super) claim_identity: PhysicalBackingClaimIdentity,
144 pub(super) projection_count: usize,
145}
146
147#[derive(Debug)]
148pub(super) struct SubmissionWaveProjectionCapacity {
149 pub(super) physical_offset_bytes: u64,
150 pub(super) capacity_size_bytes: u64,
151}
152
153#[derive(Debug)]
154pub(super) struct SubmissionWaveDomainCapacityLayout {
155 pub(super) physical_size_bytes: u64,
156 pub(super) projections: Vec<SubmissionWaveProjectionCapacity>,
157}
158
159pub(super) fn compile_submission_wave_domain_layout(
160 domain: &DynamicPoolDomainSpec,
161 nodes: &[PlanNode],
162) -> Result<Option<SubmissionWaveDomainLayout>, VNextError> {
163 if domain.pool.invocation_liveness_mode() == InvocationLivenessMode::NoInvocationResources {
164 return Ok(None);
165 }
166
167 let canonical_projections = domain
168 .descriptors
169 .iter()
170 .enumerate()
171 .filter(|(_, descriptor)| descriptor.lifetime() == AllocationLifetime::Invocation)
172 .collect::<Vec<_>>();
173 let projection_by_resource = canonical_projections
174 .iter()
175 .enumerate()
176 .map(|(projection_index, (descriptor_index, descriptor))| {
177 (
178 descriptor.base_resource_id(),
179 (*descriptor_index, projection_index),
180 )
181 })
182 .collect::<BTreeMap<_, _>>();
183 if canonical_projections.is_empty()
184 || projection_by_resource.len() != canonical_projections.len()
185 {
186 return Err(invalid_resource(
187 "submission wave layout requires unique invocation descriptors",
188 ));
189 }
190
191 let liveness = domain.pool.invocation_liveness();
192 let mut covered_projections = std::collections::BTreeSet::new();
193 let mut rows = Vec::with_capacity(liveness.len());
194 for node in nodes {
195 let Ok(row_index) = liveness.binary_search_by(|row| row.node_id().cmp(node.id())) else {
196 continue;
197 };
198 let row = &liveness[row_index];
199 let projections = row
200 .resource_ids()
201 .iter()
202 .map(|resource_id| {
203 let &(descriptor_index, projection_index) =
204 projection_by_resource.get(resource_id).ok_or_else(|| {
205 invalid_resource(
206 "submission wave liveness references a descriptor outside its pool",
207 )
208 })?;
209 if !covered_projections.insert(projection_index) {
210 return Err(invalid_resource(
211 "submission wave liveness repeats one invocation descriptor",
212 ));
213 }
214 Ok(SubmissionWaveProjectionLayout {
215 descriptor_index,
216 projection_index,
217 })
218 })
219 .collect::<Result<Vec<_>, VNextError>>()?;
220 rows.push(SubmissionWaveRowLayout { projections });
221 }
222 if rows.len() != liveness.len()
223 || covered_projections.len() != canonical_projections.len()
224 || covered_projections
225 .iter()
226 .copied()
227 .ne(0..canonical_projections.len())
228 {
229 return Err(invalid_resource(
230 "submission wave layout does not cover immutable plan invocation resources exactly",
231 ));
232 }
233
234 Ok(Some(SubmissionWaveDomainLayout {
235 rows,
236 claim_identity: PhysicalBackingClaimIdentity::new(
237 domain.pool_id().clone(),
238 canonical_projections
239 .iter()
240 .map(|(_, descriptor)| descriptor.base_resource_id().clone())
241 .collect(),
242 )?,
243 projection_count: canonical_projections.len(),
244 }))
245}
246
247pub(super) fn compile_submission_wave_reusable_capacity_layouts(
248 domains: &[DynamicPoolDomainSpec],
249 layouts: &[Option<SubmissionWaveDomainLayout>],
250 reusable_execution: Option<&ReusableExecutionMemoryPlan>,
251) -> Result<
252 BTreeMap<ReusableExecutionBucketId, Vec<Option<SubmissionWaveDomainCapacityLayout>>>,
253 VNextError,
254> {
255 let Some(reusable_execution) = reusable_execution else {
256 return Ok(BTreeMap::new());
257 };
258 if domains.len() != layouts.len() {
259 return Err(invalid_resource(
260 "submission wave reusable capacity layout count differs from dynamic pool domains",
261 ));
262 }
263
264 reusable_execution
265 .buckets()
266 .iter()
267 .map(|resolved| {
268 let bucket = resolved.bucket();
269 let capacity = bucket.capacity();
270 let capacity_shape = DynamicResourceShape::from_validated(
271 capacity.maximum_sequences(),
272 capacity.maximum_tokens(),
273 capacity.maximum_pages(),
274 );
275 let compiled = domains
276 .iter()
277 .zip(layouts)
278 .map(|(domain, layout)| {
279 let Some(layout) = layout else {
280 return Ok(None);
281 };
282 let mode = domain.pool.invocation_liveness_mode();
283 let mut projections = (0..layout.projection_count)
284 .map(|_| None)
285 .collect::<Vec<_>>();
286 let mut physical_size_bytes = 0_u64;
287 for row in &layout.rows {
288 let row_base = match mode {
289 InvocationLivenessMode::TotalOrderReuse => 0,
290 InvocationLivenessMode::ConservativeConcurrent => physical_size_bytes,
291 InvocationLivenessMode::NoInvocationResources => unreachable!(),
292 };
293 let mut row_bytes = 0_u64;
294 for projection_layout in &row.projections {
295 let descriptor = domain
296 .descriptors
297 .get(projection_layout.descriptor_index)
298 .ok_or_else(|| {
299 invalid_resource(
300 "reusable submission layout references a descriptor outside its pool",
301 )
302 })?;
303 if descriptor.lifetime() != AllocationLifetime::Invocation {
304 return Err(invalid_resource(
305 "reusable submission layout references a non-Invocation descriptor",
306 ));
307 }
308 let capacity_size_bytes =
309 descriptor.evaluate_request_bytes_for_shape(capacity_shape)?;
310 let physical_offset_bytes =
311 row_base.checked_add(row_bytes).ok_or_else(|| {
312 invalid_resource(
313 "reusable submission projection offset overflows u64",
314 )
315 })?;
316 row_bytes =
317 row_bytes.checked_add(capacity_size_bytes).ok_or_else(|| {
318 invalid_resource(
319 "reusable submission row capacity overflows u64",
320 )
321 })?;
322 if projections[projection_layout.projection_index]
323 .replace(SubmissionWaveProjectionCapacity {
324 physical_offset_bytes,
325 capacity_size_bytes,
326 })
327 .is_some()
328 {
329 return Err(invalid_resource(
330 "reusable submission layout repeats one canonical projection",
331 ));
332 }
333 }
334 physical_size_bytes = match mode {
335 InvocationLivenessMode::TotalOrderReuse => {
336 physical_size_bytes.max(row_bytes)
337 }
338 InvocationLivenessMode::ConservativeConcurrent => physical_size_bytes
339 .checked_add(row_bytes)
340 .ok_or_else(|| {
341 invalid_resource(
342 "reusable submission pool capacity overflows u64",
343 )
344 })?,
345 InvocationLivenessMode::NoInvocationResources => unreachable!(),
346 };
347 }
348 let projections =
349 projections
350 .into_iter()
351 .collect::<Option<Vec<_>>>()
352 .ok_or_else(|| {
353 invalid_resource(
354 "reusable submission layout left a projection uncompiled",
355 )
356 })?;
357 if projections.is_empty() || physical_size_bytes == 0 {
358 return Err(invalid_resource(
359 "reusable submission layout compiled empty capacity",
360 ));
361 }
362 Ok(Some(SubmissionWaveDomainCapacityLayout {
363 physical_size_bytes,
364 projections,
365 }))
366 })
367 .collect::<Result<Vec<_>, VNextError>>()?;
368 Ok((bucket.bucket_id().clone(), compiled))
369 })
370 .collect()
371}
372
373pub(super) struct ResidentChunkBacking<B> {
374 pub(super) buffer: B,
376 pub(super) _grant: DeviceCapacityGrant,
377 pub(super) identity: BackingChunkIdentity,
378 pub(super) descriptor: BufferDescriptor,
379}
380
381pub(super) struct ResidentChunkState<B> {
382 pub(super) backing: Arc<ResidentChunkBacking<B>>,
383 pub(super) live_segments: u64,
384}
385
386#[derive(Debug, Clone, Copy, PartialEq, Eq)]
387pub(super) struct DynamicBackingClaimOccupancy {
388 pub(super) scope: DynamicBackingClaimScope,
389 pub(super) residency: DynamicBackingClaimResidency,
390 pub(super) physical_bytes: u64,
391 pub(super) segment_count: u64,
392}
393
394pub(super) fn rollback_free_extent_journal<B>(
395 states: &mut [std::sync::MutexGuard<'_, DynamicBackingPoolState<B>>],
396 journals: &[Vec<Vec<BackingSegment>>],
397) -> Result<(), VNextError> {
398 for group_index in (0..journals.len()).rev() {
399 for segments in journals[group_index].iter().rev() {
400 for segment in segments.iter().rev() {
401 if let Err(error) = states[group_index].allocator.release(segment) {
402 states[group_index].poisoned = true;
403 return Err(invalid_resource(format!(
404 "dynamic backing rollback failed and poisoned its pool: {error}"
405 )));
406 }
407 }
408 }
409 }
410 Ok(())
411}
412
413pub(super) struct DynamicBackingPoolState<B> {
414 pub(super) resident_bytes: u64,
415 pub(super) pending_growth_bytes: u64,
416 pub(super) next_chunk_ordinal: u32,
417 pub(super) next_chunk_generation: u64,
418 pub(super) chunks: BTreeMap<u32, ResidentChunkState<B>>,
419 pub(super) allocator: FreeExtentIndex,
420 pub(super) live_occupancy: DynamicPoolLiveOccupancyStatus,
421 pub(super) quarantined: Vec<QuarantinedDynamicChunk<B>>,
422 pub(super) poisoned: bool,
423}
424
425#[derive(Debug, Clone, Copy, PartialEq, Eq, Serialize)]
426#[serde(rename_all = "snake_case")]
427pub enum DynamicChunkQuarantineReason {
428 DescriptorMismatch,
429 PublicationRejected,
430}
431
432pub(super) struct QuarantinedDynamicChunk<B> {
433 pub(super) backing: Arc<ResidentChunkBacking<B>>,
434 pub(super) reason: DynamicChunkQuarantineReason,
435}
436
437pub(super) struct DynamicBackingPool<R>
438where
439 R: DeviceRuntime,
440{
441 pub(super) instance_id: u64,
442 pub(super) domain: DynamicPoolDomainSpec,
443 pub(super) logical_admission: LogicalAdmissionCoordinator,
444 pub(super) maintenance: Mutex<()>,
445 pub(super) next_extent_generation: AtomicU64,
446 pub(super) state: Mutex<DynamicBackingPoolState<R::Buffer>>,
447}
448
449pub(super) struct PendingGrowthGuard<R>
450where
451 R: DeviceRuntime,
452{
453 pub(super) pool: Arc<DynamicBackingPool<R>>,
454 pub(super) bytes: u64,
455 pub(super) armed: bool,
456}
457
458impl<R> PendingGrowthGuard<R>
459where
460 R: DeviceRuntime,
461{
462 pub(super) fn disarm(&mut self) {
463 self.armed = false;
464 }
465}
466
467impl<R> Drop for PendingGrowthGuard<R>
468where
469 R: DeviceRuntime,
470{
471 fn drop(&mut self) {
472 if self.armed {
473 self.pool.cancel_pending_growth(self.bytes);
474 }
475 }
476}
477
478pub(super) trait BackingExtentOwner: Send + Sync {
479 fn instance_id(&self) -> u64;
480 fn release_segments(
481 &self,
482 claim_identity: &PhysicalBackingClaimIdentity,
483 occupancy: DynamicBackingClaimOccupancy,
484 segments: &[BackingSegment],
485 );
486}
487
488pub(super) struct BackingSegmentLease {
489 pub(super) owner: Arc<dyn BackingExtentOwner>,
490 pub(super) owner_instance_id: u64,
491 pub(super) claim_identity: PhysicalBackingClaimIdentity,
492 pub(super) segment_generation: u64,
493 pub(super) occupancy: DynamicBackingClaimOccupancy,
494 pub(super) segments: Vec<BackingSegment>,
495 pub(super) size_bytes: u64,
496 pub(super) initialization: Option<Arc<BackingInitializationCell>>,
497 pub(super) released: bool,
498}
499
500#[derive(Debug, Clone, Copy, PartialEq, Eq, Serialize)]
501#[serde(rename_all = "snake_case")]
502pub enum BackingInitializationStatus {
503 Pending,
504 Prepared,
505 InFlight,
506 Initialized,
507 Poisoned,
508}
509
510const BACKING_INITIALIZATION_PENDING: u8 = 0;
511const BACKING_INITIALIZATION_PREPARED: u8 = 1;
512const BACKING_INITIALIZATION_IN_FLIGHT: u8 = 2;
513const BACKING_INITIALIZATION_INITIALIZED: u8 = 3;
514const BACKING_INITIALIZATION_POISONED: u8 = 4;
515
516#[derive(Debug)]
517enum BackingInitializationState {
518 Pending,
519 Prepared { wave_fingerprint: String },
520 InFlight { wave_fingerprint: String },
521 Initialized,
522 Poisoned,
523}
524
525#[derive(Debug)]
526pub(super) struct BackingInitializationCell {
527 target_fingerprint: String,
528 status: AtomicU8,
529 state: Mutex<BackingInitializationState>,
530}
531
532impl BackingInitializationCell {
533 fn new(target_fingerprint: String) -> Self {
534 Self {
535 target_fingerprint,
536 status: AtomicU8::new(BACKING_INITIALIZATION_PENDING),
537 state: Mutex::new(BackingInitializationState::Pending),
538 }
539 }
540
541 pub(super) fn target_fingerprint(&self) -> &str {
542 &self.target_fingerprint
543 }
544
545 pub(super) fn status(&self) -> Result<BackingInitializationStatus, VNextError> {
546 match self.status.load(Ordering::Acquire) {
547 BACKING_INITIALIZATION_PENDING => Ok(BackingInitializationStatus::Pending),
548 BACKING_INITIALIZATION_PREPARED => Ok(BackingInitializationStatus::Prepared),
549 BACKING_INITIALIZATION_IN_FLIGHT => Ok(BackingInitializationStatus::InFlight),
550 BACKING_INITIALIZATION_INITIALIZED => Ok(BackingInitializationStatus::Initialized),
551 BACKING_INITIALIZATION_POISONED => Ok(BackingInitializationStatus::Poisoned),
552 _ => Err(invalid_resource(
553 "backing initialization status contains an invalid value",
554 )),
555 }
556 }
557
558 pub(super) fn prepare(&self, wave_fingerprint: &str) -> Result<bool, VNextError> {
559 let mut state = match self.state.lock() {
560 Ok(state) => state,
561 Err(poisoned) => {
562 let mut state = poisoned.into_inner();
563 *state = BackingInitializationState::Poisoned;
564 self.status
565 .store(BACKING_INITIALIZATION_POISONED, Ordering::Release);
566 return Err(invalid_resource("backing initialization state is poisoned"));
567 }
568 };
569 match &*state {
570 BackingInitializationState::Pending => {
571 *state = BackingInitializationState::Prepared {
572 wave_fingerprint: wave_fingerprint.to_owned(),
573 };
574 self.status
575 .store(BACKING_INITIALIZATION_PREPARED, Ordering::Release);
576 Ok(true)
577 }
578 BackingInitializationState::Initialized => Ok(false),
579 BackingInitializationState::Prepared {
580 wave_fingerprint: current,
581 } if current == wave_fingerprint => Ok(true),
582 BackingInitializationState::Prepared { .. }
583 | BackingInitializationState::InFlight { .. } => Err(invalid_resource(
584 "backing initialization is owned by another submission wave",
585 )),
586 BackingInitializationState::Poisoned => Err(invalid_resource(
587 "backing initialization authority is fail-closed",
588 )),
589 }
590 }
591
592 pub(super) fn mark_in_flight(&self, wave_fingerprint: &str) -> Result<(), VNextError> {
593 let mut state = match self.state.lock() {
594 Ok(state) => state,
595 Err(poisoned) => {
596 let mut state = poisoned.into_inner();
597 *state = BackingInitializationState::Poisoned;
598 self.status
599 .store(BACKING_INITIALIZATION_POISONED, Ordering::Release);
600 return Err(invalid_resource("backing initialization state is poisoned"));
601 }
602 };
603 match &*state {
604 BackingInitializationState::Prepared {
605 wave_fingerprint: current,
606 } if current == wave_fingerprint => {
607 *state = BackingInitializationState::InFlight {
608 wave_fingerprint: wave_fingerprint.to_owned(),
609 };
610 self.status
611 .store(BACKING_INITIALIZATION_IN_FLIGHT, Ordering::Release);
612 Ok(())
613 }
614 _ => {
615 *state = BackingInitializationState::Poisoned;
616 self.status
617 .store(BACKING_INITIALIZATION_POISONED, Ordering::Release);
618 Err(invalid_resource(
619 "backing initialization fence was installed from an invalid state",
620 ))
621 }
622 }
623 }
624
625 pub(super) fn finish(&self, wave_fingerprint: &str, succeeded: bool) -> Result<(), VNextError> {
626 let mut state = match self.state.lock() {
627 Ok(state) => state,
628 Err(poisoned) => {
629 let mut state = poisoned.into_inner();
630 *state = BackingInitializationState::Poisoned;
631 self.status
632 .store(BACKING_INITIALIZATION_POISONED, Ordering::Release);
633 return Err(invalid_resource("backing initialization state is poisoned"));
634 }
635 };
636 match &*state {
637 BackingInitializationState::InFlight {
638 wave_fingerprint: current,
639 } if current == wave_fingerprint => {
640 *state = if succeeded {
641 BackingInitializationState::Initialized
642 } else {
643 BackingInitializationState::Poisoned
644 };
645 self.status.store(
646 if succeeded {
647 BACKING_INITIALIZATION_INITIALIZED
648 } else {
649 BACKING_INITIALIZATION_POISONED
650 },
651 Ordering::Release,
652 );
653 Ok(())
654 }
655 _ => {
656 *state = BackingInitializationState::Poisoned;
657 self.status
658 .store(BACKING_INITIALIZATION_POISONED, Ordering::Release);
659 Err(invalid_resource(
660 "backing initialization completed from an invalid state",
661 ))
662 }
663 }
664 }
665
666 pub(super) fn rollback_prepared(&self, wave_fingerprint: &str) {
667 let mut state = self
668 .state
669 .lock()
670 .unwrap_or_else(std::sync::PoisonError::into_inner);
671 match &*state {
672 BackingInitializationState::Prepared {
673 wave_fingerprint: current,
674 } if current == wave_fingerprint => {
675 *state = BackingInitializationState::Pending;
676 self.status
677 .store(BACKING_INITIALIZATION_PENDING, Ordering::Release);
678 }
679 BackingInitializationState::Initialized | BackingInitializationState::Pending => {}
680 _ => {
681 *state = BackingInitializationState::Poisoned;
682 self.status
683 .store(BACKING_INITIALIZATION_POISONED, Ordering::Release);
684 }
685 }
686 }
687
688 pub(super) fn mark_indeterminate(&self) {
689 let mut state = self
690 .state
691 .lock()
692 .unwrap_or_else(std::sync::PoisonError::into_inner);
693 *state = BackingInitializationState::Poisoned;
694 self.status
695 .store(BACKING_INITIALIZATION_POISONED, Ordering::Release);
696 }
697}
698
699impl Drop for BackingSegmentLease {
700 fn drop(&mut self) {
701 if !self.released {
702 self.owner
703 .release_segments(&self.claim_identity, self.occupancy, &self.segments);
704 self.released = true;
705 }
706 }
707}
708
709impl<R> BackingExtentOwner for DynamicBackingPool<R>
710where
711 R: DeviceRuntime,
712{
713 fn instance_id(&self) -> u64 {
714 self.instance_id
715 }
716
717 fn release_segments(
718 &self,
719 claim_identity: &PhysicalBackingClaimIdentity,
720 occupancy: DynamicBackingClaimOccupancy,
721 segments: &[BackingSegment],
722 ) {
723 let mut state = match self.state.lock() {
724 Ok(state) => state,
725 Err(poisoned) => {
726 let mut state = poisoned.into_inner();
727 state.poisoned = true;
728 return;
729 }
730 };
731 if state.poisoned {
732 return;
733 }
734 let segment_count = u64::try_from(segments.len()).ok();
735 let physical_bytes = segments.iter().try_fold(0_u64, |total, segment| {
736 total.checked_add(segment.length_bytes())
737 });
738 if claim_identity.pool_id() != self.domain.pool_id()
739 || Some(occupancy.segment_count) != segment_count
740 || Some(occupancy.physical_bytes) != physical_bytes
741 {
742 state.poisoned = true;
743 return;
744 }
745 let next_occupancy = match state.live_occupancy.checked_without_claim(occupancy) {
746 Ok(next) => next,
747 Err(_) => {
748 state.poisoned = true;
749 return;
750 }
751 };
752 for segment in segments {
753 if segment.pool_id() != self.domain.pool_id() {
754 state.poisoned = true;
755 return;
756 }
757 let Some(chunk) = state.chunks.get_mut(&segment.chunk_ordinal()) else {
758 state.poisoned = true;
759 return;
760 };
761 if chunk.backing.identity != *segment.chunk() || chunk.live_segments == 0 {
762 state.poisoned = true;
763 return;
764 }
765 }
766 for segment in segments {
767 if state.allocator.release(segment).is_err() {
768 state.poisoned = true;
769 return;
770 }
771 let chunk = state
772 .chunks
773 .get_mut(&segment.chunk_ordinal())
774 .expect("validated released chunk remains installed");
775 chunk.live_segments -= 1;
776 }
777 state.live_occupancy = next_occupancy;
778 drop(state);
779 if self
780 .logical_admission
781 .notify_domain_availability_changed(self.domain.domain_id)
782 .is_err()
783 {
784 let mut state = self
785 .state
786 .lock()
787 .unwrap_or_else(std::sync::PoisonError::into_inner);
788 state.poisoned = true;
789 }
790 }
791}
792
793#[derive(Debug, Clone, PartialEq, Eq, Serialize)]
794pub struct DynamicPoolGrowthReceipt {
795 pub(super) pool_id: DynamicBackingPoolId,
796 pub(super) chunk: BackingChunkIdentity,
797 pub(super) chunk_bytes: u64,
798 pub(super) published_capacity_bytes: u64,
799 pub(super) capacity_epoch: u64,
800}
801
802#[derive(Debug, Clone, PartialEq, Eq, Serialize)]
803pub struct DynamicPoolResourceContract {
804 pub(super) resource_id: ResourceId,
805 pub(super) demand: DynamicResourceDemand,
806 pub(super) lifetime: AllocationLifetime,
807 pub(super) kind: AllocationKind,
808 pub(super) physical_allocation_quantum_bytes: u64,
809 pub(super) initialization: StateInitialization,
810}
811
812impl DynamicPoolResourceContract {
813 fn from_descriptor(descriptor: &DynamicResourceDescriptor) -> Self {
814 Self {
815 resource_id: descriptor.base_resource_id().clone(),
816 demand: descriptor.demand().clone(),
817 lifetime: descriptor.lifetime(),
818 kind: descriptor.kind().clone(),
819 physical_allocation_quantum_bytes: descriptor.physical_allocation_quantum_bytes(),
820 initialization: descriptor.initialization(),
821 }
822 }
823
824 pub fn resource_id(&self) -> &ResourceId {
825 &self.resource_id
826 }
827
828 pub fn demand(&self) -> &DynamicResourceDemand {
829 &self.demand
830 }
831
832 pub const fn lifetime(&self) -> AllocationLifetime {
833 self.lifetime
834 }
835
836 pub fn kind(&self) -> &AllocationKind {
837 &self.kind
838 }
839
840 pub const fn physical_allocation_quantum_bytes(&self) -> u64 {
841 self.physical_allocation_quantum_bytes
842 }
843
844 pub const fn initialization(&self) -> StateInitialization {
845 self.initialization
846 }
847}
848
849#[derive(Debug, Clone, PartialEq, Eq, Serialize)]
850pub struct DynamicPoolContractStatus {
851 pub(super) compatibility: PoolCompatibilityKey,
852 pub(super) resources: Vec<DynamicPoolResourceContract>,
853 pub(super) minimum_request_bytes: u64,
854 pub(super) minimum_sequence_bytes: u64,
855 pub(super) minimum_step_bytes: u64,
856 pub(super) minimum_invocation_peak_bytes: u64,
857 pub(super) reusable_workspace_ceiling_bytes: u64,
858 pub(super) provisioning: DynamicPoolProvisioningPolicy,
859 pub(super) invocation_liveness_mode: InvocationLivenessMode,
860}
861
862impl DynamicPoolContractStatus {
863 pub(super) fn from_domain(domain: &DynamicPoolDomainSpec) -> Self {
864 Self {
865 compatibility: domain.pool.compatibility().clone(),
866 resources: domain
867 .descriptors
868 .iter()
869 .map(DynamicPoolResourceContract::from_descriptor)
870 .collect(),
871 minimum_request_bytes: domain.pool.minimum_request_bytes(),
872 minimum_sequence_bytes: domain.pool.minimum_sequence_bytes(),
873 minimum_step_bytes: domain.pool.minimum_step_bytes(),
874 minimum_invocation_peak_bytes: domain.pool.minimum_invocation_peak_bytes(),
875 reusable_workspace_ceiling_bytes: domain.pool.reusable_workspace_ceiling_bytes(),
876 provisioning: domain.pool.provisioning().clone(),
877 invocation_liveness_mode: domain.pool.invocation_liveness_mode(),
878 }
879 }
880
881 pub fn compatibility(&self) -> &PoolCompatibilityKey {
882 &self.compatibility
883 }
884
885 pub fn resources(&self) -> &[DynamicPoolResourceContract] {
886 &self.resources
887 }
888
889 pub const fn minimum_request_bytes(&self) -> u64 {
890 self.minimum_request_bytes
891 }
892
893 pub const fn minimum_sequence_bytes(&self) -> u64 {
894 self.minimum_sequence_bytes
895 }
896
897 pub const fn minimum_step_bytes(&self) -> u64 {
898 self.minimum_step_bytes
899 }
900
901 pub const fn minimum_invocation_peak_bytes(&self) -> u64 {
902 self.minimum_invocation_peak_bytes
903 }
904
905 pub const fn reusable_workspace_ceiling_bytes(&self) -> u64 {
906 self.reusable_workspace_ceiling_bytes
907 }
908
909 pub fn provisioning(&self) -> &DynamicPoolProvisioningPolicy {
910 &self.provisioning
911 }
912
913 pub const fn invocation_liveness_mode(&self) -> InvocationLivenessMode {
914 self.invocation_liveness_mode
915 }
916}
917
918#[derive(Debug, Clone, PartialEq, Eq, Serialize)]
919pub struct DynamicPoolStatus {
920 pub(super) pool_id: DynamicBackingPoolId,
921 pub(super) domain_id: CapacityDomainId,
922 pub(super) contract: DynamicPoolContractStatus,
923 pub(super) storage_profile: DynamicStorageProfile,
924 pub(super) resident_bytes: u64,
925 pub(super) pending_growth_bytes: u64,
926 pub(super) free_bytes: u64,
927 pub(super) largest_contiguous_bytes: u64,
928 pub(super) resident_chunks: usize,
929 pub(super) live_segments: u64,
930 pub(super) live_occupancy: DynamicPoolLiveOccupancyStatus,
931 pub(super) quarantined_chunks: usize,
932 pub(super) quarantined_bytes: u64,
933 pub(super) descriptor_mismatch_chunks: usize,
934 pub(super) publication_rejected_chunks: usize,
935 pub(super) poisoned: bool,
936}
937
938impl DynamicPoolStatus {
939 pub fn pool_id(&self) -> &DynamicBackingPoolId {
940 &self.pool_id
941 }
942
943 pub const fn domain_id(&self) -> CapacityDomainId {
944 self.domain_id
945 }
946
947 pub fn contract(&self) -> &DynamicPoolContractStatus {
948 &self.contract
949 }
950
951 pub const fn storage_profile(&self) -> DynamicStorageProfile {
952 self.storage_profile
953 }
954
955 pub const fn resident_bytes(&self) -> u64 {
956 self.resident_bytes
957 }
958
959 pub const fn pending_growth_bytes(&self) -> u64 {
960 self.pending_growth_bytes
961 }
962
963 pub const fn free_bytes(&self) -> u64 {
964 self.free_bytes
965 }
966
967 pub const fn largest_contiguous_bytes(&self) -> u64 {
968 self.largest_contiguous_bytes
969 }
970
971 pub const fn resident_chunks(&self) -> usize {
972 self.resident_chunks
973 }
974
975 pub const fn live_segments(&self) -> u64 {
976 self.live_segments
977 }
978
979 pub const fn live_occupancy(&self) -> &DynamicPoolLiveOccupancyStatus {
980 &self.live_occupancy
981 }
982
983 pub const fn quarantined_chunks(&self) -> usize {
984 self.quarantined_chunks
985 }
986
987 pub const fn quarantined_bytes(&self) -> u64 {
988 self.quarantined_bytes
989 }
990
991 pub const fn descriptor_mismatch_chunks(&self) -> usize {
992 self.descriptor_mismatch_chunks
993 }
994
995 pub const fn publication_rejected_chunks(&self) -> usize {
996 self.publication_rejected_chunks
997 }
998
999 pub const fn poisoned(&self) -> bool {
1000 self.poisoned
1001 }
1002}
1003
1004#[derive(Debug, Clone, PartialEq, Eq, Serialize)]
1005pub struct DynamicPoolIdleReclaim {
1006 pub(super) pool_id: DynamicBackingPoolId,
1007 pub(super) chunks: Vec<BackingChunkIdentity>,
1008 pub(super) reclaimed_bytes: u64,
1009 pub(super) published_capacity_bytes: u64,
1010}
1011
1012impl DynamicPoolIdleReclaim {
1013 pub fn pool_id(&self) -> &DynamicBackingPoolId {
1014 &self.pool_id
1015 }
1016
1017 pub fn chunks(&self) -> &[BackingChunkIdentity] {
1018 &self.chunks
1019 }
1020
1021 pub const fn reclaimed_bytes(&self) -> u64 {
1022 self.reclaimed_bytes
1023 }
1024
1025 pub const fn published_capacity_bytes(&self) -> u64 {
1026 self.published_capacity_bytes
1027 }
1028}
1029
1030#[derive(Debug, Clone, PartialEq, Eq, Serialize)]
1031pub struct DynamicPoolRebalanceReceipt {
1032 pub(super) pools: Vec<DynamicPoolIdleReclaim>,
1033 pub(super) reclaimed_chunks: usize,
1034 pub(super) reclaimed_bytes: u64,
1035 pub(super) logical_capacity_epoch: u64,
1036 pub(super) plan_device_capacity_epoch: u64,
1037 pub(super) process_device_capacity_epoch: u64,
1038}
1039
1040impl DynamicPoolRebalanceReceipt {
1041 pub fn pools(&self) -> &[DynamicPoolIdleReclaim] {
1042 &self.pools
1043 }
1044
1045 pub const fn reclaimed_chunks(&self) -> usize {
1046 self.reclaimed_chunks
1047 }
1048
1049 pub const fn reclaimed_bytes(&self) -> u64 {
1050 self.reclaimed_bytes
1051 }
1052
1053 pub const fn logical_capacity_epoch(&self) -> u64 {
1054 self.logical_capacity_epoch
1055 }
1056
1057 pub const fn plan_device_capacity_epoch(&self) -> u64 {
1058 self.plan_device_capacity_epoch
1059 }
1060
1061 pub const fn process_device_capacity_epoch(&self) -> u64 {
1062 self.process_device_capacity_epoch
1063 }
1064}
1065
1066pub(super) struct DynamicDeviceCapacityBlocked {
1067 pub(super) pressure: DeviceCapacityPressure,
1068 pub(super) availability: DeviceCapacityAvailabilitySnapshot,
1069 pub(super) planned_domains: Vec<CapacityDomainId>,
1070}
1071
1072impl DynamicPoolGrowthReceipt {
1073 pub fn pool_id(&self) -> &DynamicBackingPoolId {
1074 &self.pool_id
1075 }
1076
1077 pub fn chunk(&self) -> &BackingChunkIdentity {
1078 &self.chunk
1079 }
1080
1081 pub const fn chunk_bytes(&self) -> u64 {
1082 self.chunk_bytes
1083 }
1084
1085 pub const fn published_capacity_bytes(&self) -> u64 {
1086 self.published_capacity_bytes
1087 }
1088
1089 pub const fn capacity_epoch(&self) -> u64 {
1090 self.capacity_epoch
1091 }
1092}
1093
1094#[derive(Debug, Clone, PartialEq, Eq, Serialize)]
1095pub struct DynamicPoolGrowthRequest {
1096 pool_id: DynamicBackingPoolId,
1097 requested_bytes: u64,
1098}
1099
1100impl DynamicPoolGrowthRequest {
1101 pub fn new(pool_id: DynamicBackingPoolId, requested_bytes: u64) -> Result<Self, VNextError> {
1102 if requested_bytes == 0 {
1103 return Err(invalid_resource(
1104 "dynamic pool growth must request non-zero bytes",
1105 ));
1106 }
1107 Ok(Self {
1108 pool_id,
1109 requested_bytes,
1110 })
1111 }
1112
1113 pub fn pool_id(&self) -> &DynamicBackingPoolId {
1114 &self.pool_id
1115 }
1116
1117 pub const fn requested_bytes(&self) -> u64 {
1118 self.requested_bytes
1119 }
1120}
1121
1122#[derive(Debug, Clone, Copy, PartialEq, Eq, Serialize)]
1123#[serde(rename_all = "snake_case")]
1124pub enum DynamicBackingDeferralReason {
1125 GrowthRequired,
1126 FragmentedContiguous,
1127}
1128
1129#[derive(Debug, Clone, Copy, PartialEq, Eq, Serialize)]
1133#[serde(rename_all = "snake_case")]
1134pub enum DynamicBackingClaimScope {
1135 Plan,
1136 Request,
1137 Sequence,
1138 Checkpoint,
1141 Step,
1142 Invocation,
1143 InitialSequenceBundle,
1144}
1145
1146impl DynamicBackingClaimScope {
1147 pub(super) const fn accepts(self, lifetime: AllocationLifetime) -> bool {
1148 match self {
1149 Self::Plan => matches!(lifetime, AllocationLifetime::Plan),
1150 Self::Request => matches!(lifetime, AllocationLifetime::Request),
1151 Self::Sequence => matches!(lifetime, AllocationLifetime::Sequence),
1152 Self::Checkpoint => matches!(lifetime, AllocationLifetime::Sequence),
1153 Self::Step => matches!(lifetime, AllocationLifetime::Step),
1154 Self::Invocation => matches!(lifetime, AllocationLifetime::Invocation),
1155 Self::InitialSequenceBundle => matches!(
1156 lifetime,
1157 AllocationLifetime::Request | AllocationLifetime::Sequence
1158 ),
1159 }
1160 }
1161
1162 pub const fn lifetime(self) -> Option<AllocationLifetime> {
1163 match self {
1164 Self::Plan => Some(AllocationLifetime::Plan),
1165 Self::Request => Some(AllocationLifetime::Request),
1166 Self::Sequence => Some(AllocationLifetime::Sequence),
1167 Self::Checkpoint => None,
1168 Self::Step => Some(AllocationLifetime::Step),
1169 Self::Invocation => Some(AllocationLifetime::Invocation),
1170 Self::InitialSequenceBundle => None,
1171 }
1172 }
1173}
1174
1175impl From<AllocationLifetime> for DynamicBackingClaimScope {
1176 fn from(lifetime: AllocationLifetime) -> Self {
1177 match lifetime {
1178 AllocationLifetime::Plan => Self::Plan,
1179 AllocationLifetime::Request => Self::Request,
1180 AllocationLifetime::Sequence => Self::Sequence,
1181 AllocationLifetime::Step => Self::Step,
1182 AllocationLifetime::Invocation => Self::Invocation,
1183 }
1184 }
1185}
1186
1187#[derive(Debug, Clone, Copy, PartialEq, Eq, Serialize)]
1188#[serde(rename_all = "snake_case")]
1189pub(super) enum DynamicBackingClaimResidency {
1190 Transient,
1191 LaneStable,
1192}
1193
1194#[derive(Debug, Clone, Copy, Default, PartialEq, Eq, Serialize)]
1195pub struct DynamicPoolOccupancyCounter {
1196 pub(super) claim_count: u64,
1197 pub(super) segment_count: u64,
1198 pub(super) physical_bytes: u64,
1199}
1200
1201impl DynamicPoolOccupancyCounter {
1202 pub const fn claim_count(&self) -> u64 {
1203 self.claim_count
1204 }
1205
1206 pub const fn segment_count(&self) -> u64 {
1207 self.segment_count
1208 }
1209
1210 pub const fn physical_bytes(&self) -> u64 {
1211 self.physical_bytes
1212 }
1213
1214 fn checked_add_claim(&mut self, claim: DynamicBackingClaimOccupancy) -> Result<(), VNextError> {
1215 self.claim_count = self
1216 .claim_count
1217 .checked_add(1)
1218 .ok_or_else(|| invalid_resource("dynamic live claim count overflows u64"))?;
1219 self.segment_count = self
1220 .segment_count
1221 .checked_add(claim.segment_count)
1222 .ok_or_else(|| invalid_resource("dynamic live claim segment count overflows u64"))?;
1223 self.physical_bytes = self
1224 .physical_bytes
1225 .checked_add(claim.physical_bytes)
1226 .ok_or_else(|| invalid_resource("dynamic live claim physical bytes overflow u64"))?;
1227 Ok(())
1228 }
1229
1230 fn checked_remove_claim(
1231 &mut self,
1232 claim: DynamicBackingClaimOccupancy,
1233 ) -> Result<(), VNextError> {
1234 self.claim_count = self
1235 .claim_count
1236 .checked_sub(1)
1237 .ok_or_else(|| invalid_resource("dynamic live claim count underflows u64"))?;
1238 self.segment_count = self
1239 .segment_count
1240 .checked_sub(claim.segment_count)
1241 .ok_or_else(|| invalid_resource("dynamic live claim segment count underflows u64"))?;
1242 self.physical_bytes = self
1243 .physical_bytes
1244 .checked_sub(claim.physical_bytes)
1245 .ok_or_else(|| invalid_resource("dynamic live claim physical bytes underflow u64"))?;
1246 Ok(())
1247 }
1248}
1249
1250#[derive(Debug, Clone, Copy, Default, PartialEq, Eq, Serialize)]
1251pub struct DynamicPoolResidencyOccupancyStatus {
1252 pub(super) total: DynamicPoolOccupancyCounter,
1253 pub(super) plan: DynamicPoolOccupancyCounter,
1254 pub(super) request: DynamicPoolOccupancyCounter,
1255 pub(super) sequence: DynamicPoolOccupancyCounter,
1256 pub(super) checkpoint: DynamicPoolOccupancyCounter,
1257 pub(super) step: DynamicPoolOccupancyCounter,
1258 pub(super) invocation: DynamicPoolOccupancyCounter,
1259 pub(super) initial_sequence_bundle: DynamicPoolOccupancyCounter,
1260}
1261
1262impl DynamicPoolResidencyOccupancyStatus {
1263 pub const fn total(&self) -> &DynamicPoolOccupancyCounter {
1264 &self.total
1265 }
1266
1267 pub const fn plan(&self) -> &DynamicPoolOccupancyCounter {
1268 &self.plan
1269 }
1270
1271 pub const fn request(&self) -> &DynamicPoolOccupancyCounter {
1272 &self.request
1273 }
1274
1275 pub const fn sequence(&self) -> &DynamicPoolOccupancyCounter {
1276 &self.sequence
1277 }
1278
1279 pub const fn checkpoint(&self) -> &DynamicPoolOccupancyCounter {
1280 &self.checkpoint
1281 }
1282
1283 pub const fn step(&self) -> &DynamicPoolOccupancyCounter {
1284 &self.step
1285 }
1286
1287 pub const fn invocation(&self) -> &DynamicPoolOccupancyCounter {
1288 &self.invocation
1289 }
1290
1291 pub const fn initial_sequence_bundle(&self) -> &DynamicPoolOccupancyCounter {
1292 &self.initial_sequence_bundle
1293 }
1294
1295 fn counter_mut_for_scope(
1296 &mut self,
1297 scope: DynamicBackingClaimScope,
1298 ) -> &mut DynamicPoolOccupancyCounter {
1299 match scope {
1300 DynamicBackingClaimScope::Plan => &mut self.plan,
1301 DynamicBackingClaimScope::Request => &mut self.request,
1302 DynamicBackingClaimScope::Sequence => &mut self.sequence,
1303 DynamicBackingClaimScope::Checkpoint => &mut self.checkpoint,
1304 DynamicBackingClaimScope::Step => &mut self.step,
1305 DynamicBackingClaimScope::Invocation => &mut self.invocation,
1306 DynamicBackingClaimScope::InitialSequenceBundle => &mut self.initial_sequence_bundle,
1307 }
1308 }
1309
1310 fn checked_with_claim(&self, claim: DynamicBackingClaimOccupancy) -> Result<Self, VNextError> {
1311 let mut next = *self;
1312 next.total.checked_add_claim(claim)?;
1313 next.counter_mut_for_scope(claim.scope)
1314 .checked_add_claim(claim)?;
1315 Ok(next)
1316 }
1317
1318 fn checked_without_claim(
1319 &self,
1320 claim: DynamicBackingClaimOccupancy,
1321 ) -> Result<Self, VNextError> {
1322 let mut next = *self;
1323 next.total.checked_remove_claim(claim)?;
1324 next.counter_mut_for_scope(claim.scope)
1325 .checked_remove_claim(claim)?;
1326 Ok(next)
1327 }
1328}
1329
1330#[derive(Debug, Clone, Copy, Default, PartialEq, Eq, Serialize)]
1331pub struct DynamicPoolLiveOccupancyStatus {
1332 pub(super) total: DynamicPoolOccupancyCounter,
1333 pub(super) transient: DynamicPoolResidencyOccupancyStatus,
1334 pub(super) lane_stable: DynamicPoolResidencyOccupancyStatus,
1335}
1336
1337impl DynamicPoolLiveOccupancyStatus {
1338 pub const fn total(&self) -> &DynamicPoolOccupancyCounter {
1339 &self.total
1340 }
1341
1342 pub const fn transient(&self) -> &DynamicPoolResidencyOccupancyStatus {
1343 &self.transient
1344 }
1345
1346 pub const fn lane_stable(&self) -> &DynamicPoolResidencyOccupancyStatus {
1347 &self.lane_stable
1348 }
1349
1350 fn residency_mut(
1351 &mut self,
1352 residency: DynamicBackingClaimResidency,
1353 ) -> &mut DynamicPoolResidencyOccupancyStatus {
1354 match residency {
1355 DynamicBackingClaimResidency::Transient => &mut self.transient,
1356 DynamicBackingClaimResidency::LaneStable => &mut self.lane_stable,
1357 }
1358 }
1359
1360 pub(super) fn checked_with_claim(
1361 &self,
1362 claim: DynamicBackingClaimOccupancy,
1363 ) -> Result<Self, VNextError> {
1364 let mut next = *self;
1365 next.total.checked_add_claim(claim)?;
1366 let updated = (*next.residency_mut(claim.residency)).checked_with_claim(claim)?;
1367 *next.residency_mut(claim.residency) = updated;
1368 Ok(next)
1369 }
1370
1371 pub(super) fn checked_without_claim(
1372 &self,
1373 claim: DynamicBackingClaimOccupancy,
1374 ) -> Result<Self, VNextError> {
1375 let mut next = *self;
1376 next.total.checked_remove_claim(claim)?;
1377 let updated = (*next.residency_mut(claim.residency)).checked_without_claim(claim)?;
1378 *next.residency_mut(claim.residency) = updated;
1379 Ok(next)
1380 }
1381}
1382
1383#[derive(Debug, Clone, PartialEq, Eq, Serialize)]
1384pub struct DynamicBackingBlocker {
1385 pub(super) pool_id: DynamicBackingPoolId,
1386 pub(super) domain_id: CapacityDomainId,
1387 pub(super) reason: DynamicBackingDeferralReason,
1388 pub(super) requested_bytes: u64,
1389 pub(super) free_bytes: u64,
1390 pub(super) largest_contiguous_bytes: u64,
1391 pub(super) free_extent_layout_fingerprint: String,
1392 #[serde(skip_serializing_if = "Option::is_none")]
1393 pub(super) contiguous_claim_bytes_descending: Option<Vec<u64>>,
1394}
1395
1396impl DynamicBackingBlocker {
1397 pub fn pool_id(&self) -> &DynamicBackingPoolId {
1398 &self.pool_id
1399 }
1400
1401 pub const fn domain_id(&self) -> CapacityDomainId {
1402 self.domain_id
1403 }
1404
1405 pub const fn reason(&self) -> DynamicBackingDeferralReason {
1406 self.reason
1407 }
1408
1409 pub const fn requested_bytes(&self) -> u64 {
1410 self.requested_bytes
1411 }
1412
1413 pub const fn free_bytes(&self) -> u64 {
1414 self.free_bytes
1415 }
1416
1417 pub const fn largest_contiguous_bytes(&self) -> u64 {
1418 self.largest_contiguous_bytes
1419 }
1420
1421 pub fn free_extent_layout_fingerprint(&self) -> &str {
1422 &self.free_extent_layout_fingerprint
1423 }
1424
1425 pub fn contiguous_claim_bytes_descending(&self) -> Option<&[u64]> {
1426 self.contiguous_claim_bytes_descending.as_deref()
1427 }
1428}
1429
1430#[derive(Debug, Clone, PartialEq, Eq, Serialize)]
1431pub struct DynamicBackingPackingEnvelope {
1432 pub(super) pool_id: DynamicBackingPoolId,
1433 pub(super) domain_id: CapacityDomainId,
1434 pub(super) claim_bytes_descending: Vec<u64>,
1435}
1436
1437impl DynamicBackingPackingEnvelope {
1438 pub(super) fn new(
1439 pool_id: DynamicBackingPoolId,
1440 domain_id: CapacityDomainId,
1441 mut claim_bytes_descending: Vec<u64>,
1442 ) -> Result<Self, VNextError> {
1443 if claim_bytes_descending.is_empty() || claim_bytes_descending.contains(&0) {
1444 return Err(invalid_resource(
1445 "dynamic backing packing envelope contains empty or zero-sized demand",
1446 ));
1447 }
1448 claim_bytes_descending.sort_unstable_by(|left, right| right.cmp(left));
1449 claim_bytes_descending
1450 .iter()
1451 .try_fold(0_u64, |total, bytes| {
1452 total.checked_add(*bytes).ok_or_else(|| {
1453 invalid_resource("dynamic backing packing envelope bytes overflow u64")
1454 })
1455 })?;
1456 Ok(Self {
1457 pool_id,
1458 domain_id,
1459 claim_bytes_descending,
1460 })
1461 }
1462
1463 pub fn pool_id(&self) -> &DynamicBackingPoolId {
1464 &self.pool_id
1465 }
1466
1467 pub const fn domain_id(&self) -> CapacityDomainId {
1468 self.domain_id
1469 }
1470
1471 pub fn claim_bytes_descending(&self) -> &[u64] {
1472 &self.claim_bytes_descending
1473 }
1474
1475 pub(super) fn total_bytes(&self) -> Result<u64, VNextError> {
1476 self.claim_bytes_descending
1477 .iter()
1478 .try_fold(0_u64, |total, bytes| {
1479 total.checked_add(*bytes).ok_or_else(|| {
1480 invalid_resource("dynamic backing packing envelope bytes overflow u64")
1481 })
1482 })
1483 }
1484}
1485
1486#[derive(Debug, Clone, PartialEq, Eq, Serialize)]
1487pub struct DynamicBackingDeferred {
1488 pub(super) blockers: Vec<DynamicBackingBlocker>,
1489 pub(super) epochs: CapacityEpochs,
1490 pub(super) wait_condition: CapacityWaitCondition,
1491 pub(super) scope: DynamicBackingClaimScope,
1492 pub(super) protected_immediate: CapacityVector,
1493 pub(super) protected_packing_envelopes: Vec<DynamicBackingPackingEnvelope>,
1494}
1495
1496impl DynamicBackingDeferred {
1497 pub fn blockers(&self) -> &[DynamicBackingBlocker] {
1498 &self.blockers
1499 }
1500
1501 pub const fn release_epoch(&self) -> u64 {
1502 self.epochs.release_epoch()
1503 }
1504
1505 pub const fn capacity_epoch(&self) -> u64 {
1506 self.epochs.capacity_epoch()
1507 }
1508
1509 pub const fn epochs(&self) -> CapacityEpochs {
1510 self.epochs
1511 }
1512
1513 pub fn wait_condition(&self) -> &CapacityWaitCondition {
1514 &self.wait_condition
1515 }
1516
1517 pub const fn scope(&self) -> DynamicBackingClaimScope {
1518 self.scope
1519 }
1520
1521 pub const fn lifetime(&self) -> Option<AllocationLifetime> {
1522 self.scope.lifetime()
1523 }
1524
1525 pub fn protected_immediate(&self) -> &CapacityVector {
1528 &self.protected_immediate
1529 }
1530
1531 pub fn protected_packing_envelopes(&self) -> &[DynamicBackingPackingEnvelope] {
1534 &self.protected_packing_envelopes
1535 }
1536}
1537
1538#[derive(Clone)]
1539pub(super) enum DynamicPoolGrowthIntent {
1540 Additional(DynamicPoolGrowthRequest),
1541 Minimum(DynamicBackingPoolId),
1542 RevalidatedDeferral(DynamicBackingBlocker),
1543 RevalidatedAdmissionPressure {
1546 pool_id: DynamicBackingPoolId,
1547 required_free_bytes: u64,
1548 },
1549}
1550
1551impl DynamicPoolGrowthIntent {
1552 pub(super) fn pool_id(&self) -> &DynamicBackingPoolId {
1553 match self {
1554 Self::Additional(request) => request.pool_id(),
1555 Self::Minimum(pool_id) => pool_id,
1556 Self::RevalidatedDeferral(blocker) => blocker.pool_id(),
1557 Self::RevalidatedAdmissionPressure { pool_id, .. } => pool_id,
1558 }
1559 }
1560}
1561
1562pub(super) struct PlannedDynamicGrowth<R>
1563where
1564 R: DeviceRuntime,
1565{
1566 pub(super) pool: Arc<DynamicBackingPool<R>>,
1567 pub(super) chunk: BackingChunkIdentity,
1568 pub(super) expected_resource_id: ResourceId,
1569 pub(super) chunk_bytes: u64,
1570}
1571
1572pub(super) struct AllocatedDynamicGrowth<B> {
1573 pub(super) backing: Arc<ResidentChunkBacking<B>>,
1574}
1575
1576#[derive(Clone)]
1577pub(super) struct IdleChunkReclaimCandidate {
1578 pub(super) pool_index: usize,
1579 pub(super) chunk: BackingChunkIdentity,
1580 pub(super) chunk_bytes: u64,
1581}
1582
1583#[derive(Debug, Clone, PartialEq, Eq, PartialOrd, Ord, Serialize)]
1584pub struct PhysicalBackingClaimIdentity {
1585 pool_id: DynamicBackingPoolId,
1586 resource_ids: Vec<ResourceId>,
1587}
1588
1589impl PhysicalBackingClaimIdentity {
1590 pub(super) fn new(
1591 pool_id: DynamicBackingPoolId,
1592 mut resource_ids: Vec<ResourceId>,
1593 ) -> Result<Self, VNextError> {
1594 resource_ids.sort();
1595 if resource_ids.is_empty() || resource_ids.windows(2).any(|pair| pair[0] == pair[1]) {
1596 return Err(invalid_resource(
1597 "physical backing claim identity requires unique logical resources",
1598 ));
1599 }
1600 Ok(Self {
1601 pool_id,
1602 resource_ids,
1603 })
1604 }
1605
1606 pub fn pool_id(&self) -> &DynamicBackingPoolId {
1607 &self.pool_id
1608 }
1609
1610 pub fn resource_ids(&self) -> &[ResourceId] {
1611 &self.resource_ids
1612 }
1613
1614 pub const fn is_shared(&self) -> bool {
1615 self.resource_ids.len() > 1
1616 }
1617}
1618
1619#[derive(Clone)]
1620pub(super) struct EvaluatedBackingProjection<'a> {
1621 pub(super) descriptor: &'a DynamicResourceDescriptor,
1622 pub(super) physical_offset_bytes: u64,
1623 pub(super) logical_size_bytes: u64,
1624 pub(super) capacity_size_bytes: u64,
1625}
1626
1627#[derive(Clone)]
1628pub(super) struct EvaluatedBackingRequest<'a> {
1629 pub(super) domain: &'a DynamicPoolDomainSpec,
1630 pub(super) claim_identity: PhysicalBackingClaimIdentity,
1631 pub(super) capacity_size_bytes: u64,
1632 pub(super) reusable_execution_bucket_id: Option<ReusableExecutionBucketId>,
1633 pub(super) projections: Vec<EvaluatedBackingProjection<'a>>,
1634}
1635
1636pub(super) struct PreparedBackingExtent<R>
1637where
1638 R: DeviceRuntime,
1639{
1640 pub(super) pool: Arc<DynamicBackingPool<R>>,
1641 pub(super) claim_identity: PhysicalBackingClaimIdentity,
1642 pub(super) segment_generation: u64,
1643 pub(super) occupancy: DynamicBackingClaimOccupancy,
1644 pub(super) segments: Vec<BackingSegment>,
1645 pub(super) capacity_size_bytes: u64,
1646 pub(super) projections: Vec<LogicalBackingSliceEvidence>,
1647}
1648
1649pub(super) struct PreparedBackingClaim<R>
1650where
1651 R: DeviceRuntime,
1652{
1653 pub(super) extents: Vec<PreparedBackingExtent<R>>,
1654 pub(super) committed: bool,
1655}
1656
1657impl<R> PreparedBackingClaim<R>
1658where
1659 R: DeviceRuntime,
1660{
1661 pub(super) fn empty() -> Self {
1662 Self {
1663 extents: Vec::new(),
1664 committed: false,
1665 }
1666 }
1667
1668 pub(super) fn commit(mut self) -> Vec<LogicalBackingSliceAuthority> {
1669 let mut slices = Vec::new();
1670 for extent in std::mem::take(&mut self.extents) {
1671 let initialization = extent
1672 .projections
1673 .iter()
1674 .any(|projection| projection.initialization == StateInitialization::Zero)
1675 .then(|| {
1676 Arc::new(BackingInitializationCell::new(
1677 backing_initialization_target_fingerprint(&extent),
1678 ))
1679 });
1680 let owner: Arc<dyn BackingExtentOwner> = extent.pool;
1681 let segment_lease = Arc::new(BackingSegmentLease {
1682 owner_instance_id: owner.instance_id(),
1683 owner,
1684 claim_identity: extent.claim_identity,
1685 segment_generation: extent.segment_generation,
1686 occupancy: extent.occupancy,
1687 segments: extent.segments,
1688 size_bytes: extent.capacity_size_bytes,
1689 initialization,
1690 released: false,
1691 });
1692 slices.extend(extent.projections.into_iter().map(|evidence| {
1693 LogicalBackingSliceAuthority {
1694 evidence,
1695 segment_lease: Arc::clone(&segment_lease),
1696 reusable_lane: None,
1697 }
1698 }));
1699 }
1700 slices.sort_by(|left, right| left.resource_id().cmp(right.resource_id()));
1701 self.committed = true;
1702 slices
1703 }
1704}
1705
1706fn backing_initialization_target_fingerprint<R>(extent: &PreparedBackingExtent<R>) -> String
1707where
1708 R: DeviceRuntime,
1709{
1710 let mut hasher = Sha256::new();
1711 hasher.update(b"ferrum.runtime-vnext.backing-initialization-target.v1\0");
1712 hasher.update(extent.pool.instance_id.to_be_bytes());
1713 hasher.update(extent.segment_generation.to_be_bytes());
1714 hasher.update(extent.claim_identity.pool_id().as_str().as_bytes());
1715 for resource_id in extent.claim_identity.resource_ids() {
1716 hasher.update([0]);
1717 hasher.update(resource_id.as_str().as_bytes());
1718 }
1719 for segment in &extent.segments {
1720 hasher.update(segment.chunk_ordinal().to_be_bytes());
1721 hasher.update(segment.chunk_generation().to_be_bytes());
1722 hasher.update(segment.offset_bytes().to_be_bytes());
1723 hasher.update(segment.length_bytes().to_be_bytes());
1724 }
1725 for projection in extent
1726 .projections
1727 .iter()
1728 .filter(|projection| projection.initialization == StateInitialization::Zero)
1729 {
1730 hasher.update([1]);
1731 hasher.update(projection.resource_id.as_str().as_bytes());
1732 hasher.update(projection.physical_offset_bytes.to_be_bytes());
1733 hasher.update(projection.capacity_size_bytes.to_be_bytes());
1734 }
1735 format!("sha256/{:x}", hasher.finalize())
1736}
1737
1738impl<R> Drop for PreparedBackingClaim<R>
1739where
1740 R: DeviceRuntime,
1741{
1742 fn drop(&mut self) {
1743 if self.committed {
1744 return;
1745 }
1746 for extent in self.extents.iter().rev() {
1747 extent.pool.rollback_prepared(
1748 &extent.claim_identity,
1749 extent.occupancy,
1750 &extent.segments,
1751 );
1752 }
1753 }
1754}
1755
1756pub(super) enum BackingPrepareDecision<R>
1757where
1758 R: DeviceRuntime,
1759{
1760 Prepared(PreparedBackingClaim<R>),
1761 Deferred(DynamicBackingDeferred),
1762}
1763
1764#[derive(Debug)]
1771pub(super) struct BackingClaimCertificate {
1772 allocations: Box<[Arc<LogicalBackingSliceAllocationEvidence>]>,
1773 physical_capacity: CapacityVector,
1774 reusable_execution_bucket_id: Option<ReusableExecutionBucketId>,
1775 physical_claim_count: usize,
1776 has_shared_physical_claims: bool,
1777 fingerprint: String,
1778}
1779
1780#[derive(Debug)]
1781pub(super) struct BoundBackingClaimCertificate {
1782 fingerprint: String,
1783 physical_claim_count: usize,
1784 has_shared_physical_claims: bool,
1785}
1786
1787impl BoundBackingClaimCertificate {
1788 pub(super) fn fingerprint(&self) -> &str {
1789 &self.fingerprint
1790 }
1791
1792 pub(super) const fn physical_claim_count(&self) -> usize {
1793 self.physical_claim_count
1794 }
1795
1796 pub(super) const fn has_shared_physical_claims(&self) -> bool {
1797 self.has_shared_physical_claims
1798 }
1799}
1800
1801impl BackingClaimCertificate {
1802 pub(super) fn from_slices(
1803 backing_slices: &[LogicalBackingSliceAuthority],
1804 ) -> Result<Self, VNextError> {
1805 if backing_slices
1806 .windows(2)
1807 .any(|pair| pair[0].resource_id() >= pair[1].resource_id())
1808 {
1809 return Err(invalid_resource(
1810 "backing claim certificate requires canonical unique logical projections",
1811 ));
1812 }
1813 let reusable_execution_bucket_id = backing_slices
1814 .first()
1815 .and_then(|slice| slice.evidence().reusable_execution_bucket_id())
1816 .cloned();
1817 if backing_slices.iter().any(|slice| {
1818 slice.evidence().reusable_execution_bucket_id() != reusable_execution_bucket_id.as_ref()
1819 }) {
1820 return Err(invalid_resource(
1821 "one backing certificate cannot mix reusable execution buckets",
1822 ));
1823 }
1824
1825 let mut backing_by_domain = BTreeMap::<CapacityDomainId, u64>::new();
1826 let mut physical_claims = BTreeMap::<
1827 PhysicalBackingClaimIdentity,
1828 (Arc<BackingSegmentLease>, CapacityDomainId, u64),
1829 >::new();
1830 let mut has_shared_physical_claims = false;
1831 for slice in backing_slices {
1832 let evidence = slice.evidence();
1833 let claim_identity = evidence.physical_claim_identity();
1834 if claim_identity.pool_id() != evidence.pool_id()
1835 || claim_identity
1836 .resource_ids()
1837 .binary_search(evidence.resource_id())
1838 .is_err()
1839 || slice.segment_lease.claim_identity != *claim_identity
1840 || slice.segment_lease.segment_generation != evidence.segment_generation()
1841 || slice.segment_lease.size_bytes != evidence.physical_size_bytes()
1842 || evidence.size_bytes() == 0
1843 || evidence.size_bytes() > evidence.capacity_size_bytes()
1844 || evidence
1845 .physical_offset_bytes()
1846 .checked_add(evidence.capacity_size_bytes())
1847 .is_none_or(|end| end > evidence.physical_size_bytes())
1848 {
1849 return Err(invalid_resource(
1850 "logical backing projection differs from its physical claim authority",
1851 ));
1852 }
1853 has_shared_physical_claims |= claim_identity.is_shared();
1854 match physical_claims.entry(claim_identity.clone()) {
1855 std::collections::btree_map::Entry::Vacant(entry) => {
1856 let total = backing_by_domain.entry(slice.domain_id()).or_default();
1857 *total = total
1858 .checked_add(evidence.physical_size_bytes())
1859 .ok_or_else(|| {
1860 invalid_resource("certified backing domain bytes overflow u64")
1861 })?;
1862 entry.insert((
1863 Arc::clone(&slice.segment_lease),
1864 slice.domain_id(),
1865 evidence.physical_size_bytes(),
1866 ));
1867 }
1868 std::collections::btree_map::Entry::Occupied(entry) => {
1869 let (lease, domain_id, size_bytes) = entry.get();
1870 if !Arc::ptr_eq(lease, &slice.segment_lease)
1871 || *domain_id != slice.domain_id()
1872 || *size_bytes != evidence.physical_size_bytes()
1873 {
1874 return Err(invalid_resource(
1875 "shared logical projections do not retain one physical claim",
1876 ));
1877 }
1878 }
1879 }
1880 }
1881 let physical_capacity = if backing_by_domain.is_empty() {
1882 CapacityVector::empty()
1883 } else {
1884 CapacityVector::new(
1885 backing_by_domain
1886 .into_iter()
1887 .map(|(domain, bytes)| CapacityEntry::new(domain, CapacityUnits::new(bytes)))
1888 .collect::<Result<Vec<_>, _>>()?,
1889 )?
1890 };
1891 let allocations = backing_slices
1892 .iter()
1893 .map(|slice| Arc::clone(&slice.evidence.allocation))
1894 .collect::<Vec<_>>()
1895 .into_boxed_slice();
1896 let mut hasher = Sha256::new();
1897 hasher.update(b"ferrum.runtime-vnext.backing-claim-certificate.v1\0");
1898 for allocation in &allocations {
1899 let fingerprint = allocation.fingerprint.as_bytes();
1900 hasher.update(
1901 u64::try_from(fingerprint.len())
1902 .map_err(|_| {
1903 invalid_resource(
1904 "backing allocation fingerprint length exceeds portable range",
1905 )
1906 })?
1907 .to_be_bytes(),
1908 );
1909 hasher.update(fingerprint);
1910 }
1911 Ok(Self {
1912 allocations,
1913 physical_capacity,
1914 reusable_execution_bucket_id,
1915 physical_claim_count: physical_claims.len(),
1916 has_shared_physical_claims,
1917 fingerprint: format!("{:x}", hasher.finalize()),
1918 })
1919 }
1920
1921 pub(super) fn bind(
1922 &self,
1923 backing_slices: &[LogicalBackingSliceAuthority],
1924 demand: &super::AdmissionDemand,
1925 ) -> Result<BoundBackingClaimCertificate, VNextError> {
1926 if backing_slices.len() != self.allocations.len() {
1927 return Err(invalid_resource(
1928 "bound backing projection count differs from its physical certificate",
1929 ));
1930 }
1931 let mut hasher = Sha256::new();
1932 hasher.update(b"ferrum.runtime-vnext.bound-backing-claim.v1\0");
1933 hasher.update(self.fingerprint.as_bytes());
1934 for (slice, allocation) in backing_slices.iter().zip(&self.allocations) {
1935 if !Arc::ptr_eq(&slice.evidence.allocation, allocation)
1936 || slice.evidence.size_bytes() == 0
1937 || slice.evidence.size_bytes() > allocation.capacity_size_bytes
1938 {
1939 return Err(invalid_resource(
1940 "bound logical projection differs from its certified allocation",
1941 ));
1942 }
1943 hasher.update(slice.evidence.size_bytes().to_be_bytes());
1944 }
1945 let physical_covers_logical = self.physical_capacity.entries().len()
1946 == demand.immediate_claim().entries().len()
1947 && self.physical_capacity.entries().iter().all(|physical| {
1948 demand
1949 .immediate_claim()
1950 .units_for(physical.domain())
1951 .is_some_and(|logical| physical.units().get() >= logical.get())
1952 });
1953 let claim_matches = if self.reusable_execution_bucket_id.is_some() {
1954 physical_covers_logical
1955 } else {
1956 self.physical_capacity == *demand.immediate_claim()
1957 };
1958 if !claim_matches {
1959 return Err(invalid_resource(
1960 "certified physical backing does not cover the evaluated logical demand",
1961 ));
1962 }
1963 Ok(BoundBackingClaimCertificate {
1964 fingerprint: format!("{:x}", hasher.finalize()),
1965 physical_claim_count: self.physical_claim_count,
1966 has_shared_physical_claims: self.has_shared_physical_claims,
1967 })
1968 }
1969}
1970
1971#[doc(hidden)]
1972#[derive(Debug, PartialEq, Eq, Serialize)]
1973pub struct LogicalBackingSliceAllocationEvidence {
1974 pub(in crate::vnext::resource) domain_id: CapacityDomainId,
1975 pub(in crate::vnext::resource) pool_id: DynamicBackingPoolId,
1976 pub(in crate::vnext::resource) resource_id: ResourceId,
1977 pub(in crate::vnext::resource) pool_instance_id: u64,
1978 pub(in crate::vnext::resource) physical_claim_identity: PhysicalBackingClaimIdentity,
1979 #[serde(skip_serializing_if = "Option::is_none")]
1980 pub(in crate::vnext::resource) reusable_execution_bucket_id: Option<ReusableExecutionBucketId>,
1981 pub(in crate::vnext::resource) segment_generation: u64,
1982 pub(in crate::vnext::resource) segments: Vec<BackingSegment>,
1983 pub(in crate::vnext::resource) physical_offset_bytes: u64,
1984 pub(in crate::vnext::resource) capacity_size_bytes: u64,
1985 pub(in crate::vnext::resource) physical_size_bytes: u64,
1986 pub(in crate::vnext::resource) alignment_bytes: u64,
1987 pub(in crate::vnext::resource) usage: BufferUsage,
1988 pub(in crate::vnext::resource) element_type: ElementType,
1989 pub(in crate::vnext::resource) storage_profile: DynamicStorageProfile,
1990 pub(in crate::vnext::resource) initialization: StateInitialization,
1991 #[serde(skip)]
1992 pub(in crate::vnext::resource) fingerprint: String,
1993}
1994
1995#[derive(Debug, Clone, PartialEq, Eq)]
1996pub struct LogicalBackingSliceEvidence {
1997 pub(super) allocation: Arc<LogicalBackingSliceAllocationEvidence>,
1998 pub(in crate::vnext::resource) logical_size_bytes: u64,
1999}
2000
2001impl std::ops::Deref for LogicalBackingSliceEvidence {
2002 type Target = LogicalBackingSliceAllocationEvidence;
2003
2004 fn deref(&self) -> &Self::Target {
2005 self.allocation.as_ref()
2006 }
2007}
2008
2009impl Serialize for LogicalBackingSliceEvidence {
2010 fn serialize<S>(&self, serializer: S) -> Result<S::Ok, S::Error>
2011 where
2012 S: serde::Serializer,
2013 {
2014 #[derive(Serialize)]
2015 struct Wire<'a> {
2016 domain_id: CapacityDomainId,
2017 pool_id: &'a DynamicBackingPoolId,
2018 resource_id: &'a ResourceId,
2019 pool_instance_id: u64,
2020 physical_claim_identity: &'a PhysicalBackingClaimIdentity,
2021 #[serde(skip_serializing_if = "Option::is_none")]
2022 reusable_execution_bucket_id: Option<&'a ReusableExecutionBucketId>,
2023 segment_generation: u64,
2024 segments: &'a [BackingSegment],
2025 physical_offset_bytes: u64,
2026 #[serde(rename = "size_bytes")]
2027 logical_size_bytes: u64,
2028 capacity_size_bytes: u64,
2029 physical_size_bytes: u64,
2030 alignment_bytes: u64,
2031 usage: BufferUsage,
2032 element_type: ElementType,
2033 storage_profile: DynamicStorageProfile,
2034 initialization: StateInitialization,
2035 }
2036
2037 Wire {
2038 domain_id: self.domain_id,
2039 pool_id: &self.pool_id,
2040 resource_id: &self.resource_id,
2041 pool_instance_id: self.pool_instance_id,
2042 physical_claim_identity: &self.physical_claim_identity,
2043 reusable_execution_bucket_id: self.reusable_execution_bucket_id.as_ref(),
2044 segment_generation: self.segment_generation,
2045 segments: &self.segments,
2046 physical_offset_bytes: self.physical_offset_bytes,
2047 logical_size_bytes: self.logical_size_bytes,
2048 capacity_size_bytes: self.capacity_size_bytes,
2049 physical_size_bytes: self.physical_size_bytes,
2050 alignment_bytes: self.alignment_bytes,
2051 usage: self.usage,
2052 element_type: self.element_type,
2053 storage_profile: self.storage_profile,
2054 initialization: self.initialization,
2055 }
2056 .serialize(serializer)
2057 }
2058}
2059
2060impl LogicalBackingSliceEvidence {
2061 pub fn domain_id(&self) -> CapacityDomainId {
2062 self.domain_id
2063 }
2064
2065 pub fn resource_id(&self) -> &ResourceId {
2066 &self.resource_id
2067 }
2068
2069 pub fn pool_id(&self) -> &DynamicBackingPoolId {
2070 &self.pool_id
2071 }
2072
2073 pub fn pool_instance_id(&self) -> u64 {
2074 self.pool_instance_id
2075 }
2076
2077 pub fn segment_generation(&self) -> u64 {
2078 self.segment_generation
2079 }
2080
2081 pub fn physical_claim_identity(&self) -> &PhysicalBackingClaimIdentity {
2082 &self.physical_claim_identity
2083 }
2084
2085 pub fn reusable_execution_bucket_id(&self) -> Option<&ReusableExecutionBucketId> {
2086 self.reusable_execution_bucket_id.as_ref()
2087 }
2088
2089 pub fn segments(&self) -> &[BackingSegment] {
2090 &self.segments
2091 }
2092
2093 pub fn physical_offset_bytes(&self) -> u64 {
2094 self.physical_offset_bytes
2095 }
2096
2097 pub const fn size_bytes(&self) -> u64 {
2098 self.logical_size_bytes
2099 }
2100
2101 pub fn capacity_size_bytes(&self) -> u64 {
2102 self.capacity_size_bytes
2103 }
2104
2105 pub fn physical_size_bytes(&self) -> u64 {
2106 self.physical_size_bytes
2107 }
2108
2109 pub fn alignment_bytes(&self) -> u64 {
2110 self.alignment_bytes
2111 }
2112
2113 pub fn usage(&self) -> BufferUsage {
2114 self.usage
2115 }
2116
2117 pub fn element_type(&self) -> ElementType {
2118 self.element_type
2119 }
2120
2121 pub fn storage_profile(&self) -> DynamicStorageProfile {
2122 self.storage_profile
2123 }
2124
2125 pub fn initialization(&self) -> StateInitialization {
2126 self.initialization
2127 }
2128}
2129
2130#[must_use = "a logical backing authority owns its physical arena extents"]
2131pub struct LogicalBackingSliceAuthority {
2132 pub(in crate::vnext::resource) evidence: LogicalBackingSliceEvidence,
2133 pub(in crate::vnext::resource) segment_lease: Arc<BackingSegmentLease>,
2134 pub(super) reusable_lane: Option<ExecutionLaneId>,
2135}
2136
2137impl LogicalBackingSliceAuthority {
2138 pub fn evidence(&self) -> &LogicalBackingSliceEvidence {
2139 &self.evidence
2140 }
2141
2142 pub(in crate::vnext::resource) fn retained(&self) -> Self {
2143 Self {
2144 evidence: self.evidence.clone(),
2145 segment_lease: Arc::clone(&self.segment_lease),
2146 reusable_lane: self.reusable_lane,
2147 }
2148 }
2149
2150 pub(in crate::vnext::resource) fn retained_for_lane(&self, lane_id: ExecutionLaneId) -> Self {
2151 Self {
2152 evidence: self.evidence.clone(),
2153 segment_lease: Arc::clone(&self.segment_lease),
2154 reusable_lane: Some(lane_id),
2155 }
2156 }
2157
2158 pub(crate) const fn reusable_address_scope(&self) -> Option<DeviceReusableAddressScope> {
2159 match self.reusable_lane {
2160 Some(lane_id) => Some(DeviceReusableAddressScope::ExecutionLane(lane_id)),
2161 None => None,
2162 }
2163 }
2164
2165 pub fn domain_id(&self) -> CapacityDomainId {
2166 self.evidence.domain_id
2167 }
2168
2169 pub fn resource_id(&self) -> &ResourceId {
2170 &self.evidence.resource_id
2171 }
2172
2173 pub const fn size_bytes(&self) -> u64 {
2174 self.evidence.logical_size_bytes
2175 }
2176
2177 pub fn capacity_size_bytes(&self) -> u64 {
2178 self.evidence.capacity_size_bytes
2179 }
2180
2181 pub fn initialization_status(&self) -> Result<Option<BackingInitializationStatus>, VNextError> {
2182 self.segment_lease
2183 .initialization
2184 .as_ref()
2185 .map(|cell| cell.status())
2186 .transpose()
2187 }
2188
2189 pub(in crate::vnext::resource) fn initialization_cell(
2190 &self,
2191 ) -> Option<&Arc<BackingInitializationCell>> {
2192 self.segment_lease.initialization.as_ref()
2193 }
2194}
2195
2196pub struct LogicalBackingBufferView<'a, B> {
2197 pub(in crate::vnext::resource) bindings: Vec<LogicalBackingSegmentBinding<B>>,
2198 pub(super) authorities: &'a [LogicalBackingSliceAuthority],
2199 pub(super) logical_size_bytes: u64,
2200 pub(super) capacity_size_bytes: u64,
2201 pub(super) alignment_bytes: u64,
2202 pub(super) usage: BufferUsage,
2203 pub(super) element_type: ElementType,
2204 pub(super) storage_profile: DynamicStorageProfile,
2205}
2206
2207pub(crate) struct LogicalBackingSegmentBinding<B> {
2208 pub(in crate::vnext::resource) segment: BackingSegment,
2209 pub(in crate::vnext::resource) chunk: Arc<ResidentChunkBacking<B>>,
2210 pub(in crate::vnext::resource) retention: DeviceBufferRetention,
2211}
2212
2213impl<B> LogicalBackingSegmentBinding<B> {
2214 pub(crate) fn segment(&self) -> &BackingSegment {
2215 &self.segment
2216 }
2217
2218 pub(crate) fn chunk(&self) -> &BackingChunkIdentity {
2219 self.segment.chunk()
2220 }
2221
2222 pub(crate) fn buffer(&self) -> &B {
2223 &self.chunk.buffer
2224 }
2225
2226 pub(crate) fn descriptor(&self) -> &BufferDescriptor {
2227 &self.chunk.descriptor
2228 }
2229
2230 pub(crate) fn retention(&self) -> DeviceBufferRetention {
2231 self.retention.clone()
2232 }
2233}
2234
2235impl<'a, B> LogicalBackingBufferView<'a, B> {
2236 pub(crate) fn segment_bindings(&self) -> &[LogicalBackingSegmentBinding<B>] {
2237 &self.bindings
2238 }
2239
2240 pub const fn size_bytes(&self) -> u64 {
2241 self.logical_size_bytes
2242 }
2243
2244 pub const fn capacity_size_bytes(&self) -> u64 {
2245 self.capacity_size_bytes
2246 }
2247
2248 pub const fn alignment_bytes(&self) -> u64 {
2249 self.alignment_bytes
2250 }
2251
2252 pub const fn usage(&self) -> BufferUsage {
2253 self.usage
2254 }
2255
2256 pub const fn element_type(&self) -> ElementType {
2257 self.element_type
2258 }
2259
2260 pub const fn storage_profile(&self) -> DynamicStorageProfile {
2261 self.storage_profile
2262 }
2263
2264 pub fn committed_evidence_segments(&self) -> impl Iterator<Item = &BackingSegment> {
2265 self.authorities
2266 .iter()
2267 .flat_map(|authority| authority.evidence.segments())
2268 }
2269
2270 pub fn slice(&self) -> &'a LogicalBackingSliceEvidence {
2273 &self
2274 .authorities
2275 .first()
2276 .expect("logical backing views always contain an authority")
2277 .evidence
2278 }
2279}
2280
2281impl<R> DynamicBackingPool<R>
2282where
2283 R: DeviceRuntime,
2284{
2285 pub(super) fn allocation_quantum(&self) -> u64 {
2286 match self.domain.pool.compatibility().profile().allocator() {
2287 DynamicStorageAllocator::LinearArena => {
2288 self.domain.pool.compatibility().alignment_bytes()
2289 }
2290 DynamicStorageAllocator::FixedBlockArena { block_bytes } => {
2291 block_bytes.max(self.domain.pool.compatibility().alignment_bytes())
2292 }
2293 }
2294 }
2295
2296 fn cancel_pending_growth(&self, bytes: u64) {
2297 let mut state = match self.state.lock() {
2298 Ok(state) => state,
2299 Err(poisoned) => poisoned.into_inner(),
2300 };
2301 if state.pending_growth_bytes < bytes {
2302 state.poisoned = true;
2303 return;
2304 }
2305 state.pending_growth_bytes -= bytes;
2306 }
2307
2308 fn rollback_prepared(
2309 &self,
2310 claim_identity: &PhysicalBackingClaimIdentity,
2311 occupancy: DynamicBackingClaimOccupancy,
2312 segments: &[BackingSegment],
2313 ) {
2314 let mut state = match self.state.lock() {
2315 Ok(state) => state,
2316 Err(poisoned) => {
2317 let mut state = poisoned.into_inner();
2318 state.poisoned = true;
2319 return;
2320 }
2321 };
2322 if state.poisoned {
2323 return;
2324 }
2325 let segment_count = u64::try_from(segments.len()).ok();
2326 let physical_bytes = segments.iter().try_fold(0_u64, |total, segment| {
2327 total.checked_add(segment.length_bytes())
2328 });
2329 if claim_identity.pool_id() != self.domain.pool_id()
2330 || Some(occupancy.segment_count) != segment_count
2331 || Some(occupancy.physical_bytes) != physical_bytes
2332 {
2333 state.poisoned = true;
2334 return;
2335 }
2336 let next_occupancy = match state.live_occupancy.checked_without_claim(occupancy) {
2337 Ok(next) => next,
2338 Err(_) => {
2339 state.poisoned = true;
2340 return;
2341 }
2342 };
2343 for segment in segments.iter().rev() {
2344 let valid = state
2345 .chunks
2346 .get(&segment.chunk_ordinal())
2347 .is_some_and(|chunk| {
2348 chunk.backing.identity == *segment.chunk() && chunk.live_segments != 0
2349 });
2350 if !valid || state.allocator.release(segment).is_err() {
2351 state.poisoned = true;
2352 return;
2353 }
2354 state
2355 .chunks
2356 .get_mut(&segment.chunk_ordinal())
2357 .expect("validated prepared chunk remains installed")
2358 .live_segments -= 1;
2359 }
2360 state.live_occupancy = next_occupancy;
2361 drop(state);
2362 if self
2363 .logical_admission
2364 .notify_domain_availability_changed(self.domain.domain_id)
2365 .is_err()
2366 {
2367 let mut state = self
2368 .state
2369 .lock()
2370 .unwrap_or_else(std::sync::PoisonError::into_inner);
2371 state.poisoned = true;
2372 }
2373 }
2374}