vyre_runtime/megakernel/protocol/
codec.rs

1use super::{
2    control, debug, slot, DebugRecord, ProtocolError, CONTROL_MIN_WORDS, MAX_DEBUG_RECORDS,
3    MAX_ENCODED_DEBUG_RECORDS, MAX_ENCODED_OBSERVABLE_SLOTS, MAX_ENCODED_RING_SLOTS,
4    MAX_OBSERVABLE_SLOTS, MAX_RING_SLOTS, SLOT_WORDS, STATUS_WORD,
5};
6
7/// Return the number of bytes required by a control buffer with `observable_slots`.
8#[must_use]
9pub fn control_byte_len(observable_slots: u32) -> Option<usize> {
10    if observable_slots > MAX_OBSERVABLE_SLOTS {
11        return None;
12    }
13    let words = control::OBSERVABLE_BASE.checked_add(observable_slots)?;
14    words_to_bytes(words.max(CONTROL_MIN_WORDS))
15}
16
17/// Return the number of bytes required by a ring buffer with `slot_count` slots.
18#[must_use]
19pub fn ring_byte_len(slot_count: u32) -> Option<usize> {
20    if slot_count > MAX_RING_SLOTS {
21        return None;
22    }
23    let words = slot_count.checked_mul(SLOT_WORDS)?;
24    words_to_bytes(words)
25}
26
27/// Return the number of bytes required by a debug-log buffer.
28#[must_use]
29pub fn debug_log_byte_len(record_capacity: u32) -> Option<usize> {
30    if record_capacity > MAX_DEBUG_RECORDS {
31        return None;
32    }
33    let record_words = record_capacity.checked_mul(debug::RECORD_WORDS)?;
34    let words = debug::RECORDS_BASE.checked_add(record_words)?;
35    words_to_bytes(words)
36}
37
38fn control_encode_capacity(observable_slots: u32) -> Result<usize, ProtocolError> {
39    if observable_slots > MAX_ENCODED_OBSERVABLE_SLOTS {
40        return Err(ProtocolError::ByteLengthOverflow {
41            buffer: "control",
42            fix: "shard observable results or reduce observable_slots to the megakernel allocation cap before encoding control",
43        });
44    }
45    control_byte_len(observable_slots).ok_or(ProtocolError::ByteLengthOverflow {
46        buffer: "control",
47        fix: "shard observable results or reduce observable_slots to the megakernel protocol cap before encoding control",
48    })
49}
50
51fn ring_encode_capacity(slot_count: u32) -> Result<usize, ProtocolError> {
52    if slot_count > MAX_ENCODED_RING_SLOTS {
53        return Err(ProtocolError::ByteLengthOverflow {
54            buffer: "ring",
55            fix: "split the dispatch into smaller ring shards before encoding; slot_count exceeds the megakernel allocation cap or host address space",
56        });
57    }
58    ring_byte_len(slot_count).ok_or(ProtocolError::ByteLengthOverflow {
59        buffer: "ring",
60        fix: "split the dispatch into smaller ring shards before encoding; slot_count exceeds the megakernel protocol cap or host address space",
61    })
62}
63
64fn debug_log_encode_capacity(record_capacity: u32) -> Result<usize, ProtocolError> {
65    if record_capacity > MAX_ENCODED_DEBUG_RECORDS {
66        return Err(ProtocolError::ByteLengthOverflow {
67            buffer: "debug_log",
68            fix:
69                "reduce debug-log record_capacity to the megakernel allocation cap before encoding",
70        });
71    }
72    debug_log_byte_len(record_capacity).ok_or(ProtocolError::ByteLengthOverflow {
73        buffer: "debug_log",
74        fix: "reduce debug-log record_capacity to the megakernel protocol cap before encoding",
75    })
76}
77
78/// Encode a control-buffer payload.
79///
80/// # Errors
81///
82/// Returns [`ProtocolError`] when the requested observable region overflows
83/// host address space.
84pub fn encode_control(
85    shutdown: bool,
86    tenant_count: u32,
87    observable_slots: u32,
88) -> Result<Vec<u8>, ProtocolError> {
89    try_encode_control(shutdown, tenant_count, observable_slots)
90}
91
92/// Strictly encode a control-buffer payload.
93///
94/// # Errors
95///
96/// Returns [`ProtocolError`] when the requested observable region overflows
97/// host address space.
98pub fn try_encode_control(
99    shutdown: bool,
100    tenant_count: u32,
101    observable_slots: u32,
102) -> Result<Vec<u8>, ProtocolError> {
103    let total_bytes = control_encode_capacity(observable_slots)?;
104    let mut bytes = Vec::new();
105    try_reserve_protocol_capacity(
106        &mut bytes,
107        total_bytes,
108        "control",
109        "control encode could not reserve host staging bytes; reduce observable_slots or reuse a preallocated control buffer",
110    )?;
111    try_encode_control_into(shutdown, tenant_count, observable_slots, &mut bytes)?;
112    Ok(bytes)
113}
114
115/// Strictly encode a control-buffer payload into caller-owned storage.
116///
117/// Clears and resizes `dst` to the exact control-buffer byte length, reusing
118/// any existing allocation.
119///
120/// # Errors
121///
122/// Returns [`ProtocolError`] when the requested observable region overflows
123/// host address space.
124pub fn try_encode_control_into(
125    shutdown: bool,
126    tenant_count: u32,
127    observable_slots: u32,
128    dst: &mut Vec<u8>,
129) -> Result<(), ProtocolError> {
130    let total_bytes = control_encode_capacity(observable_slots)?;
131    dst.clear();
132    try_reserve_protocol_capacity(
133        dst,
134        total_bytes,
135        "control",
136        "control encode could not reserve caller-owned staging bytes; reduce observable_slots or reuse a larger control buffer",
137    )?;
138    dst.resize(total_bytes, 0);
139
140    if shutdown {
141        write_word(
142            dst,
143            control_word_index(control::SHUTDOWN, "shutdown word")?,
144            1,
145        );
146    }
147    write_word(
148        dst,
149        control_word_index(control::TENANT_BASE, "tenant base word")?,
150        control::TENANT_BASE + 1,
151    );
152
153    let tenant_table_start = control_word_index(control::TENANT_BASE, "tenant base word")?
154        .checked_add(1)
155        .ok_or(ProtocolError::ByteLengthOverflow {
156            buffer: "control",
157            fix: "tenant table start overflowed usize; reduce control protocol constants",
158        })?;
159    let requested_tenant_words =
160        usize::try_from(tenant_count).map_err(|_| ProtocolError::ByteLengthOverflow {
161            buffer: "control",
162            fix: "tenant_count cannot fit host usize; split tenant tables before encoding",
163        })?;
164    let tenant_table_end = core::cmp::min(
165        tenant_table_start
166            .checked_add(requested_tenant_words)
167            .ok_or(ProtocolError::ByteLengthOverflow {
168                buffer: "control",
169                fix: "tenant table end overflowed usize; split tenant tables before encoding",
170            })?,
171        control_word_index(control::TENANT_QUOTA_BASE, "tenant quota base word")?,
172    );
173    for word_idx in tenant_table_start..tenant_table_end {
174        write_word(dst, word_idx, !0u32);
175    }
176
177    let quota_table_start =
178        control_word_index(control::TENANT_QUOTA_BASE, "tenant quota base word")?;
179    let quota_table_end = core::cmp::min(
180        quota_table_start
181            .checked_add(requested_tenant_words)
182            .ok_or(ProtocolError::ByteLengthOverflow {
183                buffer: "control",
184                fix: "quota table end overflowed usize; split tenant tables before encoding",
185            })?,
186        control_word_index(control::TENANT_FAIRNESS_BASE, "tenant fairness base word")?,
187    );
188    for word_idx in quota_table_start..quota_table_end {
189        write_word(dst, word_idx, 1_000_000);
190    }
191    Ok(())
192}
193
194/// Encode an empty ring buffer with `slot_count` slots.
195///
196/// # Errors
197///
198/// Returns [`ProtocolError`] when the requested ring size overflows host
199/// address space.
200pub fn encode_empty_ring(slot_count: u32) -> Result<Vec<u8>, ProtocolError> {
201    try_encode_empty_ring(slot_count)
202}
203
204/// Strictly encode an empty ring buffer with `slot_count` slots.
205///
206/// # Errors
207///
208/// Returns [`ProtocolError`] when the requested ring size overflows host
209/// address space.
210pub fn try_encode_empty_ring(slot_count: u32) -> Result<Vec<u8>, ProtocolError> {
211    let total_bytes = ring_encode_capacity(slot_count)?;
212    let mut bytes = Vec::new();
213    try_reserve_protocol_capacity(
214        &mut bytes,
215        total_bytes,
216        "ring",
217        "ring encode could not reserve host staging bytes; split the dispatch into smaller ring shards or reuse a preallocated ring buffer",
218    )?;
219    try_encode_empty_ring_into(slot_count, &mut bytes)?;
220    Ok(bytes)
221}
222
223/// Strictly encode an empty ring buffer into caller-owned storage.
224///
225/// Clears and resizes `dst` to the exact ring byte length, reusing allocation.
226///
227/// # Errors
228///
229/// Returns [`ProtocolError`] when the requested ring size overflows host
230/// address space.
231pub fn try_encode_empty_ring_into(slot_count: u32, dst: &mut Vec<u8>) -> Result<(), ProtocolError> {
232    let total_bytes = ring_encode_capacity(slot_count)?;
233    dst.clear();
234    try_reserve_protocol_capacity(
235        dst,
236        total_bytes,
237        "ring",
238        "ring encode could not reserve caller-owned staging bytes; split the dispatch into smaller ring shards or reuse a larger ring buffer",
239    )?;
240    dst.resize(total_bytes, 0);
241    Ok(())
242}
243
244/// Encode an empty PRINTF channel buffer.
245///
246/// # Errors
247///
248/// Returns [`ProtocolError`] when the requested debug-log size overflows host
249/// address space.
250pub fn encode_empty_debug_log(record_capacity: u32) -> Result<Vec<u8>, ProtocolError> {
251    try_encode_empty_debug_log(record_capacity)
252}
253
254/// Strictly encode an empty PRINTF channel buffer.
255///
256/// # Errors
257///
258/// Returns [`ProtocolError`] when the requested debug-log size overflows host
259/// address space.
260pub fn try_encode_empty_debug_log(record_capacity: u32) -> Result<Vec<u8>, ProtocolError> {
261    let total_bytes = debug_log_encode_capacity(record_capacity)?;
262    let mut bytes = Vec::new();
263    try_reserve_protocol_capacity(
264        &mut bytes,
265        total_bytes,
266        "debug_log",
267        "debug-log encode could not reserve host staging bytes; reduce record_capacity or reuse a preallocated debug-log buffer",
268    )?;
269    try_encode_empty_debug_log_into(record_capacity, &mut bytes)?;
270    Ok(bytes)
271}
272
273/// Strictly encode an empty PRINTF channel buffer into caller-owned storage.
274///
275/// Clears and resizes `dst` to the exact debug-log byte length, reusing allocation.
276///
277/// # Errors
278///
279/// Returns [`ProtocolError`] when the requested debug-log size overflows host
280/// address space.
281pub fn try_encode_empty_debug_log_into(
282    record_capacity: u32,
283    dst: &mut Vec<u8>,
284) -> Result<(), ProtocolError> {
285    let total_bytes = debug_log_encode_capacity(record_capacity)?;
286    dst.clear();
287    try_reserve_protocol_capacity(
288        dst,
289        total_bytes,
290        "debug_log",
291        "debug-log encode could not reserve caller-owned staging bytes; reduce record_capacity or reuse a larger debug-log buffer",
292    )?;
293    dst.resize(total_bytes, 0);
294    Ok(())
295}
296
297/// Decode the kernel's `done_count` from a control buffer.
298///
299/// On a malformed or truncated control buffer this returns `0` **and emits a
300/// `tracing::error!`** so the stall is operator-visible. Prefer
301/// [`try_read_done_count`] in new code, a silent `0` is indistinguishable from
302/// the initial kernel state and can cause the host pump to stall indefinitely.
303#[must_use]
304pub fn read_done_count(control_bytes: &[u8]) -> u32 {
305    match try_read_done_count(control_bytes) {
306        Ok(v) => v,
307        Err(e) => {
308            tracing::error!(
309                error = %e,
310                buf_len = control_bytes.len(),
311                "read_done_count: malformed control buffer, returning 0 instead of real count. Fix: ensure the DMA readback covers the full control buffer produced by the matching encoder."
312            );
313            0
314        }
315    }
316}
317
318/// Read the epoch counter from a control buffer.
319///
320/// On a malformed or truncated control buffer this returns `0` **and emits a
321/// `tracing::error!`** so the stall is operator-visible. Prefer
322/// [`try_read_epoch`] in new code, a silent `0` lets the host pump conclude no
323/// batch has completed, stalling dispatch indefinitely.
324#[must_use]
325pub fn read_epoch(control_bytes: &[u8]) -> u32 {
326    match try_read_epoch(control_bytes) {
327        Ok(v) => v,
328        Err(e) => {
329            tracing::error!(
330                error = %e,
331                buf_len = control_bytes.len(),
332                "read_epoch: malformed control buffer, returning 0 instead of real epoch. Fix: ensure the DMA readback covers the full control buffer produced by the matching encoder."
333            );
334            0
335        }
336    }
337}
338
339/// Strictly decode the kernel's `done_count` from a control buffer.
340///
341/// # Errors
342///
343/// Returns [`ProtocolError`] when the buffer is not word-aligned or is too
344/// short to contain the fixed control header.
345pub fn try_read_done_count(control_bytes: &[u8]) -> Result<u32, ProtocolError> {
346    read_required_word(
347        "control",
348        control_bytes,
349        control_word_index(control::DONE_COUNT, "done-count word")?,
350    )
351}
352
353/// Strictly decode the epoch counter from a control buffer.
354///
355/// # Errors
356///
357/// Returns [`ProtocolError`] when the buffer is not word-aligned or is too
358/// short to contain the epoch word.
359pub fn try_read_epoch(control_bytes: &[u8]) -> Result<u32, ProtocolError> {
360    read_required_word(
361        "control",
362        control_bytes,
363        control_word_index(control::EPOCH, "epoch word")?,
364    )
365}
366
367/// Read an observable result word from a control buffer.
368///
369/// On a malformed or truncated control buffer this returns `0` **and emits a
370/// `tracing::error!`** so the miss is operator-visible. Prefer
371/// [`try_read_observable`] in new code, a silent `0` is indistinguishable from
372/// an observable slot that was never written.
373#[must_use]
374pub fn read_observable(control_bytes: &[u8], index: u32) -> u32 {
375    match try_read_observable(control_bytes, index) {
376        Ok(v) => v,
377        Err(e) => {
378            tracing::error!(
379                error = %e,
380                buf_len = control_bytes.len(),
381                index,
382                "read_observable: malformed control buffer, returning 0 instead of real observable. Fix: ensure the DMA readback covers the full control buffer produced by the matching encoder."
383            );
384            0
385        }
386    }
387}
388
389/// Strictly read an observable result word from a control buffer.
390///
391/// # Errors
392///
393/// Returns [`ProtocolError`] when the buffer is not word-aligned, the index
394/// overflows the observable word offset, or the word is outside the buffer.
395pub fn try_read_observable(control_bytes: &[u8], index: u32) -> Result<u32, ProtocolError> {
396    let word_idx = control_word_index(
397        control::OBSERVABLE_BASE
398            .checked_add(index)
399            .ok_or(ProtocolError::ByteLengthOverflow {
400                buffer: "control",
401                fix: "observable index overflows the protocol word offset; shard observable reads",
402            })?,
403        "observable word index",
404    )?;
405    read_required_word("control", control_bytes, word_idx)
406}
407
408/// Read per-opcode metrics counters from a control buffer.
409#[must_use]
410pub fn read_metrics(control_bytes: &[u8]) -> Vec<(u32, u32)> {
411    let mut result = Vec::new();
412    read_metrics_into(control_bytes, &mut result);
413    result
414}
415
416/// Read per-opcode metrics counters into caller-owned storage.
417///
418/// Clears `out`, then reuses its allocation.
419///
420/// On an allocation failure this returns with `out` cleared **and emits a
421/// `tracing::error!`** so the empty result is operator-visible. An empty `out`
422/// on non-empty control bytes therefore always indicates either (a) no non-zero
423/// counters or (b) an OOM that was surfaced loudly. Prefer
424/// [`try_read_metrics_into`] in code paths where the caller can propagate errors.
425pub fn read_metrics_into(control_bytes: &[u8], out: &mut Vec<(u32, u32)>) {
426    out.clear();
427    let Ok(metrics_base) = control_word_index(control::METRICS_BASE, "metrics base word") else {
428        return;
429    };
430    let available_words = control_bytes.len() / 4;
431    if available_words <= metrics_base {
432        return;
433    }
434    let available_slots = (available_words - metrics_base).min(control::METRICS_SLOTS as usize);
435    let nonzero = count_nonzero_metrics_truncated(control_bytes, metrics_base, available_slots);
436    if let Err(e) = try_reserve_target_capacity(out, nonzero) {
437        tracing::error!(
438            error = %e,
439            nonzero_count = nonzero,
440            "read_metrics_into: allocation failed (returning empty metrics. Fix: reduce metrics fanout or decode into a pre-allocated scratch vector)."
441        );
442        return;
443    }
444    for slot in 0..available_slots {
445        let word_idx = metrics_base + slot;
446        let Some(count) = read_word_unaligned(control_bytes, word_idx) else {
447            break;
448        };
449        if count > 0 {
450            out.push((slot as u32, count));
451        }
452    }
453}
454
455/// Strictly read per-opcode metrics counters from a control buffer.
456///
457/// # Errors
458///
459/// Returns [`ProtocolError`] when the buffer is not word-aligned or is too
460/// short for the fixed metrics window.
461pub fn try_read_metrics(control_bytes: &[u8]) -> Result<Vec<(u32, u32)>, ProtocolError> {
462    let mut result = Vec::new();
463    try_read_metrics_into(control_bytes, &mut result)?;
464    Ok(result)
465}
466
467/// Strictly read per-opcode metrics counters into caller-owned storage.
468///
469/// Clears `out`, then reuses its allocation.
470///
471/// # Errors
472///
473/// Returns [`ProtocolError`] when the buffer is not word-aligned or is too
474/// short for the fixed metrics window.
475pub fn try_read_metrics_into(
476    control_bytes: &[u8],
477    out: &mut Vec<(u32, u32)>,
478) -> Result<(), ProtocolError> {
479    validate_word_aligned("control", control_bytes)?;
480    out.clear();
481    if let Ok(words) = bytemuck::try_cast_slice::<u8, u32>(control_bytes) {
482        try_reserve_target_capacity(
483            out,
484            count_nonzero_metrics_words_strict(words, control_bytes.len())?,
485        )?;
486        for i in 0..control::METRICS_SLOTS {
487            let word_idx = metrics_word_index(i)?;
488            let count =
489                words
490                    .get(word_idx)
491                    .copied()
492                    .map(u32::from_le)
493                    .ok_or(ProtocolError::MissingWord {
494                        buffer: "control",
495                        word_idx,
496                        byte_len: control_bytes.len(),
497                        fix: "decode only control buffers produced by the matching megakernel protocol encoder",
498                    })?;
499            if count > 0 {
500                out.push((i, count));
501            }
502        }
503        return Ok(());
504    }
505    try_reserve_target_capacity(out, count_nonzero_metrics_unaligned_strict(control_bytes)?)?;
506    for i in 0..control::METRICS_SLOTS {
507        let word_idx = metrics_word_index(i)?;
508        let count = read_word_unaligned(control_bytes, word_idx)
509            .ok_or(ProtocolError::MissingWord {
510            buffer: "control",
511            word_idx,
512            byte_len: control_bytes.len(),
513            fix: "decode only control buffers produced by the matching megakernel protocol encoder",
514        })?;
515        if count > 0 {
516            out.push((i, count));
517        }
518    }
519    Ok(())
520}
521
522mod debug_log;
523
524pub use debug_log::{
525    read_debug_log, read_debug_log_into, try_read_debug_log, try_read_debug_log_into,
526};
527
528/// Count DONE slots in a ring-buffer readback.
529///
530/// Returns `None` when the supplied bytes cannot contain `item_count` whole
531/// slots. This is intentionally part of the protocol module: DONE status is an
532/// ABI word, not a backend-specific readback rule.
533#[must_use]
534pub fn count_done_ring_slots(ring_bytes: &[u8], item_count: usize) -> Option<u64> {
535    if item_count == 0 {
536        return None;
537    }
538    let slot_words = usize::try_from(SLOT_WORDS).ok()?;
539    let required_bytes = item_count.checked_mul(slot_words)?.checked_mul(4)?;
540    if ring_bytes.len() < required_bytes {
541        return None;
542    }
543    let status_word = usize::try_from(STATUS_WORD).ok()?;
544    let words = bytemuck::try_cast_slice::<u8, u32>(ring_bytes).ok();
545    let done = (0..item_count)
546        .filter(|slot_idx| {
547            let word_idx = slot_idx
548                .checked_mul(slot_words)
549                .and_then(|base| base.checked_add(status_word));
550            word_idx.and_then(|idx| read_word_from_optional_words(words, ring_bytes, idx))
551                == Some(slot::DONE)
552        })
553        .count();
554    u64::try_from(done).ok()
555}
556
557/// Strictly count DONE slots in a ring-buffer readback.
558///
559/// # Errors
560///
561/// Returns [`ProtocolError`] when `ring_bytes` cannot contain `item_count`
562/// complete ring slots or when the byte count overflows the host protocol
563/// domain.
564pub fn try_count_done_ring_slots(
565    ring_bytes: &[u8],
566    item_count: usize,
567) -> Result<u64, ProtocolError> {
568    if item_count == 0 {
569        return Ok(0);
570    }
571    validate_word_aligned("ring", ring_bytes)?;
572    let slot_words =
573        usize::try_from(SLOT_WORDS).map_err(|_| ProtocolError::ByteLengthOverflow {
574            buffer: "ring",
575            fix: "keep SLOT_WORDS representable in host usize before decoding ring status",
576        })?;
577    let required_bytes = item_count
578        .checked_mul(slot_words)
579        .and_then(|words| words.checked_mul(4))
580        .ok_or(ProtocolError::ByteLengthOverflow {
581            buffer: "ring",
582            fix: "split the dispatch before ring status decode overflows host address space",
583        })?;
584    if ring_bytes.len() < required_bytes {
585        return Err(ProtocolError::MissingWord {
586            buffer: "ring",
587            word_idx: required_bytes / 4,
588            byte_len: ring_bytes.len(),
589            fix: "decode only full ring readbacks sized for the submitted megakernel item_count",
590        });
591    }
592    let status_word =
593        usize::try_from(STATUS_WORD).map_err(|_| ProtocolError::ByteLengthOverflow {
594            buffer: "ring",
595            fix: "keep STATUS_WORD representable in host usize before decoding ring status",
596        })?;
597    let words = bytemuck::try_cast_slice::<u8, u32>(ring_bytes).ok();
598    let mut done = 0_u64;
599    for slot_idx in 0..item_count {
600        let word_idx = slot_idx
601            .checked_mul(slot_words)
602            .and_then(|base| base.checked_add(status_word))
603            .ok_or(ProtocolError::ByteLengthOverflow {
604                buffer: "ring",
605                fix: "split the dispatch before ring status word indexing overflows host address space",
606            })?;
607        if read_word_from_optional_words(words, ring_bytes, word_idx) == Some(slot::DONE) {
608            done = done
609                .checked_add(1)
610                .ok_or(ProtocolError::ByteLengthOverflow {
611                    buffer: "ring",
612                    fix: "split the dispatch before DONE slot count exceeds u64",
613                })?;
614        }
615    }
616    Ok(done)
617}
618
619fn try_reserve_target_capacity<T>(
620    out: &mut Vec<T>,
621    target_capacity: usize,
622) -> Result<(), ProtocolError> {
623    try_reserve_protocol_capacity(
624        out,
625        target_capacity,
626        "control",
627        "host metrics decode could not reserve output records; reduce metrics fanout or decode into a reused scratch vector",
628    )
629}
630
631pub(super) fn try_reserve_protocol_capacity<T>(
632    out: &mut Vec<T>,
633    target_capacity: usize,
634    buffer: &'static str,
635    fix: &'static str,
636) -> Result<(), ProtocolError> {
637    vyre_foundation::allocation::try_reserve_vec_to_capacity(out, target_capacity)
638        .map_err(|_| ProtocolError::ByteLengthOverflow { buffer, fix })
639}
640
641fn count_nonzero_metrics_words_strict(
642    words: &[u32],
643    byte_len: usize,
644) -> Result<usize, ProtocolError> {
645    let mut count = 0usize;
646    for i in 0..control::METRICS_SLOTS {
647        let word_idx = metrics_word_index(i)?;
648        let word = words
649            .get(word_idx)
650            .copied()
651            .map(u32::from_le)
652            .ok_or(ProtocolError::MissingWord {
653            buffer: "control",
654            word_idx,
655            byte_len,
656            fix: "decode only control buffers produced by the matching megakernel protocol encoder",
657        })?;
658        if word > 0 {
659            count += 1;
660        }
661    }
662    Ok(count)
663}
664
665fn count_nonzero_metrics_unaligned_strict(control_bytes: &[u8]) -> Result<usize, ProtocolError> {
666    let mut count = 0usize;
667    for i in 0..control::METRICS_SLOTS {
668        let word_idx = metrics_word_index(i)?;
669        let word = read_word_unaligned(control_bytes, word_idx)
670            .ok_or(ProtocolError::MissingWord {
671            buffer: "control",
672            word_idx,
673            byte_len: control_bytes.len(),
674            fix: "decode only control buffers produced by the matching megakernel protocol encoder",
675        })?;
676        if word > 0 {
677            count += 1;
678        }
679    }
680    Ok(count)
681}
682
683fn count_nonzero_metrics_truncated(
684    control_bytes: &[u8],
685    metrics_base: usize,
686    available_slots: usize,
687) -> usize {
688    let mut count = 0usize;
689    for slot in 0..available_slots {
690        if read_word_unaligned(control_bytes, metrics_base + slot).unwrap_or(0) > 0 {
691            count += 1;
692        }
693    }
694    count
695}
696
697fn metrics_word_index(slot: u32) -> Result<usize, ProtocolError> {
698    let word =
699        control::METRICS_BASE
700            .checked_add(slot)
701            .ok_or(ProtocolError::ByteLengthOverflow {
702                buffer: "control",
703                fix: "metrics slot index overflows the protocol word offset; shard metrics reads",
704            })?;
705    control_word_index(word, "metrics word index")
706}
707
708fn control_word_index(word: u32, label: &'static str) -> Result<usize, ProtocolError> {
709    usize::try_from(word).map_err(|_| ProtocolError::ByteLengthOverflow {
710        buffer: "control",
711        fix: match label {
712            "observable word index" => {
713                "observable word index cannot fit host usize; shard observable reads"
714            }
715            "metrics word index" => "metrics word index cannot fit host usize; shard metrics reads",
716            _ => "control word index cannot fit host usize; shard protocol reads",
717        },
718    })
719}
720
721pub(crate) fn read_word(bytes: &[u8], word_idx: usize) -> Option<u32> {
722    if let Ok(words) = bytemuck::try_cast_slice::<u8, u32>(bytes) {
723        return words.get(word_idx).copied().map(u32::from_le);
724    }
725    read_word_unaligned(bytes, word_idx)
726}
727
728fn read_word_from_optional_words(
729    words: Option<&[u32]>,
730    bytes: &[u8],
731    word_idx: usize,
732) -> Option<u32> {
733    if let Some(words) = words {
734        return words.get(word_idx).copied().map(u32::from_le);
735    }
736    read_word_unaligned(bytes, word_idx)
737}
738
739fn read_word_unaligned(bytes: &[u8], word_idx: usize) -> Option<u32> {
740    let off = word_idx.checked_mul(4)?;
741    let end = off.checked_add(4)?;
742    let word = bytes.get(off..end)?;
743    Some(u32::from_le_bytes(word.try_into().ok()?))
744}
745
746fn read_required_word(
747    buffer: &'static str,
748    bytes: &[u8],
749    word_idx: usize,
750) -> Result<u32, ProtocolError> {
751    validate_word_aligned(buffer, bytes)?;
752    read_word(bytes, word_idx).ok_or(ProtocolError::MissingWord {
753        buffer,
754        word_idx,
755        byte_len: bytes.len(),
756        fix: "decode only buffers produced by the matching megakernel protocol encoder",
757    })
758}
759
760fn validate_word_aligned(buffer: &'static str, bytes: &[u8]) -> Result<(), ProtocolError> {
761    if bytes.len() % 4 == 0 {
762        Ok(())
763    } else {
764        Err(ProtocolError::MisalignedByteLength {
765            buffer,
766            byte_len: bytes.len(),
767            fix: "pass whole u32 protocol words; do not decode partial DMA/readback buffers",
768        })
769    }
770}
771
772pub(crate) fn write_word(bytes: &mut [u8], word_idx: usize, value: u32) {
773    let off = word_idx * 4;
774    bytes[off..off + 4].copy_from_slice(&value.to_le_bytes());
775}
776
777pub(super) fn words_to_bytes(words: u32) -> Option<usize> {
778    usize::try_from(words).ok()?.checked_mul(4)
779}
vyre_runtime/megakernel/protocol/codec.rs

vyre_runtime/megakernel/protocol/
codec.rs
