prns_runtime_embassy/manifold/driver/pooled_topology/
mod.rs

1use embassy_futures::select::{select6, Either6};
2use embassy_futures::yield_now;
3use embassy_sync::blocking_mutex::raw::RawMutex;
4use embassy_sync::channel::Receiver;
5use heapless::Vec as HeaplessVec;
6
7use crate::engine::{
8    ClassifiedInboundPacket, Departure, EngineState, IngestIo, IssuedCommand, Journaled,
9    ProofRequest,
10};
11use crate::interfaces::InterfaceIfac;
12use crate::interfaces::{AttachedInterfaces, InboundPacket, InterfaceDescriptor, InterfaceId};
13use crate::manifold::grant::{FrameTarget, ManifoldLaneReader};
14use crate::manifold::interface_seam::{EMBEDDED_MAX_LINK_MTU, EMBEDDED_MAX_WIRE_FRAME_LEN};
15use crate::manifold::kernel::{fire_due_reason, merge_wake_schedules_delta};
16use crate::manifold::timers::{wait_for_due_reason, wait_for_pacer};
17use crate::manifold::{AppDeciders, Host};
18use crate::routing::links::resources::ResourceOffer;
19use crate::runtime::{InterfaceInspectionStore, ManifoldPersistence};
20use crate::storage::{DirtyInterfaceSet, StorageLayout};
21
22use super::egress::{
23    flush_due_pacers, ifac_for, route_reaction, soonest_pacer_release, InterfacePacer,
24    ManifoldEgress, PooledEgress,
25};
26use super::packet_phy::retain_packet_phy;
27
28/// Changes the live descriptor set without reallocating the fixed lane pool.
29#[repr(C)]
30pub enum InterfaceLifecycle {
31    Add {
32        descriptor: InterfaceDescriptor,
33    },
34    Remove {
35        id: InterfaceId,
36    },
37    Update {
38        descriptor: InterfaceDescriptor,
39    },
40    Retag {
41        old_id: InterfaceId,
42        new_id: InterfaceId,
43        descriptor: InterfaceDescriptor,
44    },
45}
46
47fn clamp_to_embedded_ceiling(mut descriptor: InterfaceDescriptor) -> InterfaceDescriptor {
48    if let Some(mtu) = descriptor.hardware_mtu {
49        descriptor.hardware_mtu = Some(mtu.min(EMBEDDED_MAX_LINK_MTU));
50    }
51    descriptor
52}
53
54fn inbound_source(
55    lane_id: InterfaceId,
56    stamped_source: InterfaceId,
57    descriptors: &[InterfaceDescriptor],
58) -> InterfaceId {
59    if descriptors
60        .iter()
61        .any(|descriptor| descriptor.id == lane_id)
62    {
63        lane_id
64    } else {
65        stamped_source
66    }
67}
68
69/// Borrowed lanes and channels for one pooled-topology manifold run.
70pub struct PooledWiring<
71    'run,
72    M: RawMutex + 'static,
73    const LANE_COUNT: usize,
74    const INTERFACE_CAPACITY: usize,
75    const NOTIFY: usize,
76    const COMMANDS: usize,
77    const LIFECYCLE: usize,
78> {
79    pub descriptors: &'run mut HeaplessVec<InterfaceDescriptor, INTERFACE_CAPACITY>,
80    pub ifacs: &'run mut HeaplessVec<InterfaceIfac, LANE_COUNT>,
81    pub inbound:
82        &'run mut HeaplessVec<(InterfaceId, &'static mut dyn ManifoldLaneReader), LANE_COUNT>,
83    pub egress: &'run mut PooledEgress<LANE_COUNT>,
84    pub notify: Receiver<'run, M, InterfaceId, NOTIFY>,
85    pub commands: Receiver<'run, M, IssuedCommand, COMMANDS>,
86    pub lifecycle: Receiver<'run, M, InterfaceLifecycle, LIFECYCLE>,
87}
88
89/// Runs a mutable descriptor set over a fixed lane pool; `LANE_COUNT` bounds pacers.
90pub(crate) async fn run_pooled<
91    S,
92    H,
93    M,
94    Store,
95    const LANE_COUNT: usize,
96    const INTERFACE_CAPACITY: usize,
97    const NOTIFY: usize,
98    const COMMANDS: usize,
99    const LIFECYCLE: usize,
100>(
101    engine: &mut EngineState<S>,
102    host: &mut H,
103    wiring: PooledWiring<'_, M, LANE_COUNT, INTERFACE_CAPACITY, NOTIFY, COMMANDS, LIFECYCLE>,
104    mut on_journaled: impl FnMut(Journaled<'_>),
105    deciders: AppDeciders<impl FnMut(&ProofRequest) -> bool, impl FnMut(&ResourceOffer) -> bool>,
106    store: &Store,
107    persistence: &mut impl ManifoldPersistence<S>,
108) where
109    S: StorageLayout,
110    H: Host,
111    M: RawMutex + 'static,
112    Store: InterfaceInspectionStore,
113{
114    let AppDeciders {
115        mut should_prove,
116        mut should_accept_resource,
117    } = deciders;
118    let PooledWiring {
119        descriptors,
120        ifacs,
121        inbound,
122        egress,
123        notify,
124        commands,
125        lifecycle,
126    } = wiring;
127    let mut pacers: HeaplessVec<InterfacePacer, LANE_COUNT> = HeaplessVec::new();
128    for descriptor in descriptors.iter_mut() {
129        *descriptor = clamp_to_embedded_ceiling(*descriptor);
130        engine.interface_attached(descriptor.id, host.now());
131        if let Some(lane) = egress.lane_for(descriptor.id) {
132            if !pacers.iter().any(|pacer| pacer.id == lane) {
133                let _ = pacers.push(InterfacePacer::from_descriptor(lane, descriptor));
134            }
135        }
136    }
137    let mut wake_schedules = engine.wake_schedules(AttachedInterfaces::new(&*descriptors));
138    loop {
139        let wake = wake_schedules.soonest(host.now());
140        let pacer_wake = soonest_pacer_release(&pacers);
141
142        let persistence_deadline = persistence.deadline(host.now());
143        match select6(
144            notify.receive(),
145            commands.receive(),
146            wait_for_due_reason(&*host, wake),
147            wait_for_pacer(&*host, pacer_wake),
148            lifecycle.receive(),
149            wait_for_persistence(&*host, persistence_deadline),
150        )
151        .await
152        {
153            Either6::First(_) => {
154                while notify.try_receive().is_ok() {}
155                for (lane_id, lane) in inbound.iter_mut() {
156                    while let Some((target, packet_phy, frame)) = lane.try_read() {
157                        let FrameTarget::Direct(stamped_source) = target else {
158                            lane.release();
159                            continue;
160                        };
161                        // A dedicated lane's live key is authoritative. Runtime retagging updates
162                        // that key atomically with the descriptor, while the already-constructed
163                        // interface seam can still have stamped a queued frame with its prior id.
164                        // Fleet lanes have no descriptor of their own, so their per-member stamp
165                        // remains authoritative instead.
166                        let source = inbound_source(*lane_id, stamped_source, descriptors);
167                        let mut unmasked = [0u8; EMBEDDED_MAX_WIRE_FRAME_LEN];
168                        let bytes = match ifac_for(ifacs, *lane_id) {
169                            Some(entry) => {
170                                let Some(clean_len) =
171                                    entry.context.unmask_inbound(frame, &mut unmasked)
172                                else {
173                                    lane.release();
174                                    continue;
175                                };
176                                &mut unmasked[..clean_len]
177                            }
178                            None => frame,
179                        };
180                        let now = host.now();
181                        let packet = ClassifiedInboundPacket::classify(InboundPacket {
182                            arrived_at: now,
183                            source_interface: source,
184                            bytes,
185                        });
186                        retain_packet_phy(store, &packet, packet_phy);
187                        let delta = engine.ingest_classified_into(
188                            packet,
189                            IngestIo {
190                                interfaces: AttachedInterfaces::new(&*descriptors),
191                                now,
192                                fill_entropy: &mut |entropy| host.fill_entropy(entropy),
193                                should_prove: &mut should_prove,
194                                should_accept_resource: &mut should_accept_resource,
195                                sink: &mut |reaction| {
196                                    route_reaction(
197                                        reaction,
198                                        &mut *egress,
199                                        ifacs,
200                                        &mut pacers,
201                                        now,
202                                        &mut |journaled| {
203                                            persistence.observe(&journaled, now);
204                                            on_journaled(journaled);
205                                        },
206                                    )
207                                },
208                            },
209                        );
210                        lane.release();
211                        merge_wake_schedules_delta(
212                            &mut wake_schedules,
213                            delta,
214                            &*engine,
215                            AttachedInterfaces::new(&*descriptors),
216                        );
217                    }
218                }
219            }
220            Either6::Second(issued) => {
221                let now = host.now();
222                let delta = engine.ingest_command_into(
223                    issued,
224                    AttachedInterfaces::new(&*descriptors),
225                    now,
226                    &mut |entropy| host.fill_entropy(entropy),
227                    &mut |reaction| {
228                        route_reaction(
229                            reaction,
230                            &mut *egress,
231                            ifacs,
232                            &mut pacers,
233                            now,
234                            &mut |journaled| {
235                                persistence.observe(&journaled, now);
236                                on_journaled(journaled);
237                            },
238                        )
239                    },
240                );
241                merge_wake_schedules_delta(
242                    &mut wake_schedules,
243                    delta,
244                    &*engine,
245                    AttachedInterfaces::new(&*descriptors),
246                );
247            }
248            Either6::Third(reason) => {
249                let now = host.now();
250                let delta = fire_due_reason(
251                    &mut *engine,
252                    reason,
253                    now,
254                    AttachedInterfaces::new(&*descriptors),
255                    &mut |bytes| host.fill_entropy(bytes),
256                    &mut |reaction| {
257                        route_reaction(
258                            reaction,
259                            &mut *egress,
260                            ifacs,
261                            &mut pacers,
262                            now,
263                            &mut |journaled| {
264                                persistence.observe(&journaled, now);
265                                on_journaled(journaled);
266                            },
267                        )
268                    },
269                );
270                merge_wake_schedules_delta(
271                    &mut wake_schedules,
272                    delta,
273                    &*engine,
274                    AttachedInterfaces::new(&*descriptors),
275                );
276            }
277            Either6::Fourth(()) => {
278                let now = host.now();
279                flush_due_pacers(&mut pacers, now, &mut *egress, ifacs);
280            }
281            Either6::Fifth(message) => match message {
282                InterfaceLifecycle::Add { descriptor } => {
283                    let descriptor = clamp_to_embedded_ceiling(descriptor);
284                    let id = descriptor.id;
285                    let present = descriptors.iter().any(|existing| existing.id == id);
286                    if !present {
287                        engine.interface_attached(id, host.now());
288                        let _ = descriptors.push(descriptor);
289                        if let Some(lane) = egress.lane_for(id) {
290                            if !pacers.iter().any(|pacer| pacer.id == lane) {
291                                let _ =
292                                    pacers.push(InterfacePacer::from_descriptor(lane, &descriptor));
293                            }
294                        }
295                        wake_schedules =
296                            engine.wake_schedules(AttachedInterfaces::new(&*descriptors));
297                    }
298                    #[cfg(feature = "log")]
299                    log::info!(
300                        target: "personal_hopspot_esp32",
301                        "manifold: Add kind={:?} present={present} descriptors={}",
302                        id.kind(),
303                        descriptors.len()
304                    );
305                }
306                InterfaceLifecycle::Remove { id } => {
307                    let now = host.now();
308                    let departed_lane = egress.lane_for(id);
309                    engine.interface_departed(id, Departure::Forgotten, now);
310                    let found = descriptors
311                        .iter()
312                        .position(|descriptor| descriptor.id == id);
313                    if let Some(pos) = found {
314                        let _ = descriptors.swap_remove(pos);
315                    }
316                    #[cfg(feature = "log")]
317                    log::info!(
318                        target: "personal_hopspot_esp32",
319                        "manifold: Remove kind={:?} found={} descriptors={}",
320                        id.kind(),
321                        found.is_some(),
322                        descriptors.len()
323                    );
324                    if let Some(lane) = departed_lane {
325                        let lane_still_serves_a_descriptor = descriptors
326                            .iter()
327                            .any(|descriptor| egress.lane_for(descriptor.id) == Some(lane));
328                        if !lane_still_serves_a_descriptor {
329                            if let Some(pos) = pacers.iter().position(|pacer| pacer.id == lane) {
330                                let _ = pacers.swap_remove(pos);
331                            }
332                        }
333                    }
334                    engine.cull_expired_routes(
335                        now,
336                        AttachedInterfaces::new(&*descriptors),
337                        &mut |reaction| {
338                            route_reaction(
339                                reaction,
340                                &mut *egress,
341                                ifacs,
342                                &mut pacers,
343                                now,
344                                &mut |journaled| {
345                                    persistence.observe(&journaled, now);
346                                    on_journaled(journaled);
347                                },
348                            )
349                        },
350                    );
351                    wake_schedules = engine.wake_schedules(AttachedInterfaces::new(&*descriptors));
352                }
353                InterfaceLifecycle::Update { descriptor } => {
354                    let descriptor = clamp_to_embedded_ceiling(descriptor);
355                    if let Some(slot) = descriptors
356                        .iter()
357                        .position(|existing| existing.id == descriptor.id)
358                    {
359                        descriptors[slot] = descriptor;
360                        if let Some(lane) = egress.lane_for(descriptor.id) {
361                            if let Some(pos) = pacers.iter().position(|pacer| pacer.id == lane) {
362                                pacers[pos] = InterfacePacer::from_descriptor(lane, &descriptor);
363                            }
364                        }
365                        wake_schedules =
366                            engine.wake_schedules(AttachedInterfaces::new(&*descriptors));
367                    }
368                }
369                InterfaceLifecycle::Retag {
370                    old_id,
371                    new_id,
372                    descriptor,
373                } => {
374                    let descriptor = clamp_to_embedded_ceiling(descriptor);
375                    let present = descriptors
376                        .iter()
377                        .position(|existing| existing.id == old_id);
378                    let collides = descriptors.iter().any(|existing| existing.id == new_id);
379                    if let (Some(slot), false) = (present, collides) {
380                        let old_lane = egress.lane_for(old_id);
381                        descriptors[slot] = descriptor;
382                        egress.retag(old_id, new_id);
383                        if let Some(entry) = inbound.iter_mut().find(|(id, _)| *id == old_id) {
384                            entry.0 = new_id;
385                        }
386                        if let Some(entry) = ifacs.iter_mut().find(|entry| entry.id == old_id) {
387                            entry.id = new_id;
388                        }
389                        if let (Some(old_lane), Some(new_lane)) =
390                            (old_lane, egress.lane_for(new_id))
391                        {
392                            if let Some(pos) = pacers.iter().position(|pacer| pacer.id == old_lane)
393                            {
394                                pacers[pos] =
395                                    InterfacePacer::from_descriptor(new_lane, &descriptor);
396                            }
397                        }
398                        wake_schedules =
399                            engine.wake_schedules(AttachedInterfaces::new(&*descriptors));
400                    }
401                }
402            },
403            Either6::Sixth(()) => {}
404        }
405        let now = host.now();
406        if persistence
407            .deadline(now)
408            .is_some_and(|deadline| deadline.0 <= now.0)
409        {
410            persistence.progress(engine, now).await;
411            yield_now().await;
412        }
413        if Store::RETAINS_COUNTS {
414            let mut dirty = engine.take_dirty_interfaces();
415            let mut changed = false;
416            dirty.drain(|interface| {
417                if descriptors
418                    .iter()
419                    .any(|descriptor| descriptor.id == interface)
420                {
421                    store.set_interface_counts(interface, engine.interface_counts(interface));
422                } else {
423                    store.forget_interface(interface);
424                }
425                changed = true;
426            });
427            if changed {
428                store.signal_interface_counts_changed();
429            }
430        }
431    }
432}
433
434async fn wait_for_persistence(host: &impl Host, deadline: Option<crate::engine::InstantMillis>) {
435    match deadline {
436        Some(deadline) => host.sleep_until(deadline).await,
437        None => core::future::pending().await,
438    }
439}
440
441#[cfg(test)]
442mod tests;
prns_runtime_embassy/manifold/driver/pooled_topology/mod.rs

prns_runtime_embassy/manifold/driver/pooled_topology/
mod.rs
