bacnet_network/
layer_ingress.rs

1//! Turning a transport's received NPDUs into APDUs and network controls, in
2//! the task that reads them (#1713).
3//!
4//! [`NetworkLayer::start`](super::NetworkLayer::start) returns an
5//! [`ApduReceiver`], which does this inside its own `recv`: there is no
6//! dispatch task between the transport and the application, and no second
7//! queue. [`NetworkLayer::start_with_admission`](super::NetworkLayer::start_with_admission)
8//! runs the same [`Ingress`] in a dispatch task in front of its tracked queue.
9
10use std::sync::Arc;
11use std::sync::atomic::AtomicU64;
12use std::task::{Context, Poll, ready};
13
14use bacnet_router_core::node::Node;
15use bacnet_transport::port::ReceivedNpdu;
16use bacnet_transport::warn_throttle::WarnThrottle;
17use std::sync::{Mutex, MutexGuard, PoisonError};
18
19use tokio::sync::mpsc;
20use tracing::debug;
21
22use super::receive::{self, Inbound, NodeTotals};
23use super::{AdmissionSender, ReceivedApdu, ReceivedNetworkControl, next_ingress_sequence};
24
25/// A layer's receive state: the router core's node, which applies the
26/// receive rules (#1795), and where network controls go.
27pub(crate) struct Ingress {
28    node: Node,
29    totals: Arc<NodeTotals>,
30    drop_warning: WarnThrottle,
31    controls: Option<AdmissionSender<ReceivedNetworkControl>>,
32    control_sequence: Arc<AtomicU64>,
33}
34
35impl std::fmt::Debug for Ingress {
36    fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
37        f.debug_struct("Ingress")
38            .field("controls", &self.controls.is_some())
39            .finish_non_exhaustive()
40    }
41}
42
43impl Ingress {
44    pub(crate) fn new(
45        totals: Arc<NodeTotals>,
46        controls: Option<AdmissionSender<ReceivedNetworkControl>>,
47        control_sequence: Arc<AtomicU64>,
48    ) -> Self {
49        Self {
50            node: Node::new(),
51            totals,
52            drop_warning: WarnThrottle::default(),
53            controls,
54            control_sequence,
55        }
56    }
57
58    /// The APDU `received` carries, if it carries one. A network message
59    /// goes to the control receiver, if there is one, without waiting; a
60    /// frame the node drops is counted. Nothing here awaits.
61    pub(crate) fn take(&mut self, mut received: ReceivedNpdu) -> Option<ReceivedApdu> {
62        // Moved, not cloned: nothing below reads the frame again.
63        let frame = std::mem::take(&mut received.npdu);
64        let inbound = receive::admit(
65            &mut self.node,
66            &self.totals,
67            &self.drop_warning,
68            &received.source_mac,
69            received.link_layer_group,
70            frame,
71            self.controls.is_some(),
72        );
73        match inbound {
74            Inbound::Apdu {
75                apdu,
76                source_network,
77                is_group,
78                global_broadcast,
79            } => Some(ReceivedApdu {
80                router_response: None,
81                direct_response: received.direct_response,
82                apdu,
83                source_mac: received.source_mac,
84                ingress_network: None,
85                source_network,
86                link_layer_group: received.link_layer_group,
87                is_group,
88                global_broadcast,
89                data_attributes: received.data_attributes,
90                provenance: received.provenance,
91                reply_tx: received.reply_tx,
92            }),
93            Inbound::Control(Some(npdu)) => {
94                let tx = self.controls.as_ref()?;
95                let control = ReceivedNetworkControl {
96                    npdu,
97                    source_mac: received.source_mac,
98                    link_layer_group: received.link_layer_group,
99                    data_attributes: received.data_attributes,
100                    provenance: received.provenance,
101                    ingress_sequence: next_ingress_sequence(&self.control_sequence),
102                };
103                if let Err(mpsc::error::TrySendError::Closed(_)) = tx.try_send(control) {
104                    self.controls = None;
105                    debug!("Network-control receiver closed; resuming discard behavior");
106                }
107                None
108            }
109            Inbound::Control(None) | Inbound::Dropped => None,
110        }
111    }
112}
113
114/// The transport queue an [`ApduReceiver`] takes from, shared with its
115/// layer so [`NetworkLayer::stop`](super::NetworkLayer::stop) can close it.
116/// Never locked across an await.
117pub(crate) type SharedNpduQueue = Arc<Mutex<mpsc::Receiver<ReceivedNpdu>>>;
118
119/// The APDUs a [`NetworkLayer`](super::NetworkLayer) receives, from
120/// [`NetworkLayer::start`](super::NetworkLayer::start).
121///
122/// Each [`recv`](Self::recv) takes the transport's next NPDUs and decodes
123/// them in the caller's task until one carries an APDU. Network messages met
124/// on the way go to the control receiver, if one was enabled, so **controls
125/// flow only while this receiver is polled**: while its consumer is busy in
126/// an inline await they wait in the transport's queue, which adds latency
127/// but cannot deadlock. That queue is the only buffer: under overload it
128/// bounds memory, and the transport counts the NPDUs it drops. See the
129/// [receive-queue contract](super#receive-queue-admission).
130///
131/// [`NetworkLayer::stop`](super::NetworkLayer::stop), or dropping the layer,
132/// closes the transport's queue: the receiver yields what was queued before
133/// then, then `None`, even when the transport keeps its sender alive.
134#[derive(Debug)]
135pub struct ApduReceiver {
136    npdu_rx: SharedNpduQueue,
137    ingress: Ingress,
138}
139
140impl ApduReceiver {
141    pub(crate) fn new(npdu_rx: SharedNpduQueue, ingress: Ingress) -> Self {
142        Self { npdu_rx, ingress }
143    }
144
145    fn queue(&self) -> MutexGuard<'_, mpsc::Receiver<ReceivedNpdu>> {
146        self.npdu_rx.lock().unwrap_or_else(PoisonError::into_inner)
147    }
148
149    /// Receive the next APDU, or `None` once the transport's queue is closed,
150    /// by a stop or [`Self::close`], and every NPDU in it has been taken.
151    ///
152    /// Cancel safe, so it can sit in a `select!`: the only wait is the
153    /// transport queue's own receive, and an NPDU taken from it is decoded
154    /// and handed on before this returns or waits again. Dropping the future
155    /// loses nothing, and it holds no lock while it waits.
156    pub async fn recv(&mut self) -> Option<ReceivedApdu> {
157        std::future::poll_fn(|cx| self.poll_recv(cx)).await
158    }
159
160    /// Poll for the next APDU, as [`Self::recv`] does.
161    pub fn poll_recv(&mut self, cx: &mut Context<'_>) -> Poll<Option<ReceivedApdu>> {
162        let mut queue = self.npdu_rx.lock().unwrap_or_else(PoisonError::into_inner);
163        loop {
164            let Some(received) = ready!(queue.poll_recv(cx)) else {
165                return Poll::Ready(None);
166            };
167            if let Some(apdu) = self.ingress.take(received) {
168                return Poll::Ready(Some(apdu));
169            }
170        }
171    }
172
173    /// Receive an APDU the transport has already queued, without waiting.
174    /// `Empty` means none is queued now; `Disconnected`, that the queue is
175    /// closed and drained.
176    pub fn try_recv(&mut self) -> Result<ReceivedApdu, mpsc::error::TryRecvError> {
177        let mut queue = self.npdu_rx.lock().unwrap_or_else(PoisonError::into_inner);
178        loop {
179            let received = queue.try_recv()?;
180            if let Some(apdu) = self.ingress.take(received) {
181                return Ok(apdu);
182            }
183        }
184    }
185
186    /// Stop the transport handing up more NPDUs, keeping those it already
187    /// queued for [`Self::recv`] to drain.
188    pub fn close(&mut self) {
189        self.queue().close();
190    }
191
192    /// Whether the transport can no longer hand up NPDUs: the layer stopped,
193    /// or this receiver was closed. Queued ones may remain.
194    pub fn is_closed(&self) -> bool {
195        self.queue().is_closed()
196    }
197}
198
199/// Close `queue`, if its receiver still exists, so its stream ends once
200/// drained. Run by the layer's stop and drop.
201pub(crate) fn close_queue(queue: &std::sync::Weak<Mutex<mpsc::Receiver<ReceivedNpdu>>>) {
202    if let Some(queue) = queue.upgrade() {
203        queue.lock().unwrap_or_else(PoisonError::into_inner).close();
204    }
205}
bacnet_network/layer_ingress.rs

bacnet_network/
layer_ingress.rs
