rhdl_ubft/
broadcast.rs

1//! Sending: one RDMA write to every replica, each a single packet.
2//!
3//! **Problem.** ubft_new has two broadcast styles, and both lose messages.
4//! The fast path's `bcast_vote_rr` sends to one queue pair, then the state
5//! machine moves on, so a vote reaches one replica. The slow path loops over
6//! the queue pairs but skips any destination whose stream is full that
7//! cycle. Its 64-byte certify share is eight 64-bit beats, each marked
8//! `last`, so the RoCE stack sees eight packets, not one.
9//!
10//! **Idea.** [`Broadcaster`] sends the same write to the replica's `N − 1`
11//! peers, on local queue pairs `qbase .. qbase + N − 1` (`qbase = (N − 1)·id`;
12//! peer `j` is at rank `j` below `id` and `j − 1` above, as the RoCE stack
13//! programs them), and waits for each meta and data beat to be accepted.
14//! Nothing is dropped. A write is one packet: its data beats are contiguous,
15//! and `tlast` is set on the last beat only, so a 64-byte Ed25519 share is
16//! one beat and a 26-beat DSig share is one packet. The controller records
17//! the same words in its own HBM ([`crate::writer`]); there is no connection
18//! to self.
19//!
20//! **Result.** Every peer gets every vote, and the replica counts its own.
21//! That is what lets the fast path reach its `2f + 1` quorum at all.
22
23use rhdl::prelude::*;
24
25use crate::ports::{Axis256, Axis512, idle256, idle512};
26
27/// A broadcast request.
28#[derive(Clone, Copy, Debug, PartialEq, Digital, Default)]
29pub struct BroadcastReq {
30    /// Start this cycle (ignored while busy).
31    pub start: bool,
32    /// First queue pair (`id·N`).
33    pub qbase: b24,
34    /// Replicas (`node_num`).
35    pub n: b5,
36    /// Destination word index (byte address = 4 · word).
37    pub word: b32,
38    /// Bytes in the write: up to 64 from `payload`, or `beats · 64` from
39    /// the share RAM.
40    pub len: b16,
41    /// The bytes, from byte 0 (one-beat writes).
42    pub payload: [b64; 8],
43    /// Data beats per write (1, or the share's beats).
44    pub beats: b5,
45    /// Take the data beats from the share RAM (`ram_addr` / `ram_data`).
46    pub from_ram: bool,
47}
48
49/// No broadcast.
50#[kernel]
51pub fn idle_bcast_req() -> BroadcastReq {
52    BroadcastReq { start: false, qbase: b24(0), n: b5(0), word: b32(0), len: b16(0), payload: [b64(0); 8], beats: b5(0), from_ram: false }
53}
54
55/// Inputs of [`Broadcaster`].
56#[derive(Clone, Copy, Debug, PartialEq, Digital, Default)]
57pub struct BroadcastIn {
58    /// The request.
59    pub req: BroadcastReq,
60    /// `m_axis_tx_meta` `tready`.
61    pub meta_ready: bool,
62    /// `m_axis_tx_data` `tready`.
63    pub data_ready: bool,
64    /// The share RAM's output: the beat at last cycle's `ram_addr`.
65    pub ram_data: [b64; 8],
66}
67
68/// Outputs of [`Broadcaster`].
69#[derive(Clone, Copy, Debug, PartialEq, Digital, Default)]
70pub struct BroadcastOut {
71    /// Sending.
72    pub busy: bool,
73    /// The last write was accepted this cycle.
74    pub done: bool,
75    /// `m_axis_tx_meta`.
76    pub meta: Axis256,
77    /// `m_axis_tx_data`.
78    pub data: Axis512,
79    /// Share RAM read address.
80    pub ram_addr: b5,
81}
82
83/// The broadcaster's registers.
84#[derive(Clone, Copy, Debug, PartialEq, Digital, Default)]
85pub struct BroadcastState {
86    /// Sending.
87    pub busy: bool,
88    /// The request being sent.
89    pub req: BroadcastReq,
90    /// Destination replica (0..n).
91    pub dest: b5,
92    /// This destination's meta beat was accepted.
93    pub meta_sent: bool,
94    /// This destination's data beats are all accepted.
95    pub data_sent: bool,
96    /// Data beat being sent.
97    pub beat: b5,
98    /// `ram_addr` changed last cycle, so `ram_data` is not yet the new beat.
99    pub stale: bool,
100}
101
102/// The RoCE request word for a write of `len` bytes to `raddr` on queue
103/// pair `qpn` (op 1, laddr 0: data on the stream).
104#[kernel]
105pub fn write_meta(qpn: b24, raddr: b48, len: b32) -> [b64; 4] {
106    // op = 1 in bits 0..3, qpn in 3..27, laddr = 0 in 27..75, raddr in
107    // 75..123, len in 123..155.
108    let lane0 = b64(1) | (qpn.resize::<64>() << b64(3));
109    let lane1 = (raddr.resize::<64>() << b64(11)) | (len.resize::<64>() << b64(59));
110    let lane2 = (len >> b32(5)).resize::<64>();
111    [lane0, lane1, lane2, b64(0)]
112}
113
114/// `tkeep` for `len` bytes from byte 0.
115#[kernel]
116pub fn keep_bytes(len: b16) -> b64 {
117    if len >= b16(64) { !b64(0) } else { (b64(1) << len.resize::<64>()) - b64(1) }
118}
119
120/// Sends one write to every replica (see the module docs).
121#[derive(Clone, Debug, Synchronous, SynchronousDQ)]
122pub struct Broadcaster {
123    state: rhdl_fpga::core::dff::DFF<BroadcastState>,
124}
125
126impl Default for Broadcaster {
127    fn default() -> Self {
128        Self { state: rhdl_fpga::core::dff::DFF::new(BroadcastState::default()) }
129    }
130}
131
132impl SynchronousIO for Broadcaster {
133    type I = BroadcastIn;
134    type O = BroadcastOut;
135    type Kernel = broadcast_kernel;
136}
137
138/// The broadcaster's logic.
139#[kernel]
140pub fn broadcast_kernel(_cr: ClockReset, i: BroadcastIn, q: BroadcasterQ) -> (BroadcastOut, BroadcasterD) {
141    let r = q.state;
142    let mut s = r;
143    let mut o = BroadcastOut { busy: r.busy, done: false, meta: idle256(), data: idle512(), ram_addr: r.beat };
144    // No peers (a one-replica run): done at once, nothing sent.
145    if !r.busy && i.req.start && i.req.n == b5(0) {
146        o.done = true;
147    }
148    if !r.busy && i.req.start && i.req.n != b5(0) {
149        s.busy = true;
150        s.req = i.req;
151        s.dest = b5(0);
152        s.meta_sent = false;
153        s.data_sent = false;
154        s.beat = b5(0);
155        s.stale = true;
156    }
157    if r.busy {
158        let qpn = r.req.qbase + r.dest.resize::<24>();
159        let raddr: b48 = (r.req.word.resize::<48>()) << b48(2);
160        o.meta.valid = !r.meta_sent;
161        o.meta.data = write_meta(qpn, raddr, r.req.len.resize());
162        o.meta.last = true;
163        let final_beat = r.beat + b5(1) == r.req.beats;
164        // RAM beats need one settled cycle after the address changes.
165        let have = !r.req.from_ram || !r.stale;
166        o.data.valid = !r.data_sent && have;
167        o.data.data = if r.req.from_ram { i.ram_data } else { r.req.payload };
168        o.data.keep = if r.req.from_ram { !b64(0) } else { keep_bytes(r.req.len) };
169        o.data.last = final_beat;
170        s.stale = false;
171        let meta_ok = r.meta_sent || i.meta_ready;
172        s.meta_sent = meta_ok;
173        let mut data_ok = r.data_sent;
174        if !r.data_sent && have && i.data_ready {
175            if final_beat {
176                data_ok = true;
177            } else {
178                s.beat = r.beat + b5(1);
179                s.stale = true;
180            }
181        }
182        s.data_sent = data_ok;
183        if meta_ok && data_ok {
184            s.meta_sent = false;
185            s.data_sent = false;
186            s.beat = b5(0);
187            s.stale = true;
188            s.dest = r.dest + b5(1);
189            if r.dest + b5(1) == r.req.n {
190                s.busy = false;
191                o.done = true;
192            }
193        }
194    }
195    (o, BroadcasterD { state: s })
196}
rhdl_ubft/broadcast.rs

rhdl_ubft/
broadcast.rs
