ferrum_interfaces/vnext/operation/
compiled_submission_wave.rs1use std::sync::Arc;
2
3use serde::Serialize;
4
5use super::super::{
6 DeviceId, DeviceRuntime, ExecutablePlanView, ExecutionLane, ExecutionLaneId, NodeInvocationId,
7 PlanHash, PlanId, PreparedStepSubmissionWave, SpanId, TrustedActiveSequenceBinding, VNextError,
8};
9use super::compiled_identity::{
10 CompiledSubmissionWaveIdentity, CompiledSubmissionWaveNodeIdentityTemplate,
11 SubmissionWaveParticipantIdentitySeed,
12};
13use super::foundation::{canonical_operation_fingerprint, invalid_operation};
14use super::{BatchOperationIdentity, OperationDispatch};
15
16impl OperationDispatch {
17 pub fn compile_submission_wave_identity<R>(
18 resolved: &dyn ExecutablePlanView,
19 lane: &Arc<ExecutionLane<R>>,
20 ) -> Result<CompiledSubmissionWaveIdentity, VNextError>
21 where
22 R: DeviceRuntime,
23 {
24 let plan = resolved.execution_plan();
25 let nodes = plan.payload().nodes();
26 if nodes.is_empty()
27 || lane.descriptor() != resolved.device()
28 || lane.descriptor() != resolved.capabilities().device()
29 || lane.descriptor().id != *plan.payload().device_id()
30 || lane.descriptor().runtime_implementation_fingerprint
31 != plan.payload().device_runtime_implementation_fingerprint()
32 {
33 return Err(invalid_operation(
34 "compiled submission-wave identity requires one exact plan/runtime/lane topology",
35 ));
36 }
37 let nodes = nodes
38 .iter()
39 .enumerate()
40 .map(|(node_index, node)| {
41 Ok(CompiledSubmissionWaveNodeIdentityTemplate::new(
42 u32::try_from(node_index).map_err(|_| {
43 invalid_operation("compiled submission-wave node index exceeds u32")
44 })?,
45 node.id().clone(),
46 node.operation_id().clone(),
47 node.selection().selected_provider().clone(),
48 node.provider_implementation_fingerprint().to_owned(),
49 node.provider_execution_semantics(),
50 ))
51 })
52 .collect::<Result<Vec<_>, VNextError>>()?;
53 #[derive(Serialize)]
54 struct FingerprintInput<'a> {
55 domain: &'static str,
56 plan_id: &'a PlanId,
57 plan_hash: &'a PlanHash,
58 device_id: &'a DeviceId,
59 runtime_implementation_fingerprint: &'a str,
60 lane_id: ExecutionLaneId,
61 nodes: &'a [CompiledSubmissionWaveNodeIdentityTemplate],
62 }
63 let fingerprint = canonical_operation_fingerprint(
64 &FingerprintInput {
65 domain: "ferrum.runtime-vnext.compiled-submission-wave-identity.v2",
66 plan_id: plan.payload().plan_id(),
67 plan_hash: plan.plan_hash(),
68 device_id: plan.payload().device_id(),
69 runtime_implementation_fingerprint: plan
70 .payload()
71 .device_runtime_implementation_fingerprint(),
72 lane_id: lane.id(),
73 nodes: &nodes,
74 },
75 "compiled submission-wave identity encode failed",
76 )?;
77 Ok(CompiledSubmissionWaveIdentity::from_validated(
78 plan.payload().plan_id().clone(),
79 plan.plan_hash().clone(),
80 plan.payload().device_id().clone(),
81 plan.payload()
82 .device_runtime_implementation_fingerprint()
83 .to_owned(),
84 lane.id(),
85 nodes,
86 fingerprint,
87 ))
88 }
89
90 pub fn bind_compiled_submission_wave_identity<'binding, R, I>(
91 topology: &CompiledSubmissionWaveIdentity,
92 active_bindings: I,
93 wave: &PreparedStepSubmissionWave<R>,
94 lane: &Arc<ExecutionLane<R>>,
95 ) -> Result<BatchOperationIdentity, VNextError>
96 where
97 R: DeviceRuntime,
98 I: Clone + ExactSizeIterator<Item = &'binding TrustedActiveSequenceBinding>,
99 {
100 let Some(first_node) = wave.nodes().first() else {
101 return Err(invalid_operation(
102 "compiled submission wave requires a non-empty immutable plan",
103 ));
104 };
105 let plan_evidence = first_node.plan_evidence_ref();
106 let frames = first_node.participant_frames();
107 if active_bindings.len() == 0
108 || active_bindings.len() != frames.len()
109 || active_bindings.len() != first_node.participants().len()
110 || active_bindings.len() != first_node.participant_session_identities().len()
111 || wave.execution_lane_id() != topology.lane_id()
112 || lane.id() != topology.lane_id()
113 || !Arc::ptr_eq(first_node.runtime(), lane.runtime_arc())
114 || lane.descriptor().id != *topology.device_id()
115 || lane.descriptor().runtime_implementation_fingerprint
116 != topology.runtime_implementation_fingerprint()
117 || plan_evidence.plan_id() != topology.plan_id()
118 || plan_evidence.plan_hash() != topology.plan_hash()
119 || plan_evidence.device_id() != topology.device_id()
120 || plan_evidence.runtime_implementation_fingerprint()
121 != topology.runtime_implementation_fingerprint()
122 || wave.claimed_backing().plan_hash() != topology.plan_hash()
123 || wave.node_count() != topology.node_count()
124 || wave.nodes().iter().enumerate().any(|(node_index, node)| {
125 topology
126 .node_id_at(node_index)
127 .is_none_or(|compiled_node_id| {
128 node.node_id() != compiled_node_id
129 || node.participant_frames() != frames
130 || node.work_shape().fingerprint()
131 != first_node.work_shape().fingerprint()
132 })
133 })
134 {
135 return Err(invalid_operation(
136 "compiled submission-wave topology differs from its exact plan, lane, work, or participant authority",
137 ));
138 }
139
140 let participant_seeds = first_node
141 .participants()
142 .zip(frames.iter().copied())
143 .zip(first_node.participant_session_identities())
144 .zip(active_bindings)
145 .map(
146 |(((participant, frame), (session_epoch, session_fingerprint)), active)| {
147 active.ensure_open_for_emission()?;
148 if frame.sequence_authority() != participant.sequence_authority()
149 || frame.request_authority() != participant.request_authority()
150 || active.sequence_authority() != participant.sequence_authority()
151 || active.coordinator_id() != participant.coordinator_id()
152 || active.run_id() != participant.run_id()
153 || active.request_id() != participant.request_id()
154 || !active
155 .matches_sequence_session(session_epoch, session_fingerprint)
156 || active.plan().plan_id() != topology.plan_id()
157 || active.plan().plan_hash() != topology.plan_hash()
158 || active.plan().device_id() != topology.device_id()
159 || active.runtime_implementation_fingerprint()
160 != topology.runtime_implementation_fingerprint()
161 {
162 return Err(invalid_operation(
163 "compiled submission-wave participant differs from its live sequence session",
164 ));
165 }
166 let node_count = u64::try_from(topology.node_count()).map_err(|_| {
167 invalid_operation("compiled submission-wave node count exceeds u64")
168 })?;
169 let completed_frames = frame.frame_id().get() - 1;
170 let last_node_index = node_count - 1;
171 let last_node_invocation = completed_frames
172 .checked_mul(node_count)
173 .and_then(|value| value.checked_add(last_node_index))
174 .and_then(|value| value.checked_add(1))
175 .ok_or_else(|| {
176 invalid_operation(
177 "compiled submission-wave node invocation id space is exhausted",
178 )
179 })?;
180 NodeInvocationId::try_from(last_node_invocation)?;
181 let events_per_frame = node_count
182 .checked_mul(3)
183 .and_then(|value| value.checked_add(2))
184 .ok_or_else(|| {
185 invalid_operation(
186 "compiled submission-wave event sequence space is exhausted",
187 )
188 })?;
189 completed_frames
190 .checked_mul(events_per_frame)
191 .and_then(|value| value.checked_add(last_node_index.checked_mul(3)?))
192 .and_then(|value| value.checked_add(5))
193 .ok_or_else(|| {
194 invalid_operation(
195 "compiled submission-wave event sequence space is exhausted",
196 )
197 })?;
198 let span_root = format!("vnext/request/{}", active.fingerprint());
199 let node_span = SpanId::new(format!(
200 "{span_root}/frame/{}/node/{last_node_invocation}",
201 frame.frame_id()
202 ))?;
203 SpanId::new(format!("{node_span}/operation"))?;
204 let provisioning = active.static_provisioning_identity();
205 Ok(SubmissionWaveParticipantIdentitySeed::new(
206 frame,
207 active.run_id().clone(),
208 active.request_id().clone(),
209 active.static_pool_id(),
210 active
211 .static_pool_identity_fingerprint_ref()
212 .map(str::to_owned),
213 provisioning.map(|identity| identity.run_id().clone()),
214 provisioning.map(|identity| identity.request_id().clone()),
215 provisioning.map(|identity| identity.transaction_id().clone()),
216 active.sequence_authority().sparse_id(),
217 active.sequence_authority().generation(),
218 active.activation_epoch(),
219 active.runtime_implementation_fingerprint().to_owned(),
220 active.fingerprint().to_owned(),
221 span_root,
222 ))
223 },
224 )
225 .collect::<Result<Vec<_>, VNextError>>()?;
226
227 BatchOperationIdentity::from_compiled_wave(
228 topology.clone(),
229 wave.batch_step_id(),
230 wave.batch_invocation_id(),
231 wave.fingerprint().to_owned(),
232 first_node.work_shape().fingerprint().to_owned(),
233 participant_seeds,
234 )
235 }
236}