prns_runtime_embassy/runtime/node_facade/command_handle/
mod.rs

1use core::cell::RefCell;
2
3use embassy_sync::blocking_mutex::raw::RawMutex;
4use embassy_sync::blocking_mutex::Mutex as BlockingMutex;
5use embassy_sync::channel::Sender;
6use embassy_sync::signal::Signal;
7use portable_atomic::{AtomicU64, Ordering};
8
9use crate::engine::{
10    CloseLink, CommandId, IssuedCommand, PacketReceiptDelivered, PrnsCommand, Respond, RespondData,
11    RespondPayload, SendSinglePacket, SendSinglePacketFailure, SendSinglePacketPayload, Settlement,
12};
13use crate::routing::links::LinkId;
14use crate::wire::DestinationHash;
15
16use super::super::request_endpoints::RespondToken;
17use super::super::{PrnsNodeApi, SendError};
18
19const NO_AWAITER: u64 = u64::MAX;
20
21/// Fixed completion storage for at most `N` concurrently awaited commands.
22pub struct CompletionPool<M: RawMutex, const N: usize> {
23    next_id: AtomicU64,
24    awaited: BlockingMutex<M, RefCell<[u64; N]>>,
25    slots: [Signal<M, Settlement>; N],
26}
27
28impl<M: RawMutex, const N: usize> Default for CompletionPool<M, N> {
29    fn default() -> Self {
30        Self::new()
31    }
32}
33
34impl<M: RawMutex, const N: usize> CompletionPool<M, N> {
35    #[must_use]
36    pub const fn new() -> Self {
37        Self {
38            next_id: AtomicU64::new(0),
39            awaited: BlockingMutex::new(RefCell::new([NO_AWAITER; N])),
40            slots: [const { Signal::new() }; N],
41        }
42    }
43
44    fn mint(&self) -> CommandId {
45        loop {
46            let id = self.next_id.fetch_add(1, Ordering::Relaxed);
47            if id != NO_AWAITER {
48                return CommandId(id);
49            }
50        }
51    }
52
53    fn claim(&self, id: CommandId) -> Option<usize> {
54        self.awaited.lock(|cell| {
55            let mut awaited = cell.borrow_mut();
56            let slot = awaited.iter().position(|&a| a == NO_AWAITER)?;
57            self.slots[slot].reset();
58            awaited[slot] = id.0;
59            Some(slot)
60        })
61    }
62
63    fn release(&self, slot: usize, id: CommandId) {
64        self.awaited.lock(|cell| {
65            let mut awaited = cell.borrow_mut();
66            if awaited[slot] == id.0 {
67                awaited[slot] = NO_AWAITER;
68                self.slots[slot].reset();
69            }
70        });
71    }
72
73    fn settle(&self, id: CommandId, settlement: Settlement) -> bool {
74        self.awaited.lock(|cell| {
75            let mut awaited = cell.borrow_mut();
76            match awaited.iter().position(|&a| a == id.0) {
77                Some(slot) => {
78                    awaited[slot] = NO_AWAITER;
79                    self.slots[slot].signal(settlement);
80                    true
81                }
82                None => false,
83            }
84        })
85    }
86
87    async fn parked(&self, slot: usize) -> Settlement {
88        self.slots[slot].wait().await
89    }
90}
91
92pub struct PrnsNodeHandle<'a, M: RawMutex, const COMMANDS: usize, const N: usize> {
93    commands: Sender<'a, M, IssuedCommand, COMMANDS>,
94    pool: &'a CompletionPool<M, N>,
95}
96
97impl<M: RawMutex, const COMMANDS: usize, const N: usize> Clone
98    for PrnsNodeHandle<'_, M, COMMANDS, N>
99{
100    fn clone(&self) -> Self {
101        *self
102    }
103}
104
105impl<M: RawMutex, const COMMANDS: usize, const N: usize> Copy
106    for PrnsNodeHandle<'_, M, COMMANDS, N>
107{
108}
109
110impl<'a, M: RawMutex, const COMMANDS: usize, const N: usize> PrnsNodeHandle<'a, M, COMMANDS, N> {
111    #[must_use]
112    pub fn new(
113        commands: Sender<'a, M, IssuedCommand, COMMANDS>,
114        pool: &'a CompletionPool<M, N>,
115    ) -> Self {
116        Self { commands, pool }
117    }
118
119    /// Queues a command without awaiting settlement and returns its ID, or `None` when the command lane is full.
120    pub fn issue(&self, command: PrnsCommand) -> Option<CommandId> {
121        let id = self.pool.mint();
122        self.commands.try_send(IssuedCommand { id, command }).ok()?;
123        Some(id)
124    }
125
126    /// Sends one packet and awaits proof, returning `Busy` when all `N` completion slots are claimed.
127    pub async fn send_single_packet(
128        &self,
129        destination: DestinationHash,
130        data: &[u8],
131    ) -> Result<PacketReceiptDelivered, SendError<SendSinglePacketFailure>> {
132        let payload =
133            SendSinglePacketPayload::from_slice(data).map_err(|()| SendError::PayloadTooLarge)?;
134        let id = self.pool.mint();
135        let slot = self.pool.claim(id).ok_or(SendError::Busy)?;
136        let _guard = SlotGuard {
137            pool: self.pool,
138            slot,
139            id,
140        };
141        self.commands
142            .try_send(IssuedCommand {
143                id,
144                command: PrnsCommand::SendSinglePacket(SendSinglePacket {
145                    destination,
146                    payload,
147                }),
148            })
149            .map_err(|_| SendError::NodeStopped)?;
150        match self.pool.parked(slot).await {
151            Settlement::SendSinglePacket(result) => result.map_err(SendError::Failed),
152            _ => Err(SendError::NodeStopped),
153        }
154    }
155
156    /// Responds inline; returns `false` when the body exceeds the link MDU or the command lane is full.
157    pub fn respond_packed(&self, responder: RespondToken, packed: &[u8]) -> bool {
158        match RespondData::from_slice(packed) {
159            Ok(data) => self.respond_owned_packed(responder, data),
160            Err(_) => false,
161        }
162    }
163
164    /// Moves a prebuilt response into the command lane, returning `false` when full.
165    pub fn respond_owned_packed(&self, responder: RespondToken, data: RespondData) -> bool {
166        self.issue(PrnsCommand::Respond(Respond {
167            link_id: responder.link_id,
168            request_id: responder.request_id,
169            payload: RespondPayload::Packed(data),
170        }))
171        .is_some()
172    }
173
174    pub fn respond_static_bytes(&self, responder: RespondToken, data: &'static [u8]) -> bool {
175        self.issue(PrnsCommand::Respond(Respond {
176            link_id: responder.link_id,
177            request_id: responder.request_id,
178            payload: RespondPayload::StaticBytes(data),
179        }))
180        .is_some()
181    }
182
183    #[cfg(feature = "large-static-responses")]
184    pub fn respond_static_file(
185        &self,
186        responder: RespondToken,
187        name: &'static str,
188        bytes: &'static [u8],
189    ) -> bool {
190        self.issue(PrnsCommand::Respond(Respond {
191            link_id: responder.link_id,
192            request_id: responder.request_id,
193            payload: RespondPayload::StaticFile { name, bytes },
194        }))
195        .is_some()
196    }
197
198    /// Sever an active link. Returns `false` if the command lane is full.
199    pub fn close_link(&self, link_id: LinkId) -> bool {
200        self.issue(PrnsCommand::CloseLink(CloseLink { link_id }))
201            .is_some()
202    }
203
204    pub(super) fn settle(&self, id: CommandId, settlement: Settlement) -> bool {
205        self.pool.settle(id, settlement)
206    }
207}
208
209struct SlotGuard<'a, M: RawMutex, const N: usize> {
210    pool: &'a CompletionPool<M, N>,
211    slot: usize,
212    id: CommandId,
213}
214
215impl<M: RawMutex, const N: usize> Drop for SlotGuard<'_, M, N> {
216    fn drop(&mut self) {
217        self.pool.release(self.slot, self.id);
218    }
219}
220
221impl<M: RawMutex, const COMMANDS: usize, const N: usize> PrnsNodeApi
222    for PrnsNodeHandle<'_, M, COMMANDS, N>
223{
224    fn issue(&self, command: PrnsCommand) -> Option<CommandId> {
225        self.issue(command)
226    }
227
228    async fn send_single_packet(
229        &self,
230        destination: DestinationHash,
231        data: &[u8],
232    ) -> Result<PacketReceiptDelivered, SendError<SendSinglePacketFailure>> {
233        self.send_single_packet(destination, data).await
234    }
235
236    fn respond_packed(&self, responder: RespondToken, packed: &[u8]) -> bool {
237        self.respond_packed(responder, packed)
238    }
239
240    fn close_link(&self, link_id: LinkId) -> bool {
241        self.close_link(link_id)
242    }
243}
244
245#[cfg(test)]
246mod tests;
prns_runtime_embassy/runtime/node_facade/command_handle/mod.rs

prns_runtime_embassy/runtime/node_facade/command_handle/
mod.rs
