rhdl_ubft/kv/

kvstore.rs

//! An explicit, bounded compatibility frontend for the current replica.
//!
//! CPU word: opcode[1:0], value[31:2], key[63:32]. Opcode one is PUT;
//! other opcodes remain GET as upstream. PUT supports key16/value16 and
//! rejects wider values rather than silently truncating. GET uses key32
//! zero-extended to the table's key64. All response fields are initialized.
//!
//! The current controller duplicates each `(value32, sequence32)` decision
//! onto output and remote. This adapter buffers both, verifies equality,
//! and applies each accepted sequence once within a coordinated session.
//! Only repeats of the latest applied sequence/value are drained; older replays
//! fault. Reset discards this deduplication history. This is not an adapter for
//! historical mutually exclusive local/remote ownership. Sequence zero,
//! gaps, backwards movement, conflicting duplicates, and malformed beats
//! fault closed. A session must reset before sequence wrap; reset must also
//! reset/drain the table, controller, and all transport queues together. The
//! first nonzero sequence is accepted: attaching mid-history requires an
//! externally established consistent snapshot, not just a plausible sequence.
//!
//! Transfers occur on rising edges with valid and ready asserted. Producers
//! hold valid and payload until transfer during healthy operation. A terminal
//! fault requires coordinated recovery; reset suppresses all handshakes.
//!
//! One local PUT may wait while remote decisions continue to apply. A matching
//! decided value completes that PUT, but the wire has no client/request ID:
//! this proves value convergence, not attribution to a particular proposal.
//! GET is a local read, not a linearizable distributed read. A successful PUT
//! response says the table update completed, not that the protocol is safe.
use super::hash_table::{HashTableIn, HashTableOut, LookupReq, UpdateReq};
use rhdl::prelude::*;

/// One 256-bit controller decision. Reserved lanes must be zero, last true.
#[derive(Clone, Copy, Debug, PartialEq, Digital, Default)]
pub struct Decision {
    /// Decision offered; hold payload and last until the associated ready.
    pub valid: bool,
    /// Lane0: key in bits15:0, value in bits31:16, sequence in bits63:32.
    /// Lanes1 through 3 must be zero.
    pub data: [b64; 4],
    /// Must be true: every decision is exactly one beat.
    pub last: bool,
}

/// Observable CPU result. Status: 0 success, 1 PUT range error,
/// 2 table failure. Protocol fault is reported separately and is sticky.
#[derive(Clone, Copy, Debug, PartialEq, Digital, Default)]
pub struct KvResult {
    /// Result held until result_ready; no second local operation is admitted meanwhile.
    pub valid: bool,
    /// Original CPU opcode; one is PUT, all other encodings are GET.
    pub opcode: b2,
    /// Original full 32-bit CPU key.
    pub key: b32,
    /// GET value or zero on miss; PUT attempted value after table response,
    /// including table failure. Range rejection returns zero.
    pub value: b16,
    /// GET found the key; false for every PUT result.
    pub hit: bool,
    /// GET completed (hit is separate), or a matching PUT decision was installed.
    pub success: bool,
    /// Zero completion, one PUT range rejection, two table insertion failure.
    pub status: b2,
}

#[derive(Clone, Copy, Debug, PartialEq, Digital, Default)]
/// CPU, controller, and table inputs for one outstanding local operation.
pub struct KvIn {
    /// CPU operation offered; retain it until operation_ready.
    pub operation_valid: bool,
    /// CPU word: key63:32, value31:2, opcode1:0.
    pub operation: b64,
    /// CPU accepts a held result this edge.
    pub result_ready: bool,
    /// Controller accepts the offered group-zero proposal this edge.
    pub proposal_ready: bool,
    /// Controller output copy of the decision.
    pub output: Decision,
    /// Controller remote copy, paired and checked against output.
    pub remote: Decision,
    /// Responses and admission readiness from the attached table.
    pub table: HashTableOut,
}

#[derive(Clone, Copy, Debug, PartialEq, Digital, Default)]
/// CPU results and requests to the external controller and local table.
pub struct KvOut {
    /// Accepts one CPU operation; low while local work/result is pending or faulted.
    pub operation_ready: bool,
    /// Held observable CPU completion or explicit range/table error.
    pub result: KvResult,
    /// A group-zero PUT proposal is offered; retained until proposal_ready.
    pub proposal_valid: bool,
    /// ProposedValue: low32 = key16 | value16<<16, high32 = group zero.
    pub proposal: b64,
    /// Room for one output copy; independent of the local PUT wait.
    pub output_ready: bool,
    /// Room for one remote copy; applying waits for both matching copies.
    pub remote_ready: bool,
    /// Lookup/update requests and response-consumer readiness.
    pub table: HashTableIn,
    /// Terminal until coordinated reset; stops new requests and application.
    /// An already-held result remains consumable. Pending work need not complete.
    pub fault: bool,
    /// Wrapping diagnostic counters, not safety identifiers.
    /// Wrapping count of successfully applied sequences, including upserts.
    pub applied: b32,
    /// Wrapping count of paired repeats of the most recently applied sequence/value.
    pub duplicates: b32,
}

