rhdl_ubft/

register.rs

//! Single-writer registers: recovery layer 1 (docs/rhdl_ubft-recovery-design.md).
//!
//! **Problem.** Consistent broadcast needs memory that each replica owns,
//! every replica can read, and a read can't see half a write. RDMA writes
//! of several words reach memory in pieces, so a reader racing a writer can
//! see a mix of an old and a new value.
//!
//! **First attempt, and why it failed.** The first version followed uBFT
//! (LPD-EPFL/ubft `504eba0`, `swmr/`): a checksum over incarnation and value,
//! and a read was valid when the checksum matched. Atlas found two payloads
//! with the same checksum at the same incarnation, so a read mixing their
//! words looks complete. That works with whole-word tearing, and with whole
//! 64-byte beats (tests/register_integrity.rs). No checksum fixes this: the
//! payload is partly chosen by others (a broadcaster's signature, stored in
//! an honest receiver's register), so anyone who knows the function can aim
//! for a collision. uBFT's XXH3 has the same weakness in principle.
//!
//! **Idea.** Tag every word with the write it belongs to. A subslot is an
//! incarnation word, then the payload packed 24 bits per word, each word
//! carrying the incarnation's low 8 bits on top:
//!
//! ```text
//! word 0:  incarnation (32 bits)
//! word j:  [tag(incarnation) | 24 payload bits]      j = 1 .. S − 1
//! tag(i) = 0x80 | (i & 0x7f)
//! ```
//!
//! A read is valid only if every word's tag equals `tag` of the incarnation
//! word. The tag's top bit is always set, so zeroed memory never matches a
//! tag. (An independent review found the first version, which used the
//! incarnation's low byte, accepted an all-zero subslot under incarnation
//! 256.) A read mixing words of two writes has mismatched tags whatever
//! the payload, so nobody can choose values to get one through. As in uBFT,
//! there are two subslots, the incarnation's parity picks the one to write,
//! and the reader keeps the valid subslot with the higher incarnation.
//!
//! **Assumptions** (stated here; they must be enforced by future layer-2
//! engines, which do not exist yet; these kernels cannot enforce them):
//! - a 32-bit aligned word is written and read whole (the AXI strobe unit);
//! - tags repeat every 128 incarnations (64 writes to one subslot), so the
//!   incarnation must advance by exactly one per write (no caller-chosen
//!   jumps), and a read must not overlap more than one write. The writer
//!   starts a write at least `COOLDOWN` cycles after the previous one became
//!   visible, and a reader accepts a sample only if its whole span, queues
//!   included, is strictly less than `COOLDOWN` cycles (the strict audit's
//!   bound; then mixed subslots differ by two generations). Without those
//!   engines the tags alias: a mix of writes 1 and 129 (or 257) of one
//!   subslot passes (review witnesses, agent/ubft-register-generation
//!   0f323f3). uBFT has the same kind
//!   of timing rule (a 1 ms write cooldown).
//! - **Open (review finding):** with replicated registers, a memory host
//!   that missed 128 or more updates can hold a subslot whose mix of an old
//!   and a new write shares a tag. Logical incarnations alone cannot rule
//!   that out. The fix planned for layer 2 is a physical generation per host,
//!   stamped by that host's trusted receive path on every applied write and
//!   checked by readers, distinct from the logical incarnation.
//!
//! Incarnation 0 means "never written". Cleared memory has incarnation 0
//! and so never reads as a value.
//!
//! **Cost.** Payload density is 24/32: a 64-byte signature takes 22 words
//! plus the incarnation word.

use rhdl::prelude::*;

/// Payload bits per tagged word.
pub const CHUNK_BITS: usize = 24;

/// Words in a subslot holding `w` 32-bit payload words: the incarnation
/// word plus `⌈32w / 24⌉` tagged words.
pub const fn subslot_words(w: usize) -> usize {
    1 + (32 * w).div_ceil(CHUNK_BITS)
}

/// A subslot for `incarnation` holding `v`. `S` must be [`subslot_words`]`(W)`.
#[kernel]
pub fn pack_subslot<const W: usize, const S: usize>(incarnation: b32, v: [b32; W]) -> [b32; S] {
    let tag: b32 = (b32(0x80) | (incarnation & b32(0x7f))) << b32(24);
    let mut out = [b32(0); S];
    out[0] = incarnation;
    // (Kernel loops need a constant end, and kernels take shifts and
    // masks, not / or %.)
    for jj in 1..S {
        // Chunk j is payload bits 24j .. 24j + 24.
        let j = jj - 1;
        let a = (24 * j) >> 5;
        let o = (24 * j) & 31;
        let lo: b64 = v[a].resize::<64>() >> b64(o as u128);
        let hi: b64 = if a + 1 < W {
            v[a + 1].resize::<64>() << b64((32 - o) as u128)
        } else {
            b64(0)
        };
        let chunk: b32 = ((lo | hi) & b64(0xff_ffff)).resize();
        out[1 + j] = tag | chunk;
    }
    out
}

