1use core::marker::PhantomData;
2
3use embassy_sync::blocking_mutex::raw::RawMutex;
4use embassy_sync::channel::{Receiver, Sender};
5use embassy_sync::signal::Signal;
6use embassy_sync::zerocopy_channel;
7use heapless::Vec as HeaplessVec;
8use portable_atomic::{AtomicBool, AtomicU32, Ordering};
9use static_cell::{ConstStaticCell, StaticCell};
10
11use crate::engine::IssuedCommand;
12use crate::interfaces::{IfacContext, InterfaceDescriptor, InterfaceId, InterfaceIfac};
13use crate::manifold::driver::{
14 embassy_grant_lane, EmbassyGrantConsumer, EmbassyGrantProducer, EmbassyInterfaceSeam,
15 InterfaceLifecycle, PooledEgress,
16};
17use crate::manifold::grant::{FrameSlot, ManifoldLaneReader, ManifoldLaneWriter};
18use crate::manifold::interface_seam::EMBEDDED_MAX_LINK_MTU;
19
20use super::command_handle::PrnsNodeHandle;
21use super::interface_lifecycle::{Fleet, FleetWire};
22use super::node_lifecycle::ManifoldWiring;
23
24#[must_use]
25pub const fn minimum_manifold_notification_capacity(lane_count: usize, lane_depth: usize) -> usize {
26 assert!(lane_count > 0);
27 assert!(lane_depth > 0);
28 assert!(lane_count <= usize::MAX / lane_depth);
29 lane_count * lane_depth
30}
31
32type LaneBuffer<const FRAME: usize, const DEPTH: usize> = [FrameSlot<FRAME>; DEPTH];
33type LaneChannel<M, const FRAME: usize> = zerocopy_channel::Channel<'static, M, FrameSlot<FRAME>>;
34
35#[derive(Debug, PartialEq, Eq)]
36pub enum LaneClaimError {
37 AlreadyClaimed,
38 DuplicateInterfaceId { id: InterfaceId },
39 LaneSetFull { capacity: usize },
40 NotificationCapacityExceeded { required: usize, capacity: usize },
41 FrameCapacityExceeded { required: usize, capacity: usize },
42 EmptyOutboundBuffer,
43}
44
45pub struct StaticManifoldLane<
46 M: RawMutex + 'static,
47 const FRAME: usize,
48 const INBOUND_DEPTH: usize,
49 const OUTBOUND_DEPTH: usize = INBOUND_DEPTH,
50> {
51 taken: AtomicBool,
52 ingress_pressure_events: AtomicU32,
53 egress_pressure_events: AtomicU32,
54 inbound_buffer: ConstStaticCell<LaneBuffer<FRAME, INBOUND_DEPTH>>,
55 inbound_channel: StaticCell<LaneChannel<M, FRAME>>,
56 manifold_inbound: StaticCell<EmbassyGrantConsumer<'static, M, FRAME>>,
57 outbound_buffer: ConstStaticCell<LaneBuffer<FRAME, OUTBOUND_DEPTH>>,
58 outbound_channel: StaticCell<LaneChannel<M, FRAME>>,
59 manifold_outbound: StaticCell<EmbassyGrantProducer<'static, M, FRAME>>,
60}
61
62impl<
63 M: RawMutex + Sync + 'static,
64 const FRAME: usize,
65 const INBOUND_DEPTH: usize,
66 const OUTBOUND_DEPTH: usize,
67 > StaticManifoldLane<M, FRAME, INBOUND_DEPTH, OUTBOUND_DEPTH>
68{
69 #[must_use]
70 pub const fn new() -> Self {
71 Self {
72 taken: AtomicBool::new(false),
73 ingress_pressure_events: AtomicU32::new(0),
74 egress_pressure_events: AtomicU32::new(0),
75 inbound_buffer: ConstStaticCell::new([const { FrameSlot::empty() }; INBOUND_DEPTH]),
76 inbound_channel: StaticCell::new(),
77 manifold_inbound: StaticCell::new(),
78 outbound_buffer: ConstStaticCell::new([const { FrameSlot::empty() }; OUTBOUND_DEPTH]),
79 outbound_channel: StaticCell::new(),
80 manifold_outbound: StaticCell::new(),
81 }
82 }
83
84 fn try_take(
85 &'static self,
86 id: InterfaceId,
87 outbound_wake: Option<&'static Signal<M, ()>>,
88 external_outbound: Option<&'static mut [FrameSlot<FRAME>]>,
89 ) -> Result<TakenManifoldLane<M, FRAME>, LaneClaimError> {
90 if external_outbound
91 .as_ref()
92 .map_or(OUTBOUND_DEPTH == 0, |buffer| buffer.is_empty())
93 {
94 return Err(LaneClaimError::EmptyOutboundBuffer);
95 }
96 if self
97 .taken
98 .compare_exchange(false, true, Ordering::AcqRel, Ordering::Acquire)
99 .is_err()
100 {
101 return Err(LaneClaimError::AlreadyClaimed);
102 }
103
104 let inbound_channel = self
105 .inbound_channel
106 .init(zerocopy_channel::Channel::new(self.inbound_buffer.take()));
107 let (mut interface_inbound, manifold_inbound) = embassy_grant_lane(inbound_channel);
108 interface_inbound.set_pressure_counter(&self.ingress_pressure_events);
109
110 let outbound_buffer =
111 external_outbound.unwrap_or_else(|| self.outbound_buffer.take().as_mut_slice());
112 let outbound_channel = self
113 .outbound_channel
114 .init(zerocopy_channel::Channel::new(outbound_buffer));
115 let (mut manifold_outbound, interface_outbound) = embassy_grant_lane(outbound_channel);
116 manifold_outbound.set_pressure_counter(&self.egress_pressure_events);
117 if let Some(wake) = outbound_wake {
118 manifold_outbound.set_outbound_wake(wake);
119 }
120
121 Ok(TakenManifoldLane {
122 interface: InterfaceLane {
123 id,
124 inbound: interface_inbound,
125 outbound: interface_outbound,
126 },
127 manifold_inbound: self.manifold_inbound.init(manifold_inbound),
128 manifold_outbound: self.manifold_outbound.init(manifold_outbound),
129 })
130 }
131
132 #[must_use]
133 pub fn ingress_pressure_events(&self) -> u32 {
134 self.ingress_pressure_events.load(Ordering::Relaxed)
135 }
136
137 #[must_use]
138 pub fn egress_pressure_events(&self) -> u32 {
139 self.egress_pressure_events.load(Ordering::Relaxed)
140 }
141}
142
143impl<
144 M: RawMutex + Sync + 'static,
145 const FRAME: usize,
146 const INBOUND_DEPTH: usize,
147 const OUTBOUND_DEPTH: usize,
148 > Default for StaticManifoldLane<M, FRAME, INBOUND_DEPTH, OUTBOUND_DEPTH>
149{
150 fn default() -> Self {
151 Self::new()
152 }
153}
154
155struct TakenManifoldLane<M: RawMutex + 'static, const FRAME: usize> {
156 interface: InterfaceLane<M, FRAME>,
157 manifold_inbound: &'static mut EmbassyGrantConsumer<'static, M, FRAME>,
158 manifold_outbound: &'static mut EmbassyGrantProducer<'static, M, FRAME>,
159}
160
161pub struct ManifoldLaneSet<M: RawMutex + 'static, const LANE_COUNT: usize, const NOTIFY: usize> {
162 inbound: HeaplessVec<(InterfaceId, &'static mut dyn ManifoldLaneReader), LANE_COUNT>,
163 egress: PooledEgress<LANE_COUNT>,
164 initial: HeaplessVec<InterfaceDescriptor, LANE_COUNT>,
165 ifacs: HeaplessVec<InterfaceIfac, LANE_COUNT>,
166 notification_capacity: usize,
167 mutex: PhantomData<M>,
168}
169
170impl<M: RawMutex + Sync + 'static, const LANE_COUNT: usize, const NOTIFY: usize>
171 ManifoldLaneSet<M, LANE_COUNT, NOTIFY>
172{
173 #[must_use]
174 pub fn new() -> Self {
175 Self {
176 inbound: HeaplessVec::new(),
177 egress: PooledEgress::new(),
178 initial: HeaplessVec::new(),
179 ifacs: HeaplessVec::new(),
180 notification_capacity: 0,
181 mutex: PhantomData,
182 }
183 }
184
185 pub fn claim_interface<
186 const FRAME: usize,
187 const INBOUND_DEPTH: usize,
188 const OUTBOUND_DEPTH: usize,
189 >(
190 &mut self,
191 storage: &'static StaticManifoldLane<M, FRAME, INBOUND_DEPTH, OUTBOUND_DEPTH>,
192 descriptor: InterfaceDescriptor,
193 ) -> Result<InterfaceLane<M, FRAME>, LaneClaimError> {
194 self.claim_interface_configuration(storage, descriptor, None, None)
195 }
196
197 pub fn claim_interface_with_outbound_buffer<
199 const FRAME: usize,
200 const INBOUND_DEPTH: usize,
201 const OUTBOUND_DEPTH: usize,
202 >(
203 &mut self,
204 storage: &'static StaticManifoldLane<M, FRAME, INBOUND_DEPTH, OUTBOUND_DEPTH>,
205 descriptor: InterfaceDescriptor,
206 outbound_buffer: &'static mut [FrameSlot<FRAME>],
207 ) -> Result<InterfaceLane<M, FRAME>, LaneClaimError> {
208 self.claim_interface_configuration(storage, descriptor, None, Some(outbound_buffer))
209 }
210
211 pub fn claim_interface_with_ifac<
212 const FRAME: usize,
213 const INBOUND_DEPTH: usize,
214 const OUTBOUND_DEPTH: usize,
215 >(
216 &mut self,
217 storage: &'static StaticManifoldLane<M, FRAME, INBOUND_DEPTH, OUTBOUND_DEPTH>,
218 descriptor: InterfaceDescriptor,
219 context: IfacContext,
220 ) -> Result<InterfaceLane<M, FRAME>, LaneClaimError> {
221 self.claim_interface_configuration(storage, descriptor, Some(context), None)
222 }
223
224 fn claim_interface_configuration<
225 const FRAME: usize,
226 const INBOUND_DEPTH: usize,
227 const OUTBOUND_DEPTH: usize,
228 >(
229 &mut self,
230 storage: &'static StaticManifoldLane<M, FRAME, INBOUND_DEPTH, OUTBOUND_DEPTH>,
231 mut descriptor: InterfaceDescriptor,
232 context: Option<IfacContext>,
233 external_outbound: Option<&'static mut [FrameSlot<FRAME>]>,
234 ) -> Result<InterfaceLane<M, FRAME>, LaneClaimError> {
235 self.validate_claim::<INBOUND_DEPTH>(descriptor.id)?;
236 if let Some(mtu) = descriptor.hardware_mtu {
237 descriptor.hardware_mtu = Some(mtu.min(EMBEDDED_MAX_LINK_MTU));
238 }
239 let required = descriptor
240 .hardware_mtu
241 .unwrap_or(crate::wire::BROADCAST_MTU)
242 + context.as_ref().map_or(0, |ifac| ifac.ifac_size().bytes());
243 if required > FRAME {
244 return Err(LaneClaimError::FrameCapacityExceeded {
245 required,
246 capacity: FRAME,
247 });
248 }
249
250 let id = descriptor.id;
251 let taken = storage.try_take(id, None, external_outbound)?;
252 self.register_lane::<INBOUND_DEPTH>(id, taken.manifold_inbound, taken.manifold_outbound);
253 if self.initial.push(descriptor).is_err() {
254 unreachable!()
255 }
256 if let Some(context) = context {
257 if self.ifacs.push(InterfaceIfac { id, context }).is_err() {
258 unreachable!()
259 }
260 }
261 Ok(taken.interface)
262 }
263
264 pub fn claim_supervisor<
265 const FRAME: usize,
266 const INBOUND_DEPTH: usize,
267 const OUTBOUND_DEPTH: usize,
268 >(
269 &mut self,
270 storage: &'static StaticManifoldLane<M, FRAME, INBOUND_DEPTH, OUTBOUND_DEPTH>,
271 supervisor: InterfaceId,
272 outbound_wake: &'static Signal<M, ()>,
273 ) -> Result<SupervisorLane<M, FRAME>, LaneClaimError> {
274 self.claim_supervisor_configuration(storage, supervisor, None, outbound_wake, None)
275 }
276
277 pub fn claim_supervisor_with_outbound_buffer<
283 const FRAME: usize,
284 const INBOUND_DEPTH: usize,
285 const OUTBOUND_DEPTH: usize,
286 >(
287 &mut self,
288 storage: &'static StaticManifoldLane<M, FRAME, INBOUND_DEPTH, OUTBOUND_DEPTH>,
289 supervisor: InterfaceId,
290 outbound_wake: &'static Signal<M, ()>,
291 outbound_buffer: &'static mut [FrameSlot<FRAME>],
292 ) -> Result<SupervisorLane<M, FRAME>, LaneClaimError> {
293 self.claim_supervisor_configuration(
294 storage,
295 supervisor,
296 None,
297 outbound_wake,
298 Some(outbound_buffer),
299 )
300 }
301
302 pub fn claim_supervisor_with_ifac<
303 const FRAME: usize,
304 const INBOUND_DEPTH: usize,
305 const OUTBOUND_DEPTH: usize,
306 >(
307 &mut self,
308 storage: &'static StaticManifoldLane<M, FRAME, INBOUND_DEPTH, OUTBOUND_DEPTH>,
309 supervisor: InterfaceId,
310 context: IfacContext,
311 outbound_wake: &'static Signal<M, ()>,
312 ) -> Result<SupervisorLane<M, FRAME>, LaneClaimError> {
313 self.claim_supervisor_configuration(storage, supervisor, Some(context), outbound_wake, None)
314 }
315
316 fn claim_supervisor_configuration<
317 const FRAME: usize,
318 const INBOUND_DEPTH: usize,
319 const OUTBOUND_DEPTH: usize,
320 >(
321 &mut self,
322 storage: &'static StaticManifoldLane<M, FRAME, INBOUND_DEPTH, OUTBOUND_DEPTH>,
323 supervisor: InterfaceId,
324 context: Option<IfacContext>,
325 outbound_wake: &'static Signal<M, ()>,
326 external_outbound: Option<&'static mut [FrameSlot<FRAME>]>,
327 ) -> Result<SupervisorLane<M, FRAME>, LaneClaimError> {
328 self.validate_claim::<INBOUND_DEPTH>(supervisor)?;
329 let taken = storage.try_take(supervisor, Some(outbound_wake), external_outbound)?;
330 self.register_lane::<INBOUND_DEPTH>(
331 supervisor,
332 taken.manifold_inbound,
333 taken.manifold_outbound,
334 );
335 if let Some(context) = context {
336 if self
337 .ifacs
338 .push(InterfaceIfac {
339 id: supervisor,
340 context,
341 })
342 .is_err()
343 {
344 unreachable!()
345 }
346 }
347 Ok(SupervisorLane {
348 lane: taken.interface,
349 outbound_wake,
350 })
351 }
352
353 fn validate_claim<const DEPTH: usize>(&self, id: InterfaceId) -> Result<(), LaneClaimError> {
354 if self.inbound.iter().any(|(existing, _)| *existing == id) {
355 return Err(LaneClaimError::DuplicateInterfaceId { id });
356 }
357 if self.inbound.len() == LANE_COUNT {
358 return Err(LaneClaimError::LaneSetFull {
359 capacity: LANE_COUNT,
360 });
361 }
362 let required = self.notification_capacity.saturating_add(DEPTH);
363 if required > NOTIFY {
364 return Err(LaneClaimError::NotificationCapacityExceeded {
365 required,
366 capacity: NOTIFY,
367 });
368 }
369 Ok(())
370 }
371
372 fn register_lane<const DEPTH: usize>(
373 &mut self,
374 id: InterfaceId,
375 inbound: &'static mut dyn ManifoldLaneReader,
376 outbound: &'static mut dyn ManifoldLaneWriter,
377 ) {
378 if self.inbound.push((id, inbound)).is_err() {
379 unreachable!()
380 }
381 if self.egress.push(id, outbound).is_err() {
382 unreachable!()
383 }
384 self.notification_capacity += DEPTH;
385 }
386
387 pub fn into_manifold_wiring<
388 const COMMANDS: usize,
389 const LIFECYCLE: usize,
390 const COMPLETIONS: usize,
391 >(
392 self,
393 notify: Receiver<'static, M, InterfaceId, NOTIFY>,
394 commands: Receiver<'static, M, IssuedCommand, COMMANDS>,
395 lifecycle: Receiver<'static, M, InterfaceLifecycle, LIFECYCLE>,
396 handle: PrnsNodeHandle<'static, M, COMMANDS, COMPLETIONS>,
397 ) -> ManifoldWiring<M, LANE_COUNT, NOTIFY, COMMANDS, LIFECYCLE, COMPLETIONS> {
398 ManifoldWiring {
399 inbound: self.inbound,
400 egress: self.egress,
401 initial: self.initial,
402 ifacs: self.ifacs,
403 notify,
404 commands,
405 lifecycle,
406 handle,
407 }
408 }
409}
410
411impl<M: RawMutex + Sync + 'static, const LANE_COUNT: usize, const NOTIFY: usize> Default
412 for ManifoldLaneSet<M, LANE_COUNT, NOTIFY>
413{
414 fn default() -> Self {
415 Self::new()
416 }
417}
418
419pub struct InterfaceLane<M: RawMutex + 'static, const FRAME: usize> {
420 id: InterfaceId,
421 inbound: EmbassyGrantProducer<'static, M, FRAME>,
422 outbound: EmbassyGrantConsumer<'static, M, FRAME>,
423}
424
425impl<M: RawMutex + 'static, const FRAME: usize> InterfaceLane<M, FRAME> {
426 pub fn into_seam<const NOTIFY: usize>(
427 self,
428 notify: Sender<'static, M, InterfaceId, NOTIFY>,
429 fill_entropy: fn(&mut [u8]),
430 ) -> EmbassyInterfaceSeam<'static, M, NOTIFY, FRAME> {
431 EmbassyInterfaceSeam::new(self.id, self.inbound, notify, self.outbound, fill_entropy)
432 }
433}
434
435pub struct SupervisorLane<M: RawMutex + 'static, const FRAME: usize> {
436 lane: InterfaceLane<M, FRAME>,
437 outbound_wake: &'static Signal<M, ()>,
438}
439
440impl<M: RawMutex + 'static, const FRAME: usize> SupervisorLane<M, FRAME> {
441 pub fn into_fleet<const NOTIFY: usize, const LIFECYCLE: usize>(
442 self,
443 notify: Sender<'static, M, InterfaceId, NOTIFY>,
444 lifecycle: Sender<'static, M, InterfaceLifecycle, LIFECYCLE>,
445 ) -> Fleet<M, FRAME, NOTIFY, LIFECYCLE> {
446 Fleet::new(
447 FleetWire {
448 inbound: self.lane.inbound,
449 outbound: self.lane.outbound,
450 notify,
451 outbound_wake: self.outbound_wake,
452 },
453 lifecycle,
454 )
455 }
456}
457
458#[cfg(test)]
459mod tests;