prns_runtime_embassy/runtime/node_facade/interface_lifecycle/
mod.rs

1use embassy_sync::blocking_mutex::raw::RawMutex;
2use embassy_sync::channel::Sender;
3use embassy_sync::signal::Signal;
4
5use crate::interfaces::{InterfaceDescriptor, InterfaceId};
6use crate::manifold::driver::{EmbassyGrantConsumer, EmbassyGrantProducer, InterfaceLifecycle};
7use crate::manifold::grant::{FrameTarget, GrantConsumer, GrantProducer};
8
9#[derive(Debug, Clone, Copy, PartialEq, Eq)]
10pub enum InboundDeliveryError {
11    FrameTooLarge { len: usize, capacity: usize },
12    LaneFull,
13}
14
15#[derive(Debug, Clone, PartialEq, Eq)]
16pub struct OutboundFrame<const FRAME: usize> {
17    target: FrameTarget,
18    bytes: [u8; FRAME],
19    len: usize,
20}
21
22impl<const FRAME: usize> OutboundFrame<FRAME> {
23    #[must_use]
24    pub fn target(&self) -> FrameTarget {
25        self.target
26    }
27
28    #[must_use]
29    pub fn bytes(&self) -> &[u8] {
30        &self.bytes[..self.len]
31    }
32
33    #[must_use]
34    pub fn is_empty(&self) -> bool {
35        self.len == 0
36    }
37
38    #[must_use]
39    pub fn len(&self) -> usize {
40        self.len
41    }
42}
43
44pub(super) struct FleetWire<M: RawMutex + 'static, const FRAME: usize, const NOTIFY: usize> {
45    pub(super) inbound: EmbassyGrantProducer<'static, M, FRAME>,
46    pub(super) outbound: EmbassyGrantConsumer<'static, M, FRAME>,
47    pub(super) notify: Sender<'static, M, InterfaceId, NOTIFY>,
48    pub(super) outbound_wake: &'static Signal<M, ()>,
49}
50
51pub struct Fleet<
52    M: RawMutex + 'static,
53    const FRAME: usize,
54    const NOTIFY: usize,
55    const LIFECYCLE: usize,
56> {
57    wire: FleetWire<M, FRAME, NOTIFY>,
58    lifecycle: Sender<'static, M, InterfaceLifecycle, LIFECYCLE>,
59}
60
61impl<M: RawMutex + 'static, const FRAME: usize, const NOTIFY: usize, const LIFECYCLE: usize>
62    Fleet<M, FRAME, NOTIFY, LIFECYCLE>
63{
64    #[must_use]
65    pub(super) fn new(
66        wire: FleetWire<M, FRAME, NOTIFY>,
67        lifecycle: Sender<'static, M, InterfaceLifecycle, LIFECYCLE>,
68    ) -> Self {
69        Self { wire, lifecycle }
70    }
71
72    pub async fn register_member(&self, descriptor: InterfaceDescriptor) {
73        self.lifecycle
74            .send(InterfaceLifecycle::Add { descriptor })
75            .await;
76    }
77
78    pub async fn deregister_member(&self, id: InterfaceId) {
79        self.lifecycle.send(InterfaceLifecycle::Remove { id }).await;
80    }
81
82    pub fn try_deliver_inbound(
83        &mut self,
84        child: InterfaceId,
85        bytes: &[u8],
86    ) -> Result<(), InboundDeliveryError> {
87        if bytes.len() > FRAME {
88            return Err(InboundDeliveryError::FrameTooLarge {
89                len: bytes.len(),
90                capacity: FRAME,
91            });
92        }
93        let Some(grant) = self.wire.inbound.try_grant() else {
94            self.wire.inbound.note_pressure();
95            return Err(InboundDeliveryError::LaneFull);
96        };
97        grant.fill_for(child, bytes);
98        self.wire.inbound.commit();
99        let _ = self.wire.notify.try_send(child);
100        Ok(())
101    }
102
103    /// Delivers one member frame with end-to-end backpressure.
104    ///
105    /// Supervisors backed by a reliable transport should use this path: when the manifold is
106    /// still consuming the previous frame, the transport can stop admitting further frames
107    /// instead of silently turning transient scheduler latency into packet loss.
108    pub async fn deliver_inbound(
109        &mut self,
110        child: InterfaceId,
111        bytes: &[u8],
112    ) -> Result<(), InboundDeliveryError> {
113        if bytes.len() > FRAME {
114            return Err(InboundDeliveryError::FrameTooLarge {
115                len: bytes.len(),
116                capacity: FRAME,
117            });
118        }
119        let grant = self.wire.inbound.grant().await;
120        grant.fill_for(child, bytes);
121        self.wire.inbound.commit();
122        self.wire.notify.send(child).await;
123        Ok(())
124    }
125
126    pub async fn next_outbound(&mut self) -> OutboundFrame<FRAME> {
127        self.wire.outbound.release();
128        let slot = self.wire.outbound.peek().await;
129        let target = slot.target;
130        let len = slot.len;
131        let mut bytes = [0; FRAME];
132        bytes[..len].copy_from_slice(slot.frame());
133        self.wire.outbound.release();
134        OutboundFrame { target, bytes, len }
135    }
136
137    /// Waits for a shared-lane commit; drain with [`try_next_outbound`](Self::try_next_outbound) after waking.
138    pub async fn outbound_ready(&self) {
139        self.wire.outbound_wake.wait().await;
140    }
141
142    pub fn try_next_outbound(&mut self) -> Option<OutboundFrame<FRAME>> {
143        let slot = self.wire.outbound.try_peek()?;
144        let target = slot.target;
145        let len = slot.len;
146        let mut bytes = [0; FRAME];
147        bytes[..len].copy_from_slice(slot.frame());
148        self.wire.outbound.release();
149        Some(OutboundFrame { target, bytes, len })
150    }
151}
152
153#[cfg(test)]
154mod tests;
prns_runtime_embassy/runtime/node_facade/interface_lifecycle/mod.rs

prns_runtime_embassy/runtime/node_facade/interface_lifecycle/
mod.rs
