ferrum_kernels/backend/cpu/
vnext_runtime.rs

1//! Host execution for the production plan runtime. CPU commands run in order
2//! under a runtime-wide submission lock; errors after execution starts produce
3//! failed, quiescent fences, never permission to retry a partially written batch.
4
5use std::fmt;
6use std::panic::{catch_unwind, AssertUnwindSafe};
7use std::sync::atomic::{AtomicU64, Ordering};
8use std::sync::{Arc, Mutex};
9
10use ferrum_interfaces::vnext::{
11    BufferDescriptor, CopyRegion, DefinitelyNotSubmitted, DeviceAllocationPermit, DeviceClass,
12    DeviceCommandBatch, DeviceCommandPhase, DeviceComputePathRequirement, DeviceDescriptor,
13    DeviceErrorReport, DeviceRuntime, DeviceSubmissionAttribution, DeviceTerminal,
14    DeviceTerminalReceipt, DeviceTimingMeasurement, DeviceTimingMode,
15    DeviceTimingUnavailableReason, FenceIndeterminate, FenceQuery, HostTransferLayout, StreamState,
16    VNextError, DEVICE_COPY_NATIVE_OPERATION_ID, DEVICE_ZERO_NATIVE_OPERATION_ID,
17    HOST_UPLOAD_NATIVE_OPERATION_ID,
18};
19
20mod command;
21mod host_memory;
22mod memory;
23
24use command::CommandKind;
25pub use command::CpuDeviceCommand;
26pub(crate) use command::CpuKernelLaunch;
27pub(crate) use host_memory::{host_memory_available, host_memory_capacity};
28pub(crate) use memory::CpuBufferRegion;
29pub use memory::CpuDeviceBuffer;
30pub(crate) use memory::CpuRegionSet;
31use memory::{MemoryBudget, Storage};
32
33static NEXT_RUNTIME_INSTANCE: AtomicU64 = AtomicU64::new(1);
34
35#[derive(Debug, Clone, PartialEq, Eq)]
36pub struct CpuRuntimeError(String);
37
38impl CpuRuntimeError {
39    pub(crate) fn new(message: impl Into<String>) -> Self {
40        Self(message.into())
41    }
42}
43
44impl fmt::Display for CpuRuntimeError {
45    fn fmt(&self, formatter: &mut fmt::Formatter<'_>) -> fmt::Result {
46        formatter.write_str(&self.0)
47    }
48}
49
50impl std::error::Error for CpuRuntimeError {}
51
52impl From<VNextError> for CpuRuntimeError {
53    fn from(error: VNextError) -> Self {
54        Self::new(error.to_string())
55    }
56}
57
58pub struct CpuDeviceStream {
59    runtime_instance: u64,
60    state: StreamState,
61    failure: Option<CpuRuntimeError>,
62}
63
64pub struct CpuDeviceFence {
65    runtime_instance: u64,
66    failure: Option<CpuRuntimeError>,
67    attribution: Option<DeviceSubmissionAttribution>,
68    timing_mode: DeviceTimingMode,
69}
70
71impl CpuDeviceFence {
72    fn receipt(&self) -> DeviceTerminalReceipt<CpuRuntimeError> {
73        let terminal = match &self.failure {
74            Some(error) => DeviceTerminal::FailedButQuiescent(error.clone()),
75            None => DeviceTerminal::Succeeded,
76        };
77        // Host execution still produces an exact completion fence. Missing
78        // device clocks are unavailable observations, never a compute failure
79        // or a reason to label a host wall-clock duration as device time.
80        DeviceTerminalReceipt::profiled_with_submission_timing(
81            terminal,
82            if self.timing_mode.completion_enabled() {
83                DeviceTimingMeasurement::Unavailable(
84                    DeviceTimingUnavailableReason::BackendUnsupported,
85                )
86            } else {
87                DeviceTimingMeasurement::NotRequested
88            },
89            if self.timing_mode.physical_span_attribution_enabled() {
90                DeviceTimingMeasurement::Unavailable(
91                    DeviceTimingUnavailableReason::BackendUnsupported,
92                )
93            } else {
94                DeviceTimingMeasurement::NotRequested
95            },
96        )
97    }
98}
99
100pub struct CpuDeviceRuntime {
101    descriptor: DeviceDescriptor,
102    runtime_instance: u64,
103    budget: Arc<MemoryBudget>,
104    execution: Mutex<()>,
105}
106
107impl CpuDeviceRuntime {
108    // Only the CPU composition may create a production descriptor. Keep the
109    // public product constructor there, alongside its exact capability registry.
110    pub(crate) fn new(descriptor: DeviceDescriptor) -> Result<Self, CpuRuntimeError> {
111        descriptor.validate()?;
112        if descriptor.class != DeviceClass::Host || descriptor.ordinal != 0 {
113            return Err(CpuRuntimeError::new(
114                "CPU runtime requires the host device descriptor",
115            ));
116        }
117        let runtime_instance = NEXT_RUNTIME_INSTANCE
118            .fetch_update(Ordering::Relaxed, Ordering::Relaxed, |value| {
119                value.checked_add(1)
120            })
121            .map_err(|_| CpuRuntimeError::new("CPU runtime identity exhausted"))?;
122        Ok(Self {
123            budget: MemoryBudget::new(descriptor.total_memory_bytes),
124            descriptor,
125            runtime_instance,
126            execution: Mutex::new(()),
127        })
128    }
129
130    pub fn resident_bytes(&self) -> u64 {
131        self.budget.used()
132    }
133    pub fn peak_resident_bytes(&self) -> u64 {
134        self.budget.peak()
135    }
136
137    fn validate_buffer(&self, buffer: &CpuDeviceBuffer) -> Result<(), CpuRuntimeError> {
138        if buffer.runtime_instance == self.runtime_instance {
139            Ok(())
140        } else {
141            Err(CpuRuntimeError::new(
142                "CPU buffer belongs to another runtime",
143            ))
144        }
145    }
146
147    fn validate_stream(&self, stream: &CpuDeviceStream) -> Result<(), CpuRuntimeError> {
148        if stream.runtime_instance != self.runtime_instance {
149            return Err(CpuRuntimeError::new(
150                "CPU stream belongs to another runtime",
151            ));
152        }
153        if let Some(error) = &stream.failure {
154            return Err(error.clone());
155        }
156        Ok(())
157    }
158
159    fn execute_entries(
160        &self,
161        stream: &mut CpuDeviceStream,
162        entries: Vec<(DeviceCommandPhase, Option<u32>, CpuDeviceCommand)>,
163        attribution_required: bool,
164        timing_mode: DeviceTimingMode,
165    ) -> Result<CpuDeviceFence, DefinitelyNotSubmitted<CpuRuntimeError>> {
166        self.validate_stream(stream)
167            .map_err(DefinitelyNotSubmitted::new)?;
168        if entries.is_empty() {
169            return Err(DefinitelyNotSubmitted::new(CpuRuntimeError::new(
170                "CPU submission is empty",
171            )));
172        }
173        // Every command is validated before any command is executed.
174        for (_, _, command) in &entries {
175            command
176                .validate_runtime(self.runtime_instance)
177                .map_err(DefinitelyNotSubmitted::new)?;
178        }
179        let attribution = if attribution_required {
180            let rows = entries
181                .iter()
182                .enumerate()
183                .map(|(index, (phase, node, command))| {
184                    let index = u32::try_from(index)
185                        .map_err(|_| CpuRuntimeError::new("CPU command index exceeds u32"))?;
186                    command.attribution(index, *node, *phase)
187                })
188                .collect::<Result<Vec<_>, CpuRuntimeError>>()
189                .map_err(DefinitelyNotSubmitted::new)?;
190            Some(DeviceSubmissionAttribution::new(rows).ok_or_else(|| {
191                DefinitelyNotSubmitted::new(CpuRuntimeError::new(
192                    "invalid CPU submission attribution",
193                ))
194            })?)
195        } else {
196            None
197        };
198        let _execution = self.execution.lock().map_err(|_| {
199            DefinitelyNotSubmitted::new(CpuRuntimeError::new("CPU execution lock was poisoned"))
200        })?;
201        stream.state = StreamState::Submitted;
202        let result = catch_unwind(AssertUnwindSafe(|| {
203            for (_, _, command) in &entries {
204                command.execute()?;
205            }
206            Ok::<(), CpuRuntimeError>(())
207        }))
208        .unwrap_or_else(|_| {
209            Err(CpuRuntimeError::new(
210                "CPU command panicked; submission is quiescent and its stream is failed",
211            ))
212        });
213        let failure = result.err();
214        if let Some(error) = &failure {
215            stream.state = StreamState::Failed;
216            stream.failure = Some(error.clone());
217        }
218        // Synchronous execution is now quiescent. Do not claim full-batch work
219        // attribution if execution stopped partway through a provider command.
220        Ok(CpuDeviceFence {
221            runtime_instance: self.runtime_instance,
222            attribution: if failure.is_none() { attribution } else { None },
223            failure,
224            timing_mode,
225        })
226    }
227}
228
229impl DeviceRuntime for CpuDeviceRuntime {
230    type Buffer = CpuDeviceBuffer;
231    type Stream = CpuDeviceStream;
232    type Command = CpuDeviceCommand;
233    type Fence = CpuDeviceFence;
234    type Error = CpuRuntimeError;
235
236    fn descriptor(&self) -> &DeviceDescriptor {
237        &self.descriptor
238    }
239    fn attention_execution_policy(&self) -> ferrum_types::AttentionExecutionPolicy {
240        ferrum_types::AttentionExecutionPolicy::Portable
241    }
242
243    fn allocate(&self, permit: DeviceAllocationPermit<'_>) -> Result<Self::Buffer, Self::Error> {
244        let request = permit.into_request();
245        CpuDeviceBuffer::allocate(
246            BufferDescriptor {
247                resource_id: request.resource_id().clone(),
248                size_bytes: request.size_bytes(),
249                alignment_bytes: request.alignment_bytes(),
250                usage: request.usage(),
251                element_type: request.element_type(),
252            },
253            self.runtime_instance,
254            &self.budget,
255        )
256    }
257
258    fn buffer_descriptor(&self, buffer: &Self::Buffer) -> BufferDescriptor {
259        buffer.descriptor.clone()
260    }
261
262    fn create_stream(&self) -> Result<Self::Stream, Self::Error> {
263        Ok(CpuDeviceStream {
264            runtime_instance: self.runtime_instance,
265            state: StreamState::Ready,
266            failure: None,
267        })
268    }
269
270    fn stream_state(&self, stream: &Self::Stream) -> StreamState {
271        if stream.runtime_instance == self.runtime_instance {
272            stream.state
273        } else {
274            StreamState::Failed
275        }
276    }
277
278    fn encode_copy(
279        &self,
280        source: &Self::Buffer,
281        destination: &Self::Buffer,
282        region: CopyRegion,
283    ) -> Result<Self::Command, Self::Error> {
284        self.validate_buffer(source)?;
285        self.validate_buffer(destination)?;
286        region.validate_bounds(&source.descriptor, &destination.descriptor)?;
287        if source.descriptor.element_type != destination.descriptor.element_type {
288            return Err(CpuRuntimeError::new("CPU copy element types differ"));
289        }
290        let source = source.region(
291            region.source_offset_bytes()..region.source_offset_bytes() + region.length_bytes(),
292        )?;
293        let destination = destination.region(
294            region.destination_offset_bytes()
295                ..region.destination_offset_bytes() + region.length_bytes(),
296        )?;
297        Ok(CpuDeviceCommand::transfer(
298            DEVICE_COPY_NATIVE_OPERATION_ID.as_str(),
299            CommandKind::Copy {
300                source,
301                destination,
302            },
303        ))
304    }
305
306    fn encode_upload(
307        &self,
308        source: &[u8],
309        layout: HostTransferLayout,
310        destination: &Self::Buffer,
311        offset: u64,
312    ) -> Result<Self::Command, Self::Error> {
313        self.validate_buffer(destination)?;
314        layout.validate_bytes(source.len())?;
315        if layout.element_type() != destination.descriptor.element_type {
316            return Err(CpuRuntimeError::new("CPU upload element types differ"));
317        }
318        let size = layout.byte_len()?;
319        let end = offset
320            .checked_add(size)
321            .ok_or_else(|| CpuRuntimeError::new("CPU upload range overflows"))?;
322        let destination = destination.region(offset..end)?;
323        let mut owned = Storage::new(size, 1, &self.budget)?;
324        owned.bytes_mut().copy_from_slice(source);
325        Ok(CpuDeviceCommand::transfer(
326            HOST_UPLOAD_NATIVE_OPERATION_ID.as_str(),
327            CommandKind::Upload {
328                source: owned,
329                destination,
330            },
331        ))
332    }
333
334    fn encode_zero(
335        &self,
336        destination: &Self::Buffer,
337        offset: u64,
338        size: u64,
339    ) -> Result<Self::Command, Self::Error> {
340        self.validate_buffer(destination)?;
341        let end = offset
342            .checked_add(size)
343            .ok_or_else(|| CpuRuntimeError::new("CPU zero range overflows"))?;
344        let destination = destination.region(offset..end)?;
345        Ok(CpuDeviceCommand::transfer(
346            DEVICE_ZERO_NATIVE_OPERATION_ID.as_str(),
347            CommandKind::Zero { destination },
348        ))
349    }
350
351    fn submit(
352        &self,
353        stream: &mut Self::Stream,
354        commands: DeviceCommandBatch<Self::Command>,
355    ) -> Result<Self::Fence, DefinitelyNotSubmitted<Self::Error>> {
356        let timing_mode = commands.timing_mode();
357        validate_submission_requirements(
358            commands.compute_path_requirement(),
359            commands
360                .reusable_execution_capture()
361                .map(|capture| (capture.node_count(), capture.eager_boundary_node_indices())),
362        )
363        .map_err(DefinitelyNotSubmitted::new)?;
364        let attribution = commands
365            .attribution_requirement()
366            .logical_execution_path_required();
367        let entries = commands
368            .into_entries()
369            .into_iter()
370            .map(|entry| {
371                let (phase, node, work, command) = entry.into_parts();
372                let command = match work {
373                    Some(work) => command.bind_logical_work(work)?,
374                    None => command,
375                };
376                Ok((phase, node, command))
377            })
378            .collect::<Result<Vec<_>, CpuRuntimeError>>()
379            .map_err(DefinitelyNotSubmitted::new)?;
380        self.execute_entries(stream, entries, attribution, timing_mode)
381    }
382
383    fn submission_attribution(&self, fence: &Self::Fence) -> Option<DeviceSubmissionAttribution> {
384        (fence.runtime_instance == self.runtime_instance)
385            .then(|| fence.attribution.clone())
386            .flatten()
387    }
388
389    fn query_fence(&self, fence: &Self::Fence) -> FenceQuery<Self::Error> {
390        if fence.runtime_instance != self.runtime_instance {
391            return FenceQuery::Indeterminate(CpuRuntimeError::new(
392                "CPU fence belongs to another runtime",
393            ));
394        }
395        FenceQuery::Terminal(fence.receipt())
396    }
397
398    fn wait_fence(
399        &self,
400        fence: &Self::Fence,
401    ) -> Result<DeviceTerminalReceipt<Self::Error>, FenceIndeterminate<Self::Error>> {
402        if fence.runtime_instance != self.runtime_instance {
403            return Err(FenceIndeterminate::new(CpuRuntimeError::new(
404                "CPU fence belongs to another runtime",
405            )));
406        }
407        Ok(fence.receipt())
408    }
409
410    fn synchronize(&self, stream: &mut Self::Stream) -> Result<(), Self::Error> {
411        self.validate_stream(stream)?;
412        let _execution = self
413            .execution
414            .lock()
415            .map_err(|_| CpuRuntimeError::new("CPU execution lock was poisoned"))?;
416        stream.state = StreamState::Ready;
417        Ok(())
418    }
419
420    fn readback(
421        &self,
422        stream: &mut Self::Stream,
423        source: &Self::Buffer,
424        region: CopyRegion,
425        layout: HostTransferLayout,
426    ) -> Result<Vec<u8>, Self::Error> {
427        self.validate_buffer(source)?;
428        self.validate_stream(stream)?;
429        if layout.element_type() != source.descriptor.element_type {
430            return Err(CpuRuntimeError::new("CPU readback element types differ"));
431        }
432        let output_size = layout.byte_len()?;
433        let output_end = region
434            .destination_offset_bytes()
435            .checked_add(region.length_bytes())
436            .filter(|end| *end <= output_size)
437            .ok_or_else(|| CpuRuntimeError::new("CPU readback exceeds host destination layout"))?;
438        let source_end = region
439            .source_offset_bytes()
440            .checked_add(region.length_bytes())
441            .ok_or_else(|| CpuRuntimeError::new("CPU readback source range overflows"))?;
442        let source = source.region(region.source_offset_bytes()..source_end)?;
443        let output_len = usize::try_from(output_size)
444            .map_err(|_| CpuRuntimeError::new("CPU readback exceeds usize"))?;
445        let output_start = usize::try_from(region.destination_offset_bytes())
446            .map_err(|_| CpuRuntimeError::new("CPU readback offset exceeds usize"))?;
447        let output_end = usize::try_from(output_end)
448            .map_err(|_| CpuRuntimeError::new("CPU readback end exceeds usize"))?;
449        let _reservation = self.budget.reserve(output_size)?;
450        let mut output = Vec::new();
451        output
452            .try_reserve_exact(output_len)
453            .map_err(|error| CpuRuntimeError::new(error.to_string()))?;
454        output.resize(output_len, 0);
455        let _execution = self
456            .execution
457            .lock()
458            .map_err(|_| CpuRuntimeError::new("CPU execution lock was poisoned"))?;
459        source.with_read(|bytes| output[output_start..output_end].copy_from_slice(bytes))?;
460        stream.state = StreamState::Ready;
461        Ok(output)
462    }
463
464    fn describe_error(&self, error: &Self::Error) -> Result<DeviceErrorReport, VNextError> {
465        DeviceErrorReport::new("cpu_runtime", error.to_string(), false)
466    }
467}
468
469fn validate_submission_requirements(
470    path: DeviceComputePathRequirement,
471    capture: Option<(u32, &[u32])>,
472) -> Result<(), CpuRuntimeError> {
473    if matches!(
474        path,
475        DeviceComputePathRequirement::ReplayedOnly
476            | DeviceComputePathRequirement::ReplayedWithDeclaredEagerBoundaries
477    ) {
478        return Err(CpuRuntimeError::new(
479            "CPU runtime supports eager execution only",
480        ));
481    }
482    // Core attaches this metadata to the full eager encoding even when every
483    // provider declares an eager boundary. It is not a replay requirement.
484    // Accept that topology without publishing a reusable executable catalog.
485    if let Some((nodes, boundaries)) = capture {
486        if nodes == 0 || !boundaries.iter().copied().eq(0..nodes) {
487            return Err(CpuRuntimeError::new(
488                "CPU capture topology contains a node that was not declared eager",
489            ));
490        }
491    }
492    Ok(())
493}
494
495#[cfg(test)]
496mod tests;
ferrum_kernels/backend/cpu/vnext_runtime.rs

ferrum_kernels/backend/cpu/
vnext_runtime.rs