#[derive(Clone, Copy, Debug, PartialEq, Digital, Default)]
/// Separate local-operation and replicated-application state machines.
pub struct KvState {
    /// 0 idle, 1 offer proposal, 2 await matching decision, 3 offer lookup,
    /// 4 await lookup result. This machine serves one local CPU operation.
    pub local_phase: b3,
    /// 0 await paired decisions, 1 offer table update, 2 await update result.
    /// It advances independently of the local CPU machine.
    pub apply_phase: b2,
    /// Original key of the outstanding CPU operation.
    pub cpu_key: b32,
    /// Original opcode of the outstanding CPU operation.
    pub cpu_opcode: b2,
    /// Packed key16/value16 offered for the local PUT.
    pub proposed: b32,
    /// CPU result retained until its consumer accepts it.
    pub result: KvResult,
    /// One buffered controller output copy.
    pub output: Decision,
    /// One buffered controller remote copy.
    pub remote: Decision,
    /// Packed key16/value16 of the decision currently entering the table.
    pub applying: b32,
    /// Sequence belonging to applying.
    pub applying_seq: b32,
    /// A table update has completed successfully in this session.
    pub have_last: bool,
    /// Most recently applied sequence; meaningful when have_last.
    pub last_seq: b32,
    /// Value paired with last_seq, used to check immediate replays.
    pub last_value: b32,
    /// Latched terminal framing, sequence, response, or application error.
    pub fault: bool,
    /// Wrapping successful-application diagnostic counter.
    pub applied: b32,
    /// Wrapping accepted immediate-replay diagnostic counter.
    pub duplicates: b32,
}

/// Lossless-within-profile replacement for the upstream KV frontend.
#[derive(Clone, Debug, Synchronous, SynchronousDQ)]
pub struct LegacyKv16 {
    state: rhdl_fpga::core::dff::DFF<KvState>,
}
impl Default for LegacyKv16 {
    fn default() -> Self {
        Self {
            state: rhdl_fpga::core::dff::DFF::new(KvState::default()),
        }
    }
}
impl SynchronousIO for LegacyKv16 {
    type I = KvIn;
    type O = KvOut;
    type Kernel = kv_kernel;
}

