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 Step,
1139 Invocation,
1140 InitialSequenceBundle,
1141}
1142
1143impl DynamicBackingClaimScope {
1144 pub(super) const fn accepts(self, lifetime: AllocationLifetime) -> bool {
1145 match self {
1146 Self::Plan => matches!(lifetime, AllocationLifetime::Plan),
1147 Self::Request => matches!(lifetime, AllocationLifetime::Request),
1148 Self::Sequence => matches!(lifetime, AllocationLifetime::Sequence),
1149 Self::Step => matches!(lifetime, AllocationLifetime::Step),
1150 Self::Invocation => matches!(lifetime, AllocationLifetime::Invocation),
1151 Self::InitialSequenceBundle => matches!(
1152 lifetime,
1153 AllocationLifetime::Request | AllocationLifetime::Sequence
1154 ),
1155 }
1156 }
1157
1158 pub const fn lifetime(self) -> Option<AllocationLifetime> {
1159 match self {
1160 Self::Plan => Some(AllocationLifetime::Plan),
1161 Self::Request => Some(AllocationLifetime::Request),
1162 Self::Sequence => Some(AllocationLifetime::Sequence),
1163 Self::Step => Some(AllocationLifetime::Step),
1164 Self::Invocation => Some(AllocationLifetime::Invocation),
1165 Self::InitialSequenceBundle => None,
1166 }
1167 }
1168}
1169
1170impl From<AllocationLifetime> for DynamicBackingClaimScope {
1171 fn from(lifetime: AllocationLifetime) -> Self {
1172 match lifetime {
1173 AllocationLifetime::Plan => Self::Plan,
1174 AllocationLifetime::Request => Self::Request,
1175 AllocationLifetime::Sequence => Self::Sequence,
1176 AllocationLifetime::Step => Self::Step,
1177 AllocationLifetime::Invocation => Self::Invocation,
1178 }
1179 }
1180}
1181
1182#[derive(Debug, Clone, Copy, PartialEq, Eq, Serialize)]
1183#[serde(rename_all = "snake_case")]
1184pub(super) enum DynamicBackingClaimResidency {
1185 Transient,
1186 LaneStable,
1187}
1188
1189#[derive(Debug, Clone, Copy, Default, PartialEq, Eq, Serialize)]
1190pub struct DynamicPoolOccupancyCounter {
1191 pub(super) claim_count: u64,
1192 pub(super) segment_count: u64,
1193 pub(super) physical_bytes: u64,
1194}
1195
1196impl DynamicPoolOccupancyCounter {
1197 pub const fn claim_count(&self) -> u64 {
1198 self.claim_count
1199 }
1200
1201 pub const fn segment_count(&self) -> u64 {
1202 self.segment_count
1203 }
1204
1205 pub const fn physical_bytes(&self) -> u64 {
1206 self.physical_bytes
1207 }
1208
1209 fn checked_add_claim(&mut self, claim: DynamicBackingClaimOccupancy) -> Result<(), VNextError> {
1210 self.claim_count = self
1211 .claim_count
1212 .checked_add(1)
1213 .ok_or_else(|| invalid_resource("dynamic live claim count overflows u64"))?;
1214 self.segment_count = self
1215 .segment_count
1216 .checked_add(claim.segment_count)
1217 .ok_or_else(|| invalid_resource("dynamic live claim segment count overflows u64"))?;
1218 self.physical_bytes = self
1219 .physical_bytes
1220 .checked_add(claim.physical_bytes)
1221 .ok_or_else(|| invalid_resource("dynamic live claim physical bytes overflow u64"))?;
1222 Ok(())
1223 }
1224
1225 fn checked_remove_claim(
1226 &mut self,
1227 claim: DynamicBackingClaimOccupancy,
1228 ) -> Result<(), VNextError> {
1229 self.claim_count = self
1230 .claim_count
1231 .checked_sub(1)
1232 .ok_or_else(|| invalid_resource("dynamic live claim count underflows u64"))?;
1233 self.segment_count = self
1234 .segment_count
1235 .checked_sub(claim.segment_count)
1236 .ok_or_else(|| invalid_resource("dynamic live claim segment count underflows u64"))?;
1237 self.physical_bytes = self
1238 .physical_bytes
1239 .checked_sub(claim.physical_bytes)
1240 .ok_or_else(|| invalid_resource("dynamic live claim physical bytes underflow u64"))?;
1241 Ok(())
1242 }
1243}
1244
1245#[derive(Debug, Clone, Copy, Default, PartialEq, Eq, Serialize)]
1246pub struct DynamicPoolResidencyOccupancyStatus {
1247 pub(super) total: DynamicPoolOccupancyCounter,
1248 pub(super) plan: DynamicPoolOccupancyCounter,
1249 pub(super) request: DynamicPoolOccupancyCounter,
1250 pub(super) sequence: DynamicPoolOccupancyCounter,
1251 pub(super) step: DynamicPoolOccupancyCounter,
1252 pub(super) invocation: DynamicPoolOccupancyCounter,
1253 pub(super) initial_sequence_bundle: DynamicPoolOccupancyCounter,
1254}
1255
1256impl DynamicPoolResidencyOccupancyStatus {
1257 pub const fn total(&self) -> &DynamicPoolOccupancyCounter {
1258 &self.total
1259 }
1260
1261 pub const fn plan(&self) -> &DynamicPoolOccupancyCounter {
1262 &self.plan
1263 }
1264
1265 pub const fn request(&self) -> &DynamicPoolOccupancyCounter {
1266 &self.request
1267 }
1268
1269 pub const fn sequence(&self) -> &DynamicPoolOccupancyCounter {
1270 &self.sequence
1271 }
1272
1273 pub const fn step(&self) -> &DynamicPoolOccupancyCounter {
1274 &self.step
1275 }
1276
1277 pub const fn invocation(&self) -> &DynamicPoolOccupancyCounter {
1278 &self.invocation
1279 }
1280
1281 pub const fn initial_sequence_bundle(&self) -> &DynamicPoolOccupancyCounter {
1282 &self.initial_sequence_bundle
1283 }
1284
1285 fn counter_mut_for_scope(
1286 &mut self,
1287 scope: DynamicBackingClaimScope,
1288 ) -> &mut DynamicPoolOccupancyCounter {
1289 match scope {
1290 DynamicBackingClaimScope::Plan => &mut self.plan,
1291 DynamicBackingClaimScope::Request => &mut self.request,
1292 DynamicBackingClaimScope::Sequence => &mut self.sequence,
1293 DynamicBackingClaimScope::Step => &mut self.step,
1294 DynamicBackingClaimScope::Invocation => &mut self.invocation,
1295 DynamicBackingClaimScope::InitialSequenceBundle => &mut self.initial_sequence_bundle,
1296 }
1297 }
1298
1299 fn checked_with_claim(&self, claim: DynamicBackingClaimOccupancy) -> Result<Self, VNextError> {
1300 let mut next = *self;
1301 next.total.checked_add_claim(claim)?;
1302 next.counter_mut_for_scope(claim.scope)
1303 .checked_add_claim(claim)?;
1304 Ok(next)
1305 }
1306
1307 fn checked_without_claim(
1308 &self,
1309 claim: DynamicBackingClaimOccupancy,
1310 ) -> Result<Self, VNextError> {
1311 let mut next = *self;
1312 next.total.checked_remove_claim(claim)?;
1313 next.counter_mut_for_scope(claim.scope)
1314 .checked_remove_claim(claim)?;
1315 Ok(next)
1316 }
1317}
1318
1319#[derive(Debug, Clone, Copy, Default, PartialEq, Eq, Serialize)]
1320pub struct DynamicPoolLiveOccupancyStatus {
1321 pub(super) total: DynamicPoolOccupancyCounter,
1322 pub(super) transient: DynamicPoolResidencyOccupancyStatus,
1323 pub(super) lane_stable: DynamicPoolResidencyOccupancyStatus,
1324}
1325
1326impl DynamicPoolLiveOccupancyStatus {
1327 pub const fn total(&self) -> &DynamicPoolOccupancyCounter {
1328 &self.total
1329 }
1330
1331 pub const fn transient(&self) -> &DynamicPoolResidencyOccupancyStatus {
1332 &self.transient
1333 }
1334
1335 pub const fn lane_stable(&self) -> &DynamicPoolResidencyOccupancyStatus {
1336 &self.lane_stable
1337 }
1338
1339 fn residency_mut(
1340 &mut self,
1341 residency: DynamicBackingClaimResidency,
1342 ) -> &mut DynamicPoolResidencyOccupancyStatus {
1343 match residency {
1344 DynamicBackingClaimResidency::Transient => &mut self.transient,
1345 DynamicBackingClaimResidency::LaneStable => &mut self.lane_stable,
1346 }
1347 }
1348
1349 pub(super) fn checked_with_claim(
1350 &self,
1351 claim: DynamicBackingClaimOccupancy,
1352 ) -> Result<Self, VNextError> {
1353 let mut next = *self;
1354 next.total.checked_add_claim(claim)?;
1355 let updated = (*next.residency_mut(claim.residency)).checked_with_claim(claim)?;
1356 *next.residency_mut(claim.residency) = updated;
1357 Ok(next)
1358 }
1359
1360 pub(super) fn checked_without_claim(
1361 &self,
1362 claim: DynamicBackingClaimOccupancy,
1363 ) -> Result<Self, VNextError> {
1364 let mut next = *self;
1365 next.total.checked_remove_claim(claim)?;
1366 let updated = (*next.residency_mut(claim.residency)).checked_without_claim(claim)?;
1367 *next.residency_mut(claim.residency) = updated;
1368 Ok(next)
1369 }
1370}
1371
1372#[derive(Debug, Clone, PartialEq, Eq, Serialize)]
1373pub struct DynamicBackingBlocker {
1374 pub(super) pool_id: DynamicBackingPoolId,
1375 pub(super) domain_id: CapacityDomainId,
1376 pub(super) reason: DynamicBackingDeferralReason,
1377 pub(super) requested_bytes: u64,
1378 pub(super) free_bytes: u64,
1379 pub(super) largest_contiguous_bytes: u64,
1380 pub(super) free_extent_layout_fingerprint: String,
1381 #[serde(skip_serializing_if = "Option::is_none")]
1382 pub(super) contiguous_claim_bytes_descending: Option<Vec<u64>>,
1383}
1384
1385impl DynamicBackingBlocker {
1386 pub fn pool_id(&self) -> &DynamicBackingPoolId {
1387 &self.pool_id
1388 }
1389
1390 pub const fn domain_id(&self) -> CapacityDomainId {
1391 self.domain_id
1392 }
1393
1394 pub const fn reason(&self) -> DynamicBackingDeferralReason {
1395 self.reason
1396 }
1397
1398 pub const fn requested_bytes(&self) -> u64 {
1399 self.requested_bytes
1400 }
1401
1402 pub const fn free_bytes(&self) -> u64 {
1403 self.free_bytes
1404 }
1405
1406 pub const fn largest_contiguous_bytes(&self) -> u64 {
1407 self.largest_contiguous_bytes
1408 }
1409
1410 pub fn free_extent_layout_fingerprint(&self) -> &str {
1411 &self.free_extent_layout_fingerprint
1412 }
1413
1414 pub fn contiguous_claim_bytes_descending(&self) -> Option<&[u64]> {
1415 self.contiguous_claim_bytes_descending.as_deref()
1416 }
1417}
1418
1419#[derive(Debug, Clone, PartialEq, Eq, Serialize)]
1420pub struct DynamicBackingPackingEnvelope {
1421 pub(super) pool_id: DynamicBackingPoolId,
1422 pub(super) domain_id: CapacityDomainId,
1423 pub(super) claim_bytes_descending: Vec<u64>,
1424}
1425
1426impl DynamicBackingPackingEnvelope {
1427 pub(super) fn new(
1428 pool_id: DynamicBackingPoolId,
1429 domain_id: CapacityDomainId,
1430 mut claim_bytes_descending: Vec<u64>,
1431 ) -> Result<Self, VNextError> {
1432 if claim_bytes_descending.is_empty() || claim_bytes_descending.contains(&0) {
1433 return Err(invalid_resource(
1434 "dynamic backing packing envelope contains empty or zero-sized demand",
1435 ));
1436 }
1437 claim_bytes_descending.sort_unstable_by(|left, right| right.cmp(left));
1438 claim_bytes_descending
1439 .iter()
1440 .try_fold(0_u64, |total, bytes| {
1441 total.checked_add(*bytes).ok_or_else(|| {
1442 invalid_resource("dynamic backing packing envelope bytes overflow u64")
1443 })
1444 })?;
1445 Ok(Self {
1446 pool_id,
1447 domain_id,
1448 claim_bytes_descending,
1449 })
1450 }
1451
1452 pub fn pool_id(&self) -> &DynamicBackingPoolId {
1453 &self.pool_id
1454 }
1455
1456 pub const fn domain_id(&self) -> CapacityDomainId {
1457 self.domain_id
1458 }
1459
1460 pub fn claim_bytes_descending(&self) -> &[u64] {
1461 &self.claim_bytes_descending
1462 }
1463
1464 pub(super) fn total_bytes(&self) -> Result<u64, VNextError> {
1465 self.claim_bytes_descending
1466 .iter()
1467 .try_fold(0_u64, |total, bytes| {
1468 total.checked_add(*bytes).ok_or_else(|| {
1469 invalid_resource("dynamic backing packing envelope bytes overflow u64")
1470 })
1471 })
1472 }
1473}
1474
1475#[derive(Debug, Clone, PartialEq, Eq, Serialize)]
1476pub struct DynamicBackingDeferred {
1477 pub(super) blockers: Vec<DynamicBackingBlocker>,
1478 pub(super) epochs: CapacityEpochs,
1479 pub(super) wait_condition: CapacityWaitCondition,
1480 pub(super) scope: DynamicBackingClaimScope,
1481 pub(super) protected_immediate: CapacityVector,
1482 pub(super) protected_packing_envelopes: Vec<DynamicBackingPackingEnvelope>,
1483}
1484
1485impl DynamicBackingDeferred {
1486 pub fn blockers(&self) -> &[DynamicBackingBlocker] {
1487 &self.blockers
1488 }
1489
1490 pub const fn release_epoch(&self) -> u64 {
1491 self.epochs.release_epoch()
1492 }
1493
1494 pub const fn capacity_epoch(&self) -> u64 {
1495 self.epochs.capacity_epoch()
1496 }
1497
1498 pub const fn epochs(&self) -> CapacityEpochs {
1499 self.epochs
1500 }
1501
1502 pub fn wait_condition(&self) -> &CapacityWaitCondition {
1503 &self.wait_condition
1504 }
1505
1506 pub const fn scope(&self) -> DynamicBackingClaimScope {
1507 self.scope
1508 }
1509
1510 pub const fn lifetime(&self) -> Option<AllocationLifetime> {
1511 self.scope.lifetime()
1512 }
1513
1514 pub fn protected_immediate(&self) -> &CapacityVector {
1517 &self.protected_immediate
1518 }
1519
1520 pub fn protected_packing_envelopes(&self) -> &[DynamicBackingPackingEnvelope] {
1523 &self.protected_packing_envelopes
1524 }
1525}
1526
1527#[derive(Clone)]
1528pub(super) enum DynamicPoolGrowthIntent {
1529 Additional(DynamicPoolGrowthRequest),
1530 Minimum(DynamicBackingPoolId),
1531 RevalidatedDeferral(DynamicBackingBlocker),
1532}
1533
1534impl DynamicPoolGrowthIntent {
1535 pub(super) fn pool_id(&self) -> &DynamicBackingPoolId {
1536 match self {
1537 Self::Additional(request) => request.pool_id(),
1538 Self::Minimum(pool_id) => pool_id,
1539 Self::RevalidatedDeferral(blocker) => blocker.pool_id(),
1540 }
1541 }
1542}
1543
1544pub(super) struct PlannedDynamicGrowth<R>
1545where
1546 R: DeviceRuntime,
1547{
1548 pub(super) pool: Arc<DynamicBackingPool<R>>,
1549 pub(super) chunk: BackingChunkIdentity,
1550 pub(super) expected_resource_id: ResourceId,
1551 pub(super) chunk_bytes: u64,
1552}
1553
1554pub(super) struct AllocatedDynamicGrowth<B> {
1555 pub(super) backing: Arc<ResidentChunkBacking<B>>,
1556}
1557
1558#[derive(Clone)]
1559pub(super) struct IdleChunkReclaimCandidate {
1560 pub(super) pool_index: usize,
1561 pub(super) chunk: BackingChunkIdentity,
1562 pub(super) chunk_bytes: u64,
1563}
1564
1565#[derive(Debug, Clone, PartialEq, Eq, PartialOrd, Ord, Serialize)]
1566pub struct PhysicalBackingClaimIdentity {
1567 pool_id: DynamicBackingPoolId,
1568 resource_ids: Vec<ResourceId>,
1569}
1570
1571impl PhysicalBackingClaimIdentity {
1572 pub(super) fn new(
1573 pool_id: DynamicBackingPoolId,
1574 mut resource_ids: Vec<ResourceId>,
1575 ) -> Result<Self, VNextError> {
1576 resource_ids.sort();
1577 if resource_ids.is_empty() || resource_ids.windows(2).any(|pair| pair[0] == pair[1]) {
1578 return Err(invalid_resource(
1579 "physical backing claim identity requires unique logical resources",
1580 ));
1581 }
1582 Ok(Self {
1583 pool_id,
1584 resource_ids,
1585 })
1586 }
1587
1588 pub fn pool_id(&self) -> &DynamicBackingPoolId {
1589 &self.pool_id
1590 }
1591
1592 pub fn resource_ids(&self) -> &[ResourceId] {
1593 &self.resource_ids
1594 }
1595
1596 pub const fn is_shared(&self) -> bool {
1597 self.resource_ids.len() > 1
1598 }
1599}
1600
1601#[derive(Clone)]
1602pub(super) struct EvaluatedBackingProjection<'a> {
1603 pub(super) descriptor: &'a DynamicResourceDescriptor,
1604 pub(super) physical_offset_bytes: u64,
1605 pub(super) logical_size_bytes: u64,
1606 pub(super) capacity_size_bytes: u64,
1607}
1608
1609#[derive(Clone)]
1610pub(super) struct EvaluatedBackingRequest<'a> {
1611 pub(super) domain: &'a DynamicPoolDomainSpec,
1612 pub(super) claim_identity: PhysicalBackingClaimIdentity,
1613 pub(super) capacity_size_bytes: u64,
1614 pub(super) reusable_execution_bucket_id: Option<ReusableExecutionBucketId>,
1615 pub(super) projections: Vec<EvaluatedBackingProjection<'a>>,
1616}
1617
1618pub(super) struct PreparedBackingExtent<R>
1619where
1620 R: DeviceRuntime,
1621{
1622 pub(super) pool: Arc<DynamicBackingPool<R>>,
1623 pub(super) claim_identity: PhysicalBackingClaimIdentity,
1624 pub(super) segment_generation: u64,
1625 pub(super) occupancy: DynamicBackingClaimOccupancy,
1626 pub(super) segments: Vec<BackingSegment>,
1627 pub(super) capacity_size_bytes: u64,
1628 pub(super) projections: Vec<LogicalBackingSliceEvidence>,
1629}
1630
1631pub(super) struct PreparedBackingClaim<R>
1632where
1633 R: DeviceRuntime,
1634{
1635 pub(super) extents: Vec<PreparedBackingExtent<R>>,
1636 pub(super) committed: bool,
1637}
1638
1639impl<R> PreparedBackingClaim<R>
1640where
1641 R: DeviceRuntime,
1642{
1643 pub(super) fn empty() -> Self {
1644 Self {
1645 extents: Vec::new(),
1646 committed: false,
1647 }
1648 }
1649
1650 pub(super) fn commit(mut self) -> Vec<LogicalBackingSliceAuthority> {
1651 let mut slices = Vec::new();
1652 for extent in std::mem::take(&mut self.extents) {
1653 let initialization = extent
1654 .projections
1655 .iter()
1656 .any(|projection| projection.initialization == StateInitialization::Zero)
1657 .then(|| {
1658 Arc::new(BackingInitializationCell::new(
1659 backing_initialization_target_fingerprint(&extent),
1660 ))
1661 });
1662 let owner: Arc<dyn BackingExtentOwner> = extent.pool;
1663 let segment_lease = Arc::new(BackingSegmentLease {
1664 owner_instance_id: owner.instance_id(),
1665 owner,
1666 claim_identity: extent.claim_identity,
1667 segment_generation: extent.segment_generation,
1668 occupancy: extent.occupancy,
1669 segments: extent.segments,
1670 size_bytes: extent.capacity_size_bytes,
1671 initialization,
1672 released: false,
1673 });
1674 slices.extend(extent.projections.into_iter().map(|evidence| {
1675 LogicalBackingSliceAuthority {
1676 evidence,
1677 segment_lease: Arc::clone(&segment_lease),
1678 reusable_lane: None,
1679 }
1680 }));
1681 }
1682 slices.sort_by(|left, right| left.resource_id().cmp(right.resource_id()));
1683 self.committed = true;
1684 slices
1685 }
1686}
1687
1688fn backing_initialization_target_fingerprint<R>(extent: &PreparedBackingExtent<R>) -> String
1689where
1690 R: DeviceRuntime,
1691{
1692 let mut hasher = Sha256::new();
1693 hasher.update(b"ferrum.runtime-vnext.backing-initialization-target.v1\0");
1694 hasher.update(extent.pool.instance_id.to_be_bytes());
1695 hasher.update(extent.segment_generation.to_be_bytes());
1696 hasher.update(extent.claim_identity.pool_id().as_str().as_bytes());
1697 for resource_id in extent.claim_identity.resource_ids() {
1698 hasher.update([0]);
1699 hasher.update(resource_id.as_str().as_bytes());
1700 }
1701 for segment in &extent.segments {
1702 hasher.update(segment.chunk_ordinal().to_be_bytes());
1703 hasher.update(segment.chunk_generation().to_be_bytes());
1704 hasher.update(segment.offset_bytes().to_be_bytes());
1705 hasher.update(segment.length_bytes().to_be_bytes());
1706 }
1707 for projection in extent
1708 .projections
1709 .iter()
1710 .filter(|projection| projection.initialization == StateInitialization::Zero)
1711 {
1712 hasher.update([1]);
1713 hasher.update(projection.resource_id.as_str().as_bytes());
1714 hasher.update(projection.physical_offset_bytes.to_be_bytes());
1715 hasher.update(projection.capacity_size_bytes.to_be_bytes());
1716 }
1717 format!("sha256/{:x}", hasher.finalize())
1718}
1719
1720impl<R> Drop for PreparedBackingClaim<R>
1721where
1722 R: DeviceRuntime,
1723{
1724 fn drop(&mut self) {
1725 if self.committed {
1726 return;
1727 }
1728 for extent in self.extents.iter().rev() {
1729 extent.pool.rollback_prepared(
1730 &extent.claim_identity,
1731 extent.occupancy,
1732 &extent.segments,
1733 );
1734 }
1735 }
1736}
1737
1738pub(super) enum BackingPrepareDecision<R>
1739where
1740 R: DeviceRuntime,
1741{
1742 Prepared(PreparedBackingClaim<R>),
1743 Deferred(DynamicBackingDeferred),
1744}
1745
1746#[derive(Debug)]
1753pub(super) struct BackingClaimCertificate {
1754 allocations: Box<[Arc<LogicalBackingSliceAllocationEvidence>]>,
1755 physical_capacity: CapacityVector,
1756 reusable_execution_bucket_id: Option<ReusableExecutionBucketId>,
1757 physical_claim_count: usize,
1758 has_shared_physical_claims: bool,
1759 fingerprint: String,
1760}
1761
1762#[derive(Debug)]
1763pub(super) struct BoundBackingClaimCertificate {
1764 fingerprint: String,
1765 physical_claim_count: usize,
1766 has_shared_physical_claims: bool,
1767}
1768
1769impl BoundBackingClaimCertificate {
1770 pub(super) fn fingerprint(&self) -> &str {
1771 &self.fingerprint
1772 }
1773
1774 pub(super) const fn physical_claim_count(&self) -> usize {
1775 self.physical_claim_count
1776 }
1777
1778 pub(super) const fn has_shared_physical_claims(&self) -> bool {
1779 self.has_shared_physical_claims
1780 }
1781}
1782
1783impl BackingClaimCertificate {
1784 pub(super) fn from_slices(
1785 backing_slices: &[LogicalBackingSliceAuthority],
1786 ) -> Result<Self, VNextError> {
1787 if backing_slices
1788 .windows(2)
1789 .any(|pair| pair[0].resource_id() >= pair[1].resource_id())
1790 {
1791 return Err(invalid_resource(
1792 "backing claim certificate requires canonical unique logical projections",
1793 ));
1794 }
1795 let reusable_execution_bucket_id = backing_slices
1796 .first()
1797 .and_then(|slice| slice.evidence().reusable_execution_bucket_id())
1798 .cloned();
1799 if backing_slices.iter().any(|slice| {
1800 slice.evidence().reusable_execution_bucket_id() != reusable_execution_bucket_id.as_ref()
1801 }) {
1802 return Err(invalid_resource(
1803 "one backing certificate cannot mix reusable execution buckets",
1804 ));
1805 }
1806
1807 let mut backing_by_domain = BTreeMap::<CapacityDomainId, u64>::new();
1808 let mut physical_claims = BTreeMap::<
1809 PhysicalBackingClaimIdentity,
1810 (Arc<BackingSegmentLease>, CapacityDomainId, u64),
1811 >::new();
1812 let mut has_shared_physical_claims = false;
1813 for slice in backing_slices {
1814 let evidence = slice.evidence();
1815 let claim_identity = evidence.physical_claim_identity();
1816 if claim_identity.pool_id() != evidence.pool_id()
1817 || claim_identity
1818 .resource_ids()
1819 .binary_search(evidence.resource_id())
1820 .is_err()
1821 || slice.segment_lease.claim_identity != *claim_identity
1822 || slice.segment_lease.segment_generation != evidence.segment_generation()
1823 || slice.segment_lease.size_bytes != evidence.physical_size_bytes()
1824 || evidence.size_bytes() == 0
1825 || evidence.size_bytes() > evidence.capacity_size_bytes()
1826 || evidence
1827 .physical_offset_bytes()
1828 .checked_add(evidence.capacity_size_bytes())
1829 .is_none_or(|end| end > evidence.physical_size_bytes())
1830 {
1831 return Err(invalid_resource(
1832 "logical backing projection differs from its physical claim authority",
1833 ));
1834 }
1835 has_shared_physical_claims |= claim_identity.is_shared();
1836 match physical_claims.entry(claim_identity.clone()) {
1837 std::collections::btree_map::Entry::Vacant(entry) => {
1838 let total = backing_by_domain.entry(slice.domain_id()).or_default();
1839 *total = total
1840 .checked_add(evidence.physical_size_bytes())
1841 .ok_or_else(|| {
1842 invalid_resource("certified backing domain bytes overflow u64")
1843 })?;
1844 entry.insert((
1845 Arc::clone(&slice.segment_lease),
1846 slice.domain_id(),
1847 evidence.physical_size_bytes(),
1848 ));
1849 }
1850 std::collections::btree_map::Entry::Occupied(entry) => {
1851 let (lease, domain_id, size_bytes) = entry.get();
1852 if !Arc::ptr_eq(lease, &slice.segment_lease)
1853 || *domain_id != slice.domain_id()
1854 || *size_bytes != evidence.physical_size_bytes()
1855 {
1856 return Err(invalid_resource(
1857 "shared logical projections do not retain one physical claim",
1858 ));
1859 }
1860 }
1861 }
1862 }
1863 let physical_capacity = if backing_by_domain.is_empty() {
1864 CapacityVector::empty()
1865 } else {
1866 CapacityVector::new(
1867 backing_by_domain
1868 .into_iter()
1869 .map(|(domain, bytes)| CapacityEntry::new(domain, CapacityUnits::new(bytes)))
1870 .collect::<Result<Vec<_>, _>>()?,
1871 )?
1872 };
1873 let allocations = backing_slices
1874 .iter()
1875 .map(|slice| Arc::clone(&slice.evidence.allocation))
1876 .collect::<Vec<_>>()
1877 .into_boxed_slice();
1878 let mut hasher = Sha256::new();
1879 hasher.update(b"ferrum.runtime-vnext.backing-claim-certificate.v1\0");
1880 for allocation in &allocations {
1881 let fingerprint = allocation.fingerprint.as_bytes();
1882 hasher.update(
1883 u64::try_from(fingerprint.len())
1884 .map_err(|_| {
1885 invalid_resource(
1886 "backing allocation fingerprint length exceeds portable range",
1887 )
1888 })?
1889 .to_be_bytes(),
1890 );
1891 hasher.update(fingerprint);
1892 }
1893 Ok(Self {
1894 allocations,
1895 physical_capacity,
1896 reusable_execution_bucket_id,
1897 physical_claim_count: physical_claims.len(),
1898 has_shared_physical_claims,
1899 fingerprint: format!("{:x}", hasher.finalize()),
1900 })
1901 }
1902
1903 pub(super) fn bind(
1904 &self,
1905 backing_slices: &[LogicalBackingSliceAuthority],
1906 demand: &super::AdmissionDemand,
1907 ) -> Result<BoundBackingClaimCertificate, VNextError> {
1908 if backing_slices.len() != self.allocations.len() {
1909 return Err(invalid_resource(
1910 "bound backing projection count differs from its physical certificate",
1911 ));
1912 }
1913 let mut hasher = Sha256::new();
1914 hasher.update(b"ferrum.runtime-vnext.bound-backing-claim.v1\0");
1915 hasher.update(self.fingerprint.as_bytes());
1916 for (slice, allocation) in backing_slices.iter().zip(&self.allocations) {
1917 if !Arc::ptr_eq(&slice.evidence.allocation, allocation)
1918 || slice.evidence.size_bytes() == 0
1919 || slice.evidence.size_bytes() > allocation.capacity_size_bytes
1920 {
1921 return Err(invalid_resource(
1922 "bound logical projection differs from its certified allocation",
1923 ));
1924 }
1925 hasher.update(slice.evidence.size_bytes().to_be_bytes());
1926 }
1927 let physical_covers_logical = self.physical_capacity.entries().len()
1928 == demand.immediate_claim().entries().len()
1929 && self.physical_capacity.entries().iter().all(|physical| {
1930 demand
1931 .immediate_claim()
1932 .units_for(physical.domain())
1933 .is_some_and(|logical| physical.units().get() >= logical.get())
1934 });
1935 let claim_matches = if self.reusable_execution_bucket_id.is_some() {
1936 physical_covers_logical
1937 } else {
1938 self.physical_capacity == *demand.immediate_claim()
1939 };
1940 if !claim_matches {
1941 return Err(invalid_resource(
1942 "certified physical backing does not cover the evaluated logical demand",
1943 ));
1944 }
1945 Ok(BoundBackingClaimCertificate {
1946 fingerprint: format!("{:x}", hasher.finalize()),
1947 physical_claim_count: self.physical_claim_count,
1948 has_shared_physical_claims: self.has_shared_physical_claims,
1949 })
1950 }
1951}
1952
1953#[doc(hidden)]
1954#[derive(Debug, PartialEq, Eq, Serialize)]
1955pub struct LogicalBackingSliceAllocationEvidence {
1956 pub(in crate::vnext::resource) domain_id: CapacityDomainId,
1957 pub(in crate::vnext::resource) pool_id: DynamicBackingPoolId,
1958 pub(in crate::vnext::resource) resource_id: ResourceId,
1959 pub(in crate::vnext::resource) pool_instance_id: u64,
1960 pub(in crate::vnext::resource) physical_claim_identity: PhysicalBackingClaimIdentity,
1961 #[serde(skip_serializing_if = "Option::is_none")]
1962 pub(in crate::vnext::resource) reusable_execution_bucket_id: Option<ReusableExecutionBucketId>,
1963 pub(in crate::vnext::resource) segment_generation: u64,
1964 pub(in crate::vnext::resource) segments: Vec<BackingSegment>,
1965 pub(in crate::vnext::resource) physical_offset_bytes: u64,
1966 pub(in crate::vnext::resource) capacity_size_bytes: u64,
1967 pub(in crate::vnext::resource) physical_size_bytes: u64,
1968 pub(in crate::vnext::resource) alignment_bytes: u64,
1969 pub(in crate::vnext::resource) usage: BufferUsage,
1970 pub(in crate::vnext::resource) element_type: ElementType,
1971 pub(in crate::vnext::resource) storage_profile: DynamicStorageProfile,
1972 pub(in crate::vnext::resource) initialization: StateInitialization,
1973 #[serde(skip)]
1974 pub(in crate::vnext::resource) fingerprint: String,
1975}
1976
1977#[derive(Debug, Clone, PartialEq, Eq)]
1978pub struct LogicalBackingSliceEvidence {
1979 pub(super) allocation: Arc<LogicalBackingSliceAllocationEvidence>,
1980 pub(in crate::vnext::resource) logical_size_bytes: u64,
1981}
1982
1983impl std::ops::Deref for LogicalBackingSliceEvidence {
1984 type Target = LogicalBackingSliceAllocationEvidence;
1985
1986 fn deref(&self) -> &Self::Target {
1987 self.allocation.as_ref()
1988 }
1989}
1990
1991impl Serialize for LogicalBackingSliceEvidence {
1992 fn serialize<S>(&self, serializer: S) -> Result<S::Ok, S::Error>
1993 where
1994 S: serde::Serializer,
1995 {
1996 #[derive(Serialize)]
1997 struct Wire<'a> {
1998 domain_id: CapacityDomainId,
1999 pool_id: &'a DynamicBackingPoolId,
2000 resource_id: &'a ResourceId,
2001 pool_instance_id: u64,
2002 physical_claim_identity: &'a PhysicalBackingClaimIdentity,
2003 #[serde(skip_serializing_if = "Option::is_none")]
2004 reusable_execution_bucket_id: Option<&'a ReusableExecutionBucketId>,
2005 segment_generation: u64,
2006 segments: &'a [BackingSegment],
2007 physical_offset_bytes: u64,
2008 #[serde(rename = "size_bytes")]
2009 logical_size_bytes: u64,
2010 capacity_size_bytes: u64,
2011 physical_size_bytes: u64,
2012 alignment_bytes: u64,
2013 usage: BufferUsage,
2014 element_type: ElementType,
2015 storage_profile: DynamicStorageProfile,
2016 initialization: StateInitialization,
2017 }
2018
2019 Wire {
2020 domain_id: self.domain_id,
2021 pool_id: &self.pool_id,
2022 resource_id: &self.resource_id,
2023 pool_instance_id: self.pool_instance_id,
2024 physical_claim_identity: &self.physical_claim_identity,
2025 reusable_execution_bucket_id: self.reusable_execution_bucket_id.as_ref(),
2026 segment_generation: self.segment_generation,
2027 segments: &self.segments,
2028 physical_offset_bytes: self.physical_offset_bytes,
2029 logical_size_bytes: self.logical_size_bytes,
2030 capacity_size_bytes: self.capacity_size_bytes,
2031 physical_size_bytes: self.physical_size_bytes,
2032 alignment_bytes: self.alignment_bytes,
2033 usage: self.usage,
2034 element_type: self.element_type,
2035 storage_profile: self.storage_profile,
2036 initialization: self.initialization,
2037 }
2038 .serialize(serializer)
2039 }
2040}
2041
2042impl LogicalBackingSliceEvidence {
2043 pub fn domain_id(&self) -> CapacityDomainId {
2044 self.domain_id
2045 }
2046
2047 pub fn resource_id(&self) -> &ResourceId {
2048 &self.resource_id
2049 }
2050
2051 pub fn pool_id(&self) -> &DynamicBackingPoolId {
2052 &self.pool_id
2053 }
2054
2055 pub fn pool_instance_id(&self) -> u64 {
2056 self.pool_instance_id
2057 }
2058
2059 pub fn segment_generation(&self) -> u64 {
2060 self.segment_generation
2061 }
2062
2063 pub fn physical_claim_identity(&self) -> &PhysicalBackingClaimIdentity {
2064 &self.physical_claim_identity
2065 }
2066
2067 pub fn reusable_execution_bucket_id(&self) -> Option<&ReusableExecutionBucketId> {
2068 self.reusable_execution_bucket_id.as_ref()
2069 }
2070
2071 pub fn segments(&self) -> &[BackingSegment] {
2072 &self.segments
2073 }
2074
2075 pub fn physical_offset_bytes(&self) -> u64 {
2076 self.physical_offset_bytes
2077 }
2078
2079 pub const fn size_bytes(&self) -> u64 {
2080 self.logical_size_bytes
2081 }
2082
2083 pub fn capacity_size_bytes(&self) -> u64 {
2084 self.capacity_size_bytes
2085 }
2086
2087 pub fn physical_size_bytes(&self) -> u64 {
2088 self.physical_size_bytes
2089 }
2090
2091 pub fn alignment_bytes(&self) -> u64 {
2092 self.alignment_bytes
2093 }
2094
2095 pub fn usage(&self) -> BufferUsage {
2096 self.usage
2097 }
2098
2099 pub fn element_type(&self) -> ElementType {
2100 self.element_type
2101 }
2102
2103 pub fn storage_profile(&self) -> DynamicStorageProfile {
2104 self.storage_profile
2105 }
2106
2107 pub fn initialization(&self) -> StateInitialization {
2108 self.initialization
2109 }
2110}
2111
2112#[must_use = "a logical backing authority owns its physical arena extents"]
2113pub struct LogicalBackingSliceAuthority {
2114 pub(in crate::vnext::resource) evidence: LogicalBackingSliceEvidence,
2115 pub(in crate::vnext::resource) segment_lease: Arc<BackingSegmentLease>,
2116 pub(super) reusable_lane: Option<ExecutionLaneId>,
2117}
2118
2119impl LogicalBackingSliceAuthority {
2120 pub fn evidence(&self) -> &LogicalBackingSliceEvidence {
2121 &self.evidence
2122 }
2123
2124 pub(in crate::vnext::resource) fn retained(&self) -> Self {
2125 Self {
2126 evidence: self.evidence.clone(),
2127 segment_lease: Arc::clone(&self.segment_lease),
2128 reusable_lane: self.reusable_lane,
2129 }
2130 }
2131
2132 pub(in crate::vnext::resource) fn retained_for_lane(&self, lane_id: ExecutionLaneId) -> Self {
2133 Self {
2134 evidence: self.evidence.clone(),
2135 segment_lease: Arc::clone(&self.segment_lease),
2136 reusable_lane: Some(lane_id),
2137 }
2138 }
2139
2140 pub(crate) const fn reusable_address_scope(&self) -> Option<DeviceReusableAddressScope> {
2141 match self.reusable_lane {
2142 Some(lane_id) => Some(DeviceReusableAddressScope::ExecutionLane(lane_id)),
2143 None => None,
2144 }
2145 }
2146
2147 pub fn domain_id(&self) -> CapacityDomainId {
2148 self.evidence.domain_id
2149 }
2150
2151 pub fn resource_id(&self) -> &ResourceId {
2152 &self.evidence.resource_id
2153 }
2154
2155 pub const fn size_bytes(&self) -> u64 {
2156 self.evidence.logical_size_bytes
2157 }
2158
2159 pub fn capacity_size_bytes(&self) -> u64 {
2160 self.evidence.capacity_size_bytes
2161 }
2162
2163 pub fn initialization_status(&self) -> Result<Option<BackingInitializationStatus>, VNextError> {
2164 self.segment_lease
2165 .initialization
2166 .as_ref()
2167 .map(|cell| cell.status())
2168 .transpose()
2169 }
2170
2171 pub(in crate::vnext::resource) fn initialization_cell(
2172 &self,
2173 ) -> Option<&Arc<BackingInitializationCell>> {
2174 self.segment_lease.initialization.as_ref()
2175 }
2176}
2177
2178pub struct LogicalBackingBufferView<'a, B> {
2179 pub(in crate::vnext::resource) bindings: Vec<LogicalBackingSegmentBinding<B>>,
2180 pub(super) authorities: &'a [LogicalBackingSliceAuthority],
2181 pub(super) logical_size_bytes: u64,
2182 pub(super) capacity_size_bytes: u64,
2183 pub(super) alignment_bytes: u64,
2184 pub(super) usage: BufferUsage,
2185 pub(super) element_type: ElementType,
2186 pub(super) storage_profile: DynamicStorageProfile,
2187}
2188
2189pub(crate) struct LogicalBackingSegmentBinding<B> {
2190 pub(in crate::vnext::resource) segment: BackingSegment,
2191 pub(in crate::vnext::resource) chunk: Arc<ResidentChunkBacking<B>>,
2192 pub(in crate::vnext::resource) retention: DeviceBufferRetention,
2193}
2194
2195impl<B> LogicalBackingSegmentBinding<B> {
2196 pub(crate) fn segment(&self) -> &BackingSegment {
2197 &self.segment
2198 }
2199
2200 pub(crate) fn chunk(&self) -> &BackingChunkIdentity {
2201 self.segment.chunk()
2202 }
2203
2204 pub(crate) fn buffer(&self) -> &B {
2205 &self.chunk.buffer
2206 }
2207
2208 pub(crate) fn descriptor(&self) -> &BufferDescriptor {
2209 &self.chunk.descriptor
2210 }
2211
2212 pub(crate) fn retention(&self) -> DeviceBufferRetention {
2213 self.retention.clone()
2214 }
2215}
2216
2217impl<'a, B> LogicalBackingBufferView<'a, B> {
2218 pub(crate) fn segment_bindings(&self) -> &[LogicalBackingSegmentBinding<B>] {
2219 &self.bindings
2220 }
2221
2222 pub const fn size_bytes(&self) -> u64 {
2223 self.logical_size_bytes
2224 }
2225
2226 pub const fn capacity_size_bytes(&self) -> u64 {
2227 self.capacity_size_bytes
2228 }
2229
2230 pub const fn alignment_bytes(&self) -> u64 {
2231 self.alignment_bytes
2232 }
2233
2234 pub const fn usage(&self) -> BufferUsage {
2235 self.usage
2236 }
2237
2238 pub const fn element_type(&self) -> ElementType {
2239 self.element_type
2240 }
2241
2242 pub const fn storage_profile(&self) -> DynamicStorageProfile {
2243 self.storage_profile
2244 }
2245
2246 pub fn committed_evidence_segments(&self) -> impl Iterator<Item = &BackingSegment> {
2247 self.authorities
2248 .iter()
2249 .flat_map(|authority| authority.evidence.segments())
2250 }
2251
2252 pub fn slice(&self) -> &'a LogicalBackingSliceEvidence {
2255 &self
2256 .authorities
2257 .first()
2258 .expect("logical backing views always contain an authority")
2259 .evidence
2260 }
2261}
2262
2263impl<R> DynamicBackingPool<R>
2264where
2265 R: DeviceRuntime,
2266{
2267 pub(super) fn allocation_quantum(&self) -> u64 {
2268 match self.domain.pool.compatibility().profile().allocator() {
2269 DynamicStorageAllocator::LinearArena => {
2270 self.domain.pool.compatibility().alignment_bytes()
2271 }
2272 DynamicStorageAllocator::FixedBlockArena { block_bytes } => {
2273 block_bytes.max(self.domain.pool.compatibility().alignment_bytes())
2274 }
2275 }
2276 }
2277
2278 fn cancel_pending_growth(&self, bytes: u64) {
2279 let mut state = match self.state.lock() {
2280 Ok(state) => state,
2281 Err(poisoned) => poisoned.into_inner(),
2282 };
2283 if state.pending_growth_bytes < bytes {
2284 state.poisoned = true;
2285 return;
2286 }
2287 state.pending_growth_bytes -= bytes;
2288 }
2289
2290 fn rollback_prepared(
2291 &self,
2292 claim_identity: &PhysicalBackingClaimIdentity,
2293 occupancy: DynamicBackingClaimOccupancy,
2294 segments: &[BackingSegment],
2295 ) {
2296 let mut state = match self.state.lock() {
2297 Ok(state) => state,
2298 Err(poisoned) => {
2299 let mut state = poisoned.into_inner();
2300 state.poisoned = true;
2301 return;
2302 }
2303 };
2304 if state.poisoned {
2305 return;
2306 }
2307 let segment_count = u64::try_from(segments.len()).ok();
2308 let physical_bytes = segments.iter().try_fold(0_u64, |total, segment| {
2309 total.checked_add(segment.length_bytes())
2310 });
2311 if claim_identity.pool_id() != self.domain.pool_id()
2312 || Some(occupancy.segment_count) != segment_count
2313 || Some(occupancy.physical_bytes) != physical_bytes
2314 {
2315 state.poisoned = true;
2316 return;
2317 }
2318 let next_occupancy = match state.live_occupancy.checked_without_claim(occupancy) {
2319 Ok(next) => next,
2320 Err(_) => {
2321 state.poisoned = true;
2322 return;
2323 }
2324 };
2325 for segment in segments.iter().rev() {
2326 let valid = state
2327 .chunks
2328 .get(&segment.chunk_ordinal())
2329 .is_some_and(|chunk| {
2330 chunk.backing.identity == *segment.chunk() && chunk.live_segments != 0
2331 });
2332 if !valid || state.allocator.release(segment).is_err() {
2333 state.poisoned = true;
2334 return;
2335 }
2336 state
2337 .chunks
2338 .get_mut(&segment.chunk_ordinal())
2339 .expect("validated prepared chunk remains installed")
2340 .live_segments -= 1;
2341 }
2342 state.live_occupancy = next_occupancy;
2343 drop(state);
2344 if self
2345 .logical_admission
2346 .notify_domain_availability_changed(self.domain.domain_id)
2347 .is_err()
2348 {
2349 let mut state = self
2350 .state
2351 .lock()
2352 .unwrap_or_else(std::sync::PoisonError::into_inner);
2353 state.poisoned = true;
2354 }
2355 }
2356}