1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
165
166
167
168
169
170
171
172
173
174
175
176
177
178
179
180
181
182
183
184
185
186
187
188
189
190
191
192
193
194
195
196
197
198
199
200
201
202
203
204
205
206
207
208
209
210
211
212
213
214
215
216
217
218
219
220
221
222
223
224
225
226
227
228
229
230
231
232
233
234
235
236
237
238
239
240
241
//! Inbound dispatch: what the router does with each frame a port receives.
//!
//! Every port runs one [`PortDispatch`] task. For each frame it takes the
//! engine's lock, hands the frame to the engine (`bacnet-router-core`,
//! #1679), drains the sends the engine asks for onto the port queues, and
//! lets the lock go. The engine decides everything Clause 6 asks of a
//! router, in the order the router always followed: the address bounds,
//! the broadcast rules, network messages (routed first when they name
//! another network), local delivery, routing, rejects and the search for
//! unknown networks. Then, outside the lock, the frame's APDU goes to the
//! local application queue when the engine delivers it here, and a reject
//! addressed to the router goes to its own network-control consumer.
use std::sync::Arc;
use bacnet_encoding::npdu::Npdu;
use bacnet_npdu::{MAX_MAC_LEN, NpduLimits};
use bacnet_router_core::PortId;
use bacnet_router_core::router::{Discard, Frame, Received};
use bacnet_transport::port::{ReceivedNpdu, TransportProvenance};
use tokio::sync::{mpsc, watch};
use tracing::{debug, warn};
use super::actions::drain;
use super::control_policy::ControlGate;
use super::engine::Engine;
use super::hosted_host::{EngineWarnings, HostedHost};
use super::local_control::LocalControl;
use super::local_delivery::Delivery;
use super::send_queue;
use crate::layer::{AdmissionSender, ReceivedApdu};
/// What the router, every port's dispatch and the sweep share.
pub(super) struct Shared {
/// The engine and its clock.
pub(super) engine: Engine,
/// The RB-09 wire-control gate.
pub(super) gate: Arc<ControlGate>,
/// Each port's network number, by port index.
pub(super) networks: Vec<u16>,
/// The throttles for the engine's per-frame warnings.
pub(super) warnings: EngineWarnings,
/// The smallest NPDU limits among the ports, as they change: what the
/// router's application ports report (#2070).
pub(super) application_limits: watch::Sender<NpduLimits>,
}
impl Shared {
/// The state the dispatch tasks share, for `engine` on ports with
/// `networks`, gated by `gate`.
pub(super) fn new(engine: Engine, gate: Arc<ControlGate>, networks: Vec<u16>) -> Self {
let limits = super::limits::smallest(engine.lock().ports());
Self {
engine,
gate,
networks,
warnings: EngineWarnings::default(),
application_limits: watch::Sender::new(limits),
}
}
/// Queue the startup announcements and send them on `send_txs`.
pub(super) fn announce(&self, send_txs: &[send_queue::Sender]) {
let mut router = self.engine.lock();
router.start();
drain(&mut router, &[], send_txs, self.engine.now());
}
/// Run the engine's sweep if it is due, and return when the next is.
pub(super) fn sweep(&self) -> Option<bacnet_router_core::Instant> {
let mut host = HostedHost::new(
&self.gate,
&self.warnings,
TransportProvenance::unverified(),
);
let mut router = self.engine.lock();
router.on_deadline(self.engine.now(), &mut host);
// The sweep queues no sends; were one to come, it would be refused
// here rather than left waiting.
drain(&mut router, &[], &[], self.engine.now());
router.next_deadline()
}
}
/// Where the dispatch tasks put what they don't route: the port queues and
/// the local consumers. Only the dispatch tasks hold it, so when
/// [`BACnetRouter::stop`](super::BACnetRouter::stop) ends them, the local
/// receivers see their queues close.
pub(super) struct Outputs {
pub(super) replies: Arc<super::replies::ReplyLifetime>,
/// Each port's send queue, by port index.
pub(super) send_txs: Vec<send_queue::Sender>,
/// The local application receive queue.
pub(super) local_tx: AdmissionSender<ReceivedApdu>,
/// The router's own network-control consumer.
pub(super) local_control: LocalControl,
}
/// What the engine made of a frame, owned, so the frame can move on.
enum Outcome {
/// Its APDU is for the router's application.
Local(Delivery),
/// A reject addressed to the router: its control information, read
/// from the engine's view, and where its data starts, when the router
/// has a network-control consumer.
ToRouter(Option<(Npdu, usize)>),
/// Discarded, and counted, for this reason.
Discarded(Discard),
/// Routed, answered or taken: the sends say the rest.
Handled,
}
/// One port's dispatch task.
pub(super) struct PortDispatch {
/// What the router shares.
pub(super) shared: Arc<Shared>,
/// Where the frames go.
pub(super) outputs: Arc<Outputs>,
/// This port's index.
pub(super) port: u8,
}
impl PortDispatch {
/// Dispatch every frame the port receives until its channel closes.
pub(super) async fn run(self, mut rx: mpsc::Receiver<ReceivedNpdu>) {
while let Some(received) = rx.recv().await {
self.dispatch(received);
}
}
/// Dispatch one frame. Nothing here waits: the engine's lock is held
/// for the call and its drain only, and every queue takes `try_send`.
pub(super) fn dispatch(&self, received: ReceivedNpdu) {
let (shared, outputs) = (&*self.shared, &*self.outputs);
let port = PortId(self.port);
let frame = Frame {
port,
source: &received.source_mac,
link_group: received.link_layer_group,
npdu: &received.npdu,
};
let mut host = HostedHost::new(&shared.gate, &shared.warnings, received.provenance.clone());
let outcome = {
let mut router = shared.engine.lock();
let failures = router.counters().write_failures;
let outcome = match router.receive(frame, shared.engine.now(), &mut host) {
Received::Local(local) => Outcome::Local(Delivery::of(
&local,
router.ports()[usize::from(port.0)].max_npdu.unicast,
)),
Received::ToRouter(npdu) => Outcome::ToRouter(
outputs
.local_control
.wanted()
.then(|| (Npdu::header_from(&npdu), npdu.payload_offset)),
),
Received::Discarded(reason) => Outcome::Discarded(reason),
Received::Handled => Outcome::Handled,
};
// A rewrite or relay that couldn't be written; one the frame's
// own result names is logged below.
if router.counters().write_failures != failures
&& !matches!(outcome, Outcome::Discarded(Discard::Unwritable))
{
warn!(port = self.port, "Failed to encode forwarded NPDU");
}
drain(
&mut router,
&received.data_attributes,
&outputs.send_txs,
shared.engine.now(),
);
outcome
};
let network = shared
.networks
.get(usize::from(self.port))
.copied()
.unwrap_or_default();
match outcome {
Outcome::Local(delivery) => {
let _ = outputs.local_tx.try_send_apdu(delivery.application(
received,
network,
&self.outputs,
));
}
Outcome::ToRouter(Some((header, offset))) => {
outputs.local_control.deliver(header, offset, &received);
}
Outcome::ToRouter(None) => {}
Outcome::Discarded(reason) => {
log_discard(reason, self.port, received.source_mac.len());
}
Outcome::Handled => {}
}
}
}
/// Log a discarded frame as the router logged it before the engine. Hop
/// counts are logged by the engine's event; denials, refused network
/// messages and unroutable rejects drop without a word, as they did.
fn log_discard(reason: Discard, port: u8, source_len: usize) {
match reason {
Discard::SourceTooLong { .. } => warn!(
octets = source_len,
limit = MAX_MAC_LEN,
"Dropping a frame whose link-layer source MAC is over the limit"
),
Discard::AddressTooLong { field, length } => {
warn!(
error = %format_args!(
"NPDU {}={length} exceeds the {MAX_MAC_LEN}-octet address limit",
field.length_octet()
),
port,
"Router refused an NPDU"
);
}
Discard::Malformed(fault) => warn!(error = %fault, port, "Router decode failed"),
Discard::GlobalWithDadr => {
warn!("Dropping a global broadcast NPDU that also names a DADR")
}
Discard::BroadcastNotUnconfirmed => {
warn!("Dropping a broadcast NPDU whose APDU isn't an Unconfirmed-Request")
}
Discard::GroupDadr => warn!(
port,
"Dropping a routed NPDU for a group DADR its port's broadcast doesn't stand for, \
or that isn't an Unconfirmed-Request"
),
Discard::Unwritable => warn!(port, "Failed to encode forwarded NPDU"),
Discard::UnknownPort
| Discard::HopCount
| Discard::Denied
| Discard::BadMessage
| Discard::Unroutable => {}
// A reason a later engine adds.
_ => debug!(port, ?reason, "Router dropped a frame"),
}
}