/// Which subslot a write of `incarnation` goes to (0 or 1, by parity).
#[kernel]
pub fn subslot_of(incarnation: b32) -> b1 {
    incarnation.resize()
}

/// One subslot as read: whether every word belongs to one write, and the
/// payload.
#[kernel]
pub fn unpack_subslot<const W: usize, const S: usize>(s: [b32; S]) -> (bool, [b32; W]) {
    let tag: b32 = b32(0x80) | (s[0] & b32(0x7f));
    let mut ok = s[0] != b32(0);
    for j in 1..S {
        if (s[j] >> b32(24)) != tag {
            ok = false;
        }
    }
    // Scatter chunk j (payload bits 24j .. 24j + 24) into the one or two
    // payload words it covers.
    let mut acc = [b64(0); W];
    for jj in 1..S {
        let j = jj - 1;
        let a = (24 * j) >> 5;
        let o = (24 * j) & 31;
        let chunk: b64 = (s[1 + j] & b32(0xff_ffff)).resize();
        if a < W {
            acc[a] = acc[a] | (chunk << b64(o as u128));
        }
        if o > 8 && a + 1 < W {
            acc[a + 1] = acc[a + 1] | (chunk >> b64((32 - o) as u128));
        }
    }
    let mut v = [b32(0); W];
    for i in 0..W {
        v[i] = acc[i].resize();
    }
    (ok, v)
}

/// Value selection and fault observation answer different questions. Identical
/// valid subslots can supply a value while still revealing a same-host tie.
#[derive(Clone, Copy, Debug, PartialEq, Digital)]
pub struct RegisterInspection<const W: usize> {
    pub valid: bool,
    pub physical: b32,
    pub payload: [b32; W],
    /// Both subslots validate and claim the same physical generation.
    pub same_host_tie: bool,
    /// A same-host tie also disagrees on at least one decoded payload word.
    pub conflicting_payloads: bool,
}

/// Inspect one host's pair without losing the distinction between a usable
/// identical copy and a contradictory pair. Never compare different hosts here.
#[kernel]
pub fn inspect_register<const W: usize, const S: usize>(
    a: [b32; S],
    b: [b32; S],
) -> RegisterInspection<W> {
    let (ok_a, va) = unpack_subslot::<W, S>(a);
    let (ok_b, vb) = unpack_subslot::<W, S>(b);
    let tie = ok_a && ok_b && a[0] == b[0];
    let mut differs = false;
    for i in 0..W {
        if va[i] != vb[i] {
            differs = true;
        }
    }
    let mut result = RegisterInspection::<W> {
        valid: false,
        physical: b32(0),
        payload: [b32(0); W],
        same_host_tie: tie,
        conflicting_payloads: tie && differs,
    };
    if !(tie && differs) {
        if ok_a && (!ok_b || a[0] > b[0]) {
            result.valid = true;
            result.physical = a[0];
            result.payload = va;
        } else if ok_b {
            result.valid = true;
            result.physical = b[0];
            result.payload = vb;
        }
    }
    result
}

/// The register read from its two subslots `a` (subslot 0) and `b`: whether
/// either is valid, the higher valid incarnation, and its value.
/// Equal valid generations with different decoded payloads are ambiguous and
/// return invalid/default. Identical copies remain acceptable. This comparison
/// is within one host's register; equal generations across hosts are normal.
#[kernel]
pub fn select_register<const W: usize, const S: usize>(
    a: [b32; S],
    b: [b32; S],
) -> (bool, b32, [b32; W]) {
    // Preserve the approved R-32 interface for callers that only select values.
    let inspected = inspect_register::<W, S>(a, b);
    (inspected.valid, inspected.physical, inspected.payload)
}

#[cfg(test)]
mod tests {
    use super::*;

    const W: usize = 4;
    const S: usize = 7;

    fn v(x: u32) -> [b32; W] {
        std::array::from_fn(|i| {
            b32(u128::from(
                x.wrapping_mul(0x9e37_79b9)
                    .wrapping_add(i as u32 * 0x0101_0101),
            ))
        })
    }

    fn write(reg: &mut [[b32; S]; 2], inc: u32, x: u32) {
        let s = pack_subslot::<W, S>(b32(u128::from(inc)), v(x));
        reg[subslot_of(b32(u128::from(inc))).raw() as usize] = s;
    }

    #[test]
    fn sizes() {
        assert_eq!(subslot_words(4), S);
        assert_eq!(subslot_words(16), 23);
        assert_eq!(subslot_words(45), 61);
    }

    /// Lower the real selector, including its ambiguity rejection, in the
    /// synchronous compilation mode used by the register controllers.
    #[test]
    fn ambiguity_policy_lowers_synchronously() -> anyhow::Result<()> {
        // Embedded at compilation, not read from the live tree when the test
        // runs. This fingerprints this selector source, not the whole crate.
        let source_fingerprint = include_bytes!("register.rs")
            .iter()
            .fold(0xcbf29ce484222325u64, |hash, byte| {
                (hash ^ u64::from(*byte)).wrapping_mul(0x100000001b3)
            });
        println!("REGISTER_SOURCE_FNV={source_fingerprint:016x}");
        let _ = compile_design::<select_register<W, S>>(CompilationMode::Synchronous)?;
        Ok(())
    }

