rhdl_ubft/ctb/
broadcaster.rs

1//! The CTB broadcaster (docs/rhdl_ubft-ctb.md): upstream
2//! `tail-cb/broadcaster.hpp` as a step kernel.
3//!
4//! - Each message gets the next index and goes out as one data record to
5//!   every receiver (the transport fans it out).
6//! - Every message is kept for the tail. While the slow path is on, each
7//!   kept message is digested, signed under the CTB context (role 2) and the
8//!   signature sent, including messages admitted before the slow path was
9//!   turned on (a late fallback must be able to decide them). A message
10//!   pushed out of the tail before its signature is done is dropped from
11//!   signing, as upstream, since no receiver still holds it.
12//! - Every response carries `(slot, idx)`; a stale one is dropped.
13//! - No `throw`: an index at the wrap guard is a fault and stops admission.
14use rhdl::prelude::*;
15
16use super::ctb_context;
17
18/// One-time configuration.
19#[derive(Clone, Copy, Debug, PartialEq, Digital, Default)]
20pub struct BcConfig {
21    /// Configure this cycle (clears all state; indices restart at 0).
22    pub init: bool,
23    /// This replica (the broadcaster id).
24    pub me: b5,
25    /// The run (nonzero).
26    pub session: b64,
27}
28
29/// A message to broadcast.
30#[derive(Clone, Copy, Debug, PartialEq, Digital)]
31pub struct BcMsg<const P: usize> {
32    pub valid: bool,
33    pub len: b5,
34    pub payload: [b32; P],
35}
36
37/// Inputs.
38#[derive(Clone, Copy, Debug, PartialEq, Digital)]
39pub struct BcIn<const P: usize> {
40    pub cfg: BcConfig,
41    pub msg: BcMsg<P>,
42    /// Sign messages from now on (the slow path).
43    pub slow: bool,
44    /// The transport takes the data record.
45    pub data_ready: bool,
46    pub digest_ready: bool,
47    pub sign_ready: bool,
48    /// The transport takes the signature record.
49    pub sig_ready: bool,
50    /// A digest (see [`super::receiver::DigestIn`]).
51    pub digest: super::receiver::DigestIn,
52    /// A signature for `(slot, idx)`.
53    pub signed: BcSigned,
54}
55
56/// A signature from the service.
57#[derive(Clone, Copy, Debug, PartialEq, Digital, Default)]
58pub struct BcSigned {
59    pub valid: bool,
60    /// The generation of the request (see [`BcSignReq::epoch`]).
61    pub epoch: b32,
62    pub slot: b4,
63    pub idx: b32,
64    pub sig: [b64; 8],
65}
66
67/// The data record to send to every receiver.
68#[derive(Clone, Copy, Debug, PartialEq, Digital)]
69pub struct BcData<const P: usize> {
70    pub valid: bool,
71    pub idx: b32,
72    pub len: b5,
73    pub payload: [b32; P],
74}
75
76/// Sign the CTB context for `(slot, idx)`.
77#[derive(Clone, Copy, Debug, PartialEq, Digital, Default)]
78pub struct BcSignReq {
79    pub valid: bool,
80    /// The broadcaster's generation (bumped at every configuration);
81    /// responses of an older one are dropped.
82    pub epoch: b32,
83    pub slot: b4,
84    pub idx: b32,
85    /// The 48 signed bytes.
86    pub context: [b64; 6],
87}
88
89/// The signature record to send to every receiver.
90#[derive(Clone, Copy, Debug, PartialEq, Digital, Default)]
91pub struct BcSigOut {
92    pub valid: bool,
93    pub idx: b32,
94    pub sig: [b64; 8],
95}
96
97/// Outputs.
98#[derive(Clone, Copy, Debug, PartialEq, Digital)]
99pub struct BcOut<const P: usize> {
100    /// `msg` is taken this cycle (exactly when `data` is taken).
101    pub msg_ready: bool,
102    /// The data record (the message, with its index).
103    pub data: BcData<P>,
104    pub digest_req: super::receiver::DigestReq<P>,
105    pub sign: BcSignReq,
106    pub sig: BcSigOut,
107    /// Messages sent so far (the next index).
108    pub next_idx: b32,
109    pub fault: bool,
110}
111
112/// A message in the tail, while it is being signed.
113#[derive(Clone, Copy, Debug, PartialEq, Digital)]
114pub struct BcEntry<const P: usize> {
115    pub valid: bool,
116    pub idx: b32,
117    pub len: b5,
118    pub payload: [b32; P],
119    pub dig_sent: bool,
120    pub dig_have: bool,
121    pub digest: [b64; 4],
122    pub sign_sent: bool,
123    pub sig_have: bool,
124    pub sig: [b64; 8],
125}
126
127/// State.
128#[derive(Clone, Copy, Debug, PartialEq, Digital)]
129pub struct BcState<const T: usize, const P: usize> {
130    pub configured: bool,
131    pub epoch: b32,
132    pub me: b5,
133    pub session: b64,
134    pub next_idx: b32,
135    pub fault: bool,
136    /// The slow path is on (registered, so requests depend on registers).
137    pub slow: bool,
138    pub e: [BcEntry<P>; T],
139}
140
141impl<const P: usize> BcEntry<P> {
142    pub fn empty() -> Self {
143        Self { valid: false, idx: b32(0), len: b5(0), payload: [b32(0); P], dig_sent: false, dig_have: false, digest: [b64(0); 4], sign_sent: false, sig_have: false, sig: [b64(0); 8] }
144    }
145}
146
147impl<const T: usize, const P: usize> BcState<T, P> {
148    pub fn idle() -> Self {
149        Self { configured: false, epoch: b32(0), me: b5(0), session: b64(0), next_idx: b32(0), fault: false, slow: false, e: [BcEntry::empty(); T] }
150    }
151}
152
153impl<const P: usize> Default for BcMsg<P> {
154    fn default() -> Self {
155        Self { valid: false, len: b5(0), payload: [b32(0); P] }
156    }
157}
158
159impl<const P: usize> Default for BcIn<P> {
160    fn default() -> Self {
161        Self {
162            cfg: BcConfig::default(),
163            msg: BcMsg::default(),
164            slow: false,
165            data_ready: false,
166            digest_ready: false,
167            sign_ready: false,
168            sig_ready: false,
169            digest: Default::default(),
170            signed: BcSigned::default(),
171        }
172    }
173}
174
175impl<const T: usize, const P: usize> Default for BcState<T, P> {
176    fn default() -> Self {
177        Self::idle()
178    }
179}
180
181/// A zero entry, for kernels.
182#[kernel]
183pub fn empty_bc_entry<const P: usize>() -> BcEntry<P> {
184    BcEntry::<P> { valid: false, idx: b32(0), len: b5(0), payload: [b32(0); P], dig_sent: false, dig_have: false, digest: [b64(0); 4], sign_sent: false, sig_have: false, sig: [b64(0); 8] }
185}
186
187/// One cycle of the broadcaster.
188#[kernel]
189pub fn bc_step<const T: usize, const P: usize>(i: BcIn<P>, r: BcState<T, P>) -> (BcOut<P>, BcState<T, P>) {
190    let mut s = r;
191    let mask = b32(T as u128) - b32(1);
192    if i.cfg.init {
193        s.configured = i.cfg.session != b64(0);
194        s.epoch = r.epoch + b32(1);
195        s.me = i.cfg.me;
196        s.session = i.cfg.session;
197        s.next_idx = b32(0);
198        s.fault = i.cfg.session == b64(0);
199        for k in 0..T {
200            s.e[k] = empty_bc_entry::<P>();
201        }
202    }
203    let run = r.configured && !r.fault && !i.cfg.init;
204    s.slow = i.slow;
205    let mut o = BcOut::<P> {
206        msg_ready: false,
207        data: BcData::<P> { valid: false, idx: b32(0), len: b5(0), payload: [b32(0); P] },
208        digest_req: super::receiver::DigestReq::<P> { valid: false, epoch: b32(0), slot: b4(0), idx: b32(0), len: b5(0), payload: [b32(0); P] },
209        sign: BcSignReq { valid: false, epoch: b32(0), slot: b4(0), idx: b32(0), context: [b64(0); 6] },
210        sig: BcSigOut { valid: false, idx: b32(0), sig: [b64(0); 8] },
211        next_idx: r.next_idx,
212        fault: r.fault,
213    };
214
215    // Responses (dropped unless they match the entry still in the tail).
216    if run && i.digest.valid && i.digest.epoch == r.epoch {
217        let k = i.digest.slot;
218        if s.e[k].valid && s.e[k].idx == i.digest.idx && s.e[k].dig_sent {
219            s.e[k].dig_have = true;
220            s.e[k].digest = i.digest.digest;
221        }
222    }
223    if run && i.signed.valid && i.signed.epoch == r.epoch {
224        let k = i.signed.slot;
225        if s.e[k].valid && s.e[k].idx == i.signed.idx && s.e[k].sign_sent {
226            s.e[k].sig_have = true;
227            s.e[k].sig = i.signed.sig;
228        }
229    }
230
231    // Requests, from registered state.
232    if run {
233        let mut d_done = false;
234        let mut g_done = false;
235        let mut x_done = false;
236        for k in 0..T {
237            let e = r.e[k];
238            let slot = b4(k as u128);
239            if !d_done && r.slow && e.valid && !e.dig_sent {
240                d_done = true;
241                o.digest_req = super::receiver::DigestReq::<P> { valid: true, epoch: r.epoch, slot: slot, idx: e.idx, len: e.len, payload: e.payload };
242                if i.digest_ready {
243                    s.e[k].dig_sent = true;
244                }
245            }
246            if !g_done && r.slow && e.valid && e.dig_have && !e.sign_sent {
247                g_done = true;
248                o.sign = BcSignReq { valid: true, epoch: r.epoch, slot: slot, idx: e.idx, context: ctb_context(r.session, r.me, e.idx, e.digest) };
249                if i.sign_ready {
250                    s.e[k].sign_sent = true;
251                }
252            }
253            if !x_done && e.valid && e.sig_have {
254                x_done = true;
255                o.sig = BcSigOut { valid: true, idx: e.idx, sig: e.sig };
256                if i.sig_ready {
257                    // Sent: the entry is done.
258                    s.e[k] = empty_bc_entry::<P>();
259                }
260            }
261        }
262    }
263
264    // A new message: its data record goes out as it is taken.
265    if run {
266        // A message that cannot be sent (too long, or at the wrap guard) is
267        // taken and refused: nothing goes out, and the fault stops admission.
268        let bad = r.next_idx >= b32(0xFFFF_0000) || i.msg.len > b5(P as u128);
269        o.data = BcData::<P> { valid: i.msg.valid && !bad, idx: r.next_idx, len: i.msg.len, payload: i.msg.payload };
270        o.msg_ready = i.data_ready || bad;
271        if i.msg.valid && (i.data_ready || bad) {
272            if bad {
273                s.fault = true;
274            } else {
275                s.next_idx = r.next_idx + b32(1);
276                // Kept whether or not the slow path is on now. Takes the slot
277                // over; an older message still there has left the tail and
278                // is no longer signed.
279                let k: b4 = (r.next_idx & mask).resize();
280                let mut n = empty_bc_entry::<P>();
281                n.valid = true;
282                n.idx = r.next_idx;
283                n.len = i.msg.len;
284                n.payload = i.msg.payload;
285                s.e[k] = n;
286            }
287        }
288    }
289    (o, s)
290}
rhdl_ubft/ctb/broadcaster.rs

rhdl_ubft/ctb/
broadcaster.rs
