vyre_runtime/megakernel/
descriptor.rs

1//! Typed host-side descriptors for publishing work into the megakernel ring.
2//!
3//! Wrappers such as VyreOffload should not have to hand-assemble
4//! `(opcode, tenant_id, args)` tuples or know when to switch to the
5//! packed-slot path. These descriptors provide an additive typed API
6//! over the existing wire protocol.
7
8use super::staging_reserve::reserve_vec_capacity as reserve_descriptor_vec;
9use crate::PipelineError;
10
11use smallvec::SmallVec;
12
13const ARGS_PER_SLOT_USIZE: usize = 12;
14
15use super::{protocol, ResidentWorkQueue};
16
17/// Built-in megakernel opcodes exposed as a typed host API.
18#[derive(Debug, Clone, Copy, PartialEq, Eq)]
19pub enum BuiltinOpcode {
20    /// No-op heartbeat / probe.
21    Nop,
22    /// `control[arg1] = arg0`.
23    StoreU32,
24    /// `atomic_add(control[arg1], arg0)`.
25    AtomicAdd,
26    /// `control[OBSERVABLE_BASE + arg1] = control[arg0]`.
27    LoadU32,
28    /// Compare-and-swap on `control[arg0]`.
29    CompareSwap,
30    /// Copy `arg2` words from `control[arg0]` to `control[arg1]`.
31    Memcpy,
32    /// Single DFA transition step.
33    DfaStep,
34    /// Batch fence / epoch bump.
35    BatchFence,
36    /// Emit a debug log record.
37    Printf,
38    /// Set `SHUTDOWN=1`.
39    Shutdown,
40}
41
42impl BuiltinOpcode {
43    /// Underlying wire opcode.
44    #[must_use]
45    pub const fn into_wire(self) -> u32 {
46        match self {
47            Self::Nop => protocol::opcode::NOP,
48            Self::StoreU32 => protocol::opcode::STORE_U32,
49            Self::AtomicAdd => protocol::opcode::ATOMIC_ADD,
50            Self::LoadU32 => protocol::opcode::LOAD_U32,
51            Self::CompareSwap => protocol::opcode::COMPARE_SWAP,
52            Self::Memcpy => protocol::opcode::MEMCPY,
53            Self::DfaStep => protocol::opcode::DFA_STEP,
54            Self::BatchFence => protocol::opcode::BATCH_FENCE,
55            Self::Printf => protocol::opcode::PRINTF,
56            Self::Shutdown => protocol::opcode::SHUTDOWN,
57        }
58    }
59}
60
61/// A slot opcode can target either a builtin wire opcode or a caller-defined
62/// extension registered via an opcode handler (see `handlers` module).
63#[derive(Debug, Clone, Copy, PartialEq, Eq)]
64pub enum SlotOpcode {
65    /// One of the frozen builtins in [`protocol::opcode`].
66    Builtin(BuiltinOpcode),
67    /// A custom extension opcode.
68    Custom(u32),
69}
70
71impl SlotOpcode {
72    /// Underlying wire opcode.
73    #[must_use]
74    pub const fn into_wire(self) -> u32 {
75        match self {
76            Self::Builtin(op) => op.into_wire(),
77            Self::Custom(op) => op,
78        }
79    }
80}
81
82/// One packed inner-op inside a `PACKED_SLOT`.
83#[derive(Debug, Clone, PartialEq, Eq)]
84pub struct PackedOpDescriptor {
85    /// Inner opcode id. Must fit in `u8` due to the current wire format.
86    pub opcode: u8,
87    /// Positional `u32` arguments for the inner opcode.
88    pub args: Vec<u32>,
89}
90
91impl PackedOpDescriptor {
92    /// Convenience constructor.
93    #[must_use]
94    pub fn new(opcode: u8, args: Vec<u32>) -> Self {
95        Self { opcode, args }
96    }
97}
98
99/// One top-level slot publication request.
100#[derive(Debug, Clone, PartialEq, Eq)]
101pub enum SlotDescriptor {
102    /// Publish one normal slot.
103    Single {
104        /// Tenant id used for the runtime's authorization mask.
105        tenant_id: u32,
106        /// Slot opcode.
107        opcode: SlotOpcode,
108        /// Positional `u32` arguments.
109        args: Vec<u32>,
110    },
111    /// Publish one packed slot containing several inner ops.
112    Packed {
113        /// Tenant id used for the runtime's authorization mask.
114        tenant_id: u32,
115        /// Inner packed ops.
116        ops: Vec<PackedOpDescriptor>,
117    },
118}
119
120impl SlotDescriptor {
121    /// Build a simple slot descriptor.
122    #[must_use]
123    pub fn single(tenant_id: u32, opcode: SlotOpcode, args: Vec<u32>) -> Self {
124        Self::Single {
125            tenant_id,
126            opcode,
127            args,
128        }
129    }
130
131    /// Build a packed-slot descriptor.
132    #[must_use]
133    pub fn packed(tenant_id: u32, ops: Vec<PackedOpDescriptor>) -> Self {
134        Self::Packed { tenant_id, ops }
135    }
136
137    /// Publish this slot into the ring at `slot_idx`.
138    ///
139    /// # Errors
140    ///
141    /// Propagates any wire-level publication error from the underlying ring
142    /// protocol helpers.
143    pub fn publish_into(&self, ring_bytes: &mut [u8], slot_idx: u32) -> Result<(), PipelineError> {
144        match self {
145            Self::Single {
146                tenant_id,
147                opcode,
148                args,
149            } => ResidentWorkQueue::publish_slot(
150                ring_bytes,
151                slot_idx,
152                *tenant_id,
153                opcode.into_wire(),
154                args,
155            ),
156            Self::Packed { tenant_id, ops } => {
157                ResidentWorkQueue::publish_packed_descriptors(ring_bytes, slot_idx, *tenant_id, ops)
158            }
159        }
160    }
161}
162
163/// A typed batch publication request.
164#[derive(Debug, Clone, PartialEq, Eq)]
165pub struct BatchDescriptor {
166    /// Slot index where the first item should be written.
167    pub start_slot: u32,
168    /// Items to publish in order.
169    pub items: Vec<SlotDescriptor>,
170}
171
172impl BatchDescriptor {
173    /// Convenience constructor.
174    #[must_use]
175    pub fn new(start_slot: u32, items: Vec<SlotDescriptor>) -> Self {
176        Self { start_slot, items }
177    }
178
179    /// Publish all items into the ring. Returns the number of slots consumed.
180    ///
181    /// # Errors
182    ///
183    /// Propagates any slot publication error.
184    pub fn publish_into(&self, ring_bytes: &mut [u8]) -> Result<u32, PipelineError> {
185        let item_count = u32::try_from(self.items.len()).map_err(|_| PipelineError::QueueFull {
186            queue: "submission",
187            fix: "batch size exceeds u32::MAX slots",
188        })?;
189        if item_count > 0 {
190            self.start_slot
191                .checked_add(item_count - 1)
192                .ok_or(PipelineError::QueueFull {
193                    queue: "submission",
194                    fix: "batch start plus item count overflows u32; split the descriptor batch before publishing",
195                })?;
196        }
197        for (slot_offset, item) in (0..item_count).zip(self.items.iter()) {
198            let slot_idx = self
199                .start_slot
200                .checked_add(slot_offset)
201                .ok_or(PipelineError::QueueFull {
202                queue: "submission",
203                fix:
204                    "batch slot index overflowed u32; split the descriptor batch before publishing",
205            })?;
206            item.publish_into(ring_bytes, slot_idx)?;
207        }
208        Ok(item_count)
209    }
210}
211
212/// Classification for items published inside a window descriptor.
213#[derive(Debug, Clone, Copy, PartialEq, Eq)]
214pub enum WindowClass {
215    /// Required work that must converge for the window to be usable.
216    Required,
217    /// Lookahead work that improves the next step but is not on the immediate critical path.
218    Lookahead,
219}
220
221impl WindowClass {
222    /// Stable on-the-wire encoding  -  `Required` = 0, `Lookahead` = 1.
223    #[must_use]
224    pub const fn into_wire(self) -> u32 {
225        match self {
226            Self::Required => 0,
227            Self::Lookahead => 1,
228        }
229    }
230}
231
232/// A ticketed window of related slot publications.
233///
234/// Each emitted slot receives a stable prefix of `[window_ticket, class_tag]`
235/// followed by the caller-supplied payload, so wrappers can submit required and
236/// lookahead work as one structured batch without hand-assembling the prefix.
237#[derive(Debug, Clone, PartialEq, Eq)]
238pub struct WindowDescriptor {
239    /// Slot index where the first window item should be written.
240    pub start_slot: u32,
241    /// Tenant id used for all emitted slots.
242    pub tenant_id: u32,
243    /// Slot opcode shared by all emitted slots.
244    pub opcode: SlotOpcode,
245    /// Stable ticket id correlating every slot in this window.
246    pub ticket: u32,
247    /// Required entries for the window.
248    pub required: Vec<Vec<u32>>,
249    /// Lookahead entries for the window.
250    pub lookahead: Vec<Vec<u32>>,
251}
252
253impl WindowDescriptor {
254    /// Convenience constructor.
255    #[must_use]
256    pub fn new(
257        start_slot: u32,
258        tenant_id: u32,
259        opcode: SlotOpcode,
260        ticket: u32,
261        required: Vec<Vec<u32>>,
262        lookahead: Vec<Vec<u32>>,
263    ) -> Self {
264        Self {
265            start_slot,
266            tenant_id,
267            opcode,
268            ticket,
269            required,
270            lookahead,
271        }
272    }
273
274    /// Convert the window into a typed batch publication.
275    ///
276    /// # Panics
277    ///
278    /// Panics when `try_into_batch` fails (oversized payload, OOM, or slot
279    /// overflow). The panic message names the failing condition and directs the
280    /// caller to `try_into_batch`. Panicking here is intentional: silently
281    /// returning an empty `BatchDescriptor` would publish zero ring slots and
282    /// lose all window work with no operator signal (Law 10). In production
283    /// code, call `try_into_batch` directly so errors can be propagated.
284    #[must_use]
285    pub fn into_batch(self) -> BatchDescriptor {
286        self.try_into_batch().unwrap_or_else(|e| {
287            panic!(
288                "WindowDescriptor::into_batch failed: {e}. Fix: call try_into_batch() and propagate the error instead of using the infallible wrapper."
289            )
290        })
291    }
292
293    /// Convert the window into a typed batch publication with explicit staging
294    /// and ABI-bound errors.
295    pub fn try_into_batch(&self) -> Result<BatchDescriptor, PipelineError> {
296        let item_count = self
297            .required
298            .len()
299            .checked_add(self.lookahead.len())
300            .ok_or(PipelineError::QueueFull {
301            queue: "submission",
302            fix:
303                "window item count overflowed usize; split the window before materializing a batch",
304        })?;
305        let mut items = Vec::new();
306        reserve_descriptor_vec(&mut items, item_count, "window batch item")?;
307        for payload in &self.required {
308            let mut args = window_payload_args(self.ticket, WindowClass::Required, payload)?;
309            args.push(self.ticket);
310            args.push(WindowClass::Required.into_wire());
311            args.extend(payload.iter().copied());
312            items.push(SlotDescriptor::single(self.tenant_id, self.opcode, args));
313        }
314        for payload in &self.lookahead {
315            let mut args = window_payload_args(self.ticket, WindowClass::Lookahead, payload)?;
316            args.push(self.ticket);
317            args.push(WindowClass::Lookahead.into_wire());
318            args.extend(payload.iter().copied());
319            items.push(SlotDescriptor::single(self.tenant_id, self.opcode, args));
320        }
321        Ok(BatchDescriptor::new(self.start_slot, items))
322    }
323
324    /// Publish the full window into the ring and return the number of emitted slots.
325    pub fn publish_into(&self, ring_bytes: &mut [u8]) -> Result<u32, PipelineError> {
326        let consumed = self
327            .required
328            .len()
329            .checked_add(self.lookahead.len())
330            .ok_or(PipelineError::QueueFull {
331                queue: "submission",
332                fix: "window item count overflowed usize; split the window before publishing",
333            })?;
334        let consumed_u32 = u32::try_from(consumed).map_err(|_| PipelineError::QueueFull {
335            queue: "submission",
336            fix: "window size exceeds u32::MAX slots; split the window before publishing",
337        })?;
338        if consumed_u32 == 0 {
339            return Ok(0);
340        }
341        self.start_slot
342            .checked_add(consumed_u32 - 1)
343            .ok_or(PipelineError::QueueFull {
344                queue: "submission",
345                fix: "window start plus item count overflows u32; split the window before publishing",
346            })?;
347
348        let mut slot_offset = 0u32;
349        let mut args = SmallVec::<[u32; ARGS_PER_SLOT_USIZE]>::new();
350        for payload in &self.required {
351            publish_window_payload(
352                ring_bytes,
353                self.start_slot,
354                &mut slot_offset,
355                self.tenant_id,
356                self.opcode,
357                self.ticket,
358                WindowClass::Required,
359                payload,
360                &mut args,
361            )?;
362        }
363        for payload in &self.lookahead {
364            publish_window_payload(
365                ring_bytes,
366                self.start_slot,
367                &mut slot_offset,
368                self.tenant_id,
369                self.opcode,
370                self.ticket,
371                WindowClass::Lookahead,
372                payload,
373                &mut args,
374            )?;
375        }
376        Ok(slot_offset)
377    }
378}
379
380fn window_payload_args(
381    _ticket: u32,
382    _class: WindowClass,
383    payload: &[u32],
384) -> Result<Vec<u32>, PipelineError> {
385    let required_args = payload
386        .len()
387        .checked_add(2)
388        .ok_or(PipelineError::QueueFull {
389            queue: "submission",
390            fix: "window payload argument count overflowed usize; split the payload before materializing a batch",
391        })?;
392    if required_args > ARGS_PER_SLOT_USIZE {
393        return Err(PipelineError::QueueFull {
394            queue: "submission",
395            fix: "too many args for one window payload; ticket plus class plus payload must fit in 12 u32 args",
396        });
397    }
398    let mut args = Vec::new();
399    reserve_descriptor_vec(&mut args, required_args, "window payload arg")?;
400    Ok(args)
401}
402
403fn publish_window_payload(
404    ring_bytes: &mut [u8],
405    start_slot: u32,
406    slot_offset: &mut u32,
407    tenant_id: u32,
408    opcode: SlotOpcode,
409    ticket: u32,
410    class: WindowClass,
411    payload: &[u32],
412    args: &mut SmallVec<[u32; ARGS_PER_SLOT_USIZE]>,
413) -> Result<(), PipelineError> {
414    let slot_idx = start_slot
415        .checked_add(*slot_offset)
416        .ok_or(PipelineError::QueueFull {
417            queue: "submission",
418            fix: "window slot index overflowed u32; split the window before publishing",
419        })?;
420    args.clear();
421    let required_args = payload
422        .len()
423        .checked_add(2)
424        .ok_or(PipelineError::QueueFull {
425        queue: "submission",
426        fix: "window payload argument count overflowed usize; split the payload before publishing",
427    })?;
428    if required_args > ARGS_PER_SLOT_USIZE {
429        return Err(PipelineError::QueueFull {
430            queue: "submission",
431            fix: "too many args for one window payload; ticket plus class plus payload must fit in 12 u32 args",
432        });
433    }
434    args.push(ticket);
435    args.push(class.into_wire());
436    args.extend_from_slice(payload);
437    ResidentWorkQueue::publish_slot(ring_bytes, slot_idx, tenant_id, opcode.into_wire(), args)?;
438    *slot_offset = slot_offset.checked_add(1).ok_or(PipelineError::QueueFull {
439        queue: "submission",
440        fix: "window slot count overflowed u32; split the window before publishing",
441    })?;
442    Ok(())
443}
444
445#[cfg(test)]
446mod tests;
vyre_runtime/megakernel/descriptor.rs

vyre_runtime/megakernel/
descriptor.rs