    /// Packing round-trips every bit of the payload.
    #[test]
    fn pack_round_trips() {
        for x in [0u32, 1, 0xdead_beef, u32::MAX] {
            for bit in 0..(32 * W) {
                let mut p = v(x);
                p[bit / 32] = p[bit / 32] ^ b32(1 << (bit % 32));
                let (ok, got) = unpack_subslot::<W, S>(pack_subslot::<W, S>(b32(5), p));
                assert!(ok);
                assert_eq!(got, p);
            }
        }
    }

    /// A never-written register reads as empty; then each write is seen,
    /// newest first, whichever subslot it went to.
    #[test]
    fn reads_follow_writes() {
        let mut reg = [[b32(0); S]; 2];
        assert!(!select_register::<W, S>(reg[0], reg[1]).0);
        for inc in 1..=5u32 {
            write(&mut reg, inc, 100 + inc);
            let (ok, got_inc, val) = select_register::<W, S>(reg[0], reg[1]);
            assert!(ok);
            assert_eq!(got_inc.raw(), u128::from(inc));
            assert_eq!(val, v(100 + inc));
        }
    }

    /// Every way to mix words of two writes of one subslot is rejected, and
    /// the reader falls back to the other subslot's complete value. Checked
    /// for every subset of words taken from the older write.
    #[test]
    fn any_mix_of_two_writes_falls_back_to_the_previous_value() {
        let mut reg = [[b32(0); S]; 2];
        write(&mut reg, 1, 11);
        write(&mut reg, 2, 22);
        let fresh = pack_subslot::<W, S>(b32(3), v(33));
        for mask in 1u32..(1 << S) - 1 {
            let mut s = fresh;
            for k in 0..S {
                if mask >> k & 1 == 1 {
                    s[k] = reg[1][k];
                }
            }
            let (ok, inc, val) = select_register::<W, S>(reg[0], s);
            assert!(ok);
            assert_eq!((inc.raw(), val), (2, v(22)), "mask {mask:#b}");
        }
    }

    /// Review finding: an all-zero subslot under an incarnation whose low
    /// bits are zero is not a value.
    #[test]
    fn zeroed_memory_never_matches_a_tag() {
        for inc in [128u32, 256, 0x8000_0000, 1, 0x7f] {
            let mut s = [b32(0); S];
            s[0] = b32(u128::from(inc));
            assert!(!unpack_subslot::<W, S>(s).0, "incarnation {inc}");
            assert!(!select_register::<W, S>(s, [b32(0); S]).0);
        }
    }

    /// Atlas's chosen-payload witness against the old checksum: two
    /// payloads that collided, mixed word by word, are rejected here.
    #[test]
    fn chosen_colliding_payloads_cannot_mix() {
        let old = [b32(0x1b3), b32(0x0100_0366), b32(0), b32(0)];
        let new = [b32(0x1b2), b32(0x0080_0366), b32(1), b32(1)];
        let prior = [b32(9); 4];
        let old_slot = pack_subslot::<W, S>(b32(1), old);
        let current = pack_subslot::<W, S>(b32(2), prior);
        let new_slot = pack_subslot::<W, S>(b32(3), new);
        for k in 0..S {
            let mut torn = new_slot;
            torn[k] = old_slot[k];
            let (ok, _, value) = select_register::<W, S>(current, torn);
            assert!(ok);
            assert_eq!(value, prior, "word {k}");
        }
    }

    /// The kernels as hardware: the RHDL VM and the generated Verilog agree
    /// with the native code on fresh, mixed and empty registers.
    #[test]
    fn kernels_match_vm_and_generated_verilog() -> anyhow::Result<()> {
        use rhdl::core::sim::testbench::kernel::test_kernel_vm_and_verilog_synchronous;
        // These plain Digital values belong to one synchronous register step.
        // Keep the same VM/RTL/netlist checks as the lowering control's mode;
        // this fixture does not model asynchronous clock-domain crossings.
        let mut cases = Vec::new();
        let empty = [b32(0); S];
        for inc in [1u32, 2, 3, 0xffff_fffe] {
            let a = pack_subslot::<W, S>(b32(u128::from(inc)), v(inc));
            let b = pack_subslot::<W, S>(b32(u128::from(inc + 1)), v(inc + 1));
            let mut torn = b;
            torn[S - 1] = a[S - 1];
            cases.push((a, b));
            cases.push((b, a));
            cases.push((a, torn));
            cases.push((empty, a));
        }
        cases.push((empty, empty));
        test_kernel_vm_and_verilog_synchronous::<select_register<W, S>, _, _, _>(
            select_register::<W, S>,
            cases.into_iter(),
        )?;
        let packs = (1u32..6).map(|i| (b32(u128::from(i)), v(i)));
        test_kernel_vm_and_verilog_synchronous::<pack_subslot<W, S>, _, _, _>(
            pack_subslot::<W, S>,
            packs,
        )?;
        Ok(())
    }
}