vyre_runtime/megakernel/telemetry/
ring_state.rs

1//! Decoded views of the megakernel ring and control buffers: slot status,
2//! per-window aggregates, occupancy counts, and the scratch a repeated decode
3//! reuses.
4
5use super::slot;
6use super::{TelemetryDecodeCapacityEvidence, TELEMETRY_DECODE_CAPACITY_SCHEMA_VERSION};
7use rustc_hash::FxHashMap;
8
9
10/// Decoded top-level ring slot state.
11#[derive(Debug, Clone, Copy, PartialEq, Eq)]
12pub enum RingStatus {
13    /// Slot is free.
14    Empty,
15    /// Slot is published and waiting for a worker.
16    Published,
17    /// Slot has been claimed by a worker.
18    Claimed,
19    /// Slot completed and can be recycled.
20    Done,
21    /// Slot is waiting for an asynchronous IO continuation.
22    WaitIo,
23    /// Slot yielded execution back to the scheduler.
24    Yield,
25    /// Slot is heavily contested and has been requeued.
26    Requeue,
27    /// Slot hit a hardware or software fault constraint.
28    Fault,
29    /// Unknown raw wire value.
30    Unknown(u32),
31}
32
33impl RingStatus {
34    #[must_use]
35    pub(super) fn from_raw(raw: u32) -> Self {
36        match raw {
37            slot::EMPTY => Self::Empty,
38            slot::PUBLISHED => Self::Published,
39            slot::CLAIMED => Self::Claimed,
40            slot::DONE => Self::Done,
41            slot::WAIT_IO => Self::WaitIo,
42            slot::YIELD => Self::Yield,
43            slot::REQUEUE => Self::Requeue,
44            slot::FAULT => Self::Fault,
45            other => Self::Unknown(other),
46        }
47    }
48
49    /// Raw wire discriminant for sketching, replay, and compact telemetry.
50    #[must_use]
51    pub const fn raw(self) -> u32 {
52        match self {
53            Self::Empty => slot::EMPTY,
54            Self::Published => slot::PUBLISHED,
55            Self::Claimed => slot::CLAIMED,
56            Self::Done => slot::DONE,
57            Self::WaitIo => slot::WAIT_IO,
58            Self::Yield => slot::YIELD,
59            Self::Requeue => slot::REQUEUE,
60            Self::Fault => slot::FAULT,
61            Self::Unknown(raw) => raw,
62        }
63    }
64
65    /// Whether this status still represents in-flight work rather than a
66    /// terminal slot outcome.
67    #[must_use]
68    pub const fn is_active(self) -> bool {
69        matches!(
70            self,
71            Self::Published | Self::Claimed | Self::WaitIo | Self::Yield | Self::Requeue
72        )
73    }
74}
75
76/// Snapshot of one ring slot.
77#[derive(Debug, Clone, PartialEq, Eq)]
78pub struct RingSlotSnapshot {
79    /// Zero-based slot index.
80    pub slot_idx: u32,
81    /// Current state.
82    pub status: RingStatus,
83    /// Tenant id assigned to the slot.
84    pub tenant_id: u32,
85    /// Top-level opcode currently stored in the slot.
86    pub opcode: u32,
87    /// First three argument words, useful for quick debugging.
88    pub args_prefix: [u32; 3],
89}
90
91/// Aggregated telemetry for one ticketed route window.
92#[derive(Debug, Clone, PartialEq, Eq)]
93pub struct WindowTelemetry {
94    /// Stable ticket id encoded in `arg0`.
95    pub ticket: u32,
96    /// Tenant id shared by all emitted slots in this window.
97    pub tenant_id: u32,
98    /// Opcode shared by the window payload slots.
99    pub opcode: u32,
100    /// Number of required slots in the window.
101    pub required_slots: u32,
102    /// Number of lookahead slots in the window.
103    pub lookahead_slots: u32,
104    /// Number of slots currently published.
105    pub published: u32,
106    /// Number of slots currently claimed.
107    pub claimed: u32,
108    /// Number of slots completed.
109    pub done: u32,
110    /// Number of slots waiting for I/O.
111    pub wait_io: u32,
112    /// Number of yielded slots.
113    pub yield_count: u32,
114    /// Number of requeued slots.
115    pub requeue: u32,
116    /// Number of faulted slots.
117    pub fault: u32,
118}
119
120impl WindowTelemetry {
121    /// Whether this ticket still has unfinished work in the ring.
122    #[must_use]
123    pub const fn is_active(&self) -> bool {
124        self.published > 0
125            || self.claimed > 0
126            || self.wait_io > 0
127            || self.yield_count > 0
128            || self.requeue > 0
129    }
130}
131
132/// Slot occupancy counts across the ring.
133#[derive(Debug, Clone, Copy, PartialEq, Eq, Default)]
134pub struct RingOccupancy {
135    /// Number of empty slots.
136    pub empty: u32,
137    /// Number of published slots.
138    pub published: u32,
139    /// Number of claimed slots.
140    pub claimed: u32,
141    /// Number of done slots.
142    pub done: u32,
143    /// Number of slots waiting for IO.
144    pub wait_io: u32,
145    /// Number of slots yielded.
146    pub yield_count: u32,
147    /// Number of requeued slots.
148    pub requeue: u32,
149    /// Number of faulted slots.
150    pub fault: u32,
151    /// Number of slots with unrecognized raw status values.
152    pub unknown: u32,
153}
154
155impl RingOccupancy {
156    /// Total slots represented by this occupancy snapshot.
157    #[must_use]
158    pub fn total_slots(&self) -> u32 {
159        checked_status_sum(
160            [
161                self.empty,
162                self.published,
163                self.claimed,
164                self.done,
165                self.wait_io,
166                self.yield_count,
167                self.requeue,
168                self.fault,
169                self.unknown,
170            ],
171            "total ring slots",
172        )
173    }
174
175    /// Host-visible active queue depth: all non-empty slots that are not done.
176    #[must_use]
177    pub fn queue_depth(&self) -> u32 {
178        checked_status_sum(
179            [
180                self.published,
181                self.claimed,
182                self.wait_io,
183                self.yield_count,
184                self.requeue,
185                self.fault,
186                self.unknown,
187            ],
188            "ring queue depth",
189        )
190    }
191}
192
193
194/// Structured view of the control buffer.
195#[derive(Debug, Clone, PartialEq, Eq, Default)]
196pub struct ControlSnapshot {
197    /// Shutdown flag.
198    pub shutdown: bool,
199    /// Total drained slots.
200    pub done_count: u32,
201    /// Epoch value (batch fences).
202    pub epoch: u32,
203    /// Non-zero opcode metrics.
204    pub metrics: Vec<(u32, u32)>,
205    /// Per-tenant fairness counters (cumulative).
206    pub tenant_fairness: Vec<u32>,
207    /// Per-priority fairness counters (cumulative).
208    pub priority_fairness: Vec<u32>,
209}
210
211/// Aggregated runtime performance counters derived from one telemetry snapshot.
212#[derive(Debug, Clone, Copy, PartialEq, Eq)]
213pub struct ResidentRuntimeCounters {
214    /// Total ring slots represented by the snapshot.
215    pub total_slots: u32,
216    /// Active queue depth: published/claimed/waiting/requeued/fault/unknown slots.
217    pub queue_depth: u32,
218    /// Empty ring slots, used as the host-visible idle-capacity signal.
219    pub gpu_idle_slots: u32,
220    /// Idle slots in parts per million of the ring size.
221    pub gpu_idle_ppm: u32,
222    /// Active frontier density in basis points of the ring size.
223    pub frontier_density_bps: u16,
224    /// Occupancy proxy in basis points: non-idle slots divided by total slots.
225    pub occupancy_proxy_bps: u16,
226    /// Total slots the GPU has drained according to the control buffer.
227    pub drained_slots: u32,
228    /// Done slots visible in the ring snapshot and pending reclaim.
229    pub unreclaimed_done_slots: u32,
230    /// Sum of tenant fairness counters.
231    pub tenant_fairness_total: u64,
232    /// Max minus min non-zero tenant fairness counter.
233    pub tenant_fairness_skew: u32,
234    /// Sum of priority fairness counters.
235    pub priority_fairness_total: u64,
236    /// Requeued slots visible in the ring.
237    pub requeue_slots: u32,
238    /// Faulted slots visible in the ring.
239    pub fault_slots: u32,
240}
241
242/// Watchdog view computed from two host-visible telemetry snapshots.
243#[derive(Debug, Clone, Copy, PartialEq, Eq)]
244pub struct ResidentWatchdogSnapshot {
245    /// Increase in drained slots between the previous and current snapshot.
246    pub done_delta: u32,
247    /// Current active queue depth.
248    pub queue_depth: u32,
249    /// Current faulted slots.
250    pub fault_slots: u32,
251    /// Current requeued slots.
252    pub requeue_slots: u32,
253    /// Current idle slots in parts per million.
254    pub gpu_idle_ppm: u32,
255    /// True when work remains queued but no drain progress was observed.
256    pub suspected_stall: bool,
257}
258
259/// Combined host-visible telemetry for a megakernel run.
260#[derive(Debug, Clone, PartialEq, Eq, Default)]
261pub struct RingTelemetry {
262    /// Decoded control-buffer snapshot.
263    pub control: ControlSnapshot,
264    /// Occupancy summary.
265    pub occupancy: RingOccupancy,
266    /// All decoded slots.
267    pub slots: Vec<RingSlotSnapshot>,
268    /// Decoded ticketed windows for any caller-specified window opcodes.
269    pub windows: Vec<WindowTelemetry>,
270}
271
272impl RingTelemetry {
273    /// Build capacity evidence for a strict caller-owned telemetry decode.
274    #[must_use]
275    pub fn decode_capacity_evidence(
276        &self,
277        scratch: &TelemetryDecodeScratch,
278    ) -> TelemetryDecodeCapacityEvidence {
279        TelemetryDecodeCapacityEvidence {
280            schema_version: TELEMETRY_DECODE_CAPACITY_SCHEMA_VERSION,
281            decoded_slot_count: self.slots.len(),
282            slot_output_capacity: self.slots.capacity(),
283            decoded_window_count: self.windows.len(),
284            window_output_capacity: self.windows.capacity(),
285            window_opcode_scratch_capacity: scratch.window_opcodes.capacity(),
286            window_accumulator_scratch_capacity: scratch.windows.capacity(),
287            uses_caller_owned_scratch: true,
288        }
289    }
290}
291
292/// Caller-owned scratch for repeated megakernel telemetry decodes.
293///
294/// Long-running supervisors poll telemetry at high frequency. Reusing this
295/// scratch keeps each sample to straight-line buffer rewrites rather than
296/// per-poll map allocation.
297#[derive(Debug, Default)]
298pub struct TelemetryDecodeScratch {
299    pub(super) window_opcodes: Vec<u32>,
300    pub(super) windows: FxHashMap<(u32, u32), WindowAccumulator>,
301}
302
303impl TelemetryDecodeScratch {
304    /// Construct empty decode scratch.
305    #[must_use]
306    pub fn new() -> Self {
307        Self {
308            window_opcodes: Vec::new(),
309            windows: FxHashMap::default(),
310        }
311    }
312
313    /// Clear retained decode rows without releasing allocated scratch capacity.
314    pub fn clear(&mut self) {
315        self.window_opcodes.clear();
316        self.windows.clear();
317    }
318}
319
320#[derive(Debug, Clone, Copy, PartialEq, Eq, Default)]
321pub(super) struct WindowAccumulator {
322    pub(super) tenant_id: u32,
323    pub(super) opcode: u32,
324    pub(super) required_slots: u32,
325    pub(super) lookahead_slots: u32,
326    pub(super) published: u32,
327    pub(super) claimed: u32,
328    pub(super) done: u32,
329    pub(super) wait_io: u32,
330    pub(super) yield_count: u32,
331    pub(super) requeue: u32,
332    pub(super) fault: u32,
333}
334
335pub(super) fn checked_status_sum<const N: usize>(values: [u32; N], label: &'static str) -> u32 {
336    let _ = label;
337    values
338        .into_iter()
339        .fold(0_u32, |acc, value| acc.saturating_add(value))
340}
vyre_runtime/megakernel/telemetry/ring_state.rs

vyre_runtime/megakernel/telemetry/
ring_state.rs
