rhdl_ubft/
fetch.rs

1//! Reading mailboxes: one burst instead of one read per word.
2//!
3//! **Problem.** ubft_new reads HBM through `volatile int *`, one 32-bit
4//! word per AXI transaction. Counting a slot's votes takes `N` reads, a
5//! 64-byte signature share 16, and each pays the full memory latency (HLS
6//! assumes 64 cycles). Polling is most of what the kernel does while it
7//! waits for peers, so its loop crawls.
8//!
9//! **Idea.** Everything one poll needs sits in at most 32 consecutive words
10//! (a slot's two-word votes from 16 replicas, a prepare, a record).
11//! [`Fetcher`] reads them with one INCR burst of one to three 64-byte beats,
12//! whatever the alignment (the mailboxes start at byte 1,000,764, which is
13//! 60 mod 64), and hands back the 32 words starting at the requested one.
14//!
15//! **Result.** A poll costs one memory latency plus one or two beats,
16//! instead of `N` latencies.
17
18use rhdl::prelude::*;
19
20use crate::ports::{AxiRead, AxiReadResp, idle_read};
21
22/// A fetch request: `count` (1..=16) words from word `word` of the buffer
23/// at byte address `base`.
24#[derive(Clone, Copy, Debug, PartialEq, Digital, Default)]
25pub struct FetchReq {
26    /// Start a fetch this cycle (ignored while busy).
27    pub start: bool,
28    /// Buffer base (byte address, 64-byte aligned, as XRT allocates).
29    pub base: b64,
30    /// First word.
31    pub word: b32,
32    /// Words wanted, 1..=32.
33    pub count: b6,
34}
35
36/// No fetch.
37#[kernel]
38pub fn idle_fetch_req() -> FetchReq {
39    FetchReq { start: false, base: b64(0), word: b32(0), count: b6(0) }
40}
41
42/// Inputs of [`Fetcher`].
43#[derive(Clone, Copy, Debug, PartialEq, Digital, Default)]
44pub struct FetchIn {
45    /// The request.
46    pub req: FetchReq,
47    /// The memory's answers.
48    pub axi: AxiReadResp,
49}
50
51/// Outputs of [`Fetcher`].
52#[derive(Clone, Copy, Debug, PartialEq, Digital, Default)]
53pub struct FetchOut {
54    /// A fetch is in flight.
55    pub busy: bool,
56    /// The fetch finished this cycle; `words` holds the result.
57    pub done: bool,
58    /// The words from the requested one on (those past `count` are
59    /// whatever follows in memory).
60    pub words: [b32; 32],
61    /// Requests to the memory.
62    pub axi: AxiRead,
63}
64
65/// The fetcher's registers.
66#[derive(Clone, Copy, Debug, PartialEq, Digital)]
67pub struct FetchState {
68    /// 0 idle, 2 reading, 3 done (one cycle).
69    pub phase: b2,
70    /// Aligned byte address of the next burst to request.
71    pub addr: b64,
72    /// Beats in the whole fetch (1..=3).
73    pub total: b2,
74    /// Beats of the next burst to request; 0 when all are requested.
75    pub ar_beats: b2,
76    /// Beats of the second burst, if the fetch crosses a 4 KiB boundary.
77    pub ar_next: b2,
78    /// The requested word's position in the first beat (0..16).
79    pub offset: b4,
80    /// Beats received.
81    pub got: b2,
82    /// The beats, as 48 words.
83    pub buf: [b32; 48],
84}
85
86impl Default for FetchState {
87    fn default() -> Self {
88        Self { phase: b2(0), addr: b64(0), total: b2(0), ar_beats: b2(0), ar_next: b2(0), offset: b4(0), got: b2(0), buf: [b32(0); 48] }
89    }
90}
91
92/// One-burst mailbox reader (see the module docs).
93#[derive(Clone, Debug, Synchronous, SynchronousDQ)]
94pub struct Fetcher {
95    state: rhdl_fpga::core::dff::DFF<FetchState>,
96}
97
98impl Default for Fetcher {
99    fn default() -> Self {
100        Self { state: rhdl_fpga::core::dff::DFF::new(FetchState::default()) }
101    }
102}
103
104impl SynchronousIO for Fetcher {
105    type I = FetchIn;
106    type O = FetchOut;
107    type Kernel = fetch_kernel;
108}
109
110/// The fetcher's logic.
111#[kernel]
112pub fn fetch_kernel(_cr: ClockReset, i: FetchIn, q: FetcherQ) -> (FetchOut, FetcherD) {
113    let r = q.state;
114    let mut s = r;
115    let mut o = FetchOut { busy: false, done: false, words: [b32(0); 32], axi: idle_read() };
116    o.busy = r.phase != b2(0);
117    if r.phase == b2(0) && i.req.start {
118        // Byte address of the first word, and its place in its beat.
119        let byte = i.req.base + (i.req.word.resize::<64>() << b64(2));
120        let offset: b4 = (byte >> b64(2)).resize();
121        s.addr = byte & !b64(63);
122        s.offset = offset;
123        // One more beat each time the words run past a line.
124        let end: b7 = offset.resize::<7>() + i.req.count.resize::<7>();
125        let total = if end > b7(32) { b2(3) } else if end > b7(16) { b2(2) } else { b2(1) };
126        s.total = total;
127        // An AXI4 burst may not cross a 4 KiB boundary: split the fetch at
128        // it (Codex strict audit: base 0, word 1023, count 2).
129        let to_boundary: b7 = b7(64) - ((byte >> b64(6)) & b64(63)).resize::<7>();
130        let first = if to_boundary < total.resize::<7>() { to_boundary.resize::<2>() } else { total };
131        s.ar_beats = first;
132        s.ar_next = total - first;
133        s.got = b2(0);
134        s.phase = b2(2);
135    }
136    if r.phase == b2(2) {
137        // Requests: the first burst, then the second if there is one.
138        if r.ar_beats != b2(0) {
139            o.axi.ar_valid = true;
140            o.axi.ar_addr = r.addr;
141            o.axi.ar_len = (r.ar_beats - b2(1)).resize();
142            if i.axi.ar_ready {
143                s.addr = r.addr + (r.ar_beats.resize::<64>() << b64(6));
144                s.ar_beats = r.ar_next;
145                s.ar_next = b2(0);
146            }
147        }
148        // Data: beats arrive in order; the fetch ends on its last beat
149        // (counted, so a split fetch's first r_last does not end it).
150        o.axi.r_ready = true;
151        if i.axi.r_valid {
152            for k in 0..8 {
153                let lo: b32 = i.axi.r_data[k].resize();
154                let hi: b32 = (i.axi.r_data[k] >> b64(32)).resize();
155                if r.got == b2(0) {
156                    s.buf[2 * k] = lo;
157                    s.buf[2 * k + 1] = hi;
158                } else if r.got == b2(1) {
159                    s.buf[16 + 2 * k] = lo;
160                    s.buf[16 + 2 * k + 1] = hi;
161                } else {
162                    s.buf[32 + 2 * k] = lo;
163                    s.buf[32 + 2 * k + 1] = hi;
164                }
165            }
166            s.got = r.got + b2(1);
167            if r.got + b2(1) == r.total {
168                s.phase = b2(3);
169            }
170        }
171    }
172    if r.phase == b2(3) {
173        o.done = true;
174        s.phase = b2(0);
175    }
176    // The 32 words from `offset` on.
177    for k in 0..32 {
178        let mut w = b32(0);
179        for j in 0..16 {
180            if r.offset == b4(j as u128) {
181                w = r.buf[j + k];
182            }
183        }
184        o.words[k] = w;
185    }
186    (o, FetcherD { state: s })
187}
188
189#[cfg(test)]
190mod tests {
191    use super::*;
192    use crate::ports::AxiReadResp;
193
194    /// Drives a fetch against a memory whose word `k` holds `k`. The memory
195    /// panics on a burst that crosses a 4 KiB boundary (AXI4 forbids it).
196    fn fetch(word: u32, count: u32) -> Vec<u32> {
197        let f = Fetcher::default();
198        let mut st = f.init();
199        let cr = |level| rhdl::core::clock_reset(rhdl::core::clock::clock(level), rhdl::core::types::reset::reset(false));
200        let mut queue: std::collections::VecDeque<(u64, u64)> = Default::default();
201        let mut start = true;
202        for _ in 0..200 {
203            let mut resp = AxiReadResp { ar_ready: true, ..Default::default() };
204            if let Some(&(addr, left)) = queue.front() {
205                resp.r_valid = true;
206                resp.r_data = std::array::from_fn(|k| {
207                    let w = addr / 4 + 2 * k as u64;
208                    b64(u128::from(w) | (u128::from(w + 1) << 32))
209                });
210                resp.r_last = left == 0;
211            }
212            let i = FetchIn { req: FetchReq { start, base: b64(0), word: b32(word.into()), count: b6(count.into()) }, axi: resp };
213            let o = f.sim(cr(false), i, &mut st);
214            if o.done {
215                return o.words[..count as usize].iter().map(|w| w.raw() as u32).collect();
216            }
217            if o.axi.ar_valid {
218                let (addr, beats) = (o.axi.ar_addr.raw() as u64, o.axi.ar_len.raw() as u64 + 1);
219                assert_eq!(addr / 4096, (addr + 64 * beats - 1) / 4096, "burst at {addr:#x} of {beats} beats crosses 4 KiB");
220                for b in 0..beats {
221                    queue.push_back((addr + 64 * b, beats - 1 - b));
222                }
223            }
224            if resp.r_valid && o.axi.r_ready {
225                queue.pop_front();
226            }
227            f.sim(cr(true), i, &mut st);
228            start = false;
229        }
230        panic!("fetch of word {word} count {count} never finished");
231    }
232
233    /// Every start word around a 4 KiB boundary, every length: the words
234    /// come back in order and no burst crosses the boundary.
235    #[test]
236    fn fetches_never_cross_4_kib() {
237        for word in 1000..1030u32 {
238            for count in [1u32, 2, 5, 16, 17, 32] {
239                let got = fetch(word, count);
240                assert_eq!(got, (word..word + count).collect::<Vec<_>>(), "word {word} count {count}");
241            }
242        }
243    }
244}
rhdl_ubft/fetch.rs

rhdl_ubft/
fetch.rs
