sim_lib_server/site/
single.rs

1use std::sync::Arc;
2
3use sim_kernel::{Cx, Error, EvalFabric, EvalReply, ReadPolicy, Result, Symbol};
4use sim_lib_stream_core::{ClockDomain, LatencyClass, StreamEndpoint, StreamEndpointKind};
5
6use crate::{
7    Coroutine, FrameKind, ServerAddress, ServerFrame, decode_frame_payload,
8    eval_request_from_frame, server_frame_from_reply,
9};
10
11use super::core::{
12    EvalSite, Site, SiteKind, eval_site_clock_domain, eval_site_endpoint_kind,
13    eval_site_latency_class, reply_codec_for_frame, site_endpoint_id,
14};
15use super::pipeline::{enforce_trigger_eval_policy, eval_request_from_trigger_frame};
16
17/// An eval site that evaluates requests directly in the local runtime.
18#[derive(Clone)]
19pub struct LocalEvalSite {
20    address: ServerAddress,
21    codecs: Vec<Symbol>,
22}
23
24impl LocalEvalSite {
25    /// Creates a local site answering at `address` over `codecs`.
26    pub fn new(address: ServerAddress, codecs: Vec<Symbol>) -> Self {
27        Self { address, codecs }
28    }
29}
30
31impl EvalSite for LocalEvalSite {
32    fn site_kind(&self) -> &'static str {
33        "local"
34    }
35
36    fn address(&self) -> &ServerAddress {
37        &self.address
38    }
39
40    fn codecs(&self) -> &[Symbol] {
41        &self.codecs
42    }
43
44    fn answer(&self, cx: &mut Cx, frame: ServerFrame) -> Result<ServerFrame> {
45        match frame.kind {
46            FrameKind::Request => {
47                let consistency = frame.envelope.consistency;
48                let reply_codec = reply_codec_for_frame(self, &frame);
49                let request = eval_request_from_frame(cx, &frame)?;
50                let reply = realize_locally(cx, request)?;
51                server_frame_from_reply(cx, &reply_codec, reply, consistency)
52            }
53            FrameKind::Trigger { .. } => {
54                realize_trigger_locally(cx, &frame)?;
55                Ok(frame)
56            }
57            FrameKind::Notify => {
58                realize_notify_locally(cx, &frame)?;
59                Ok(frame)
60            }
61            _ => Err(Error::Eval(format!(
62                "local eval site cannot answer frame kind {}",
63                frame.kind.as_symbol()
64            ))),
65        }
66    }
67
68    fn as_any(&self) -> &dyn std::any::Any {
69        self
70    }
71}
72
73impl StreamEndpoint for LocalEvalSite {
74    fn endpoint_id(&self) -> Symbol {
75        site_endpoint_id(SiteKind::Local, &self.address)
76    }
77
78    fn endpoint_kind(&self) -> StreamEndpointKind {
79        eval_site_endpoint_kind()
80    }
81
82    fn clock_domain(&self) -> ClockDomain {
83        eval_site_clock_domain()
84    }
85
86    fn latency_class(&self) -> LatencyClass {
87        eval_site_latency_class()
88    }
89}
90
91impl Site for LocalEvalSite {
92    fn kind(&self) -> SiteKind {
93        SiteKind::Local
94    }
95}
96
97/// An eval site that delegates requests to a distributed [`EvalFabric`].
98#[derive(Clone)]
99pub struct FabricEvalSite {
100    kind: &'static str,
101    address: ServerAddress,
102    codecs: Vec<Symbol>,
103    fabric: Arc<dyn EvalFabric>,
104}
105
106impl FabricEvalSite {
107    /// Creates a fabric site labeled `kind`, answering at `address` over
108    /// `codecs`, backed by `fabric`.
109    pub fn new(
110        kind: &'static str,
111        address: ServerAddress,
112        codecs: Vec<Symbol>,
113        fabric: Arc<dyn EvalFabric>,
114    ) -> Self {
115        Self {
116            kind,
117            address,
118            codecs,
119            fabric,
120        }
121    }
122}
123
124impl EvalSite for FabricEvalSite {
125    fn site_kind(&self) -> &'static str {
126        self.kind
127    }
128
129    fn address(&self) -> &ServerAddress {
130        &self.address
131    }
132
133    fn codecs(&self) -> &[Symbol] {
134        &self.codecs
135    }
136
137    fn answer(&self, cx: &mut Cx, frame: ServerFrame) -> Result<ServerFrame> {
138        match frame.kind {
139            FrameKind::Request => {
140                let consistency = frame.envelope.consistency;
141                let reply_codec = reply_codec_for_frame(self, &frame);
142                let request = eval_request_from_frame(cx, &frame)?;
143                let reply = self.fabric.realize(cx, request)?;
144                server_frame_from_reply(cx, &reply_codec, reply, consistency)
145            }
146            FrameKind::Trigger { .. } => {
147                let request = eval_request_from_trigger_frame(cx, &frame)?;
148                let _ = self.fabric.realize(cx, request)?;
149                Ok(frame)
150            }
151            FrameKind::StreamStart | FrameKind::StreamChunk | FrameKind::StreamEnd => Ok(frame),
152            _ => Err(Error::Eval(format!(
153                "fabric eval site cannot answer frame kind {}",
154                frame.kind.as_symbol()
155            ))),
156        }
157    }
158
159    fn as_eval_fabric(&self) -> Option<&dyn EvalFabric> {
160        Some(self.fabric.as_ref())
161    }
162
163    fn as_any(&self) -> &dyn std::any::Any {
164        self
165    }
166}
167
168impl StreamEndpoint for FabricEvalSite {
169    fn endpoint_id(&self) -> Symbol {
170        site_endpoint_id(SiteKind::Fabric, &self.address)
171    }
172
173    fn endpoint_kind(&self) -> StreamEndpointKind {
174        eval_site_endpoint_kind()
175    }
176
177    fn clock_domain(&self) -> ClockDomain {
178        eval_site_clock_domain()
179    }
180
181    fn latency_class(&self) -> LatencyClass {
182        eval_site_latency_class()
183    }
184}
185
186impl Site for FabricEvalSite {
187    fn kind(&self) -> SiteKind {
188        SiteKind::Fabric
189    }
190}
191
192/// An eval site that resumes a [`Coroutine`] with each request's expression.
193#[derive(Clone)]
194pub struct CoroutineEvalSite {
195    address: ServerAddress,
196    codecs: Vec<Symbol>,
197    coroutine: Arc<Coroutine>,
198}
199
200impl CoroutineEvalSite {
201    /// Creates a coroutine site answering at `address` over `codecs`, resuming
202    /// `coroutine`.
203    pub fn new(address: ServerAddress, codecs: Vec<Symbol>, coroutine: Arc<Coroutine>) -> Self {
204        Self {
205            address,
206            codecs,
207            coroutine,
208        }
209    }
210
211    /// Returns the coroutine this site resumes.
212    pub fn coroutine(&self) -> &Arc<Coroutine> {
213        &self.coroutine
214    }
215}
216
217impl EvalSite for CoroutineEvalSite {
218    fn site_kind(&self) -> &'static str {
219        "coroutine"
220    }
221
222    fn address(&self) -> &ServerAddress {
223        &self.address
224    }
225
226    fn codecs(&self) -> &[Symbol] {
227        &self.codecs
228    }
229
230    fn answer(&self, cx: &mut Cx, frame: ServerFrame) -> Result<ServerFrame> {
231        match frame.kind {
232            FrameKind::Request => {
233                let consistency = frame.envelope.consistency;
234                let reply_codec = reply_codec_for_frame(self, &frame);
235                let request = eval_request_from_frame(cx, &frame)?;
236                let input = cx.factory().expr(request.expr)?;
237                let reply = self.coroutine.resume(cx, input)?;
238                let diagnostics = cx.take_diagnostics();
239                server_frame_from_reply(
240                    cx,
241                    &reply_codec,
242                    EvalReply {
243                        value: reply,
244                        diagnostics,
245                        trace: None,
246                    },
247                    consistency,
248                )
249            }
250            FrameKind::Trigger { .. } => {
251                let expr = frame.decode_expr(cx, ReadPolicy::default())?;
252                enforce_trigger_eval_policy(cx, &expr)?;
253                let input = cx.factory().expr(expr)?;
254                let _ = self.coroutine.resume(cx, input)?;
255                Ok(frame)
256            }
257            FrameKind::Notify => {
258                let expr = decode_frame_payload(
259                    cx,
260                    &frame.codec,
261                    &frame.payload,
262                    ReadPolicy::default(),
263                    Default::default(),
264                )?;
265                let input = cx.factory().expr(expr)?;
266                let _ = self.coroutine.resume(cx, input)?;
267                Ok(frame)
268            }
269            _ => Err(Error::Eval(format!(
270                "coroutine eval site cannot answer frame kind {}",
271                frame.kind.as_symbol()
272            ))),
273        }
274    }
275
276    fn as_any(&self) -> &dyn std::any::Any {
277        self
278    }
279}
280
281impl StreamEndpoint for CoroutineEvalSite {
282    fn endpoint_id(&self) -> Symbol {
283        site_endpoint_id(SiteKind::Coroutine, &self.address)
284    }
285
286    fn endpoint_kind(&self) -> StreamEndpointKind {
287        eval_site_endpoint_kind()
288    }
289
290    fn clock_domain(&self) -> ClockDomain {
291        eval_site_clock_domain()
292    }
293
294    fn latency_class(&self) -> LatencyClass {
295        eval_site_latency_class()
296    }
297}
298
299impl Site for CoroutineEvalSite {
300    fn kind(&self) -> SiteKind {
301        SiteKind::Coroutine
302    }
303}
304
305fn realize_locally(cx: &mut Cx, request: sim_kernel::EvalRequest) -> Result<EvalReply> {
306    for capability in &request.required_capabilities {
307        cx.require(capability)?;
308    }
309    let value = cx.eval_expr(request.expr)?;
310    if let Some(shape) = &request.result_shape {
311        let Some(shape_object) = shape.object().as_shape() else {
312            return Err(Error::TypeMismatch {
313                expected: "shape",
314                found: "non-shape",
315            });
316        };
317        let matched = shape_object.check_value(cx, value.clone())?;
318        if !matched.accepted {
319            return Err(Error::WrongShape {
320                expected: shape_object.id().unwrap_or(sim_kernel::ShapeId(0)),
321                diagnostics: matched.diagnostics,
322            });
323        }
324    }
325    Ok(EvalReply {
326        value,
327        diagnostics: cx.take_diagnostics(),
328        trace: request
329            .trace
330            .then(|| cx.factory().symbol(Symbol::new("local")).ok())
331            .flatten(),
332    })
333}
334
335fn realize_notify_locally(cx: &mut Cx, frame: &ServerFrame) -> Result<()> {
336    for capability in &frame.envelope.required_capabilities {
337        cx.require(capability)?;
338    }
339    let expr = decode_frame_payload(
340        cx,
341        &frame.codec,
342        &frame.payload,
343        ReadPolicy::default(),
344        Default::default(),
345    )?;
346    cx.eval_expr(expr)?;
347    Ok(())
348}
349
350fn realize_trigger_locally(cx: &mut Cx, frame: &ServerFrame) -> Result<()> {
351    for capability in &frame.envelope.required_capabilities {
352        cx.require(capability)?;
353    }
354    let expr = frame.decode_expr(cx, ReadPolicy::default())?;
355    enforce_trigger_eval_policy(cx, &expr)?;
356    cx.eval_expr(expr)?;
357    Ok(())
358}
sim_lib_server/site/single.rs

sim_lib_server/site/
single.rs
