umsh_ulcp_runtime/
lan_bearer.rs

1//! The board-agnostic half of a LAN bearer: what a radio does with a
2//! datagram from the group before its node sees it.
3//!
4//! The socket, the group membership, and the DNS-SD responder are the
5//! board's, because they are its network stack's. Everything that decides
6//! about a datagram is here, where it can be tested on a host:
7//!
8//! - **Validation.** One UMSH frame per datagram, no header. Empty and
9//!   oversized datagrams, and ones whose version bits name no protocol,
10//!   are dropped and counted. MeshCore v1 frames are recognized and left
11//!   alone: this node does not carry them across the LAN.
12//! - **The ingress limit.** A token bucket over the datagrams handed to
13//!   the node, so a busy LAN cannot spend the radio's airtime faster than
14//!   the operator allows. The duty ceiling still applies behind it.
15//! - **The hand-off.** A [`BridgePort`] the radio multiplexer drains into
16//!   the node as [`RxOrigin::Bearer`](umsh_hal::RxOrigin), and into the
17//!   ULCP session, which hears the LAN as the device does.
18//! - **The egress.** Every UMSH frame the radio puts on the air, as the
19//!   multiplexer reports it, queued for the board's socket task. The LAN
20//!   works as backhaul does: whether a frame is carried is the node's
21//!   ordinary forwarding decision, and the LAN sees what came of it.
22//!
23//! Spec: `docs/protocol/src/lan-bearer.md`.
24
25use core::cell::RefCell;
26use core::task::{Context, Poll};
27
28use embassy_sync::blocking_mutex::{Mutex, raw::RawMutex};
29use embassy_sync::waitqueue::WakerRegistration;
30use heapless::{Deque, Vec};
31use umsh_lan_bearer::Datagram;
32use umsh_radio_loraphy::MAX_PAYLOAD;
33use umsh_ulcp::stats::{Counter, StatsLedger};
34
35use crate::radio_mux::BridgePort;
36
37// The device's default (what PROP_LAN_BEARER_GROUP reports after a reset)
38// and the transport's must be the same group.
39const _: () = {
40    let [a, b, c, d] = umsh_ulcp::lan::DEFAULT_GROUP;
41    let [e, f, g, h] = umsh_lan_bearer::DEFAULT_GROUP;
42    assert!(a == e && b == f && c == g && d == h);
43    assert!(umsh_ulcp::lan::DEFAULT_PORT == umsh_lan_bearer::DEFAULT_PORT);
44};
45
46/// Datagrams waiting for the multiplexer. The node drains them as fast as
47/// it receives anything; a few absorb a burst while it is busy.
48const QUEUE: usize = 4;
49
50/// Transmitted frames waiting for the socket task, which sends each at
51/// once; a few absorb a burst while it is busy.
52const OUTGOING: usize = 4;
53
54/// What became of one datagram.
55#[derive(Clone, Copy, Debug, PartialEq, Eq)]
56pub enum Verdict {
57    /// Handed toward the node.
58    Accepted,
59    /// The bearer is not running.
60    Inactive,
61    /// Empty, oversized, or of an unknown protocol version.
62    Malformed,
63    /// A MeshCore v1 frame, which this node does not carry across the
64    /// LAN.
65    Foreign,
66    /// Over the ingress limit.
67    Limited,
68    /// The queue toward the node was full.
69    Overflow,
70}
71
72/// What a datagram is, judged by its first octet alone—as a radio judges
73/// a reception, before parsing anything.
74pub fn classify(datagram: &[u8]) -> Verdict {
75    match umsh_lan_bearer::classify(datagram) {
76        Datagram::Umsh => Verdict::Accepted,
77        Datagram::MeshCoreV1 => Verdict::Foreign,
78        Datagram::Malformed => Verdict::Malformed,
79    }
80}
81
82/// The ingress limit: a bucket of datagrams, refilled continuously at the
83/// configured rate.
84///
85/// The bucket holds a tenth of a minute's allowance, and never less than
86/// one datagram, so a quiet LAN can deliver a short burst and a busy one
87/// is held to the rate.
88#[derive(Clone, Copy, Debug)]
89pub struct IngressLimit {
90    per_minute: u16,
91    /// Tokens in thousandths of a datagram, so a slow rate still refills
92    /// between datagrams.
93    milli_tokens: u64,
94    last_ms: u64,
95}
96
97impl IngressLimit {
98    pub const fn new(per_minute: u16, now_ms: u64) -> Self {
99        let mut limit = Self {
100            per_minute,
101            milli_tokens: 0,
102            last_ms: now_ms,
103        };
104        limit.milli_tokens = limit.capacity_milli();
105        limit
106    }
107
108    const fn capacity_milli(&self) -> u64 {
109        let per_minute = self.per_minute as u64;
110        let tenth = per_minute / 10;
111        (if tenth == 0 { 1 } else { tenth }) * 1000
112    }
113
114    /// Take one datagram's token, if the bucket has one.
115    pub fn admit(&mut self, now_ms: u64) -> bool {
116        let per_minute = u64::from(self.per_minute.max(1));
117        let elapsed_ms = now_ms.saturating_sub(self.last_ms);
118        // per_minute datagrams per 60,000 ms is per_minute milli-tokens
119        // per 60 ms. Time is consumed only as far as it earned whole
120        // milli-tokens, so frequent calls lose nothing to rounding.
121        let refill = elapsed_ms.saturating_mul(per_minute) / 60;
122        self.last_ms = self.last_ms.saturating_add(refill * 60 / per_minute);
123        self.milli_tokens = self.milli_tokens.saturating_add(refill);
124        if self.milli_tokens >= self.capacity_milli() {
125            self.milli_tokens = self.capacity_milli();
126            self.last_ms = now_ms;
127        }
128        if self.milli_tokens >= 1000 {
129            self.milli_tokens -= 1000;
130            true
131        } else {
132            false
133        }
134    }
135}
136
137struct State {
138    active: bool,
139    limit: IngressLimit,
140    queue: Deque<Vec<u8, MAX_PAYLOAD>, QUEUE>,
141    waker: WakerRegistration,
142    outgoing: Deque<Vec<u8, MAX_PAYLOAD>, OUTGOING>,
143    outgoing_waker: WakerRegistration,
144}
145
146/// The LAN bearer's port into the radio multiplexer.
147///
148/// A board keeps one in a `static`, offers it every datagram its socket
149/// receives, and attaches it to the multiplexer with
150/// [`RxOrigin::Bearer`](umsh_hal::RxOrigin) as its origin.
151pub struct LanPort<M: RawMutex> {
152    state: Mutex<M, RefCell<State>>,
153    stats: Option<&'static StatsLedger>,
154}
155
156impl<M: RawMutex> LanPort<M> {
157    pub const fn new(stats: Option<&'static StatsLedger>) -> Self {
158        Self {
159            state: Mutex::new(RefCell::new(State {
160                active: false,
161                limit: IngressLimit::new(umsh_ulcp::lan::DEFAULT_INGRESS_LIMIT, 0),
162                queue: Deque::new(),
163                waker: WakerRegistration::new(),
164                outgoing: Deque::new(),
165                outgoing_waker: WakerRegistration::new(),
166            })),
167            stats,
168        }
169    }
170
171    /// Start or stop accepting datagrams, at `per_minute`. Stopping drops
172    /// whatever was waiting in either direction: frames from before the
173    /// bearer went away are not worth carrying when it comes back.
174    pub fn configure(&self, active: bool, per_minute: u16, now_ms: u64) {
175        self.state.lock(|cell| {
176            let mut state = cell.borrow_mut();
177            if !active {
178                state.queue.clear();
179                state.outgoing.clear();
180            }
181            if active && (!state.active || state.limit.per_minute != per_minute) {
182                state.limit = IngressLimit::new(per_minute, now_ms);
183            }
184            state.active = active;
185        });
186    }
187
188    /// Judge one datagram from the group and, if it passes, queue it for
189    /// the node.
190    pub fn offer(&self, datagram: &[u8], now_ms: u64) -> Verdict {
191        let verdict = self.state.lock(|cell| {
192            let mut state = cell.borrow_mut();
193            if !state.active {
194                return Verdict::Inactive;
195            }
196            match classify(datagram) {
197                Verdict::Accepted => {}
198                other => return other,
199            }
200            if !state.limit.admit(now_ms) {
201                return Verdict::Limited;
202            }
203            let frame = Vec::from_slice(datagram).expect("classified as fitting");
204            if state.queue.push_back(frame).is_err() {
205                return Verdict::Overflow;
206            }
207            state.waker.wake();
208            Verdict::Accepted
209        });
210        if let Some(stats) = self.stats {
211            match verdict {
212                Verdict::Accepted => stats.bump(Counter::LanRx),
213                Verdict::Limited | Verdict::Overflow => stats.bump(Counter::LanRxLimited),
214                Verdict::Malformed => stats.bump(Counter::LanRxMalformed),
215                Verdict::Inactive | Verdict::Foreign => {}
216            }
217        }
218        verdict
219    }
220}
221
222impl<M: RawMutex> LanPort<M> {
223    /// Wait for the next frame to send to the group.
224    pub async fn outgoing(&self) -> Vec<u8, MAX_PAYLOAD> {
225        core::future::poll_fn(|cx| {
226            self.state.lock(|cell| {
227                let mut state = cell.borrow_mut();
228                state.outgoing_waker.register(cx.waker());
229                match state.outgoing.pop_front() {
230                    Some(frame) => Poll::Ready(frame),
231                    None => Poll::Pending,
232                }
233            })
234        })
235        .await
236    }
237}
238
239impl<M: RawMutex> BridgePort for LanPort<M> {
240    /// Queue a frame the radio put on the air for the group. A MeshCore
241    /// frame is left off, as one from the group is.
242    fn transmitted(&self, data: &[u8]) {
243        if classify(data) != Verdict::Accepted {
244            return;
245        }
246        let outcome = self.state.lock(|cell| {
247            let mut state = cell.borrow_mut();
248            if !state.active {
249                return None;
250            }
251            let frame = Vec::from_slice(data).expect("classified as fitting");
252            // The oldest goes first: the newest frame is the one a
253            // neighbor on the LAN is still waiting for.
254            let dropped = state.outgoing.is_full() && state.outgoing.pop_front().is_some();
255            let _ = state.outgoing.push_back(frame);
256            state.outgoing_waker.wake();
257            Some(dropped)
258        });
259        if let (Some(stats), Some(dropped)) = (self.stats, outcome) {
260            stats.bump(Counter::LanExported);
261            if dropped {
262                stats.bump(Counter::LanExportDropped);
263            }
264        }
265    }
266
267    fn poll_receive(&self, cx: &mut Context<'_>) -> Poll<Vec<u8, MAX_PAYLOAD>> {
268        self.state.lock(|cell| {
269            let mut state = cell.borrow_mut();
270            state.waker.register(cx.waker());
271            match state.queue.pop_front() {
272                Some(frame) => Poll::Ready(frame),
273                None => Poll::Pending,
274            }
275        })
276    }
277}
278
279#[cfg(test)]
280mod tests {
281    use super::*;
282    use embassy_sync::blocking_mutex::raw::NoopRawMutex;
283
284    const UMSH: u8 = 0b1100_0000;
285
286    #[test]
287    fn datagrams_are_judged_by_their_first_octet() {
288        assert_eq!(classify(&[]), Verdict::Malformed);
289        assert_eq!(classify(&[UMSH, 1, 2]), Verdict::Accepted);
290        assert_eq!(classify(&[0b0000_0000, 1]), Verdict::Foreign);
291        assert_eq!(classify(&[0b0100_0000, 1]), Verdict::Malformed);
292        assert_eq!(classify(&[UMSH; MAX_PAYLOAD + 1]), Verdict::Malformed);
293        assert_eq!(classify(&[UMSH; MAX_PAYLOAD]), Verdict::Accepted);
294    }
295
296    #[test]
297    fn the_ingress_limit_allows_a_burst_then_the_rate() {
298        // 60 a minute: a burst of six, then one a second.
299        let mut limit = IngressLimit::new(60, 0);
300        for _ in 0..6 {
301            assert!(limit.admit(0));
302        }
303        assert!(!limit.admit(0));
304        assert!(!limit.admit(999));
305        assert!(limit.admit(1_000));
306        assert!(!limit.admit(1_000));
307        // A long quiet spell refills no more than the bucket holds.
308        for _ in 0..6 {
309            assert!(limit.admit(600_000));
310        }
311        assert!(!limit.admit(600_000));
312    }
313
314    #[test]
315    fn a_slow_limit_still_admits_one() {
316        let mut limit = IngressLimit::new(1, 0);
317        assert!(limit.admit(0));
318        assert!(!limit.admit(59_999));
319        assert!(limit.admit(60_000));
320    }
321
322    #[test]
323    fn the_port_counts_and_queues_what_it_accepts() {
324        static STATS: StatsLedger = StatsLedger::new();
325        let port = LanPort::<NoopRawMutex>::new(Some(&STATS));
326        assert_eq!(port.offer(&[UMSH, 1], 0), Verdict::Inactive);
327        port.configure(true, 10, 0);
328        assert_eq!(port.offer(&[UMSH, 1], 0), Verdict::Accepted);
329        assert_eq!(port.offer(&[UMSH, 2], 0), Verdict::Limited);
330        assert_eq!(port.offer(&[0x40], 0), Verdict::Malformed);
331        assert_eq!(port.offer(&[0x00], 0), Verdict::Foreign);
332        assert_eq!(STATS.get(Counter::LanRx), 1);
333        assert_eq!(STATS.get(Counter::LanRxLimited), 1);
334        assert_eq!(STATS.get(Counter::LanRxMalformed), 1);
335
336        let waker = core::task::Waker::noop();
337        let mut cx = Context::from_waker(waker);
338        match port.poll_receive(&mut cx) {
339            Poll::Ready(frame) => assert_eq!(frame.as_slice(), &[UMSH, 1]),
340            Poll::Pending => panic!("the accepted datagram was not queued"),
341        }
342        assert!(port.poll_receive(&mut cx).is_pending());
343
344        port.offer(&[UMSH, 3], 60_000);
345        port.configure(false, 10, 60_000);
346        assert!(port.poll_receive(&mut cx).is_pending(), "stopping drops it");
347    }
348
349    /// What the radio sends goes to the group while the bearer runs, oldest
350    /// first, newest kept when it overflows; MeshCore stays off the LAN.
351    #[test]
352    fn transmitted_frames_queue_for_the_group() {
353        let port = LanPort::<NoopRawMutex>::new(None);
354        let next = || embassy_futures::block_on(port.outgoing());
355        port.transmitted(&[UMSH, 0]);
356        port.configure(true, 60, 0);
357        port.transmitted(&[0b0000_0000, 1]);
358        for tag in 1..=OUTGOING as u8 + 1 {
359            port.transmitted(&[UMSH, tag]);
360        }
361        for tag in 2..=OUTGOING as u8 + 1 {
362            assert_eq!(next().as_slice(), &[UMSH, tag]);
363        }
364        port.transmitted(&[UMSH, 9]);
365        port.configure(false, 60, 0);
366        port.configure(true, 60, 0);
367        port.transmitted(&[UMSH, 10]);
368        assert_eq!(next().as_slice(), &[UMSH, 10], "stopping drops what waited");
369    }
370}
umsh_ulcp_runtime/lan_bearer.rs

umsh_ulcp_runtime/
lan_bearer.rs
