diff --git a/crates/ziggurat-driver/src/zigbee_stack.rs b/crates/ziggurat-driver/src/zigbee_stack.rs index 584d417..969be98 100644 --- a/crates/ziggurat-driver/src/zigbee_stack.rs +++ b/crates/ziggurat-driver/src/zigbee_stack.rs @@ -98,8 +98,6 @@ pub enum EnqueueError { pub enum DeliveryError { #[error("route discovery timed out")] RouteDiscoveryTimeout, - #[error("route not active after discovery completed")] - RouteInactiveAfterDiscovery, #[error("next hop {next_hop:?} did not ACK")] NwkNoAck { next_hop: Ieee802154Address }, #[error("transmit rejected due to CCA failure")] diff --git a/crates/ziggurat-driver/src/zigbee_stack/joining.rs b/crates/ziggurat-driver/src/zigbee_stack/joining.rs index f35450e..16e8bec 100644 --- a/crates/ziggurat-driver/src/zigbee_stack/joining.rs +++ b/crates/ziggurat-driver/src/zigbee_stack/joining.rs @@ -296,11 +296,12 @@ impl ZigbeeStack { return; } - // Routers resolve their own conflicts after hearing the notification; our - // mapping for the address is ambiguous until the keeper re-announces + // Routers resolve their own conflicts after hearing the notification. Until + // one announces itself we cannot tell which device a frame to the address + // would reach, so sends held for it fail let mut core = self.core(); core.nib.address_map.forget_address(address); - core.nib.routing.remove_route(address); + self.forget_nwk_state_in(core, address); } /// The address-conflict report reactor: a single long-lived task owning the diff --git a/crates/ziggurat-driver/src/zigbee_stack/nwk.rs b/crates/ziggurat-driver/src/zigbee_stack/nwk.rs index 36c5503..48dddea 100644 --- a/crates/ziggurat-driver/src/zigbee_stack/nwk.rs +++ b/crates/ziggurat-driver/src/zigbee_stack/nwk.rs @@ -26,9 +26,10 @@ use ziggurat_zigbee::nwk::next_hop::{self, NextHop}; use ziggurat_zigbee::sends::{Unicast as UnicastRecord, UnroutedUnicast}; use super::{ - AddrConflictSource, Broadcast, Confirm, DeliveryError, EnqueueError, Frame, MAX_DEPTH, Nib, - NwkSecurityMode, PROTOCOL_VERSION, RouteDirective, Schedule, SendActions, SendHandle, SendId, - SendMode, SendSlot, TxPolicy, TxPriority, Unicast, ZigbeeCore, ZigbeeNotification, ZigbeeStack, + AddrConflictSource, Broadcast, Confirm, CoreGuard, DeliveryError, EnqueueError, Frame, + MAX_DEPTH, Nib, NwkSecurityMode, PROTOCOL_VERSION, RouteDirective, Schedule, SendActions, + SendHandle, SendId, SendMode, SendSlot, TxPolicy, TxPriority, Unicast, ZigbeeCore, + ZigbeeNotification, ZigbeeStack, }; impl ZigbeeStack { @@ -45,17 +46,17 @@ impl ZigbeeStack { return; } - // The device can be reached again, so sends held for it are kept. The route - // request they wait on went out before it joined. With that discovery gone, the - // timer pass starts a new one. - core.nib.routing.forget(self.state.network_address, nwk); - self.commit_sends(core, SendActions::new()); + self.forget_nwk_state_in(core, nwk); } - /// The device at `nwk` left, so what we know about the address belongs to a session - /// that is over. Held sends fail: nothing will answer a route request for it. + /// The device at `nwk` left or joined again, or two devices claim the address, so + /// what we know about it no longer holds. Held sends fail: they wait on a route + /// request the device never heard, and forgetting the address drops that discovery. pub(super) fn forget_nwk_state(&self, nwk: Nwk) { - let mut core = self.core(); + self.forget_nwk_state_in(self.core(), nwk); + } + + pub(super) fn forget_nwk_state_in(&self, mut core: CoreGuard<'_>, nwk: Nwk) { core.nib.routing.forget(self.state.network_address, nwk); let actions = core.sends.fail_unrouted(nwk); self.commit_sends(core, actions); @@ -247,18 +248,24 @@ impl ZigbeeStack { NwkCommand::RouteRecord(cmd) => { tracing::trace!("Route record command frame received: {cmd:?}"); let source = nwk_frame.nwk_header.source; - let changed = self + let change = self .core() .nib .routing .store_route_record(source, cmd.relays.clone()); - if changed { + if change.source { self.push_notification(ZigbeeNotification::RouteRecord { destination: source, relays: cmd.relays.clone(), }); } + + // Sends held for the device, or for a relay on its path, can now be + // source routed + if change.any { + self.send_timer_wake.notify_one(); + } } NwkCommand::RouteRequest(cmd) => { self.handle_route_request(nwk_frame, cmd.clone(), sender_nwk); @@ -419,8 +426,6 @@ impl ZigbeeStack { tracing::debug!("Holding frame until a route to {destination:?} is known"); let slot = Arc::new(SendSlot::default()); - // The first unrouted send to a destination starts its discovery - let discovering = core.sends.has_unrouted(destination); let (id, mut actions) = core.sends.enqueue_unrouted_unicast( UnroutedUnicast { frame: Frame::nwk(nwk_frame, security), @@ -432,8 +437,22 @@ impl ZigbeeStack { now, &self.tunables, ); - if !discovering { - actions.extend(self.start_route_discovery(&mut core, destination)); + + // The routing table starts a discovery unless the send can wait on + // one: a discovery of ours still underway, or one the timer pass starts + let route_request_identifier = core.nib.routing.want_route( + self.state.network_address, + destination, + now, + self.tunables.route_discovery_time(), + self.tunables.pending_route_discovery_attempts(), + ); + if let Some(route_request_identifier) = route_request_identifier { + actions.extend(self.broadcast_discovery_request( + &mut core, + destination, + route_request_identifier, + )); } (SendHandle::new(id, slot), actions) diff --git a/crates/ziggurat-driver/src/zigbee_stack/route.rs b/crates/ziggurat-driver/src/zigbee_stack/route.rs index db8ed19..4f41d44 100644 --- a/crates/ziggurat-driver/src/zigbee_stack/route.rs +++ b/crates/ziggurat-driver/src/zigbee_stack/route.rs @@ -35,6 +35,7 @@ impl ZigbeeStack { // responder seeds the first hop and each relay adds the link it forwards across. let updated_path_cost = route_reply_cmd.path_cost; + let now = self.core_now(); let outcome = self.core().nib.routing.accept_route_reply( self.state.network_address, route_reply_cmd.originator_nwk, @@ -42,6 +43,7 @@ impl ZigbeeStack { route_reply_cmd.responder_nwk, nwk_frame.nwk_header.source, updated_path_cost, + now, ); let (next_hop_nwk, path_cost) = match outcome.disposition { @@ -435,24 +437,16 @@ impl ZigbeeStack { } } - /// Begin or restart ad-hoc route discovery toward a destination: the routing entry - /// enters `DiscoveryUnderway` and a route request is broadcast. The send timer - /// watches the discovery on behalf of the unrouted sends to it; this only kicks it - /// off. Under a core lock the caller holds and commits. - pub(super) fn start_route_discovery( + /// Broadcast the route request of a discovery the routing table started or + /// restarted toward a destination. Under a core lock the caller holds and commits. + pub(super) fn broadcast_discovery_request( &self, core: &mut ZigbeeCore, destination: Nwk, + route_request_identifier: u8, ) -> SendActions { tracing::debug!("Sending route discovery for NWK {destination:?}"); - let route_request_identifier = core.nib.routing.begin_discovery( - self.state.network_address, - destination, - self.core_now(), - self.tunables.route_discovery_time(), - ); - // If we know the EUI64 corresponding to the NWK, use it let destination_eui64 = core.nib.address_map.eui64_for(destination); diff --git a/crates/ziggurat-driver/src/zigbee_stack/sends.rs b/crates/ziggurat-driver/src/zigbee_stack/sends.rs index 636bf4d..fe977b7 100644 --- a/crates/ziggurat-driver/src/zigbee_stack/sends.rs +++ b/crates/ziggurat-driver/src/zigbee_stack/sends.rs @@ -18,7 +18,6 @@ use ziggurat_zigbee::sends::{Action, Broadcast as BroadcastRecord, Unicast as Un use crate::runtime::Runtime; -use super::routing::Status as RouteStatus; use super::{ Confirm, CoreGuard, DeliveryError, Frame, Nib, NwkSecurityMode, RouteDirective, Schedule, SendActions, SendHandle, SendId, SendMode, SendSlot, Transmit, TxPolicy, ZigbeeCore, @@ -178,32 +177,41 @@ impl ZigbeeStack { .map(Ieee802154Address::Eui64) } - /// The next hop of an unrouted frame once its route is known, and the address it - /// polls under if it is a sleepy child. The route can have vanished in the race, - /// which fails the send. - fn resolve_routed_unicast( + /// Resolve the route of an unrouted frame again: its next hop and the address it + /// polls under if it is a sleepy child. `None` means no route yet. + fn resolve_unrouted_unicast( &self, nib: &Nib, frame: &mut Frame, now: CoreInstant, - ) -> Result<(Nwk, Option), DeliveryError> { + ) -> Option<(Nwk, Option)> { let Frame::Nwk { nwk_frame, .. } = frame else { unreachable!("only NWK unicasts await a route") }; + // A forced discovery holds the frame only while it is underway. Once it is + // answered or has lapsed, the frame takes any route like every other frame + let mut policy = self.next_hop_policy(); + policy.force_route_discovery &= nib.routing.own_discovery_underway( + self.state.network_address, + nwk_frame.nwk_header.destination, + now, + ); + let resolved = next_hop::resolve( &nib.routing, &nib.neighbors, nwk_frame, &SendMode::Route(RouteDirective::StackDecides), - self.next_hop_policy(), + policy, now, ); match resolved { - NextHop::Resolved(next_hop) => Ok((next_hop, Self::poll_address(nib, next_hop))), - NextHop::NeedDiscovery | NextHop::Discard => { - Err(DeliveryError::RouteInactiveAfterDiscovery) + NextHop::Resolved(next_hop) => Some((next_hop, Self::poll_address(nib, next_hop))), + NextHop::NeedDiscovery => None, + NextHop::Discard => { + unreachable!("a frame is only held for a route if it allows discovery") } } } @@ -298,16 +306,15 @@ impl ZigbeeStack { } } - /// One timer pass: settle the broadcasts whose passive-ack quorum was heard, route - /// the sends whose route became active, restart or fail the discoveries that - /// lapsed, then run the table's timed transitions. Returns when the next pass is - /// due: the table's next deadline, or that of a route discovery unrouted sends wait - /// on, which lives in the routing table. + /// One timer pass: settle the broadcasts whose passive-ack quorum was heard, run + /// the send tables' pass against the routing and neighbor tables, then let the + /// routing table start a discovery for, or give up on, each destination a send + /// still waits on with none underway. Returns when the next pass is due. fn drive_sends(&self) -> Option { let now = self.core_now(); let mut guard = self.core(); - // A plain `&mut ZigbeeCore`, so the closure below borrows `nib` beside `sends` + // A plain `&mut ZigbeeCore`, so the closures below borrow `nib` beside `sends` let core = &mut *guard; let mut actions = SendActions::new(); @@ -317,37 +324,38 @@ impl ZigbeeStack { } } - for destination in core.sends.unrouted_destinations() { - if core.nib.routing.route_status(destination) == Some(RouteStatus::Active) { - actions.extend(core.sends.route_unicasts( - destination, - now, - &self.tunables, - |send| self.resolve_routed_unicast(&core.nib, send, now), - )); - } else if core - .nib - .routing - .discovery_deadline(destination, now) - .is_none() - { - if core.sends.retry_discovery(destination) { - actions.extend(self.start_route_discovery(core, destination)); - } else { - core.nib.routing.mark_discovery_failed(destination); - actions.extend(core.sends.fail_unrouted(destination)); - } - } + actions.extend(core.sends.tick(now, &self.tunables, |frame| { + self.resolve_unrouted_unicast(&core.nib, frame, now) + })); + + let discoveries = core.nib.routing.tick_discoveries( + self.state.network_address, + now, + self.tunables.route_discovery_time(), + |destination| core.sends.has_unrouted(destination), + ); + + for (destination, route_request_identifier) in discoveries.restarted { + actions.extend(self.broadcast_discovery_request( + core, + destination, + route_request_identifier, + )); } - actions.extend(core.sends.tick(now)); + for destination in discoveries.unreachable { + actions.extend(core.sends.fail_unrouted(destination)); + } - let discoveries = core - .sends - .unrouted_destinations() - .into_iter() - .filter_map(|destination| core.nib.routing.discovery_deadline(destination, now)); - let next_wake = core.sends.next_wake().into_iter().chain(discoveries).min(); + let next_wake = [ + core.sends.next_wake(), + core.nib + .routing + .next_discovery_lapse(self.state.network_address, now), + ] + .into_iter() + .flatten() + .min(); // This pass is the timer task's own: waking the timer here would make it spin self.run_send_actions(guard, actions); diff --git a/crates/ziggurat-protocol/src/bridge.rs b/crates/ziggurat-protocol/src/bridge.rs index aeb863e..f24060c 100644 --- a/crates/ziggurat-protocol/src/bridge.rs +++ b/crates/ziggurat-protocol/src/bridge.rs @@ -420,7 +420,6 @@ pub(crate) const fn send_status(result: &Result<(), DeliveryError>) -> SendStatu Ok(()) => SendStatus::Success, Err(err) => match err { DeliveryError::RouteDiscoveryTimeout => SendStatus::RouteDiscoveryTimeout, - DeliveryError::RouteInactiveAfterDiscovery => SendStatus::RouteInactiveAfterDiscovery, DeliveryError::NwkNoAck { .. } => SendStatus::NwkNoAck, DeliveryError::CcaFailure => SendStatus::CcaFailure, DeliveryError::TransmitFailed(_) => SendStatus::TransmitFailed, diff --git a/crates/ziggurat-protocol/src/wire.rs b/crates/ziggurat-protocol/src/wire.rs index c44869b..845cf67 100644 --- a/crates/ziggurat-protocol/src/wire.rs +++ b/crates/ziggurat-protocol/src/wire.rs @@ -713,11 +713,14 @@ pub enum SendStatus { Success = 0, RouteDiscoveryTimeout = 1, RouteDiscoveryNoEntry = 2, + /// No longer sent. Kept so a confirmation from older firmware still decodes. RouteInactiveAfterDiscovery = 3, NwkNoAck = 4, CcaFailure = 5, TransmitFailed = 6, ApsAckTimeout = 7, + /// No longer sent. Kept so a confirmation from older firmware still decodes. + BroadcastQuorumNotReached = 8, IndirectExpired = 9, FrameBudgetExhausted = 10, Cancelled = 11, diff --git a/crates/ziggurat-zigbee/src/nwk/routing.rs b/crates/ziggurat-zigbee/src/nwk/routing.rs index d927f0c..da19c9e 100644 --- a/crates/ziggurat-zigbee/src/nwk/routing.rs +++ b/crates/ziggurat-zigbee/src/nwk/routing.rs @@ -8,6 +8,9 @@ use ziggurat_ieee_802154::types::Nwk; use crate::nwk::frame::BROADCAST_ALL_ROUTERS_AND_COORDINATOR; +#[cfg(test)] +mod tests; + pub type RouteRequestId = u8; const UNKNOWN_NEXT_HOP: Nwk = Nwk(0xFFFF); @@ -16,6 +19,8 @@ const UNKNOWN_NEXT_HOP: Nwk = Nwk(0xFFFF); pub enum Status { Active = 0, DiscoveryUnderway = 1, + /// Never set: the spec defines the value but an unanswered discovery deletes its + /// entry (3.6.4.5.1.7). DiscoveryFailed = 2, Inactive = 3, } @@ -102,6 +107,17 @@ impl TableEntry { } } +/// What storing a route record changed. +#[derive(Debug, Clone, Copy, PartialEq, Eq)] +pub struct RouteRecordChange { + /// The source route to the record's sender changed. Only these are worth the + /// client persisting. + pub source: bool, + /// Some stored source route changed: the sender's, or that of a relay on its path. + /// A send held for any of them may now have a route. + pub any: bool, +} + #[derive(Debug)] pub struct DiscoveryEntry { /// A sequence number for a route request command frame that is incremented each time @@ -121,7 +137,8 @@ pub struct DiscoveryEntry { /// device. pub forward_cost: u8, /// The accumulated path cost from the current device to the destination device. - pub residual_cost: u8, + /// `None` until a route reply is accepted. + pub residual_cost: Option, /// A countdown timer indicating the number of milliseconds until route discovery /// expires. The initial value is `nwkcRouteDiscoveryTime`. pub expiration_time: Instant, @@ -179,6 +196,19 @@ pub struct RouteReplyOutcome { pub update: Option, } +/// What a timer pass decided about the route discoveries we started. +#[derive(Debug, Default, PartialEq, Eq)] +#[must_use = "restarted discoveries need their route request broadcast, and frames \ + held for an unreachable destination must fail"] +pub struct DiscoveryTick { + /// Discoveries started for frames that wait with none underway: the destination, + /// and the identifier to put in the new route request. + pub restarted: Vec<(Nwk, RouteRequestId)>, + /// Destinations given up on, with no restart left. Frames held for a route to them + /// fail. + pub unreachable: Vec, +} + /// The NWK routing layer: route table, route discovery table, and route record table, /// with the decision logic for route request/reply processing. #[derive(Debug, Default)] @@ -193,6 +223,12 @@ pub struct Routing { discovery_table: FlatMap<(Nwk, RouteRequestId), DiscoveryEntry>, route_record_table: FlatMap>, + /// The destinations our frames wait for a route to, with how many more discoveries + /// each may get. An entry lasts from the first frame that wants the route until a + /// timer pass finds no frame waiting, or gives up. The discoveries themselves are + /// only in the two spec tables above. + discoveries_left: FlatMap, + /// Implied from the spec: "notice that this 8-bit identifier is distinct from the /// 16-bit Routing Sequence Number. The former is used to discern route requests /// originating in a particular router; the latter is used to identify stale routing @@ -241,10 +277,116 @@ impl Routing { }) } - pub fn mark_discovery_failed(&mut self, destination: Nwk) { - if let Some(entry) = self.route_table.get_mut(&destination) { - entry.status = Status::DiscoveryFailed; + /// When the last live discovery we started toward a destination lapses. + fn own_discovery_lapse( + &self, + own_address: Nwk, + destination: Nwk, + now: Instant, + ) -> Option { + self.discovery_table + .values() + .filter(|entry| { + entry.source_address == own_address + && entry.destination_address == destination + && entry.expiration_time > now + }) + .map(|entry| entry.expiration_time) + .max() + } + + /// Spec 3.6.4.3: a frame waits on a discovery that is underway only if we started + /// it. The status alone does not say so: relaying another router's route request + /// sets it too, even after our own discovery was answered and its route lost. So + /// a discovery of ours must also be live and have no reply yet. + pub fn own_discovery_underway(&self, own_address: Nwk, destination: Nwk, now: Instant) -> bool { + self.route_status(destination) == Some(Status::DiscoveryUnderway) + && self.discovery_table.values().any(|entry| { + entry.source_address == own_address + && entry.destination_address == destination + && entry.expiration_time > now + && entry.residual_cost.is_none() + }) + } + + /// One of our frames needs a route to `destination`. Returns the identifier to put + /// in the route request to broadcast, or `None` when the frame waits on a discovery + /// that exists or that the next timer pass starts. A discovery we started that is + /// still underway counts as the first of the frame's `attempts`: its route request + /// is on the network whatever became of the frames that wanted it. + pub fn want_route( + &mut self, + own_address: Nwk, + destination: Nwk, + now: Instant, + route_discovery_time: Duration, + attempts: u8, + ) -> Option { + if self.discoveries_left.contains_key(&destination) { + return None; + } + + self.discoveries_left + .insert(destination, attempts.saturating_sub(1)); + + if self.own_discovery_underway(own_address, destination, now) { + return None; + } + + Some(self.begin_discovery(own_address, destination, now, route_discovery_time)) + } + + /// Decide what becomes of each destination our frames wait for a route to. + /// `wanted` says whether a frame still waits, after the frames the routing table + /// can now route have left. + pub fn tick_discoveries( + &mut self, + own_address: Nwk, + now: Instant, + route_discovery_time: Duration, + wanted: impl Fn(Nwk) -> bool, + ) -> DiscoveryTick { + let mut tick = DiscoveryTick::default(); + + self.expire_discoveries(now); + + let destinations: Vec = self.discoveries_left.keys().copied().collect(); + + for destination in destinations { + if !wanted(destination) { + self.discoveries_left.remove(&destination); + continue; + } + + if self.own_discovery_underway(own_address, destination, now) { + continue; + } + + // A frame waits and no discovery of ours is underway: it lapsed, or its + // route was lost before the frame could use it + let discoveries_left = self.discoveries_left.get_mut(&destination).unwrap(); + + if *discoveries_left > 0 { + *discoveries_left -= 1; + let request_id = + self.begin_discovery(own_address, destination, now, route_discovery_time); + tick.restarted.push((destination, request_id)); + } else { + self.discoveries_left.remove(&destination); + tick.unreachable.push(destination); + } } + + tick + } + + /// When the next discovery a frame of ours waits on lapses, or `None` when none + /// is live. + pub fn next_discovery_lapse(&self, own_address: Nwk, now: Instant) -> Option { + self.discoveries_left + .keys() + .filter_map(|destination| self.own_discovery_lapse(own_address, *destination, now)) + .min() } /// A frame was forwarded along the route: bump its usage counters. @@ -279,8 +421,14 @@ impl Routing { } } - /// Remove the route to a destination, returning whether an entry was removed. + /// Remove the route to a destination, returning whether an entry was removed. An + /// entry in `DiscoveryUnderway` holds no route and stays: without it the reply to + /// the discovery is dropped (spec 3.6.4.5.2). pub fn remove_route(&mut self, destination: Nwk) -> bool { + if self.route_status(destination) == Some(Status::DiscoveryUnderway) { + return false; + } + self.route_table.remove(&destination).is_some() } @@ -291,6 +439,7 @@ impl Routing { // unknown responder. That router gets no route through us for that discovery. self.route_table.remove(&nwk); self.route_record_table.remove(&nwk); + self.discoveries_left.remove(&nwk); // Only our own discoveries toward it are dropped. Discoveries we relay stay // until they expire: they are the only duplicate check for late copies of @@ -308,22 +457,33 @@ impl Routing { self.route_table.insert(destination, entry); } - /// Returns whether the stored relay list for `source` changed, so the client only - /// persists a genuinely new source route (most route records are redundant). - pub fn store_route_record(&mut self, source: Nwk, relays: Vec) -> bool { + /// Store the route record `source` sent, and the routes to its relays that follow + /// from it. Most route records are redundant, so the result says what changed. + pub fn store_route_record(&mut self, source: Nwk, relays: Vec) -> RouteRecordChange { + let mut relay_changed = false; + // Spec 3.6.4.5.5: the new route also replaces any existing source routes to // the intermediary relays. The relays between relay `i` and us form its own // route; the last relay delivered the record to us directly. for i in 1..relays.len() { - self.route_record_table - .insert(relays[i - 1], relays[i..].to_vec()); + relay_changed |= self.replace_route_record(relays[i - 1], relays[i..].to_vec()); } if let Some(&last) = relays.last() { - self.route_record_table.insert(last, Vec::new()); + relay_changed |= self.replace_route_record(last, Vec::new()); } - let changed = self.route_record_table.get(&source) != Some(&relays); - self.route_record_table.insert(source, relays); + let source_changed = self.replace_route_record(source, relays); + + RouteRecordChange { + source: source_changed, + any: source_changed || relay_changed, + } + } + + /// Store one source route, returning whether it differs from the one it replaces. + fn replace_route_record(&mut self, destination: Nwk, relays: Vec) -> bool { + let changed = self.route_record_table.get(&destination) != Some(&relays); + self.route_record_table.insert(destination, relays); changed } @@ -332,16 +492,13 @@ impl Routing { self.route_record_table.remove(&destination).is_some() } - /// Spec 3.6.4.8.1: a router reported it could not deliver a frame we originated - /// toward `destination` (`LinkFailure`, `SourceRouteFailure` or a legacy - /// equivalent). Both the routing entry and the stored route record go: a record - /// that reached the failing router as a plain next hop, such as a direct record - /// (no relays) for a device that moved away, is as stale as a broken source route, - /// and only its removal lets the next transmission discover a fresh route. - /// Returns whether a route record was removed. + /// Spec 3.6.4.8.1: a router could not deliver a frame of ours toward `destination`. + /// The route record goes with the routing entry: it wins over the routing table, so + /// a stale one would keep the broken path in use. Returns whether a route record + /// was removed. pub fn forget_route_to(&mut self, destination: Nwk) -> bool { - self.route_table.remove(&destination); - self.route_record_table.remove(&destination).is_some() + self.remove_route(destination); + self.remove_route_record(destination) } /// Spec 3.6.4.3.2: relaying a source-routed frame proves the concentrator that @@ -381,7 +538,7 @@ impl Routing { /// Prepare table state for a route discovery we originate: the routing entry enters /// `DiscoveryUnderway` and a discovery entry keyed by our own address is created. /// Returns the request identifier to put in the route request command. - pub fn begin_discovery( + fn begin_discovery( &mut self, own_address: Nwk, destination: Nwk, @@ -411,7 +568,7 @@ impl Routing { source_address: own_address, sender_address: UNKNOWN_NEXT_HOP, forward_cost: 0, - residual_cost: u8::MAX, + residual_cost: None, expiration_time: now + route_discovery_time, destination_address: destination, }); @@ -443,7 +600,7 @@ impl Routing { source_address: own_address, sender_address: UNKNOWN_NEXT_HOP, forward_cost: 0, - residual_cost: u8::MAX, + residual_cost: None, expiration_time: now + route_discovery_time, destination_address: BROADCAST_ALL_ROUTERS_AND_COORDINATOR, }, @@ -452,17 +609,6 @@ impl Routing { request_id } - /// The expiration time of the live discovery toward a destination, if any. - pub fn discovery_deadline(&self, destination: Nwk, now: Instant) -> Option { - self.discovery_table.values().find_map(|entry| { - if entry.expiration_time > now && entry.destination_address == destination { - Some(entry.expiration_time) - } else { - None - } - }) - } - /// Deduplicate a received route request and track the best path back to its /// originator. Returns false for duplicates with a strictly worse forward cost /// (our own requests, seeded at cost 0, still never echo back); an equal-cost @@ -504,7 +650,7 @@ impl Routing { source_address: originator, sender_address: sender, forward_cost: updated_path_cost, - residual_cost: u8::MAX, + residual_cost: None, expiration_time: now + route_discovery_time, destination_address: destination, }, @@ -542,8 +688,8 @@ impl Routing { } } - /// Before relaying a route request: track that discovery toward the destination is - /// underway, so frames for it are held off until it completes. + /// Before relaying a route request: mark the destination's entry as + /// `DiscoveryUnderway`, so the reply we relay back finds it (spec 3.6.4.5.1.2). pub fn note_relayed_discovery(&mut self, destination: Nwk) { self.route_table .entry(destination) @@ -560,6 +706,7 @@ impl Routing { /// Process a received route reply against the discovery and route tables and /// decide whether to drop it, accept it for our own request, or relay it onward. /// `updated_path_cost` is the advertised cost plus the sender's outgoing link cost. + #[allow(clippy::too_many_arguments)] pub fn accept_route_reply( &mut self, own_address: Nwk, @@ -568,7 +715,11 @@ impl Routing { responder: Nwk, sender: Nwk, updated_path_cost: u8, + now: Instant, ) -> RouteReplyOutcome { + // A reply to a discovery that has lapsed finds no entry and is dropped + self.expire_discoveries(now); + let Some(discovery_entry) = self.discovery_table.get_mut(&(originator, request_id)) else { tracing::debug!("Route reply for unknown route discovery, ignoring"); return RouteReplyOutcome { @@ -588,10 +739,12 @@ impl Routing { // If we are the originator, handling is simplified if originator == own_address { if routing_entry.status == Status::Active - && updated_path_cost >= discovery_entry.residual_cost + && discovery_entry + .residual_cost + .is_some_and(|residual_cost| updated_path_cost >= residual_cost) { tracing::debug!( - "Ignoring route reply for us with higher cost ({} > {})", + "Ignoring route reply for us with higher cost ({} > {:?})", updated_path_cost, discovery_entry.residual_cost ); @@ -611,7 +764,7 @@ impl Routing { routing_entry.status = Status::Active; routing_entry.next_hop_address = sender; routing_entry.path_cost = updated_path_cost; - discovery_entry.residual_cost = updated_path_cost; + discovery_entry.residual_cost = Some(updated_path_cost); let update = (previous_next_hop != Some(sender)).then_some(RouteUpdate { destination: responder, @@ -627,9 +780,12 @@ impl Routing { // Otherwise, we need to decide if we need to update our own routes and possibly // relay the frame - if updated_path_cost >= discovery_entry.residual_cost { + if discovery_entry + .residual_cost + .is_some_and(|residual_cost| updated_path_cost >= residual_cost) + { tracing::debug!( - "Ignoring unsolicited route reply with higher cost ({} > {})", + "Ignoring unsolicited route reply with higher cost ({} > {:?})", updated_path_cost, discovery_entry.residual_cost ); @@ -642,7 +798,7 @@ impl Routing { routing_entry.status = Status::Active; routing_entry.next_hop_address = sender; routing_entry.path_cost = updated_path_cost; - discovery_entry.residual_cost = updated_path_cost; + discovery_entry.residual_cost = Some(updated_path_cost); // The reply travels back along the path the request came from let next_hop = discovery_entry.sender_address; @@ -711,57 +867,3 @@ impl Routing { } } } - -#[cfg(test)] -mod test { - use super::*; - use alloc::vec; - - const DESTINATION: Nwk = Nwk(0x0391); - const RELAY: Nwk = Nwk(0xEF2F); - const MAX_SOURCE_ROUTE: u8 = 12; - - #[test] - fn test_forget_route_to_drops_a_direct_route_record() { - let mut routing = Routing::default(); - routing.store_route_record(DESTINATION, vec![]); - assert_eq!( - routing.route_to(DESTINATION, MAX_SOURCE_ROUTE), - Some(Route::NextHop(DESTINATION)) - ); - - assert!(routing.forget_route_to(DESTINATION)); - assert_eq!(routing.route_to(DESTINATION, MAX_SOURCE_ROUTE), None); - } - - #[test] - fn test_forget_route_to_drops_a_source_route_record() { - let mut routing = Routing::default(); - routing.store_route_record(DESTINATION, vec![RELAY]); - - assert!(routing.forget_route_to(DESTINATION)); - assert_eq!(routing.route_to(DESTINATION, MAX_SOURCE_ROUTE), None); - } - - #[test] - fn test_forget_route_to_drops_the_routing_entry() { - let mut routing = Routing::default(); - routing.restore_route(DESTINATION, RELAY, 1); - assert_eq!(routing.next_hop(DESTINATION), Some(RELAY)); - - assert!(!routing.forget_route_to(DESTINATION)); - assert_eq!(routing.next_hop(DESTINATION), None); - } - - #[test] - fn test_forget_route_to_keeps_other_destinations() { - let mut routing = Routing::default(); - routing.store_route_record(DESTINATION, vec![RELAY]); - - routing.forget_route_to(DESTINATION); - assert_eq!( - routing.route_to(RELAY, MAX_SOURCE_ROUTE), - Some(Route::NextHop(RELAY)) - ); - } -} diff --git a/crates/ziggurat-zigbee/src/nwk/routing/tests.rs b/crates/ziggurat-zigbee/src/nwk/routing/tests.rs new file mode 100644 index 0000000..4e34080 --- /dev/null +++ b/crates/ziggurat-zigbee/src/nwk/routing/tests.rs @@ -0,0 +1,777 @@ +use alloc::vec; +use alloc::vec::Vec; +use core::time::Duration; + +use ziggurat_ieee_802154::types::Nwk; + +use crate::Instant; +use crate::nwk::commands::NwkRouteRequestManyToOne; + +use super::*; + +const OWN: Nwk = Nwk(0x0000); +const DEVICE: Nwk = Nwk(0x1111); +const OTHER_DEVICE: Nwk = Nwk(0x2222); +const ROUTER: Nwk = Nwk(0x3333); +const NEIGHBOR: Nwk = Nwk(0x4444); + +const START: Instant = Instant::from_micros(0); +const ROUTE_DISCOVERY_TIME: Duration = Duration::from_secs(10); +const MAX_SOURCE_ROUTE: u8 = 11; + +/// A route request relayed to us by `NEIGHBOR`. +#[derive(Clone, Copy)] +struct RouteRequest { + originator: Nwk, + destination: Nwk, + path_cost: u8, +} + +/// Receive a route request. Returns whether it was accepted as new or as a better path. +fn receive(routing: &mut Routing, request: RouteRequest, now: Instant) -> bool { + routing + .accept_route_request( + request.originator, + 1, + request.destination, + NEIGHBOR, + request.path_cost, + NwkRouteRequestManyToOne::NotManyToOne, + now, + ROUTE_DISCOVERY_TIME, + ) + .accepted +} + +// Issue #88: a device that joins again never heard the route request we sent before it +// joined. Our discovery toward it must go, or it blocks a new one until it expires. +#[test] +fn test_forget_drops_our_discovery_toward_the_device() { + let mut routing = Routing::default(); + routing.begin_discovery(OWN, DEVICE, START, ROUTE_DISCOVERY_TIME); + assert_eq!( + routing.own_discovery_lapse(OWN, DEVICE, START), + Some(START + ROUTE_DISCOVERY_TIME) + ); + assert_eq!( + routing.route_status(DEVICE), + Some(Status::DiscoveryUnderway) + ); + + routing.forget(OWN, DEVICE); + + assert_eq!(routing.own_discovery_lapse(OWN, DEVICE, START), None); + assert_eq!(routing.route_status(DEVICE), None); +} + +#[test] +fn test_forget_drops_the_route_and_the_route_record() { + let mut routing = Routing::default(); + routing.restore_route(DEVICE, NEIGHBOR, 3); + routing.store_route_record(DEVICE, vec![ROUTER]); + assert!(routing.route_to(DEVICE, MAX_SOURCE_ROUTE).is_some()); + + routing.forget(OWN, DEVICE); + + assert_eq!(routing.next_hop(DEVICE), None); + assert_eq!(routing.route_to(DEVICE, MAX_SOURCE_ROUTE), None); +} + +#[test] +fn test_forget_leaves_other_destinations_alone() { + let mut routing = Routing::default(); + routing.restore_route(OTHER_DEVICE, NEIGHBOR, 3); + routing.begin_discovery(OWN, ROUTER, START, ROUTE_DISCOVERY_TIME); + + routing.forget(OWN, DEVICE); + + assert_eq!(routing.next_hop(OTHER_DEVICE), Some(NEIGHBOR)); + assert_eq!( + routing.own_discovery_lapse(OWN, ROUTER, START), + Some(START + ROUTE_DISCOVERY_TIME) + ); +} + +// Route requests bypass the broadcast transaction table, so the discovery entry is the +// only duplicate check for a relayed request. Forgetting its originator must not make +// a late copy look new, or we rebroadcast it. +#[test] +fn test_forget_keeps_the_duplicate_check_for_a_request_the_device_originated() { + let mut routing = Routing::default(); + let request = RouteRequest { + originator: DEVICE, + destination: OTHER_DEVICE, + path_cost: 5, + }; + assert!(receive(&mut routing, request, START)); + + routing.forget(OWN, DEVICE); + + let late_copy = RouteRequest { + path_cost: 7, + ..request + }; + assert!(!receive(&mut routing, late_copy, START)); +} + +#[test] +fn test_forget_keeps_the_duplicate_check_for_a_request_toward_the_device() { + let mut routing = Routing::default(); + let request = RouteRequest { + originator: OTHER_DEVICE, + destination: DEVICE, + path_cost: 5, + }; + assert!(receive(&mut routing, request, START)); + + routing.forget(OWN, DEVICE); + + let late_copy = RouteRequest { + path_cost: 7, + ..request + }; + assert!(!receive(&mut routing, late_copy, START)); +} + +// A relayed request marks its destination `DiscoveryUnderway`. Only the expiry of the +// discovery entry retires that mark, so the entry must survive forgetting the +// originator. +#[test] +fn test_relayed_discovery_still_expires_after_its_originator_is_forgotten() { + let mut routing = Routing::default(); + let request = RouteRequest { + originator: DEVICE, + destination: OTHER_DEVICE, + path_cost: 5, + }; + assert!(receive(&mut routing, request, START)); + routing.note_relayed_discovery(OTHER_DEVICE); + assert_eq!( + routing.route_status(OTHER_DEVICE), + Some(Status::DiscoveryUnderway) + ); + + routing.forget(OWN, DEVICE); + + // Any later request runs the expiry + let unrelated = RouteRequest { + originator: ROUTER, + destination: NEIGHBOR, + path_cost: 5, + }; + let after_expiry = START + ROUTE_DISCOVERY_TIME + Duration::from_secs(1); + receive(&mut routing, unrelated, after_expiry); + + assert_eq!(routing.route_status(OTHER_DEVICE), None); +} + +// Most route records repeat what is stored. Only a changed one is reported. +#[test] +fn test_route_record_reports_only_a_changed_source_route() { + let mut routing = Routing::default(); + let changed = RouteRecordChange { + source: true, + any: true, + }; + let unchanged = RouteRecordChange { + source: false, + any: false, + }; + + let first = routing.store_route_record(DEVICE, vec![ROUTER, NEIGHBOR]); + let repeat = routing.store_route_record(DEVICE, vec![ROUTER, NEIGHBOR]); + let new_path = routing.store_route_record(DEVICE, vec![NEIGHBOR]); + + assert_eq!([first, repeat, new_path], [changed, unchanged, changed]); +} + +// A record also carries the routes to its relays. An unchanged record restores a relay +// route that was dropped: the device's own route is the same, but a send held for the +// relay can be routed again. +#[test] +fn test_unchanged_route_record_reports_a_relay_route_it_restores() { + let mut routing = Routing::default(); + routing.store_route_record(DEVICE, vec![ROUTER, NEIGHBOR]); + assert!(routing.remove_route_record(ROUTER)); + + let change = routing.store_route_record(DEVICE, vec![ROUTER, NEIGHBOR]); + + assert_eq!( + change, + RouteRecordChange { + source: false, + any: true, + } + ); + assert_eq!( + routing.route_to(ROUTER, MAX_SOURCE_ROUTE), + Some(Route::SourceRouted(vec![NEIGHBOR])) + ); +} + +// A route record wins over the routing table, so a reported failure must remove it +// too, whether it is direct or a source route. +#[test] +fn test_forget_route_to_drops_the_route_record_and_the_routing_entry() { + for relays in [vec![], vec![ROUTER]] { + let mut routing = Routing::default(); + routing.restore_route(DEVICE, NEIGHBOR, 3); + routing.store_route_record(DEVICE, relays); + + assert!(routing.forget_route_to(DEVICE)); + assert_eq!(routing.route_to(DEVICE, MAX_SOURCE_ROUTE), None); + } +} + +#[test] +fn test_forget_route_to_reports_only_a_removed_route_record() { + let mut routing = Routing::default(); + routing.restore_route(DEVICE, NEIGHBOR, 3); + + assert!(!routing.forget_route_to(DEVICE)); + assert_eq!(routing.next_hop(DEVICE), None); +} + +#[test] +fn test_forget_route_to_leaves_other_destinations_alone() { + let mut routing = Routing::default(); + routing.store_route_record(DEVICE, vec![ROUTER]); + + routing.forget_route_to(DEVICE); + + assert_eq!( + routing.route_to(ROUTER, MAX_SOURCE_ROUTE), + Some(Route::NextHop(ROUTER)) + ); +} + +/// A frame of ours wants a route to `DEVICE`. Returns whether that starts a discovery. +fn want(routing: &mut Routing, now: Instant, attempts: u8) -> bool { + routing + .want_route(OWN, DEVICE, now, ROUTE_DISCOVERY_TIME, attempts) + .is_some() +} + +/// Run a timer pass, with a frame still waiting for the route or not. +fn tick(routing: &mut Routing, now: Instant, wanted: bool) -> DiscoveryTick { + routing.tick_discoveries(OWN, now, ROUTE_DISCOVERY_TIME, |_| wanted) +} + +#[test] +fn test_wanting_a_route_starts_one_discovery() { + let mut routing = Routing::default(); + + assert!(want(&mut routing, START, 1)); + assert!(!want(&mut routing, START, 1)); + + assert_eq!( + routing.route_status(DEVICE), + Some(Status::DiscoveryUnderway) + ); + assert_eq!( + routing.next_discovery_lapse(OWN, START), + Some(START + ROUTE_DISCOVERY_TIME) + ); +} + +// Spec 3.6.4.5.1.7: an unanswered discovery deletes its routing entry. +#[test] +fn test_lapsed_discovery_without_a_restart_makes_the_destination_unreachable() { + let mut routing = Routing::default(); + assert!(want(&mut routing, START, 1)); + + let lapsed = START + ROUTE_DISCOVERY_TIME; + assert_eq!(tick(&mut routing, lapsed, true).unreachable, vec![DEVICE]); + + assert_eq!(routing.route_status(DEVICE), None); + // The next frame starts over + assert!(want(&mut routing, lapsed, 1)); +} + +#[test] +fn test_lapsed_discovery_is_restarted_while_a_frame_waits() { + let mut routing = Routing::default(); + assert!(want(&mut routing, START, 2)); + + let lapsed = START + ROUTE_DISCOVERY_TIME; + let restarted = tick(&mut routing, lapsed, true).restarted; + + assert_eq!(restarted.len(), 1); + assert_eq!(restarted[0].0, DEVICE); + assert_eq!( + routing.next_discovery_lapse(OWN, lapsed), + Some(lapsed + ROUTE_DISCOVERY_TIME) + ); + + // The restart was the last one + let lapsed_again = lapsed + ROUTE_DISCOVERY_TIME; + assert_eq!( + tick(&mut routing, lapsed_again, true).unreachable, + vec![DEVICE] + ); +} + +// A discovery nothing waits on is over, not failed: it is neither restarted nor +// reported. +#[test] +fn test_lapsed_discovery_nothing_waits_on_ends_quietly() { + let mut routing = Routing::default(); + assert!(want(&mut routing, START, 2)); + + let lapsed = START + ROUTE_DISCOVERY_TIME; + assert_eq!(tick(&mut routing, lapsed, false), DiscoveryTick::default()); + + assert!(want(&mut routing, lapsed, 2)); +} + +#[test] +fn test_live_discovery_is_left_alone() { + let mut routing = Routing::default(); + assert!(want(&mut routing, START, 1)); + + let live = START + Duration::from_secs(1); + assert_eq!(tick(&mut routing, live, true), DiscoveryTick::default()); + assert_eq!(tick(&mut routing, live, false), DiscoveryTick::default()); +} + +// The route request is on the network whatever became of the frames that wanted it. +// A later frame waits on it, and it is that frame's one attempt. +#[test] +fn test_new_frame_waits_on_a_discovery_whose_frames_are_gone() { + let mut routing = Routing::default(); + assert!(want(&mut routing, START, 1)); + + let live = START + Duration::from_secs(1); + assert_eq!(tick(&mut routing, live, false), DiscoveryTick::default()); + assert!(!want(&mut routing, live, 1)); + + let lapsed = START + ROUTE_DISCOVERY_TIME; + assert_eq!(tick(&mut routing, lapsed, true).unreachable, vec![DEVICE]); +} + +// Spec 3.6.4.3: relaying a route request also marks the destination's entry as +// `DiscoveryUnderway`. A frame of ours does not wait on another router's discovery. +#[test] +fn test_relayed_discovery_toward_the_destination_is_not_ours() { + let mut routing = Routing::default(); + let request = RouteRequest { + originator: ROUTER, + destination: DEVICE, + path_cost: 3, + }; + assert!(receive(&mut routing, request, START)); + routing.note_relayed_discovery(DEVICE); + + assert!(want(&mut routing, START, 1)); +} + +#[test] +fn test_relayed_discovery_does_not_keep_ours_alive() { + let mut routing = Routing::default(); + assert!(want(&mut routing, START, 1)); + + let request = RouteRequest { + originator: ROUTER, + destination: DEVICE, + path_cost: 3, + }; + assert!(receive( + &mut routing, + request, + START + Duration::from_secs(5) + )); + + let lapsed = START + ROUTE_DISCOVERY_TIME; + assert_eq!(routing.next_discovery_lapse(OWN, lapsed), None); + assert_eq!(tick(&mut routing, lapsed, true).unreachable, vec![DEVICE]); +} + +// A relay can report an earlier frame undeliverable after the next frame started a +// discovery. The report is about the old route. The discovery must still take its reply. +#[test] +fn test_removing_the_route_keeps_a_discovery_underway() { + let mut routing = Routing::default(); + routing.restore_route(DEVICE, NEIGHBOR, 3); + assert!(routing.remove_route(DEVICE)); + + let request_id = routing + .want_route(OWN, DEVICE, START, ROUTE_DISCOVERY_TIME, 1) + .unwrap(); + assert!(!routing.remove_route(DEVICE)); + assert!(!routing.forget_route_to(DEVICE)); + + let reply = reply(&mut routing, request_id, START); + assert_eq!(reply.disposition, RouteReplyDisposition::Established); + assert_eq!(routing.next_hop(DEVICE), Some(NEIGHBOR)); +} + +/// `DEVICE` answers our route request through `NEIGHBOR`. +fn reply(routing: &mut Routing, request_id: RouteRequestId, now: Instant) -> RouteReplyOutcome { + routing.accept_route_reply(OWN, OWN, request_id, DEVICE, NEIGHBOR, 3, now) +} + +/// Start a discovery toward `DEVICE` and have the device answer it. +fn answered_discovery(routing: &mut Routing, attempts: u8) { + let request_id = routing + .want_route(OWN, DEVICE, START, ROUTE_DISCOVERY_TIME, attempts) + .unwrap(); + + let reply = reply(routing, request_id, START); + assert_eq!(reply.disposition, RouteReplyDisposition::Established); + assert_eq!(routing.next_hop(DEVICE), Some(NEIGHBOR)); +} + +// The reply makes the route active, so no discovery is underway any more. If the route +// is lost later, the next frame starts a new discovery instead of waiting on a request +// that was already answered. +#[test] +fn test_frame_after_an_answered_discovery_starts_a_new_one() { + let mut routing = Routing::default(); + answered_discovery(&mut routing, 1); + + assert_eq!(tick(&mut routing, START, false), DiscoveryTick::default()); + + routing.invalidate_via(NEIGHBOR); + assert!(want(&mut routing, START, 1)); +} + +// Relaying another router's route request marks the entry `DiscoveryUnderway` again. +// Our answered discovery is still in the table, but its request is spent. +#[test] +fn test_relayed_discovery_does_not_revive_our_answered_one() { + let mut routing = Routing::default(); + answered_discovery(&mut routing, 1); + assert_eq!(tick(&mut routing, START, false), DiscoveryTick::default()); + routing.invalidate_via(NEIGHBOR); + + let request = RouteRequest { + originator: ROUTER, + destination: DEVICE, + path_cost: 3, + }; + assert!(receive(&mut routing, request, START)); + routing.note_relayed_discovery(DEVICE); + assert_eq!( + routing.route_status(DEVICE), + Some(Status::DiscoveryUnderway) + ); + + assert!(want(&mut routing, START, 1)); +} + +// A reply can carry the highest path cost. It still answers the discovery. +#[test] +fn test_reply_with_the_highest_path_cost_answers_the_discovery() { + let mut routing = Routing::default(); + let request_id = routing + .want_route(OWN, DEVICE, START, ROUTE_DISCOVERY_TIME, 1) + .unwrap(); + + let reply = routing.accept_route_reply(OWN, OWN, request_id, DEVICE, NEIGHBOR, u8::MAX, START); + assert_eq!(reply.disposition, RouteReplyDisposition::Established); + assert_eq!(tick(&mut routing, START, false), DiscoveryTick::default()); + + routing.invalidate_via(NEIGHBOR); + routing.note_relayed_discovery(DEVICE); + + assert!(want(&mut routing, START, 1)); +} + +// The route can be lost between the reply and the pass that would have released the +// held frame. The frame must not be left waiting with no discovery behind it. +#[test] +fn test_answered_discovery_is_restarted_for_a_frame_its_route_did_not_release() { + let mut routing = Routing::default(); + answered_discovery(&mut routing, 2); + routing.invalidate_via(NEIGHBOR); + + assert_eq!(tick(&mut routing, START, true).restarted.len(), 1); + + // With no restart left the frame fails instead + let mut routing = Routing::default(); + answered_discovery(&mut routing, 1); + routing.invalidate_via(NEIGHBOR); + + assert_eq!(tick(&mut routing, START, true).unreachable, vec![DEVICE]); +} + +#[test] +fn test_forget_ends_our_discovery_toward_the_device() { + let mut routing = Routing::default(); + assert!(want(&mut routing, START, 1)); + + routing.forget(OWN, DEVICE); + + assert_eq!(routing.next_discovery_lapse(OWN, START), None); + assert!(want(&mut routing, START, 1)); +} + +// Nothing wakes the timer for a discovery no frame waits on, so its entry can outlive +// its expiry. A reply must not find it. +#[test] +fn test_reply_to_a_lapsed_discovery_is_dropped() { + let mut routing = Routing::default(); + let request_id = routing + .want_route(OWN, DEVICE, START, ROUTE_DISCOVERY_TIME, 1) + .unwrap(); + + let late = reply(&mut routing, request_id, START + ROUTE_DISCOVERY_TIME); + + assert_eq!(late.disposition, RouteReplyDisposition::Drop); + assert_eq!(routing.route_status(DEVICE), None); +} + +/// One event in the life of our route discovery toward `DEVICE`. +#[derive(Debug, Clone, Copy)] +enum Step { + /// A frame is held for the route. + Hold, + /// The held frames are gone: cancelled, or past their deadline. + Release, + /// Time passes until every discovery has lapsed. + Lapse, + Tick, + /// The device answers the latest discovery. + Reply, + /// We relay another router's route request toward the device. + Relay, + /// A router reports that it could not deliver a frame of ours to the device: a + /// route we had is lost. + ForgetRoute, +} + +const STEPS: [Step; 7] = [ + Step::Hold, + Step::Release, + Step::Lapse, + Step::Tick, + Step::Reply, + Step::Relay, + Step::ForgetRoute, +]; + +/// What `Routing` should hold about `DEVICE`. +#[derive(Debug, Default)] +struct Model { + /// A frame waits for the route. + wanted: bool, + /// How many more discoveries the waiting frames may get. Kept from the first frame + /// that wants the route until a pass finds no frame waiting, or gives up. + discoveries_left: Option, + /// The status of the routing entry, if there is one. + status: Option, + /// A discovery we started has not lapsed. + live: bool, + /// A discovery we relay has not lapsed. + relayed_live: bool, + /// The device answered the latest discovery. + replied: bool, +} + +impl Model { + /// A route request of ours is on the network and may still be answered. This is + /// what a waiting frame needs, whatever the routing entry says. + const fn request_pending(&self) -> bool { + self.live && !self.replied + } + + fn underway(&self) -> bool { + self.request_pending() && self.status == Some(Status::DiscoveryUnderway) + } + + /// Lapsed discoveries are only cleaned up by a pass, a received route request or + /// the next discovery. The last one to go takes an unanswered routing entry with it. + fn expire(&mut self) { + if !self.live && !self.relayed_live && self.status == Some(Status::DiscoveryUnderway) { + self.status = None; + } + } + + fn begin_discovery(&mut self) { + self.expire(); + self.status = Some(Status::DiscoveryUnderway); + self.live = true; + self.replied = false; + } +} + +/// Run `steps` against `Routing` and the model, comparing them after each one. +fn check_own_discovery(steps: &[Step], attempts: u8) { + let mut routing = Routing::default(); + let mut model = Model::default(); + let mut now = START; + let mut request_id = None; + + for step in steps { + match *step { + // The driver only holds a frame when it has no route + Step::Hold if routing.next_hop(DEVICE).is_some() => {} + Step::Hold => { + let started = routing.want_route(OWN, DEVICE, now, ROUTE_DISCOVERY_TIME, attempts); + + // Only the first frame to wait can start a discovery, and a discovery + // of ours that is still underway serves it + let starts = model.discoveries_left.is_none() && !model.underway(); + assert_eq!(started.is_some(), starts, "{steps:?}"); + + // The first frame to wait is never left to a request that is spent + assert!( + started.is_some() + || model.discoveries_left.is_some() + || model.request_pending(), + "{steps:?}" + ); + + if model.discoveries_left.is_none() { + model.discoveries_left = Some(attempts - 1); + } + if let Some(started) = started { + request_id = Some(started); + model.begin_discovery(); + } + model.wanted = true; + } + Step::Release => model.wanted = false, + Step::Lapse => { + now = now + ROUTE_DISCOVERY_TIME; + model.live = false; + model.relayed_live = false; + } + Step::Tick => { + // The send tables go first: a held frame leaves once it has a next hop + if routing.next_hop(DEVICE).is_some() { + model.wanted = false; + } + + let tick = tick(&mut routing, now, model.wanted); + let mut restarted = false; + let mut unreachable = false; + + model.expire(); + + match model.discoveries_left { + None => {} + Some(_) if !model.wanted => model.discoveries_left = None, + Some(_) if model.underway() => {} + Some(0) => { + model.discoveries_left = None; + // The driver fails the frames held for the destination + model.wanted = false; + unreachable = true; + } + Some(left) => { + model.discoveries_left = Some(left - 1); + model.begin_discovery(); + restarted = true; + } + } + + let destinations: Vec = tick.restarted.iter().map(|r| r.0).collect(); + assert_eq!( + destinations, + [DEVICE][..usize::from(restarted)], + "{steps:?}" + ); + assert_eq!( + tick.unreachable, + [DEVICE][..usize::from(unreachable)], + "{steps:?}" + ); + + if let Some(restart) = tick.restarted.first() { + request_id = Some(restart.1); + } + + // A pass leaves every waiting frame with a request that may still be + // answered, and with a wake for when it lapses + assert!(!model.wanted || model.request_pending(), "{steps:?}"); + assert!( + !model.wanted || routing.next_discovery_lapse(OWN, now).is_some(), + "{steps:?}" + ); + } + Step::Reply if model.request_pending() => { + let reply = reply(&mut routing, request_id.unwrap(), now); + assert_eq!( + reply.disposition, + RouteReplyDisposition::Established, + "{steps:?}" + ); + + // The held frames leave at the next pass, if the route is still there + model.status = Some(Status::Active); + model.replied = true; + } + // A reply that comes after its discovery lapsed is too late + Step::Reply if !model.live && request_id.is_some() => { + let reply = reply(&mut routing, request_id.unwrap(), now); + assert_eq!(reply.disposition, RouteReplyDisposition::Drop, "{steps:?}"); + + model.expire(); + } + Step::Reply => {} + Step::Relay => { + let request = RouteRequest { + originator: ROUTER, + destination: DEVICE, + path_cost: 3, + }; + assert!(receive(&mut routing, request, now), "{steps:?}"); + routing.note_relayed_discovery(DEVICE); + + model.expire(); + model.relayed_live = true; + if model.status != Some(Status::Active) { + model.status = Some(Status::DiscoveryUnderway); + } + } + Step::ForgetRoute => { + routing.forget_route_to(DEVICE); + + // An entry that waits for a reply holds no route + if model.status != Some(Status::DiscoveryUnderway) { + model.status = None; + } + } + } + + assert_eq!(routing.route_status(DEVICE), model.status, "{steps:?}"); + assert_eq!( + routing.next_discovery_lapse(OWN, now).is_some(), + model.discoveries_left.is_some() && model.live, + "{steps:?}" + ); + + // A frame that waits always has something that will release or fail it: a + // route, or a count a later pass acts on + assert!( + !model.wanted || model.discoveries_left.is_some() || routing.next_hop(DEVICE).is_some(), + "a frame waits with no route and nothing a pass acts on: {steps:?}" + ); + } +} + +/// Check every sequence of seven events. +fn check_every_short_sequence(attempts: u8) { + const LENGTH: u32 = 7; + + for mut sequence in 0..STEPS.len().pow(LENGTH) { + let mut steps = Vec::new(); + for _ in 0..LENGTH { + steps.push(STEPS[sequence % STEPS.len()]); + sequence /= STEPS.len(); + } + + check_own_discovery(&steps, attempts); + } +} + +#[test] +fn test_own_discovery_matches_the_model_with_one_attempt() { + check_every_short_sequence(1); +} + +#[test] +fn test_own_discovery_matches_the_model_with_two_attempts() { + check_every_short_sequence(2); +} diff --git a/crates/ziggurat-zigbee/src/sends/broadcasts.rs b/crates/ziggurat-zigbee/src/sends/broadcasts.rs index 3945e59..7d4d33f 100644 --- a/crates/ziggurat-zigbee/src/sends/broadcasts.rs +++ b/crates/ziggurat-zigbee/src/sends/broadcasts.rs @@ -294,6 +294,10 @@ impl> Broadcasts { waiters } + pub(super) fn contains(&self, id: SendId) -> bool { + self.records.contains_key(&id) + } + /// `None` if a copy is in the radio right now, or the send is unknown. pub(super) fn cancel(&mut self, id: SendId) -> Option> { let record = self.records.get(&id)?; diff --git a/crates/ziggurat-zigbee/src/sends/mod.rs b/crates/ziggurat-zigbee/src/sends/mod.rs index 9527595..fab4888 100644 --- a/crates/ziggurat-zigbee/src/sends/mod.rs +++ b/crates/ziggurat-zigbee/src/sends/mod.rs @@ -7,9 +7,9 @@ //! - [`Unicasts`](unicasts::Unicasts): a frame with a known next hop, delivered directly //! (with retries) or to a sleepy child when it polls. //! - [`UnroutedUnicasts`](unrouted::UnroutedUnicasts): a unicast with no route yet, held -//! until its destination's route discovery ends. The driver runs the discovery against its -//! routing table and reports when the route is known or the discovery lapsed; the -//! send then moves to the unicasts or fails. +//! until a timer pass finds a next hop for it. The routing table decides when a route +//! discovery is started or given up on; the driver reports a destination it gave up +//! on, and the sends held for it fail. //! - [`Broadcasts`](broadcasts::Broadcasts): a frame sent once, or rebroadcast on a //! schedule until its passive-ack quorum is heard. //! @@ -26,6 +26,8 @@ //! deadline fails with the error its state implies. mod broadcasts; +#[cfg(test)] +mod tests; mod unicasts; mod unrouted; @@ -252,6 +254,10 @@ pub struct Sends { unicasts: Unicasts, unrouted_unicasts: UnroutedUnicasts, broadcasts: Broadcasts, + /// The broadcast carrying the latest route request toward a destination, while it + /// still has copies to send. Kept to stop those copies when the discovery is + /// replaced or abandoned. + route_requests: FlatMap, } impl Default for Sends { @@ -261,6 +267,7 @@ impl Default for Sends { unicasts: Unicasts::default(), unrouted_unicasts: UnroutedUnicasts::default(), broadcasts: Broadcasts::default(), + route_requests: FlatMap::new(), } } } @@ -299,10 +306,8 @@ impl> Sends { (id, Actions::new()) } - /// Enqueue a unicast with no route yet, held until its destination's discovery - /// ends. The caller starts that discovery if the destination was not already - /// awaited; the first unrouted send to a destination fixes how often it is - /// restarted. + /// Enqueue a unicast with no route yet, held until a timer pass finds it a next + /// hop. The caller tells the routing table a route is wanted. pub fn enqueue_unrouted_unicast( &mut self, send: UnroutedUnicast, @@ -400,87 +405,64 @@ impl> Sends { self.unicasts.ack_received(ack) } - /// Every destination with unrouted sends. - pub fn unrouted_destinations(&self) -> Vec { - self.unrouted_unicasts.destinations() - } - - /// Whether any unrouted send waits on `destination`'s route discovery. + /// Whether any unrouted send waits for a route to `destination`. pub fn has_unrouted(&self, destination: Nwk) -> bool { - self.unrouted_unicasts.has_unrouted(destination) - } - - /// `destination`'s discovery lapsed: consume a restart of it, if one is left and - /// an unrouted send still waits on it. The caller restarts the discovery. - pub fn retry_discovery(&mut self, destination: Nwk) -> bool { - self.unrouted_unicasts.retry_discovery(destination) + self.unrouted_unicasts.has_sends_to(destination) } - /// The driver broadcast `request`, the route request of `destination`'s discovery. - /// Copies of the request it replaces still to come are not sent. + /// The driver broadcast `request`, a route request toward `destination`. Copies of + /// the request it replaces still to come are not sent. pub fn route_request_sent(&mut self, destination: Nwk, request: SendId) -> Actions { - match self - .unrouted_unicasts - .route_request_sent(destination, request) - { + match self.route_requests.insert(destination, request) { Some(previous) => self.broadcasts.stop(previous), None => Actions::new(), } } - /// Fail every unrouted send to `destination`: its discovery is over without a - /// route. Copies of its route request still to come are not sent. + /// Fail every unrouted send to `destination`: no route to it will be found. Copies + /// of its route request still to come are not sent. pub fn fail_unrouted(&mut self, destination: Nwk) -> Actions { - let (request, mut actions) = self.unrouted_unicasts.fail(destination); + let mut actions = self.unrouted_unicasts.fail(destination); - if let Some(request) = request { + if let Some(request) = self.route_requests.remove(&destination) { actions.extend(self.broadcasts.stop(request)); } actions } - /// A route to `destination` is known: move every unrouted send to it to the unicasts, - /// keeping the deadline it was admitted with. `resolve` finds each frame's next hop - /// (rewriting the frame for a source route) and the address the device polls under - /// if the frame must wait for its poll, or fails the send. - pub fn route_unicasts( + /// Move every unrouted send that `next_hop` finds a next hop for. + fn route_unicasts( &mut self, - destination: Nwk, now: Instant, tunables: &Tunables, - mut resolve: impl FnMut(&mut F) -> Result<(Nwk, Option), E>, - ) -> Actions { - // Fail the sends past their deadline first, so routing never revives one - let mut actions = self.unrouted_unicasts.tick(now); - - for (id, mut unrouted) in self.unrouted_unicasts.route(destination) { - match resolve(&mut unrouted.frame) { - Ok((next_hop, poll_address)) => self.unicasts.enqueue( - id, - Unicast { - frame: unrouted.frame, - priority: unrouted.priority, - waiter: unrouted.waiter, - confirm: unrouted.confirm, - next_hop, - poll_address, - }, - now, - Some(unrouted.deadline), - None, - tunables, - ), - Err(err) => actions.push(Action::Settled { - waiter: unrouted.waiter, - result: Err(err), + mut next_hop: impl FnMut(&mut F) -> Option<(Nwk, Option)>, + ) { + for id in self.unrouted_unicasts.ids() { + // `next_hop` may rewrite the frame for a source route. Without a next hop + // the send stays held. + let Some((next_hop, poll_address)) = next_hop(self.unrouted_unicasts.frame_mut(id)) + else { + continue; + }; + + let unrouted = self.unrouted_unicasts.take(id); + self.unicasts.enqueue( + id, + Unicast { frame: unrouted.frame, - failed_hop: None, - }), - } + priority: unrouted.priority, + waiter: unrouted.waiter, + confirm: unrouted.confirm, + next_hop, + poll_address, + }, + now, + Some(unrouted.deadline), + None, + tunables, + ); } - - actions } /// The key of every broadcast awaiting its passive-ack quorum. @@ -505,6 +487,7 @@ impl> Sends { let mut waiters = self.unicasts.abandon(); waiters.extend(self.unrouted_unicasts.abandon()); waiters.extend(self.broadcasts.abandon()); + self.route_requests.clear(); waiters } @@ -537,11 +520,26 @@ impl> Sends { Some(actions) } - /// The timer pass: re-queue what is due and fail what is past its deadline. - pub fn tick(&mut self, now: Instant) -> Actions { - let mut actions = self.unicasts.tick(now); - actions.extend(self.unrouted_unicasts.tick(now)); + /// The timer pass over the three tables. + pub fn tick( + &mut self, + now: Instant, + tunables: &Tunables, + // A frame's next hop, and the address its device polls under if the frame must + // wait for its poll + next_hop: impl FnMut(&mut F) -> Option<(Nwk, Option)>, + ) -> Actions { + // Fail the sends past their deadline first, so routing never revives one + let mut actions = self.unrouted_unicasts.tick(now); + self.route_unicasts(now, tunables, next_hop); + + actions.extend(self.unicasts.tick(now)); actions.extend(self.broadcasts.tick(now)); + + // A route request with no copy left to send needs no stopping + self.route_requests + .retain(|_, request| self.broadcasts.contains(*request)); + actions } diff --git a/crates/ziggurat-zigbee/src/sends/tests.rs b/crates/ziggurat-zigbee/src/sends/tests.rs new file mode 100644 index 0000000..99f3e07 --- /dev/null +++ b/crates/ziggurat-zigbee/src/sends/tests.rs @@ -0,0 +1,256 @@ +use alloc::vec::Vec; +use core::time::Duration; + +use ziggurat_ieee_802154::types::Nwk; + +use crate::Instant; +use crate::constants::Tunables; + +use super::*; + +type TestSends = Sends<&'static str, u8, SendFailure>; +type TestActions = Actions<&'static str, u8, SendFailure>; + +/// A send's waiter and how it settled. +type Verdict = (u8, Result<(), SendFailure>); + +const DESTINATION: Nwk = Nwk(0x1234); +const NEXT_HOP: Nwk = Nwk(0x5678); + +// The waiters of the sends a test enqueues +const HELD_UNICAST: u8 = 1; +const ROUTE_REQUEST: u8 = 2; +const SECOND_ROUTE_REQUEST: u8 = 3; + +const HELD_TIMED_OUT: Verdict = (HELD_UNICAST, Err(SendFailure::RouteDiscoveryTimeout)); +const REQUEST_CANCELLED: Verdict = (ROUTE_REQUEST, Err(SendFailure::Cancelled)); + +/// The send tables, with the tunables and the clock a test drives them with. +struct Fixture { + sends: TestSends, + tunables: Tunables, + now: Instant, +} + +impl Fixture { + fn new() -> Self { + Self { + sends: TestSends::default(), + tunables: Tunables::default(), + now: Instant::from_micros(0), + } + } + + fn advance(&mut self, duration: Duration) { + self.now = self.now + duration; + } + + /// Hold a unicast to `DESTINATION` until a route to it is known. + fn hold_unicast(&mut self) -> SendId { + let (id, actions) = self.sends.enqueue_unrouted_unicast( + UnroutedUnicast { + frame: "held unicast", + priority: TxPriority::UserNormal, + waiter: HELD_UNICAST, + confirm: Confirm::OnHandoff, + destination: DESTINATION, + }, + self.now, + &self.tunables, + ); + assert!(settled(actions).is_empty()); + + id + } + + /// Whether a unicast is still held for `DESTINATION`. + fn holding(&self) -> bool { + self.sends.has_unrouted(DESTINATION) + } + + /// Fail every unicast held for `DESTINATION`, as the driver does when the device + /// leaves or joins again. + fn fail_held(&mut self) -> Vec { + settled(self.sends.fail_unrouted(DESTINATION)) + } + + /// Broadcast the route request of `DESTINATION`'s discovery, as the driver does: + /// four copies, the first due now. Returns what registering the request settles. + fn broadcast_route_request(&mut self, waiter: u8) -> Vec { + let (request, actions) = self.sends.enqueue_broadcast( + Broadcast { + frame: "route request", + priority: TxPriority::UserNormal, + waiter, + schedule: Schedule::FixedInterval { + interval: Duration::from_millis(254), + copies_remaining: 4, + next_attempt: self.now, + }, + }, + self.now, + &self.tunables, + ); + assert!(settled(actions).is_empty()); + + settled(self.sends.route_request_sent(DESTINATION, request)) + } + + /// Run a timer pass in which the routing table offers this next hop toward + /// `DESTINATION`. Returns what the pass settles. + fn pass(&mut self, next_hop: Option) -> Vec { + settled(self.sends.tick(self.now, &self.tunables, |_| { + next_hop.map(|next_hop| (next_hop, None)) + })) + } + + fn dequeue(&mut self) -> Option> { + self.sends.dequeue(self.now) + } + + /// Report a successful transmit. Returns what it settles. + fn transmitted(&mut self, in_flight: InFlight<&'static str>) -> Vec { + settled( + self.sends + .transmitted(in_flight, Ok(()), self.now, Duration::ZERO, Duration::ZERO), + ) + } + + /// Transmit the next frame that is due. Returns what that settles. + fn transmit_next(&mut self) -> Vec { + let in_flight = self.dequeue().unwrap(); + self.transmitted(in_flight) + } +} + +/// The waiter and verdict of every send the actions settle. +fn settled(actions: TestActions) -> Vec { + actions + .into_iter() + .filter_map(|action| match action { + Action::Settled { waiter, result, .. } => Some((waiter, result)), + Action::HandedOff(_) => None, + }) + .collect() +} + +#[test] +fn test_held_unicast_waits_without_a_next_hop() { + let mut fixture = Fixture::new(); + fixture.hold_unicast(); + + assert!(fixture.pass(None).is_empty()); + assert!(fixture.holding()); + assert!(fixture.dequeue().is_none()); +} + +// A held unicast waits for a next hop, not for a discovery: any pass that finds one +// releases it. +#[test] +fn test_held_unicast_is_routed_once_a_next_hop_is_found() { + let mut fixture = Fixture::new(); + fixture.hold_unicast(); + + assert!(fixture.pass(Some(NEXT_HOP)).is_empty()); + assert!(!fixture.holding()); + + let in_flight = fixture.dequeue().unwrap(); + assert_eq!(in_flight.frame, "held unicast"); + assert!(matches!( + in_flight.transmit(), + Transmit::Unicast { next_hop, .. } if next_hop == NEXT_HOP + )); +} + +#[test] +fn test_held_unicast_fails_at_its_deadline() { + let mut fixture = Fixture::new(); + fixture.hold_unicast(); + + fixture.advance(fixture.tunables.handoff_timeout()); + + // The deadline comes first: a next hop does not revive a late send + assert_eq!( + fixture.pass(Some(NEXT_HOP)), + [(HELD_UNICAST, Err(SendFailure::HandoffTimeout))] + ); + assert!(!fixture.holding()); +} + +#[test] +fn test_held_unicast_can_be_cancelled() { + let mut fixture = Fixture::new(); + let id = fixture.hold_unicast(); + + let cancelled = settled(fixture.sends.cancel(id).unwrap()); + + assert_eq!(cancelled, [(HELD_UNICAST, Err(SendFailure::Cancelled))]); + assert!(!fixture.holding()); +} + +// The routing table gave up on the destination, or the device left or joined again. +#[test] +fn test_failing_held_unicasts_leaves_nothing_held() { + let mut fixture = Fixture::new(); + fixture.hold_unicast(); + + assert_eq!(fixture.fail_held(), [HELD_TIMED_OUT]); + assert!(!fixture.holding()); +} + +// An abandoned discovery must not keep broadcasting its route request. +#[test] +fn test_failing_held_unicasts_stops_their_route_request() { + let mut fixture = Fixture::new(); + fixture.hold_unicast(); + assert!(fixture.broadcast_route_request(ROUTE_REQUEST).is_empty()); + + // The first of the four copies goes out + assert!(fixture.pass(None).is_empty()); + assert!(fixture.transmit_next().is_empty()); + + assert_eq!(fixture.fail_held(), [HELD_TIMED_OUT, REQUEST_CANCELLED]); + + // The other three copies are never queued + fixture.advance(Duration::from_secs(1)); + assert!(fixture.pass(None).is_empty()); + assert!(fixture.dequeue().is_none()); +} + +// A copy in the radio cannot be recalled. It completes, and no copy follows it. +#[test] +fn test_stopped_route_request_in_the_radio_sends_no_further_copy() { + let mut fixture = Fixture::new(); + fixture.hold_unicast(); + assert!(fixture.broadcast_route_request(ROUTE_REQUEST).is_empty()); + + assert!(fixture.pass(None).is_empty()); + let first_copy = fixture.dequeue().unwrap(); + + // Only the held unicast settles: the route request is on loan to the radio + assert_eq!(fixture.fail_held(), [HELD_TIMED_OUT]); + assert!(fixture.transmitted(first_copy).is_empty()); + + // One copy reached the air, which completes a route request + fixture.advance(Duration::from_secs(1)); + assert_eq!(fixture.pass(None), [(ROUTE_REQUEST, Ok(()))]); + assert!(fixture.dequeue().is_none()); +} + +// A restarted discovery broadcasts a new route request. The one it replaces stops. +#[test] +fn test_new_route_request_stops_the_one_it_replaces() { + let mut fixture = Fixture::new(); + fixture.hold_unicast(); + assert!(fixture.broadcast_route_request(ROUTE_REQUEST).is_empty()); + + assert_eq!( + fixture.broadcast_route_request(SECOND_ROUTE_REQUEST), + [REQUEST_CANCELLED] + ); + + // Only the new request has a copy to send + assert!(fixture.pass(None).is_empty()); + assert!(fixture.transmit_next().is_empty()); + assert!(fixture.dequeue().is_none()); +} diff --git a/crates/ziggurat-zigbee/src/sends/unrouted.rs b/crates/ziggurat-zigbee/src/sends/unrouted.rs index 65eb637..5052477 100644 --- a/crates/ziggurat-zigbee/src/sends/unrouted.rs +++ b/crates/ziggurat-zigbee/src/sends/unrouted.rs @@ -1,9 +1,7 @@ -//! Unicasts with no route yet, held until their destination's route discovery ends. +//! Unicasts with no route yet, held until one is known or their deadline passes. //! -//! Discovery runs once per destination and the whole bucket moves on together, so ten -//! frames to one device ride a single discovery. The discovery itself lives in the -//! routing table; this table counts the restarts it has left and remembers the -//! broadcast of its route request, to stop it if the discovery is abandoned. +//! This table knows nothing of route discovery. The routing table decides when one is +//! started, restarted or given up on; a held send only waits for a next hop. use alloc::vec::Vec; use core::marker::PhantomData; @@ -38,21 +36,9 @@ pub(super) struct Record { pub(super) deadline: Instant, } -/// What this table tracks of a destination's route discovery. -#[derive(Debug)] -struct Discovery { - /// Restarts left. The first discovery is the driver's, started at admission; these - /// are the ones after it. - restarts_left: u8, - /// The broadcast of the latest route request. - request: Option, -} - #[derive(Debug)] pub(super) struct UnroutedUnicasts { records: FlatMap>, - /// One per destination with unrouted sends. - discoveries: FlatMap, _error: PhantomData E>, } @@ -60,7 +46,6 @@ impl Default for UnroutedUnicasts { fn default() -> Self { Self { records: FlatMap::new(), - discoveries: FlatMap::new(), _error: PhantomData, } } @@ -71,12 +56,14 @@ impl> UnroutedUnicasts { self.records.contains_key(&id) } - pub(super) fn has_unrouted(&self, destination: Nwk) -> bool { - self.discoveries.contains_key(&destination) + /// Whether any send waits for a route to `destination`. + pub(super) fn has_sends_to(&self, destination: Nwk) -> bool { + self.records + .values() + .any(|send| send.destination == destination) } - /// Hold a send. It fails if still unrouted within the handoff timeout. The first - /// unrouted send to a destination fixes how often its discovery may be restarted. + /// Hold a send. It fails if still unrouted within the handoff timeout. pub(super) fn enqueue( &mut self, id: SendId, @@ -84,7 +71,6 @@ impl> UnroutedUnicasts { now: Instant, tunables: &Tunables, ) { - let destination = send.destination; self.records.insert( id, Record { @@ -92,60 +78,24 @@ impl> UnroutedUnicasts { priority: send.priority, waiter: send.waiter, confirm: send.confirm, - destination, + destination: send.destination, deadline: now + tunables.handoff_timeout(), }, ); - - if !self.discoveries.contains_key(&destination) { - self.discoveries.insert( - destination, - Discovery { - // The tunable counts discoveries; the first one is not a restart - restarts_left: tunables - .pending_route_discovery_attempts() - .saturating_sub(1), - request: None, - }, - ); - } - } - - pub(super) fn destinations(&self) -> Vec { - self.discoveries.keys().copied().collect() } - /// The driver broadcast a route request for `destination`'s discovery. Returns the - /// broadcast of the request it replaces, for the caller to stop. - pub(super) fn route_request_sent( - &mut self, - destination: Nwk, - request: SendId, - ) -> Option { - self.discoveries - .get_mut(&destination) - .unwrap() - .request - .replace(request) + /// The id of every held send, oldest first. + pub(super) fn ids(&self) -> Vec { + self.records.keys().copied().collect() } - /// Take every send to `destination`: its route is known. - pub(super) fn route(&mut self, destination: Nwk) -> Vec<(SendId, Record)> { - self.discoveries.remove(&destination); - self.take(destination) + pub(super) fn frame_mut(&mut self, id: SendId) -> &mut F { + &mut self.records.get_mut(&id).unwrap().frame } - fn take(&mut self, destination: Nwk) -> Vec<(SendId, Record)> { - let ids: Vec = self - .records - .iter() - .filter(|(_, send)| send.destination == destination) - .map(|(id, _)| *id) - .collect(); - - ids.into_iter() - .map(|id| (id, self.records.remove(&id).unwrap())) - .collect() + /// Take a held send: it has a route. + pub(super) fn take(&mut self, id: SendId) -> Record { + self.records.remove(&id).unwrap() } fn settled(send: Record, failure: SendFailure) -> Action { @@ -157,44 +107,22 @@ impl> UnroutedUnicasts { } } - /// Consume a restart of `destination`'s discovery, if one is left and a send still - /// waits on it. - pub(super) fn retry_discovery(&mut self, destination: Nwk) -> bool { - let waiting = self + /// Fail every send to `destination`: no route to it will be found. + pub(super) fn fail(&mut self, destination: Nwk) -> Actions { + let ids: Vec = self .records - .values() - .any(|send| send.destination == destination); - - let discovery = self.discoveries.get_mut(&destination).unwrap(); - // A discovery nothing waits on any more is not restarted - if discovery.restarts_left == 0 || !waiting { - return false; - } - - discovery.restarts_left -= 1; - true - } - - /// Fail every send to `destination`: its discovery is over without a route. Returns - /// the broadcast of its route request, for the caller to stop. - pub(super) fn fail(&mut self, destination: Nwk) -> (Option, Actions) { - let request = self - .discoveries - .remove(&destination) - .and_then(|discovery| discovery.request); - - let actions = self - .take(destination) - .into_iter() - .map(|(_, send)| Self::settled(send, SendFailure::RouteDiscoveryTimeout)) + .iter() + .filter(|(_, send)| send.destination == destination) + .map(|(id, _)| *id) .collect(); - (request, actions) + ids.into_iter() + .map(|id| Self::settled(self.take(id), SendFailure::RouteDiscoveryTimeout)) + .collect() } /// Drop every record and return its waiter: the stack is shutting down. pub(super) fn abandon(&mut self) -> Vec { - self.discoveries.clear(); let mut waiters = Vec::new(); while let Some((_, send)) = self.records.pop_first() { waiters.push(send.waiter); @@ -202,8 +130,6 @@ impl> UnroutedUnicasts { waiters } - /// The destination stays awaited, so a new send to it does not start a second - /// discovery; it lapses on its own once nothing waits on it. pub(super) fn cancel(&mut self, id: SendId) -> Actions { let send = self.records.remove(&id).unwrap(); let mut actions = Actions::new();