/// CPU phases: idle, propose, wait-decision, lookup-request, lookup-response.
/// Apply phases: idle, update-request, update-response.
#[kernel]
pub fn kv_kernel(cr: ClockReset, i: KvIn, q: LegacyKv16Q) -> (KvOut, LegacyKv16D) {
    // r is this cycle's registered snapshot; s is the next snapshot. All state
    // tests use r, so accepting a proposal cannot also complete it in this edge.
    let r = q.state;
    let mut s = r;
    let mut o = KvOut {
        operation_ready: false,
        result: r.result,
        proposal_valid: false,
        proposal: r.proposed.resize::<64>(),
        output_ready: false,
        remote_ready: false,
        table: HashTableIn {
            lookup: LookupReq {
                valid: false,
                key: b64(0),
                source: false,
            },
            update: UpdateReq {
                valid: false,
                op: b1(0),
                key: b64(0),
                value: b16(0),
                source: true,
            },
            lookup_response_ready: false,
            update_response_ready: false,
        },
        fault: r.fault,
        applied: r.applied,
        duplicates: r.duplicates,
    };
    if r.result.valid && i.result_ready {
        s.result.valid = false;
    }
    if !r.fault {
        // One buffer per stream absorbs skew without assuming simultaneous
        // arrival. A full buffer backpressures only its own input stream.
        o.output_ready = !r.output.valid;
        o.remote_ready = !r.remote.valid;
        if o.output_ready && i.output.valid {
            s.output = i.output;
        }
        if o.remote_ready && i.remote.valid {
            s.remote = i.remote;
        }

        o.operation_ready = r.local_phase == b3(0) && !r.result.valid && i.table.initialized;
        if o.operation_ready && i.operation_valid {
            let key: b32 = (i.operation >> b64(32)).resize();
            let value: b30 = (i.operation >> b64(2)).resize();
            let opcode: b2 = i.operation.resize();
            s.cpu_key = key;
            s.cpu_opcode = opcode;
            if opcode == b2(1) {
                if (key >> b32(16)) != b32(0) || (value >> b30(16)) != b30(0) {
                    s.result = KvResult {
                        valid: true,
                        opcode,
                        key,
                        value: b16(0),
                        hit: false,
                        success: false,
                        status: b2(1),
                    };
                } else {
                    s.proposed = key | (value.resize::<32>() << b32(16));
                    s.local_phase = b3(1);
                }
            } else {
                s.local_phase = b3(3);
            }
        }
        if r.local_phase == b3(1) {
            o.proposal_valid = true;
            if i.proposal_ready {
                s.local_phase = b3(2);
            }
        }

        // Fair table arbitration is owned by HashTable; no local CPU wait
        // prevents remote updates from reaching it.
        if r.local_phase == b3(3) {
            o.table.lookup = LookupReq {
                valid: true,
                key: r.cpu_key.resize::<64>(),
                source: false,
            };
            if i.table.lookup_request_ready {
                s.local_phase = b3(4);
            }
        }
        if r.local_phase == b3(4) {
            o.table.lookup_response_ready = true;
            if i.table.lookup.valid {
                if i.table.lookup.key != r.cpu_key.resize::<64>() || i.table.lookup.source {
                    s.fault = true;
                } else {
                    s.result = KvResult {
                        valid: true,
                        opcode: r.cpu_opcode,
                        key: r.cpu_key,
                        value: if i.table.lookup.hit {
                            i.table.lookup.value
                        } else {
                            b16(0)
                        },
                        hit: i.table.lookup.hit,
                        success: true,
                        status: b2(0),
                    };
                    s.local_phase = b3(0);
                }
            }
        }

        if r.apply_phase == b2(0) && r.output.valid && r.remote.valid {
            // The two ports carry copies of one decision, not two independent
            // writes. Validate their full framing before touching table state.
            let mut malformed = !r.output.last || !r.remote.last;
            for lane in 0..4 {
                if r.output.data[lane] != r.remote.data[lane] {
                    malformed = true;
                }
                if lane > 0 && r.remote.data[lane] != b64(0) {
                    malformed = true;
                }
            }
            let seq: b32 = (r.remote.data[0] >> b64(32)).resize();
            let value: b32 = r.remote.data[0].resize();
            if seq == b32(0) {
                malformed = true;
            }
            if r.have_last {
                if seq == r.last_seq {
                    if value != r.last_value {
                        malformed = true;
                    }
                } else if r.last_seq == b32(0xffff_ffff) || seq != r.last_seq + b32(1) {
                    malformed = true;
                }
            }
            if malformed {
                s.fault = true;
            } else {
                s.output.valid = false;
                s.remote.valid = false;
                if r.have_last && seq == r.last_seq {
                    // Only an identical replay of the last applied decision is
                    // harmless. Older decisions cannot silently rewind state.
                    s.duplicates = r.duplicates + b32(1);
                } else {
                    s.applying = value;
                    s.applying_seq = seq;
                    s.apply_phase = b2(1);
                }
            }
        }
        if r.apply_phase == b2(1) {
            let key: b16 = r.applying.resize();
            let value: b16 = (r.applying >> b32(16)).resize();
            o.table.update = UpdateReq {
                valid: true,
                op: b1(0),
                key: key.resize::<64>(),
                value,
                source: true,
            };
            if i.table.update_request_ready {
                s.apply_phase = b2(2);
            }
        }
        if r.apply_phase == b2(2) {
            o.table.update_response_ready = true;
            if i.table.update.valid {
                let key: b16 = r.applying.resize();
                let value: b16 = (r.applying >> b32(16)).resize();
                let malformed = i.table.update.op != b1(0)
                    || i.table.update.key != key.resize::<64>()
                    || i.table.update.value != value
                    || !i.table.update.source;
                if malformed {
                    s.fault = true;
                } else {
                    if !i.table.update.success {
                        s.fault = true;
                    }
                    if i.table.update.success {
                        // Advance replay protection after application succeeds,
                        // never merely when the controller announces a decision.
                        s.have_last = true;
                        s.last_seq = r.applying_seq;
                        s.last_value = r.applying;
                        s.applied = r.applied + b32(1);
                    }
                    if r.local_phase == b3(2) && r.proposed == r.applying {
                        // This is value matching, not requester identification:
                        // another client may have proposed the same PUT.
                        s.result = KvResult {
                            valid: true,
                            opcode: b2(1),
                            key: r.cpu_key,
                            value,
                            hit: false,
                            success: i.table.update.success,
                            status: if i.table.update.success { b2(0) } else { b2(2) },
                        };
                        s.local_phase = b3(0);
                    }
                    s.apply_phase = b2(0);
                }
            }
        }
    }
    if cr.reset.any() {
        o.operation_ready = false;
        o.result.valid = false;
        o.proposal_valid = false;
        o.output_ready = false;
        o.remote_ready = false;
        o.table.lookup.valid = false;
        o.table.update.valid = false;
        o.table.lookup_response_ready = false;
        o.table.update_response_ready = false;
    }
    (o, LegacyKv16D { state: s })
}

#[cfg(test)]
#[path = "kvstore_tests.rs"]
mod tests;