sim_lib_server/
server.rs

1use std::{
2    sync::{
3        Arc, Mutex,
4        atomic::{AtomicU8, AtomicU64, Ordering},
5    },
6    time::Instant,
7};
8
9use sim_citizen_derive::non_citizen;
10use sim_kernel::{ClassRef, Cx, Expr, Object, Result, Symbol, Value};
11
12use crate::{
13    EvalSite, FrameRouter, IsolationPolicy, ServerAddress, ServerFrame, ServerRuntime,
14    TriggerHandle, symbol_list_value,
15};
16
17static NEXT_SERVER_ID: AtomicU64 = AtomicU64::new(1);
18
19/// Threading strategy a server uses to service connections.
20#[derive(Clone, Debug, PartialEq, Eq)]
21pub enum ThreadMode {
22    /// Run on the calling main thread.
23    Main,
24    /// Run cooperatively, yielding to the host scheduler.
25    Coop,
26    /// Spawn a dedicated thread per connection.
27    Spawn,
28    /// Service connections from a shared worker pool.
29    Pool,
30    /// Run the wrapped mode as a coroutine.
31    Coroutine(Box<ThreadMode>),
32}
33
34impl ThreadMode {
35    /// Parses a thread mode from an expression: a bare symbol (`main`, `coop`, `spawn`,
36    /// `pool`) or a `(coroutine <base>)` list.
37    pub fn from_expr(expr: &Expr) -> Result<Self> {
38        match expr {
39            Expr::Symbol(symbol) => match symbol.name.as_ref() {
40                "main" => Ok(Self::Main),
41                "coop" => Ok(Self::Coop),
42                "spawn" => Ok(Self::Spawn),
43                "pool" => Ok(Self::Pool),
44                other => Err(sim_kernel::Error::Eval(format!(
45                    "unsupported thread mode {other}"
46                ))),
47            },
48            Expr::List(items) | Expr::Vector(items) => {
49                let Some(Expr::Symbol(head)) = items.first() else {
50                    return Err(sim_kernel::Error::TypeMismatch {
51                        expected: "thread mode list starting with a symbol",
52                        found: "non-symbol",
53                    });
54                };
55                if head.name.as_ref() != "coroutine" {
56                    return Err(sim_kernel::Error::Eval(format!(
57                        "unsupported thread mode {}",
58                        head
59                    )));
60                }
61                let base = match items.get(1) {
62                    Some(expr) => Self::from_expr(expr)?,
63                    None => Self::Coop,
64                };
65                Ok(Self::Coroutine(Box::new(base)))
66            }
67            _ => Err(sim_kernel::Error::TypeMismatch {
68                expected: "thread mode expression",
69                found: "non-thread-mode",
70            }),
71        }
72    }
73
74    /// Renders this thread mode back to its expression form.
75    pub fn as_expr(&self) -> Expr {
76        match self {
77            Self::Main => Expr::Symbol(Symbol::new("main")),
78            Self::Coop => Expr::Symbol(Symbol::new("coop")),
79            Self::Spawn => Expr::Symbol(Symbol::new("spawn")),
80            Self::Pool => Expr::Symbol(Symbol::new("pool")),
81            Self::Coroutine(base) => {
82                Expr::List(vec![Expr::Symbol(Symbol::new("coroutine")), base.as_expr()])
83            }
84        }
85    }
86
87    /// Returns whether this thread mode can be used in the current environment.
88    pub fn is_available_now(&self) -> bool {
89        match self {
90            Self::Main | Self::Coop | Self::Spawn | Self::Pool => true,
91            Self::Coroutine(base) => matches!(base.as_ref(), Self::Main | Self::Coop),
92        }
93    }
94
95    /// Returns whether this mode is a coroutine variant.
96    pub fn is_coroutine(&self) -> bool {
97        matches!(self, Self::Coroutine(_))
98    }
99}
100
101/// Lifecycle status of a running server.
102#[derive(Clone, Copy, Debug, PartialEq, Eq)]
103pub enum ServerStatus {
104    /// Server is accepting and servicing connections.
105    Running,
106    /// Server is paused and not servicing connections.
107    Suspended,
108    /// Server has stopped.
109    Stopped,
110}
111
112impl ServerStatus {
113    fn as_u8(self) -> u8 {
114        match self {
115            Self::Running => 0,
116            Self::Suspended => 1,
117            Self::Stopped => 2,
118        }
119    }
120
121    fn from_u8(value: u8) -> Self {
122        match value {
123            0 => Self::Running,
124            1 => Self::Suspended,
125            2 => Self::Stopped,
126            _ => Self::Stopped,
127        }
128    }
129
130    fn as_symbol(self) -> Symbol {
131        Symbol::new(match self {
132            Self::Running => "running",
133            Self::Suspended => "suspended",
134            Self::Stopped => "stopped",
135        })
136    }
137}
138
139#[non_citizen(
140    reason = "live server handle; reconstruct configuration via server/Address descriptor and start ops",
141    kind = "handle",
142    descriptor = "server/Address"
143)]
144/// Live server handle: an address bound to an [`EvalSite`], its codec and threading
145/// configuration, lifecycle status, triggers, and optional [`ServerRuntime`].
146pub struct Server {
147    id: u64,
148    name: Option<Symbol>,
149    address: ServerAddress,
150    default_codec: Symbol,
151    supported_codecs: Vec<Symbol>,
152    thread: ThreadMode,
153    isolation: IsolationPolicy,
154    status: AtomicU8,
155    site: Arc<dyn EvalSite>,
156    spec: Vec<(Symbol, Expr)>,
157    router: Arc<FrameRouter>,
158    triggers: Arc<Mutex<Vec<Arc<TriggerHandle>>>>,
159    runtime: Option<Arc<ServerRuntime>>,
160    started_at: Instant,
161}
162
163impl Server {
164    /// Builds a server without an attached runtime.
165    #[allow(clippy::too_many_arguments)]
166    pub fn new(
167        address: ServerAddress,
168        default_codec: Symbol,
169        supported_codecs: Vec<Symbol>,
170        thread: ThreadMode,
171        isolation: IsolationPolicy,
172        name: Option<Symbol>,
173        site: Arc<dyn EvalSite>,
174        spec: Vec<(Symbol, Expr)>,
175    ) -> Result<Self> {
176        Self::with_runtime(
177            address,
178            default_codec,
179            supported_codecs,
180            thread,
181            isolation,
182            name,
183            site,
184            spec,
185            None,
186        )
187    }
188
189    /// Builds a server, optionally attaching a [`ServerRuntime`], after verifying the
190    /// address transport is available.
191    #[allow(clippy::too_many_arguments)]
192    pub fn with_runtime(
193        address: ServerAddress,
194        default_codec: Symbol,
195        supported_codecs: Vec<Symbol>,
196        thread: ThreadMode,
197        isolation: IsolationPolicy,
198        name: Option<Symbol>,
199        site: Arc<dyn EvalSite>,
200        spec: Vec<(Symbol, Expr)>,
201        runtime: Option<Arc<ServerRuntime>>,
202    ) -> Result<Self> {
203        address.ensure_transport_available()?;
204        Ok(Self {
205            id: NEXT_SERVER_ID.fetch_add(1, Ordering::Relaxed),
206            name,
207            address,
208            default_codec,
209            supported_codecs,
210            thread,
211            isolation,
212            status: AtomicU8::new(ServerStatus::Running.as_u8()),
213            site,
214            spec,
215            router: Arc::new(FrameRouter::default()),
216            triggers: Arc::new(Mutex::new(Vec::new())),
217            runtime,
218            started_at: Instant::now(),
219        })
220    }
221
222    /// Returns this server's unique id.
223    pub fn id(&self) -> u64 {
224        self.id
225    }
226
227    /// Returns the server's name, if one was assigned.
228    pub fn name(&self) -> Option<&Symbol> {
229        self.name.as_ref()
230    }
231
232    /// Returns the address the server is bound to.
233    pub fn address(&self) -> &ServerAddress {
234        &self.address
235    }
236
237    /// Returns the codec used by default for frames.
238    pub fn default_codec(&self) -> &Symbol {
239        &self.default_codec
240    }
241
242    /// Returns the codecs the server is willing to negotiate.
243    pub fn supported_codecs(&self) -> &[Symbol] {
244        &self.supported_codecs
245    }
246
247    /// Returns the server's threading mode.
248    pub fn thread(&self) -> &ThreadMode {
249        &self.thread
250    }
251
252    /// Returns the eval site that handles incoming frames.
253    pub fn site(&self) -> &Arc<dyn EvalSite> {
254        &self.site
255    }
256
257    /// Returns the isolation policy applied to sessions.
258    pub fn isolation(&self) -> &IsolationPolicy {
259        &self.isolation
260    }
261
262    /// Returns the configuration spec entries the server was started with.
263    pub fn spec(&self) -> &[(Symbol, Expr)] {
264        &self.spec
265    }
266
267    /// Returns the attached runtime, if the server is listening.
268    pub fn runtime(&self) -> Option<&Arc<ServerRuntime>> {
269        self.runtime.as_ref()
270    }
271
272    /// Returns the current lifecycle status.
273    pub fn status(&self) -> ServerStatus {
274        ServerStatus::from_u8(self.status.load(Ordering::Relaxed))
275    }
276
277    /// Sets the lifecycle status.
278    pub fn set_status(&self, status: ServerStatus) {
279        self.status.store(status.as_u8(), Ordering::Relaxed);
280    }
281
282    /// Returns the elapsed time since the server started, in milliseconds.
283    pub fn uptime_millis(&self) -> u64 {
284        self.started_at.elapsed().as_millis() as u64
285    }
286
287    /// Registers a trigger handle to be tracked and stopped with the server.
288    pub fn register_trigger(&self, trigger: Arc<TriggerHandle>) -> Result<()> {
289        self.triggers
290            .lock()
291            .map_err(|_| sim_kernel::Error::PoisonedLock("server triggers"))?
292            .push(trigger);
293        Ok(())
294    }
295
296    /// Stops every registered trigger.
297    pub fn stop_triggers(&self) -> Result<()> {
298        for trigger in self.trigger_snapshots()? {
299            trigger.stop()?;
300        }
301        Ok(())
302    }
303
304    /// Buffers `frame` as inbound and delivers it to the eval site, as if fired by a trigger.
305    pub fn deliver_trigger_frame(&self, cx: &mut Cx, frame: ServerFrame) -> Result<()> {
306        self.router.push_inbound(frame.clone())?;
307        let _ = self.site.answer(cx, frame)?;
308        Ok(())
309    }
310
311    /// Returns a snapshot of the currently registered trigger handles.
312    pub fn trigger_snapshots(&self) -> Result<Vec<Arc<TriggerHandle>>> {
313        Ok(self
314            .triggers
315            .lock()
316            .map_err(|_| sim_kernel::Error::PoisonedLock("server triggers"))?
317            .clone())
318    }
319
320    /// Returns a table value reflecting the server's configuration and live state.
321    pub fn reflect_value(&self, cx: &mut Cx) -> Result<Value> {
322        let mut entries = table_entries(self, cx)?;
323        entries.extend(live_state_entries(self, cx)?);
324        cx.factory().table(entries)
325    }
326
327    /// Returns a table value summarizing health: status, uptime, and session and message counts.
328    pub fn health_value(&self, cx: &mut Cx) -> Result<Value> {
329        let (sessions, connections, messages_sent, messages_received) = self
330            .runtime
331            .as_ref()
332            .map(|runtime| {
333                (
334                    runtime.session_count(),
335                    runtime.connection_count(),
336                    runtime.messages_sent(),
337                    runtime.messages_received(),
338                )
339            })
340            .unwrap_or((0, 0, 0, 0));
341        cx.factory().table(vec![
342            (
343                Symbol::new("status"),
344                cx.factory().symbol(self.status().as_symbol())?,
345            ),
346            (
347                Symbol::new("uptime"),
348                cx.factory().string(self.uptime_millis().to_string())?,
349            ),
350            (
351                Symbol::new("sessions"),
352                cx.factory().string(sessions.to_string())?,
353            ),
354            (
355                Symbol::new("connections"),
356                cx.factory().string(connections.to_string())?,
357            ),
358            (
359                Symbol::new("messages-sent"),
360                cx.factory().string(messages_sent.to_string())?,
361            ),
362            (
363                Symbol::new("messages-received"),
364                cx.factory().string(messages_received.to_string())?,
365            ),
366        ])
367    }
368
369    /// Returns a list value of the runtime's sessions, or an empty list if not listening.
370    pub fn sessions_value(&self, cx: &mut Cx) -> Result<Value> {
371        let Some(runtime) = &self.runtime else {
372            return cx.factory().list(Vec::new());
373        };
374        let sessions = runtime
375            .sessions()?
376            .into_iter()
377            .map(|session| session.as_value(cx))
378            .collect::<Result<Vec<_>>>()?;
379        cx.factory().list(sessions)
380    }
381}
382
383impl Object for Server {
384    fn display(&self, _cx: &mut Cx) -> Result<String> {
385        Ok("#<server>".to_owned())
386    }
387
388    fn as_any(&self) -> &dyn std::any::Any {
389        self
390    }
391}
392
393impl sim_kernel::ObjectCompat for Server {
394    fn class(&self, cx: &mut Cx) -> Result<ClassRef> {
395        cx.factory().class_stub(
396            sim_kernel::ClassId(0),
397            Symbol::qualified("server", "Server"),
398        )
399    }
400    fn as_expr(&self, cx: &mut Cx) -> Result<Expr> {
401        self.as_table(cx)?.object().as_expr(cx)
402    }
403    fn as_table(&self, cx: &mut Cx) -> Result<Value> {
404        let mut entries = table_entries(self, cx)?;
405        entries.extend(live_state_entries(self, cx)?);
406        cx.factory().table(entries)
407    }
408}
409
410impl Clone for Server {
411    fn clone(&self) -> Self {
412        Self {
413            id: self.id,
414            name: self.name.clone(),
415            address: self.address.clone(),
416            default_codec: self.default_codec.clone(),
417            supported_codecs: self.supported_codecs.clone(),
418            thread: self.thread.clone(),
419            isolation: self.isolation.clone(),
420            status: AtomicU8::new(self.status().as_u8()),
421            site: self.site.clone(),
422            spec: self.spec.clone(),
423            router: self.router.clone(),
424            triggers: self.triggers.clone(),
425            runtime: self.runtime.clone(),
426            started_at: self.started_at,
427        }
428    }
429}
430
431fn table_entries(server: &Server, cx: &mut Cx) -> Result<Vec<(Symbol, Value)>> {
432    let name = match server.name() {
433        Some(name) => cx.factory().symbol(name.clone())?,
434        None => cx.factory().nil()?,
435    };
436    let spec_entries = server
437        .spec()
438        .iter()
439        .map(|(key, value)| {
440            cx.factory()
441                .expr(Expr::List(vec![Expr::Symbol(key.clone()), value.clone()]))
442        })
443        .collect::<Result<Vec<_>>>()?;
444    let spec = cx.factory().list(spec_entries)?;
445    let address = server.address.as_value(cx)?;
446    let default_codec = cx.factory().symbol(server.default_codec.clone())?;
447    let supported_codecs = symbol_list_value(cx, &server.supported_codecs)?;
448    let thread = cx.factory().expr(server.thread.as_expr())?;
449    let site_kind = cx.factory().string(server.site.site_kind().to_owned())?;
450    let site_address = server.site.address().as_value(cx)?;
451    let site_codecs = symbol_list_value(cx, server.site.codecs())?;
452    let isolation = server.isolation.as_value(cx)?;
453    let listening = cx.factory().bool(server.runtime.is_some())?;
454    let next_msg_id = cx
455        .factory()
456        .string(server.router.peek_next_msg_id().to_string())?;
457    Ok(vec![
458        (
459            Symbol::new("kind"),
460            cx.factory().symbol(Symbol::new("server"))?,
461        ),
462        (
463            Symbol::new("id"),
464            cx.factory().string(server.id.to_string())?,
465        ),
466        (Symbol::new("name"), name),
467        (Symbol::new("address"), address),
468        (Symbol::new("default-codec"), default_codec),
469        (Symbol::new("supported-codecs"), supported_codecs),
470        (Symbol::new("thread"), thread),
471        (Symbol::new("site-kind"), site_kind),
472        (Symbol::new("site-address"), site_address),
473        (Symbol::new("site-codecs"), site_codecs),
474        (Symbol::new("isolation"), isolation),
475        (Symbol::new("listening"), listening),
476        (Symbol::new("spec"), spec),
477        (Symbol::new("next-msg-id"), next_msg_id),
478    ])
479}
480
481fn live_state_entries(server: &Server, cx: &mut Cx) -> Result<Vec<(Symbol, Value)>> {
482    let trigger_values = server
483        .trigger_snapshots()?
484        .into_iter()
485        .map(|trigger| trigger.reflect_value(cx))
486        .collect::<Result<Vec<_>>>()?;
487    let triggers = cx.factory().list(trigger_values)?;
488    let (sessions, connections, messages_sent, messages_received) = server
489        .runtime
490        .as_ref()
491        .map(|runtime| {
492            (
493                runtime.session_count(),
494                runtime.connection_count(),
495                runtime.messages_sent(),
496                runtime.messages_received(),
497            )
498        })
499        .unwrap_or((0, 0, 0, 0));
500    Ok(vec![
501        (
502            Symbol::new("status"),
503            cx.factory().symbol(server.status().as_symbol())?,
504        ),
505        (
506            Symbol::new("uptime"),
507            cx.factory().string(server.uptime_millis().to_string())?,
508        ),
509        (
510            Symbol::new("sessions"),
511            cx.factory().string(sessions.to_string())?,
512        ),
513        (
514            Symbol::new("connections"),
515            cx.factory().string(connections.to_string())?,
516        ),
517        (
518            Symbol::new("messages-sent"),
519            cx.factory().string(messages_sent.to_string())?,
520        ),
521        (
522            Symbol::new("messages-received"),
523            cx.factory().string(messages_received.to_string())?,
524        ),
525        (Symbol::new("triggers"), triggers),
526        (Symbol::new("line-driver"), cx.factory().nil()?),
527    ])
528}
sim_lib_server/server.rs

sim_lib_server/
server.rs
