use serde::{ser::SerializeSeq, Serialize, Serializer};
use sha2::{Digest, Sha256};
use std::collections::BTreeMap;
use std::fmt;
use std::num::NonZeroU64;
use std::ops::Range;
use std::panic::{catch_unwind, AssertUnwindSafe};
use std::sync::atomic::{AtomicBool, AtomicU64, Ordering};
use std::sync::{Arc, Mutex, MutexGuard, OnceLock, Weak};
use std::time::Instant;
use super::{
classify_device_error, defer_device_cleanup, translate_step_participant_readback_range,
AllocationKind, AllocationLifetime, BackingInitializationEncodeError, BatchOperationIdentity,
BatchParticipantAuthority, BufferUsage, CopyRegion, DeferredDeviceCleanupDisposition,
DeferredDeviceCleanupDomainId, DeferredDeviceCleanupTask, DefinitelyNotSubmittedRetryAuthority,
DefinitelyNotSubmittedWaveRetryAuthority, DeviceCommandBatch, DeviceDescriptor,
DeviceExecutionTiming, DeviceReusableExecutionPlan, DeviceReusableExecutionPreparation,
DeviceReusableExecutionPreparationState, DeviceReusableExecutionProgram, DeviceRuntime,
DeviceSubmissionExecutionTiming, DeviceSubmissionTimingSink, DeviceTerminal,
DeviceTerminalReceipt, DeviceTimingMeasurement, DeviceTimingMode,
DeviceTimingUnavailableReason, ExecutionIdentityEnvelope, ExecutionLaneId, FenceQuery,
HostTransferLayout, IdentifiedFailure, InvocationResourceLease, LogicalBackingBufferView,
NodeId, PreparedStepSubmissionWave, RequestStateHazardTerminalDisposition, ResourceId,
StreamState, VNextError,
};
mod readback_collection;
pub use readback_collection::*;
fn invalid_completion(reason: impl Into<String>) -> VNextError {
VNextError::InvalidExecutionPlan {
reason: reason.into(),
}
}
fn canonical_completion_fingerprint(value: &impl Serialize) -> String {
format!(
"{:x}",
Sha256::digest(
serde_json::to_vec(value).expect("trusted completion evidence must serialize")
)
)
}
#[derive(Debug)]
pub enum ExecutionLaneCreationError<E> {
Contract(VNextError),
Device(E),
}
impl<E: fmt::Display> fmt::Display for ExecutionLaneCreationError<E> {
fn fmt(&self, formatter: &mut fmt::Formatter<'_>) -> fmt::Result {
match self {
Self::Contract(error) => write!(formatter, "execution lane contract failed: {error}"),
Self::Device(error) => write!(formatter, "execution lane creation failed: {error}"),
}
}
}
impl<E: std::error::Error + 'static> std::error::Error for ExecutionLaneCreationError<E> {}
struct ExecutionLaneState<S> {
stream: S,
in_flight: u64,
fail_closed: bool,
}
#[derive(Debug, Clone)]
pub struct ExecutionLaneReusableExecutionCatalog {
epoch: u64,
programs: Vec<DeviceReusableExecutionProgram>,
}
impl ExecutionLaneReusableExecutionCatalog {
pub const fn epoch(&self) -> u64 {
self.epoch
}
pub fn programs(&self) -> &[DeviceReusableExecutionProgram] {
&self.programs
}
pub fn into_parts(self) -> (u64, Vec<DeviceReusableExecutionProgram>) {
(self.epoch, self.programs)
}
}
enum LaneReadbackError<E> {
Contract(VNextError),
Device(E),
}
struct LaneReadback {
bytes: Vec<u8>,
timing: DeviceTimingMeasurement<CompletionReadbackTiming>,
}
#[must_use = "scheduler-owned lanes must outlive every fence enqueued on them"]
pub struct ExecutionLane<R: DeviceRuntime> {
id: ExecutionLaneId,
runtime: Arc<R>,
descriptor: DeviceDescriptor,
fail_closed: AtomicBool,
reusable_execution_epoch: AtomicU64,
state: Mutex<ExecutionLaneState<R::Stream>>,
}
impl<R: DeviceRuntime> fmt::Debug for ExecutionLane<R> {
fn fmt(&self, formatter: &mut fmt::Formatter<'_>) -> fmt::Result {
formatter
.debug_struct("ExecutionLane")
.field("id", &self.id)
.field("device_id", &self.descriptor.id)
.field(
"runtime_implementation_fingerprint",
&self.descriptor.runtime_implementation_fingerprint,
)
.finish_non_exhaustive()
}
}
impl<R: DeviceRuntime> ExecutionLane<R> {
pub fn create(runtime: Arc<R>) -> Result<Arc<Self>, ExecutionLaneCreationError<R::Error>> {
runtime
.descriptor()
.validate()
.map_err(ExecutionLaneCreationError::Contract)?;
let descriptor = runtime.descriptor().clone();
let stream = runtime
.create_stream()
.map_err(ExecutionLaneCreationError::Device)?;
if runtime.descriptor() != &descriptor
|| runtime.stream_state(&stream) != StreamState::Ready
{
return Err(ExecutionLaneCreationError::Contract(invalid_completion(
"execution lane creation requires a stable runtime descriptor and ready stream",
)));
}
Ok(Arc::new(Self {
id: ExecutionLaneId::mint().map_err(ExecutionLaneCreationError::Contract)?,
runtime,
descriptor,
fail_closed: AtomicBool::new(false),
reusable_execution_epoch: AtomicU64::new(1),
state: Mutex::new(ExecutionLaneState {
stream,
in_flight: 0,
fail_closed: false,
}),
}))
}
pub fn descriptor(&self) -> &DeviceDescriptor {
&self.descriptor
}
pub const fn id(&self) -> ExecutionLaneId {
self.id
}
pub fn is_reusable(&self) -> bool {
if self.fail_closed.load(Ordering::Acquire) {
return false;
}
self.state
.lock()
.map(|state| !state.fail_closed && !self.fail_closed.load(Ordering::Acquire))
.unwrap_or(false)
}
pub fn is_fail_closed(&self) -> bool {
!self.is_reusable()
}
pub fn in_flight_count(&self) -> u64 {
self.state
.lock()
.map(|state| state.in_flight)
.unwrap_or(u64::MAX)
}
pub fn reusable_execution_epoch(&self) -> u64 {
self.reusable_execution_epoch.load(Ordering::Acquire)
}
fn with_quiescent_stream<T>(
&self,
operation: &'static str,
action: impl FnOnce(&R, &mut R::Stream) -> Result<T, R::Error>,
) -> Result<T, VNextError> {
if self.fail_closed.load(Ordering::Acquire) {
return Err(invalid_completion(format!(
"fail-closed execution lane cannot {operation}"
)));
}
let mut state = self
.state
.lock()
.map_err(|_| invalid_completion("execution lane state mutex is poisoned"))?;
if state.fail_closed || self.fail_closed.load(Ordering::Acquire) {
return Err(invalid_completion(format!(
"fail-closed execution lane cannot {operation}"
)));
}
if state.in_flight != 0
|| !self.current_descriptor_matches_snapshot()
|| self.runtime.stream_state(&state.stream) != StreamState::Ready
{
return Err(invalid_completion(format!(
"{operation} requires a stable, quiescent execution lane"
)));
}
let result = catch_unwind(AssertUnwindSafe(|| {
action(self.runtime.as_ref(), &mut state.stream)
}));
match result {
Ok(Ok(value))
if self.current_descriptor_matches_snapshot()
&& self.runtime.stream_state(&state.stream) == StreamState::Ready =>
{
Ok(value)
}
Ok(Ok(_)) => {
state.fail_closed = true;
self.fail_closed.store(true, Ordering::Release);
Err(invalid_completion(format!(
"{operation} changed the execution lane state"
)))
}
Ok(Err(error)) => {
state.fail_closed = true;
self.fail_closed.store(true, Ordering::Release);
Err(invalid_completion(format!("{operation} failed: {error}")))
}
Err(_) => {
state.fail_closed = true;
self.fail_closed.store(true, Ordering::Release);
Err(invalid_completion(format!(
"device runtime panicked while attempting to {operation}"
)))
}
}
}
pub fn configure_reusable_executables(
&self,
plan: DeviceReusableExecutionPlan,
) -> Result<DeviceReusableExecutionPreparation, VNextError> {
self.with_quiescent_stream("configure reusable executables", |runtime, stream| {
runtime.configure_reusable_executables(stream, plan)
})
}
pub fn seal_reusable_executables(
&self,
) -> Result<DeviceReusableExecutionPreparation, VNextError> {
self.with_quiescent_stream("seal reusable executables", |runtime, stream| {
runtime.seal_reusable_executables(stream)
})
}
pub fn reusable_executable_preparation(
&self,
) -> Result<DeviceReusableExecutionPreparation, VNextError> {
self.with_quiescent_stream(
"inspect reusable executable preparation",
|runtime, stream| runtime.reusable_executable_preparation(stream),
)
}
pub fn reusable_execution_catalog(
&self,
) -> Result<ExecutionLaneReusableExecutionCatalog, VNextError> {
self.with_quiescent_stream("inspect reusable execution catalog", |runtime, stream| {
runtime.reusable_execution_catalog(stream).map(|programs| {
ExecutionLaneReusableExecutionCatalog {
epoch: self.reusable_execution_epoch(),
programs,
}
})
})
}
pub(crate) fn trim_reusable_executables_if_quiescent(&self) -> Result<bool, VNextError> {
if self.fail_closed.load(Ordering::Acquire) {
return Err(invalid_completion(
"fail-closed execution lane cannot trim reusable executables",
));
}
let mut state = self
.state
.lock()
.map_err(|_| invalid_completion("execution lane state mutex is poisoned"))?;
if state.fail_closed || self.fail_closed.load(Ordering::Acquire) {
return Err(invalid_completion(
"fail-closed execution lane cannot trim reusable executables",
));
}
if state.in_flight != 0 {
return Ok(false);
}
if !self.current_descriptor_matches_snapshot()
|| self.runtime.stream_state(&state.stream) != StreamState::Ready
{
state.fail_closed = true;
self.fail_closed.store(true, Ordering::Release);
return Err(invalid_completion(
"reusable executable trim requires a stable, quiescent execution lane",
));
}
let trimmed = catch_unwind(AssertUnwindSafe(|| {
let preparation = self
.runtime
.reusable_executable_preparation(&state.stream)?;
if preparation.state() == DeviceReusableExecutionPreparationState::Ready
&& preparation.resident_executables() != 0
{
return Ok(None);
}
self.runtime
.trim_reusable_executables(&mut state.stream)
.map(Some)
}));
match trimmed {
Ok(Ok(None))
if self.current_descriptor_matches_snapshot()
&& self.runtime.stream_state(&state.stream) == StreamState::Ready =>
{
Ok(false)
}
Ok(Ok(Some(trim)))
if self.current_descriptor_matches_snapshot()
&& self.runtime.stream_state(&state.stream) == StreamState::Ready =>
{
if trim.released_executables() != 0 {
let epoch = self.reusable_execution_epoch.load(Ordering::Relaxed);
let Some(next_epoch) = epoch.checked_add(1) else {
state.fail_closed = true;
self.fail_closed.store(true, Ordering::Release);
return Err(invalid_completion(
"reusable execution catalog epoch is exhausted",
));
};
self.reusable_execution_epoch
.store(next_epoch, Ordering::Release);
}
Ok(true)
}
Ok(Ok(_)) => {
state.fail_closed = true;
self.fail_closed.store(true, Ordering::Release);
Err(invalid_completion(
"reusable executable trim changed the execution lane state",
))
}
Ok(Err(error)) => {
state.fail_closed = true;
self.fail_closed.store(true, Ordering::Release);
Err(invalid_completion(format!(
"device reusable executable inspection or trim failed: {error}"
)))
}
Err(_) => {
state.fail_closed = true;
self.fail_closed.store(true, Ordering::Release);
Err(invalid_completion(
"device runtime panicked while trimming reusable executables",
))
}
}
}
pub(crate) fn runtime(&self) -> &R {
&self.runtime
}
pub(crate) fn runtime_arc(&self) -> &Arc<R> {
&self.runtime
}
pub(crate) fn current_descriptor_matches_snapshot(&self) -> bool {
self.runtime.descriptor() == &self.descriptor
}
pub(crate) fn reserve_enqueue(&self) -> Result<ExecutionLaneEnqueue<'_, R>, VNextError> {
if self.fail_closed.load(Ordering::Acquire) {
return Err(invalid_completion("execution lane is fail-closed"));
}
if !self.current_descriptor_matches_snapshot() {
return Err(invalid_completion(
"execution lane runtime descriptor differs from its creation snapshot",
));
}
let state = self
.state
.lock()
.map_err(|_| invalid_completion("execution lane state mutex is poisoned"))?;
if self.fail_closed.load(Ordering::Acquire)
|| state.fail_closed
|| !matches!(
self.runtime.stream_state(&state.stream),
StreamState::Ready | StreamState::Submitted
)
{
return Err(invalid_completion(
"execution lane is failed or not enqueue-capable",
));
}
Ok(ExecutionLaneEnqueue { lane: self, state })
}
fn query_fence(&self, fence: &R::Fence) -> FenceQuery<R::Error> {
self.runtime.query_fence(fence)
}
fn wait_fence(
&self,
fence: &R::Fence,
) -> Result<DeviceTerminalReceipt<R::Error>, super::FenceIndeterminate<R::Error>> {
self.runtime.wait_fence(fence)
}
fn finish_one_terminal(&self) -> Result<(), QuiescentCompletionContractFailure> {
let mut state = match self.state.lock() {
Ok(state) => state,
Err(poisoned) => poisoned.into_inner(),
};
if state.in_flight == 0 {
state.fail_closed = true;
self.fail_closed.store(true, Ordering::Release);
return Err(QuiescentCompletionContractFailure::new(
"completion lane in-flight accounting underflowed",
));
}
state.in_flight -= 1;
let descriptor_matches = catch_unwind(AssertUnwindSafe(|| {
self.current_descriptor_matches_snapshot()
}))
.unwrap_or(false);
let stream_not_failed = catch_unwind(AssertUnwindSafe(|| {
!matches!(
self.runtime.stream_state(&state.stream),
StreamState::Failed
)
}))
.unwrap_or(false);
if !descriptor_matches || !stream_not_failed {
state.fail_closed = true;
self.fail_closed.store(true, Ordering::Release);
}
if !descriptor_matches {
Err(QuiescentCompletionContractFailure::new(
"completion runtime descriptor drifted at terminal accounting",
))
} else if !stream_not_failed {
Err(QuiescentCompletionContractFailure::new(
"completion lane stream entered failed state at terminal accounting",
))
} else {
Ok(())
}
}
fn readback_buffer(
&self,
backing: &LogicalBackingBufferView<'_, R::Buffer>,
expected_usage: BufferUsage,
logical_offset_bytes: u64,
output_layout: HostTransferLayout,
timing_mode: DeviceTimingMode,
) -> Result<LaneReadback, LaneReadbackError<R::Error>> {
let timing_started = timing_mode.completion_enabled().then(Instant::now);
let output_bytes = output_layout
.byte_len()
.map_err(LaneReadbackError::Contract)?;
let logical_end = logical_offset_bytes
.checked_add(output_bytes)
.ok_or_else(|| {
LaneReadbackError::Contract(invalid_completion(
"completion readback logical range overflows u64",
))
})?;
let element_bytes = output_layout.element_type().size_bytes();
if backing.usage() != expected_usage
|| backing.element_type() != output_layout.element_type()
|| logical_end > backing.size_bytes()
|| logical_offset_bytes % element_bytes != 0
|| output_bytes % element_bytes != 0
{
return Err(LaneReadbackError::Contract(invalid_completion(
"completion readback must select an aligned typed backing range with matching usage and element type",
)));
}
if self.fail_closed.load(Ordering::Acquire) || !self.current_descriptor_matches_snapshot() {
return Err(LaneReadbackError::Contract(invalid_completion(
"completion readback requires its original reusable execution lane",
)));
}
let mut state = self.state.lock().map_err(|_| {
LaneReadbackError::Contract(invalid_completion(
"execution lane state mutex is poisoned during completion readback",
))
})?;
if state.fail_closed
|| !matches!(
self.runtime.stream_state(&state.stream),
StreamState::Ready | StreamState::Submitted
)
{
return Err(LaneReadbackError::Contract(invalid_completion(
"completion readback lane is failed or not readable",
)));
}
let output_capacity = usize::try_from(output_bytes).map_err(|_| {
LaneReadbackError::Contract(invalid_completion(
"completion readback output exceeds host address space",
))
})?;
let mut output = Vec::with_capacity(output_capacity);
let mut readback_calls = 0_u32;
let mut logical_cursor = 0_u64;
for binding in backing.segment_bindings() {
let segment = binding.segment();
let segment_logical_end = logical_cursor
.checked_add(segment.length_bytes())
.ok_or_else(|| {
LaneReadbackError::Contract(invalid_completion(
"completion readback backing coverage overflows u64",
))
})?;
let overlap_start = logical_cursor.max(logical_offset_bytes);
let overlap_end = segment_logical_end.min(logical_end);
if overlap_start < overlap_end {
let within_segment = overlap_start - logical_cursor;
let source_offset = segment
.offset_bytes()
.checked_add(within_segment)
.ok_or_else(|| {
LaneReadbackError::Contract(invalid_completion(
"completion readback physical offset overflows u64",
))
})?;
let length = overlap_end - overlap_start;
if source_offset % element_bytes != 0 || length % element_bytes != 0 {
return Err(LaneReadbackError::Contract(invalid_completion(
"completion readback backing segments split an element",
)));
}
let actual = self.runtime.buffer_descriptor(binding.buffer());
if &actual != binding.descriptor()
|| source_offset
.checked_add(length)
.is_none_or(|end| end > actual.size_bytes)
{
return Err(LaneReadbackError::Contract(invalid_completion(
"completion readback backing descriptor differs from its committed extent",
)));
}
let piece_layout =
HostTransferLayout::new(output_layout.element_type(), length / element_bytes)
.map_err(LaneReadbackError::Contract)?;
let region = CopyRegion::new(source_offset, 0, length)
.map_err(LaneReadbackError::Contract)?;
readback_calls = readback_calls.checked_add(1).ok_or_else(|| {
LaneReadbackError::Contract(invalid_completion(
"completion readback call count exceeds u32",
))
})?;
let piece = match self.runtime.readback(
&mut state.stream,
binding.buffer(),
region,
piece_layout,
) {
Ok(piece) => piece,
Err(error) => {
state.fail_closed = true;
self.fail_closed.store(true, Ordering::Release);
return Err(LaneReadbackError::Device(error));
}
};
if piece.len() != usize::try_from(length).unwrap_or(usize::MAX) {
state.fail_closed = true;
self.fail_closed.store(true, Ordering::Release);
return Err(LaneReadbackError::Contract(invalid_completion(
"device runtime returned an invalid completion readback byte count",
)));
}
output.extend_from_slice(&piece);
}
logical_cursor = segment_logical_end;
}
if output.len() != output_capacity || !self.current_descriptor_matches_snapshot() {
state.fail_closed = true;
self.fail_closed.store(true, Ordering::Release);
return Err(LaneReadbackError::Contract(invalid_completion(
"completion readback did not cover its exact logical output",
)));
}
let timing = match timing_started {
Some(started) => {
let elapsed_ns = u64::try_from(started.elapsed().as_nanos()).map_err(|_| {
LaneReadbackError::Contract(invalid_completion(
"completion readback host duration exceeds u64 nanoseconds",
))
})?;
let bytes = u64::try_from(output.len()).map_err(|_| {
LaneReadbackError::Contract(invalid_completion(
"completion readback output length exceeds u64",
))
})?;
DeviceTimingMeasurement::Measured(CompletionReadbackTiming::new(
elapsed_ns,
readback_calls,
bytes,
))
}
None => DeviceTimingMeasurement::NotRequested,
};
Ok(LaneReadback {
bytes: output,
timing,
})
}
fn drain(&self, retires_submitted_fence: bool) -> bool {
self.fail_closed.store(true, Ordering::Release);
let mut state = match self.state.lock() {
Ok(state) => state,
Err(poisoned) => poisoned.into_inner(),
};
state.fail_closed = true;
let drained = catch_unwind(AssertUnwindSafe(|| {
self.runtime.synchronize(&mut state.stream).is_ok()
&& self.current_descriptor_matches_snapshot()
&& self.runtime.stream_state(&state.stream) == StreamState::Ready
}))
.unwrap_or(false);
if !drained {
state.fail_closed = true;
self.fail_closed.store(true, Ordering::Release);
return false;
}
if retires_submitted_fence {
let Some(remaining) = state.in_flight.checked_sub(1) else {
state.fail_closed = true;
self.fail_closed.store(true, Ordering::Release);
return false;
};
state.in_flight = remaining;
}
true
}
pub(crate) fn fail_closed(&self) {
self.fail_closed.store(true, Ordering::Release);
}
}
pub(crate) enum LaneSubmitOutcome<F, E> {
Submitted(F),
DefinitelyNotSubmitted(E),
PossiblySubmittedPanic,
}
pub(crate) struct ExecutionLaneEnqueue<'a, R: DeviceRuntime> {
lane: &'a ExecutionLane<R>,
state: MutexGuard<'a, ExecutionLaneState<R::Stream>>,
}
impl<R: DeviceRuntime> ExecutionLaneEnqueue<'_, R> {
fn submit_via(
&mut self,
submit: impl FnOnce(
&R,
&mut R::Stream,
) -> Result<R::Fence, super::DefinitelyNotSubmitted<R::Error>>,
) -> LaneSubmitOutcome<R::Fence, R::Error> {
let runtime = self.lane.runtime.as_ref();
let stream = &mut self.state.stream;
let submission = catch_unwind(AssertUnwindSafe(|| submit(runtime, stream)));
match submission {
Ok(Ok(fence)) => {
if let Some(next) = self.state.in_flight.checked_add(1) {
self.state.in_flight = next;
} else {
self.state.fail_closed = true;
self.lane.fail_closed.store(true, Ordering::Release);
}
LaneSubmitOutcome::Submitted(fence)
}
Ok(Err(not_submitted)) => {
LaneSubmitOutcome::DefinitelyNotSubmitted(not_submitted.into_error())
}
Err(_) => {
self.state.fail_closed = true;
self.lane.fail_closed.store(true, Ordering::Release);
LaneSubmitOutcome::PossiblySubmittedPanic
}
}
}
pub(crate) fn submit(
&mut self,
commands: super::DeviceCommandBatch<R::Command>,
) -> LaneSubmitOutcome<R::Fence, R::Error> {
self.submit_via(|runtime, stream| runtime.submit(stream, commands))
}
pub(crate) fn submit_with_timing<S>(
&mut self,
commands: DeviceCommandBatch<R::Command>,
timing_sink: &S,
) -> LaneSubmitOutcome<R::Fence, R::Error>
where
S: DeviceSubmissionTimingSink,
{
self.submit_via(|runtime, stream| runtime.submit_with_timing(stream, commands, timing_sink))
}
}
#[derive(Debug, Clone, Copy, PartialEq, Eq, PartialOrd, Ord, Hash, Serialize)]
#[serde(transparent)]
pub struct CompletionSlotId(NonZeroU64);
impl CompletionSlotId {
pub const fn get(self) -> u64 {
self.0.get()
}
}
#[derive(Debug)]
struct SubmittedOperationReceiptData {
slot_id: CompletionSlotId,
batch_identity: BatchOperationIdentity,
participants: OnceLock<Vec<SubmittedOperationParticipantReceipt>>,
fingerprint: String,
}
#[derive(Debug, Clone)]
#[must_use = "physical submission evidence must be recorded"]
pub struct SubmittedOperationReceipt {
data: Arc<SubmittedOperationReceiptData>,
}
impl PartialEq for SubmittedOperationReceipt {
fn eq(&self, other: &Self) -> bool {
self.data.slot_id == other.data.slot_id
&& self.data.batch_identity == other.data.batch_identity
&& self.data.fingerprint == other.data.fingerprint
}
}
impl Eq for SubmittedOperationReceipt {}
impl Serialize for SubmittedOperationReceipt {
fn serialize<S>(&self, serializer: S) -> Result<S::Ok, S::Error>
where
S: Serializer,
{
#[derive(Serialize)]
struct Wire<'a> {
slot_id: CompletionSlotId,
batch_identity: &'a BatchOperationIdentity,
participants: &'a [SubmittedOperationParticipantReceipt],
fingerprint: &'a str,
}
Wire {
slot_id: self.data.slot_id,
batch_identity: &self.data.batch_identity,
participants: self.participants(),
fingerprint: &self.data.fingerprint,
}
.serialize(serializer)
}
}
#[derive(Debug, PartialEq, Eq, Serialize)]
struct SubmittedOperationParticipantReceiptData {
slot_id: CompletionSlotId,
participant_index: u32,
identity: ExecutionIdentityEnvelope,
batch_submission_fingerprint: String,
}
#[derive(Debug, Clone, PartialEq, Eq)]
#[must_use = "participant submission projections must remain linked to the physical batch"]
pub struct SubmittedOperationParticipantReceipt {
data: Arc<SubmittedOperationParticipantReceiptData>,
}
impl SubmittedOperationParticipantReceipt {
fn new(
slot_id: CompletionSlotId,
participant_index: u32,
identity: ExecutionIdentityEnvelope,
batch_submission_fingerprint: String,
) -> Self {
Self {
data: Arc::new(SubmittedOperationParticipantReceiptData {
slot_id,
participant_index,
identity,
batch_submission_fingerprint,
}),
}
}
}
impl Serialize for SubmittedOperationParticipantReceipt {
fn serialize<S>(&self, serializer: S) -> Result<S::Ok, S::Error>
where
S: Serializer,
{
self.data.as_ref().serialize(serializer)
}
}
impl SubmittedOperationReceipt {
fn new(
slot_id: CompletionSlotId,
batch_identity: BatchOperationIdentity,
fingerprint: String,
) -> Self {
Self {
data: Arc::new(SubmittedOperationReceiptData {
slot_id,
batch_identity,
participants: OnceLock::new(),
fingerprint,
}),
}
}
pub fn slot_id(&self) -> CompletionSlotId {
self.data.slot_id
}
pub fn batch_identity(&self) -> &BatchOperationIdentity {
&self.data.batch_identity
}
pub fn participants(&self) -> &[SubmittedOperationParticipantReceipt] {
self.data.participants.get_or_init(|| {
self.data
.batch_identity
.participants()
.iter()
.map(|participant| {
SubmittedOperationParticipantReceipt::new(
self.data.slot_id,
participant.participant_index(),
participant.identity().clone(),
self.data.fingerprint.clone(),
)
})
.collect()
})
}
#[doc(hidden)]
pub fn has_materialized_participant_receipts(&self) -> bool {
self.data.participants.get().is_some()
}
pub fn fingerprint(&self) -> &str {
&self.data.fingerprint
}
}
impl SubmittedOperationParticipantReceipt {
pub fn slot_id(&self) -> CompletionSlotId {
self.data.slot_id
}
pub fn participant_index(&self) -> u32 {
self.data.participant_index
}
pub fn identity(&self) -> &ExecutionIdentityEnvelope {
&self.data.identity
}
pub fn batch_submission_fingerprint(&self) -> &str {
&self.data.batch_submission_fingerprint
}
}
#[derive(Debug, Clone, PartialEq, Eq, Serialize)]
#[serde(rename_all = "snake_case", tag = "status", content = "failure")]
pub enum OperationCompletionDisposition {
Succeeded,
FailedButQuiescent(Vec<IdentifiedFailure>),
ContractFailedButQuiescent(QuiescentCompletionContractFailure),
}
#[derive(Debug, Clone, PartialEq, Eq, Serialize)]
#[serde(rename_all = "snake_case", tag = "status", content = "failure")]
pub enum OperationParticipantCompletionDisposition {
Succeeded,
FailedButQuiescent(IdentifiedFailure),
ContractFailedButQuiescent(QuiescentCompletionContractFailure),
}
#[derive(Debug, Clone, PartialEq, Eq, Serialize)]
#[must_use = "participant completion projections must remain linked to the physical batch"]
pub struct OperationParticipantCompletionReceipt {
submission: SubmittedOperationParticipantReceipt,
disposition: OperationParticipantCompletionDisposition,
batch_completion_fingerprint: String,
}
impl OperationParticipantCompletionReceipt {
pub fn submission(&self) -> &SubmittedOperationParticipantReceipt {
&self.submission
}
pub fn disposition(&self) -> &OperationParticipantCompletionDisposition {
&self.disposition
}
pub fn batch_completion_fingerprint(&self) -> &str {
&self.batch_completion_fingerprint
}
}
#[derive(Debug, Clone, PartialEq, Eq, Serialize)]
#[must_use = "a quiescent contract failure is a terminal operation outcome"]
pub struct QuiescentCompletionContractFailure {
reason: String,
}
impl QuiescentCompletionContractFailure {
fn new(reason: impl Into<String>) -> Self {
Self {
reason: reason.into(),
}
}
pub fn reason(&self) -> &str {
&self.reason
}
}
#[derive(Debug, Clone, Copy, PartialEq, Eq, Serialize)]
pub struct CompletionFenceTiming {
timing_mode: DeviceTimingMode,
device_execution: DeviceTimingMeasurement<DeviceExecutionTiming>,
blocking_wait_host_ns: DeviceTimingMeasurement<u64>,
}
impl CompletionFenceTiming {
fn new(
timing_mode: DeviceTimingMode,
device_execution: DeviceTimingMeasurement<DeviceExecutionTiming>,
blocking_wait_host_ns: DeviceTimingMeasurement<u64>,
) -> Self {
let device_execution = match (timing_mode, device_execution) {
(DeviceTimingMode::Off, _) => DeviceTimingMeasurement::NotRequested,
(
DeviceTimingMode::Completion
| DeviceTimingMode::Replay
| DeviceTimingMode::Kernel
| DeviceTimingMode::Verification,
DeviceTimingMeasurement::NotRequested,
) => DeviceTimingMeasurement::Unavailable(
DeviceTimingUnavailableReason::BackendUnsupported,
),
(
DeviceTimingMode::Completion
| DeviceTimingMode::Replay
| DeviceTimingMode::Kernel
| DeviceTimingMode::Verification,
measurement,
) => measurement,
};
Self {
timing_mode,
device_execution,
blocking_wait_host_ns,
}
}
pub const fn device_execution(self) -> DeviceTimingMeasurement<DeviceExecutionTiming> {
self.device_execution
}
pub const fn blocking_wait_host_ns(self) -> DeviceTimingMeasurement<u64> {
self.blocking_wait_host_ns
}
pub const fn timing_mode(self) -> DeviceTimingMode {
self.timing_mode
}
}
#[derive(Debug, Clone, Copy, PartialEq, Eq, Serialize)]
pub struct CompletionReadbackTiming {
host_elapsed_ns: u64,
calls: u32,
bytes: u64,
}
impl CompletionReadbackTiming {
const fn new(host_elapsed_ns: u64, calls: u32, bytes: u64) -> Self {
Self {
host_elapsed_ns,
calls,
bytes,
}
}
pub const fn host_elapsed_ns(self) -> u64 {
self.host_elapsed_ns
}
pub const fn calls(self) -> u32 {
self.calls
}
pub const fn bytes(self) -> u64 {
self.bytes
}
}
#[derive(Debug, Clone)]
#[must_use = "a terminal completion receipt releases one exact invocation"]
pub struct OperationCompletionReceipt {
submission: SubmittedOperationReceipt,
disposition: OperationCompletionDisposition,
fence_timing: CompletionFenceTiming,
submission_timing: DeviceTimingMeasurement<DeviceSubmissionExecutionTiming>,
participants: OnceLock<Vec<OperationParticipantCompletionReceipt>>,
fingerprint: String,
}
impl PartialEq for OperationCompletionReceipt {
fn eq(&self, other: &Self) -> bool {
self.submission == other.submission
&& self.disposition == other.disposition
&& self.fence_timing == other.fence_timing
&& self.submission_timing == other.submission_timing
&& self.fingerprint == other.fingerprint
}
}
impl Eq for OperationCompletionReceipt {}
impl Serialize for OperationCompletionReceipt {
fn serialize<S>(&self, serializer: S) -> Result<S::Ok, S::Error>
where
S: Serializer,
{
#[derive(Serialize)]
struct Wire<'a> {
submission: &'a SubmittedOperationReceipt,
disposition: &'a OperationCompletionDisposition,
fence_timing: CompletionFenceTiming,
submission_timing: &'a DeviceTimingMeasurement<DeviceSubmissionExecutionTiming>,
participants: &'a [OperationParticipantCompletionReceipt],
fingerprint: &'a str,
}
Wire {
submission: &self.submission,
disposition: &self.disposition,
fence_timing: self.fence_timing,
submission_timing: &self.submission_timing,
participants: self.participants(),
fingerprint: &self.fingerprint,
}
.serialize(serializer)
}
}
impl OperationCompletionReceipt {
fn new(
submission: SubmittedOperationReceipt,
disposition: OperationCompletionDisposition,
fence_timing: CompletionFenceTiming,
submission_timing: DeviceTimingMeasurement<DeviceSubmissionExecutionTiming>,
) -> Result<Self, VNextError> {
match &disposition {
OperationCompletionDisposition::Succeeded
| OperationCompletionDisposition::ContractFailedButQuiescent(_) => {}
OperationCompletionDisposition::FailedButQuiescent(failures) => {
if failures.len() != submission.participants().len()
|| failures
.iter()
.zip(submission.participants())
.any(|(failure, participant)| failure.identity() != participant.identity())
{
return Err(invalid_completion(
"batch completion failures differ from participant submission projections",
));
}
}
}
#[derive(Serialize)]
struct CompletionFingerprintInput<'a> {
domain: &'static str,
submission_fingerprint: &'a str,
disposition: &'a OperationCompletionDisposition,
}
let fingerprint = canonical_completion_fingerprint(&CompletionFingerprintInput {
domain: "ferrum.runtime-vnext.batch-operation-completion.v1",
submission_fingerprint: submission.fingerprint(),
disposition: &disposition,
});
Ok(Self {
submission,
disposition,
fence_timing,
submission_timing,
participants: OnceLock::new(),
fingerprint,
})
}
pub fn submission(&self) -> &SubmittedOperationReceipt {
&self.submission
}
pub fn disposition(&self) -> &OperationCompletionDisposition {
&self.disposition
}
pub const fn fence_timing(&self) -> CompletionFenceTiming {
self.fence_timing
}
pub const fn submission_timing(
&self,
) -> &DeviceTimingMeasurement<DeviceSubmissionExecutionTiming> {
&self.submission_timing
}
pub fn participants(&self) -> &[OperationParticipantCompletionReceipt] {
self.participants.get_or_init(|| {
self.submission
.participants()
.iter()
.enumerate()
.map(|(index, submission)| {
let disposition = match &self.disposition {
OperationCompletionDisposition::Succeeded => {
OperationParticipantCompletionDisposition::Succeeded
}
OperationCompletionDisposition::FailedButQuiescent(failures) => {
OperationParticipantCompletionDisposition::FailedButQuiescent(
failures[index].clone(),
)
}
OperationCompletionDisposition::ContractFailedButQuiescent(failure) => {
OperationParticipantCompletionDisposition::ContractFailedButQuiescent(
failure.clone(),
)
}
};
OperationParticipantCompletionReceipt {
submission: submission.clone(),
disposition,
batch_completion_fingerprint: self.fingerprint.clone(),
}
})
.collect()
})
}
pub fn fingerprint(&self) -> &str {
&self.fingerprint
}
}
#[derive(Debug, Clone, Copy, PartialEq, Eq, Serialize)]
#[serde(rename_all = "snake_case")]
pub enum CompletionRecoveryCause {
SubmissionIndeterminate,
FenceIndeterminate,
FenceObservationPanicked,
}
#[derive(Debug, Clone, PartialEq, Eq, Serialize)]
#[must_use = "a lane-drain receipt proves quiescence for one exact completion slot"]
pub struct CompletionDrainReceipt {
slot_id: CompletionSlotId,
batch_identity: BatchOperationIdentity,
submission: Option<SubmittedOperationReceipt>,
cause: CompletionRecoveryCause,
had_submission_fence: bool,
}
impl CompletionDrainReceipt {
pub const fn slot_id(&self) -> CompletionSlotId {
self.slot_id
}
pub fn batch_identity(&self) -> &BatchOperationIdentity {
&self.batch_identity
}
pub fn submission(&self) -> Option<&SubmittedOperationReceipt> {
self.submission.as_ref()
}
pub const fn cause(&self) -> CompletionRecoveryCause {
self.cause
}
pub const fn had_submission_fence(&self) -> bool {
self.had_submission_fence
}
}
#[derive(Debug, Clone)]
struct CompletionQuarantineFreshness(Arc<AtomicBool>);
impl CompletionQuarantineFreshness {
fn current() -> Self {
Self(Arc::new(AtomicBool::new(true)))
}
fn is_current(&self) -> bool {
self.0.load(Ordering::Acquire)
}
fn invalidate(&self) {
self.0.store(false, Ordering::Release);
}
}
#[derive(Debug, Clone, Serialize)]
#[must_use = "a quarantine receipt identifies retained device-visible ownership"]
pub struct CompletionQuarantineReceipt {
slot_id: CompletionSlotId,
batch_identity: BatchOperationIdentity,
submission: Option<SubmittedOperationReceipt>,
cause: CompletionRecoveryCause,
had_submission_fence: bool,
device_id: super::DeviceId,
runtime_implementation_fingerprint: String,
#[serde(skip)]
freshness: CompletionQuarantineFreshness,
}
impl PartialEq for CompletionQuarantineReceipt {
fn eq(&self, other: &Self) -> bool {
self.slot_id == other.slot_id
&& self.batch_identity == other.batch_identity
&& self.submission == other.submission
&& self.cause == other.cause
&& self.had_submission_fence == other.had_submission_fence
&& self.device_id == other.device_id
&& self.runtime_implementation_fingerprint == other.runtime_implementation_fingerprint
}
}
impl Eq for CompletionQuarantineReceipt {}
impl CompletionQuarantineReceipt {
pub const fn slot_id(&self) -> CompletionSlotId {
self.slot_id
}
pub fn batch_identity(&self) -> &BatchOperationIdentity {
&self.batch_identity
}
pub fn submission(&self) -> Option<&SubmittedOperationReceipt> {
self.submission.as_ref()
}
pub const fn cause(&self) -> CompletionRecoveryCause {
self.cause
}
pub const fn had_submission_fence(&self) -> bool {
self.had_submission_fence
}
pub fn device_id(&self) -> &super::DeviceId {
&self.device_id
}
pub fn runtime_implementation_fingerprint(&self) -> &str {
&self.runtime_implementation_fingerprint
}
pub fn is_current(&self) -> bool {
self.freshness.is_current()
}
}
#[derive(Debug)]
#[must_use = "completion recovery must be recorded as drained or quarantined"]
pub enum CompletionRecoveryOutcome {
Drained(CompletionDrainReceipt),
Quarantined(CompletionQuarantineReceipt),
}
#[derive(Debug)]
pub enum CompletionSweepObservation {
Observed(CompletionObservation),
Failed(VNextError),
}
#[derive(Debug)]
pub struct CompletionSweepEntry {
slot_id: CompletionSlotId,
observation: CompletionSweepObservation,
}
impl CompletionSweepEntry {
pub const fn slot_id(&self) -> CompletionSlotId {
self.slot_id
}
pub fn observation(&self) -> &CompletionSweepObservation {
&self.observation
}
pub fn into_observation(self) -> CompletionSweepObservation {
self.observation
}
}
#[derive(Debug)]
#[must_use = "a bounded completion sweep contains scheduler-owned progress evidence"]
pub struct CompletionSweepReceipt {
entries: Vec<CompletionSweepEntry>,
retained_after: usize,
quarantined_after: usize,
}
impl CompletionSweepReceipt {
pub fn entries(&self) -> &[CompletionSweepEntry] {
&self.entries
}
pub fn into_entries(self) -> Vec<CompletionSweepEntry> {
self.entries
}
pub const fn retained_after(&self) -> usize {
self.retained_after
}
pub const fn quarantined_after(&self) -> usize {
self.quarantined_after
}
}
enum CompletionResourceLease<R: DeviceRuntime> {
Invocation(InvocationResourceLease<R>),
Wave(PreparedStepSubmissionWave<R>),
}
impl<R: DeviceRuntime> CompletionResourceLease<R> {
fn runtime(&self) -> &Arc<R> {
match self {
Self::Invocation(invocation) => invocation.runtime(),
Self::Wave(wave) => wave.runtime(),
}
}
fn deferred_cleanup_domain(&self) -> DeferredDeviceCleanupDomainId {
match self {
Self::Invocation(invocation) => invocation.deferred_cleanup_domain(),
Self::Wave(wave) => wave.deferred_cleanup_domain(),
}
}
fn mark_submission_fence_installed(&mut self) -> Result<(), VNextError> {
match self {
Self::Invocation(invocation) => invocation.mark_submission_fence_installed(),
Self::Wave(wave) => wave.mark_submission_fence_installed(),
}
}
fn encode_backing_initializations(
&self,
runtime: &R,
commands: &mut DeviceCommandBatch<R::Command>,
) -> Result<usize, BackingInitializationEncodeError<R::Error>> {
match self {
Self::Invocation(invocation) => {
invocation.encode_backing_initializations(runtime, commands)
}
Self::Wave(wave) => wave.encode_backing_initializations(runtime, commands),
}
}
fn mark_submission_indeterminate(&mut self) {
match self {
Self::Invocation(invocation) => invocation.mark_submission_indeterminate(),
Self::Wave(wave) => wave.mark_submission_indeterminate(),
}
}
fn finish_backing_initializations(&mut self, succeeded: bool) -> Result<(), VNextError> {
match self {
Self::Invocation(invocation) => invocation.finish_backing_initializations(succeeded),
Self::Wave(wave) => wave.finish_backing_initializations(succeeded),
}
}
fn finish_request_state_hazards(
&mut self,
disposition: RequestStateHazardTerminalDisposition,
) -> Result<(), VNextError> {
match self {
Self::Invocation(invocation) => invocation.finish_request_state_hazards(disposition),
Self::Wave(wave) => wave.finish_request_state_hazards(disposition),
}
}
fn backing_view(
&self,
node_id: &NodeId,
participant_index: u32,
resource_id: &ResourceId,
) -> Result<LogicalBackingBufferView<'_, R::Buffer>, VNextError> {
let participant_index = usize::try_from(participant_index).map_err(|_| {
invalid_completion("completion readback participant index exceeds host address space")
})?;
match self {
Self::Invocation(invocation) => {
if invocation.node_id() != node_id {
return Err(invalid_completion(
"completion readback node differs from its invocation",
));
}
let is_shared = invocation
.backing_slices()
.iter()
.chain(invocation.step_resources().backing_slices())
.any(|authority| authority.resource_id() == resource_id);
if is_shared {
return invocation.backing_view(resource_id);
}
let participant = invocation
.participants()
.nth(participant_index)
.ok_or_else(|| {
invalid_completion(
"completion readback participant is absent from its invocation",
)
})?;
invocation.step_resources().participant_backing_view(
BatchParticipantAuthority::new(
participant.sequence_authority(),
participant.request_authority(),
),
resource_id,
)
}
Self::Wave(wave) => {
let node_index = wave
.nodes()
.iter()
.position(|node| node.node_id() == node_id)
.ok_or_else(|| {
invalid_completion("completion readback node is absent from its wave")
})?;
let is_shared = wave
.claimed_backing()
.backing_slices()
.iter()
.chain(wave.step_resources().backing_slices())
.any(|authority| authority.resource_id() == resource_id);
if is_shared {
return wave.backing_view(node_index, resource_id);
}
let participant = wave.nodes()[node_index]
.participants()
.nth(participant_index)
.ok_or_else(|| {
invalid_completion(
"completion readback participant is absent from its wave node",
)
})?;
wave.step_resources().participant_backing_view(
BatchParticipantAuthority::new(
participant.sequence_authority(),
participant.request_authority(),
),
resource_id,
)
}
}
}
fn readback_range(
&self,
node_id: &NodeId,
participant_index: u32,
resource_id: &ResourceId,
semantic_range: Range<u64>,
) -> Result<Range<u64>, VNextError> {
let participant_index = usize::try_from(participant_index).map_err(|_| {
invalid_completion("completion readback participant index exceeds host address space")
})?;
let (step, work_shape) = match self {
Self::Invocation(invocation) => {
if invocation.node_id() != node_id {
return Err(invalid_completion(
"completion readback node differs from its invocation",
));
}
(invocation.step_resources(), invocation.work_shape())
}
Self::Wave(wave) => {
let node = wave
.nodes()
.iter()
.find(|node| node.node_id() == node_id)
.ok_or_else(|| {
invalid_completion("completion readback node is absent from its wave")
})?;
(wave.step_resources(), node.work_shape())
}
};
let descriptor = step.dynamic_descriptor(resource_id)?;
if descriptor.lifetime() == AllocationLifetime::Step
&& descriptor.kind() == &AllocationKind::Value
{
translate_step_participant_readback_range(
descriptor.demand(),
work_shape,
participant_index,
semantic_range,
)
} else {
Ok(semantic_range)
}
}
}
enum CompletionRecord<R: DeviceRuntime> {
Reserved,
InFlight {
resources: CompletionResourceLease<R>,
lane: Arc<ExecutionLane<R>>,
fence: R::Fence,
batch_identity: BatchOperationIdentity,
receipt: SubmittedOperationReceipt,
timing_mode: DeviceTimingMode,
recovery_state: CompletionRecoveryState,
},
SubmissionIndeterminate {
resources: CompletionResourceLease<R>,
lane: Arc<ExecutionLane<R>>,
batch_identity: BatchOperationIdentity,
},
Quarantined {
ownership: CompletionQuarantineOwnership<R>,
receipt: CompletionQuarantineReceipt,
},
Reaped,
}
enum BoundCompletionObservation<T> {
Pending,
Terminal {
completion: OperationCompletionReceipt,
terminal: T,
},
Indeterminate(Vec<IdentifiedFailure>),
SubmissionIndeterminate,
ObservationPanicked,
Quarantined(CompletionQuarantineReceipt),
}
#[derive(Debug, Clone, Copy, PartialEq, Eq)]
enum CompletionRecoveryState {
Unobserved,
QueryIndeterminate,
DrainEligible(CompletionRecoveryCause),
}
enum CompletionQuarantineOwnership<R: DeviceRuntime> {
InFlight {
resources: CompletionResourceLease<R>,
lane: Arc<ExecutionLane<R>>,
fence: R::Fence,
batch_identity: BatchOperationIdentity,
submission: SubmittedOperationReceipt,
},
SubmissionIndeterminate {
resources: CompletionResourceLease<R>,
lane: Arc<ExecutionLane<R>>,
batch_identity: BatchOperationIdentity,
},
}
impl<R: DeviceRuntime> CompletionQuarantineOwnership<R> {
fn lane(&self) -> &Arc<ExecutionLane<R>> {
match self {
Self::InFlight { lane, .. } | Self::SubmissionIndeterminate { lane, .. } => lane,
}
}
fn deferred_cleanup_domain(&self) -> DeferredDeviceCleanupDomainId {
match self {
Self::InFlight { resources, .. } | Self::SubmissionIndeterminate { resources, .. } => {
resources.deferred_cleanup_domain()
}
}
}
fn resources_mut(&mut self) -> &mut CompletionResourceLease<R> {
match self {
Self::InFlight { resources, .. } | Self::SubmissionIndeterminate { resources, .. } => {
resources
}
}
}
}
impl<R: DeviceRuntime> CompletionRecord<R> {
fn deferred_cleanup_domain(&self) -> Option<DeferredDeviceCleanupDomainId> {
match self {
Self::InFlight { resources, .. } | Self::SubmissionIndeterminate { resources, .. } => {
Some(resources.deferred_cleanup_domain())
}
Self::Quarantined { ownership, .. } => Some(ownership.deferred_cleanup_domain()),
Self::Reserved | Self::Reaped => None,
}
}
fn finish_request_state_hazards_after_drain(&mut self) -> Result<(), VNextError> {
let resources = match self {
Self::InFlight { resources, .. } | Self::SubmissionIndeterminate { resources, .. } => {
resources
}
Self::Quarantined { ownership, .. } => ownership.resources_mut(),
Self::Reserved | Self::Reaped => return Ok(()),
};
resources.finish_request_state_hazards(
RequestStateHazardTerminalDisposition::IndeterminateAfterDrain,
)
}
}
type SharedCompletionRecord<R> = Arc<Mutex<CompletionRecord<R>>>;
struct CompletionReaperState<R: DeviceRuntime> {
next_slot: Option<NonZeroU64>,
sweep_cursor: Option<CompletionSlotId>,
slots: BTreeMap<CompletionSlotId, SharedCompletionRecord<R>>,
}
#[must_use = "the scheduler must retain its completion reaper"]
pub struct CompletionReaper<R: DeviceRuntime> {
state: Mutex<CompletionReaperState<R>>,
}
pub const MAX_COMPLETION_SWEEP_SLOTS: usize = 64;
impl<R: DeviceRuntime> fmt::Debug for CompletionReaper<R> {
fn fmt(&self, formatter: &mut fmt::Formatter<'_>) -> fmt::Result {
formatter
.debug_struct("CompletionReaper")
.field("retained", &self.retained_count())
.finish_non_exhaustive()
}
}
impl<R: DeviceRuntime> CompletionReaper<R> {
pub fn new() -> Arc<Self> {
Arc::new(Self {
state: Mutex::new(CompletionReaperState {
next_slot: NonZeroU64::new(1),
sweep_cursor: None,
slots: BTreeMap::new(),
}),
})
}
pub fn retained_count(&self) -> usize {
self.state
.lock()
.map(|state| state.slots.len())
.unwrap_or(usize::MAX)
}
pub fn quarantined_count(&self) -> usize {
let records = match self.state.lock() {
Ok(state) => state.slots.values().cloned().collect::<Vec<_>>(),
Err(_) => return usize::MAX,
};
records
.iter()
.filter(|record| {
record
.lock()
.map(|record| matches!(&*record, CompletionRecord::Quarantined { .. }))
.unwrap_or(true)
})
.count()
}
pub fn poll_bounded(&self, maximum_slots: usize) -> Result<CompletionSweepReceipt, VNextError> {
if maximum_slots == 0 || maximum_slots > MAX_COMPLETION_SWEEP_SLOTS {
return Err(invalid_completion(format!(
"completion sweep size must be in 1..={MAX_COMPLETION_SWEEP_SLOTS}"
)));
}
let slot_ids = {
let mut state = self
.state
.lock()
.map_err(|_| invalid_completion("completion reaper state mutex is poisoned"))?;
let keys = state.slots.keys().copied().collect::<Vec<_>>();
let start = state
.sweep_cursor
.and_then(|cursor| keys.iter().position(|slot_id| *slot_id > cursor))
.unwrap_or(0);
let slot_ids = keys
.iter()
.cycle()
.skip(start)
.take(maximum_slots.min(keys.len()))
.copied()
.collect::<Vec<_>>();
if let Some(last) = slot_ids.last() {
state.sweep_cursor = Some(*last);
}
slot_ids
};
let entries = slot_ids
.into_iter()
.map(|slot_id| CompletionSweepEntry {
slot_id,
observation: match self.poll_bound(slot_id) {
Ok(observation) => CompletionSweepObservation::Observed(observation),
Err(error) => CompletionSweepObservation::Failed(error),
},
})
.collect();
let records = self
.state
.lock()
.map_err(|_| invalid_completion("completion reaper state mutex is poisoned"))?
.slots
.values()
.cloned()
.collect::<Vec<_>>();
let retained_after = records.len();
let quarantined_after = records
.iter()
.filter(|record| {
record
.lock()
.map(|record| matches!(&*record, CompletionRecord::Quarantined { .. }))
.unwrap_or(true)
})
.count();
Ok(CompletionSweepReceipt {
entries,
retained_after,
quarantined_after,
})
}
pub fn recover_slot_by_draining_lane(
&self,
slot_id: CompletionSlotId,
) -> Result<CompletionRecoveryOutcome, VNextError> {
self.recover_bound(slot_id)
}
pub fn wait_slot_for_recovery(
&self,
slot_id: CompletionSlotId,
) -> Result<CompletionObservation, VNextError> {
self.wait_bound(slot_id)
}
pub(crate) fn reserve(
reaper: &Arc<Self>,
invocation: InvocationResourceLease<R>,
lane: Arc<ExecutionLane<R>>,
batch_identity: BatchOperationIdentity,
) -> Result<CompletionReservation<R>, VNextError> {
Self::reserve_resources(
reaper,
CompletionResourceLease::Invocation(invocation),
lane,
batch_identity,
)
}
pub(crate) fn reserve_wave(
reaper: &Arc<Self>,
wave: PreparedStepSubmissionWave<R>,
lane: Arc<ExecutionLane<R>>,
batch_identity: BatchOperationIdentity,
) -> Result<CompletionReservation<R>, VNextError> {
Self::reserve_resources(
reaper,
CompletionResourceLease::Wave(wave),
lane,
batch_identity,
)
}
fn reserve_resources(
reaper: &Arc<Self>,
resources: CompletionResourceLease<R>,
lane: Arc<ExecutionLane<R>>,
batch_identity: BatchOperationIdentity,
) -> Result<CompletionReservation<R>, VNextError> {
if !Arc::ptr_eq(resources.runtime(), lane.runtime_arc()) {
return Err(invalid_completion(
"completion lane runtime is not the submission resource runtime instance",
));
}
let mut state = reaper
.state
.lock()
.map_err(|_| invalid_completion("completion reaper state mutex is poisoned"))?;
let raw = state
.next_slot
.ok_or_else(|| invalid_completion("completion slot identity space is exhausted"))?;
state.next_slot = raw.get().checked_add(1).and_then(NonZeroU64::new);
let slot_id = CompletionSlotId(raw);
let record = Arc::new(Mutex::new(CompletionRecord::Reserved));
if state.slots.insert(slot_id, Arc::clone(&record)).is_some() {
return Err(invalid_completion("completion slot identity was reused"));
}
drop(state);
#[derive(Serialize)]
struct SubmissionFingerprintInput<'a> {
domain: &'static str,
slot_id: CompletionSlotId,
batch_identity_fingerprint: &'a str,
}
let fingerprint = canonical_completion_fingerprint(&SubmissionFingerprintInput {
domain: "ferrum.runtime-vnext.batch-operation-submission.v1",
slot_id,
batch_identity_fingerprint: batch_identity.fingerprint(),
});
let receipt = SubmittedOperationReceipt::new(slot_id, batch_identity.clone(), fingerprint);
let record_receipt = receipt.clone();
Ok(CompletionReservation {
reaper: Arc::clone(reaper),
record,
slot_id,
resources: Some(resources),
lane: Some(lane),
batch_identity: Some(batch_identity),
receipt: Some(receipt),
record_receipt: Some(record_receipt),
submission_may_have_happened: false,
finished: false,
})
}
fn lookup(&self, slot_id: CompletionSlotId) -> Result<SharedCompletionRecord<R>, VNextError> {
self.state
.lock()
.map_err(|_| invalid_completion("completion reaper state mutex is poisoned"))?
.slots
.get(&slot_id)
.cloned()
.ok_or_else(|| invalid_completion("completion slot is unknown or already reaped"))
}
fn remove_exact(&self, slot_id: CompletionSlotId, record: &SharedCompletionRecord<R>) {
let mut state = match self.state.lock() {
Ok(state) => state,
Err(poisoned) => poisoned.into_inner(),
};
if state
.slots
.get(&slot_id)
.is_some_and(|current| Arc::ptr_eq(current, record))
{
state.slots.remove(&slot_id);
}
}
pub(crate) fn poll_bound(
&self,
slot_id: CompletionSlotId,
) -> Result<CompletionObservation, VNextError> {
Self::map_plain_observation(self.observe_bound_with(slot_id, false, |_, _, _, _, _| ())?)
}
pub(crate) fn wait_bound(
&self,
slot_id: CompletionSlotId,
) -> Result<CompletionObservation, VNextError> {
Self::map_plain_observation(self.observe_bound_with(slot_id, true, |_, _, _, _, _| ())?)
}
pub(crate) fn wait_bound_with_readback(
&self,
slot_id: CompletionSlotId,
request: CompletionReadbackRequest,
) -> Result<CompletionReadbackObservation, VNextError> {
let observation = self.observe_bound_with(
slot_id,
true,
|resources, lane, batch_identity, disposition, timing_mode| {
attempt_completion_readback(
resources,
lane,
batch_identity,
disposition,
timing_mode,
request,
)
},
)?;
Ok(match observation {
BoundCompletionObservation::Pending => CompletionReadbackObservation::Pending,
BoundCompletionObservation::Terminal {
completion,
terminal,
} => CompletionReadbackObservation::Terminal(CompletionReadbackReceipt::new(
completion, terminal,
)),
BoundCompletionObservation::Indeterminate(failures) => {
CompletionReadbackObservation::Indeterminate(failures)
}
BoundCompletionObservation::SubmissionIndeterminate => {
CompletionReadbackObservation::SubmissionIndeterminate
}
BoundCompletionObservation::ObservationPanicked => {
CompletionReadbackObservation::ObservationPanicked
}
BoundCompletionObservation::Quarantined(receipt) => {
CompletionReadbackObservation::Quarantined(receipt)
}
})
}
pub(crate) fn wait_bound_with_readbacks(
&self,
slot_id: CompletionSlotId,
request: CompletionReadbackBatchRequest,
) -> Result<CompletionReadbackBatchObservation, VNextError> {
self.validate_bound_readback_batch(slot_id, &request)?;
let observation = self.observe_bound_with(
slot_id,
true,
|resources, lane, batch_identity, disposition, timing_mode| {
attempt_completion_readbacks(
resources,
lane,
batch_identity,
disposition,
timing_mode,
request,
)
},
)?;
Ok(match observation {
BoundCompletionObservation::Pending => CompletionReadbackBatchObservation::Pending,
BoundCompletionObservation::Terminal {
completion,
terminal,
} => CompletionReadbackBatchObservation::Terminal(CompletionReadbackBatchReceipt::new(
completion, terminal,
)),
BoundCompletionObservation::Indeterminate(failures) => {
CompletionReadbackBatchObservation::Indeterminate(failures)
}
BoundCompletionObservation::SubmissionIndeterminate => {
CompletionReadbackBatchObservation::SubmissionIndeterminate
}
BoundCompletionObservation::ObservationPanicked => {
CompletionReadbackBatchObservation::ObservationPanicked
}
BoundCompletionObservation::Quarantined(receipt) => {
CompletionReadbackBatchObservation::Quarantined(receipt)
}
})
}
pub(crate) fn wait_bound_with_readback_collection(
&self,
slot_id: CompletionSlotId,
request: CompletionReadbackCollectionRequest,
) -> Result<CompletionReadbackCollectionObservation, VNextError> {
self.validate_bound_readback_collection(slot_id, &request)?;
let observation = self.observe_bound_with(
slot_id,
true,
|resources, lane, batch_identity, disposition, timing_mode| {
attempt_completion_readback_collection(
resources,
lane,
batch_identity,
disposition,
timing_mode,
request,
)
},
)?;
Ok(match observation {
BoundCompletionObservation::Pending => CompletionReadbackBatchObservation::Pending,
BoundCompletionObservation::Terminal {
completion,
terminal,
} => CompletionReadbackBatchObservation::Terminal(CompletionReadbackBatchReceipt::new(
completion, terminal,
)),
BoundCompletionObservation::Indeterminate(failures) => {
CompletionReadbackBatchObservation::Indeterminate(failures)
}
BoundCompletionObservation::SubmissionIndeterminate => {
CompletionReadbackBatchObservation::SubmissionIndeterminate
}
BoundCompletionObservation::ObservationPanicked => {
CompletionReadbackBatchObservation::ObservationPanicked
}
BoundCompletionObservation::Quarantined(receipt) => {
CompletionReadbackBatchObservation::Quarantined(receipt)
}
})
}
fn validate_bound_readback_batch(
&self,
slot_id: CompletionSlotId,
request: &CompletionReadbackBatchRequest,
) -> Result<(), VNextError> {
let record = self.lookup(slot_id)?;
let guard = record
.lock()
.map_err(|_| invalid_completion("completion slot mutex is poisoned"))?;
match &*guard {
CompletionRecord::InFlight { batch_identity, .. }
| CompletionRecord::SubmissionIndeterminate { batch_identity, .. } => {
request.validate_for(batch_identity)
}
CompletionRecord::Reserved => Err(invalid_completion(
"completion slot has not reached submission",
)),
CompletionRecord::Quarantined { .. } => Ok(()),
CompletionRecord::Reaped => {
Err(invalid_completion("completion slot is already reaped"))
}
}
}
fn validate_bound_readback_collection(
&self,
slot_id: CompletionSlotId,
request: &CompletionReadbackCollectionRequest,
) -> Result<(), VNextError> {
let record = self.lookup(slot_id)?;
let guard = record
.lock()
.map_err(|_| invalid_completion("completion slot mutex is poisoned"))?;
match &*guard {
CompletionRecord::InFlight { batch_identity, .. }
| CompletionRecord::SubmissionIndeterminate { batch_identity, .. } => {
request.validate_for(batch_identity)
}
CompletionRecord::Reserved => Err(invalid_completion(
"completion slot has not reached submission",
)),
CompletionRecord::Quarantined { .. } => Ok(()),
CompletionRecord::Reaped => {
Err(invalid_completion("completion slot is already reaped"))
}
}
}
fn map_plain_observation(
observation: BoundCompletionObservation<()>,
) -> Result<CompletionObservation, VNextError> {
Ok(match observation {
BoundCompletionObservation::Pending => CompletionObservation::Pending,
BoundCompletionObservation::Terminal { completion, .. } => {
CompletionObservation::Terminal(completion)
}
BoundCompletionObservation::Indeterminate(failures) => {
CompletionObservation::Indeterminate(failures)
}
BoundCompletionObservation::SubmissionIndeterminate => {
CompletionObservation::SubmissionIndeterminate
}
BoundCompletionObservation::ObservationPanicked => {
CompletionObservation::ObservationPanicked
}
BoundCompletionObservation::Quarantined(receipt) => {
CompletionObservation::Quarantined(receipt)
}
})
}
fn observe_bound_with<T, F>(
&self,
slot_id: CompletionSlotId,
blocking: bool,
terminal_action: F,
) -> Result<BoundCompletionObservation<T>, VNextError>
where
F: FnOnce(
&CompletionResourceLease<R>,
&Arc<ExecutionLane<R>>,
&BatchOperationIdentity,
&OperationCompletionDisposition,
DeviceTimingMode,
) -> T,
{
let record = self.lookup(slot_id)?;
let mut guard = record
.lock()
.map_err(|_| invalid_completion("completion slot mutex is poisoned"))?;
let observation = match &*guard {
CompletionRecord::Reserved => {
return Err(invalid_completion(
"completion slot has not reached submission",
));
}
CompletionRecord::SubmissionIndeterminate { .. } => {
FenceObservation::SubmissionIndeterminate
}
CompletionRecord::Quarantined { receipt, .. } => {
FenceObservation::Quarantined(receipt.clone())
}
CompletionRecord::Reaped => {
return Err(invalid_completion("completion slot is already reaped"));
}
CompletionRecord::InFlight {
lane,
fence,
batch_identity,
timing_mode,
..
} if blocking => {
let wait_started = timing_mode.completion_enabled().then(Instant::now);
match catch_unwind(AssertUnwindSafe(|| lane.wait_fence(fence))) {
Ok(Ok(terminal)) => {
let wait_timing = match wait_started {
Some(started) => u64::try_from(started.elapsed().as_nanos()).map_or(
DeviceTimingMeasurement::Unavailable(
DeviceTimingUnavailableReason::DurationOverflow,
),
DeviceTimingMeasurement::Measured,
),
None => DeviceTimingMeasurement::NotRequested,
};
catch_unwind(AssertUnwindSafe(|| {
terminal_observation(
lane,
batch_identity,
terminal,
*timing_mode,
wait_timing,
)
}))
.unwrap_or(FenceObservation::ObservationPanicked)
}
Ok(Err(indeterminate)) => catch_unwind(AssertUnwindSafe(|| {
classify_batch_device_error(
lane.runtime(),
batch_identity,
indeterminate.error(),
)
}))
.map_or(
FenceObservation::ObservationPanicked,
|classified| match classified {
Ok(failure) => FenceObservation::Indeterminate(failure),
Err(error) => FenceObservation::ContractIndeterminate(error),
},
),
Err(_) => FenceObservation::ObservationPanicked,
}
}
CompletionRecord::InFlight {
lane,
fence,
batch_identity,
timing_mode,
..
} => match catch_unwind(AssertUnwindSafe(|| lane.query_fence(fence))) {
Ok(FenceQuery::Pending) => FenceObservation::Pending,
Ok(FenceQuery::Terminal(terminal)) => catch_unwind(AssertUnwindSafe(|| {
terminal_observation(
lane,
batch_identity,
terminal,
*timing_mode,
if timing_mode.completion_enabled() {
DeviceTimingMeasurement::Measured(0)
} else {
DeviceTimingMeasurement::NotRequested
},
)
}))
.unwrap_or(FenceObservation::ObservationPanicked),
Ok(FenceQuery::Indeterminate(error)) => catch_unwind(AssertUnwindSafe(|| {
classify_batch_device_error(lane.runtime(), batch_identity, &error)
}))
.map_or(
FenceObservation::ObservationPanicked,
|classified| match classified {
Ok(failure) => FenceObservation::Indeterminate(failure),
Err(error) => FenceObservation::ContractIndeterminate(error),
},
),
Err(_) => FenceObservation::ObservationPanicked,
},
};
let recovery_state = match &observation {
FenceObservation::Indeterminate(_) | FenceObservation::ContractIndeterminate(_) => {
Some(if blocking {
CompletionRecoveryState::DrainEligible(
CompletionRecoveryCause::FenceIndeterminate,
)
} else {
CompletionRecoveryState::QueryIndeterminate
})
}
FenceObservation::ObservationPanicked => Some(if blocking {
CompletionRecoveryState::DrainEligible(
CompletionRecoveryCause::FenceObservationPanicked,
)
} else {
CompletionRecoveryState::QueryIndeterminate
}),
_ => None,
};
if let Some(recovery_state) = recovery_state {
if let CompletionRecord::InFlight {
recovery_state: current,
..
} = &mut *guard
{
if !matches!(current, CompletionRecoveryState::DrainEligible(_)) {
*current = recovery_state;
}
}
}
if let FenceObservation::Terminal(mut disposition, fence_timing, submission_timing) =
observation
{
let old = std::mem::replace(&mut *guard, CompletionRecord::Reaped);
let CompletionRecord::InFlight {
resources,
lane,
batch_identity,
receipt,
timing_mode,
..
} = old
else {
unreachable!("terminal observation came from an in-flight record")
};
let terminal = terminal_action(
&resources,
&lane,
&batch_identity,
&disposition,
timing_mode,
);
if let Err(failure) = lane.finish_one_terminal() {
disposition = OperationCompletionDisposition::ContractFailedButQuiescent(failure);
}
let initialization_succeeded =
matches!(disposition, OperationCompletionDisposition::Succeeded);
let mut resources = resources;
if let Err(error) = resources.finish_backing_initializations(initialization_succeeded) {
disposition = OperationCompletionDisposition::ContractFailedButQuiescent(
QuiescentCompletionContractFailure::new(format!(
"backing initialization terminal transition failed: {error}"
)),
);
}
let hazard_disposition =
if matches!(disposition, OperationCompletionDisposition::Succeeded) {
RequestStateHazardTerminalDisposition::Succeeded
} else {
RequestStateHazardTerminalDisposition::FailedButQuiescent
};
if let Err(error) = resources.finish_request_state_hazards(hazard_disposition) {
disposition = OperationCompletionDisposition::ContractFailedButQuiescent(
QuiescentCompletionContractFailure::new(format!(
"request-state hazard terminal transition failed: {error}"
)),
);
}
drop(guard);
self.remove_exact(slot_id, &record);
return OperationCompletionReceipt::new(
receipt,
disposition,
fence_timing,
submission_timing,
)
.map(|completion| BoundCompletionObservation::Terminal {
completion,
terminal,
});
}
Ok(match observation {
FenceObservation::Pending => BoundCompletionObservation::Pending,
FenceObservation::Indeterminate(failure) => {
BoundCompletionObservation::Indeterminate(failure)
}
FenceObservation::SubmissionIndeterminate => {
BoundCompletionObservation::SubmissionIndeterminate
}
FenceObservation::ObservationPanicked => {
BoundCompletionObservation::ObservationPanicked
}
FenceObservation::ContractIndeterminate(error) => return Err(error),
FenceObservation::Quarantined(receipt) => {
BoundCompletionObservation::Quarantined(receipt)
}
FenceObservation::Terminal(_, _, _) => {
unreachable!("terminal observation returned through the reaping branch")
}
})
}
fn recover_bound(
&self,
slot_id: CompletionSlotId,
) -> Result<CompletionRecoveryOutcome, VNextError> {
let record = self.lookup(slot_id)?;
let mut guard = record
.lock()
.map_err(|_| invalid_completion("completion slot mutex is poisoned"))?;
let (lane, batch_identity, submission, cause, had_submission_fence) = match &*guard {
CompletionRecord::InFlight {
lane,
batch_identity,
receipt,
recovery_state: CompletionRecoveryState::DrainEligible(cause),
..
} => (
Arc::clone(lane),
batch_identity.clone(),
Some(receipt.clone()),
*cause,
true,
),
CompletionRecord::SubmissionIndeterminate {
lane,
batch_identity,
..
} => (
Arc::clone(lane),
batch_identity.clone(),
None,
CompletionRecoveryCause::SubmissionIndeterminate,
false,
),
CompletionRecord::Quarantined { ownership, receipt } => (
Arc::clone(ownership.lane()),
receipt.batch_identity.clone(),
receipt.submission.clone(),
receipt.cause,
receipt.had_submission_fence,
),
CompletionRecord::InFlight { .. } => {
return Err(invalid_completion(
"completion slot has no blocking indeterminate fence observation",
));
}
CompletionRecord::Reserved => {
return Err(invalid_completion(
"completion slot has not reached submission",
));
}
CompletionRecord::Reaped => {
return Err(invalid_completion("completion slot is already reaped"));
}
};
lane.fail_closed();
if !lane.drain(had_submission_fence) {
if let CompletionRecord::Quarantined { receipt, .. } = &*guard {
return Ok(CompletionRecoveryOutcome::Quarantined(receipt.clone()));
}
let old = std::mem::replace(&mut *guard, CompletionRecord::Reaped);
let ownership = match old {
CompletionRecord::InFlight {
resources,
lane,
fence,
batch_identity,
receipt,
..
} => CompletionQuarantineOwnership::InFlight {
resources,
lane,
fence,
batch_identity,
submission: receipt,
},
CompletionRecord::SubmissionIndeterminate {
resources,
lane,
batch_identity,
} => CompletionQuarantineOwnership::SubmissionIndeterminate {
resources,
lane,
batch_identity,
},
_ => unreachable!("recovery source was validated before lane drain"),
};
let receipt = CompletionQuarantineReceipt {
slot_id,
batch_identity,
submission,
cause,
had_submission_fence,
device_id: lane.descriptor.id.clone(),
runtime_implementation_fingerprint: lane
.descriptor
.runtime_implementation_fingerprint
.clone(),
freshness: CompletionQuarantineFreshness::current(),
};
*guard = CompletionRecord::Quarantined {
ownership,
receipt: receipt.clone(),
};
return Ok(CompletionRecoveryOutcome::Quarantined(receipt));
}
let mut old = std::mem::replace(&mut *guard, CompletionRecord::Reaped);
if let CompletionRecord::Quarantined { receipt, .. } = &old {
receipt.freshness.invalidate();
}
let hazard_result = old.finish_request_state_hazards_after_drain();
drop(guard);
self.remove_exact(slot_id, &record);
drop(old);
hazard_result?;
Ok(CompletionRecoveryOutcome::Drained(CompletionDrainReceipt {
slot_id,
batch_identity,
submission,
cause,
had_submission_fence,
}))
}
}
enum FenceObservation {
Pending,
Terminal(
OperationCompletionDisposition,
CompletionFenceTiming,
DeviceTimingMeasurement<DeviceSubmissionExecutionTiming>,
),
Indeterminate(Vec<IdentifiedFailure>),
ContractIndeterminate(VNextError),
SubmissionIndeterminate,
ObservationPanicked,
Quarantined(CompletionQuarantineReceipt),
}
fn terminal_observation<R: DeviceRuntime>(
lane: &ExecutionLane<R>,
batch_identity: &BatchOperationIdentity,
terminal: DeviceTerminalReceipt<R::Error>,
timing_mode: DeviceTimingMode,
blocking_wait_host_ns: DeviceTimingMeasurement<u64>,
) -> FenceObservation {
let (terminal, device_execution, submission_timing) = terminal.into_parts();
let timing = CompletionFenceTiming::new(timing_mode, device_execution, blocking_wait_host_ns);
let submission_timing = match (timing_mode, submission_timing) {
(DeviceTimingMode::Off | DeviceTimingMode::Completion, _) => {
DeviceTimingMeasurement::NotRequested
}
(
DeviceTimingMode::Replay | DeviceTimingMode::Kernel | DeviceTimingMode::Verification,
DeviceTimingMeasurement::NotRequested,
) => {
DeviceTimingMeasurement::Unavailable(DeviceTimingUnavailableReason::BackendUnsupported)
}
(
DeviceTimingMode::Replay | DeviceTimingMode::Kernel | DeviceTimingMode::Verification,
measurement,
) => measurement,
};
let descriptor_is_stable = catch_unwind(AssertUnwindSafe(|| {
lane.current_descriptor_matches_snapshot()
}))
.unwrap_or(false);
if !descriptor_is_stable {
return FenceObservation::Terminal(
OperationCompletionDisposition::ContractFailedButQuiescent(
QuiescentCompletionContractFailure::new(
"completion runtime descriptor differs from its execution lane snapshot",
),
),
timing,
submission_timing,
);
}
FenceObservation::Terminal(
match terminal {
DeviceTerminal::Succeeded => OperationCompletionDisposition::Succeeded,
DeviceTerminal::FailedButQuiescent(error) => {
match classify_batch_device_error(lane.runtime(), batch_identity, &error) {
Ok(failures) => OperationCompletionDisposition::FailedButQuiescent(failures),
Err(error) => OperationCompletionDisposition::ContractFailedButQuiescent(
QuiescentCompletionContractFailure::new(error.to_string()),
),
}
}
},
timing,
submission_timing,
)
}
fn classify_batch_device_error<R: DeviceRuntime>(
runtime: &R,
batch_identity: &BatchOperationIdentity,
error: &R::Error,
) -> Result<Vec<IdentifiedFailure>, VNextError> {
batch_identity
.participants()
.iter()
.map(|participant| classify_device_error(runtime, participant.identity().clone(), error))
.collect()
}
struct DeferredCompletionCleanup<R: DeviceRuntime> {
records: BTreeMap<CompletionSlotId, SharedCompletionRecord<R>>,
}
impl<R: DeviceRuntime> DeferredCompletionCleanup<R> {
fn new(records: BTreeMap<CompletionSlotId, SharedCompletionRecord<R>>) -> Self {
Self { records }
}
}
impl<R: DeviceRuntime> DeferredDeviceCleanupTask for DeferredCompletionCleanup<R> {
fn try_cleanup(&mut self) -> DeferredDeviceCleanupDisposition {
let slot_ids = self.records.keys().copied().collect::<Vec<_>>();
let mut retryable = false;
let mut quarantined = false;
for slot_id in slot_ids {
let Some(record) = self.records.get(&slot_id).cloned() else {
continue;
};
let quiescent = catch_unwind(AssertUnwindSafe(|| {
cleanup_dropped_completion_record(&record)
}))
.unwrap_or(false);
if quiescent {
if self
.records
.get(&slot_id)
.is_some_and(|current| Arc::ptr_eq(current, &record))
{
self.records.remove(&slot_id);
}
continue;
}
retryable = true;
quarantined |= record
.lock()
.map(|record| matches!(&*record, CompletionRecord::Quarantined { .. }))
.unwrap_or(true);
}
if self.records.is_empty() {
DeferredDeviceCleanupDisposition::Completed
} else if quarantined {
DeferredDeviceCleanupDisposition::Quarantined
} else {
debug_assert!(retryable);
DeferredDeviceCleanupDisposition::Retryable
}
}
}
fn fail_close_completion_record<R: DeviceRuntime>(record: &SharedCompletionRecord<R>) {
let guard = match record.lock() {
Ok(guard) => guard,
Err(poisoned) => poisoned.into_inner(),
};
match &*guard {
CompletionRecord::InFlight { lane, .. }
| CompletionRecord::SubmissionIndeterminate { lane, .. } => lane.fail_closed(),
CompletionRecord::Quarantined { ownership, .. } => ownership.lane().fail_closed(),
CompletionRecord::Reserved | CompletionRecord::Reaped => {}
}
}
fn cleanup_dropped_completion_record<R: DeviceRuntime>(record: &SharedCompletionRecord<R>) -> bool {
let mut guard = match record.lock() {
Ok(guard) => guard,
Err(poisoned) => poisoned.into_inner(),
};
let (quiescent, drained) = match &*guard {
CompletionRecord::InFlight { lane, fence, .. } => {
let queried = catch_unwind(AssertUnwindSafe(|| lane.query_fence(fence)))
.is_ok_and(|query| matches!(query, FenceQuery::Terminal(_)));
let waited = queried
|| catch_unwind(AssertUnwindSafe(|| lane.wait_fence(fence)))
.is_ok_and(|result| result.is_ok());
if waited {
(true, false)
} else {
let drained = lane.drain(true);
(drained, drained)
}
}
CompletionRecord::SubmissionIndeterminate { lane, .. } => {
let drained = lane.drain(false);
(drained, drained)
}
CompletionRecord::Quarantined { ownership, receipt } => {
let drained = ownership.lane().drain(receipt.had_submission_fence);
(drained, drained)
}
CompletionRecord::Reserved | CompletionRecord::Reaped => (true, false),
};
if !quiescent {
match &*guard {
CompletionRecord::InFlight { lane, .. }
| CompletionRecord::SubmissionIndeterminate { lane, .. } => lane.fail_closed(),
CompletionRecord::Quarantined { ownership, .. } => ownership.lane().fail_closed(),
CompletionRecord::Reserved | CompletionRecord::Reaped => {}
}
return false;
}
let old = std::mem::replace(&mut *guard, CompletionRecord::Reaped);
if let CompletionRecord::Quarantined { receipt, .. } = &old {
receipt.freshness.invalidate();
}
if !drained {
if let CompletionRecord::InFlight { lane, .. } = &old {
let _ = lane.finish_one_terminal();
}
}
drop(guard);
drop(old);
true
}
impl<R: DeviceRuntime> Drop for CompletionReaper<R> {
fn drop(&mut self) {
let records = match self.state.get_mut() {
Ok(state) => std::mem::take(&mut state.slots),
Err(poisoned) => std::mem::take(&mut poisoned.into_inner().slots),
};
if records.is_empty() {
return;
}
for record in records.values() {
fail_close_completion_record(record);
}
for (slot_id, record) in records {
let domain = record
.lock()
.unwrap_or_else(std::sync::PoisonError::into_inner)
.deferred_cleanup_domain();
if let Some(domain) = domain {
defer_device_cleanup(
domain,
DeferredCompletionCleanup::new(BTreeMap::from([(slot_id, record)])),
);
}
}
}
}
#[derive(Debug, Clone, PartialEq, Eq, Serialize)]
#[must_use = "a completion readback request identifies one exact typed logical backing range"]
pub struct CompletionReadbackRequest {
node_id: NodeId,
participant_index: u32,
resource_id: ResourceId,
expected_usage: BufferUsage,
logical_offset_bytes: u64,
output_layout: HostTransferLayout,
}
impl CompletionReadbackRequest {
pub fn new(
node_id: NodeId,
participant_index: u32,
resource_id: ResourceId,
logical_offset_bytes: u64,
output_layout: HostTransferLayout,
) -> Result<Self, VNextError> {
Self::new_typed(
node_id,
participant_index,
resource_id,
BufferUsage::Activations,
logical_offset_bytes,
output_layout,
)
}
pub fn new_typed(
node_id: NodeId,
participant_index: u32,
resource_id: ResourceId,
expected_usage: BufferUsage,
logical_offset_bytes: u64,
output_layout: HostTransferLayout,
) -> Result<Self, VNextError> {
let output_bytes = output_layout.byte_len()?;
if logical_offset_bytes.checked_add(output_bytes).is_none() {
return Err(invalid_completion(
"completion readback logical range overflows u64",
));
}
Ok(Self {
node_id,
participant_index,
resource_id,
expected_usage,
logical_offset_bytes,
output_layout,
})
}
pub fn node_id(&self) -> &NodeId {
&self.node_id
}
pub fn resource_id(&self) -> &ResourceId {
&self.resource_id
}
pub const fn expected_usage(&self) -> BufferUsage {
self.expected_usage
}
pub const fn participant_index(&self) -> u32 {
self.participant_index
}
pub const fn logical_offset_bytes(&self) -> u64 {
self.logical_offset_bytes
}
pub const fn output_layout(&self) -> HostTransferLayout {
self.output_layout
}
}
#[derive(Debug, Clone, PartialEq, Eq, Serialize)]
#[must_use = "a completion readback batch must cover one exact plan-node participant set"]
pub struct CompletionReadbackBatchRequest {
requests: Vec<CompletionReadbackRequest>,
}
impl CompletionReadbackBatchRequest {
pub fn new(requests: Vec<CompletionReadbackRequest>) -> Result<Self, VNextError> {
let Some(first) = requests.first() else {
return Err(invalid_completion(
"completion readback batch cannot be empty",
));
};
if requests.iter().enumerate().any(|(index, request)| {
usize::try_from(request.participant_index()).ok() != Some(index)
|| request.node_id() != first.node_id()
|| request.resource_id() != first.resource_id()
|| request.expected_usage() != first.expected_usage()
|| request.output_layout().element_type() != first.output_layout().element_type()
}) {
return Err(invalid_completion(
"completion readback batch must use canonical participant order and one typed node/resource/element group",
));
}
Ok(Self { requests })
}
pub fn requests(&self) -> &[CompletionReadbackRequest] {
&self.requests
}
pub fn len(&self) -> usize {
self.requests.len()
}
pub fn is_empty(&self) -> bool {
self.requests.is_empty()
}
fn validate_for(&self, batch_identity: &BatchOperationIdentity) -> Result<(), VNextError> {
let node_id = self.requests[0].node_id();
let node_index = batch_identity.node_index(node_id).ok_or_else(|| {
invalid_completion("completion readback batch node is absent from its submission")
})?;
let participant_count = batch_identity
.node_participant_count(node_index)
.ok_or_else(|| {
invalid_completion("completion readback batch node is absent from its submission")
})?;
if participant_count != self.requests.len() {
return Err(invalid_completion(
"completion readback batch must cover every submitted node participant exactly once",
));
}
Ok(())
}
fn into_requests(self) -> Vec<CompletionReadbackRequest> {
self.requests
}
}
#[derive(Debug, Clone, PartialEq, Eq, Serialize)]
#[must_use = "successful completion output bytes are exact readback evidence"]
pub struct CompletionReadbackOutput {
request: CompletionReadbackRequest,
bytes: Vec<u8>,
sha256: String,
#[serde(skip)]
timing: DeviceTimingMeasurement<CompletionReadbackTiming>,
}
impl CompletionReadbackOutput {
fn new(request: CompletionReadbackRequest, readback: LaneReadback) -> Result<Self, VNextError> {
request.output_layout.validate_bytes(readback.bytes.len())?;
let sha256 = format!("{:x}", Sha256::digest(&readback.bytes));
Ok(Self {
request,
bytes: readback.bytes,
sha256,
timing: readback.timing,
})
}
pub fn request(&self) -> &CompletionReadbackRequest {
&self.request
}
pub fn bytes(&self) -> &[u8] {
&self.bytes
}
pub fn sha256(&self) -> &str {
&self.sha256
}
pub const fn timing(&self) -> DeviceTimingMeasurement<CompletionReadbackTiming> {
self.timing
}
pub fn into_bytes(self) -> Vec<u8> {
self.bytes
}
}
#[derive(Debug, Clone, PartialEq, Eq, Serialize)]
#[serde(rename_all = "snake_case", tag = "status", content = "detail")]
pub enum CompletionReadbackDisposition {
Succeeded(CompletionReadbackOutput),
NotAttempted(CompletionReadbackRequest),
FailedButQuiescent {
request: CompletionReadbackRequest,
failures: Vec<IdentifiedFailure>,
},
ContractFailedButQuiescent {
request: CompletionReadbackRequest,
failure: QuiescentCompletionContractFailure,
},
}
#[derive(Serialize)]
#[serde(rename_all = "snake_case", tag = "status", content = "detail")]
enum CompletionReadbackDispositionFingerprint<'a> {
Succeeded {
request: &'a CompletionReadbackRequest,
output_sha256: &'a str,
},
NotAttempted {
request: &'a CompletionReadbackRequest,
},
FailedButQuiescent {
request: &'a CompletionReadbackRequest,
failures: &'a [IdentifiedFailure],
},
ContractFailedButQuiescent {
request: &'a CompletionReadbackRequest,
failure: &'a str,
},
}
impl<'a> From<&'a CompletionReadbackDisposition> for CompletionReadbackDispositionFingerprint<'a> {
fn from(disposition: &'a CompletionReadbackDisposition) -> Self {
match disposition {
CompletionReadbackDisposition::Succeeded(output) => Self::Succeeded {
request: output.request(),
output_sha256: output.sha256(),
},
CompletionReadbackDisposition::NotAttempted(request) => Self::NotAttempted { request },
CompletionReadbackDisposition::FailedButQuiescent { request, failures } => {
Self::FailedButQuiescent { request, failures }
}
CompletionReadbackDisposition::ContractFailedButQuiescent { request, failure } => {
Self::ContractFailedButQuiescent {
request,
failure: failure.reason(),
}
}
}
}
}
struct CompletionReadbackDispositionFingerprints<'a>(&'a [CompletionReadbackDisposition]);
impl Serialize for CompletionReadbackDispositionFingerprints<'_> {
fn serialize<S>(&self, serializer: S) -> Result<S::Ok, S::Error>
where
S: serde::Serializer,
{
let mut sequence = serializer.serialize_seq(Some(self.0.len()))?;
for disposition in self.0 {
sequence
.serialize_element(&CompletionReadbackDispositionFingerprint::from(disposition))?;
}
sequence.end()
}
}
fn completion_readback_batch_fingerprint(
completion_fingerprint: &str,
dispositions: &[CompletionReadbackDisposition],
) -> String {
#[derive(Serialize)]
struct FingerprintInput<'a> {
domain: &'static str,
completion_fingerprint: &'a str,
dispositions: CompletionReadbackDispositionFingerprints<'a>,
}
canonical_completion_fingerprint(&FingerprintInput {
domain: "ferrum.runtime-vnext.completion-readback-batch.v2",
completion_fingerprint,
dispositions: CompletionReadbackDispositionFingerprints(dispositions),
})
}
#[derive(Debug, Clone, PartialEq, Eq, Serialize)]
#[must_use = "a terminal readback receipt couples output evidence to its exact completion"]
pub struct CompletionReadbackReceipt {
completion: OperationCompletionReceipt,
disposition: CompletionReadbackDisposition,
readback_timing: Option<DeviceTimingMeasurement<CompletionReadbackTiming>>,
fingerprint: String,
}
impl CompletionReadbackReceipt {
fn new(
completion: OperationCompletionReceipt,
disposition: CompletionReadbackDisposition,
) -> Self {
let readback_timing = completion
.fence_timing()
.timing_mode()
.completion_enabled()
.then(|| readback_timing_for_disposition(&disposition));
#[derive(Serialize)]
struct FingerprintInput<'a> {
domain: &'static str,
completion_fingerprint: &'a str,
request: &'a CompletionReadbackRequest,
output_sha256: Option<&'a str>,
failures: Option<&'a [IdentifiedFailure]>,
contract_failure: Option<&'a str>,
}
let (request, output_sha256, failures, contract_failure) = match &disposition {
CompletionReadbackDisposition::Succeeded(output) => {
(output.request(), Some(output.sha256()), None, None)
}
CompletionReadbackDisposition::NotAttempted(request) => (request, None, None, None),
CompletionReadbackDisposition::FailedButQuiescent { request, failures } => {
(request, None, Some(failures.as_slice()), None)
}
CompletionReadbackDisposition::ContractFailedButQuiescent { request, failure } => {
(request, None, None, Some(failure.reason()))
}
};
let fingerprint = canonical_completion_fingerprint(&FingerprintInput {
domain: "ferrum.runtime-vnext.completion-readback.v1",
completion_fingerprint: completion.fingerprint(),
request,
output_sha256,
failures,
contract_failure,
});
Self {
completion,
disposition,
readback_timing,
fingerprint,
}
}
pub fn completion(&self) -> &OperationCompletionReceipt {
&self.completion
}
pub fn disposition(&self) -> &CompletionReadbackDisposition {
&self.disposition
}
pub const fn readback_timing(
&self,
) -> Option<DeviceTimingMeasurement<CompletionReadbackTiming>> {
self.readback_timing
}
pub fn fingerprint(&self) -> &str {
&self.fingerprint
}
}
#[derive(Debug, Clone, PartialEq, Eq, Serialize)]
#[must_use = "a terminal batch readback receipt owns all participant output evidence"]
pub struct CompletionReadbackBatchReceipt {
completion: OperationCompletionReceipt,
dispositions: Vec<CompletionReadbackDisposition>,
readback_timings: Option<Vec<DeviceTimingMeasurement<CompletionReadbackTiming>>>,
fingerprint: String,
}
impl CompletionReadbackBatchReceipt {
fn new(
completion: OperationCompletionReceipt,
dispositions: Vec<CompletionReadbackDisposition>,
) -> Self {
let readback_timings = completion
.fence_timing()
.timing_mode()
.completion_enabled()
.then(|| {
dispositions
.iter()
.map(readback_timing_for_disposition)
.collect()
});
let fingerprint =
completion_readback_batch_fingerprint(completion.fingerprint(), &dispositions);
Self {
completion,
dispositions,
readback_timings,
fingerprint,
}
}
pub fn completion(&self) -> &OperationCompletionReceipt {
&self.completion
}
pub fn dispositions(&self) -> &[CompletionReadbackDisposition] {
&self.dispositions
}
pub fn readback_timings(&self) -> Option<&[DeviceTimingMeasurement<CompletionReadbackTiming>]> {
self.readback_timings.as_deref()
}
pub fn fingerprint(&self) -> &str {
&self.fingerprint
}
}
fn readback_timing_for_disposition(
disposition: &CompletionReadbackDisposition,
) -> DeviceTimingMeasurement<CompletionReadbackTiming> {
match disposition {
CompletionReadbackDisposition::Succeeded(output) => output.timing(),
CompletionReadbackDisposition::NotAttempted(_)
| CompletionReadbackDisposition::FailedButQuiescent { .. }
| CompletionReadbackDisposition::ContractFailedButQuiescent { .. } => {
DeviceTimingMeasurement::NotRequested
}
}
}
fn attempt_completion_readback<R: DeviceRuntime>(
resources: &CompletionResourceLease<R>,
lane: &Arc<ExecutionLane<R>>,
batch_identity: &BatchOperationIdentity,
completion_disposition: &OperationCompletionDisposition,
timing_mode: DeviceTimingMode,
request: CompletionReadbackRequest,
) -> CompletionReadbackDisposition {
if !matches!(
completion_disposition,
OperationCompletionDisposition::Succeeded
) {
return CompletionReadbackDisposition::NotAttempted(request);
}
let byte_len = match request.output_layout().byte_len() {
Ok(byte_len) => byte_len,
Err(error) => {
return CompletionReadbackDisposition::ContractFailedButQuiescent {
request,
failure: QuiescentCompletionContractFailure::new(error.to_string()),
};
}
};
let semantic_end = match request.logical_offset_bytes().checked_add(byte_len) {
Some(end) => end,
None => {
return CompletionReadbackDisposition::ContractFailedButQuiescent {
request,
failure: QuiescentCompletionContractFailure::new(
"completion readback semantic range overflows u64",
),
};
}
};
let readback_range = match resources.readback_range(
request.node_id(),
request.participant_index(),
request.resource_id(),
request.logical_offset_bytes()..semantic_end,
) {
Ok(range) => range,
Err(error) => {
return CompletionReadbackDisposition::ContractFailedButQuiescent {
request,
failure: QuiescentCompletionContractFailure::new(error.to_string()),
};
}
};
let backing = match resources.backing_view(
request.node_id(),
request.participant_index(),
request.resource_id(),
) {
Ok(backing) => backing,
Err(error) => {
return CompletionReadbackDisposition::ContractFailedButQuiescent {
request,
failure: QuiescentCompletionContractFailure::new(error.to_string()),
};
}
};
let readback = catch_unwind(AssertUnwindSafe(|| {
lane.readback_buffer(
&backing,
request.expected_usage(),
readback_range.start,
request.output_layout(),
timing_mode,
)
}));
match readback {
Ok(Ok(readback)) => match CompletionReadbackOutput::new(request.clone(), readback) {
Ok(output) => CompletionReadbackDisposition::Succeeded(output),
Err(error) => CompletionReadbackDisposition::ContractFailedButQuiescent {
request,
failure: QuiescentCompletionContractFailure::new(error.to_string()),
},
},
Ok(Err(LaneReadbackError::Contract(error))) => {
CompletionReadbackDisposition::ContractFailedButQuiescent {
request,
failure: QuiescentCompletionContractFailure::new(error.to_string()),
}
}
Ok(Err(LaneReadbackError::Device(error))) => {
match classify_batch_device_error(lane.runtime(), batch_identity, &error) {
Ok(failures) => {
CompletionReadbackDisposition::FailedButQuiescent { request, failures }
}
Err(error) => CompletionReadbackDisposition::ContractFailedButQuiescent {
request,
failure: QuiescentCompletionContractFailure::new(error.to_string()),
},
}
}
Err(_) => {
lane.fail_closed();
CompletionReadbackDisposition::ContractFailedButQuiescent {
request,
failure: QuiescentCompletionContractFailure::new(
"device runtime panicked during completion readback",
),
}
}
}
}
fn attempt_completion_readbacks<R: DeviceRuntime>(
resources: &CompletionResourceLease<R>,
lane: &Arc<ExecutionLane<R>>,
batch_identity: &BatchOperationIdentity,
completion_disposition: &OperationCompletionDisposition,
timing_mode: DeviceTimingMode,
request: CompletionReadbackBatchRequest,
) -> Vec<CompletionReadbackDisposition> {
request
.into_requests()
.into_iter()
.map(|request| {
attempt_completion_readback(
resources,
lane,
batch_identity,
completion_disposition,
timing_mode,
request,
)
})
.collect()
}
fn attempt_completion_readback_collection<R: DeviceRuntime>(
resources: &CompletionResourceLease<R>,
lane: &Arc<ExecutionLane<R>>,
batch_identity: &BatchOperationIdentity,
completion_disposition: &OperationCompletionDisposition,
timing_mode: DeviceTimingMode,
request: CompletionReadbackCollectionRequest,
) -> Vec<CompletionReadbackDisposition> {
request
.into_requests()
.into_iter()
.map(|request| {
attempt_completion_readback(
resources,
lane,
batch_identity,
completion_disposition,
timing_mode,
request,
)
})
.collect()
}
#[derive(Debug)]
#[must_use = "nonterminal completion readback observations retain invocation ownership"]
pub enum CompletionReadbackObservation {
Pending,
Terminal(CompletionReadbackReceipt),
Indeterminate(Vec<IdentifiedFailure>),
SubmissionIndeterminate,
ObservationPanicked,
Quarantined(CompletionQuarantineReceipt),
}
#[derive(Debug)]
#[must_use = "nonterminal batch readback observations retain invocation ownership"]
pub enum CompletionReadbackBatchObservation {
Pending,
Terminal(CompletionReadbackBatchReceipt),
Indeterminate(Vec<IdentifiedFailure>),
SubmissionIndeterminate,
ObservationPanicked,
Quarantined(CompletionQuarantineReceipt),
}
#[derive(Debug)]
#[must_use = "nonterminal completion observations retain invocation ownership"]
pub enum CompletionObservation {
Pending,
Terminal(OperationCompletionReceipt),
Indeterminate(Vec<IdentifiedFailure>),
SubmissionIndeterminate,
ObservationPanicked,
Quarantined(CompletionQuarantineReceipt),
}
#[must_use = "an indeterminate submission must be drained or retained"]
pub struct IndeterminateSubmissionHandle<R: DeviceRuntime> {
reaper: Weak<CompletionReaper<R>>,
slot_id: CompletionSlotId,
}
impl<R: DeviceRuntime> Clone for IndeterminateSubmissionHandle<R> {
fn clone(&self) -> Self {
Self {
reaper: Weak::clone(&self.reaper),
slot_id: self.slot_id,
}
}
}
impl<R: DeviceRuntime> fmt::Debug for IndeterminateSubmissionHandle<R> {
fn fmt(&self, formatter: &mut fmt::Formatter<'_>) -> fmt::Result {
formatter
.debug_struct("IndeterminateSubmissionHandle")
.field("slot_id", &self.slot_id)
.finish_non_exhaustive()
}
}
impl<R: DeviceRuntime> IndeterminateSubmissionHandle<R> {
pub const fn slot_id(&self) -> CompletionSlotId {
self.slot_id
}
pub fn recover_by_draining_lane(&self) -> Result<CompletionRecoveryOutcome, VNextError> {
self.reaper
.upgrade()
.ok_or_else(|| invalid_completion("completion reaper owner was dropped"))?
.recover_bound(self.slot_id)
}
}
#[must_use = "a submitted operation handle observes its exact completion slot"]
pub struct CompletionHandle<R: DeviceRuntime> {
reaper: Weak<CompletionReaper<R>>,
receipt: SubmittedOperationReceipt,
}
impl<R: DeviceRuntime> Clone for CompletionHandle<R> {
fn clone(&self) -> Self {
Self {
reaper: Weak::clone(&self.reaper),
receipt: self.receipt.clone(),
}
}
}
impl<R: DeviceRuntime> fmt::Debug for CompletionHandle<R> {
fn fmt(&self, formatter: &mut fmt::Formatter<'_>) -> fmt::Result {
formatter
.debug_struct("CompletionHandle")
.field("receipt", &self.receipt)
.finish_non_exhaustive()
}
}
impl<R: DeviceRuntime> CompletionHandle<R> {
pub fn receipt(&self) -> &SubmittedOperationReceipt {
&self.receipt
}
pub fn batch_identity(&self) -> &BatchOperationIdentity {
self.receipt.batch_identity()
}
pub fn slot_id(&self) -> CompletionSlotId {
self.receipt.slot_id()
}
pub fn poll(&self) -> Result<CompletionObservation, VNextError> {
self.reaper
.upgrade()
.ok_or_else(|| invalid_completion("completion reaper owner was dropped"))?
.poll_bound(self.slot_id())
}
pub fn wait(&self) -> Result<CompletionObservation, VNextError> {
self.reaper
.upgrade()
.ok_or_else(|| invalid_completion("completion reaper owner was dropped"))?
.wait_bound(self.slot_id())
}
pub fn wait_with_readback(
&self,
request: CompletionReadbackRequest,
) -> Result<CompletionReadbackObservation, VNextError> {
self.reaper
.upgrade()
.ok_or_else(|| invalid_completion("completion reaper owner was dropped"))?
.wait_bound_with_readback(self.slot_id(), request)
}
pub fn wait_with_readbacks(
&self,
request: CompletionReadbackBatchRequest,
) -> Result<CompletionReadbackBatchObservation, VNextError> {
self.reaper
.upgrade()
.ok_or_else(|| invalid_completion("completion reaper owner was dropped"))?
.wait_bound_with_readbacks(self.slot_id(), request)
}
pub fn wait_with_readback_collection(
&self,
request: CompletionReadbackCollectionRequest,
) -> Result<CompletionReadbackCollectionObservation, VNextError> {
self.reaper
.upgrade()
.ok_or_else(|| invalid_completion("completion reaper owner was dropped"))?
.wait_bound_with_readback_collection(self.slot_id(), request)
}
}
pub(crate) struct CompletionReservation<R: DeviceRuntime> {
reaper: Arc<CompletionReaper<R>>,
record: SharedCompletionRecord<R>,
slot_id: CompletionSlotId,
resources: Option<CompletionResourceLease<R>>,
lane: Option<Arc<ExecutionLane<R>>>,
batch_identity: Option<BatchOperationIdentity>,
receipt: Option<SubmittedOperationReceipt>,
record_receipt: Option<SubmittedOperationReceipt>,
submission_may_have_happened: bool,
finished: bool,
}
impl<R: DeviceRuntime> CompletionReservation<R> {
pub(crate) fn invocation(&self) -> &InvocationResourceLease<R> {
match self
.resources
.as_ref()
.expect("live completion reservation owns submission resources")
{
CompletionResourceLease::Invocation(invocation) => invocation,
CompletionResourceLease::Wave(_) => {
unreachable!("single-operation reservation cannot own a submission wave")
}
}
}
pub(crate) fn wave(&self) -> &PreparedStepSubmissionWave<R> {
match self
.resources
.as_ref()
.expect("live completion reservation owns submission resources")
{
CompletionResourceLease::Wave(wave) => wave,
CompletionResourceLease::Invocation(_) => {
unreachable!("wave reservation cannot own single-operation resources")
}
}
}
pub(crate) fn backing_view(
&self,
node_id: &NodeId,
participant_index: u32,
resource_id: &ResourceId,
) -> Result<LogicalBackingBufferView<'_, R::Buffer>, VNextError> {
self.resources
.as_ref()
.expect("live completion reservation owns submission resources")
.backing_view(node_id, participant_index, resource_id)
}
pub(crate) fn encode_backing_initializations(
&self,
runtime: &R,
commands: &mut DeviceCommandBatch<R::Command>,
) -> Result<usize, BackingInitializationEncodeError<R::Error>> {
self.resources
.as_ref()
.expect("live completion reservation owns submission resources")
.encode_backing_initializations(runtime, commands)
}
pub(crate) fn mark_submission_started(&mut self) {
self.submission_may_have_happened = true;
}
pub(crate) fn definitely_not_submitted(
mut self,
) -> Result<DefinitelyNotSubmittedRetryAuthority<R>, VNextError> {
self.remove_reserved_slot();
self.submission_may_have_happened = false;
let resources = self
.resources
.take()
.expect("reservation owns definitely-not-submitted resources");
let CompletionResourceLease::Invocation(invocation) = resources else {
return Err(invalid_completion(
"single-operation retry requested from a submission wave reservation",
));
};
let retry = invocation.definitely_not_submitted()?;
self.finished = true;
Ok(retry)
}
pub(crate) fn definitely_not_submitted_wave(
mut self,
) -> Result<DefinitelyNotSubmittedWaveRetryAuthority<R>, VNextError> {
self.remove_reserved_slot();
self.submission_may_have_happened = false;
let resources = self
.resources
.take()
.expect("reservation owns definitely-not-submitted resources");
let CompletionResourceLease::Wave(wave) = resources else {
return Err(invalid_completion(
"wave retry requested from a single-operation reservation",
));
};
let retry = wave.definitely_not_submitted()?;
self.finished = true;
Ok(retry)
}
pub(crate) fn arm(
mut self,
fence: R::Fence,
timing_mode: DeviceTimingMode,
) -> Result<CompletionHandle<R>, (VNextError, CompletionHandle<R>)> {
let resources = self.resources.take().expect("reservation owns resources");
let lane = self.lane.take().expect("reservation owns lane");
let batch_identity = self
.batch_identity
.take()
.expect("reservation owns batch identity");
let receipt = self.receipt.take().expect("reservation owns receipt");
let record_receipt = self
.record_receipt
.take()
.expect("reservation owns record receipt");
let mut record = match self.record.lock() {
Ok(record) => record,
Err(poisoned) => poisoned.into_inner(),
};
if !matches!(&*record, CompletionRecord::Reserved) {
lane.fail_closed();
std::mem::forget((resources, lane, fence));
self.finished = true;
panic!("completion reservation changed after submission");
}
*record = CompletionRecord::InFlight {
resources,
lane,
fence,
batch_identity,
receipt: record_receipt,
timing_mode,
recovery_state: CompletionRecoveryState::Unobserved,
};
let transition = match &mut *record {
CompletionRecord::InFlight {
resources, lane, ..
} => match resources.mark_submission_fence_installed() {
Ok(()) => Ok(()),
Err(error) => {
resources.mark_submission_indeterminate();
lane.fail_closed();
Err(error)
}
},
_ => unreachable!("completion record was just armed"),
};
drop(record);
self.finished = true;
let handle = CompletionHandle {
reaper: Arc::downgrade(&self.reaper),
receipt,
};
match transition {
Ok(()) => Ok(handle),
Err(error) => Err((error, handle)),
}
}
pub(crate) fn submission_indeterminate(mut self) -> IndeterminateSubmissionHandle<R> {
let mut resources = self.resources.take().expect("reservation owns resources");
resources.mark_submission_indeterminate();
let lane = self.lane.take().expect("reservation owns lane");
let batch_identity = self
.batch_identity
.take()
.expect("reservation owns batch identity");
let mut record = match self.record.lock() {
Ok(record) => record,
Err(poisoned) => poisoned.into_inner(),
};
if matches!(&*record, CompletionRecord::Reserved) {
*record = CompletionRecord::SubmissionIndeterminate {
resources,
lane,
batch_identity,
};
} else {
lane.fail_closed();
std::mem::forget((resources, lane, batch_identity));
}
drop(record);
self.finished = true;
IndeterminateSubmissionHandle {
reaper: Arc::downgrade(&self.reaper),
slot_id: self.slot_id,
}
}
fn remove_reserved_slot(&mut self) {
self.reaper.remove_exact(self.slot_id, &self.record);
let mut record = match self.record.lock() {
Ok(record) => record,
Err(poisoned) => poisoned.into_inner(),
};
if matches!(&*record, CompletionRecord::Reserved) {
*record = CompletionRecord::Reaped;
}
}
}
impl<R: DeviceRuntime> Drop for CompletionReservation<R> {
fn drop(&mut self) {
if self.finished {
return;
}
if self.submission_may_have_happened {
let Some(mut resources) = self.resources.take() else {
return;
};
resources.mark_submission_indeterminate();
let Some(lane) = self.lane.take() else {
std::mem::forget(resources);
return;
};
let Some(batch_identity) = self.batch_identity.take() else {
std::mem::forget((resources, lane));
return;
};
lane.fail_closed();
let mut record = match self.record.lock() {
Ok(record) => record,
Err(poisoned) => poisoned.into_inner(),
};
if matches!(&*record, CompletionRecord::Reserved) {
*record = CompletionRecord::SubmissionIndeterminate {
resources,
lane,
batch_identity,
};
} else {
std::mem::forget((resources, lane, batch_identity));
}
} else {
self.remove_reserved_slot();
}
}
}
#[cfg(test)]
mod fingerprint_tests {
use super::*;
use crate::vnext::ElementType;
const LARGE_READBACK_BYTES: usize = 512 * 1024;
fn request() -> CompletionReadbackRequest {
CompletionReadbackRequest::new(
NodeId::try_from("node.fingerprint-test".to_owned()).unwrap(),
0,
ResourceId::try_from("resource.fingerprint-test".to_owned()).unwrap(),
0,
HostTransferLayout::new(ElementType::U8, LARGE_READBACK_BYTES as u64).unwrap(),
)
.unwrap()
}
fn successful_output(fill: u8) -> CompletionReadbackOutput {
let bytes = vec![fill; LARGE_READBACK_BYTES];
let sha256 = format!("{:x}", Sha256::digest(&bytes));
CompletionReadbackOutput {
request: request(),
bytes,
sha256,
timing: DeviceTimingMeasurement::NotRequested,
}
}
#[test]
fn batch_fingerprint_is_bounded_by_digest_evidence_not_output_bytes() {
let first = vec![CompletionReadbackDisposition::Succeeded(successful_output(
0x5a,
))];
let encoded = serde_json::to_vec(&CompletionReadbackDispositionFingerprints(&first))
.expect("fingerprint evidence serializes");
assert!(encoded.len() < 1024, "encoded {} bytes", encoded.len());
let first_sha = match &first[0] {
CompletionReadbackDisposition::Succeeded(output) => output.sha256(),
_ => unreachable!(),
};
assert!(String::from_utf8(encoded).unwrap().contains(first_sha));
let completion_fingerprint = "a".repeat(64);
let first_fingerprint =
completion_readback_batch_fingerprint(&completion_fingerprint, &first);
assert_eq!(
first_fingerprint,
completion_readback_batch_fingerprint(&completion_fingerprint, &first)
);
let second = vec![CompletionReadbackDisposition::Succeeded(successful_output(
0xa5,
))];
assert_ne!(
first_fingerprint,
completion_readback_batch_fingerprint(&completion_fingerprint, &second)
);
let not_attempted = vec![CompletionReadbackDisposition::NotAttempted(request())];
assert_ne!(
first_fingerprint,
completion_readback_batch_fingerprint(&completion_fingerprint, ¬_attempted)
);
}
}