1use std::collections::{BTreeMap, BTreeSet, VecDeque};
65use std::path::PathBuf;
66use std::sync::{mpsc, Arc, Condvar, LazyLock, Mutex, MutexGuard, OnceLock, Weak};
67use std::time::{Duration, Instant};
68
69use tatara_lisp::{Span, Spanned};
70use tatara_lisp_eval::ffi::Arity;
71use tatara_lisp_eval::vm::run::CompiledClosure;
72use tatara_lisp_eval::vm::{
73 compile_program, Budget, Chunk, FnRef, ParkError, Vm, DEFAULT_MAX_DEPTH,
74};
75use tatara_lisp_eval::{EvalError, Interpreter, MapKey, Value};
76
77use blue_lang_waku::Capability;
78
79use crate::code::{self, Code, Entry};
80use crate::image::{self, Account, Arrivals, Continuation, Image, Supervision};
81use crate::mailbox::copy_data;
82use crate::pattern;
83use crate::receipt::{
84 CarriedIo, Frame, HostIo, Identities, Places, Receipt, ReceiptSink, Roots, StateRoot, Subject,
85};
86use crate::scheduling::{self, IdleLane, Scheduling};
87use crate::trace::{Carried, CarriedAccount, Journal, SliceRecord, TapeEntry, Trace};
88use crate::violation::{breach, defined_key, Violation};
89use crate::{provided_name, raised, ExitReason, Pid, Proc, ProcState, Slice, Strategy};
90
91pub const ROOT: Pid = Pid(0);
93
94pub const DEFAULT_QUANTUM: usize = 1_000;
97
98pub const DEFAULT_MAX_TAPE: usize = 100_000;
102
103pub const DEFAULT_JOURNAL: usize = 8;
106
107const SCHEDULE_RING: usize = 1_024;
109
110const KEPT_TRACES: usize = 16;
112
113const MIGRATION_FUEL: usize = 10_000_000;
115
116pub const DEFAULT_MAX_RESTARTS: usize = 3;
119
120pub const PROCESS_WORDS: &[&str] = &[
122 "spawn",
123 "spawn_link",
124 "spawn_server",
125 "send",
126 "receive",
127 "self",
128 "alive?",
129 "link",
130 "unlink",
131 "monitor",
132 "demonitor",
133 "exit",
134 "trap_exit",
135 "supervisor",
136 "children",
137 "defmigrate",
138 "def_hash",
139];
140
141pub const CARRIED_WORDS: &[&str] = &["evict", "accept"];
142
143pub const EVICTED: &str = "evicted";
144
145pub const SERVER_NEXT: &str = "%server_next";
148
149static SERVER_LOOP: LazyLock<Vec<Spanned>> = LazyLock::new(|| {
156 tatara_lisp::read_spanned(
157 "(define (%serve) (%serve_step (%server_next %state)))
158 (define (%serve_step m) (set! %state (%handler m %state)) (%serve))",
159 )
160 .expect("the server loop reads")
161});
162
163static TRACE_DIR: OnceLock<Option<PathBuf>> = OnceLock::new();
164
165pub fn set_trace_dir(dir: Option<PathBuf>) {
169 let _ = TRACE_DIR.set(dir);
170}
171
172static JOURNAL: OnceLock<usize> = OnceLock::new();
173
174pub fn set_journal_bound(bound: usize) {
178 let _ = JOURNAL.set(bound);
179}
180
181#[derive(Clone, Copy, Debug, PartialEq, Eq)]
183pub enum ClockKind {
184 Real,
187 Virtual,
191}
192
193impl ClockKind {
194 #[must_use]
195 pub fn label(self) -> &'static str {
196 match self {
197 ClockKind::Real => "real",
198 ClockKind::Virtual => "virtual",
199 }
200 }
201}
202
203struct Clock {
204 kind: ClockKind,
205 origin: Option<Instant>,
206 virtual_ms: u64,
207}
208
209impl Clock {
210 fn new(kind: ClockKind) -> Self {
211 Self {
212 kind,
213 origin: None,
214 virtual_ms: 0,
215 }
216 }
217
218 fn now(&mut self) -> u64 {
219 match self.kind {
220 ClockKind::Real => u64::try_from(
221 self.origin
222 .get_or_insert_with(Instant::now)
223 .elapsed()
224 .as_millis(),
225 )
226 .unwrap_or(u64::MAX),
227 ClockKind::Virtual => self.virtual_ms,
228 }
229 }
230
231 fn advance_to(&mut self, t: u64) -> Option<Duration> {
233 match self.kind {
234 ClockKind::Real => {
235 let now = self.now();
236 (t > now).then(|| Duration::from_millis(t - now))
237 }
238 ClockKind::Virtual => {
239 self.virtual_ms = self.virtual_ms.max(t);
240 None
241 }
242 }
243 }
244}
245
246pub type ChildFactory = Arc<dyn Fn() -> Interpreter<()> + Send + Sync>;
248
249#[derive(Clone, Copy, Debug)]
258pub struct Meter {
259 pub bound: fn() -> Option<u64>,
262 pub charged: fn(&mut i64, Option<u64>, &mut dyn FnMut()) -> bool,
265 pub charge: fn(i64),
267}
268
269#[derive(Clone, Copy, Debug)]
270pub struct Carrier {
271 pub room: usize,
272 pub queue: fn(&Value, &Value, &[String], bool) -> Result<usize, String>,
273 pub take: fn(&Value, &str, Duration, bool) -> Result<Vec<String>, String>,
274}
275
276#[derive(Clone)]
278pub struct NodeConfig {
279 pub budget: Budget,
282 pub clock: ClockKind,
283 pub max_tape: usize,
284 pub trace_dir: Option<PathBuf>,
286 pub child: ChildFactory,
287 pub frame: Option<Frame>,
289 pub journal: usize,
292 pub host_io: HostIo,
294 pub identities: Identities,
296 pub places: Places,
298 pub receipts: Option<ReceiptSink>,
300 pub state_root: Option<StateRoot>,
303 pub meter: Option<Meter>,
306 pub carrier: Option<Carrier>,
307 pub dispatch_steps: Option<usize>,
308 pub count_steps: bool,
309 pub idle_lane: IdleLane,
310}
311
312impl NodeConfig {
313 #[must_use]
317 pub fn new(child: ChildFactory) -> Self {
318 Self {
319 budget: Budget {
320 fuel: None,
321 max_depth: Some(DEFAULT_MAX_DEPTH),
322 quantum: Some(DEFAULT_QUANTUM),
323 },
324 clock: ClockKind::Virtual,
325 max_tape: DEFAULT_MAX_TAPE,
326 trace_dir: TRACE_DIR.get().cloned().flatten(),
327 child,
328 frame: None,
329 journal: JOURNAL.get().copied().unwrap_or(DEFAULT_JOURNAL),
330 host_io: HostIo::default(),
331 identities: Identities::default(),
332 places: Places::default(),
333 receipts: None,
334 state_root: None,
335 meter: None,
336 carrier: None,
337 dispatch_steps: None,
338 count_steps: false,
339 idle_lane: IdleLane::default(),
340 }
341 }
342
343 fn journals(&self) -> bool {
344 self.frame.is_some() && self.journal > 0
345 }
346
347 fn carrying(&self) -> Carrying {
348 let cap = Capability::CommScavenger;
349 Carrying {
350 carrier: self.carrier.is_some(),
351 scoped: self.frame.as_ref().is_some_and(|f| {
352 f.provided
353 .iter()
354 .any(|p| p.scope.is_some() && p.provision.capability() == cap)
355 }),
356 withheld: self
357 .frame
358 .as_ref()
359 .is_some_and(|f| f.withheld.contains(&cap)),
360 }
361 }
362}
363
364#[derive(Clone, Copy)]
365struct Carrying {
366 carrier: bool,
367 scoped: bool,
368 withheld: bool,
369}
370
371#[derive(Clone, Debug)]
372struct Sup {
373 strategy: Strategy,
374 max_restarts: usize,
375 children: Vec<Pid>,
376}
377
378#[derive(Clone, Debug)]
379enum Kind {
380 Root,
381 Plain,
382 Server,
383 Supervisor(Sup),
384}
385
386struct Info {
388 kind: Kind,
389 code: Code,
391 chunk: Option<Arc<Chunk>>,
392 def_name: Option<String>,
394 links: BTreeSet<Pid>,
395 monitors: Vec<(i64, Pid)>,
397 trap_exit: bool,
398 sup: Option<Pid>,
399 restarts: usize,
400 live: bool,
403 snapshot: Option<Value>,
405 initial: Option<Value>,
407 entered: Option<Value>,
409 held: i64,
411 bound: Option<u64>,
413 scheduling: Scheduling,
414 dispatch: Option<usize>,
415 dispatch_from: Option<usize>,
416 armed: bool,
417 kept: Option<Value>,
418 asked: Option<usize>,
419}
420
421impl Info {
422 fn new(kind: Kind) -> Self {
423 Self {
424 kind,
425 code: Code::default(),
426 chunk: None,
427 def_name: None,
428 links: BTreeSet::new(),
429 monitors: Vec::new(),
430 trap_exit: false,
431 sup: None,
432 restarts: 0,
433 live: true,
434 snapshot: None,
435 initial: None,
436 entered: None,
437 held: 0,
438 bound: None,
439 scheduling: Scheduling::Normal,
440 dispatch: None,
441 dispatch_from: None,
442 armed: false,
443 kept: None,
444 asked: None,
445 }
446 }
447
448 fn ask_dispatch(&mut self, own: Option<usize>) {
449 self.asked = own;
450 self.dispatch = match (self.dispatch, own) {
451 (Some(app), Some(own)) => Some(app.min(own)),
452 (app, own) => app.or(own),
453 };
454 }
455}
456
457#[derive(Clone, Debug)]
459enum Wait {
460 Receive {
461 pattern: Value,
462 deadline: Option<u64>,
463 },
464 ServerNext,
466}
467
468#[derive(Clone)]
469struct Migration {
470 request: u64,
471 code: Code,
472 new_name: String,
473 transform: Option<Code>,
474}
475
476#[derive(Default, Clone)]
477struct Tape {
478 entries: Vec<TapeEntry>,
479 truncated: bool,
480 carried: Option<Carried>,
481}
482
483impl Tape {
484 fn arrived(carried: Option<Vec<TapeEntry>>, image: &str) -> Self {
485 match carried {
486 Some(entries) => Self {
487 carried: Some(Carried::of(&entries, image)),
488 entries,
489 truncated: false,
490 },
491 None => Self {
492 entries: Vec::new(),
493 truncated: true,
494 carried: None,
495 },
496 }
497 }
498}
499
500#[derive(Clone, Debug)]
501struct Brought {
502 mark: CarriedAccount,
503 input_root: Option<String>,
504}
505
506struct NodeState {
507 current: Pid,
508 next_pid: u64,
509 next_ref: i64,
510 next_request: u64,
511 boxes: BTreeMap<Pid, VecDeque<Value>>,
512 info: BTreeMap<Pid, Info>,
513 pending: Vec<Proc<()>>,
516 waits: BTreeMap<Pid, Wait>,
517 tapes: BTreeMap<Pid, Tape>,
518 journals: BTreeMap<Pid, VecDeque<Value>>,
520 accounts: BTreeMap<Pid, Brought>,
521 signals: VecDeque<(Pid, Pid, Value)>,
523 migrations: BTreeMap<Pid, Migration>,
524 outcomes: BTreeMap<u64, Vec<(Pid, Result<(), String>)>>,
525 schedule: VecDeque<SliceRecord>,
526 traces: VecDeque<Trace>,
527 clock: Clock,
528 root_signal: Option<(Pid, Value)>,
530 root_due: i64,
533 evicting: BTreeSet<Pid>,
534 arrivals: Arrivals,
535 aloft: BTreeSet<Pid>,
536 doomed: BTreeMap<Pid, Value>,
537 recalled: BTreeSet<Pid>,
538}
539
540impl NodeState {
541 fn new(clock: ClockKind) -> Self {
542 let mut info = BTreeMap::new();
543 info.insert(ROOT, Info::new(Kind::Root));
544 let mut boxes = BTreeMap::new();
545 boxes.insert(ROOT, VecDeque::new());
546 Self {
547 current: ROOT,
548 next_pid: 1,
549 next_ref: 1,
550 next_request: 1,
551 boxes,
552 info,
553 pending: Vec::new(),
554 waits: BTreeMap::new(),
555 tapes: BTreeMap::new(),
556 journals: BTreeMap::new(),
557 accounts: BTreeMap::new(),
558 signals: VecDeque::new(),
559 migrations: BTreeMap::new(),
560 outcomes: BTreeMap::new(),
561 schedule: VecDeque::new(),
562 traces: VecDeque::new(),
563 clock: Clock::new(clock),
564 root_signal: None,
565 root_due: 0,
566 evicting: BTreeSet::new(),
567 arrivals: Arrivals::default(),
568 aloft: BTreeSet::new(),
569 doomed: BTreeMap::new(),
570 recalled: BTreeSet::new(),
571 }
572 }
573
574 fn running(&self) -> Pid {
575 scheduling::lane().unwrap_or(self.current)
576 }
577
578 fn supervisor(&mut self, strategy: Strategy, max_restarts: usize) -> Pid {
579 let sup = Pid(self.next_pid);
580 self.next_pid += 1;
581 self.info.insert(
582 sup,
583 Info::new(Kind::Supervisor(Sup {
584 strategy,
585 max_restarts,
586 children: Vec::new(),
587 })),
588 );
589 self.boxes.insert(sup, VecDeque::new());
590 sup
591 }
592
593 fn enrol(&mut self, sup: Pid, child: Pid) {
594 if let Some(Info {
595 kind: Kind::Supervisor(s),
596 ..
597 }) = self.info.get_mut(&sup)
598 {
599 s.children.push(child);
600 }
601 }
602
603 fn drop_child(&mut self, sup: Pid, child: Pid) {
604 if let Some(Info {
605 kind: Kind::Supervisor(s),
606 ..
607 }) = self.info.get_mut(&sup)
608 {
609 s.children.retain(|c| *c != child);
610 }
611 }
612
613 fn held_at_boundary(&self, pid: Pid) -> bool {
614 self.migrations.contains_key(&pid) || self.evicting.contains(&pid)
615 }
616
617 fn held_at_receive(&self, pid: Pid) -> bool {
618 self.evicting.contains(&pid)
619 && self
620 .info
621 .get(&pid)
622 .is_some_and(|i| matches!(i.kind, Kind::Plain))
623 }
624
625 fn opens_dispatch(&mut self, pid: Pid) -> bool {
626 match self.info.get_mut(&pid) {
627 Some(info) if info.dispatch.is_some() && !info.armed => {
628 info.armed = true;
629 true
630 }
631 _ => false,
632 }
633 }
634
635 fn deliver(&mut self, to: Pid, msg: Value) {
636 if let Some(q) = self.boxes.get_mut(&to) {
637 q.push_back(msg);
638 }
639 }
640
641 fn is_ready(&mut self, pid: Pid) -> bool {
644 let now = self.clock.now();
645 let mailbox = self.boxes.get(&pid);
646 match self.waits.get(&pid) {
647 Some(Wait::Receive { pattern, deadline }) => {
648 !self.held_at_receive(pid)
649 && (mailbox.is_some_and(|b| pattern::first_match(b, pattern).is_some())
650 || deadline.is_some_and(|d| now >= d))
651 }
652 Some(Wait::ServerNext) => {
653 mailbox.is_some_and(|b| !b.is_empty()) && !self.held_at_boundary(pid)
654 }
655 None => mailbox.is_some_and(|b| !b.is_empty()),
656 }
657 }
658
659 fn earliest_deadline(&self) -> Option<u64> {
660 self.waits
661 .values()
662 .filter_map(|w| match w {
663 Wait::Receive { deadline, .. } => *deadline,
664 Wait::ServerNext => None,
665 })
666 .min()
667 }
668
669 fn forget(&mut self, pid: Pid) {
670 self.info.remove(&pid);
671 self.boxes.remove(&pid);
672 self.waits.remove(&pid);
673 self.tapes.remove(&pid);
674 self.journals.remove(&pid);
675 self.accounts.remove(&pid);
676 self.evicting.remove(&pid);
677 }
678
679 fn note_taken(&mut self, pid: Pid, msg: &Value, bound: usize) {
681 let ring = self.journals.entry(pid).or_default();
682 if ring.len() == bound {
683 ring.pop_front();
684 if let Some(a) = self.accounts.get_mut(&pid) {
685 a.mark.dropped_oldest(ring);
686 }
687 }
688 ring.push_back(msg.clone());
689 }
690}
691
692pub struct Node {
694 config: NodeConfig,
695 me: Weak<Node>,
696 state: Mutex<NodeState>,
697 sched: Mutex<Vec<Proc<()>>>,
700 proto: OnceLock<Arc<Interpreter<()>>>,
703 lanes: Mutex<BTreeMap<Pid, mpsc::Sender<Proc<()>>>>,
704 flights: Mutex<mpsc::Receiver<Flight>>,
705 radio: mpsc::Sender<Flight>,
706 gate: Mutex<Gate>,
707 gated: Condvar,
708 rooted: Mutex<Option<Weak<()>>>,
709}
710
711#[derive(Default)]
712struct Gate {
713 driving: usize,
714 inside: usize,
715}
716
717struct Pass<'a>(&'a Node);
718
719impl Drop for Pass<'_> {
720 fn drop(&mut self) {
721 let mut g = lock(&self.0.gate);
722 g.inside -= 1;
723 self.0.gated.notify_all();
724 }
725}
726
727struct Driving<'a>(&'a Node);
728
729impl Drop for Driving<'_> {
730 fn drop(&mut self) {
731 let node = self.0;
732 let mut g = lock(&node.gate);
733 g.driving -= 1;
734 while g.driving == 0 && g.inside > 0 {
735 g = node
736 .gated
737 .wait(g)
738 .unwrap_or_else(std::sync::PoisonError::into_inner);
739 }
740 }
741}
742
743enum Flight {
744 Landed(Proc<()>, Slice, usize),
745 Poked,
746 Fell(Box<dyn std::any::Any + Send>),
747}
748
749fn lock<T>(m: &Mutex<T>) -> MutexGuard<'_, T> {
750 m.lock().unwrap_or_else(std::sync::PoisonError::into_inner)
751}
752
753fn kw(s: &str) -> Value {
754 Value::Keyword(Arc::from(s))
755}
756
757fn int(n: impl TryInto<i64>) -> Value {
758 Value::Int(n.try_into().unwrap_or(i64::MAX))
759}
760
761#[must_use]
763pub fn reason_value(r: &ExitReason) -> Value {
764 match r {
765 ExitReason::Normal => kw("normal"),
766 ExitReason::Crashed(m) => Value::list([kw("crashed"), Value::Str(Arc::from(m.as_str()))]),
767 ExitReason::Runaway(m) => Value::list([kw("runaway"), Value::Str(Arc::from(m.as_str()))]),
768 ExitReason::Killed(m) => Value::list([kw("killed"), Value::Str(Arc::from(m.as_str()))]),
769 ExitReason::Fatal(m) => Value::list([kw("fatal"), Value::Str(Arc::from(m.as_str()))]),
770 }
771}
772
773fn is_keyword(v: &Value, name: &str) -> bool {
774 matches!(v, Value::Keyword(k) if k.trim_start_matches(':') == name)
775}
776
777fn render(v: &Value) -> String {
778 v.to_string()
779}
780
781fn written(supervised: bool, reason: &ExitReason) -> bool {
786 supervised || reason.is_fatal()
787}
788
789fn refuse(word: &str, msg: impl Into<String>, span: Span) -> EvalError {
790 EvalError::native_fn(Arc::<str>::from(word), msg.into(), span)
791}
792
793fn parking(e: &ParkError) -> String {
794 match e {
795 ParkError::NoFrames => "it holds no frames: it has not started, or it has returned".into(),
796 ParkError::NoDataForm { what, at } => format!(
797 "it holds {what} in {at}, which has no data form and does not cross a wire; open it \
798 inside the process once it has moved"
799 ),
800 ParkError::OtherChunk { what, at } => format!(
801 "it runs code it was not spawned with ({what}, {at}), and its image carries only the \
802 code it was spawned with"
803 ),
804 ParkError::ChunkMismatch { parked, chunk } => format!(
805 "its frames were parked in chunk {parked}, and its code compiles here to chunk \
806 {chunk}: this blue compiles it differently from the one it left"
807 ),
808 ParkError::UnknownNative { name } => {
809 format!("it holds the native function `{name}`, which this node does not register")
810 }
811 ParkError::Malformed(why) => format!("its frames are malformed: {why}"),
812 }
813}
814
815fn pid_arg(v: &Value, word: &str, span: Span) -> Result<Pid, EvalError> {
816 match v {
817 Value::Int(n) if *n >= 0 => Ok(Pid(n.unsigned_abs())),
818 other => Err(refuse(
819 word,
820 format!("expected a pid, got {}", other.type_name()),
821 span,
822 )),
823 }
824}
825
826fn compile(forms: &[Spanned]) -> Result<Chunk, String> {
827 compile_program(forms).map_err(|e| e.to_string())
828}
829
830fn run_in(
832 interp: &mut Interpreter<()>,
833 forms: &[Spanned],
834 depth: Option<usize>,
835) -> Result<Value, String> {
836 let chunk = compile(forms)?;
837 let mut vm = Vm::with_budget(Budget {
838 fuel: Some(MIGRATION_FUEL),
839 max_depth: depth,
840 quantum: None,
841 });
842 vm.run(&chunk, interp, &mut ()).map_err(|e| e.to_string())
843}
844
845fn bind_all(interp: &mut Interpreter<()>, bindings: &[(Arc<str>, Value)]) {
846 for (n, v) in bindings {
847 interp.define_global(n.clone(), v.clone());
848 }
849}
850
851fn receive_args(args: &[Value], span: Span) -> Result<(Value, Option<u64>, Value), EvalError> {
854 let pattern = args
855 .first()
856 .cloned()
857 .unwrap_or_else(|| kw(pattern::WILDCARD));
858 let timeout = match args.get(1) {
859 None | Some(Value::Nil) => None,
860 Some(Value::Int(n)) if *n >= 0 => Some(n.unsigned_abs()),
861 Some(other) => {
862 return Err(refuse(
863 "receive",
864 format!(
865 "the timeout is milliseconds (an Int >= 0) or nil, got {}",
866 other.type_name()
867 ),
868 span,
869 ))
870 }
871 };
872 let default = args.get(2).cloned().unwrap_or(Value::Nil);
873 Ok((pattern, timeout, default))
874}
875
876#[derive(Clone)]
878enum Side {
879 Root(Arc<Node>),
882 Child(Weak<Node>),
885}
886
887impl Side {
888 fn node(&self, word: &str, span: Span) -> Result<Arc<Node>, EvalError> {
889 match self {
890 Side::Root(n) => Ok(n.clone()),
891 Side::Child(w) => w
892 .upgrade()
893 .ok_or_else(|| refuse(word, "the process's node has shut down", span)),
894 }
895 }
896}
897
898#[derive(Clone, Copy, PartialEq, Eq)]
900enum Who {
901 Root,
902 Child(Pid),
903}
904
905impl Who {
906 fn pid(self) -> Pid {
907 match self {
908 Who::Root => ROOT,
909 Who::Child(p) => p,
910 }
911 }
912}
913
914type Word = fn(&Node, Who, &[Value], Span) -> Result<Value, EvalError>;
915
916const WORDS: &[(&str, Arity, Word)] = &[
918 ("spawn", Arity::Range(1, 3), w_spawn),
919 ("spawn_link", Arity::Range(1, 3), w_spawn_link),
920 ("spawn_server", Arity::Range(2, 3), w_spawn_server),
921 ("send", Arity::Exact(2), w_send),
922 ("self", Arity::Exact(0), w_self),
923 ("alive?", Arity::Exact(1), w_alive),
924 ("link", Arity::Exact(1), w_link),
925 ("unlink", Arity::Exact(1), w_unlink),
926 ("monitor", Arity::Exact(1), w_monitor),
927 ("demonitor", Arity::Exact(1), w_demonitor),
928 ("exit", Arity::Exact(2), w_exit),
929 ("trap_exit", Arity::Exact(1), w_trap_exit),
930 ("supervisor", Arity::Range(2, 3), w_supervisor),
931 ("children", Arity::Exact(1), w_children),
932 ("defmigrate", Arity::Range(2, 3), w_defmigrate),
933 ("def_hash", Arity::Exact(1), w_def_hash),
934];
935
936const CARRIED: &[(&str, Arity, Word, Word)] = &[
937 ("evict", Arity::Exact(3), w_evict, w_evict_provided),
938 ("accept", Arity::Exact(2), w_accept, w_accept_provided),
939];
940
941pub fn install<H: 'static>(interp: &mut Interpreter<H>, node: Arc<Node>) {
944 let carrying = node.config.carrying();
945 let rooted = Arc::new(());
946 *lock(&node.rooted) = Some(Arc::downgrade(&rooted));
947 let side = Side::Root(node);
948 for (name, arity, word) in WORDS {
949 register(interp, &side, name, *arity, *word);
950 }
951 carry(interp, &side, carrying);
952 let s = side.clone();
955 interp.register_fn(
956 "receive",
957 Arity::Range(0, 3),
958 move |args: &[Value], _h: &mut H, span| {
959 let _rooted = &rooted;
960 let node = s.node("receive", span)?;
961 node.receive_root(args, span)
962 },
963 );
964}
965
966fn carry<H: 'static>(interp: &mut Interpreter<H>, side: &Side, carrying: Carrying) {
967 for (name, arity, word, provided) in CARRIED {
968 if !carrying.carrier {
969 if interp.lookup_global(name).is_some() {
970 register_as(interp, side, name, (*name).to_string(), *arity, w_uncarried);
971 }
972 continue;
973 }
974 if carrying.withheld {
975 refuse_as(interp, side, name);
976 } else {
977 register_as(interp, side, name, (*name).to_string(), *arity, *word);
978 }
979 if carrying.scoped {
980 register_as(interp, side, name, provided_name(name), *arity, *provided);
981 }
982 }
983}
984
985fn refuse_as<H: 'static>(interp: &mut Interpreter<H>, side: &Side, name: &'static str) {
986 let who = match side {
987 Side::Root(_) => "this run's frame",
988 Side::Child(_) => "this process's frame",
989 };
990 interp.register_fn(name, Arity::Any, move |_: &[Value], _h: &mut H, span| {
991 Err(breach(name, Capability::CommScavenger, who, span))
992 });
993}
994
995fn register_as<H: 'static>(
996 interp: &mut Interpreter<H>,
997 side: &Side,
998 name: &'static str,
999 bound: String,
1000 arity: Arity,
1001 word: Word,
1002) {
1003 bind_word(interp, side, name, bound, arity, word, false);
1004}
1005
1006fn register<H: 'static>(
1007 interp: &mut Interpreter<H>,
1008 side: &Side,
1009 name: &'static str,
1010 arity: Arity,
1011 word: Word,
1012) {
1013 bind_word(interp, side, name, name.to_string(), arity, word, true);
1014}
1015
1016fn bind_word<H: 'static>(
1017 interp: &mut Interpreter<H>,
1018 side: &Side,
1019 name: &'static str,
1020 bound: String,
1021 arity: Arity,
1022 word: Word,
1023 taped: bool,
1024) {
1025 let side = side.clone();
1026 interp.register_fn(bound, arity, move |args: &[Value], _h: &mut H, span| {
1027 let node = side.node(name, span)?;
1028 let _pass = node.pass(name, span)?;
1029 let who = match side {
1030 Side::Root(_) => Who::Root,
1031 Side::Child(_) => Who::Child(node.current()),
1032 };
1033 let out = word(&node, who, args, span);
1034 if let (true, Who::Child(pid)) = (taped, who) {
1035 node.record(pid, name, &out);
1036 }
1037 node.poke();
1038 out
1039 });
1040}
1041
1042fn install_child(proto: &mut Interpreter<()>, me: &Weak<Node>, carrying: Carrying) {
1044 let side = Side::Child(me.clone());
1045 for (name, arity, word) in WORDS {
1046 register(proto, &side, name, *arity, *word);
1047 }
1048 carry(proto, &side, carrying);
1049
1050 let (r, c) = (me.clone(), me.clone());
1056 proto.register_awaitable_fn(
1057 "receive",
1058 Arity::Range(0, 3),
1059 move |args: &[Value], _h: &()| {
1060 let Some(node) = r.upgrade() else { return true };
1061 let Ok(_pass) = node.pass("receive", Span::synthetic()) else {
1062 return true;
1063 };
1064 let Ok((pattern, timeout, _)) = receive_args(args, Span::synthetic()) else {
1065 return true;
1067 };
1068 let mut st = node.st();
1069 let pid = st.running();
1070 let held = st.held_at_receive(pid);
1071 if !held
1072 && st
1073 .boxes
1074 .get(&pid)
1075 .is_some_and(|b| pattern::first_match(b, &pattern).is_some())
1076 {
1077 return true;
1078 }
1079 let now = st.clock.now();
1080 let deadline = match st.waits.get(&pid) {
1081 Some(Wait::Receive { deadline, .. }) => *deadline,
1082 _ => timeout.map(|t| now.saturating_add(t)),
1083 };
1084 st.waits.insert(pid, Wait::Receive { pattern, deadline });
1085 !held && deadline.is_some_and(|d| now >= d)
1086 },
1087 move |args: &[Value], _h: &mut (), span| {
1088 let node = Side::Child(c.clone()).node("receive", span)?;
1089 let _pass = node.pass("receive", span)?;
1090 let (pattern, _, default) = receive_args(args, span)?;
1091 let out = {
1092 let mut st = node.st();
1093 let pid = st.running();
1094 st.waits.remove(&pid);
1095 let taken = st.boxes.get_mut(&pid).and_then(|b| {
1096 let i = pattern::first_match(b, &pattern)?;
1097 b.remove(i)
1098 });
1099 node.journal(&mut st, pid, taken.as_ref());
1100 Ok(taken.unwrap_or(default))
1101 };
1102 node.record(node.current(), "receive", &out);
1103 out
1104 },
1105 );
1106
1107 let (r, c) = (me.clone(), me.clone());
1111 proto.register_awaitable_fn(
1112 SERVER_NEXT,
1113 Arity::Exact(1),
1114 move |_args: &[Value], _h: &()| {
1115 let Some(node) = r.upgrade() else { return true };
1116 let Ok(_pass) = node.pass(SERVER_NEXT, Span::synthetic()) else {
1117 return true;
1118 };
1119 let mut st = node.st();
1120 let pid = st.running();
1121 let opens = st.opens_dispatch(pid);
1122 let ready = !opens
1123 && st.boxes.get(&pid).is_some_and(|b| !b.is_empty())
1124 && !st.held_at_boundary(pid);
1125 if !ready {
1126 st.waits.insert(pid, Wait::ServerNext);
1127 }
1128 ready
1129 },
1130 move |args: &[Value], _h: &mut (), span| {
1131 let node = Side::Child(c.clone()).node(SERVER_NEXT, span)?;
1132 let _pass = node.pass(SERVER_NEXT, span)?;
1133 let mut st = node.st();
1134 let pid = st.running();
1135 st.waits.remove(&pid);
1136 let taken = st.boxes.get_mut(&pid).and_then(VecDeque::pop_front);
1137 node.journal(&mut st, pid, taken.as_ref());
1138 let msg = taken.unwrap_or(Value::Nil);
1139 if let Some(info) = st.info.get_mut(&pid) {
1142 info.snapshot = Some(args.first().cloned().unwrap_or(Value::Nil));
1143 info.armed = false;
1144 }
1145 st.tapes.insert(pid, Tape::default());
1146 drop(st);
1147 let out = Ok(msg);
1148 node.record(pid, SERVER_NEXT, &out);
1149 out
1150 },
1151 );
1152}
1153
1154fn record_host_names(proto: &mut Interpreter<()>, me: &Weak<Node>) {
1157 for name in host_names() {
1158 let Some(original @ Value::NativeFn(_)) = proto.lookup_global(name) else {
1159 continue;
1160 };
1161 let Value::NativeFn(nf) = &original else {
1162 continue;
1163 };
1164 let arity = nf.arity;
1165 let wrapper = format!("%host:{name}");
1166 let me = me.clone();
1167 proto.register_higher_order_fn(
1168 wrapper.clone(),
1169 arity,
1170 move |args: &[Value], h: &mut (), caller: &tatara_lisp_eval::ffi::Caller<()>, span| {
1171 let out = caller.apply_value(&original, args.to_vec(), h, span);
1172 if let Some(node) = me.upgrade() {
1173 let pid = node.current();
1174 if pid != ROOT {
1175 node.record(pid, name, &out);
1176 }
1177 }
1178 out
1179 },
1180 );
1181 if let Some(w) = proto.lookup_global(&wrapper) {
1182 proto.define_global(name, w);
1183 }
1184 }
1185}
1186
1187#[must_use]
1189pub fn host_names() -> Vec<&'static str> {
1190 blue_lang_waku::capability::Capability::host_effects()
1191 .into_iter()
1192 .flat_map(blue_lang_waku::capability::Capability::names)
1193 .collect()
1194}
1195
1196fn unclassed(p: &mut Proc<()>, why: &str) -> Slice {
1197 let reason = ExitReason::Crashed(format!("the OS refused the idle class: {why}"));
1198 p.state = ProcState::Exited(reason.clone());
1199 Slice::Exited(reason)
1200}
1201
1202fn note_slice(st: &mut NodeState, pid: Pid, out: &Slice, burned: usize) {
1203 let outcome = match out {
1204 Slice::Yielded => "yielded",
1205 Slice::Blocked => "blocked",
1206 Slice::Done(_) => "done",
1207 Slice::Exited(_) => "exited",
1208 };
1209 if st.schedule.len() == SCHEDULE_RING {
1210 st.schedule.pop_front();
1211 }
1212 st.schedule.push_back(SliceRecord {
1213 pid: pid.0,
1214 outcome: outcome.to_string(),
1215 burned,
1216 });
1217}
1218
1219fn fly(me: &Weak<Node>, pid: Pid, queue: &mpsc::Receiver<Proc<()>>, radio: &mpsc::Sender<Flight>) {
1220 let lowered = scheduling::lower();
1221 scheduling::board(pid);
1222 while let Ok(mut p) = queue.recv() {
1223 let Some(node) = me.upgrade() else { break };
1224 let before = p.vm.burned();
1225 let ran = std::panic::catch_unwind(std::panic::AssertUnwindSafe(|| match &lowered {
1226 Ok(()) => {
1227 let out = node.slice_held(&mut p);
1228 node.dispatch_held(&mut p, out)
1229 }
1230 Err(why) => unclassed(&mut p, why),
1231 }));
1232 let flight = match ran {
1233 Ok(out) => {
1234 let burned = p.vm.burned().saturating_sub(before);
1235 Flight::Landed(p, out, burned)
1236 }
1237 Err(panic) => Flight::Fell(panic),
1238 };
1239 drop(node);
1240 if radio.send(flight).is_err() {
1241 break;
1242 }
1243 }
1244}
1245
1246enum Driven<T> {
1248 Done(T),
1249 TimedOut,
1250 Deadlock(Vec<Pid>),
1251}
1252
1253impl Node {
1254 #[must_use]
1256 pub fn new(config: NodeConfig) -> Arc<Self> {
1257 let clock = config.clock;
1258 let (radio, flights) = mpsc::channel();
1259 Arc::new_cyclic(|me| Self {
1260 config,
1261 me: me.clone(),
1262 state: Mutex::new(NodeState::new(clock)),
1263 sched: Mutex::new(Vec::new()),
1264 proto: OnceLock::new(),
1265 lanes: Mutex::new(BTreeMap::new()),
1266 flights: Mutex::new(flights),
1267 radio,
1268 gate: Mutex::new(Gate::default()),
1269 gated: Condvar::new(),
1270 rooted: Mutex::new(None),
1271 })
1272 }
1273
1274 fn st(&self) -> MutexGuard<'_, NodeState> {
1275 lock(&self.state)
1276 }
1277
1278 fn current(&self) -> Pid {
1279 self.st().running()
1280 }
1281
1282 fn pass(&self, word: &str, span: Span) -> Result<Option<Pass<'_>>, EvalError> {
1283 if scheduling::lane().is_none() {
1284 return Ok(None);
1285 }
1286 let mut g = lock(&self.gate);
1287 while g.driving == 0 {
1288 let rooted = lock(&self.rooted)
1289 .as_ref()
1290 .is_some_and(|w| w.strong_count() > 0);
1291 if !rooted {
1292 return Err(refuse(word, "the process's node has shut down", span));
1293 }
1294 g = self
1295 .gated
1296 .wait_timeout(g, Duration::from_millis(100))
1297 .map_or_else(|e| e.into_inner().0, |(g, _)| g);
1298 }
1299 g.inside += 1;
1300 Ok(Some(Pass(self)))
1301 }
1302
1303 fn driving(&self) -> Driving<'_> {
1304 lock(&self.gate).driving += 1;
1305 self.gated.notify_all();
1306 Driving(self)
1307 }
1308
1309 fn poke(&self) {
1310 if scheduling::lane().is_some() {
1311 let _ = self.radio.send(Flight::Poked);
1312 }
1313 }
1314
1315 fn proto(&self) -> Arc<Interpreter<()>> {
1316 self.proto
1317 .get_or_init(|| {
1318 let mut p = (self.config.child)();
1319 install_child(&mut p, &self.me, self.config.carrying());
1320 record_host_names(&mut p, &self.me);
1321 Arc::new(p)
1322 })
1323 .clone()
1324 }
1325
1326 fn builder(
1327 &self,
1328 bindings: &[(Arc<str>, Value)],
1329 ) -> Arc<dyn Fn() -> Interpreter<()> + Send + Sync> {
1330 let proto = self.proto();
1331 let bindings = bindings.to_vec();
1332 Arc::new(move || {
1333 let mut i = proto.fork();
1334 bind_all(&mut i, &bindings);
1335 i
1336 })
1337 }
1338
1339 fn record(&self, pid: Pid, call: &str, out: &Result<Value, EvalError>) {
1340 let max = self.config.max_tape;
1341 let mut st = self.st();
1342 let Some(tape) = st.tapes.get_mut(&pid) else {
1343 return;
1344 };
1345 if tape.truncated {
1346 return;
1347 }
1348 if tape.entries.len() >= max {
1349 tape.truncated = true;
1352 tape.entries = Vec::new();
1353 tape.carried = None;
1354 return;
1355 }
1356 tape.entries.push(TapeEntry::of(call, out));
1357 }
1358
1359 fn journal(&self, st: &mut NodeState, pid: Pid, taken: Option<&Value>) {
1361 if let (true, Some(msg)) = (self.config.journals(), taken) {
1362 st.note_taken(pid, msg, self.config.journal);
1363 }
1364 }
1365
1366 fn switch(&self, st: &mut NodeState, pid: Pid) {
1368 st.current = pid;
1369 self.config.host_io.enter(pid);
1370 }
1371
1372 fn charged<T>(&self, held: &mut i64, bound: Option<u64>, f: impl FnOnce() -> T) -> (T, bool) {
1375 let Some(meter) = self.config.meter else {
1376 return (f(), false);
1377 };
1378 let mut f = Some(f);
1379 let mut out = None;
1380 let over = (meter.charged)(held, bound, &mut || out = f.take().map(|f| f()));
1381 match (out, f) {
1382 (Some(v), _) => (v, over),
1383 (None, Some(f)) => (f(), false),
1384 (None, None) => unreachable!("a closure that ran left its value"),
1385 }
1386 }
1387
1388 fn apart<T>(&self, f: impl FnOnce() -> T) -> (T, i64) {
1390 let mut held = 0;
1391 let (value, _) = self.charged(&mut held, None, f);
1392 (value, held)
1393 }
1394
1395 fn credit(&self, to: Pid, bytes: i64) {
1397 let Some(meter) = self.config.meter.filter(|_| bytes != 0) else {
1398 return;
1399 };
1400 let mut st = self.st();
1401 if to == st.running() {
1402 drop(st);
1403 (meter.charge)(bytes);
1404 } else if to == ROOT {
1405 st.root_due += bytes;
1406 } else if let Some(info) = st.info.get_mut(&to) {
1407 info.held += bytes;
1408 }
1409 }
1410
1411 fn settle_root(&self) {
1413 let due = std::mem::take(&mut self.st().root_due);
1414 if let (Some(meter), true) = (self.config.meter, due != 0) {
1415 (meter.charge)(due);
1416 }
1417 }
1418
1419 fn slice_held(&self, p: &mut Proc<()>) -> Slice {
1422 let pid = p.pid;
1423 let (mut held, bound) = self
1424 .st()
1425 .info
1426 .get(&pid)
1427 .map_or((0, None), |i| (i.held, i.bound));
1428 let Some(limit) = bound else {
1429 return p.slice(&mut ());
1430 };
1431 let start = held;
1432 let (out, over) = self.charged(&mut held, bound, || p.slice(&mut ()));
1433 if let Some(info) = self.st().info.get_mut(&pid) {
1434 info.held += held - start;
1435 }
1436 if !over {
1437 return out;
1438 }
1439 let reason = ExitReason::memory_exhausted(limit);
1440 p.state = ProcState::Exited(reason.clone());
1441 Slice::Exited(reason)
1442 }
1443
1444 fn dispatch_held(&self, p: &mut Proc<()>, out: Slice) -> Slice {
1445 if !matches!(out, Slice::Yielded | Slice::Blocked) {
1446 return out;
1447 }
1448 let burned = p.vm.burned();
1449 let mut st = self.st();
1450 let boundary = matches!(st.waits.get(&p.pid), Some(Wait::ServerNext));
1451 let Some(info) = st.info.get_mut(&p.pid) else {
1452 return out;
1453 };
1454 let Some(limit) = info.dispatch else {
1455 return out;
1456 };
1457 if info
1458 .dispatch_from
1459 .is_some_and(|from| burned.saturating_sub(from) > limit)
1460 {
1461 info.dispatch_from = None;
1462 info.armed = false;
1463 info.kept.clone_from(&info.snapshot);
1464 let kept = info.kept.clone();
1465 drop(st);
1466 if let Some(state) = kept {
1467 p.interp.define_global("%state", state);
1468 }
1469 let reason = ExitReason::dispatch_exhausted(limit);
1470 p.state = ProcState::Exited(reason.clone());
1471 return Slice::Exited(reason);
1472 }
1473 if boundary {
1474 info.dispatch_from = info.armed.then_some(burned);
1475 }
1476 out
1477 }
1478
1479 fn slice_idle(&self, p: &mut Proc<()>) -> Slice {
1480 let ran = std::thread::scope(|s| {
1481 std::thread::Builder::new()
1482 .name("blue-idle".to_string())
1483 .stack_size(scheduling::LANE_STACK)
1484 .spawn_scoped(s, || scheduling::lower().map(|()| self.slice_held(p)))
1485 .map_err(|e| e.to_string())
1486 .and_then(|lane| {
1487 lane.join()
1488 .unwrap_or_else(|panic| std::panic::resume_unwind(panic))
1489 })
1490 });
1491 ran.unwrap_or_else(|why| unclassed(p, &why))
1492 }
1493
1494 fn launch(&self, p: Proc<()>) {
1495 let pid = p.pid;
1496 self.st().aloft.insert(pid);
1497 let mut p = p;
1498 let mut why = String::new();
1499 for _ in 0..2 {
1500 let lane = {
1501 let mut lanes = lock(&self.lanes);
1502 match lanes.get(&pid) {
1503 Some(lane) => Ok(lane.clone()),
1504 None => self.lane(pid).inspect(|lane| {
1505 lanes.insert(pid, lane.clone());
1506 }),
1507 }
1508 };
1509 match lane {
1510 Ok(lane) => match lane.send(p) {
1511 Ok(()) => return,
1512 Err(mpsc::SendError(back)) => {
1513 lock(&self.lanes).remove(&pid);
1514 why = "its lane closed".to_string();
1515 p = back;
1516 }
1517 },
1518 Err(e) => {
1519 why = e;
1520 break;
1521 }
1522 }
1523 }
1524 let out = unclassed(&mut p, &why);
1525 let _ = self.radio.send(Flight::Landed(p, out, 0));
1526 }
1527
1528 fn lane(&self, pid: Pid) -> Result<mpsc::Sender<Proc<()>>, String> {
1529 let (work, queue) = mpsc::channel::<Proc<()>>();
1530 let me = self.me.clone();
1531 let radio = self.radio.clone();
1532 std::thread::Builder::new()
1533 .name(format!("blue-idle-{}", pid.0))
1534 .stack_size(scheduling::LANE_STACK)
1535 .spawn(move || fly(&me, pid, &queue, &radio))
1536 .map(|_| work)
1537 .map_err(|e| e.to_string())
1538 }
1539
1540 fn land(&self, sched: &mut Vec<Proc<()>>) -> bool {
1541 let mut any = false;
1542 loop {
1543 let next = lock(&self.flights).try_recv();
1544 let Ok(flight) = next else { return any };
1545 any |= self.touch_down(sched, flight);
1546 }
1547 }
1548
1549 fn await_flight(&self, sched: &mut Vec<Proc<()>>, limit: Option<Duration>) {
1550 let next = {
1551 let flights = lock(&self.flights);
1552 match limit {
1553 Some(d) => flights.recv_timeout(d).ok(),
1554 None => flights.recv().ok(),
1555 }
1556 };
1557 if let Some(flight) = next {
1558 self.touch_down(sched, flight);
1559 }
1560 self.land(sched);
1561 }
1562
1563 fn touch_down(&self, sched: &mut Vec<Proc<()>>, flight: Flight) -> bool {
1564 let (p, out, burned) = match flight {
1565 Flight::Landed(p, out, burned) => (p, out, burned),
1566 Flight::Poked => return false,
1567 Flight::Fell(panic) => std::panic::resume_unwind(panic),
1568 };
1569 let pid = p.pid;
1570 let (known, doomed, recalled) = {
1571 let mut st = self.st();
1572 st.aloft.remove(&pid);
1573 note_slice(&mut st, pid, &out, burned);
1574 (
1575 st.info.contains_key(&pid),
1576 st.doomed.remove(&pid),
1577 st.recalled.remove(&pid),
1578 )
1579 };
1580 if !known {
1581 lock(&self.lanes).remove(&pid);
1582 return true;
1583 }
1584 let at = sched.partition_point(|q| q.pid < pid);
1585 sched.insert(at, p);
1586 if recalled {
1587 self.restart(sched, pid);
1588 } else {
1589 match out {
1590 Slice::Yielded | Slice::Blocked => {
1591 if let Some(reason) = doomed {
1592 self.kill(sched, pid, reason);
1593 }
1594 }
1595 out => self.conclude(sched, pid, out),
1596 }
1597 }
1598 self.adopt(sched);
1599 self.process_signals(sched);
1600 true
1601 }
1602
1603 fn conclude(&self, sched: &mut Vec<Proc<()>>, pid: Pid, out: Slice) {
1604 match out {
1605 Slice::Done(v) => self.on_exit(sched, pid, &ExitReason::Normal, kw("normal"), Some(v)),
1606 Slice::Exited(r) => {
1607 let rv = reason_value(&r);
1608 self.on_exit(sched, pid, &r, rv, None);
1609 }
1610 Slice::Yielded | Slice::Blocked => {}
1611 }
1612 }
1613
1614 fn detached(&self, pid: Pid) -> bool {
1615 self.config.idle_lane == IdleLane::Apart
1616 && self
1617 .st()
1618 .info
1619 .get(&pid)
1620 .is_some_and(|i| i.scheduling == Scheduling::Idle)
1621 }
1622
1623 fn start(
1626 &self,
1627 code: Code,
1628 kind: Kind,
1629 def_name: Option<String>,
1630 sup: Option<Pid>,
1631 scheduling: Scheduling,
1632 ) -> Result<Pid, String> {
1633 let bound = self.config.meter.and_then(|m| (m.bound)());
1634 self.apart(|| self.start_unmetered(code, kind, def_name, sup, bound, scheduling))
1635 .0
1636 }
1637
1638 fn start_unmetered(
1639 &self,
1640 code: Code,
1641 kind: Kind,
1642 def_name: Option<String>,
1643 sup: Option<Pid>,
1644 bound: Option<u64>,
1645 scheduling: Scheduling,
1646 ) -> Result<Pid, String> {
1647 let chunk = Arc::new(compile(&code.program())?);
1648 let build = self.builder(&code.bindings);
1649 let dispatch = matches!(kind, Kind::Server)
1650 .then_some(self.config.dispatch_steps)
1651 .flatten();
1652 let initial = match kind {
1653 Kind::Server => code
1654 .bindings
1655 .iter()
1656 .find(|(n, _)| &**n == "%state")
1657 .map(|(_, v)| v.clone()),
1658 _ => None,
1659 };
1660 let mut st = self.st();
1661 let pid = Pid(st.next_pid);
1662 st.next_pid += 1;
1663 let proc = Proc::new(pid, chunk.clone(), self.config.budget, build);
1664 let mut info = Info::new(kind);
1665 info.code = code;
1666 info.chunk = Some(chunk);
1667 info.def_name = def_name;
1668 info.sup = sup;
1669 info.snapshot.clone_from(&initial);
1670 info.entered.clone_from(&initial);
1671 info.initial = initial;
1672 info.bound = bound;
1673 info.scheduling = scheduling;
1674 info.dispatch = dispatch;
1675 st.info.insert(pid, info);
1676 st.boxes.insert(pid, VecDeque::new());
1677 st.tapes.insert(pid, Tape::default());
1678 st.pending.push(proc);
1679 Ok(pid)
1680 }
1681
1682 fn adopt(&self, sched: &mut Vec<Proc<()>>) {
1685 let pending = std::mem::take(&mut self.st().pending);
1686 sched.extend(pending);
1687 }
1688
1689 fn has_work(&self, sched: &[Proc<()>]) -> bool {
1690 let mut st = self.st();
1691 sched
1692 .iter()
1693 .any(|p| p.is_runnable() || (p.state.is_blocked() && st.is_ready(p.pid)))
1694 }
1695
1696 fn round(&self, sched: &mut Vec<Proc<()>>) {
1699 {
1700 let mut st = self.st();
1701 for p in sched.iter_mut() {
1702 if p.state.is_blocked() && st.is_ready(p.pid) {
1703 p.state = ProcState::Runnable;
1704 }
1705 }
1706 }
1707 let mut end = sched.len();
1708 let mut idx = 0;
1709 while idx < end {
1710 if !sched[idx].is_runnable() {
1711 idx += 1;
1712 continue;
1713 }
1714 let pid = sched[idx].pid;
1715 if self.detached(pid) {
1716 self.launch(sched.remove(idx));
1717 end -= 1;
1718 continue;
1719 }
1720 self.switch(&mut self.st(), pid);
1721 let idle = self
1722 .st()
1723 .info
1724 .get(&pid)
1725 .is_some_and(|i| i.scheduling == Scheduling::Idle);
1726 let before = sched[idx].vm.burned();
1727 let out = if idle {
1728 self.slice_idle(&mut sched[idx])
1729 } else {
1730 self.slice_held(&mut sched[idx])
1731 };
1732 let out = self.dispatch_held(&mut sched[idx], out);
1733 let burned = sched[idx].vm.burned().saturating_sub(before);
1734 {
1735 let mut st = self.st();
1736 self.switch(&mut st, ROOT);
1737 note_slice(&mut st, pid, &out, burned);
1738 }
1739 self.conclude(sched, pid, out);
1740 self.adopt(sched);
1741 self.process_signals(sched);
1742 idx += 1;
1743 }
1744 self.reap(sched);
1745 }
1746
1747 fn reap(&self, sched: &mut Vec<Proc<()>>) {
1750 let st = self.st();
1751 sched.retain(|p| {
1752 p.is_runnable()
1753 || p.state.is_blocked()
1754 || st.info.get(&p.pid).is_some_and(|i| i.sup.is_some())
1755 });
1756 let mut lanes = lock(&self.lanes);
1757 if !lanes.is_empty() {
1758 lanes.retain(|pid, _| st.aloft.contains(pid) || sched.iter().any(|p| p.pid == *pid));
1759 }
1760 }
1761
1762 fn process_signals(&self, sched: &mut Vec<Proc<()>>) {
1763 loop {
1764 let next = self.st().signals.pop_front();
1765 let Some((from, to, reason)) = next else {
1766 break;
1767 };
1768 self.signal(sched, from, to, reason);
1769 }
1770 }
1771
1772 fn signal(&self, sched: &mut Vec<Proc<()>>, from: Pid, to: Pid, reason: Value) {
1774 let normal = is_keyword(&reason, "normal");
1775 let kill = is_keyword(&reason, "kill");
1776 let mut st = self.st();
1777 let Some(info) = st.info.get(&to) else { return };
1778 let trapping = info.trap_exit && !kill;
1779 let is_sup = matches!(info.kind, Kind::Supervisor(_));
1780 if trapping {
1781 st.deliver(to, Value::list([kw("exit"), int(from.0), reason]));
1782 return;
1783 }
1784 if normal {
1785 return;
1786 }
1787 if to == ROOT {
1788 st.root_signal = Some((from, reason));
1789 return;
1790 }
1791 drop(st);
1792 let reason = if kill { kw("killed") } else { reason };
1793 if is_sup {
1794 self.shutdown_sup(sched, to, reason);
1795 } else {
1796 self.kill(sched, to, reason);
1797 }
1798 }
1799
1800 fn kill(&self, sched: &mut Vec<Proc<()>>, pid: Pid, reason: Value) {
1801 let text = render(&reason);
1802 let killed = ExitReason::Killed(text);
1803 let alive = match sched.iter_mut().find(|p| p.pid == pid) {
1804 Some(p) if p.is_runnable() || p.state.is_blocked() => {
1805 p.state = ProcState::Exited(killed.clone());
1806 true
1807 }
1808 _ => false,
1809 };
1810 if alive {
1811 self.on_exit(sched, pid, &killed, reason, None);
1812 } else {
1813 let mut st = self.st();
1814 if st.aloft.contains(&pid) {
1815 st.doomed.entry(pid).or_insert(reason);
1816 }
1817 }
1818 }
1819
1820 fn on_exit(
1823 &self,
1824 sched: &mut Vec<Proc<()>>,
1825 pid: Pid,
1826 reason: &ExitReason,
1827 value: Value,
1828 _result: Option<Value>,
1829 ) {
1830 let mut st = self.st();
1831 if !st.info.contains_key(&pid) {
1832 return;
1833 }
1834 let trace = reason
1835 .is_abnormal()
1836 .then(|| self.trace_of(&mut st, pid, reason, &value));
1837 let receipt = self.receipt_of(&st, sched, pid, reason);
1838 let (links, monitors, sup) = {
1839 let info = st.info.get_mut(&pid).expect("checked above");
1840 (
1841 std::mem::take(&mut info.links),
1842 std::mem::take(&mut info.monitors),
1843 info.sup,
1844 )
1845 };
1846 for l in &links {
1847 if let Some(li) = st.info.get_mut(l) {
1848 li.links.remove(&pid);
1849 }
1850 }
1851 for (r, watcher) in monitors {
1852 st.deliver(
1853 watcher,
1854 Value::list([kw("down"), int(r), int(pid.0), value.clone()]),
1855 );
1856 }
1857 st.waits.remove(&pid);
1858 if let Some(m) = st.migrations.remove(&pid) {
1859 st.outcomes.entry(m.request).or_default().push((
1860 pid,
1861 Err("the process exited before it reached a boundary".into()),
1862 ));
1863 }
1864 let supervised =
1865 sup.filter(|s| matches!(st.info.get(s).map(|i| &i.kind), Some(Kind::Supervisor(_))));
1866 if reason.is_fatal() {
1867 if let Some(s) = supervised {
1868 st.drop_child(s, pid);
1869 }
1870 st.forget(pid);
1871 } else if supervised.is_none() {
1872 st.forget(pid);
1873 } else if let Some(info) = st.info.get_mut(&pid) {
1874 info.live = false;
1875 }
1876 drop(st);
1877
1878 if let (Some(r), Some(sink)) = (&receipt, &self.config.receipts) {
1879 sink(r);
1880 }
1881 if let Some(t) = trace {
1882 self.keep_trace(t, written(supervised.is_some(), reason));
1886 }
1887 for l in links {
1888 self.signal(sched, pid, l, value.clone());
1889 }
1890 if let (Some(s), true) = (supervised, reason.is_restartable()) {
1891 self.supervise(sched, s, pid);
1892 }
1893 }
1894
1895 fn supervise(&self, sched: &mut Vec<Proc<()>>, sup: Pid, child: Pid) {
1896 let (spec, restarts) = {
1897 let st = self.st();
1898 let Some(Kind::Supervisor(spec)) = st.info.get(&sup).map(|i| i.kind.clone()) else {
1899 return;
1900 };
1901 let restarts = st.info.get(&child).map_or(0, |i| i.restarts);
1902 (spec, restarts)
1903 };
1904 if restarts >= spec.max_restarts {
1907 self.shutdown_sup(sched, sup, Value::list([kw("gave_up"), int(child.0)]));
1908 return;
1909 }
1910 let Some(pos) = spec.children.iter().position(|c| *c == child) else {
1911 return;
1912 };
1913 for t in spec.strategy.targets(pos, spec.children.len()) {
1914 self.restart(sched, spec.children[t]);
1915 }
1916 }
1917
1918 fn restart(&self, sched: &mut [Proc<()>], pid: Pid) {
1921 let Some(p) = sched.iter_mut().find(|p| p.pid == pid) else {
1922 let mut st = self.st();
1923 if st.aloft.contains(&pid) {
1924 st.recalled.insert(pid);
1925 }
1926 return;
1927 };
1928 let (chunk, kept) = {
1929 let mut st = self.st();
1930 if let Some(b) = st.boxes.get_mut(&pid) {
1934 b.clear();
1935 }
1936 st.waits.remove(&pid);
1937 st.tapes.insert(pid, Tape::default());
1938 st.journals.remove(&pid);
1939 st.accounts.remove(&pid);
1940 self.config.host_io.take(pid);
1941 st.info.get_mut(&pid).map_or((None, None), |info| {
1942 info.restarts += 1;
1943 info.live = true;
1944 info.held = 0;
1945 info.dispatch_from = None;
1946 info.armed = false;
1947 let kept = info.kept.take();
1948 let state = kept.clone().or_else(|| info.initial.clone());
1949 info.snapshot.clone_from(&state);
1950 info.entered = state;
1951 (info.chunk.clone(), kept)
1952 })
1953 };
1954 if let Some(c) = chunk {
1955 p.chunk = c;
1956 }
1957 p.restart();
1958 if let Some(state) = kept {
1959 p.interp.define_global("%state", state);
1960 }
1961 }
1962
1963 fn shutdown_sup(&self, sched: &mut Vec<Proc<()>>, sup: Pid, reason: Value) {
1966 let (children, links, monitors) = {
1967 let mut st = self.st();
1968 let Some(info) = st.info.remove(&sup) else {
1969 return;
1970 };
1971 st.boxes.remove(&sup);
1972 let children = match info.kind {
1973 Kind::Supervisor(s) => s.children,
1974 _ => Vec::new(),
1975 };
1976 for c in &children {
1977 if let Some(ci) = st.info.get_mut(c) {
1978 ci.sup = None;
1979 }
1980 }
1981 for l in &info.links {
1982 if let Some(li) = st.info.get_mut(l) {
1983 li.links.remove(&sup);
1984 }
1985 }
1986 (children, info.links, info.monitors)
1987 };
1988 for c in children {
1989 self.kill(sched, c, kw("shutdown"));
1990 let mut st = self.st();
1993 if st.info.get(&c).is_some_and(|i| i.sup.is_none())
1994 && !st.aloft.contains(&c)
1995 && !sched
1996 .iter()
1997 .any(|p| p.pid == c && (p.is_runnable() || p.state.is_blocked()))
1998 {
1999 st.forget(c);
2000 }
2001 }
2002 {
2003 let mut st = self.st();
2004 for (r, watcher) in monitors {
2005 st.deliver(
2006 watcher,
2007 Value::list([kw("down"), int(r), int(sup.0), reason.clone()]),
2008 );
2009 }
2010 }
2011 for l in links {
2012 self.signal(sched, sup, l, reason.clone());
2013 }
2014 }
2015
2016 fn apply_migrations(&self, sched: &mut [Proc<()>]) -> bool {
2022 let pending: Vec<Pid> = self.st().migrations.keys().copied().collect();
2023 let mut applied = false;
2024 for pid in pending {
2025 let Some(idx) = sched.iter().position(|p| p.pid == pid) else {
2026 continue;
2027 };
2028 let (ready, m, kind) = {
2029 let st = self.st();
2030 let Some(info) = st.info.get(&pid) else {
2031 continue;
2032 };
2033 let p = &sched[idx];
2034 let ready = match info.kind {
2035 Kind::Server => {
2036 p.state.is_blocked() && matches!(st.waits.get(&pid), Some(Wait::ServerNext))
2037 }
2038 Kind::Plain => p.is_runnable() || p.state.is_blocked(),
2039 _ => false,
2040 };
2041 (ready, st.migrations.get(&pid).cloned(), info.kind.clone())
2042 };
2043 let (true, Some(m)) = (ready, m) else {
2044 continue;
2045 };
2046 self.switch(&mut self.st(), pid);
2047 let result = self.migrate(&mut sched[idx], &kind, &m);
2048 let mut st = self.st();
2049 self.switch(&mut st, ROOT);
2050 st.migrations.remove(&pid);
2051 st.outcomes
2052 .entry(m.request)
2053 .or_default()
2054 .push((pid, result));
2055 applied = true;
2056 }
2057 applied
2058 }
2059
2060 fn migrate(&self, p: &mut Proc<()>, kind: &Kind, m: &Migration) -> Result<(), String> {
2065 let depth = self.config.budget.max_depth;
2066 let (old_name, initial) = {
2067 let st = self.st();
2068 let info = st.info.get(&p.pid).ok_or("the process is gone")?;
2069 (info.def_name.clone(), info.initial.clone())
2070 };
2071 let server = matches!(kind, Kind::Server);
2072 if !server && m.transform.is_some() {
2073 return Err(
2074 "a plain process keeps its state in its own frames, so there is no state to \
2075 transform; run it with spawn_server to migrate state"
2076 .into(),
2077 );
2078 }
2079
2080 let mut scratch = p.interp.fork();
2081 bind_all(&mut scratch, &m.code.bindings);
2082 run_in(&mut scratch, &m.code.defs, depth)?;
2083 let (state, initial) = if server {
2084 let old = p.interp.lookup_global("%state").unwrap_or(Value::Nil);
2085 match &m.transform {
2086 Some(t) => {
2087 bind_all(&mut scratch, &t.bindings);
2088 run_in(&mut scratch, &t.defs, depth)?;
2089 let apply = |scratch: &mut Interpreter<()>, v: Value| {
2090 scratch.define_global("%old", v);
2091 run_in(
2092 scratch,
2093 &[code::call(code::symbol("%transform"), &["%old"])],
2094 depth,
2095 )
2096 };
2097 let state = apply(&mut scratch, old)?;
2098 let initial = match initial {
2099 Some(i) => Some(apply(&mut scratch, i)?),
2100 None => None,
2101 };
2102 (Some(state), initial)
2103 }
2104 None => (Some(old), initial),
2105 }
2106 } else {
2107 (None, None)
2108 };
2109 drop(scratch);
2110
2111 bind_all(&mut p.interp, &m.code.bindings);
2113 run_in(&mut p.interp, &m.code.defs, depth)?;
2114 let new_fn = p
2115 .interp
2116 .lookup_global(&m.new_name)
2117 .ok_or_else(|| format!("the migration did not define `{}`", m.new_name))?;
2118 let alias = if server {
2119 "%handler".to_string()
2120 } else {
2121 old_name.clone().unwrap_or_default()
2122 };
2123 if !alias.is_empty() && alias != m.new_name {
2124 p.interp.define_global(alias.as_str(), new_fn);
2125 }
2126 if let Some(s) = &state {
2127 p.interp.define_global("%state", s.clone());
2128 }
2129
2130 let mut st = self.st();
2132 let info = st.info.get_mut(&p.pid).ok_or("the process is gone")?;
2133 let mut defs = info.code.defs.clone();
2134 defs.extend(m.code.defs.iter().cloned());
2135 if !alias.is_empty() && alias != m.new_name {
2136 defs.push(code::define_value(&alias, code::symbol(&m.new_name)));
2137 }
2138 if server {
2139 defs.retain(|d| !SERVER_LOOP.contains(d));
2141 defs.extend(SERVER_LOOP.iter().cloned());
2142 }
2143 let mut new_code = Code {
2144 defs,
2145 bindings: info.code.bindings.clone(),
2146 start: info.code.start.clone(),
2147 };
2148 for (n, v) in &m.code.bindings {
2149 new_code.bind(n, v.clone());
2150 }
2151 if let Some(i) = &initial {
2152 new_code.bind("%state", i.clone());
2153 }
2154 let chunk = Arc::new(compile(&new_code.program())?);
2155 info.code = new_code;
2156 info.chunk = Some(chunk);
2157 info.def_name = Some(m.new_name.clone());
2158 info.initial = initial;
2159 if state.is_some() {
2160 info.snapshot = state;
2161 }
2162 let bindings = info.code.bindings.clone();
2163 st.tapes.insert(p.pid, Tape::default());
2164 drop(st);
2165 p.build = self.builder(&bindings);
2166 Ok(())
2167 }
2168
2169 fn tally(&self) {
2172 if self
2173 .config
2174 .frame
2175 .as_ref()
2176 .is_some_and(|f| f.granted.contains(&Capability::CommScavenger))
2177 {
2178 self.config.host_io.tally();
2179 }
2180 }
2181
2182 fn carrier(&self, word: &str, span: Span) -> Result<Carrier, EvalError> {
2183 self.config
2184 .carrier
2185 .ok_or_else(|| refuse(word, "this node has no wire to carry a process", span))
2186 }
2187
2188 fn evictable(&self, pid: Pid) -> Result<(), String> {
2189 let st = self.st();
2190 let info = st
2191 .info
2192 .get(&pid)
2193 .filter(|i| i.live)
2194 .ok_or_else(|| format!("no process {}", pid.0))?;
2195 match info.kind {
2196 Kind::Server | Kind::Plain => {}
2197 Kind::Root => {
2198 return Err(format!(
2199 "process {} is the program: the door runs it on its own evaluator, the \
2200 tree-walker under `blue run`, which this node does not hold, so it has no \
2201 frames to carry; spawn the work as a process and evict that",
2202 pid.0
2203 ))
2204 }
2205 Kind::Supervisor(_) => {
2206 return Err(format!(
2207 "process {} is a supervisor: it runs no code of its own to carry; evict its \
2208 children one at a time",
2209 pid.0
2210 ))
2211 }
2212 }
2213 if st.held_at_boundary(pid) {
2214 return Err(format!("process {} is already being moved", pid.0));
2215 }
2216 drop(st);
2217 self.reachable_links(pid)
2218 }
2219
2220 fn reachable_links(&self, pid: Pid) -> Result<(), String> {
2221 let st = self.st();
2222 let deaf: Vec<String> = st
2223 .info
2224 .get(&pid)
2225 .map(|i| i.links.iter())
2226 .into_iter()
2227 .flatten()
2228 .filter(|l| !st.info.get(l).is_some_and(|i| i.trap_exit))
2229 .map(|l| l.0.to_string())
2230 .collect();
2231 if deaf.is_empty() {
2232 return Ok(());
2233 }
2234 Err(format!(
2235 "process {} is linked to {}, which does not trap exits; a link cannot follow a \
2236 process to another node, so evict breaks it and tells each partner [:exit, {}, \
2237 :{EVICTED}], which a partner hears only when it traps exits: unlink it first, or \
2238 trap exits in the partner",
2239 pid.0,
2240 deaf.join(", "),
2241 pid.0
2242 ))
2243 }
2244
2245 fn evict(
2246 &self,
2247 pid: Pid,
2248 wire: &Value,
2249 peer: &Value,
2250 loopback: bool,
2251 span: Span,
2252 ) -> Result<Value, EvalError> {
2253 let word = "evict";
2254 let carrier = self.carrier(word, span)?;
2255 self.evictable(pid).map_err(|m| refuse(word, m, span))?;
2256 self.st().evicting.insert(pid);
2257 let parked = self.drive(
2258 |st| {
2259 let info = st.info.get(&pid).filter(|i| i.live);
2260 let boundary = match (info.map(|i| &i.kind), st.waits.get(&pid)) {
2261 (None, _) => return Some(false),
2262 (Some(Kind::Server), Some(Wait::ServerNext))
2263 | (Some(Kind::Plain), Some(Wait::Receive { .. })) => true,
2264 _ => false,
2265 };
2266 (boundary && !st.aloft.contains(&pid)).then_some(true)
2267 },
2268 None,
2269 word,
2270 span,
2271 );
2272 let moved = match parked {
2273 Ok(Driven::Done(true)) => self.depart_over(pid, wire, peer, loopback, carrier),
2274 Ok(Driven::Done(false)) => Err(format!(
2275 "process {} exited before it reached a message boundary",
2276 pid.0
2277 )),
2278 Ok(Driven::Deadlock(_) | Driven::TimedOut) => Err(format!(
2279 "process {} is waiting {} and nothing can wake it, so it never reached a message \
2280 boundary",
2281 pid.0,
2282 match self.st().info.get(&pid).map(|i| &i.kind) {
2283 Some(Kind::Plain) => "outside receive",
2284 _ => "inside its handler",
2285 }
2286 )),
2287 Err(e) => {
2288 self.st().evicting.remove(&pid);
2289 return Err(e);
2290 }
2291 };
2292 self.st().evicting.remove(&pid);
2293 moved.map_err(|m| refuse(word, m, span))
2294 }
2295
2296 fn image_of(&self, pid: Pid) -> Result<Image, String> {
2297 let state = lock(&self.sched)
2298 .iter()
2299 .find(|p| p.pid == pid)
2300 .and_then(|p| p.interp.lookup_global("%state"))
2301 .unwrap_or(Value::Nil);
2302 let state = copy_data(&state).map_err(|kind| {
2303 format!(
2304 "process {}'s state holds a {kind}, which is not data and does not cross a wire",
2305 pid.0
2306 )
2307 })?;
2308 let st = self.st();
2309 let info = st.info.get(&pid).ok_or("the process is gone")?;
2310 let plain = matches!(info.kind, Kind::Plain);
2311 let mut code = info.code.clone();
2312 if !plain {
2313 code.bind("%state", state);
2314 }
2315 let supervision = match info.sup.and_then(|s| st.info.get(&s)).map(|s| &s.kind) {
2316 Some(Kind::Supervisor(spec)) => Some(Supervision {
2317 max_restarts: spec.max_restarts,
2318 restarts: info.restarts,
2319 initial: info.initial.clone().unwrap_or(Value::Nil),
2320 }),
2321 _ => None,
2322 };
2323 let tape = st
2324 .tapes
2325 .get(&pid)
2326 .filter(|t| plain && !t.truncated)
2327 .map(|t| t.entries.clone());
2328 let input_root = (!plain && self.config.frame.is_some())
2329 .then(|| {
2330 st.accounts
2331 .get(&pid)
2332 .and_then(|b| b.input_root.clone())
2333 .or_else(|| {
2334 self.config
2335 .state_root
2336 .zip(info.entered.as_ref())
2337 .map(|(root, entered)| root(entered))
2338 })
2339 })
2340 .flatten();
2341 let carries = if plain {
2342 tape.is_some()
2343 } else {
2344 input_root.is_some()
2345 };
2346 let account = self
2347 .config
2348 .frame
2349 .as_ref()
2350 .filter(|_| carries)
2351 .map(|_| Account {
2352 host_io: self.config.host_io.of(pid),
2353 journal: self.config.journals().then(|| {
2354 st.journals
2355 .get(&pid)
2356 .map(|r| r.iter().cloned().collect())
2357 .unwrap_or_default()
2358 }),
2359 input_root,
2360 });
2361 let mut image = Image {
2362 definition: info.def_name.clone(),
2363 scheduling: info.scheduling,
2364 trap_exit: info.trap_exit,
2365 code,
2366 mailbox: st
2367 .boxes
2368 .get(&pid)
2369 .map(|b| b.iter().cloned().collect())
2370 .unwrap_or_default(),
2371 supervision,
2372 dispatch_steps: info.asked,
2373 parked: None,
2374 tape,
2375 account,
2376 };
2377 drop(st);
2378 if plain {
2379 image.parked = Some(self.continuation(pid)?);
2380 }
2381 Ok(image)
2382 }
2383
2384 fn continuation(&self, pid: Pid) -> Result<Continuation, String> {
2385 let sched = lock(&self.sched);
2386 let p = sched
2387 .iter()
2388 .find(|p| p.pid == pid)
2389 .ok_or("the process is gone")?;
2390 let mut st = self.st();
2391 let now = st.clock.now();
2392 let wait_ms = match st.waits.get(&pid) {
2393 Some(Wait::Receive {
2394 deadline: Some(d), ..
2395 }) => Some(d.saturating_sub(now)),
2396 _ => None,
2397 };
2398 let info = st.info.get(&pid).ok_or("the process is gone")?;
2399 let cannot = |why: String| format!("process {} cannot move: {why}", pid.0);
2400 if !info
2401 .chunk
2402 .as_ref()
2403 .is_some_and(|c| Arc::ptr_eq(c, &p.chunk))
2404 {
2405 return Err(cannot(
2406 "a migration replaced its code, and its frames still run the code it had \
2407 before, which its image does not carry; let it restart or finish first"
2408 .to_string(),
2409 ));
2410 }
2411 let vm = p.vm.parked(&p.chunk).map_err(|e| cannot(parking(&e)))?;
2412 let functions = info
2413 .code
2414 .defs
2415 .iter()
2416 .filter_map(defined_key)
2417 .map(|name| {
2418 let index = match p.interp.lookup_global(name) {
2419 Some(Value::Foreign(any)) => any
2420 .downcast_ref::<CompiledClosure>()
2421 .filter(|c| {
2422 c.captures.is_empty()
2423 && (Arc::ptr_eq(&c.chunk, &p.chunk)
2424 || c.chunk.fingerprint() == p.chunk.fingerprint())
2425 })
2426 .map(|c| c.fn_index),
2427 _ => None,
2428 };
2429 index.map(|i| (name.to_string(), i)).ok_or_else(|| {
2430 cannot(format!(
2431 "its global `{name}` is not the function its code defines"
2432 ))
2433 })
2434 })
2435 .collect::<Result<_, _>>()?;
2436 Ok(Continuation {
2437 vm,
2438 functions,
2439 wait_ms,
2440 })
2441 }
2442
2443 fn resume(&self, pid: Pid, c: &Continuation, tape: Tape) -> Result<(), String> {
2444 let mut st = self.st();
2445 let p = st
2446 .pending
2447 .iter_mut()
2448 .find(|p| p.pid == pid)
2449 .ok_or("the process is gone")?;
2450 for (name, index) in &c.functions {
2451 let body = FnRef::Table(*index)
2452 .resolve(&p.chunk)
2453 .filter(|f| f.captures.is_empty())
2454 .ok_or_else(|| {
2455 format!(
2456 "its function `{name}` names entry {index}, which its code does not define"
2457 )
2458 })?;
2459 let function = CompiledClosure {
2460 body: Arc::new(body.clone()),
2461 captures: Vec::new(),
2462 chunk: Arc::clone(&p.chunk),
2463 globals: p.interp.globals_snapshot().clone(),
2464 fn_index: *index,
2465 };
2466 p.interp
2467 .define_global(name.as_str(), Value::Foreign(Arc::new(function)));
2468 }
2469 p.vm = Vm::from_parked(&c.vm, &p.chunk, &p.interp, self.config.budget)
2470 .map_err(|e| parking(&e))?;
2471 if let Some(ms) = c.wait_ms {
2472 let deadline = st.clock.now().saturating_add(ms);
2473 st.waits.insert(
2474 pid,
2475 Wait::Receive {
2476 pattern: kw(pattern::WILDCARD),
2477 deadline: Some(deadline),
2478 },
2479 );
2480 }
2481 st.tapes.insert(pid, tape);
2482 Ok(())
2483 }
2484
2485 fn depart_over(
2486 &self,
2487 pid: Pid,
2488 wire: &Value,
2489 peer: &Value,
2490 loopback: bool,
2491 carrier: Carrier,
2492 ) -> Result<Value, String> {
2493 let image = self.image_of(pid)?;
2494 let queue = |image: &Image| -> Result<(String, usize, usize), String> {
2495 let text = image.text();
2496 let id = image::id_of(&text);
2497 let frames = image::frames(&id, &text, carrier.room)?;
2498 self.reachable_links(pid)?;
2499 let bytes = (carrier.queue)(wire, peer, &frames, loopback)?;
2500 Ok((id, frames.len(), bytes))
2501 };
2502 let (id, frames, bytes) = match queue(&image) {
2503 Err(_) if image.tape.is_some() || image.account.is_some() => queue(&Image {
2504 tape: None,
2505 account: None,
2506 ..image
2507 }),
2508 queued => queued,
2509 }?;
2510 self.depart(pid);
2511 Ok(record(&[
2512 ("image", Value::Str(Arc::from(format!("b3:{id}")))),
2513 ("frames", int(frames)),
2514 ("bytes", int(bytes)),
2515 ]))
2516 }
2517
2518 fn depart(&self, pid: Pid) {
2519 lock(&self.sched).retain(|p| p.pid != pid);
2520 lock(&self.lanes).remove(&pid);
2521 let mut st = self.st();
2522 let (monitors, links, sup) = st
2523 .info
2524 .get_mut(&pid)
2525 .map(|i| {
2526 (
2527 std::mem::take(&mut i.monitors),
2528 std::mem::take(&mut i.links),
2529 i.sup,
2530 )
2531 })
2532 .unwrap_or_default();
2533 for (r, watcher) in monitors {
2534 st.deliver(
2535 watcher,
2536 Value::list([kw("down"), int(r), int(pid.0), kw(EVICTED)]),
2537 );
2538 }
2539 for l in links {
2540 let trapping = st.info.get_mut(&l).is_some_and(|i| {
2541 i.links.remove(&pid);
2542 i.trap_exit
2543 });
2544 if trapping {
2545 st.deliver(l, Value::list([kw("exit"), int(pid.0), kw(EVICTED)]));
2546 }
2547 }
2548 if let Some(s) = sup {
2549 st.drop_child(s, pid);
2550 }
2551 st.forget(pid);
2552 drop(st);
2553 self.config.host_io.take(pid);
2554 }
2555
2556 fn accept(
2557 &self,
2558 wire: &Value,
2559 timeout: Duration,
2560 loopback: bool,
2561 span: Span,
2562 ) -> Result<Value, EvalError> {
2563 let word = "accept";
2564 let carrier = self.carrier(word, span)?;
2565 let deadline = Instant::now() + timeout;
2566 loop {
2567 let left = deadline.saturating_duration_since(Instant::now());
2568 let frames = (carrier.take)(wire, image::FRAME, left, loopback)
2569 .map_err(|m| refuse(word, m, span))?;
2570 let (done, flaw) = {
2571 let mut st = self.st();
2572 let mut flaw = None;
2573 for f in &frames {
2574 if let Err(why) = st.arrivals.file(f) {
2575 flaw.get_or_insert(why);
2576 }
2577 }
2578 (st.arrivals.complete(), flaw)
2579 };
2580 match (done, flaw) {
2581 (Some(Ok(text)), _) => return self.admit(&text, span),
2582 (Some(Err(why)), _) | (None, Some(why)) => {
2583 return Err(raised("decode_error", why, Value::Nil, span))
2584 }
2585 (None, None) if frames.is_empty() => return Ok(Value::Nil),
2586 (None, None) => {}
2587 }
2588 }
2589 }
2590
2591 fn admit(&self, text: &str, span: Span) -> Result<Value, EvalError> {
2592 let (parsed, _) = self.apart(|| Image::parse(text));
2593 let mut image = parsed.map_err(|why| raised("decode_error", why, Value::Nil, span))?;
2594 let parked = image.parked.take();
2595 let id = image::id_of(text);
2596 let tape = Tape::arrived(image.tape.take(), &id);
2597 let account = image.account.take();
2598 let state = bound_state(&image.code);
2599 let handed = image.supervision.take();
2600 if let (Some(s), None) = (&handed, &parked) {
2601 image.code.bind("%state", s.initial.clone());
2602 }
2603 let sup = handed
2604 .as_ref()
2605 .map(|s| self.st().supervisor(Strategy::OneForOne, s.max_restarts));
2606 let started = self.start(
2607 image.code,
2608 if parked.is_some() {
2609 Kind::Plain
2610 } else {
2611 Kind::Server
2612 },
2613 image.definition,
2614 sup,
2615 image.scheduling,
2616 );
2617 let resumed = started.and_then(|pid| match &parked {
2618 Some(c) => self
2619 .apart(|| self.resume(pid, c, tape))
2620 .0
2621 .map(|()| pid)
2622 .map_err(|why| {
2623 let mut st = self.st();
2624 st.pending.retain(|p| p.pid != pid);
2625 st.forget(pid);
2626 format!("the image cannot resume here: {why}")
2627 }),
2628 None => Ok(pid),
2629 });
2630 let pid = match resumed {
2631 Ok(pid) => pid,
2632 Err(m) => {
2633 if let Some(sup) = sup {
2634 self.st().forget(sup);
2635 }
2636 return Err(refuse("accept", m, span));
2637 }
2638 };
2639 if let (Some(sup), Some(s)) = (sup, &handed) {
2640 let mut st = self.st();
2641 st.enrol(sup, pid);
2642 if parked.is_none() {
2643 if let Some(p) = st.pending.iter_mut().find(|p| p.pid == pid) {
2644 p.interp.define_global("%state", state.clone());
2645 }
2646 }
2647 if let Some(i) = st.info.get_mut(&pid) {
2648 i.restarts = s.restarts;
2649 if parked.is_none() {
2650 i.snapshot = Some(state.clone());
2651 i.entered = Some(state);
2652 }
2653 }
2654 }
2655 let (mailbox, bytes) = self.apart(|| {
2656 image
2657 .mailbox
2658 .iter()
2659 .filter_map(|m| copy_data(m).ok())
2660 .collect::<VecDeque<_>>()
2661 });
2662 {
2663 let mut st = self.st();
2664 if let Some(i) = st.info.get_mut(&pid) {
2665 i.trap_exit = image.trap_exit;
2666 i.ask_dispatch(image.dispatch_steps);
2667 }
2668 st.boxes.insert(pid, mailbox);
2669 }
2670 self.credit(pid, bytes);
2671 if let Some(a) = account {
2672 self.brought(pid, a, &id);
2673 }
2674 Ok(int(pid.0))
2675 }
2676
2677 fn brought(&self, pid: Pid, account: Account, image: &str) {
2678 if self.config.frame.is_none() {
2679 return;
2680 }
2681 let journal: Option<Vec<Value>> =
2682 account.journal.filter(|_| self.config.journals()).map(|j| {
2683 let skip = j.len().saturating_sub(self.config.journal);
2684 j.into_iter().skip(skip).collect()
2685 });
2686 let mark = CarriedAccount::of(journal.as_deref(), account.host_io, image);
2687 self.config.host_io.carry(pid, account.host_io);
2688 let mut st = self.st();
2689 if let Some(j) = journal {
2690 st.journals.insert(pid, j.into());
2691 }
2692 st.accounts.insert(
2693 pid,
2694 Brought {
2695 mark,
2696 input_root: account.input_root,
2697 },
2698 );
2699 }
2700
2701 fn trace_of(&self, st: &mut NodeState, pid: Pid, reason: &ExitReason, value: &Value) -> Trace {
2704 let info = st.info.get(&pid).expect("a live process");
2705 let mut code = info.code.clone();
2706 if let (Kind::Server, Some(s)) = (&info.kind, &info.snapshot) {
2707 code.bind("%state", s.clone());
2708 }
2709 let tape = st.tapes.get(&pid).cloned().unwrap_or_default();
2710 let bound = self.config.journal;
2711 let mailbox = self.config.journals().then(|| {
2712 let waiting = st.boxes.get(&pid);
2713 Journal {
2714 bound,
2715 recent: st
2716 .journals
2717 .get(&pid)
2718 .map(|r| r.iter().cloned().collect())
2719 .unwrap_or_default(),
2720 unconsumed: waiting
2721 .map(|b| b.iter().take(bound).cloned().collect())
2722 .unwrap_or_default(),
2723 unconsumed_total: waiting.map_or(0, VecDeque::len),
2724 }
2725 });
2726 let reason_kind = crate::trace::reason_kind(reason).to_string();
2727 let places = &self.config.places;
2728 let violation = Violation::of(&reason_kind, &tape.entries, &code.defs, places);
2729 let defined_at = info.def_name.as_deref().and_then(|d| {
2730 let form = code.defs.iter().find(|f| defined_key(f) == Some(d))?;
2731 places.locate(d, form.span)
2732 });
2733 Trace {
2734 pid: pid.0,
2735 incarnation: info.restarts,
2736 definition: info.def_name.clone(),
2737 def_hash: info
2738 .def_name
2739 .as_deref()
2740 .and_then(|d| self.config.identities.of(d)),
2741 hold: self.config.frame.clone(),
2742 host_io: self
2743 .config
2744 .frame
2745 .as_ref()
2746 .map(|_| self.config.host_io.of(pid)),
2747 mailbox,
2748 violation,
2749 defined_at,
2750 reason: value.clone(),
2751 reason_text: crate::trace::reason_text(reason),
2752 reason_kind,
2753 budget: self.config.budget,
2754 code,
2755 carried: tape.carried,
2756 carried_account: st.accounts.get(&pid).map(|b| b.mark.clone()),
2757 tape: tape.entries,
2758 truncated: tape.truncated,
2759 schedule: st.schedule.iter().cloned().collect(),
2760 clock: self.config.clock,
2761 }
2762 }
2763
2764 fn receipt_of(
2767 &self,
2768 st: &NodeState,
2769 sched: &[Proc<()>],
2770 pid: Pid,
2771 reason: &ExitReason,
2772 ) -> Option<Receipt> {
2773 let host_io = self.config.host_io.take(pid);
2774 let frame = self.config.frame.clone()?;
2775 self.config.receipts.as_ref()?;
2776 let info = st.info.get(&pid)?;
2777 let proc = sched.iter().find(|p| p.pid == pid);
2778 let steps = proc
2779 .filter(|_| self.config.count_steps)
2780 .map(|p| u64::try_from(p.vm.burned()).unwrap_or(u64::MAX));
2781 let brought = st.accounts.get(&pid);
2782 let roots = match (&info.kind, self.config.state_root, &info.entered) {
2783 (Kind::Server, Some(root), Some(entered)) => {
2784 let left = proc
2785 .and_then(|p| p.interp.lookup_global("%state"))
2786 .or_else(|| info.snapshot.clone())
2787 .unwrap_or(Value::Nil);
2788 Some(Roots {
2789 input: brought
2790 .and_then(|b| b.input_root.clone())
2791 .unwrap_or_else(|| root(entered)),
2792 output: root(&left),
2793 })
2794 }
2795 _ => None,
2796 };
2797 let carried = brought.map(|b| CarriedIo {
2798 image: b.mark.image.clone(),
2799 host_io: b.mark.host_io,
2800 input_root: b.input_root.clone().filter(|_| roots.is_some()),
2801 });
2802 Some(Receipt {
2803 subject: Subject::Actor {
2804 pid,
2805 incarnation: info.restarts,
2806 },
2807 definition: info.def_name.clone(),
2808 def_hash: info
2809 .def_name
2810 .as_deref()
2811 .and_then(|d| self.config.identities.of(d)),
2812 frame,
2813 host_io,
2814 exit_kind: crate::trace::reason_kind(reason).to_string(),
2815 exit_text: crate::trace::reason_text(reason),
2816 roots,
2817 steps,
2818 carried,
2819 })
2820 }
2821
2822 fn keep_trace(&self, trace: Trace, write: bool) {
2823 if let Some(dir) = self.config.trace_dir.as_ref().filter(|_| write) {
2824 match trace.write_to(dir) {
2825 Ok(path) => eprintln!(
2826 "blue: process {} exited ({}); trace: {}",
2827 trace.pid,
2828 trace.reason_text,
2829 path.display()
2830 ),
2831 Err(e) => eprintln!(
2832 "blue: process {} exited; its trace could not be written: {e}",
2833 trace.pid
2834 ),
2835 }
2836 }
2837 let mut st = self.st();
2838 if st.traces.len() == KEPT_TRACES {
2839 st.traces.pop_front();
2840 }
2841 st.traces.push_back(trace);
2842 }
2843
2844 #[must_use]
2846 pub fn traces(&self) -> Vec<Trace> {
2847 self.st().traces.iter().cloned().collect()
2848 }
2849
2850 fn drive<T>(
2859 &self,
2860 goal: impl FnMut(&mut NodeState) -> Option<T>,
2861 deadline: Option<u64>,
2862 word: &str,
2863 span: Span,
2864 ) -> Result<Driven<T>, EvalError> {
2865 let driving = self.driving();
2866 let (driven, _) = self.apart(|| self.drive_unmetered(goal, deadline, word, span));
2867 drop(driving);
2868 self.settle_root();
2869 driven
2870 }
2871
2872 fn drive_unmetered<T>(
2873 &self,
2874 mut goal: impl FnMut(&mut NodeState) -> Option<T>,
2875 deadline: Option<u64>,
2876 word: &str,
2877 span: Span,
2878 ) -> Result<Driven<T>, EvalError> {
2879 let mut sched = lock(&self.sched);
2880 loop {
2881 self.adopt(&mut sched);
2882 self.process_signals(&mut sched);
2883 {
2884 let mut st = self.st();
2885 if let Some((from, reason)) = st.root_signal.take() {
2886 return Err(refuse(
2887 word,
2888 format!("linked process {} exited: {}", from.0, render(&reason)),
2889 span,
2890 ));
2891 }
2892 if let Some(t) = goal(&mut st) {
2893 return Ok(Driven::Done(t));
2894 }
2895 if deadline.is_some_and(|d| st.clock.now() >= d) {
2896 return Ok(Driven::TimedOut);
2897 }
2898 }
2899 if self.apply_migrations(&mut sched) {
2900 continue;
2901 }
2902 if self.land(&mut sched) {
2903 continue;
2904 }
2905 if self.has_work(&sched) {
2906 self.round(&mut sched);
2907 continue;
2908 }
2909 let next = {
2912 let st = self.st();
2913 match (st.earliest_deadline(), deadline) {
2914 (Some(a), Some(b)) => Some(a.min(b)),
2915 (a, b) => a.or(b),
2916 }
2917 };
2918 let aloft = {
2919 let mut st = self.st();
2920 let now = st.clock.now();
2921 (!st.aloft.is_empty()).then(|| match st.clock.kind {
2922 ClockKind::Real => next.map(|t| Duration::from_millis(t.saturating_sub(now))),
2923 ClockKind::Virtual => None,
2924 })
2925 };
2926 if let Some(limit) = aloft {
2927 self.await_flight(&mut sched, limit);
2928 continue;
2929 }
2930 match next {
2931 Some(t) => {
2932 let pause = self.st().clock.advance_to(t);
2933 if let Some(d) = pause {
2934 std::thread::sleep(d);
2935 }
2936 }
2937 None => {
2938 if self.fail_stuck_migrations() {
2939 continue;
2940 }
2941 let blocked = sched
2942 .iter()
2943 .filter(|p| p.state.is_blocked())
2944 .map(|p| p.pid)
2945 .collect();
2946 return Ok(Driven::Deadlock(blocked));
2947 }
2948 }
2949 }
2950 }
2951
2952 fn fail_stuck_migrations(&self) -> bool {
2955 let mut st = self.st();
2956 let stuck: Vec<(Pid, Migration)> = std::mem::take(&mut st.migrations).into_iter().collect();
2957 let any = !stuck.is_empty();
2958 for (pid, m) in stuck {
2959 st.outcomes.entry(m.request).or_default().push((
2960 pid,
2961 Err("the process is waiting inside its handler and nothing can wake it, so it never \
2962 reached a boundary"
2963 .into()),
2964 ));
2965 }
2966 any
2967 }
2968
2969 fn receive_root(&self, args: &[Value], span: Span) -> Result<Value, EvalError> {
2970 let (pattern, timeout, default) = receive_args(args, span)?;
2971 let deadline = timeout.map(|t| self.st().clock.now().saturating_add(t));
2972 let driven = self.drive(
2973 |st| {
2974 let b = st.boxes.get_mut(&ROOT)?;
2975 let i = pattern::first_match(b, &pattern)?;
2976 b.remove(i)
2977 },
2978 deadline,
2979 "receive",
2980 span,
2981 )?;
2982 match driven {
2983 Driven::Done(msg) => Ok(msg),
2984 Driven::TimedOut => Ok(default),
2985 Driven::Deadlock(blocked) => Err(refuse(
2986 "receive",
2987 format!(
2988 "would wait forever: no process can send (deadlock; waiting: {})",
2989 blocked
2990 .iter()
2991 .map(|p| p.0.to_string())
2992 .collect::<Vec<_>>()
2993 .join(", ")
2994 ),
2995 span,
2996 )),
2997 }
2998 }
2999
3000 pub fn run_until_idle(&self) {
3003 let _ = self.drive(|_| None::<()>, None, "run", Span::synthetic());
3004 }
3005
3006 #[must_use]
3010 pub fn process_count(&self) -> usize {
3011 let sched = lock(&self.sched).len();
3012 let st = self.st();
3013 sched + st.pending.len() + st.aloft.len()
3014 }
3015
3016 #[must_use]
3018 pub fn mailbox(&self, pid: Pid) -> Vec<Value> {
3019 self.st()
3020 .boxes
3021 .get(&pid)
3022 .map(|b| b.iter().cloned().collect())
3023 .unwrap_or_default()
3024 }
3025}
3026
3027fn bound_state(code: &Code) -> Value {
3030 code.bindings
3031 .iter()
3032 .find(|(n, _)| &**n == "%state")
3033 .map_or(Value::Nil, |(_, v)| v.clone())
3034}
3035
3036fn record(fields: &[(&str, Value)]) -> Value {
3037 Value::map(
3038 fields
3039 .iter()
3040 .map(|(k, v)| (MapKey::Keyword(Arc::from(*k)), v.clone())),
3041 )
3042}
3043
3044const DISPATCH_OPTION: &str = "dispatch_steps";
3045
3046const SPAWN_OPTIONS: &[&str] = &[scheduling::OPTION];
3047
3048const SERVER_OPTIONS: &[&str] = &[scheduling::OPTION, DISPATCH_OPTION];
3049
3050#[derive(Clone, Copy, Debug, Default)]
3051struct Opts {
3052 scheduling: Scheduling,
3053 dispatch_steps: Option<usize>,
3054}
3055
3056fn dispatch_steps(v: &Value, span: Span) -> Result<usize, EvalError> {
3057 match v {
3058 Value::Int(n) if *n > 0 => Ok(usize::try_from(*n).unwrap_or(usize::MAX)),
3059 other => Err(raised(
3060 "bad_option",
3061 format!(
3062 "{DISPATCH_OPTION} is a positive integer, got {}",
3063 match other {
3064 Value::Int(n) => n.to_string(),
3065 other => other.type_name().to_string(),
3066 }
3067 ),
3068 Value::Nil,
3069 span,
3070 )),
3071 }
3072}
3073
3074fn spawn_opts(
3075 v: Option<&Value>,
3076 word: &str,
3077 takes: &[&str],
3078 span: Span,
3079) -> Result<Opts, EvalError> {
3080 let m = match v {
3081 None | Some(Value::Nil) => return Ok(Opts::default()),
3082 Some(Value::Map(m)) => m,
3083 Some(other) => {
3084 return Err(raised(
3085 "bad_option",
3086 format!(
3087 "{word}'s options are a map {{{}: …}}, got {}",
3088 scheduling::OPTION,
3089 other.type_name()
3090 ),
3091 Value::Nil,
3092 span,
3093 ))
3094 }
3095 };
3096 let mut opts = Opts::default();
3097 for (k, val) in m.iter() {
3098 let key = match k {
3099 MapKey::Keyword(s) | MapKey::Symbol(s) | MapKey::Str(s) => {
3100 s.trim_start_matches(':').to_string()
3101 }
3102 other => other.to_value().to_string(),
3103 };
3104 if !takes.contains(&key.as_str()) {
3105 let named: Vec<String> = takes.iter().map(|t| format!("{t}:")).collect();
3106 return Err(raised(
3107 "bad_option",
3108 format!("{word} takes {}, got {key}:", named.join(" or ")),
3109 Value::Nil,
3110 span,
3111 ));
3112 }
3113 if key == DISPATCH_OPTION {
3114 opts.dispatch_steps = Some(dispatch_steps(val, span)?);
3115 continue;
3116 }
3117 opts.scheduling = match val {
3118 Value::Keyword(w) => Scheduling::named(w),
3119 other => Err(scheduling::Refusal::Unknown(format!(
3120 "a {}",
3121 other.type_name()
3122 ))),
3123 }
3124 .map_err(|r| raised(r.kind(), r.to_string(), Value::Nil, span))?;
3125 }
3126 Ok(opts)
3127}
3128
3129fn spawn_args(args: &[Value], word: &str, span: Span) -> Result<Vec<Value>, EvalError> {
3130 match args.get(1) {
3131 None | Some(Value::Nil) => Ok(Vec::new()),
3132 Some(Value::List(l)) => Ok(l.iter().cloned().collect()),
3133 Some(other) => Err(refuse(
3134 word,
3135 format!("the arguments are a list, got {}", other.type_name()),
3136 span,
3137 )),
3138 }
3139}
3140
3141fn spawn_one(
3142 node: &Node,
3143 f: &Value,
3144 args: &[Value],
3145 sup: Option<Pid>,
3146 word: &str,
3147 span: Span,
3148 scheduling: Scheduling,
3149) -> Result<Pid, EvalError> {
3150 let (code, entry) = code::spawn_code(f, args).map_err(|m| refuse(word, m, span))?;
3151 let def_name = match entry {
3152 Entry::Named(n) | Entry::Builtin(n) => Some(n),
3153 Entry::Lambda(_) => None,
3154 };
3155 node.start(code, Kind::Plain, def_name, sup, scheduling)
3156 .map_err(|m| refuse(word, m, span))
3157}
3158
3159fn w_spawn(node: &Node, _who: Who, args: &[Value], span: Span) -> Result<Value, EvalError> {
3160 let a = spawn_args(args, "spawn", span)?;
3161 let class = spawn_opts(args.get(2), "spawn", SPAWN_OPTIONS, span)?.scheduling;
3162 Ok(int(spawn_one(
3163 node, &args[0], &a, None, "spawn", span, class,
3164 )?
3165 .0))
3166}
3167
3168fn w_spawn_link(node: &Node, who: Who, args: &[Value], span: Span) -> Result<Value, EvalError> {
3169 let a = spawn_args(args, "spawn_link", span)?;
3170 let class = spawn_opts(args.get(2), "spawn_link", SPAWN_OPTIONS, span)?.scheduling;
3171 let pid = spawn_one(node, &args[0], &a, None, "spawn_link", span, class)?;
3172 let mut st = node.st();
3173 let me = who.pid();
3174 if let Some(i) = st.info.get_mut(&me) {
3175 i.links.insert(pid);
3176 }
3177 if let Some(i) = st.info.get_mut(&pid) {
3178 i.links.insert(me);
3179 }
3180 Ok(int(pid.0))
3181}
3182
3183fn w_def_hash(node: &Node, _who: Who, args: &[Value], span: Span) -> Result<Value, EvalError> {
3186 let f = &args[0];
3187 if !code::is_function(f) {
3188 return Err(refuse(
3189 "def_hash",
3190 format!("expected a function, got {}", f.type_name()),
3191 span,
3192 ));
3193 }
3194 Ok(code::name_of(f)
3195 .and_then(|key| node.config.identities.of(&key))
3196 .map_or(Value::Nil, |hash| Value::Str(Arc::from(hash))))
3197}
3198
3199fn w_spawn_server(node: &Node, _who: Who, args: &[Value], span: Span) -> Result<Value, EvalError> {
3200 let opts = spawn_opts(args.get(2), "spawn_server", SERVER_OPTIONS, span)?;
3201 Ok(int(server(
3202 node,
3203 &args[0],
3204 &args[1],
3205 None,
3206 "spawn_server",
3207 span,
3208 opts,
3209 )?
3210 .0))
3211}
3212
3213fn server(
3214 node: &Node,
3215 handler: &Value,
3216 state: &Value,
3217 sup: Option<Pid>,
3218 word: &str,
3219 span: Span,
3220 opts: Opts,
3221) -> Result<Pid, EvalError> {
3222 let mut code = code::function_code(handler, "%handler").map_err(|m| refuse(word, m, span))?;
3223 code.defs.extend(SERVER_LOOP.iter().cloned());
3224 code.start = Some(code::call(code::symbol("%serve"), &[]));
3225 let state = copy_data(state).map_err(|kind| {
3226 refuse(
3227 word,
3228 format!("a server's state is copied into the process, and a {kind} is not data"),
3229 span,
3230 )
3231 })?;
3232 code.bind("%state", state);
3233 let pid = node
3234 .start(
3235 code,
3236 Kind::Server,
3237 code::name_of(handler),
3238 sup,
3239 opts.scheduling,
3240 )
3241 .map_err(|m| refuse(word, m, span))?;
3242 if let Some(info) = node.st().info.get_mut(&pid) {
3243 info.ask_dispatch(opts.dispatch_steps);
3244 }
3245 Ok(pid)
3246}
3247
3248fn w_evict(node: &Node, who: Who, args: &[Value], span: Span) -> Result<Value, EvalError> {
3249 node.tally();
3250 evicting(node, who, args, false, span)
3251}
3252
3253fn w_evict_provided(node: &Node, who: Who, args: &[Value], span: Span) -> Result<Value, EvalError> {
3254 evicting(node, who, args, true, span)
3255}
3256
3257fn evicting(
3258 node: &Node,
3259 who: Who,
3260 args: &[Value],
3261 loopback: bool,
3262 span: Span,
3263) -> Result<Value, EvalError> {
3264 if who != Who::Root {
3265 return Err(refuse(
3266 "evict",
3267 "evict waits for the process to reach a message boundary, and only the program \
3268 waits; a process asks the program to evict",
3269 span,
3270 ));
3271 }
3272 let pid = pid_arg(&args[0], "evict", span)?;
3273 node.evict(pid, &args[1], &args[2], loopback, span)
3274}
3275
3276fn w_accept(node: &Node, _who: Who, args: &[Value], span: Span) -> Result<Value, EvalError> {
3277 node.tally();
3278 accepting(node, args, false, span)
3279}
3280
3281fn w_accept_provided(
3282 node: &Node,
3283 _who: Who,
3284 args: &[Value],
3285 span: Span,
3286) -> Result<Value, EvalError> {
3287 accepting(node, args, true, span)
3288}
3289
3290fn accepting(node: &Node, args: &[Value], loopback: bool, span: Span) -> Result<Value, EvalError> {
3291 let timeout = match &args[1] {
3292 Value::Int(n) if *n >= 0 => Duration::from_millis(n.unsigned_abs()),
3293 other => {
3294 return Err(refuse(
3295 "accept",
3296 format!(
3297 "the timeout is milliseconds (an Int >= 0), got {}",
3298 other.type_name()
3299 ),
3300 span,
3301 ))
3302 }
3303 };
3304 node.accept(&args[0], timeout, loopback, span)
3305}
3306
3307fn w_uncarried(_node: &Node, _who: Who, _args: &[Value], span: Span) -> Result<Value, EvalError> {
3308 Err(refuse(
3309 "evict",
3310 "this node has no wire to carry a process",
3311 span,
3312 ))
3313}
3314
3315fn w_send(node: &Node, _who: Who, args: &[Value], span: Span) -> Result<Value, EvalError> {
3316 let to = pid_arg(&args[0], "send", span)?;
3317 let (copied, bytes) = node.apart(|| {
3318 copy_data(&args[1]).map(|msg| {
3319 node.st().deliver(to, msg);
3322 })
3323 });
3324 copied.map_err(|kind| {
3325 refuse(
3326 "send",
3327 format!("a message is copied to the receiver, and a {kind} is not data"),
3328 span,
3329 )
3330 })?;
3331 node.credit(to, bytes);
3332 Ok(args[1].clone())
3333}
3334
3335fn w_self(_node: &Node, who: Who, _args: &[Value], _span: Span) -> Result<Value, EvalError> {
3336 Ok(int(who.pid().0))
3337}
3338
3339fn w_alive(node: &Node, _who: Who, args: &[Value], span: Span) -> Result<Value, EvalError> {
3340 let pid = pid_arg(&args[0], "alive?", span)?;
3341 Ok(Value::Bool(
3342 node.st().info.get(&pid).is_some_and(|i| i.live),
3343 ))
3344}
3345
3346fn w_link(node: &Node, who: Who, args: &[Value], span: Span) -> Result<Value, EvalError> {
3347 let other = pid_arg(&args[0], "link", span)?;
3348 let me = who.pid();
3349 let mut st = node.st();
3350 if other == me || !st.info.contains_key(&other) {
3351 return Ok(Value::Bool(false));
3352 }
3353 if let Some(i) = st.info.get_mut(&me) {
3354 i.links.insert(other);
3355 }
3356 if let Some(i) = st.info.get_mut(&other) {
3357 i.links.insert(me);
3358 }
3359 Ok(Value::Bool(true))
3360}
3361
3362fn w_unlink(node: &Node, who: Who, args: &[Value], span: Span) -> Result<Value, EvalError> {
3363 let other = pid_arg(&args[0], "unlink", span)?;
3364 let me = who.pid();
3365 let mut st = node.st();
3366 if let Some(i) = st.info.get_mut(&me) {
3367 i.links.remove(&other);
3368 }
3369 if let Some(i) = st.info.get_mut(&other) {
3370 i.links.remove(&me);
3371 }
3372 Ok(Value::Bool(true))
3373}
3374
3375fn w_monitor(node: &Node, who: Who, args: &[Value], span: Span) -> Result<Value, EvalError> {
3376 let target = pid_arg(&args[0], "monitor", span)?;
3377 let me = who.pid();
3378 let mut st = node.st();
3379 let r = st.next_ref;
3380 st.next_ref += 1;
3381 match st.info.get_mut(&target) {
3382 Some(i) => i.monitors.push((r, me)),
3383 None => st.deliver(
3384 me,
3385 Value::list([kw("down"), int(r), int(target.0), kw("noproc")]),
3386 ),
3387 }
3388 Ok(int(r))
3389}
3390
3391fn w_demonitor(node: &Node, _who: Who, args: &[Value], span: Span) -> Result<Value, EvalError> {
3392 let Value::Int(r) = args[0] else {
3393 return Err(refuse(
3394 "demonitor",
3395 "expected a monitor reference (an Int)",
3396 span,
3397 ));
3398 };
3399 let mut st = node.st();
3400 for i in st.info.values_mut() {
3401 i.monitors.retain(|(x, _)| *x != r);
3402 }
3403 Ok(Value::Bool(true))
3404}
3405
3406fn w_exit(node: &Node, who: Who, args: &[Value], span: Span) -> Result<Value, EvalError> {
3407 let to = pid_arg(&args[0], "exit", span)?;
3408 let reason = copy_data(&args[1])
3409 .map_err(|kind| refuse("exit", format!("a reason is data, not a {kind}"), span))?;
3410 let me = who.pid();
3411 if to == me {
3412 return Err(refuse(
3413 "exit",
3414 format!("process {} exits: {}", me.0, render(&reason)),
3415 span,
3416 ));
3417 }
3418 node.st().signals.push_back((me, to, reason));
3419 if who == Who::Root {
3420 let mut sched = lock(&node.sched);
3422 node.adopt(&mut sched);
3423 node.process_signals(&mut sched);
3424 node.reap(&mut sched);
3425 }
3426 Ok(Value::Bool(true))
3427}
3428
3429fn truthy(v: &Value) -> bool {
3430 !matches!(v, Value::Nil | Value::Bool(false))
3431}
3432
3433fn w_trap_exit(node: &Node, who: Who, args: &[Value], _span: Span) -> Result<Value, EvalError> {
3434 let mut st = node.st();
3435 let Some(i) = st.info.get_mut(&who.pid()) else {
3436 return Ok(Value::Bool(false));
3437 };
3438 let previous = i.trap_exit;
3439 i.trap_exit = truthy(&args[0]);
3440 Ok(Value::Bool(previous))
3441}
3442
3443fn w_supervisor(node: &Node, _who: Who, args: &[Value], span: Span) -> Result<Value, EvalError> {
3444 let strategy = match &args[0] {
3445 Value::Keyword(k) => Strategy::from_keyword(k.trim_start_matches(':')),
3446 _ => None,
3447 }
3448 .ok_or_else(|| {
3449 refuse(
3450 "supervisor",
3451 "the strategy is :one_for_one, :one_for_all or :rest_for_one",
3452 span,
3453 )
3454 })?;
3455 let Value::List(specs) = &args[1] else {
3456 return Err(refuse(
3457 "supervisor",
3458 "the children are a list of functions or [f, args] pairs",
3459 span,
3460 ));
3461 };
3462 let max_restarts = match args.get(2) {
3463 None | Some(Value::Nil) => DEFAULT_MAX_RESTARTS,
3464 Some(Value::Int(n)) if *n >= 0 => usize::try_from(*n).unwrap_or(usize::MAX),
3465 Some(other) => {
3466 return Err(refuse(
3467 "supervisor",
3468 format!("max_restarts is an Int >= 0, got {}", other.type_name()),
3469 span,
3470 ))
3471 }
3472 };
3473 let sup = node.st().supervisor(strategy, max_restarts);
3474 for spec in specs.iter() {
3475 let pid = match spec {
3476 Value::List(t) if (3..=4).contains(&t.len()) && is_keyword(&t[0], "server") => server(
3478 node,
3479 &t[1],
3480 &t[2],
3481 Some(sup),
3482 "supervisor",
3483 span,
3484 spawn_opts(t.get(3), "supervisor", SERVER_OPTIONS, span)?,
3485 )?,
3486 Value::List(pair) if pair.len() == 2 && code::is_function(&pair[0]) => {
3487 let Value::List(a) = &pair[1] else {
3488 return Err(refuse(
3489 "supervisor",
3490 "a child [f, args] takes its args as a list",
3491 span,
3492 ));
3493 };
3494 let a: Vec<Value> = a.iter().cloned().collect();
3495 spawn_one(
3496 node,
3497 &pair[0],
3498 &a,
3499 Some(sup),
3500 "supervisor",
3501 span,
3502 Scheduling::Normal,
3503 )?
3504 }
3505 f => spawn_one(
3506 node,
3507 f,
3508 &[],
3509 Some(sup),
3510 "supervisor",
3511 span,
3512 Scheduling::Normal,
3513 )?,
3514 };
3515 node.st().enrol(sup, pid);
3516 }
3517 Ok(int(sup.0))
3518}
3519
3520fn w_children(node: &Node, _who: Who, args: &[Value], span: Span) -> Result<Value, EvalError> {
3521 let sup = pid_arg(&args[0], "children", span)?;
3522 let st = node.st();
3523 match st.info.get(&sup).map(|i| &i.kind) {
3524 Some(Kind::Supervisor(s)) => Ok(Value::list(s.children.iter().map(|p| int(p.0)))),
3525 _ => Ok(Value::list([])),
3526 }
3527}
3528
3529fn w_defmigrate(node: &Node, who: Who, args: &[Value], span: Span) -> Result<Value, EvalError> {
3530 let targets: Vec<Pid> = match &args[0] {
3531 Value::Int(_) => vec![pid_arg(&args[0], "defmigrate", span)?],
3532 f if code::is_function(f) => {
3533 let Some(name) = code::name_of(f) else {
3534 return Err(refuse(
3535 "defmigrate",
3536 "the target function has no name to match processes by",
3537 span,
3538 ));
3539 };
3540 let st = node.st();
3541 st.info
3542 .iter()
3543 .filter(|(_, i)| {
3544 matches!(i.kind, Kind::Plain | Kind::Server)
3545 && i.def_name.as_deref() == Some(&name)
3546 })
3547 .map(|(p, _)| *p)
3548 .collect()
3549 }
3550 other => {
3551 return Err(refuse(
3552 "defmigrate",
3553 format!(
3554 "the target is a pid or the function processes run, got {}",
3555 other.type_name()
3556 ),
3557 span,
3558 ))
3559 }
3560 };
3561 let request = {
3562 let mut st = node.st();
3563 let r = st.next_request;
3564 st.next_request += 1;
3565 r
3566 };
3567 let new_name = code::name_of(&args[1]).unwrap_or_else(|| format!("%migration{request}"));
3568 let new_code =
3569 code::function_code(&args[1], &new_name).map_err(|m| refuse("defmigrate", m, span))?;
3570 let transform = match args.get(2) {
3571 None | Some(Value::Nil) => None,
3572 Some(t) => {
3573 Some(code::function_code(t, "%transform").map_err(|m| refuse("defmigrate", m, span))?)
3574 }
3575 };
3576 let mut answers: Vec<(Pid, Result<(), String>)> = Vec::new();
3577 let mut queued = Vec::new();
3578 {
3579 let mut st = node.st();
3580 for pid in targets {
3581 match st.info.get(&pid).map(|i| &i.kind) {
3582 Some(Kind::Plain | Kind::Server) => {
3583 st.migrations.insert(
3584 pid,
3585 Migration {
3586 request,
3587 code: new_code.clone(),
3588 new_name: new_name.clone(),
3589 transform: transform.clone(),
3590 },
3591 );
3592 queued.push(pid);
3593 }
3594 _ => answers.push((pid, Err("no such process".into()))),
3595 }
3596 }
3597 }
3598 if who != Who::Root {
3599 return Ok(Value::list(
3602 queued.iter().map(|p| Value::list([int(p.0), kw("queued")])),
3603 ));
3604 }
3605 if !queued.is_empty() {
3606 let want = queued.len();
3607 let driven = node.drive(
3608 |st| {
3609 (st.outcomes.get(&request).map_or(0, Vec::len) >= want)
3610 .then(|| st.outcomes.remove(&request).unwrap_or_default())
3611 },
3612 None,
3613 "defmigrate",
3614 span,
3615 )?;
3616 if let Driven::Done(done) = driven {
3617 answers.extend(done);
3618 }
3619 }
3620 answers.sort_by_key(|(p, _)| *p);
3621 Ok(Value::list(answers.into_iter().map(|(p, r)| {
3622 Value::list([
3623 int(p.0),
3624 match r {
3625 Ok(()) => kw("ok"),
3626 Err(m) => Value::list([kw("error"), Value::Str(Arc::from(m.as_str()))]),
3627 },
3628 ])
3629 })))
3630}