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    SystemWallClock, TriggerHandle, WallClock, 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    wall_clock: Arc<dyn WallClock>,
161    started_at: Instant,
162}
163
164impl Server {
165    /// Builds a server without an attached runtime.
166    #[allow(clippy::too_many_arguments)]
167    pub fn new(
168        address: ServerAddress,
169        default_codec: Symbol,
170        supported_codecs: Vec<Symbol>,
171        thread: ThreadMode,
172        isolation: IsolationPolicy,
173        name: Option<Symbol>,
174        site: Arc<dyn EvalSite>,
175        spec: Vec<(Symbol, Expr)>,
176    ) -> Result<Self> {
177        Self::with_runtime(
178            address,
179            default_codec,
180            supported_codecs,
181            thread,
182            isolation,
183            name,
184            site,
185            spec,
186            None,
187        )
188    }
189
190    /// Builds a server, optionally attaching a [`ServerRuntime`], after verifying the
191    /// address transport is available.
192    #[allow(clippy::too_many_arguments)]
193    pub fn with_runtime(
194        address: ServerAddress,
195        default_codec: Symbol,
196        supported_codecs: Vec<Symbol>,
197        thread: ThreadMode,
198        isolation: IsolationPolicy,
199        name: Option<Symbol>,
200        site: Arc<dyn EvalSite>,
201        spec: Vec<(Symbol, Expr)>,
202        runtime: Option<Arc<ServerRuntime>>,
203    ) -> Result<Self> {
204        address.ensure_transport_available()?;
205        Ok(Self {
206            id: NEXT_SERVER_ID.fetch_add(1, Ordering::Relaxed),
207            name,
208            address,
209            default_codec,
210            supported_codecs,
211            thread,
212            isolation,
213            status: AtomicU8::new(ServerStatus::Running.as_u8()),
214            site,
215            spec,
216            router: Arc::new(FrameRouter::default()),
217            triggers: Arc::new(Mutex::new(Vec::new())),
218            runtime,
219            wall_clock: Arc::new(SystemWallClock),
220            started_at: Instant::now(),
221        })
222    }
223
224    /// Replaces the host wall-clock observation source used by this server and its triggers.
225    ///
226    /// Configure the clock before sharing the server or registering triggers.
227    pub fn with_wall_clock(mut self, wall_clock: Arc<dyn WallClock>) -> Self {
228        self.wall_clock = wall_clock;
229        self
230    }
231
232    /// Returns this server's unique id.
233    pub fn id(&self) -> u64 {
234        self.id
235    }
236
237    /// Returns the server's name, if one was assigned.
238    pub fn name(&self) -> Option<&Symbol> {
239        self.name.as_ref()
240    }
241
242    /// Returns the address the server is bound to.
243    pub fn address(&self) -> &ServerAddress {
244        &self.address
245    }
246
247    /// Returns the codec used by default for frames.
248    pub fn default_codec(&self) -> &Symbol {
249        &self.default_codec
250    }
251
252    /// Returns the codecs the server is willing to negotiate.
253    pub fn supported_codecs(&self) -> &[Symbol] {
254        &self.supported_codecs
255    }
256
257    /// Returns the server's threading mode.
258    pub fn thread(&self) -> &ThreadMode {
259        &self.thread
260    }
261
262    /// Returns the eval site that handles incoming frames.
263    pub fn site(&self) -> &Arc<dyn EvalSite> {
264        &self.site
265    }
266
267    /// Returns the isolation policy applied to sessions.
268    pub fn isolation(&self) -> &IsolationPolicy {
269        &self.isolation
270    }
271
272    /// Returns the configuration spec entries the server was started with.
273    pub fn spec(&self) -> &[(Symbol, Expr)] {
274        &self.spec
275    }
276
277    /// Returns the attached runtime, if the server is listening.
278    pub fn runtime(&self) -> Option<&Arc<ServerRuntime>> {
279        self.runtime.as_ref()
280    }
281
282    /// Returns the injectable host wall-clock source used by this server.
283    pub fn wall_clock(&self) -> &Arc<dyn WallClock> {
284        &self.wall_clock
285    }
286
287    /// Returns the current lifecycle status.
288    pub fn status(&self) -> ServerStatus {
289        ServerStatus::from_u8(self.status.load(Ordering::Relaxed))
290    }
291
292    /// Sets the lifecycle status.
293    pub fn set_status(&self, status: ServerStatus) {
294        self.status.store(status.as_u8(), Ordering::Relaxed);
295    }
296
297    /// Returns the elapsed time since the server started, in milliseconds.
298    pub fn uptime_millis(&self) -> u64 {
299        self.started_at.elapsed().as_millis() as u64
300    }
301
302    /// Registers a trigger handle to be tracked and stopped with the server.
303    pub fn register_trigger(&self, trigger: Arc<TriggerHandle>) -> Result<()> {
304        self.triggers
305            .lock()
306            .map_err(|_| sim_kernel::Error::PoisonedLock("server triggers"))?
307            .push(trigger);
308        Ok(())
309    }
310
311    /// Stops every registered trigger.
312    pub fn stop_triggers(&self) -> Result<()> {
313        for trigger in self.trigger_snapshots()? {
314            trigger.stop()?;
315        }
316        Ok(())
317    }
318
319    /// Buffers `frame` as inbound and delivers it to the eval site, as if fired by a trigger.
320    pub fn deliver_trigger_frame(&self, cx: &mut Cx, frame: ServerFrame) -> Result<()> {
321        self.router.push_inbound(frame.clone())?;
322        let _ = self.site.answer(cx, frame)?;
323        Ok(())
324    }
325
326    /// Returns a snapshot of the currently registered trigger handles.
327    pub fn trigger_snapshots(&self) -> Result<Vec<Arc<TriggerHandle>>> {
328        Ok(self
329            .triggers
330            .lock()
331            .map_err(|_| sim_kernel::Error::PoisonedLock("server triggers"))?
332            .clone())
333    }
334
335    /// Returns a table value reflecting the server's configuration and live state.
336    pub fn reflect_value(&self, cx: &mut Cx) -> Result<Value> {
337        let mut entries = table_entries(self, cx)?;
338        entries.extend(live_state_entries(self, cx)?);
339        cx.factory().table(entries)
340    }
341
342    /// Returns a table value summarizing health: status, uptime, and session and message counts.
343    pub fn health_value(&self, cx: &mut Cx) -> Result<Value> {
344        let (sessions, connections, messages_sent, messages_received) = self
345            .runtime
346            .as_ref()
347            .map(|runtime| {
348                (
349                    runtime.session_count(),
350                    runtime.connection_count(),
351                    runtime.messages_sent(),
352                    runtime.messages_received(),
353                )
354            })
355            .unwrap_or((0, 0, 0, 0));
356        cx.factory().table(vec![
357            (
358                Symbol::new("status"),
359                cx.factory().symbol(self.status().as_symbol())?,
360            ),
361            (
362                Symbol::new("uptime"),
363                cx.factory().string(self.uptime_millis().to_string())?,
364            ),
365            (
366                Symbol::new("sessions"),
367                cx.factory().string(sessions.to_string())?,
368            ),
369            (
370                Symbol::new("connections"),
371                cx.factory().string(connections.to_string())?,
372            ),
373            (
374                Symbol::new("messages-sent"),
375                cx.factory().string(messages_sent.to_string())?,
376            ),
377            (
378                Symbol::new("messages-received"),
379                cx.factory().string(messages_received.to_string())?,
380            ),
381        ])
382    }
383
384    /// Returns a list value of the runtime's sessions, or an empty list if not listening.
385    pub fn sessions_value(&self, cx: &mut Cx) -> Result<Value> {
386        let Some(runtime) = &self.runtime else {
387            return cx.factory().list(Vec::new());
388        };
389        let sessions = runtime
390            .sessions()?
391            .into_iter()
392            .map(|session| session.as_value(cx))
393            .collect::<Result<Vec<_>>>()?;
394        cx.factory().list(sessions)
395    }
396}
397
398impl Object for Server {
399    fn display(&self, _cx: &mut Cx) -> Result<String> {
400        Ok("#<server>".to_owned())
401    }
402
403    fn as_any(&self) -> &dyn std::any::Any {
404        self
405    }
406}
407
408impl sim_kernel::ObjectCompat for Server {
409    fn class(&self, cx: &mut Cx) -> Result<ClassRef> {
410        cx.factory().class_stub(
411            sim_kernel::ClassId(0),
412            Symbol::qualified("server", "Server"),
413        )
414    }
415    fn as_expr(&self, cx: &mut Cx) -> Result<Expr> {
416        self.as_table(cx)?.object().as_expr(cx)
417    }
418    fn as_table(&self, cx: &mut Cx) -> Result<Value> {
419        let mut entries = table_entries(self, cx)?;
420        entries.extend(live_state_entries(self, cx)?);
421        cx.factory().table(entries)
422    }
423}
424
425impl Clone for Server {
426    fn clone(&self) -> Self {
427        Self {
428            id: self.id,
429            name: self.name.clone(),
430            address: self.address.clone(),
431            default_codec: self.default_codec.clone(),
432            supported_codecs: self.supported_codecs.clone(),
433            thread: self.thread.clone(),
434            isolation: self.isolation.clone(),
435            status: AtomicU8::new(self.status().as_u8()),
436            site: self.site.clone(),
437            spec: self.spec.clone(),
438            router: self.router.clone(),
439            triggers: self.triggers.clone(),
440            runtime: self.runtime.clone(),
441            wall_clock: self.wall_clock.clone(),
442            started_at: self.started_at,
443        }
444    }
445}
446
447fn table_entries(server: &Server, cx: &mut Cx) -> Result<Vec<(Symbol, Value)>> {
448    let name = match server.name() {
449        Some(name) => cx.factory().symbol(name.clone())?,
450        None => cx.factory().nil()?,
451    };
452    let spec_entries = server
453        .spec()
454        .iter()
455        .map(|(key, value)| {
456            cx.factory()
457                .expr(Expr::List(vec![Expr::Symbol(key.clone()), value.clone()]))
458        })
459        .collect::<Result<Vec<_>>>()?;
460    let spec = cx.factory().list(spec_entries)?;
461    let address = server.address.as_value(cx)?;
462    let default_codec = cx.factory().symbol(server.default_codec.clone())?;
463    let supported_codecs = symbol_list_value(cx, &server.supported_codecs)?;
464    let thread = cx.factory().expr(server.thread.as_expr())?;
465    let site_kind = cx.factory().string(server.site.site_kind().to_owned())?;
466    let site_address = server.site.address().as_value(cx)?;
467    let site_codecs = symbol_list_value(cx, server.site.codecs())?;
468    let isolation = server.isolation.as_value(cx)?;
469    let listening = cx.factory().bool(server.runtime.is_some())?;
470    let next_msg_id = cx
471        .factory()
472        .string(server.router.peek_next_msg_id().to_string())?;
473    Ok(vec![
474        (
475            Symbol::new("kind"),
476            cx.factory().symbol(Symbol::new("server"))?,
477        ),
478        (
479            Symbol::new("id"),
480            cx.factory().string(server.id.to_string())?,
481        ),
482        (Symbol::new("name"), name),
483        (Symbol::new("address"), address),
484        (Symbol::new("default-codec"), default_codec),
485        (Symbol::new("supported-codecs"), supported_codecs),
486        (Symbol::new("thread"), thread),
487        (Symbol::new("site-kind"), site_kind),
488        (Symbol::new("site-address"), site_address),
489        (Symbol::new("site-codecs"), site_codecs),
490        (Symbol::new("isolation"), isolation),
491        (Symbol::new("listening"), listening),
492        (Symbol::new("spec"), spec),
493        (Symbol::new("next-msg-id"), next_msg_id),
494    ])
495}
496
497fn live_state_entries(server: &Server, cx: &mut Cx) -> Result<Vec<(Symbol, Value)>> {
498    let trigger_values = server
499        .trigger_snapshots()?
500        .into_iter()
501        .map(|trigger| trigger.reflect_value(cx))
502        .collect::<Result<Vec<_>>>()?;
503    let triggers = cx.factory().list(trigger_values)?;
504    let (sessions, connections, messages_sent, messages_received) = server
505        .runtime
506        .as_ref()
507        .map(|runtime| {
508            (
509                runtime.session_count(),
510                runtime.connection_count(),
511                runtime.messages_sent(),
512                runtime.messages_received(),
513            )
514        })
515        .unwrap_or((0, 0, 0, 0));
516    Ok(vec![
517        (
518            Symbol::new("status"),
519            cx.factory().symbol(server.status().as_symbol())?,
520        ),
521        (
522            Symbol::new("uptime"),
523            cx.factory().string(server.uptime_millis().to_string())?,
524        ),
525        (
526            Symbol::new("sessions"),
527            cx.factory().string(sessions.to_string())?,
528        ),
529        (
530            Symbol::new("connections"),
531            cx.factory().string(connections.to_string())?,
532        ),
533        (
534            Symbol::new("messages-sent"),
535            cx.factory().string(messages_sent.to_string())?,
536        ),
537        (
538            Symbol::new("messages-received"),
539            cx.factory().string(messages_received.to_string())?,
540        ),
541        (Symbol::new("triggers"), triggers),
542        (Symbol::new("line-driver"), cx.factory().nil()?),
543    ])
544}
sim_lib_server/server.rs

sim_lib_server/
server.rs
