1use std::ops::Range;
2
3use super::super::{
4 AllocationLifetime, BatchWorkShape, BufferDescriptor, BufferUsage, DeviceBufferRetention,
5 DeviceRuntime, DynamicResourceDemand, DynamicResourceShape, LeasedBufferView,
6 LogicalBackingBufferView, LogicalBackingSegmentBinding, NodeWorkContract, ResourceId,
7 ResourceTransactionIdentity, VNextError,
8};
9use super::foundation::invalid_operation;
10use super::{DynamicStorageView, ResolvedStorageComponent, ResolvedValueBinding};
11
12#[derive(Debug, Clone, Copy, PartialEq, Eq)]
13pub(super) enum ValueBindingPhysicalCoverage {
14 CanonicalComponent,
15 RuntimeTokenView,
16}
17
18pub(super) fn validate_value_binding_physical_coverage(
19 work: &NodeWorkContract,
20 binding: &ResolvedValueBinding,
21 component: &ResolvedStorageComponent,
22 descriptor: &BufferDescriptor,
23 dynamic_demand: Option<&DynamicResourceDemand>,
24 value_alignment_bytes: u64,
25) -> Result<ValueBindingPhysicalCoverage, VNextError> {
26 let required_end = component
27 .offset_bytes()
28 .checked_add(component.length_bytes())
29 .ok_or_else(|| invalid_operation("bound component range overflows u64"))?;
30 if descriptor.usage != binding.usage()
31 || descriptor.element_type != component.element_type()
32 || descriptor.alignment_bytes < value_alignment_bytes
33 || descriptor.alignment_bytes % value_alignment_bytes != 0
34 || component.offset_bytes() % value_alignment_bytes != 0
35 {
36 return Err(invalid_operation(format!(
37 "resource `{}` differs from its value binding",
38 component.resource_id()
39 )));
40 }
41
42 let Some(projection) = work.token_projection(binding.role(), binding.ordinal()) else {
43 if required_end > descriptor.size_bytes {
44 return Err(invalid_operation(format!(
45 "resource `{}` differs from its value binding",
46 component.resource_id()
47 )));
48 }
49 return Ok(ValueBindingPhysicalCoverage::CanonicalComponent);
50 };
51
52 let axis = usize::try_from(projection.axis())
53 .map_err(|_| invalid_operation("token projection axis exceeds usize"))?;
54 let canonical_extent = projection.canonical_extent();
55 let bytes_per_token = component
56 .length_bytes()
57 .checked_div(canonical_extent)
58 .filter(|bytes| *bytes > 0)
59 .ok_or_else(|| invalid_operation("token projection has zero canonical extent"))?;
60 let demand_matches = matches!(
61 dynamic_demand,
62 Some(DynamicResourceDemand::Tokens {
63 bytes_per_token: planned_bytes_per_token,
64 ..
65 }) if *planned_bytes_per_token == bytes_per_token
66 );
67 if binding.usage() != BufferUsage::Activations
68 || component.offset_bytes() != 0
69 || component.length_bytes() % canonical_extent != 0
70 || usize::try_from(projection.rank()).ok() != Some(binding.tensor().dimensions().len())
71 || binding.tensor().dimensions().get(axis) != Some(&canonical_extent)
72 || !demand_matches
73 || descriptor.size_bytes < bytes_per_token
74 {
75 return Err(invalid_operation(format!(
76 "resource `{}` differs from its token-projected value binding",
77 component.resource_id()
78 )));
79 }
80
81 Ok(ValueBindingPhysicalCoverage::RuntimeTokenView)
82}
83
84#[derive(Debug, Clone, Copy, PartialEq, Eq)]
85enum StepParticipantRangeCoordinates {
86 SourceToken,
87 ParticipantLocal,
88}
89
90pub(crate) fn translate_step_participant_upload_range(
93 demand: &DynamicResourceDemand,
94 work_shape: &BatchWorkShape,
95 participant_index: usize,
96 semantic_range: Range<u64>,
97) -> Result<Range<u64>, VNextError> {
98 translate_step_participant_range(
99 demand,
100 work_shape,
101 participant_index,
102 semantic_range,
103 StepParticipantRangeCoordinates::SourceToken,
104 )
105}
106
107pub(crate) fn translate_step_participant_readback_range(
109 demand: &DynamicResourceDemand,
110 work_shape: &BatchWorkShape,
111 participant_index: usize,
112 semantic_range: Range<u64>,
113) -> Result<Range<u64>, VNextError> {
114 translate_step_participant_range(
115 demand,
116 work_shape,
117 participant_index,
118 semantic_range,
119 StepParticipantRangeCoordinates::ParticipantLocal,
120 )
121}
122
123pub(crate) fn packed_step_token_range_to_participant_local_readback(
127 demand: &DynamicResourceDemand,
128 work_shape: &BatchWorkShape,
129 participant_index: usize,
130 packed_range: Range<u64>,
131) -> Result<Range<u64>, VNextError> {
132 if packed_range.start >= packed_range.end {
133 return Err(invalid_operation(
134 "packed participant token readback range is empty",
135 ));
136 }
137 let token_range = work_shape
138 .participant_token_ranges()
139 .get(participant_index)
140 .ok_or_else(|| invalid_operation("packed participant token readback is out of range"))?;
141 let DynamicResourceDemand::Tokens {
142 bytes_per_token,
143 maximum_tokens,
144 } = demand
145 else {
146 return Err(invalid_operation(
147 "packed participant token readback requires token demand",
148 ));
149 };
150 if work_shape.immediate_tokens() > *maximum_tokens {
151 return Err(invalid_operation(
152 "packed participant token readback exceeds its planned ceiling",
153 ));
154 }
155 let participant_start = token_range
156 .immediate_token_range()
157 .start
158 .checked_mul(*bytes_per_token)
159 .ok_or_else(|| invalid_operation("packed participant token offset overflows u64"))?;
160 let participant_end = token_range
161 .immediate_token_range()
162 .end
163 .checked_mul(*bytes_per_token)
164 .ok_or_else(|| invalid_operation("packed participant token range overflows u64"))?;
165 if packed_range.start < participant_start || packed_range.end > participant_end {
166 return Err(invalid_operation(
167 "packed participant token readback is outside its immediate span",
168 ));
169 }
170 Ok(packed_range.start - participant_start..packed_range.end - participant_start)
171}
172
173fn translate_step_participant_range(
174 demand: &DynamicResourceDemand,
175 work_shape: &BatchWorkShape,
176 participant_index: usize,
177 semantic_range: Range<u64>,
178 coordinates: StepParticipantRangeCoordinates,
179) -> Result<Range<u64>, VNextError> {
180 if semantic_range.start >= semantic_range.end {
181 return Err(invalid_operation(
182 "participant resource projection has an empty semantic range",
183 ));
184 }
185 if participant_index >= work_shape.participant_token_ranges().len() {
186 return Err(invalid_operation(
187 "participant resource projection is out of range",
188 ));
189 }
190 match demand {
191 DynamicResourceDemand::ActualSequences {
192 bytes_per_sequence,
193 maximum_sequences,
194 } => {
195 if work_shape.immediate_sequences() > *maximum_sequences
196 || semantic_range.end > *bytes_per_sequence
197 {
198 return Err(invalid_operation(
199 "participant fixed resource projection exceeds its planned stride",
200 ));
201 }
202 let base = bytes_per_sequence
203 .checked_mul(u64::try_from(participant_index).map_err(|_| {
204 invalid_operation("participant resource projection exceeds u64")
205 })?)
206 .ok_or_else(|| {
207 invalid_operation("participant fixed resource projection overflows u64")
208 })?;
209 let translated_start = base
210 .checked_add(semantic_range.start)
211 .ok_or_else(|| invalid_operation("participant fixed range overflows u64"))?;
212 let translated_end = base
213 .checked_add(semantic_range.end)
214 .ok_or_else(|| invalid_operation("participant fixed range overflows u64"))?;
215 Ok(translated_start..translated_end)
216 }
217 DynamicResourceDemand::Tokens {
218 bytes_per_token,
219 maximum_tokens,
220 } => {
221 if work_shape.immediate_tokens() > *maximum_tokens {
222 return Err(invalid_operation(
223 "participant token resource projection exceeds its planned ceiling",
224 ));
225 }
226 let token_range = &work_shape.participant_token_ranges()[participant_index];
227 let source = token_range.source_token_range();
228 let packed = token_range.immediate_token_range();
229 let source_start = source
230 .start
231 .checked_mul(*bytes_per_token)
232 .ok_or_else(|| invalid_operation("source token byte offset overflows u64"))?;
233 let source_end = source
234 .end
235 .checked_mul(*bytes_per_token)
236 .ok_or_else(|| invalid_operation("source token byte range overflows u64"))?;
237 let packed_start = packed
238 .start
239 .checked_mul(*bytes_per_token)
240 .ok_or_else(|| invalid_operation("packed token byte offset overflows u64"))?;
241 let relative_range = match coordinates {
242 StepParticipantRangeCoordinates::SourceToken => {
243 if semantic_range.start < source_start || semantic_range.end > source_end {
244 return Err(invalid_operation(
245 "participant token resource projection is outside its source span",
246 ));
247 }
248 semantic_range.start - source_start..semantic_range.end - source_start
249 }
250 StepParticipantRangeCoordinates::ParticipantLocal => {
251 let span_bytes = source_end - source_start;
252 if semantic_range.end > span_bytes {
253 return Err(invalid_operation(
254 "participant token readback exceeds its immediate span",
255 ));
256 }
257 semantic_range
258 }
259 };
260 let translated_start = packed_start
261 .checked_add(relative_range.start)
262 .ok_or_else(|| invalid_operation("packed token projection overflows u64"))?;
263 let translated_end = translated_start
264 .checked_add(relative_range.end - relative_range.start)
265 .ok_or_else(|| invalid_operation("packed token range overflows u64"))?;
266 Ok(translated_start..translated_end)
267 }
268 _ => Ok(semantic_range),
272 }
273}
274
275#[derive(Debug, Clone, Copy, PartialEq, Eq)]
276pub enum OperationBufferStorageKind {
277 StaticContiguous,
278 DynamicContiguous,
279 DynamicPaged,
280}
281
282enum OperationBufferSource<'a, B> {
283 Static {
284 view: LeasedBufferView<'a, B>,
285 retention: DeviceBufferRetention,
286 },
287 Backing(LogicalBackingBufferView<'a, B>),
288}
289
290#[derive(Debug, Clone, Copy, PartialEq, Eq)]
291enum OperationBufferCoverage {
292 Exact,
293 BackingWindow {
296 offset_bytes: u64,
297 },
298}
299
300enum OperationRegionSource<'a, B> {
301 Contiguous {
302 buffer: &'a B,
303 physical_base_offset_bytes: u64,
304 retention: DeviceBufferRetention,
305 },
306 Paged {
307 bindings: &'a [LogicalBackingSegmentBinding<B>],
308 logical_origin_bytes: u64,
309 },
310}
311
312mod copy;
313pub(crate) use copy::OperationBufferCopy;
314
315pub struct OperationBufferRegions<'a, B> {
318 storage_kind: OperationBufferStorageKind,
319 logical_offset_bytes: u64,
320 logical_length_bytes: u64,
321 source: OperationRegionSource<'a, B>,
322}
323
324impl<'a, B> OperationBufferRegions<'a, B> {
325 pub const fn storage_kind(&self) -> OperationBufferStorageKind {
326 self.storage_kind
327 }
328
329 pub const fn logical_offset_bytes(&self) -> u64 {
330 self.logical_offset_bytes
331 }
332
333 pub const fn logical_length_bytes(&self) -> u64 {
334 self.logical_length_bytes
335 }
336
337 pub fn iter(&self) -> OperationBufferRegionIter<'a, B> {
338 let logical_end_bytes = self
339 .logical_offset_bytes
340 .checked_add(self.logical_length_bytes)
341 .expect("validated operation logical range does not overflow");
342 match &self.source {
343 OperationRegionSource::Contiguous {
344 buffer,
345 physical_base_offset_bytes,
346 retention,
347 } => OperationBufferRegionIter {
348 state: OperationBufferRegionIterState::Contiguous(Some(OperationPhysicalRegion {
349 buffer: *buffer,
350 logical_offset_bytes: self.logical_offset_bytes,
351 physical_offset_bytes: physical_base_offset_bytes
352 .checked_add(self.logical_offset_bytes)
353 .expect("validated contiguous physical range does not overflow"),
354 length_bytes: self.logical_length_bytes,
355 retention: retention.clone(),
356 })),
357 },
358 OperationRegionSource::Paged {
359 bindings,
360 logical_origin_bytes,
361 } => OperationBufferRegionIter {
362 state: OperationBufferRegionIterState::Paged {
363 bindings: *bindings,
364 logical_origin_bytes: *logical_origin_bytes,
365 requested_start_bytes: logical_origin_bytes
366 .checked_add(self.logical_offset_bytes)
367 .expect("validated paged window start does not overflow"),
368 requested_end_bytes: logical_origin_bytes
369 .checked_add(logical_end_bytes)
370 .expect("validated paged window end does not overflow"),
371 next_segment: 0,
372 next_segment_logical_offset_bytes: 0,
373 },
374 },
375 }
376 }
377}
378
379pub struct OperationPhysicalRegion<'a, B> {
382 buffer: &'a B,
383 logical_offset_bytes: u64,
384 physical_offset_bytes: u64,
385 length_bytes: u64,
386 retention: DeviceBufferRetention,
387}
388
389impl<'a, B> OperationPhysicalRegion<'a, B> {
390 pub const fn logical_offset_bytes(&self) -> u64 {
391 self.logical_offset_bytes
392 }
393
394 pub const fn length_bytes(&self) -> u64 {
395 self.length_bytes
396 }
397
398 pub fn buffer_and_physical_range(
399 &self,
400 ) -> (&'a B, std::ops::Range<u64>, DeviceBufferRetention) {
401 (
402 self.buffer,
403 self.physical_offset_bytes
404 ..self
405 .physical_offset_bytes
406 .checked_add(self.length_bytes)
407 .expect("validated physical region does not overflow"),
408 self.retention.clone(),
409 )
410 }
411}
412
413pub struct OperationBufferRegionIter<'a, B> {
414 state: OperationBufferRegionIterState<'a, B>,
415}
416
417enum OperationBufferRegionIterState<'a, B> {
418 Contiguous(Option<OperationPhysicalRegion<'a, B>>),
419 Paged {
420 bindings: &'a [LogicalBackingSegmentBinding<B>],
421 logical_origin_bytes: u64,
422 requested_start_bytes: u64,
423 requested_end_bytes: u64,
424 next_segment: usize,
425 next_segment_logical_offset_bytes: u64,
426 },
427}
428
429impl<'a, B> Iterator for OperationBufferRegionIter<'a, B> {
430 type Item = OperationPhysicalRegion<'a, B>;
431
432 fn next(&mut self) -> Option<Self::Item> {
433 match &mut self.state {
434 OperationBufferRegionIterState::Contiguous(region) => region.take(),
435 OperationBufferRegionIterState::Paged {
436 bindings,
437 logical_origin_bytes,
438 requested_start_bytes,
439 requested_end_bytes,
440 next_segment,
441 next_segment_logical_offset_bytes,
442 } => {
443 while let Some(binding) = bindings.get(*next_segment) {
444 *next_segment += 1;
445 let (segment_logical_end, region) = translate_paged_segment(
446 binding.buffer(),
447 binding.retention(),
448 binding.segment().offset_bytes(),
449 binding.segment().length_bytes(),
450 *next_segment_logical_offset_bytes,
451 *requested_start_bytes,
452 *requested_end_bytes,
453 );
454 *next_segment_logical_offset_bytes = segment_logical_end;
455 if let Some(mut region) = region {
456 region.logical_offset_bytes = region
457 .logical_offset_bytes
458 .checked_sub(*logical_origin_bytes)
459 .expect("paged region is inside its validated backing window");
460 return Some(region);
461 }
462 }
463 None
464 }
465 }
466 }
467}
468
469fn translate_paged_segment<'a, B>(
470 buffer: &'a B,
471 retention: DeviceBufferRetention,
472 physical_offset_bytes: u64,
473 length_bytes: u64,
474 logical_start_bytes: u64,
475 requested_start_bytes: u64,
476 requested_end_bytes: u64,
477) -> (u64, Option<OperationPhysicalRegion<'a, B>>) {
478 let logical_end_bytes = logical_start_bytes
479 .checked_add(length_bytes)
480 .expect("validated backing segments do not overflow");
481 let translated_start = logical_start_bytes.max(requested_start_bytes);
482 let translated_end = logical_end_bytes.min(requested_end_bytes);
483 let region = (translated_start < translated_end).then(|| OperationPhysicalRegion {
484 buffer,
485 logical_offset_bytes: translated_start,
486 physical_offset_bytes: physical_offset_bytes
487 .checked_add(translated_start - logical_start_bytes)
488 .expect("validated paged physical range does not overflow"),
489 length_bytes: translated_end - translated_start,
490 retention,
491 });
492 (logical_end_bytes, region)
493}
494
495const fn operation_storage_kind(view: DynamicStorageView) -> OperationBufferStorageKind {
496 match view {
497 DynamicStorageView::Contiguous => OperationBufferStorageKind::DynamicContiguous,
498 DynamicStorageView::PagedRegions { .. } => OperationBufferStorageKind::DynamicPaged,
499 }
500}
501
502fn validate_dynamic_binding_layout(
503 storage_kind: OperationBufferStorageKind,
504 logical_size_bytes: u64,
505 mut binding_lengths: impl ExactSizeIterator<Item = u64>,
506 coverage: OperationBufferCoverage,
507) -> Result<(), VNextError> {
508 let binding_count = binding_lengths.len();
509 if storage_kind == OperationBufferStorageKind::StaticContiguous {
510 return Err(invalid_operation(
511 "dynamic backing cannot claim static storage kind",
512 ));
513 }
514 if binding_count == 0 {
515 return Err(invalid_operation(
516 "dynamic backing has no physical segment binding",
517 ));
518 }
519 if storage_kind == OperationBufferStorageKind::DynamicContiguous && binding_count != 1 {
520 return Err(invalid_operation(
521 "contiguous dynamic storage requires one physical segment binding",
522 ));
523 }
524 let covered = binding_lengths.try_fold(0_u64, |total, length_bytes| {
525 total
526 .checked_add(length_bytes)
527 .ok_or_else(|| invalid_operation("backing segment coverage overflows u64"))
528 })?;
529 let window_offset = match coverage {
530 OperationBufferCoverage::Exact => 0,
531 OperationBufferCoverage::BackingWindow { offset_bytes } => offset_bytes,
532 };
533 let required_end = window_offset
534 .checked_add(logical_size_bytes)
535 .ok_or_else(|| invalid_operation("operation backing window overflows u64"))?;
536 if covered < required_end
537 || (coverage == OperationBufferCoverage::Exact && covered != logical_size_bytes)
538 {
539 return Err(invalid_operation(
540 "dynamic backing segments do not cover the operation's exact logical view",
541 ));
542 }
543 Ok(())
544}
545
546pub(super) fn sequence_execution_shape(
547 committed: DynamicResourceShape,
548 source_end_tokens: u64,
549) -> Result<DynamicResourceShape, VNextError> {
550 if committed.sequences() != 1
551 || source_end_tokens == 0
552 || source_end_tokens > committed.tokens()
553 {
554 return Err(invalid_operation(
555 "sequence operation frontier is empty or exceeds committed backing",
556 ));
557 }
558 Ok(DynamicResourceShape::from_validated(
559 1,
560 source_end_tokens,
561 committed.pages(),
562 ))
563}
564
565pub struct OperationBufferView<'a, B> {
566 descriptor: BufferDescriptor,
567 source: OperationBufferSource<'a, B>,
568 coverage: OperationBufferCoverage,
569 allocation_lifetime: AllocationLifetime,
570 packed_batch_coordinates: bool,
571}
572
573impl<'a, B> OperationBufferView<'a, B> {
574 pub(super) fn from_static(
575 view: LeasedBufferView<'a, B>,
576 retention: DeviceBufferRetention,
577 ) -> Self {
578 Self {
579 descriptor: view.committed_descriptor().clone(),
580 source: OperationBufferSource::Static { view, retention },
581 coverage: OperationBufferCoverage::Exact,
582 allocation_lifetime: AllocationLifetime::Plan,
583 packed_batch_coordinates: false,
584 }
585 }
586
587 pub(super) fn from_backing_exact(
588 descriptor: BufferDescriptor,
589 backing: LogicalBackingBufferView<'a, B>,
590 allocation_lifetime: AllocationLifetime,
591 ) -> Self {
592 Self::from_backing(
593 descriptor,
594 backing,
595 OperationBufferCoverage::Exact,
596 allocation_lifetime,
597 )
598 }
599
600 pub(crate) fn from_backing_prefix(
601 descriptor: BufferDescriptor,
602 backing: LogicalBackingBufferView<'a, B>,
603 allocation_lifetime: AllocationLifetime,
604 ) -> Self {
605 Self::from_backing(
606 descriptor,
607 backing,
608 OperationBufferCoverage::BackingWindow { offset_bytes: 0 },
609 allocation_lifetime,
610 )
611 }
612
613 pub(super) fn from_backing_window(
614 descriptor: BufferDescriptor,
615 backing: LogicalBackingBufferView<'a, B>,
616 offset_bytes: u64,
617 allocation_lifetime: AllocationLifetime,
618 ) -> Self {
619 Self::from_backing(
620 descriptor,
621 backing,
622 OperationBufferCoverage::BackingWindow { offset_bytes },
623 allocation_lifetime,
624 )
625 }
626
627 fn from_backing(
628 descriptor: BufferDescriptor,
629 backing: LogicalBackingBufferView<'a, B>,
630 coverage: OperationBufferCoverage,
631 allocation_lifetime: AllocationLifetime,
632 ) -> Self {
633 Self {
634 descriptor,
635 source: OperationBufferSource::Backing(backing),
636 coverage,
637 allocation_lifetime,
638 packed_batch_coordinates: false,
639 }
640 }
641
642 pub(super) fn with_packed_batch_coordinates(mut self, packed: bool) -> Self {
643 self.packed_batch_coordinates = packed;
644 self
645 }
646
647 pub(super) fn validate_runtime<R>(
648 &self,
649 runtime: &R,
650 expected_static_identity: Option<&ResourceTransactionIdentity>,
651 ) -> Result<(), VNextError>
652 where
653 R: DeviceRuntime<Buffer = B>,
654 {
655 match &self.source {
656 OperationBufferSource::Static { view, .. } => {
657 let actual = runtime.buffer_descriptor(view.buffer());
658 if Some(view.identity()) != expected_static_identity
659 || &actual != view.committed_descriptor()
660 || view.generation() == 0
661 {
662 return Err(invalid_operation(format!(
663 "runtime descriptor differs from committed static resource `{}`",
664 self.resource_id()
665 )));
666 }
667 }
668 OperationBufferSource::Backing(backing_view) => {
669 let bindings = backing_view.segment_bindings();
670 if bindings.is_empty()
671 || bindings.len() != backing_view.committed_evidence_segments().count()
672 || bindings
673 .iter()
674 .zip(backing_view.committed_evidence_segments())
675 .any(|(binding, evidence)| {
676 let actual = runtime.buffer_descriptor(binding.buffer());
677 binding.segment() != evidence
678 || binding.chunk() != evidence.chunk()
679 || &actual != binding.descriptor()
680 || binding
681 .segment()
682 .offset_bytes()
683 .checked_add(binding.segment().length_bytes())
684 .is_none_or(|end| end > binding.descriptor().size_bytes)
685 })
686 {
687 return Err(invalid_operation(format!(
688 "runtime descriptor differs from a committed backing chunk for `{}`",
689 self.resource_id()
690 )));
691 }
692 }
693 }
694 Ok(())
695 }
696
697 pub fn resource_id(&self) -> &ResourceId {
698 &self.descriptor.resource_id
699 }
700
701 pub fn descriptor(&self) -> &BufferDescriptor {
702 &self.descriptor
703 }
704
705 pub const fn allocation_lifetime(&self) -> AllocationLifetime {
706 self.allocation_lifetime
707 }
708
709 pub const fn uses_packed_batch_coordinates(&self) -> bool {
710 self.packed_batch_coordinates
711 }
712
713 pub fn storage_kind(&self) -> OperationBufferStorageKind {
714 match &self.source {
715 OperationBufferSource::Static { .. } => OperationBufferStorageKind::StaticContiguous,
716 OperationBufferSource::Backing(view) => {
717 operation_storage_kind(view.storage_profile().view())
718 }
719 }
720 }
721
722 pub fn translate(
723 &self,
724 logical_offset_bytes: u64,
725 logical_length_bytes: u64,
726 ) -> Result<OperationBufferRegions<'_, B>, VNextError> {
727 let logical_end_bytes = logical_offset_bytes
728 .checked_add(logical_length_bytes)
729 .ok_or_else(|| invalid_operation("operation logical buffer range overflows u64"))?;
730 if logical_length_bytes == 0 || logical_end_bytes > self.descriptor.size_bytes {
731 return Err(invalid_operation(
732 "operation logical buffer range is empty or outside its resource",
733 ));
734 }
735 match &self.source {
736 OperationBufferSource::Static { view, retention } => Ok(OperationBufferRegions {
737 storage_kind: OperationBufferStorageKind::StaticContiguous,
738 logical_offset_bytes,
739 logical_length_bytes,
740 source: OperationRegionSource::Contiguous {
741 buffer: view.buffer(),
742 physical_base_offset_bytes: 0,
743 retention: retention.clone(),
744 },
745 }),
746 OperationBufferSource::Backing(view) => {
747 let bindings = view.segment_bindings();
748 if bindings.len() != view.committed_evidence_segments().count()
749 || bindings.iter().zip(view.committed_evidence_segments()).any(
750 |(binding, segment)| {
751 binding.segment() != segment || binding.chunk() != segment.chunk()
752 },
753 )
754 {
755 return Err(invalid_operation(
756 "dynamic backing bindings differ from committed segment evidence",
757 ));
758 }
759 let storage_kind = operation_storage_kind(view.storage_profile().view());
760 let backing_window_offset = match self.coverage {
761 OperationBufferCoverage::Exact => 0,
762 OperationBufferCoverage::BackingWindow { offset_bytes } => offset_bytes,
763 };
764 validate_dynamic_binding_layout(
765 storage_kind,
766 self.descriptor.size_bytes,
767 bindings
768 .iter()
769 .map(|binding| binding.segment().length_bytes()),
770 self.coverage,
771 )?;
772 let source = match storage_kind {
773 OperationBufferStorageKind::DynamicContiguous => {
774 let binding = &bindings[0];
775 OperationRegionSource::Contiguous {
776 buffer: binding.buffer(),
777 physical_base_offset_bytes: binding
778 .segment()
779 .offset_bytes()
780 .checked_add(backing_window_offset)
781 .ok_or_else(|| {
782 invalid_operation(
783 "participant backing physical offset overflows u64",
784 )
785 })?,
786 retention: binding.retention(),
787 }
788 }
789 OperationBufferStorageKind::DynamicPaged => OperationRegionSource::Paged {
790 bindings,
791 logical_origin_bytes: backing_window_offset,
792 },
793 OperationBufferStorageKind::StaticContiguous => unreachable!(
794 "dynamic storage kind was validated before region construction"
795 ),
796 };
797 Ok(OperationBufferRegions {
798 storage_kind,
799 logical_offset_bytes,
800 logical_length_bytes,
801 source,
802 })
803 }
804 }
805 }
806}
807
808#[cfg(test)]
809mod operation_buffer_region_tests {
810 use super::{
811 packed_step_token_range_to_participant_local_readback, sequence_execution_shape,
812 translate_paged_segment, translate_step_participant_readback_range,
813 translate_step_participant_upload_range, validate_dynamic_binding_layout,
814 validate_value_binding_physical_coverage, OperationBufferCoverage, OperationBufferRegions,
815 OperationBufferStorageKind, OperationRegionSource, ValueBindingPhysicalCoverage,
816 };
817 use crate::vnext::{
818 AliasPolicy, BatchWorkShape, BufferDescriptor, BufferUsage, DeviceBufferRetention,
819 DynamicResourceDemand, DynamicResourceShape, ElementType, NodeWorkContract, ProgramValueId,
820 ResolvedTensorLayout, ResolvedTensorSpec, ResolvedValueBinding, ResolvedValueRole,
821 ResolvedValueStorage, ResourceId, TensorAccess, TokenSpanWork,
822 };
823 use std::sync::atomic::Ordering;
824 use std::sync::Arc;
825
826 #[test]
827 fn token_projection_validates_runtime_view_instead_of_canonical_extent() {
828 let resource_id = ResourceId::new("resource.activation.token-ids").unwrap();
829 let binding = ResolvedValueBinding::new(
830 ProgramValueId::new("value.input.token-ids").unwrap(),
831 ResolvedValueRole::Input,
832 0,
833 ResolvedTensorSpec::new(
834 vec![128],
835 ElementType::U32,
836 ResolvedTensorLayout::Contiguous,
837 )
838 .unwrap(),
839 TensorAccess::Read,
840 AliasPolicy::NoAlias,
841 BufferUsage::Activations,
842 None,
843 ResolvedValueStorage::single(resource_id.clone(), 0, 512, ElementType::U32).unwrap(),
844 )
845 .unwrap();
846 let work: NodeWorkContract = serde_json::from_value(serde_json::json!({
847 "tokens": {
848 "source": {
849 "value_id": "value.input.token-ids",
850 "role": "input",
851 "ordinal": 0,
852 "axis": 0,
853 "rank": 1,
854 "canonical_extent": 128
855 },
856 "projections": [{
857 "value_id": "value.input.token-ids",
858 "role": "input",
859 "ordinal": 0,
860 "axis": 0,
861 "rank": 1,
862 "canonical_extent": 128
863 }]
864 }
865 }))
866 .unwrap();
867 let descriptor = BufferDescriptor {
868 resource_id,
869 size_bytes: 160,
870 alignment_bytes: 16,
871 usage: BufferUsage::Activations,
872 element_type: ElementType::U32,
873 };
874 let component = &binding.storage().components()[0];
875 let demand = DynamicResourceDemand::tokens(4, 128).unwrap();
876
877 assert_eq!(
878 validate_value_binding_physical_coverage(
879 &work,
880 &binding,
881 component,
882 &descriptor,
883 Some(&demand),
884 16,
885 )
886 .unwrap(),
887 ValueBindingPhysicalCoverage::RuntimeTokenView
888 );
889 assert!(validate_value_binding_physical_coverage(
890 &NodeWorkContract::Fixed,
891 &binding,
892 component,
893 &descriptor,
894 Some(&demand),
895 16,
896 )
897 .is_err());
898 assert!(validate_value_binding_physical_coverage(
899 &work,
900 &binding,
901 component,
902 &descriptor,
903 Some(&DynamicResourceDemand::tokens(8, 128).unwrap()),
904 16,
905 )
906 .is_err());
907 }
908
909 #[test]
910 fn paged_translation_uses_each_chunks_exact_buffer() {
911 struct MockBinding<'a> {
912 buffer: &'a u8,
913 physical_offset_bytes: u64,
914 length_bytes: u64,
915 }
916
917 let first_buffer = 7_u8;
918 let second_buffer = 11_u8;
919 let bindings = [
920 MockBinding {
921 buffer: &first_buffer,
922 physical_offset_bytes: 64,
923 length_bytes: 8,
924 },
925 MockBinding {
926 buffer: &second_buffer,
927 physical_offset_bytes: 200,
928 length_bytes: 12,
929 },
930 ];
931 let mut next_logical_offset = 0;
932 let translated = bindings
933 .iter()
934 .filter_map(|binding| {
935 let (logical_end, region) = translate_paged_segment(
936 binding.buffer,
937 DeviceBufferRetention::plan(Arc::new(())),
938 binding.physical_offset_bytes,
939 binding.length_bytes,
940 next_logical_offset,
941 6,
942 16,
943 );
944 next_logical_offset = logical_end;
945 region
946 })
947 .collect::<Vec<_>>();
948
949 assert_eq!(translated.len(), 2);
950 let (first, first_physical, _first_retention) = translated[0].buffer_and_physical_range();
951 assert!(std::ptr::eq(first, &first_buffer));
952 assert_eq!(translated[0].logical_offset_bytes(), 6);
953 assert_eq!(first_physical, 70..72);
954 let (second, second_physical, _second_retention) =
955 translated[1].buffer_and_physical_range();
956 assert!(std::ptr::eq(second, &second_buffer));
957 assert_eq!(translated[1].logical_offset_bytes(), 8);
958 assert_eq!(second_physical, 200..208);
959 }
960
961 #[test]
962 fn contiguous_layout_rejects_cross_chunk_bindings() {
963 let first_buffer = 7_u8;
964 let second_buffer = 11_u8;
965 let bindings = [(&first_buffer, 8_u64), (&second_buffer, 12_u64)];
966
967 let error = validate_dynamic_binding_layout(
968 OperationBufferStorageKind::DynamicContiguous,
969 20,
970 bindings.iter().map(|(_, length_bytes)| *length_bytes),
971 OperationBufferCoverage::Exact,
972 )
973 .unwrap_err();
974
975 assert!(error
976 .to_string()
977 .contains("contiguous dynamic storage requires one physical segment binding"));
978 }
979
980 #[test]
981 fn operation_prefix_view_retains_wider_backing_coverage() {
982 validate_dynamic_binding_layout(
983 OperationBufferStorageKind::DynamicPaged,
984 64,
985 [64_u64, 64].into_iter(),
986 OperationBufferCoverage::BackingWindow { offset_bytes: 0 },
987 )
988 .unwrap();
989 validate_dynamic_binding_layout(
990 OperationBufferStorageKind::DynamicContiguous,
991 64,
992 [128_u64].into_iter(),
993 OperationBufferCoverage::BackingWindow { offset_bytes: 0 },
994 )
995 .unwrap();
996
997 assert!(validate_dynamic_binding_layout(
998 OperationBufferStorageKind::DynamicPaged,
999 64,
1000 [64_u64, 64].into_iter(),
1001 OperationBufferCoverage::Exact,
1002 )
1003 .is_err());
1004 assert!(validate_dynamic_binding_layout(
1005 OperationBufferStorageKind::DynamicPaged,
1006 128,
1007 [64_u64].into_iter(),
1008 OperationBufferCoverage::BackingWindow { offset_bytes: 0 },
1009 )
1010 .is_err());
1011 }
1012
1013 #[test]
1014 fn participant_step_projection_keeps_upload_and_readback_coordinate_spaces_distinct() {
1015 let first_tokens = vec![1_u32; 20];
1016 let second_tokens = vec![2_u32; 9];
1017 let work = BatchWorkShape::test_only(vec![
1018 TokenSpanWork::from_token_ids(&first_tokens, 17..18).unwrap(),
1019 TokenSpanWork::from_token_ids(&second_tokens, 7..9).unwrap(),
1020 ])
1021 .unwrap();
1022 let tokens = DynamicResourceDemand::tokens(4, 32).unwrap();
1023
1024 assert_eq!(
1025 translate_step_participant_upload_range(&tokens, &work, 0, 68..72).unwrap(),
1026 0..4
1027 );
1028 assert_eq!(
1029 translate_step_participant_upload_range(&tokens, &work, 1, 28..36).unwrap(),
1030 4..12
1031 );
1032 assert_eq!(
1033 translate_step_participant_readback_range(&tokens, &work, 1, 0..8).unwrap(),
1034 4..12
1035 );
1036 assert_eq!(
1037 packed_step_token_range_to_participant_local_readback(&tokens, &work, 1, 4..12)
1038 .unwrap(),
1039 0..8
1040 );
1041 assert!(translate_step_participant_readback_range(&tokens, &work, 1, 28..36).is_err());
1042 assert!(
1043 packed_step_token_range_to_participant_local_readback(&tokens, &work, 1, 28..36)
1044 .is_err()
1045 );
1046 }
1047
1048 #[test]
1049 fn fixed_participant_step_projection_uses_disjoint_aligned_strides() {
1050 let tokens = [1_u32];
1051 let work = BatchWorkShape::test_only(vec![
1052 TokenSpanWork::from_token_ids(&tokens, 0..1).unwrap(),
1053 TokenSpanWork::from_token_ids(&tokens, 0..1).unwrap(),
1054 ])
1055 .unwrap();
1056 let fixed = DynamicResourceDemand::actual_sequences(16, 4).unwrap();
1057
1058 assert_eq!(
1059 translate_step_participant_upload_range(&fixed, &work, 0, 0..4).unwrap(),
1060 0..4
1061 );
1062 assert_eq!(
1063 translate_step_participant_upload_range(&fixed, &work, 1, 0..4).unwrap(),
1064 16..20
1065 );
1066 assert_eq!(
1067 translate_step_participant_readback_range(&fixed, &work, 1, 0..4).unwrap(),
1068 16..20
1069 );
1070 assert!(translate_step_participant_upload_range(&fixed, &work, 1, 0..17).is_err());
1071 assert_eq!(
1072 translate_step_participant_upload_range(
1073 &DynamicResourceDemand::fixed(16).unwrap(),
1074 &work,
1075 1,
1076 0..4,
1077 )
1078 .unwrap(),
1079 0..4
1080 );
1081 }
1082
1083 #[test]
1084 fn sequence_execution_shape_uses_the_executed_source_frontier() {
1085 let committed = DynamicResourceShape::from_validated(1, 8, 3);
1086 let projected = sequence_execution_shape(committed, 4).unwrap();
1087
1088 assert_eq!(projected.sequences(), 1);
1089 assert_eq!(projected.tokens(), 4);
1090 assert_eq!(projected.pages(), 3);
1091 assert!(sequence_execution_shape(committed, 0).is_err());
1092 assert!(sequence_execution_shape(committed, 9).is_err());
1093 assert!(
1094 sequence_execution_shape(DynamicResourceShape::from_validated(2, 8, 3), 4).is_err()
1095 );
1096 }
1097
1098 #[test]
1099 fn contiguous_translation_applies_physical_base_offset() {
1100 let buffer = 9_u8;
1101 let regions = OperationBufferRegions {
1102 storage_kind: OperationBufferStorageKind::DynamicContiguous,
1103 logical_offset_bytes: 16,
1104 logical_length_bytes: 32,
1105 source: OperationRegionSource::Contiguous {
1106 buffer: &buffer,
1107 physical_base_offset_bytes: 4096,
1108 retention: DeviceBufferRetention::plan(Arc::new(())),
1109 },
1110 };
1111
1112 let translated = regions.iter().collect::<Vec<_>>();
1113 assert_eq!(translated.len(), 1);
1114 let (actual, physical, _retention) = translated[0].buffer_and_physical_range();
1115 assert_eq!(*actual, buffer);
1116 assert_eq!(translated[0].logical_offset_bytes(), 16);
1117 assert_eq!(physical, 4112..4144);
1118 }
1119
1120 #[test]
1121 fn physical_region_retains_opaque_owner_after_translation_source_drops() {
1122 struct DropOwner(Arc<std::sync::atomic::AtomicBool>);
1123
1124 impl Drop for DropOwner {
1125 fn drop(&mut self) {
1126 self.0.store(true, Ordering::Release);
1127 }
1128 }
1129
1130 let buffer = 9_u8;
1131 let dropped = Arc::new(std::sync::atomic::AtomicBool::new(false));
1132 let owner = Arc::new(DropOwner(Arc::clone(&dropped)));
1133 let regions = OperationBufferRegions {
1134 storage_kind: OperationBufferStorageKind::DynamicContiguous,
1135 logical_offset_bytes: 0,
1136 logical_length_bytes: 8,
1137 source: OperationRegionSource::Contiguous {
1138 buffer: &buffer,
1139 physical_base_offset_bytes: 64,
1140 retention: DeviceBufferRetention::plan(Arc::clone(&owner)),
1141 },
1142 };
1143 drop(owner);
1144
1145 let translated = regions.iter().collect::<Vec<_>>();
1146 drop(regions);
1147 assert!(!dropped.load(Ordering::Acquire));
1148 let (_, physical, retention) = translated[0].buffer_and_physical_range();
1149 assert_eq!(physical, 64..72);
1150 drop(translated);
1151 assert!(!dropped.load(Ordering::Acquire));
1152
1153 drop(retention);
1154 assert!(dropped.load(Ordering::Acquire));
1155 }
1156}