sim_lib_server/
server.rs

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

sim_lib_server/
server.rs
