1use super::*;
2
3pub const MAX_VNEXT_DETERMINISM_PARTICIPANTS: usize = 32;
4
5#[derive(Debug, Clone, Copy, PartialEq, Eq)]
6pub enum VNextDeterminismPhase {
7 Prefill,
8 Decode,
9}
10
11impl VNextDeterminismPhase {
12 const fn wave_kind(self) -> VNextExecutionWaveKind {
13 match self {
14 Self::Prefill => VNextExecutionWaveKind::Prefill,
15 Self::Decode => VNextExecutionWaveKind::Decode,
16 }
17 }
18}
19
20#[derive(Debug, Clone, Copy, PartialEq, Eq)]
21pub enum VNextDeterminismInitialState {
22 Zero,
23 Nonzero,
24}
25
26#[derive(Debug, Clone, Copy, PartialEq, Eq)]
27pub enum VNextDeterminismWorkspacePoison {
28 Zero,
29 A5,
30}
31
32impl VNextDeterminismWorkspacePoison {
33 const fn fill_byte(self) -> u8 {
34 match self {
35 Self::Zero => 0,
36 Self::A5 => 0xa5,
37 }
38 }
39}
40
41#[derive(Debug, Clone, Copy, PartialEq, Eq)]
42pub enum VNextDeterminismExecutionMode {
43 Eager,
44 Replayed,
45}
46
47#[derive(Debug, Clone, PartialEq, Eq)]
48pub struct VNextDeterminismParticipantSpec {
49 token_ids: Vec<u32>,
50 immediate_range: Range<usize>,
51 maximum_sequence_tokens: usize,
52}
53
54impl VNextDeterminismParticipantSpec {
55 pub fn new(
56 token_ids: Vec<u32>,
57 immediate_range: Range<usize>,
58 maximum_sequence_tokens: usize,
59 ) -> Result<Self> {
60 if token_ids.is_empty()
61 || immediate_range.start >= immediate_range.end
62 || immediate_range.end != token_ids.len()
63 || maximum_sequence_tokens < token_ids.len()
64 {
65 return Err(FerrumError::request_validation(
66 "vNext determinism participant requires a non-empty terminal token span covered by its sequence ceiling",
67 ));
68 }
69 Ok(Self {
70 token_ids,
71 immediate_range,
72 maximum_sequence_tokens,
73 })
74 }
75
76 pub fn token_ids(&self) -> &[u32] {
77 &self.token_ids
78 }
79
80 pub fn immediate_range(&self) -> Range<usize> {
81 self.immediate_range.clone()
82 }
83
84 pub const fn maximum_sequence_tokens(&self) -> usize {
85 self.maximum_sequence_tokens
86 }
87
88 fn token_span(&self) -> Result<TokenSpanWork> {
89 TokenSpanWork::from_token_ids_with_fit(
90 &self.token_ids,
91 self.immediate_range.clone(),
92 self.maximum_sequence_tokens,
93 )
94 .map_err(|error| FerrumError::backend(error.to_string()))
95 }
96
97 fn full_extension(&self) -> Result<ResourceWorkShape> {
98 let span = TokenSpanWork::from_token_ids_with_fit(
99 &self.token_ids,
100 0..self.token_ids.len(),
101 self.maximum_sequence_tokens,
102 )
103 .map_err(|error| FerrumError::backend(error.to_string()))?;
104 ResourceWorkShape::single(span).map_err(|error| FerrumError::backend(error.to_string()))
105 }
106
107 fn immediate_token_ids(&self) -> &[u32] {
108 &self.token_ids[self.immediate_range.clone()]
109 }
110}
111
112#[derive(Debug, Clone, PartialEq, Eq)]
113pub struct VNextDeterminismExecutionSpec {
114 phase: VNextDeterminismPhase,
115 participants: Vec<VNextDeterminismParticipantSpec>,
116 initial_state: VNextDeterminismInitialState,
117 workspace_poison: VNextDeterminismWorkspacePoison,
118 mode: VNextDeterminismExecutionMode,
119}
120
121impl VNextDeterminismExecutionSpec {
122 pub fn new(
123 phase: VNextDeterminismPhase,
124 participants: Vec<VNextDeterminismParticipantSpec>,
125 initial_state: VNextDeterminismInitialState,
126 workspace_poison: VNextDeterminismWorkspacePoison,
127 mode: VNextDeterminismExecutionMode,
128 ) -> Result<Self> {
129 if participants.is_empty()
130 || participants.len() > MAX_VNEXT_DETERMINISM_PARTICIPANTS
131 || (phase == VNextDeterminismPhase::Decode
132 && participants
133 .iter()
134 .any(|participant| participant.immediate_range.len() != 1))
135 {
136 return Err(FerrumError::request_validation(format!(
137 "vNext determinism execution requires 1..={MAX_VNEXT_DETERMINISM_PARTICIPANTS} participants and one immediate token per decode participant"
138 )));
139 }
140 Ok(Self {
141 phase,
142 participants,
143 initial_state,
144 workspace_poison,
145 mode,
146 })
147 }
148
149 pub const fn phase(&self) -> VNextDeterminismPhase {
150 self.phase
151 }
152
153 pub fn participants(&self) -> &[VNextDeterminismParticipantSpec] {
154 &self.participants
155 }
156
157 pub const fn initial_state(&self) -> VNextDeterminismInitialState {
158 self.initial_state
159 }
160
161 pub const fn workspace_poison(&self) -> VNextDeterminismWorkspacePoison {
162 self.workspace_poison
163 }
164
165 pub const fn mode(&self) -> VNextDeterminismExecutionMode {
166 self.mode
167 }
168}
169
170struct PendingDeterminismAdmissions<'executor, R: DeviceRuntime> {
171 executor: &'executor VNextModelExecutor<R>,
172 request_ids: Vec<RequestId>,
173 armed: bool,
174}
175
176impl<'executor, R: DeviceRuntime> PendingDeterminismAdmissions<'executor, R> {
177 fn new(executor: &'executor VNextModelExecutor<R>) -> Self {
178 Self {
179 executor,
180 request_ids: Vec::new(),
181 armed: true,
182 }
183 }
184
185 fn track(&mut self, request_id: RequestId) {
186 self.request_ids.push(request_id);
187 }
188
189 fn disarm(&mut self) {
190 self.armed = false;
191 }
192}
193
194impl<R: DeviceRuntime> Drop for PendingDeterminismAdmissions<'_, R> {
195 fn drop(&mut self) {
196 if !self.armed {
197 return;
198 }
199 let mut sequences = self.executor.sequences.lock();
200 for request_id in &self.request_ids {
201 sequences.cancel_prefill(request_id);
202 }
203 }
204}
205
206impl<R: DeviceRuntime> VNextModelExecutor<R> {
207 async fn admit_determinism_participants(
208 &self,
209 spec: &VNextDeterminismExecutionSpec,
210 ) -> Result<Vec<(Arc<VNextPrefillSlot<R>>, Arc<VNextSequence<R>>)>> {
211 if spec.participants.iter().any(|participant| {
212 participant.maximum_sequence_tokens > self.maximum_model_tokens
213 || participant.token_ids.iter().any(|token| {
214 usize::try_from(*token).map_or(true, |token| token >= self.info.vocab_size)
215 })
216 }) {
217 return Err(FerrumError::request_validation(
218 "vNext determinism participant exceeds the resolved model token or vocabulary bound",
219 ));
220 }
221
222 let mut pending = PendingDeterminismAdmissions::new(self);
223 for participant in &spec.participants {
224 let request_id = RequestId::new();
225 pending.track(request_id.clone());
226 let input_tokens = participant
227 .token_ids
228 .iter()
229 .copied()
230 .map(TokenId::new)
231 .collect::<Vec<_>>();
232 let mut maintenance_attempts = 0_u32;
233 loop {
234 match self.try_admit_prefill(ExecutorPrefillAdmission::for_diagnostic(
235 &request_id,
236 &input_tokens,
237 participant.maximum_sequence_tokens,
238 ))? {
239 ExecutorPrefillAdmissionDecision::Admitted(receipt) => {
240 if receipt.request_id != request_id {
241 return Err(FerrumError::internal(
242 "vNext determinism admission changed request identity",
243 ));
244 }
245 break;
246 }
247 ExecutorPrefillAdmissionDecision::MaintenanceDeferred(_) => {
248 if maintenance_attempts >= MAX_BACKING_MAINTENANCE_ATTEMPTS {
249 return Err(FerrumError::resource_exhausted(
250 "vNext determinism admission backing did not converge",
251 ));
252 }
253 maintenance_attempts += 1;
254 match self.maintain_prefill_backing(&request_id)? {
255 ExecutorPrefillMaintenanceOutcome::Maintained { .. }
256 | ExecutorPrefillMaintenanceOutcome::RetryAdmission { .. } => continue,
257 ExecutorPrefillMaintenanceOutcome::WaitForRelease { .. } => {
258 return Err(FerrumError::resource_exhausted(
259 "vNext determinism admission is waiting for capacity release",
260 ))
261 }
262 ExecutorPrefillMaintenanceOutcome::NoLongerPending => {
263 return Err(FerrumError::internal(
264 "vNext determinism admission lost its retained request",
265 ))
266 }
267 }
268 }
269 ExecutorPrefillAdmissionDecision::Deferred(deferred) => {
270 return Err(Self::deferred(
271 "determinism participant admission",
272 &deferred,
273 ))
274 }
275 ExecutorPrefillAdmissionDecision::PermanentRejected(rejected) => {
276 return Err(FerrumError::resource_exhausted(format!(
277 "vNext determinism participant exceeds its immutable plan: {rejected:?}"
278 )))
279 }
280 }
281 }
282 }
283
284 let request_ids = pending.request_ids.clone();
285 let admitted = self
286 .sequences
287 .lock()
288 .begin_prefill_batch_execution(&request_ids)?;
289 pending.disarm();
290 Ok(admitted)
291 }
292
293 fn prepare_determinism_wave(
294 &self,
295 step: &Arc<StepResourceLease<R>>,
296 sequences: &[Arc<VNextSequence<R>>],
297 spans: &[TokenSpanWork],
298 ) -> Result<PreparedStepSubmissionWave<R>> {
299 if sequences.len() != spans.len() {
300 return Err(FerrumError::internal(
301 "vNext determinism-wave maintenance participants differ from the work spans",
302 ));
303 }
304 Self::validate_step_maintenance_participants(step, sequences)?;
305 let work_shape = step
306 .shared_all_invocation_work_shape(spans)
307 .map_err(|error| FerrumError::backend(error.to_string()))?;
308 let requests = self
309 .resolved_plan
310 .execution_plan()
311 .payload()
312 .nodes()
313 .iter()
314 .map(|node| {
315 InvocationResourceAdmissionRequest::for_all_step_participants(
316 node.id().clone(),
317 Arc::clone(&work_shape),
318 AdmissionFitPolicy::ImmediateOnly,
319 AdmissionPressureAction::WaitForRelease,
320 )
321 .map_err(|error| FerrumError::backend(error.to_string()))
322 })
323 .collect::<Result<Vec<_>>>()?;
324 let mut prefix_maintenance = self.prefix_pressure_maintenance();
325 let mut backing_attempts = 0_u32;
326 let mut maintenance_receipts = Vec::new();
327 loop {
328 match step
329 .try_prepare_determinism_submission_wave(requests.clone())
330 .map_err(|error| FerrumError::backend(error.to_string()))?
331 {
332 StepSubmissionWaveAdmissionDecision::Prepared(wave) => return Ok(wave),
333 StepSubmissionWaveAdmissionDecision::Deferred(deferred) => {
334 if deferred.action() != DeferredAction::AwaitBackingGrowth {
335 return Err(Self::deferred("determinism submission wave", &deferred));
336 }
337 if !prefix_maintenance.allows_backing_attempt(backing_attempts) {
338 let deferred = ExecutorExecutionCapacityDeferral::from_pending_maintenance(
339 &deferred,
340 ExecutorExecutionCapacityStage::SubmissionWave,
341 )?;
342 return Err(Self::execution_capacity_error(&deferred));
343 }
344 backing_attempts += 1;
345 let outcome = self
346 .plan_resources
347 .maintain_for_admission_deferred(&deferred)
348 .map_err(|error| FerrumError::backend(error.to_string()))?;
349 if let Some(deferred) = self.execution_maintenance_decision(
350 ExecutorExecutionCapacityStage::SubmissionWave,
351 outcome,
352 VNextExecutionMaintenanceSource::Logical(&deferred),
353 sequences.iter().map(Arc::as_ref),
354 &mut maintenance_receipts,
355 &mut prefix_maintenance,
356 )? {
357 return Err(Self::execution_capacity_error(&deferred));
358 }
359 }
360 StepSubmissionWaveAdmissionDecision::BackingDeferred(deferred) => {
361 if !prefix_maintenance.allows_backing_attempt(backing_attempts) {
362 let deferred = ExecutorExecutionCapacityDeferral::from_backing(
363 deferred.evidence(),
364 ExecutorExecutionCapacityStage::SubmissionWave,
365 )?;
366 return Err(Self::execution_capacity_error(&deferred));
367 }
368 backing_attempts += 1;
369 let outcome = deferred
370 .maintain()
371 .map_err(|error| FerrumError::backend(error.to_string()))?;
372 if let Some(deferred) = self.execution_maintenance_decision(
373 ExecutorExecutionCapacityStage::SubmissionWave,
374 outcome,
375 VNextExecutionMaintenanceSource::Backing(deferred.evidence()),
376 sequences.iter().map(Arc::as_ref),
377 &mut maintenance_receipts,
378 &mut prefix_maintenance,
379 )? {
380 return Err(Self::execution_capacity_error(&deferred));
381 }
382 }
383 StepSubmissionWaveAdmissionDecision::PermanentRejected(rejected) => {
384 return Err(FerrumError::backend(format!(
385 "vNext determinism wave exceeds its immutable plan: {rejected:?}"
386 )))
387 }
388 StepSubmissionWaveAdmissionDecision::RequestStateDeferred(deferred) => {
389 return Err(FerrumError::resource_exhausted(format!(
390 "vNext determinism wave is waiting for Request-state hazards: {:?}",
391 deferred.blockers()
392 )))
393 }
394 StepSubmissionWaveAdmissionDecision::RequestStateSplitRequired(split) => {
395 return Err(FerrumError::request_validation(format!(
396 "vNext determinism wave requires sibling split for request {:?}: {:?}",
397 split.request(),
398 split.resource_ids()
399 )))
400 }
401 StepSubmissionWaveAdmissionDecision::RequestStatePoisoned(poison) => {
402 return Err(FerrumError::backend(format!(
403 "vNext determinism Request-state resource is poisoned: {poison:?}"
404 )))
405 }
406 }
407 }
408 }
409
410 fn deterministic_scalar_bytes(element_type: ElementType, nonzero: bool) -> &'static [u8] {
411 if !nonzero {
412 return &[0, 0, 0, 0];
413 }
414 match element_type {
415 ElementType::Bool | ElementType::U8 | ElementType::I8 => &[1],
416 ElementType::F16 => &[0x00, 0x3c],
417 ElementType::Bf16 => &[0x80, 0x3f],
418 ElementType::U32 | ElementType::I32 => &[1, 0, 0, 0],
419 ElementType::F32 => &[0x00, 0x00, 0x80, 0x3f],
420 }
421 }
422
423 fn repeated_deterministic_scalar(
424 element_type: ElementType,
425 length_bytes: usize,
426 nonzero: bool,
427 ) -> Result<Vec<u8>> {
428 let element_width = usize::try_from(element_type.size_bytes())
429 .map_err(|_| FerrumError::internal("determinism element width exceeds usize"))?;
430 if length_bytes == 0 || length_bytes % element_width != 0 {
431 return Err(FerrumError::internal(
432 "determinism payload length differs from its element type",
433 ));
434 }
435 let scalar = Self::deterministic_scalar_bytes(element_type, nonzero);
436 let scalar = &scalar[..element_width];
437 Ok(scalar.iter().copied().cycle().take(length_bytes).collect())
438 }
439
440 fn determinism_external_payload(
441 &self,
442 participant: &VNextDeterminismParticipantSpec,
443 initialization: &ExecutionDeterminismInitializationSpec,
444 length_bytes: usize,
445 ) -> Result<Vec<u8>> {
446 let location = initialization.location();
447 let node_id = location.node_id();
448 let ordinal = location.ordinal();
449 if node_id == &self.io.input_node_id && ordinal == self.io.input_ordinal {
450 if location.element_type() != ElementType::U32 {
451 return Err(FerrumError::internal(
452 "vNext determinism token input is not U32",
453 ));
454 }
455 let bytes = participant
456 .immediate_token_ids()
457 .iter()
458 .flat_map(|token| token.to_le_bytes())
459 .collect::<Vec<_>>();
460 if bytes.len() != length_bytes {
461 return Err(FerrumError::internal(
462 "vNext determinism token bytes differ from the prepared input range",
463 ));
464 }
465 return Ok(bytes);
466 }
467 if node_id == &self.io.token_mask_input_node_id
468 && ordinal == self.io.token_mask_input_ordinal
469 {
470 return Self::repeated_deterministic_scalar(
471 location.element_type(),
472 length_bytes,
473 true,
474 );
475 }
476 if node_id == &self.io.repetition_penalty_input_node_id
477 && ordinal == self.io.repetition_penalty_input_ordinal
478 {
479 if location.element_type() != ElementType::F32 {
480 return Err(FerrumError::internal(
481 "vNext determinism repetition penalty is not F32",
482 ));
483 }
484 return Self::repeated_deterministic_scalar(
485 location.element_type(),
486 length_bytes,
487 true,
488 );
489 }
490 Self::repeated_deterministic_scalar(location.element_type(), length_bytes, false)
491 }
492
493 fn bind_determinism_restore(
494 &self,
495 participants: &[&VNextDeterminismParticipantSpec],
496 participant_order: SubmissionWaveDeterminismParticipantOrder,
497 initial_state: VNextDeterminismInitialState,
498 identity: &BatchOperationIdentity,
499 active_bindings: &[&TrustedActiveSequenceBinding],
500 wave: &PreparedStepSubmissionWave<R>,
501 ) -> Result<SubmissionWaveDeterminismRestore> {
502 let layout =
503 SubmissionWaveDeterminismRestoreLayout::from_prepared_wave_with_participant_order(
504 self.runtime.as_ref(),
505 self.providers.providers(),
506 &self.resolved_plan,
507 identity,
508 active_bindings.iter().copied(),
509 participant_order,
510 wave,
511 )
512 .map_err(|error| FerrumError::backend(error.to_string()))?;
513 if usize::try_from(layout.participant_count()).ok() != Some(participants.len()) {
514 return Err(FerrumError::internal(
515 "vNext determinism restore participant count drifted",
516 ));
517 }
518 let payloads = participants
519 .iter()
520 .enumerate()
521 .map(|(participant_index, participant)| {
522 let ranges = layout
523 .participant_initialization_ranges(u32::try_from(participant_index).map_err(
524 |_| {
525 FerrumError::internal("vNext determinism participant index exceeds u32")
526 },
527 )?)
528 .ok_or_else(|| {
529 FerrumError::internal("vNext determinism restore lost a participant range")
530 })?;
531 layout
532 .witness_plan()
533 .initializations()
534 .iter()
535 .zip(ranges)
536 .map(|(initialization, range)| {
537 let length_bytes = usize::try_from(range.length_bytes()).map_err(|_| {
538 FerrumError::internal("vNext determinism restore range exceeds usize")
539 })?;
540 match initialization.kind() {
541 ExecutionDeterminismInitializationKind::ExternalInput { .. } => self
542 .determinism_external_payload(
543 participant,
544 initialization,
545 length_bytes,
546 ),
547 ExecutionDeterminismInitializationKind::State { .. } => {
548 Self::repeated_deterministic_scalar(
549 initialization.location().element_type(),
550 length_bytes,
551 initial_state == VNextDeterminismInitialState::Nonzero,
552 )
553 }
554 }
555 })
556 .collect::<Result<Vec<_>>>()
557 })
558 .collect::<Result<Vec<_>>>()?;
559 layout
560 .bind(payloads)
561 .map_err(|error| FerrumError::backend(error.to_string()))
562 }
563
564 async fn determinism_dispatch_failure(
565 &self,
566 step: Arc<StepResourceLease<R>>,
567 error: SubmissionWaveDispatchError<R>,
568 ) -> FerrumError {
569 match error {
570 SubmissionWaveDispatchError::DefinitelyNotSubmitted { failures, retry } => {
571 let message = format!(
572 "operation attempt {} with {} participants was definitely not submitted: {}",
573 retry.prior_attempt(),
574 failures.len(),
575 failures
576 .first()
577 .map(|failure| failure.failure().message())
578 .unwrap_or("missing classified participant failure")
579 );
580 drop(retry);
581 self.abort_unsubmitted_step(step, FerrumError::backend(message))
582 }
583 error @ (SubmissionWaveDispatchError::Contract(_)
584 | SubmissionWaveDispatchError::Provider(_)
585 | SubmissionWaveDispatchError::Initialization(_)
586 | SubmissionWaveDispatchError::InputUpload(_)) => {
587 self.abort_unsubmitted_step(step, FerrumError::backend(error.to_string()))
588 }
589 SubmissionWaveDispatchError::SubmissionIndeterminate { recovery } => {
590 let reaper = Arc::clone(&self.reaper);
591 let recovered = self
592 .completion_worker
593 .execute(VNextCompletionTaskKind::IndeterminateRecovery, move || {
594 let recovered = recovery.recover_by_draining_lane();
595 drop(reaper);
596 recovered
597 })
598 .await;
599 let message = match recovered {
600 Ok(Ok(_)) => "vNext determinism submission was indeterminate".to_owned(),
601 Ok(Err(error)) => format!(
602 "vNext determinism submission was indeterminate and recovery failed: {error}"
603 ),
604 Err(error) => format!(
605 "vNext determinism submission recovery task failed: {error}"
606 ),
607 };
608 self.abort_step(step, message).await
609 }
610 SubmissionWaveDispatchError::PostSubmitContract { error, completion } => {
611 let message = error.to_string();
612 let reaper = Arc::clone(&self.reaper);
613 let _ = self
614 .completion_worker
615 .execute(VNextCompletionTaskKind::PostSubmitDrain, move || {
616 let observation = completion.wait();
617 drop(reaper);
618 observation
619 })
620 .await;
621 self.abort_step(step, message).await
622 }
623 }
624 }
625
626 fn abort_prepared_unsubmitted_step(
627 &self,
628 step: Arc<StepResourceLease<R>>,
629 wave: PreparedStepSubmissionWave<R>,
630 error: FerrumError,
631 ) -> FerrumError {
632 drop(wave);
633 self.abort_unsubmitted_step(step, error)
634 }
635
636 pub async fn collect_determinism_execution(
637 &self,
638 spec: &VNextDeterminismExecutionSpec,
639 ) -> Result<SubmissionWaveDeterminismEvidence> {
640 if !self.startup_preparation.lock().is_ready() {
641 return Err(FerrumError::internal(
642 "vNext determinism collection requires completed startup preparation",
643 ));
644 }
645
646 let admitted = self.admit_determinism_participants(spec).await?;
647 let mut execution_guards = admitted
648 .iter()
649 .map(|(slot, sequence)| {
650 VNextPrefillExecutionGuard::new(
651 &self.sequences,
652 Arc::clone(slot),
653 Arc::clone(sequence),
654 )
655 })
656 .collect::<Vec<_>>();
657 let mut participant_by_authority = BTreeMap::new();
658 for (participant_index, (_, sequence)) in admitted.iter().enumerate() {
659 if participant_by_authority
660 .insert(sequence.session.sequence_authority(), participant_index)
661 .is_some()
662 {
663 return Err(FerrumError::internal(
664 "vNext determinism admission duplicated sequence authority",
665 ));
666 }
667 }
668 let batch = ExecutionBatchParticipants::new(
669 admitted
670 .iter()
671 .map(|(_, sequence)| Arc::clone(&sequence.session))
672 .collect(),
673 )
674 .map_err(|error| FerrumError::backend(error.to_string()))?;
675 let canonical_indices = batch
676 .sessions()
677 .iter()
678 .map(|session| {
679 participant_by_authority
680 .remove(&session.sequence_authority())
681 .ok_or_else(|| {
682 FerrumError::internal("vNext determinism canonical participant is absent")
683 })
684 })
685 .collect::<Result<Vec<_>>>()?;
686 if !participant_by_authority.is_empty() {
687 return Err(FerrumError::internal(
688 "vNext determinism participant is absent from its canonical batch",
689 ));
690 }
691 let sequences = canonical_indices
692 .iter()
693 .map(|index| Arc::clone(&admitted[*index].1))
694 .collect::<Vec<_>>();
695 let participants = canonical_indices
696 .iter()
697 .map(|index| &spec.participants[*index])
698 .collect::<Vec<_>>();
699 let mut operation_guards = Vec::with_capacity(sequences.len());
700 for sequence in &sequences {
701 operation_guards.push(sequence.operation.lock().await);
702 }
703 for (sequence, participant) in sequences.iter().zip(&participants) {
704 self.extend_sequence(sequence, participant.full_extension()?)?;
705 }
706 let spans = participants
707 .iter()
708 .map(|participant| {
709 self.retain_checkpoint_token_evidence(
710 participant.token_span()?,
711 participant.token_ids(),
712 )
713 })
714 .collect::<Result<Vec<_>>>()?;
715 let step = match self.begin_step_for_spans_with_capacity(
716 &batch,
717 &sequences,
718 &spans,
719 spec.phase.wave_kind(),
720 )? {
721 VNextExecutionCapacityDecision::Ready(step) => step,
722 VNextExecutionCapacityDecision::Deferred(deferred) => {
723 return Err(Self::execution_capacity_error(&deferred))
724 }
725 VNextExecutionCapacityDecision::RequestStateDeferred(_) => {
726 return Err(FerrumError::internal(
727 "determinism step admission unexpectedly produced a Request-state deferral",
728 ))
729 }
730 };
731 let wave = match self.prepare_determinism_wave(&step, &sequences, &spans) {
732 Ok(wave) => wave,
733 Err(error) => return Err(self.abort_unsubmitted_step(step, error)),
734 };
735 let participant_order =
736 SubmissionWaveDeterminismParticipantOrder::from_logical_participant_sessions(
737 &wave,
738 &admitted
739 .iter()
740 .map(|(_, sequence)| Arc::clone(&sequence.session))
741 .collect::<Vec<_>>(),
742 )
743 .map_err(|error| FerrumError::backend(error.to_string()))?;
744 let active_bindings = sequences
745 .iter()
746 .map(|sequence| sequence.active_binding.as_ref())
747 .collect::<Vec<_>>();
748 let identity = match OperationDispatch::bind_compiled_submission_wave_identity(
749 &self.submission_wave_identity,
750 active_bindings.iter().copied(),
751 &wave,
752 &self.lane,
753 ) {
754 Ok(identity) => identity,
755 Err(error) => {
756 return Err(self.abort_prepared_unsubmitted_step(
757 step,
758 wave,
759 FerrumError::backend(error.to_string()),
760 ))
761 }
762 };
763 let restore = match self.bind_determinism_restore(
764 &participants,
765 participant_order,
766 spec.initial_state,
767 &identity,
768 &active_bindings,
769 &wave,
770 ) {
771 Ok(restore) => restore,
772 Err(error) => {
773 return Err(self.abort_prepared_unsubmitted_step(step, wave, error));
774 }
775 };
776
777 let submission = match spec.mode {
778 VNextDeterminismExecutionMode::Eager => {
779 OperationDispatch::encode_and_submit_determinism_eager_wave(
780 self.providers.providers(),
781 &self.resolved_plan,
782 &identity,
783 active_bindings.iter().copied(),
784 DeviceTimingMode::Off,
785 &restore,
786 spec.workspace_poison.fill_byte(),
787 wave,
788 &self.lane,
789 &self.reaper,
790 )
791 }
792 VNextDeterminismExecutionMode::Replayed => {
793 let program_id =
794 match OperationDispatch::reusable_execution_program_id_for_wave(
795 self.providers.providers(),
796 &self.resolved_plan,
797 &wave,
798 &self.lane,
799 ) {
800 Ok(Some(program_id)) => program_id,
801 Ok(None) => return Err(self.abort_prepared_unsubmitted_step(
802 step,
803 wave,
804 FerrumError::backend(
805 "vNext determinism replay has no exact reusable program identity",
806 ),
807 )),
808 Err(error) => {
809 return Err(self.abort_prepared_unsubmitted_step(
810 step,
811 wave,
812 FerrumError::backend(error.to_string()),
813 ))
814 }
815 };
816 let catalog_snapshot = self.reusable_execution_catalog.read().clone();
817 let catalog = match catalog_snapshot.as_deref() {
818 Some(catalog) if catalog.lane_epoch == self.lane.reusable_execution_epoch() => {
819 catalog
820 }
821 _ => {
822 return Err(self.abort_prepared_unsubmitted_step(
823 step,
824 wave,
825 FerrumError::backend(
826 "vNext determinism replay catalog is absent or stale",
827 ),
828 ))
829 }
830 };
831 let reusable_program = match catalog.programs.get(&program_id) {
832 Some(program) => program,
833 None => {
834 return Err(self.abort_prepared_unsubmitted_step(
835 step,
836 wave,
837 FerrumError::backend(
838 "vNext determinism replay program is absent from the sealed exact catalog",
839 ),
840 ))
841 }
842 };
843 OperationDispatch::encode_and_submit_determinism_replayed_wave(
844 self.providers.providers(),
845 &self.resolved_plan,
846 &identity,
847 active_bindings.iter().copied(),
848 DeviceTimingMode::Off,
849 &restore,
850 spec.workspace_poison.fill_byte(),
851 reusable_program,
852 wave,
853 &self.lane,
854 &self.reaper,
855 )
856 }
857 };
858 let handle = match submission {
859 Ok(handle) => handle,
860 Err(error) => return Err(self.determinism_dispatch_failure(step, error).await),
861 };
862 let reaper = Arc::clone(&self.reaper);
863 let evidence = self
864 .completion_worker
865 .execute(VNextCompletionTaskKind::WaveReadback, move || {
866 let evidence = handle.wait_into_evidence();
867 drop(reaper);
868 evidence
869 })
870 .await
871 .map_err(|error| {
872 FerrumError::backend(format!("vNext determinism completion task failed: {error}"))
873 })?
874 .map_err(|error| FerrumError::backend(error.to_string()));
875 let evidence = match evidence {
876 Ok(evidence) => evidence,
877 Err(error) => return Err(self.abort_step(step, error.to_string()).await),
878 };
879 step.try_retire_normal().map_err(|failure| {
880 FerrumError::backend(format!(
881 "vNext determinism step retirement failed: {}",
882 failure.error()
883 ))
884 })?;
885 drop(operation_guards);
886 for guard in &mut execution_guards {
887 guard.disarm();
888 }
889 {
890 let mut registry = self.sequences.lock();
891 for (slot, sequence) in &admitted {
892 registry.finish_prefill_execution(slot, sequence);
893 }
894 }
895 for (_, sequence) in admitted {
896 sequence.abort();
897 }
898 Ok(evidence)
899 }
900}
901
902#[cfg(test)]
903mod tests {
904 use super::*;
905
906 fn participant(start: usize, end: usize) -> VNextDeterminismParticipantSpec {
907 VNextDeterminismParticipantSpec::new(vec![0; end], start..end, end + 8).unwrap()
908 }
909
910 #[test]
911 fn participant_spec_requires_one_terminal_immediate_span() {
912 assert!(VNextDeterminismParticipantSpec::new(vec![], 0..0, 1).is_err());
913 assert!(VNextDeterminismParticipantSpec::new(vec![0, 1], 0..1, 2).is_err());
914 assert!(VNextDeterminismParticipantSpec::new(vec![0, 1], 1..2, 2).is_ok());
915 }
916
917 #[test]
918 fn decode_spec_rejects_multi_token_participants_and_unbounded_width() {
919 assert!(VNextDeterminismExecutionSpec::new(
920 VNextDeterminismPhase::Decode,
921 vec![participant(0, 2)],
922 VNextDeterminismInitialState::Zero,
923 VNextDeterminismWorkspacePoison::Zero,
924 VNextDeterminismExecutionMode::Eager,
925 )
926 .is_err());
927 assert!(VNextDeterminismExecutionSpec::new(
928 VNextDeterminismPhase::Decode,
929 (0..=MAX_VNEXT_DETERMINISM_PARTICIPANTS)
930 .map(|_| participant(0, 1))
931 .collect(),
932 VNextDeterminismInitialState::Zero,
933 VNextDeterminismWorkspacePoison::Zero,
934 VNextDeterminismExecutionMode::Eager,
935 )
936 .is_err());
937 }
938
939 #[test]
940 fn prefill_spec_accepts_chunk_boundary_shape() {
941 let spec = VNextDeterminismExecutionSpec::new(
942 VNextDeterminismPhase::Prefill,
943 vec![participant(4, 8)],
944 VNextDeterminismInitialState::Nonzero,
945 VNextDeterminismWorkspacePoison::A5,
946 VNextDeterminismExecutionMode::Replayed,
947 )
948 .unwrap();
949 assert_eq!(spec.participants()[0].immediate_range(), 4..8);
950 assert_eq!(spec.workspace_poison(), VNextDeterminismWorkspacePoison::A5);
951 }
952}