server/
blocking_archive.rs

1// SPDX-FileCopyrightText: 2025 Foundation Devices, Inc. <hello@foundation.xyz>
2// SPDX-License-Identifier: GPL-3.0-or-later
3
4//! archive message for server IPC
5
6use std::any::type_name;
7
8use rkyv::{
9    bytecheck::CheckBytes,
10    rancor::{self, Source as _},
11};
12use whence::WhenceExt;
13
14use crate::{AsyncMessageInit, Error, Server, ServerContext, WrongMessageTypeError};
15use crate::{SizeOfSerializer, XousDeserializer, XousSerializer, XousValidator};
16
17// ==================== core ====================
18
19/// heap allocated message that expects a response
20pub trait BlockingArchive
21where
22    Self: ArchiveCodec,
23    Self: crate::MessageId,
24    <Self::Response as rkyv::Archive>::Archived:
25        rkyv::Deserialize<Self::Response, XousDeserializer> + for<'a> CheckBytes<XousValidator<'a>>,
26{
27    /// response type for this message
28    type Response: ArchiveCodec;
29}
30
31/// serialization requirements for archive messages
32pub trait ArchiveCodec
33where
34    Self: Sized
35        + rkyv::Archive
36        + for<'a, 'b> rkyv::Serialize<XousSerializer<'a, 'b>>
37        + for<'a> rkyv::Serialize<SizeOfSerializer<'a>>,
38    Self::Archived: rkyv::Portable,
39{
40}
41
42impl<T> ArchiveCodec for T
43where
44    T: Sized
45        + rkyv::Archive
46        + for<'a, 'b> rkyv::Serialize<XousSerializer<'a, 'b>>
47        + for<'a> rkyv::Serialize<SizeOfSerializer<'a>>,
48    T::Archived: rkyv::Portable,
49{
50}
51
52// ==================== handler traits ====================
53
54/// handle archive messages synchronously
55pub trait BlockingArchiveHandler<M>
56where
57    M: BlockingArchive,
58    Self: Server,
59{
60    /// process message and return response immediately
61    fn handle(&mut self, msg: M, sender: xous::PID, context: &mut ServerContext<Self>) -> M::Response;
62}
63
64/// handle archive messages asynchronously (can defer response)
65pub trait BlockingArchiveAsyncHandler<M>
66where
67    M: BlockingArchive,
68    Self: Server,
69{
70    /// process message, response can be sent later
71    fn handle(&mut self, request: ArchiveRequest<M>, context: &mut ServerContext<Self>);
72    /// default response if handler drops without responding
73    fn default_response() -> M::Response;
74}
75
76// auto-convert sync handlers to async
77impl<T, M> BlockingArchiveAsyncHandler<M> for T
78where
79    M: BlockingArchive,
80    T: BlockingArchiveHandler<M>,
81{
82    fn handle(&mut self, request: ArchiveRequest<M>, context: &mut ServerContext<Self>) {
83        let ArchiveRequest { message, response: request } = request;
84        let response = <Self as BlockingArchiveHandler<M>>::handle(self, message, request.pid(), context);
85        if let Err(e) = request.respond(response) {
86            log::warn!("failed to respond archive {e:?}");
87        }
88    }
89
90    fn default_response() -> <M as BlockingArchive>::Response {
91        unreachable!("default value not required in sync handler")
92    }
93}
94
95/// handle async responses from other servers
96pub trait ArchiveResponseHandler<R>
97where
98    Self: Server,
99    R: ArchiveCodec,
100{
101    /// process received async response
102    fn handle_response(&mut self, response: R, sender: xous::PID, context: &mut ServerContext<Self>);
103}
104
105// ==================== types ====================
106
107/// archive request with deferred response capability
108#[derive(Debug)]
109pub struct ArchiveRequest<M: BlockingArchive> {
110    pub message: M,
111    pub response: ArchiveResponse<M::Response>,
112}
113
114/// deferred response that sends default on drop if not used
115#[derive(Debug)]
116pub struct ArchiveResponse<R: ArchiveCodec> {
117    responder: Option<Responder>,
118    pid: xous::PID,
119    default: fn() -> R,
120}
121
122impl<R: ArchiveCodec> ArchiveResponse<R> {
123    /// get sender's process ID
124    pub fn pid(&self) -> xous::PID { self.pid }
125
126    /// send response
127    pub fn respond(mut self, response: R) -> whence::Result<(), Error> {
128        let responder = self.responder.take().unwrap();
129        responder.respond(&response)
130    }
131
132    /// override default response function
133    /// will be sent on drop if [`Self::respond`] is not called
134    pub fn set_response(&mut self, f: fn() -> R) { self.default = f; }
135}
136
137// auto-send default response on drop
138impl<R: ArchiveCodec> Drop for ArchiveResponse<R> {
139    fn drop(&mut self) {
140        if let Some(responder) = self.responder.take() {
141            let default = (self.default)();
142            responder.respond(&default).ok();
143        }
144    }
145}
146
147// ==================== API ====================
148
149/// send archive message and block for response
150pub fn send_blocking_archive<M>(cid: xous::CID, msg: M) -> M::Response
151where
152    M: BlockingArchive,
153{
154    try_send_blocking_archive(cid, msg).unwrap()
155}
156
157/// send archive message, returns error instead of panic
158pub fn try_send_blocking_archive<M>(cid: xous::CID, msg: M) -> whence::Result<M::Response, Error>
159where
160    M: BlockingArchive,
161{
162    let mut buf = crate::Buffer::into_buf(&msg).whence()?;
163    buf.blocking_move(cid, M::ID as u32).whence()?;
164    buf.to_original().whence()
165}
166
167/// Message handler, used by ServerMessages::messages()
168pub fn handle_blocking_archive_message<M, S>(
169    handler: &mut S,
170    raw: xous::MessageEnvelope,
171    context: &mut ServerContext<S>,
172) where
173    M: BlockingArchive,
174    S: BlockingArchiveAsyncHandler<M>,
175    <M as rkyv::Archive>::Archived:
176        rkyv::Deserialize<M, XousDeserializer> + for<'a> CheckBytes<XousValidator<'a>>,
177{
178    let pid = raw.sender.pid().unwrap();
179    if let Err(e) = try_handle_blocking_archive_message(handler, raw, context) {
180        log::warn!("BlockingArchive handle error (PID {pid}) for {}: {e}", type_name::<M>());
181    }
182}
183
184fn try_handle_blocking_archive_message<M, S>(
185    handler: &mut S,
186    mut raw: xous::MessageEnvelope,
187    context: &mut ServerContext<S>,
188) -> whence::Result<(), Error>
189where
190    M: BlockingArchive,
191    S: BlockingArchiveAsyncHandler<M>,
192    <M as rkyv::Archive>::Archived:
193        rkyv::Deserialize<M, XousDeserializer> + for<'a> CheckBytes<XousValidator<'a>>,
194{
195    let pid = raw.sender.pid().unwrap();
196
197    match &mut raw.body {
198        xous::Message::BlockingMove(mem) => {
199            // sync case - extract message directly (no AsyncMessageInit wrapper)
200            let message: M = crate::Buffer::deserialize::<M>(mem).whence()?;
201            let request =
202                ArchiveResponse { responder: Some(Responder::Sync(raw)), pid, default: S::default_response };
203            let request = ArchiveRequest { message, response: request };
204            handler.handle(request, context);
205            Ok(())
206        }
207        xous::Message::Move(mem) => {
208            // async case - extract async wrapper
209            let init: AsyncMessageInit<M> = crate::Buffer::deserialize(mem).whence()?;
210            let request = ArchiveResponse {
211                responder: Some(Responder::Async { cid: init.cid, msg_id: init.msg_id }),
212                pid,
213                default: S::default_response,
214            };
215            let request = ArchiveRequest { message: init.msg, response: request };
216            handler.handle(request, context);
217            Ok(())
218        }
219        _ => Err(rancor::Error::new(WrongMessageTypeError)).whence(),
220    }
221}
222/// decode async response from raw envelope
223pub fn decode_archive_async_response<M>(raw: xous::MessageEnvelope) -> M
224where
225    M: ArchiveCodec,
226    <M as rkyv::Archive>::Archived:
227        rkyv::Deserialize<M, XousDeserializer> + for<'a> CheckBytes<XousValidator<'a>>,
228{
229    try_decode_archive_async_response(raw).unwrap()
230}
231
232pub fn try_decode_archive_async_response<M>(raw: xous::MessageEnvelope) -> whence::Result<M, Error>
233where
234    M: ArchiveCodec,
235    <M as rkyv::Archive>::Archived:
236        rkyv::Deserialize<M, XousDeserializer> + for<'a> CheckBytes<XousValidator<'a>>,
237{
238    crate::r#move::decode_move::<M>(&raw).whence()
239}
240
241// ==================== internal ====================
242
243#[derive(Debug)]
244enum Responder {
245    /// response goes back in same buffer (blocking call)
246    Sync(xous::MessageEnvelope),
247    /// response sent in new buffer (async call)
248    Async { cid: xous::CID, msg_id: xous::MessageId },
249}
250
251impl Responder {
252    fn respond<R>(self, response: &R) -> whence::Result<(), Error>
253    where
254        R: ArchiveCodec,
255    {
256        match self {
257            Responder::Sync(mut envelope) => {
258                let mem = envelope.body.memory_message_mut().expect("blocking move carries memory");
259                crate::Buffer::reply_move(mem, response).whence()
260            }
261            Responder::Async { cid, msg_id } => {
262                let _disconnect = defer::defer(|| {
263                    xous::disconnect(cid).ok();
264                });
265                crate::Buffer::into_buf(response).whence()?.send_nowait(cid, msg_id as u32).whence()?;
266                Ok(())
267            }
268        }
269    }
270}
server/blocking_archive.rs

server/
blocking_archive.rs
