vyre_runtime/megakernel/telemetry/
types.rs

1use super::slot;
2use rustc_hash::FxHashMap;
3
4/// Decoded top-level ring slot state.
5#[derive(Debug, Clone, Copy, PartialEq, Eq)]
6pub enum RingStatus {
7    /// Slot is free.
8    Empty,
9    /// Slot is published and waiting for a worker.
10    Published,
11    /// Slot has been claimed by a worker.
12    Claimed,
13    /// Slot completed and can be recycled.
14    Done,
15    /// Slot is waiting for an asynchronous IO continuation.
16    WaitIo,
17    /// Slot yielded execution back to the scheduler.
18    Yield,
19    /// Slot is heavily contested and has been requeued.
20    Requeue,
21    /// Slot hit a hardware or software fault constraint.
22    Fault,
23    /// Unknown raw wire value.
24    Unknown(u32),
25}
26
27impl RingStatus {
28    #[must_use]
29    pub(super) fn from_raw(raw: u32) -> Self {
30        match raw {
31            slot::EMPTY => Self::Empty,
32            slot::PUBLISHED => Self::Published,
33            slot::CLAIMED => Self::Claimed,
34            slot::DONE => Self::Done,
35            slot::WAIT_IO => Self::WaitIo,
36            slot::YIELD => Self::Yield,
37            slot::REQUEUE => Self::Requeue,
38            slot::FAULT => Self::Fault,
39            other => Self::Unknown(other),
40        }
41    }
42
43    /// Raw wire discriminant for sketching, replay, and compact telemetry.
44    #[must_use]
45    pub const fn raw(self) -> u32 {
46        match self {
47            Self::Empty => slot::EMPTY,
48            Self::Published => slot::PUBLISHED,
49            Self::Claimed => slot::CLAIMED,
50            Self::Done => slot::DONE,
51            Self::WaitIo => slot::WAIT_IO,
52            Self::Yield => slot::YIELD,
53            Self::Requeue => slot::REQUEUE,
54            Self::Fault => slot::FAULT,
55            Self::Unknown(raw) => raw,
56        }
57    }
58
59    /// Whether this status still represents in-flight work rather than a
60    /// terminal slot outcome.
61    #[must_use]
62    pub const fn is_active(self) -> bool {
63        matches!(
64            self,
65            Self::Published | Self::Claimed | Self::WaitIo | Self::Yield | Self::Requeue
66        )
67    }
68}
69
70/// Snapshot of one ring slot.
71#[derive(Debug, Clone, PartialEq, Eq)]
72pub struct RingSlotSnapshot {
73    /// Zero-based slot index.
74    pub slot_idx: u32,
75    /// Current state.
76    pub status: RingStatus,
77    /// Tenant id assigned to the slot.
78    pub tenant_id: u32,
79    /// Top-level opcode currently stored in the slot.
80    pub opcode: u32,
81    /// First three argument words, useful for quick debugging.
82    pub args_prefix: [u32; 3],
83}
84
85/// Aggregated telemetry for one ticketed route window.
86#[derive(Debug, Clone, PartialEq, Eq)]
87pub struct WindowTelemetry {
88    /// Stable ticket id encoded in `arg0`.
89    pub ticket: u32,
90    /// Tenant id shared by all emitted slots in this window.
91    pub tenant_id: u32,
92    /// Opcode shared by the window payload slots.
93    pub opcode: u32,
94    /// Number of required slots in the window.
95    pub required_slots: u32,
96    /// Number of lookahead slots in the window.
97    pub lookahead_slots: u32,
98    /// Number of slots currently published.
99    pub published: u32,
100    /// Number of slots currently claimed.
101    pub claimed: u32,
102    /// Number of slots completed.
103    pub done: u32,
104    /// Number of slots waiting for I/O.
105    pub wait_io: u32,
106    /// Number of yielded slots.
107    pub yield_count: u32,
108    /// Number of requeued slots.
109    pub requeue: u32,
110    /// Number of faulted slots.
111    pub fault: u32,
112}
113
114impl WindowTelemetry {
115    /// Whether this ticket still has unfinished work in the ring.
116    #[must_use]
117    pub const fn is_active(&self) -> bool {
118        self.published > 0
119            || self.claimed > 0
120            || self.wait_io > 0
121            || self.yield_count > 0
122            || self.requeue > 0
123    }
124}
125
126/// Slot occupancy counts across the ring.
127#[derive(Debug, Clone, Copy, PartialEq, Eq, Default)]
128pub struct RingOccupancy {
129    /// Number of empty slots.
130    pub empty: u32,
131    /// Number of published slots.
132    pub published: u32,
133    /// Number of claimed slots.
134    pub claimed: u32,
135    /// Number of done slots.
136    pub done: u32,
137    /// Number of slots waiting for IO.
138    pub wait_io: u32,
139    /// Number of slots yielded.
140    pub yield_count: u32,
141    /// Number of requeued slots.
142    pub requeue: u32,
143    /// Number of faulted slots.
144    pub fault: u32,
145    /// Number of slots with unrecognized raw status values.
146    pub unknown: u32,
147}
148
149impl RingOccupancy {
150    /// Total slots represented by this occupancy snapshot.
151    #[must_use]
152    pub fn total_slots(&self) -> u32 {
153        checked_status_sum(
154            [
155                self.empty,
156                self.published,
157                self.claimed,
158                self.done,
159                self.wait_io,
160                self.yield_count,
161                self.requeue,
162                self.fault,
163                self.unknown,
164            ],
165            "total ring slots",
166        )
167    }
168
169    /// Host-visible active queue depth: all non-empty slots that are not done.
170    #[must_use]
171    pub fn queue_depth(&self) -> u32 {
172        checked_status_sum(
173            [
174                self.published,
175                self.claimed,
176                self.wait_io,
177                self.yield_count,
178                self.requeue,
179                self.fault,
180                self.unknown,
181            ],
182            "ring queue depth",
183        )
184    }
185}
186
187/// Schema version for IO/runtime evidence emitted from megakernel telemetry.
188pub const RUNTIME_IO_EVIDENCE_SCHEMA_VERSION: u32 = 1;
189
190/// Required metric families for runtime IO/residency evidence.
191#[derive(Debug, Clone, Copy, PartialEq, Eq)]
192pub enum RuntimeEvidenceMetricFamily {
193    /// Ring occupancy metrics are present.
194    Ring,
195    /// Control-buffer decode metrics are present.
196    Control,
197    /// Host/device copy accounting metrics are present.
198    Copy,
199    /// Resident device-byte metrics are present.
200    Residency,
201}
202
203impl RuntimeEvidenceMetricFamily {
204    /// Stable evidence-family token.
205    #[must_use]
206    pub const fn as_str(self) -> &'static str {
207        match self {
208            Self::Ring => "ring",
209            Self::Control => "control",
210            Self::Copy => "copy",
211            Self::Residency => "residency",
212        }
213    }
214}
215
216/// Coverage bits for the required runtime evidence metric families.
217#[derive(Debug, Clone, Copy, PartialEq, Eq, Default)]
218pub struct RuntimeEvidenceMetricCoverage {
219    /// Ring occupancy metrics are present.
220    pub ring: bool,
221    /// Control-buffer decode metrics are present.
222    pub control: bool,
223    /// Host/device copy accounting metrics are present.
224    pub copy: bool,
225    /// Resident device-byte metrics are present.
226    pub residency: bool,
227}
228
229impl RuntimeEvidenceMetricCoverage {
230    /// Coverage with every runtime evidence family present.
231    #[must_use]
232    pub const fn complete() -> Self {
233        Self {
234            ring: true,
235            control: true,
236            copy: true,
237            residency: true,
238        }
239    }
240
241    /// Missing required metric families.
242    #[must_use]
243    pub fn missing_families(self) -> Vec<RuntimeEvidenceMetricFamily> {
244        let mut missing = Vec::new();
245        if !self.ring {
246            missing.push(RuntimeEvidenceMetricFamily::Ring);
247        }
248        if !self.control {
249            missing.push(RuntimeEvidenceMetricFamily::Control);
250        }
251        if !self.copy {
252            missing.push(RuntimeEvidenceMetricFamily::Copy);
253        }
254        if !self.residency {
255            missing.push(RuntimeEvidenceMetricFamily::Residency);
256        }
257        missing
258    }
259}
260
261/// Runtime-owned IO/residency evidence envelope for release and benchmark artifacts.
262#[derive(Debug, Clone, PartialEq, Eq)]
263pub struct MegakernelRuntimeEvidence {
264    /// Runtime evidence schema version.
265    pub schema_version: u32,
266    /// Device-resident bytes retained by the dispatch family.
267    pub resident_device_bytes: u64,
268    /// Host-visible copy bytes still required for this evidence sample.
269    pub host_copy_bytes: u64,
270    /// Host copy bytes avoided by resident handles or device-side IO.
271    pub host_copy_avoided_bytes: u64,
272    /// Ring occupancy snapshot.
273    pub ring_occupancy: RingOccupancy,
274    /// Control-buffer decode cost in nanoseconds.
275    pub control_decode_ns: u64,
276    /// Ring decode cost in nanoseconds.
277    pub ring_decode_ns: u64,
278    /// Required metric-family coverage.
279    pub coverage: RuntimeEvidenceMetricCoverage,
280}
281
282impl MegakernelRuntimeEvidence {
283    /// Construct a complete runtime evidence envelope.
284    #[must_use]
285    pub const fn complete(
286        resident_device_bytes: u64,
287        host_copy_bytes: u64,
288        host_copy_avoided_bytes: u64,
289        ring_occupancy: RingOccupancy,
290        control_decode_ns: u64,
291        ring_decode_ns: u64,
292    ) -> Self {
293        Self {
294            schema_version: RUNTIME_IO_EVIDENCE_SCHEMA_VERSION,
295            resident_device_bytes,
296            host_copy_bytes,
297            host_copy_avoided_bytes,
298            ring_occupancy,
299            control_decode_ns,
300            ring_decode_ns,
301            coverage: RuntimeEvidenceMetricCoverage::complete(),
302        }
303    }
304
305    /// Metric families missing from this evidence envelope.
306    #[must_use]
307    pub fn missing_metric_families(&self) -> Vec<RuntimeEvidenceMetricFamily> {
308        self.coverage.missing_families()
309    }
310
311    /// Whether all required runtime evidence families are present.
312    #[must_use]
313    pub fn is_complete(&self) -> bool {
314        self.schema_version == RUNTIME_IO_EVIDENCE_SCHEMA_VERSION
315            && self.missing_metric_families().is_empty()
316    }
317
318    /// Avoided host-copy bytes in basis points of total relevant copy volume.
319    #[must_use]
320    pub fn host_copy_avoidance_bps(&self) -> u16 {
321        let total = u128::from(self.host_copy_bytes)
322            .saturating_add(u128::from(self.host_copy_avoided_bytes));
323        if total == 0 {
324            return 0;
325        }
326        let bps = u128::from(self.host_copy_avoided_bytes).saturating_mul(10_000) / total;
327        bps.min(10_000) as u16
328    }
329}
330
331fn checked_status_sum<const N: usize>(values: [u32; N], label: &'static str) -> u32 {
332    let _ = label;
333    values
334        .into_iter()
335        .fold(0_u32, |acc, value| acc.saturating_add(value))
336}
337
338#[cfg(test)]
339mod evidence_tests {
340    use super::*;
341
342    #[test]
343    fn runtime_evidence_reports_missing_metric_families() {
344        let evidence = MegakernelRuntimeEvidence {
345            schema_version: RUNTIME_IO_EVIDENCE_SCHEMA_VERSION,
346            resident_device_bytes: 0,
347            host_copy_bytes: 0,
348            host_copy_avoided_bytes: 0,
349            ring_occupancy: RingOccupancy::default(),
350            control_decode_ns: 0,
351            ring_decode_ns: 0,
352            coverage: RuntimeEvidenceMetricCoverage {
353                ring: true,
354                control: false,
355                copy: true,
356                residency: false,
357            },
358        };
359
360        let missing = evidence
361            .missing_metric_families()
362            .into_iter()
363            .map(RuntimeEvidenceMetricFamily::as_str)
364            .collect::<Vec<_>>();
365
366        assert_eq!(missing, vec!["control", "residency"]);
367        assert!(!evidence.is_complete());
368    }
369
370    #[test]
371    fn runtime_evidence_records_copy_avoidance_and_occupancy() {
372        let evidence = MegakernelRuntimeEvidence::complete(
373            4096,
374            1024,
375            3072,
376            RingOccupancy {
377                empty: 1,
378                published: 2,
379                claimed: 3,
380                done: 4,
381                wait_io: 5,
382                yield_count: 6,
383                requeue: 7,
384                fault: 8,
385                unknown: 9,
386            },
387            11,
388            13,
389        );
390
391        assert!(evidence.is_complete());
392        assert_eq!(evidence.resident_device_bytes, 4096);
393        assert_eq!(evidence.ring_occupancy.total_slots(), 45);
394        assert_eq!(evidence.ring_occupancy.queue_depth(), 40);
395        assert_eq!(evidence.host_copy_avoidance_bps(), 7500);
396    }
397}
398
399/// Structured view of the control buffer.
400#[derive(Debug, Clone, PartialEq, Eq, Default)]
401pub struct ControlSnapshot {
402    /// Shutdown flag.
403    pub shutdown: bool,
404    /// Total drained slots.
405    pub done_count: u32,
406    /// Epoch value (batch fences).
407    pub epoch: u32,
408    /// Non-zero opcode metrics.
409    pub metrics: Vec<(u32, u32)>,
410    /// Per-tenant fairness counters (cumulative).
411    pub tenant_fairness: Vec<u32>,
412    /// Per-priority fairness counters (cumulative).
413    pub priority_fairness: Vec<u32>,
414}
415
416/// Aggregated runtime performance counters derived from one telemetry snapshot.
417#[derive(Debug, Clone, Copy, PartialEq, Eq)]
418pub struct MegakernelRuntimeCounters {
419    /// Total ring slots represented by the snapshot.
420    pub total_slots: u32,
421    /// Active queue depth: published/claimed/waiting/requeued/fault/unknown slots.
422    pub queue_depth: u32,
423    /// Empty ring slots, used as the host-visible idle-capacity signal.
424    pub gpu_idle_slots: u32,
425    /// Idle slots in parts per million of the ring size.
426    pub gpu_idle_ppm: u32,
427    /// Active frontier density in basis points of the ring size.
428    pub frontier_density_bps: u16,
429    /// Occupancy proxy in basis points: non-idle slots divided by total slots.
430    pub occupancy_proxy_bps: u16,
431    /// Total slots the GPU has drained according to the control buffer.
432    pub drained_slots: u32,
433    /// Done slots visible in the ring snapshot and pending reclaim.
434    pub unreclaimed_done_slots: u32,
435    /// Sum of tenant fairness counters.
436    pub tenant_fairness_total: u64,
437    /// Max minus min non-zero tenant fairness counter.
438    pub tenant_fairness_skew: u32,
439    /// Sum of priority fairness counters.
440    pub priority_fairness_total: u64,
441    /// Requeued slots visible in the ring.
442    pub requeue_slots: u32,
443    /// Faulted slots visible in the ring.
444    pub fault_slots: u32,
445}
446
447/// Watchdog view computed from two host-visible telemetry snapshots.
448#[derive(Debug, Clone, Copy, PartialEq, Eq)]
449pub struct MegakernelWatchdogSnapshot {
450    /// Increase in drained slots between the previous and current snapshot.
451    pub done_delta: u32,
452    /// Current active queue depth.
453    pub queue_depth: u32,
454    /// Current faulted slots.
455    pub fault_slots: u32,
456    /// Current requeued slots.
457    pub requeue_slots: u32,
458    /// Current idle slots in parts per million.
459    pub gpu_idle_ppm: u32,
460    /// True when work remains queued but no drain progress was observed.
461    pub suspected_stall: bool,
462}
463
464/// Combined host-visible telemetry for a megakernel run.
465#[derive(Debug, Clone, PartialEq, Eq, Default)]
466pub struct RingTelemetry {
467    /// Decoded control-buffer snapshot.
468    pub control: ControlSnapshot,
469    /// Occupancy summary.
470    pub occupancy: RingOccupancy,
471    /// All decoded slots.
472    pub slots: Vec<RingSlotSnapshot>,
473    /// Decoded ticketed windows for any caller-specified window opcodes.
474    pub windows: Vec<WindowTelemetry>,
475}
476
477/// Schema version for telemetry decode capacity evidence.
478pub const TELEMETRY_DECODE_CAPACITY_SCHEMA_VERSION: u32 = 1;
479
480/// Evidence that a telemetry decode used caller-owned output and scratch buffers.
481#[derive(Debug, Clone, Copy, PartialEq, Eq)]
482pub struct TelemetryDecodeCapacityEvidence {
483    /// Evidence schema version.
484    pub schema_version: u32,
485    /// Number of decoded ring slots in the output snapshot.
486    pub decoded_slot_count: usize,
487    /// Capacity of the caller-owned ring-slot output buffer.
488    pub slot_output_capacity: usize,
489    /// Number of decoded route-window rows in the output snapshot.
490    pub decoded_window_count: usize,
491    /// Capacity of the caller-owned route-window output buffer.
492    pub window_output_capacity: usize,
493    /// Capacity retained for sorted window-opcode scratch.
494    pub window_opcode_scratch_capacity: usize,
495    /// Capacity retained for route-window accumulator scratch.
496    pub window_accumulator_scratch_capacity: usize,
497    /// True when evidence was produced from caller-owned scratch.
498    pub uses_caller_owned_scratch: bool,
499}
500
501impl TelemetryDecodeCapacityEvidence {
502    /// Return true when output and scratch capacities cover the decoded rows.
503    #[must_use]
504    pub fn is_complete(self) -> bool {
505        self.schema_version == TELEMETRY_DECODE_CAPACITY_SCHEMA_VERSION
506            && self.uses_caller_owned_scratch
507            && self.slot_output_capacity >= self.decoded_slot_count
508            && self.window_output_capacity >= self.decoded_window_count
509            && self.window_accumulator_scratch_capacity >= self.decoded_window_count
510    }
511}
512
513impl RingTelemetry {
514    /// Build capacity evidence for a strict caller-owned telemetry decode.
515    #[must_use]
516    pub fn decode_capacity_evidence(
517        &self,
518        scratch: &TelemetryDecodeScratch,
519    ) -> TelemetryDecodeCapacityEvidence {
520        TelemetryDecodeCapacityEvidence {
521            schema_version: TELEMETRY_DECODE_CAPACITY_SCHEMA_VERSION,
522            decoded_slot_count: self.slots.len(),
523            slot_output_capacity: self.slots.capacity(),
524            decoded_window_count: self.windows.len(),
525            window_output_capacity: self.windows.capacity(),
526            window_opcode_scratch_capacity: scratch.window_opcodes.capacity(),
527            window_accumulator_scratch_capacity: scratch.windows.capacity(),
528            uses_caller_owned_scratch: true,
529        }
530    }
531}
532
533/// Caller-owned scratch for repeated megakernel telemetry decodes.
534///
535/// Long-running supervisors poll telemetry at high frequency. Reusing this
536/// scratch keeps each sample to straight-line buffer rewrites rather than
537/// per-poll map allocation.
538#[derive(Debug, Default)]
539pub struct TelemetryDecodeScratch {
540    pub(super) window_opcodes: Vec<u32>,
541    pub(super) windows: FxHashMap<(u32, u32), WindowAccumulator>,
542}
543
544impl TelemetryDecodeScratch {
545    /// Construct empty decode scratch.
546    #[must_use]
547    pub fn new() -> Self {
548        Self {
549            window_opcodes: Vec::new(),
550            windows: FxHashMap::default(),
551        }
552    }
553
554    /// Clear retained decode rows without releasing allocated scratch capacity.
555    pub fn clear(&mut self) {
556        self.window_opcodes.clear();
557        self.windows.clear();
558    }
559}
560
561#[derive(Debug, Clone, Copy, PartialEq, Eq, Default)]
562pub(super) struct WindowAccumulator {
563    pub(super) tenant_id: u32,
564    pub(super) opcode: u32,
565    pub(super) required_slots: u32,
566    pub(super) lookahead_slots: u32,
567    pub(super) published: u32,
568    pub(super) claimed: u32,
569    pub(super) done: u32,
570    pub(super) wait_io: u32,
571    pub(super) yield_count: u32,
572    pub(super) requeue: u32,
573    pub(super) fault: u32,
574}
vyre_runtime/megakernel/telemetry/types.rs

vyre_runtime/megakernel/telemetry/
types.rs
