1use std::collections::{BTreeMap, HashMap, HashSet};
2use std::path::PathBuf;
3use std::sync::atomic::{AtomicBool, AtomicU64, Ordering};
4use std::sync::Arc;
5use std::time::Instant;
6
7use crate::chunk::{Chunk, ChunkRef, Constant};
8use crate::runtime_limits::RuntimeLimits;
9use crate::value::{
10 ModuleFunctionRegistry, VmAsyncBuiltinFn, VmBuiltinFn, VmClosure, VmEnv, VmError, VmMutex,
11 VmTaskHandle, VmValue,
12};
13use crate::BuiltinId;
14
15use super::debug::DebugHook;
16use super::modules::LoadedModule;
17use super::VmBuiltinMetadata;
18
19pub(crate) struct ResolvedLazyCallable {
33 pub(crate) exports: BTreeMap<String, Arc<VmClosure>>,
34 #[allow(dead_code)]
39 pub(crate) retained_module_graph: Arc<BTreeMap<PathBuf, LoadedModule>>,
40}
41
42pub(crate) type LazyCallableResolution = Arc<ResolvedLazyCallable>;
43pub(crate) type LazyCallableModuleCache =
44 Arc<VmMutex<BTreeMap<PathBuf, Arc<tokio::sync::OnceCell<LazyCallableResolution>>>>>;
45
46pub(crate) struct ScopeSpan(u64);
48
49impl ScopeSpan {
50 pub(crate) fn new(kind: crate::tracing::SpanKind, name: String) -> Self {
51 Self(crate::tracing::span_start(kind, name))
52 }
53}
54
55pub(crate) struct ExecutionDeadlineState {
60 origin: Instant,
61 deadline_offset: AtomicU64,
63 abandoned: AtomicBool,
67}
68
69impl ExecutionDeadlineState {
70 pub(crate) fn new(origin: Instant, deadline: Option<Instant>) -> Arc<Self> {
71 Arc::new(Self {
72 origin,
73 deadline_offset: AtomicU64::new(Self::encode(origin, deadline)),
74 abandoned: AtomicBool::new(false),
75 })
76 }
77
78 #[inline]
79 pub(crate) fn is_active(&self) -> bool {
80 self.deadline_offset.load(Ordering::Acquire) != 0
81 }
82
83 #[inline]
84 pub(crate) fn is_abandoned(&self) -> bool {
85 self.abandoned.load(Ordering::Acquire)
86 }
87
88 pub(crate) fn fork(&self) -> Arc<Self> {
89 let state = Self::new(self.origin, self.current());
90 state
91 .abandoned
92 .store(self.is_abandoned(), Ordering::Release);
93 state
94 }
95
96 pub(crate) fn current(&self) -> Option<Instant> {
97 let encoded = self.deadline_offset.load(Ordering::Acquire);
98 (encoded != 0)
99 .then(|| self.origin + std::time::Duration::from_nanos(encoded.saturating_sub(1)))
100 }
101
102 pub(crate) fn install(self: &Arc<Self>, deadline: Instant) -> ExecutionDeadlineGuard {
103 let previous = self.deadline_offset.load(Ordering::Acquire);
104 let requested = Self::encode(self.origin, Some(deadline));
105 let active = if previous == 0 {
106 requested
107 } else {
108 previous.min(requested)
109 };
110 self.deadline_offset.store(active, Ordering::Release);
111 ExecutionDeadlineGuard {
112 state: Arc::clone(self),
113 previous,
114 completed: false,
115 }
116 }
117
118 fn encode(origin: Instant, deadline: Option<Instant>) -> u64 {
119 deadline.map_or(0, |deadline| {
120 let nanos = deadline.saturating_duration_since(origin).as_nanos();
121 u64::try_from(nanos)
122 .unwrap_or(u64::MAX - 1)
123 .saturating_add(1)
124 })
125 }
126}
127
128pub(crate) struct ExecutionDeadlineGuard {
129 state: Arc<ExecutionDeadlineState>,
130 previous: u64,
131 completed: bool,
132}
133
134impl ExecutionDeadlineGuard {
135 pub(crate) fn complete(mut self) {
138 self.completed = true;
139 }
140}
141
142impl Drop for ExecutionDeadlineGuard {
143 fn drop(&mut self) {
144 self.state
145 .deadline_offset
146 .store(self.previous, Ordering::Release);
147 if !self.completed {
148 self.state.abandoned.store(true, Ordering::Release);
149 crate::orchestration::clear_pipeline_on_finish();
150 }
151 }
152}
153
154impl Drop for ScopeSpan {
155 fn drop(&mut self) {
156 crate::tracing::span_end(self.0);
157 }
158}
159
160#[derive(Clone)]
161pub(crate) struct LocalSlot {
162 pub(crate) value: VmValue,
163 pub(crate) initialized: bool,
164 pub(crate) synced: bool,
165}
166
167impl Drop for LocalSlot {
168 fn drop(&mut self) {
169 if crate::value::recursion::is_recursive_container(&self.value) {
177 crate::value::recursion::dismantle(std::mem::replace(&mut self.value, VmValue::Nil));
178 }
179 }
180}
181
182#[derive(Clone)]
183pub(crate) struct InterruptHandler {
184 pub(crate) handle: i64,
185 pub(crate) signals: Vec<String>,
186 pub(crate) once: bool,
187 pub(crate) graceful_timeout_ms: Option<u64>,
188 pub(crate) handler: VmValue,
189}
190
191pub(crate) struct CallFrame {
193 pub(crate) chunk: ChunkRef,
194 pub(crate) inline_cache_set: usize,
198 pub(crate) ip: usize,
199 pub(crate) stack_base: usize,
200 pub(crate) saved_env: VmEnv,
201 pub(crate) initial_env: Option<VmEnv>,
209 pub(crate) initial_local_slots: Option<Vec<LocalSlot>>,
210 pub(crate) saved_iterator_depth: usize,
212 pub(crate) fn_name: String,
214 pub(crate) argc: usize,
216 pub(crate) saved_source_dir: Option<std::path::PathBuf>,
219 pub(crate) module_functions: Option<ModuleFunctionRegistry>,
221 pub(crate) module_state: Option<crate::value::ModuleState>,
227 pub(crate) local_slots: Vec<LocalSlot>,
229 pub(crate) local_scope_base: usize,
231 pub(crate) local_scope_depth: usize,
233}
234
235pub(crate) struct InlineCacheSite {
236 pub(crate) cache_set: usize,
237 pub(crate) slot_count: usize,
238 pub(crate) slot: Option<usize>,
239}
240
241impl CallFrame {
242 #[inline]
243 pub(crate) fn inline_cache_site_for_previous_op(&self) -> InlineCacheSite {
244 let op_offset = self.ip.saturating_sub(1);
245 InlineCacheSite {
246 cache_set: self.inline_cache_set,
247 slot_count: self.chunk.inline_cache_slot_count(),
248 slot: self.chunk.inline_cache_slot(op_offset),
249 }
250 }
251}
252
253pub(crate) struct ExceptionHandler {
255 pub(crate) catch_ip: usize,
256 pub(crate) stack_depth: usize,
257 pub(crate) frame_depth: usize,
258 pub(crate) env_scope_depth: usize,
259 pub(crate) error_type: Option<crate::value::HarnStr>,
261}
262
263pub(crate) struct TaskScope {
266 pub(crate) task_ids: Vec<String>,
269 pub(crate) frame_depth: usize,
271 pub(crate) env_scope_depth: usize,
273}
274
275pub(crate) enum IterState {
277 Vec {
278 items: Arc<Vec<VmValue>>,
279 idx: usize,
280 },
281 Dict {
282 entries: Arc<crate::value::DictMap>,
283 keys: Vec<String>,
284 idx: usize,
285 },
286 Channel {
287 receiver: std::sync::Arc<tokio::sync::Mutex<tokio::sync::mpsc::Receiver<VmValue>>>,
288 close: std::sync::Arc<crate::value::VmChannelCloseState>,
289 },
290 Generator {
291 gen: Arc<crate::value::VmGenerator>,
292 },
293 Stream {
294 stream: Arc<crate::value::VmStream>,
295 },
296 Range {
300 next: i64,
301 end: i64,
302 inclusive: bool,
303 done: bool,
304 },
305 VmIter {
306 handle: crate::vm::iter::VmIterHandle,
307 },
308}
309
310#[derive(Clone)]
311pub(crate) enum VmBuiltinDispatch {
312 Sync(VmBuiltinFn),
313 Async(VmAsyncBuiltinFn),
314}
315
316#[derive(Clone)]
317pub(crate) struct VmBuiltinEntry {
318 pub(crate) name: Arc<str>,
319 pub(crate) dispatch: VmBuiltinDispatch,
320}
321
322pub struct Vm {
324 pub(crate) stack: Vec<VmValue>,
325 pub(crate) env: VmEnv,
326 pub(crate) output: String,
327 pub(crate) builtins: Arc<BTreeMap<String, VmBuiltinFn>>,
328 pub(crate) async_builtins: Arc<BTreeMap<String, VmAsyncBuiltinFn>>,
329 pub(crate) builtin_metadata: Arc<BTreeMap<String, VmBuiltinMetadata>>,
330 pub(crate) builtins_by_id: Arc<HashMap<BuiltinId, VmBuiltinEntry>>,
333 pub(crate) builtin_id_collisions: Arc<HashSet<BuiltinId>>,
336 pub(crate) iterators: Vec<IterState>,
338 pub(crate) frames: Vec<CallFrame>,
340 pub(crate) exception_handlers: Vec<ExceptionHandler>,
342 pub(crate) spawned_tasks: BTreeMap<String, VmTaskHandle>,
344 pub(crate) sync_runtime: Arc<crate::synchronization::VmSyncRuntime>,
346 pub(crate) shared_state_runtime: Arc<crate::shared_state::VmSharedStateRuntime>,
348 pub(crate) inline_cache_sets: Vec<Vec<crate::chunk::InlineCacheEntry>>,
352 pub(crate) inline_cache_set_by_chunk: HashMap<u64, usize>,
353 pub(crate) pool_registry: Arc<crate::stdlib::pool::PoolRegistry>,
355 pub(crate) wait_for_graph: Arc<crate::wait_for_graph::VmWaitForGraph>,
357 pub(crate) held_sync_guards: Vec<crate::synchronization::VmSyncHeldGuard>,
359 pub(crate) inherited_held_keys: Arc<Vec<crate::synchronization::VmSyncHeldKey>>,
367 pub(crate) task_scopes: Vec<TaskScope>,
373 pub(crate) task_counter: u64,
375 pub(crate) runtime_context_counter: u64,
377 pub(crate) runtime_context: crate::runtime_context::RuntimeContext,
379 pub(crate) deadlines: Vec<(Instant, usize)>,
381 pub(crate) execution_deadline: Arc<ExecutionDeadlineState>,
383 pub(crate) breakpoints: BTreeMap<String, std::collections::BTreeSet<usize>>,
388 pub(crate) function_breakpoints: std::collections::BTreeSet<String>,
394 pub(crate) pending_function_bp: Option<String>,
399 pub(crate) step_mode: bool,
401 pub(crate) step_frame_depth: usize,
403 pub(crate) stopped: bool,
405 pub(crate) last_line: usize,
407 pub(crate) source_dir: Option<std::path::PathBuf>,
409 pub(crate) imported_paths: Vec<std::path::PathBuf>,
411 pub(crate) deferred_cyclic_imports: Vec<super::modules::DeferredCyclicImport>,
415 pub(crate) module_cache: Arc<BTreeMap<std::path::PathBuf, LoadedModule>>,
417 pub(crate) lazy_callable_modules: LazyCallableModuleCache,
421 pub(crate) source_cache: Arc<BTreeMap<std::path::PathBuf, String>>,
423 pub(crate) source_file: Option<String>,
425 pub(crate) source_text: Option<String>,
427 pub(crate) coverage: Option<crate::coverage::Coverage>,
430 pub(crate) bridge: Option<Arc<crate::bridge::HostBridge>>,
432 pub(crate) denied_builtins: Arc<HashSet<String>>,
434 pub(crate) cancel_token: Option<std::sync::Arc<std::sync::atomic::AtomicBool>>,
436 pub(crate) interrupt_signal_token: Option<std::sync::Arc<std::sync::Mutex<Option<String>>>>,
437 pub(crate) cancel_grace_instructions_remaining: Option<usize>,
442 pub(crate) interrupt_handlers: Vec<InterruptHandler>,
444 pub(crate) next_interrupt_handle: i64,
445 pub(crate) pending_interrupt_signal: Option<String>,
446 pub(crate) interrupted: bool,
447 pub(crate) dispatching_interrupt: bool,
448 pub(crate) interrupt_handler_deadline: Option<Instant>,
449 pub(crate) error_stack_trace: Vec<(String, usize, usize, Option<String>)>,
451 pub(crate) yield_sender: Option<tokio::sync::mpsc::Sender<Result<VmValue, VmError>>>,
454 pub(crate) project_root: Option<std::path::PathBuf>,
457 pub(crate) globals: Arc<crate::value::DictMap>,
460 pub(crate) debug_hook: Option<parking_lot::Mutex<Box<DebugHook>>>,
462 pub(crate) runtime_limits: RuntimeLimits,
464}
465
466#[derive(Clone)]
474pub struct VmBaseline {
475 builtins: Arc<BTreeMap<String, VmBuiltinFn>>,
476 async_builtins: Arc<BTreeMap<String, VmAsyncBuiltinFn>>,
477 builtin_metadata: Arc<BTreeMap<String, VmBuiltinMetadata>>,
478 builtins_by_id: Arc<HashMap<BuiltinId, VmBuiltinEntry>>,
479 builtin_id_collisions: Arc<HashSet<BuiltinId>>,
480 source_dir: Option<std::path::PathBuf>,
481 source_file: Option<String>,
482 source_text: Option<String>,
483 project_root: Option<std::path::PathBuf>,
484 globals: Arc<crate::value::DictMap>,
485 denied_builtins: Arc<HashSet<String>>,
486 runtime_limits: RuntimeLimits,
487}
488
489impl VmBaseline {
490 pub fn from_vm(vm: &Vm) -> Self {
491 Self {
492 builtins: Arc::clone(&vm.builtins),
493 async_builtins: Arc::clone(&vm.async_builtins),
494 builtin_metadata: Arc::clone(&vm.builtin_metadata),
495 builtins_by_id: Arc::clone(&vm.builtins_by_id),
496 builtin_id_collisions: Arc::clone(&vm.builtin_id_collisions),
497 source_dir: vm.source_dir.clone(),
498 source_file: vm.source_file.clone(),
499 source_text: vm.source_text.clone(),
500 project_root: vm.project_root.clone(),
501 globals: Arc::clone(&vm.globals),
502 denied_builtins: Arc::clone(&vm.denied_builtins),
503 runtime_limits: vm.runtime_limits,
504 }
505 }
506
507 pub fn instantiate(&self) -> Vm {
508 let mut source_cache = BTreeMap::new();
509 if let (Some(file), Some(text)) = (&self.source_file, &self.source_text) {
510 source_cache.insert(std::path::PathBuf::from(file), text.clone());
511 }
512 if let Some(dir) = &self.source_dir {
513 crate::stdlib::set_thread_source_dir(dir);
514 }
515
516 let mut vm = Vm {
517 stack: Vec::with_capacity(256),
518 env: VmEnv::new(),
519 output: String::new(),
520 builtins: Arc::clone(&self.builtins),
521 async_builtins: Arc::clone(&self.async_builtins),
522 builtin_metadata: Arc::clone(&self.builtin_metadata),
523 builtins_by_id: Arc::clone(&self.builtins_by_id),
524 builtin_id_collisions: Arc::clone(&self.builtin_id_collisions),
525 iterators: Vec::new(),
526 frames: Vec::new(),
527 exception_handlers: Vec::new(),
528 spawned_tasks: BTreeMap::new(),
529 sync_runtime: Arc::new(crate::synchronization::VmSyncRuntime::new()),
530 shared_state_runtime: Arc::new(crate::shared_state::VmSharedStateRuntime::new()),
531 inline_cache_sets: Vec::new(),
532 inline_cache_set_by_chunk: HashMap::new(),
533 pool_registry: crate::stdlib::pool::new_pool_registry(),
534 wait_for_graph: Arc::new(crate::wait_for_graph::VmWaitForGraph::new()),
535 held_sync_guards: Vec::new(),
536 inherited_held_keys: Arc::new(Vec::new()),
537 task_scopes: Vec::new(),
538 task_counter: 0,
539 runtime_context_counter: 0,
540 runtime_context: crate::runtime_context::RuntimeContext::root(),
541 deadlines: Vec::new(),
542 execution_deadline: super::execution::new_execution_deadline_state(None),
543 breakpoints: BTreeMap::new(),
544 function_breakpoints: std::collections::BTreeSet::new(),
545 pending_function_bp: None,
546 step_mode: false,
547 step_frame_depth: 0,
548 stopped: false,
549 last_line: 0,
550 source_dir: self.source_dir.clone(),
551 imported_paths: Vec::new(),
552 deferred_cyclic_imports: Vec::new(),
553 module_cache: Arc::new(BTreeMap::new()),
554 lazy_callable_modules: Arc::new(crate::value::VmMutex::new(BTreeMap::new())),
555 source_cache: Arc::new(source_cache),
556 source_file: self.source_file.clone(),
557 source_text: self.source_text.clone(),
558 coverage: crate::coverage::for_primary(self.source_file.as_deref()),
559 bridge: None,
560 denied_builtins: Arc::clone(&self.denied_builtins),
561 cancel_token: None,
562 interrupt_signal_token: None,
563 cancel_grace_instructions_remaining: None,
564 interrupt_handlers: Vec::new(),
565 next_interrupt_handle: 1,
566 pending_interrupt_signal: None,
567 interrupted: false,
568 dispatching_interrupt: false,
569 interrupt_handler_deadline: None,
570 error_stack_trace: Vec::new(),
571 yield_sender: None,
572 project_root: self.project_root.clone(),
573 globals: Arc::clone(&self.globals),
574 debug_hook: None,
575 runtime_limits: self.runtime_limits,
576 };
577
578 crate::stdlib::rebind_execution_state_builtins(&mut vm);
579 vm
580 }
581}
582
583impl Vm {
584 pub(crate) fn ensure_execution_available(&self) -> Result<(), VmError> {
585 if self.execution_deadline.is_abandoned() {
586 return Err(VmError::AbandonedExecution);
587 }
588 Ok(())
589 }
590
591 pub(crate) fn fresh_local_slots(chunk: &Chunk) -> Vec<LocalSlot> {
592 chunk
593 .local_slots
594 .iter()
595 .map(|_| LocalSlot {
596 value: VmValue::Nil,
597 initialized: false,
598 synced: false,
599 })
600 .collect()
601 }
602
603 pub(crate) fn bind_param_slots(
604 slots: &mut [LocalSlot],
605 func: &crate::chunk::CompiledFunction,
606 args: &[VmValue],
607 synced: bool,
608 ) {
609 Self::bind_param_slots_args(slots, func, &super::CallArgs::Slice(args), synced);
610 }
611
612 pub(crate) fn bind_param_slots_args(
613 slots: &mut [LocalSlot],
614 func: &crate::chunk::CompiledFunction,
615 args: &super::CallArgs<'_>,
616 synced: bool,
617 ) {
618 let param_count = func.params.len();
619 for (i, _param) in func.params.iter().enumerate() {
620 if i >= slots.len() {
621 break;
622 }
623 if func.has_rest_param && i == param_count - 1 {
624 let rest_args = args.to_vec_from(i);
625 slots[i].value = VmValue::List(std::sync::Arc::new(rest_args));
626 slots[i].initialized = true;
627 slots[i].synced = synced;
628 } else if let Some(arg) = args.get(i) {
629 slots[i].value = arg.clone();
630 slots[i].initialized = true;
631 slots[i].synced = synced;
632 }
633 }
634 }
635
636 pub(crate) fn visible_variables(&self) -> crate::value::DictMap {
637 let mut vars = self.env.all_variables();
638 let Some(frame) = self.frames.last() else {
639 return vars;
640 };
641 for (slot, info) in frame.local_slots.iter().zip(frame.chunk.local_slots.iter()) {
642 if slot.initialized && info.scope_depth <= frame.local_scope_depth {
643 vars.insert(crate::value::intern_key(&info.name), slot.value.clone());
644 }
645 }
646 vars
647 }
648
649 pub(crate) fn sync_current_frame_locals_to_env(&mut self) {
650 let frames = &mut self.frames;
651 let env = &mut self.env;
652 let Some(frame) = frames.last_mut() else {
653 return;
654 };
655 let local_scope_base = frame.local_scope_base;
656 let local_scope_depth = frame.local_scope_depth;
657 for (slot, info) in frame
658 .local_slots
659 .iter_mut()
660 .zip(frame.chunk.local_slots.iter())
661 {
662 if slot.initialized && !slot.synced && info.scope_depth <= local_scope_depth {
663 slot.synced = true;
664 let scope_idx = local_scope_base + info.scope_depth;
665 while env.scopes.len() <= scope_idx {
666 env.push_scope();
667 }
668 Arc::make_mut(&mut env.scopes[scope_idx].vars).insert(
672 info.name.clone(),
673 crate::value::Binding::Value {
674 value: slot.value.clone(),
675 mutable: info.mutable,
676 },
677 );
678 }
679 }
680 }
681
682 pub(crate) fn closure_call_env_for_current_frame(
683 &self,
684 closure: &crate::value::VmClosure,
685 ) -> VmEnv {
686 if closure.module_state().is_some() {
687 return closure.env.cloned_for_call();
688 }
689 let call_env = Self::closure_call_env(&self.env, closure);
690 if !closure.func.chunk.references_outer_names {
695 return call_env;
696 }
697 let mut call_env = call_env;
698 let Some(frame) = self.frames.last() else {
699 return call_env;
700 };
701 for (slot, info) in frame
702 .local_slots
703 .iter()
704 .zip(frame.chunk.local_slots.iter())
705 .filter(|(slot, info)| slot.initialized && info.scope_depth <= frame.local_scope_depth)
706 {
707 if matches!(slot.value, VmValue::Closure(_)) && !call_env.contains(&info.name) {
708 let _ = call_env.define(&info.name, slot.value.clone(), info.mutable);
709 }
710 }
711 call_env
712 }
713
714 pub(crate) fn active_local_slot_value(&self, name: &str) -> Option<VmValue> {
715 let frame = self.frames.last()?;
716 let idx = self.active_local_slot_index(name)?;
717 frame.local_slots.get(idx).map(|slot| slot.value.clone())
718 }
719
720 pub(crate) fn active_local_slot_index(&self, name: &str) -> Option<usize> {
725 let frame = self.frames.last()?;
726 for (idx, info) in frame.chunk.local_slots.iter().enumerate().rev() {
727 if info.name == name && info.scope_depth <= frame.local_scope_depth {
728 if let Some(slot) = frame.local_slots.get(idx) {
729 if slot.initialized {
730 return Some(idx);
731 }
732 }
733 }
734 }
735 None
736 }
737
738 pub(crate) fn assign_active_local_slot(
739 &mut self,
740 name: &str,
741 value: VmValue,
742 debug: bool,
743 ) -> Result<bool, VmError> {
744 let Some(frame) = self.frames.last_mut() else {
745 return Ok(false);
746 };
747 for (idx, info) in frame.chunk.local_slots.iter().enumerate().rev() {
748 if info.name == name && info.scope_depth <= frame.local_scope_depth {
749 if !debug && !info.mutable {
750 return Err(VmError::ImmutableAssignment(name.to_string()));
751 }
752 if let Some(slot) = frame.local_slots.get_mut(idx) {
753 crate::value::recursion::dismantle(std::mem::replace(&mut slot.value, value));
754 slot.initialized = true;
755 slot.synced = false;
756 return Ok(true);
757 }
758 }
759 }
760 Ok(false)
761 }
762
763 pub fn new() -> Self {
764 Self {
765 stack: Vec::with_capacity(256),
766 env: VmEnv::new(),
767 output: String::new(),
768 builtins: Arc::new(BTreeMap::new()),
769 async_builtins: Arc::new(BTreeMap::new()),
770 builtin_metadata: Arc::new(BTreeMap::new()),
771 builtins_by_id: Arc::new(HashMap::new()),
772 builtin_id_collisions: Arc::new(HashSet::new()),
773 iterators: Vec::new(),
774 frames: Vec::new(),
775 exception_handlers: Vec::new(),
776 spawned_tasks: BTreeMap::new(),
777 sync_runtime: Arc::new(crate::synchronization::VmSyncRuntime::new()),
778 shared_state_runtime: Arc::new(crate::shared_state::VmSharedStateRuntime::new()),
779 inline_cache_sets: Vec::new(),
780 inline_cache_set_by_chunk: HashMap::new(),
781 pool_registry: crate::stdlib::pool::new_pool_registry(),
782 wait_for_graph: Arc::new(crate::wait_for_graph::VmWaitForGraph::new()),
783 held_sync_guards: Vec::new(),
784 inherited_held_keys: Arc::new(Vec::new()),
785 task_scopes: Vec::new(),
786 task_counter: 0,
787 runtime_context_counter: 0,
788 runtime_context: crate::runtime_context::RuntimeContext::root(),
789 deadlines: Vec::new(),
790 execution_deadline: super::execution::new_execution_deadline_state(None),
791 breakpoints: BTreeMap::new(),
792 function_breakpoints: std::collections::BTreeSet::new(),
793 pending_function_bp: None,
794 step_mode: false,
795 step_frame_depth: 0,
796 stopped: false,
797 last_line: 0,
798 source_dir: None,
799 imported_paths: Vec::new(),
800 deferred_cyclic_imports: Vec::new(),
801 module_cache: Arc::new(BTreeMap::new()),
802 lazy_callable_modules: Arc::new(crate::value::VmMutex::new(BTreeMap::new())),
803 source_cache: Arc::new(BTreeMap::new()),
804 source_file: None,
805 source_text: None,
806 coverage: crate::coverage::for_primary(None),
807 bridge: None,
808 denied_builtins: Arc::new(HashSet::new()),
809 cancel_token: None,
810 interrupt_signal_token: None,
811 cancel_grace_instructions_remaining: None,
812 interrupt_handlers: Vec::new(),
813 next_interrupt_handle: 1,
814 pending_interrupt_signal: None,
815 interrupted: false,
816 dispatching_interrupt: false,
817 interrupt_handler_deadline: None,
818 error_stack_trace: Vec::new(),
819 yield_sender: None,
820 project_root: None,
821 globals: Arc::new(crate::value::DictMap::new()),
822 debug_hook: None,
823 runtime_limits: RuntimeLimits::default(),
824 }
825 }
826
827 pub fn baseline(&self) -> VmBaseline {
828 VmBaseline::from_vm(self)
829 }
830
831 pub fn runtime_limits(&self) -> RuntimeLimits {
833 self.runtime_limits
834 }
835
836 pub fn runtime_limit_report(&self) -> crate::RuntimeLimitsReport {
838 self.runtime_limits.report()
839 }
840
841 #[inline]
855 pub(crate) fn debugger_attached(&self) -> bool {
856 self.debug_hook.is_some()
857 || !self.breakpoints.is_empty()
858 || !self.function_breakpoints.is_empty()
859 }
860
861 pub fn set_bridge(&mut self, bridge: Arc<crate::bridge::HostBridge>) {
863 self.bridge = Some(bridge);
864 }
865
866 pub fn set_denied_builtins(&mut self, denied: HashSet<String>) {
869 self.denied_builtins = Arc::new(denied);
870 }
871
872 pub fn set_source_info(&mut self, file: &str, text: &str) {
874 self.source_file = Some(file.to_string());
875 self.source_text = Some(text.to_string());
876 if let Some(cov) = self.coverage.as_mut() {
877 cov.set_primary_file(file);
878 }
879 Arc::make_mut(&mut self.source_cache)
880 .insert(std::path::PathBuf::from(file), text.to_string());
881 }
882
883 pub fn start(&mut self, chunk: &Chunk) -> Result<(), VmError> {
885 self.ensure_execution_available()?;
886 let debugger = self.debugger_attached();
893 let initial_env = if debugger {
894 Some(self.env.clone())
895 } else {
896 None
897 };
898 let initial_local_slots = if debugger {
899 Some(Self::fresh_local_slots(chunk))
900 } else {
901 None
902 };
903 let chunk = Arc::new(chunk.clone());
904 let local_slots = Self::fresh_local_slots(&chunk);
905 let inline_cache_set = self.inline_cache_set_index_for_chunk(&chunk);
906 self.frames.push(CallFrame {
907 chunk,
908 inline_cache_set,
909 ip: 0,
910 stack_base: self.stack.len(),
911 saved_env: self.env.clone(),
912 initial_env,
913 initial_local_slots,
914 saved_iterator_depth: self.iterators.len(),
915 fn_name: String::new(),
916 argc: 0,
917 saved_source_dir: None,
918 module_functions: None,
919 module_state: None,
920 local_slots,
921 local_scope_base: self.env.scope_depth().saturating_sub(1),
922 local_scope_depth: 0,
923 });
924 Ok(())
925 }
926
927 pub(crate) fn child_vm(&self) -> Vm {
930 Vm {
931 stack: Vec::with_capacity(64),
932 env: self.env.clone(),
933 output: String::new(),
934 builtins: Arc::clone(&self.builtins),
935 async_builtins: Arc::clone(&self.async_builtins),
936 builtin_metadata: Arc::clone(&self.builtin_metadata),
937 builtins_by_id: Arc::clone(&self.builtins_by_id),
938 builtin_id_collisions: Arc::clone(&self.builtin_id_collisions),
939 iterators: Vec::new(),
940 frames: Vec::new(),
941 exception_handlers: Vec::new(),
942 spawned_tasks: BTreeMap::new(),
943 sync_runtime: self.sync_runtime.clone(),
944 shared_state_runtime: self.shared_state_runtime.clone(),
945 inline_cache_sets: Vec::new(),
946 inline_cache_set_by_chunk: HashMap::new(),
947 pool_registry: self.pool_registry.clone(),
948 wait_for_graph: self.wait_for_graph.clone(),
949 held_sync_guards: Vec::new(),
950 inherited_held_keys: Arc::new(Vec::new()),
951 task_scopes: Vec::new(),
952 task_counter: 0,
953 runtime_context_counter: self.runtime_context_counter,
954 runtime_context: self.runtime_context.clone(),
955 deadlines: self.deadlines.clone(),
956 execution_deadline: self.execution_deadline.fork(),
957 breakpoints: BTreeMap::new(),
958 function_breakpoints: std::collections::BTreeSet::new(),
959 pending_function_bp: None,
960 step_mode: false,
961 step_frame_depth: 0,
962 stopped: false,
963 last_line: 0,
964 source_dir: self.source_dir.clone(),
965 imported_paths: Vec::new(),
966 deferred_cyclic_imports: Vec::new(),
967 module_cache: Arc::clone(&self.module_cache),
968 lazy_callable_modules: Arc::clone(&self.lazy_callable_modules),
969 source_cache: Arc::clone(&self.source_cache),
970 source_file: self.source_file.clone(),
971 source_text: self.source_text.clone(),
972 coverage: crate::coverage::for_primary(self.source_file.as_deref()),
973 bridge: self.bridge.clone(),
974 denied_builtins: Arc::clone(&self.denied_builtins),
975 cancel_token: self.cancel_token.clone(),
976 interrupt_signal_token: self.interrupt_signal_token.clone(),
977 cancel_grace_instructions_remaining: None,
978 interrupt_handlers: Vec::new(),
979 next_interrupt_handle: 1,
980 pending_interrupt_signal: None,
981 interrupted: self.interrupted,
982 dispatching_interrupt: false,
983 interrupt_handler_deadline: None,
984 error_stack_trace: Vec::new(),
985 yield_sender: None,
986 project_root: self.project_root.clone(),
987 globals: Arc::clone(&self.globals),
988 debug_hook: None,
989 runtime_limits: self.runtime_limits,
990 }
991 }
992
993 pub(crate) fn child_vm_for_host(&self) -> Vm {
996 self.child_vm()
997 }
998
999 pub(crate) fn cancel_spawned_tasks(&mut self) {
1003 for (_, task) in std::mem::take(&mut self.spawned_tasks) {
1004 task.cancel_token
1005 .store(true, std::sync::atomic::Ordering::SeqCst);
1006 task.handle.abort();
1007 }
1008 }
1009
1010 pub fn set_source_dir(&mut self, dir: &std::path::Path) {
1013 let dir = crate::stdlib::process::normalize_context_path(dir);
1014 self.source_dir = Some(dir.clone());
1015 crate::stdlib::set_thread_source_dir(&dir);
1016 if self.project_root.is_none() {
1018 self.project_root = crate::stdlib::process::find_project_root(&dir);
1019 }
1020 }
1021
1022 pub fn set_project_root(&mut self, root: &std::path::Path) {
1025 self.project_root = Some(root.to_path_buf());
1026 }
1027
1028 pub(crate) fn explicit_project_root(&self) -> Option<&std::path::Path> {
1031 self.project_root.as_deref()
1032 }
1033
1034 pub fn project_root(&self) -> Option<&std::path::Path> {
1036 self.project_root.as_deref().or(self.source_dir.as_deref())
1037 }
1038
1039 pub fn builtin_names(&self) -> Vec<String> {
1041 let mut names: Vec<String> = self.builtins.keys().cloned().collect();
1042 names.extend(self.async_builtins.keys().cloned());
1043 names
1044 }
1045
1046 pub fn builtin_metadata(&self) -> Vec<VmBuiltinMetadata> {
1048 self.builtin_metadata.values().cloned().collect()
1049 }
1050
1051 pub fn builtin_metadata_for(&self, name: &str) -> Option<&VmBuiltinMetadata> {
1053 self.builtin_metadata.get(name)
1054 }
1055
1056 pub fn set_global(&mut self, name: &str, value: VmValue) {
1059 Arc::make_mut(&mut self.globals).insert(crate::value::intern_key(name), value);
1060 }
1061
1062 pub fn global(&self, name: &str) -> Option<&VmValue> {
1068 self.globals.get(name)
1069 }
1070
1071 pub fn set_harness(&mut self, harness: crate::harness::Harness) {
1077 self.set_global("harness", harness.into_vm_value());
1078 }
1079
1080 pub fn output(&self) -> &str {
1082 &self.output
1083 }
1084
1085 pub fn take_output(&mut self) -> String {
1089 std::mem::take(&mut self.output)
1090 }
1091
1092 pub fn append_output(&mut self, text: &str) {
1096 self.output.push_str(text);
1097 }
1098
1099 pub(crate) fn pop(&mut self) -> Result<VmValue, VmError> {
1100 self.stack.pop().ok_or(VmError::StackUnderflow)
1101 }
1102
1103 pub(crate) fn peek(&self) -> Result<&VmValue, VmError> {
1104 self.stack.last().ok_or(VmError::StackUnderflow)
1105 }
1106
1107 pub(crate) fn const_str(c: &Constant) -> Result<&str, VmError> {
1108 match c {
1109 Constant::String(s) => Ok(s.as_str()),
1110 _ => Err(VmError::TypeError("expected string constant".into())),
1111 }
1112 }
1113
1114 pub(crate) fn release_sync_guards_for_current_scope(&mut self) {
1115 let depth = self.env.scope_depth();
1116 self.held_sync_guards
1117 .retain(|guard| guard.env_scope_depth < depth);
1118 self.cancel_task_scopes_where(|s| s.env_scope_depth >= depth);
1121 }
1122
1123 pub(crate) fn release_sync_guards_after_unwind(
1124 &mut self,
1125 frame_depth: usize,
1126 env_scope_depth: usize,
1127 ) {
1128 self.held_sync_guards.retain(|guard| {
1129 guard.frame_depth <= frame_depth && guard.env_scope_depth <= env_scope_depth
1130 });
1131 self.cancel_task_scopes_where(|s| {
1134 !(s.frame_depth <= frame_depth && s.env_scope_depth <= env_scope_depth)
1135 });
1136 }
1137
1138 pub(crate) fn release_sync_guards_for_frame(&mut self, frame_depth: usize) {
1139 self.held_sync_guards
1140 .retain(|guard| guard.frame_depth != frame_depth);
1141 self.cancel_task_scopes_where(|s| s.frame_depth == frame_depth);
1144 }
1145
1146 pub(crate) fn adopt_sync_permit_for_current_scope(
1147 &mut self,
1148 permit: crate::value::VmSyncPermitHandle,
1149 ) {
1150 if permit.is_released()
1151 || self
1152 .held_sync_guards
1153 .iter()
1154 .any(|guard| guard._permit.same_lease(&permit))
1155 {
1156 return;
1157 }
1158 self.held_sync_guards
1159 .push(crate::synchronization::VmSyncHeldGuard {
1160 _permit: permit,
1161 frame_depth: self.frames.len(),
1162 env_scope_depth: self.env.scope_depth(),
1163 });
1164 }
1165
1166 pub(crate) fn deregister_task_from_scopes(&mut self, id: &str) {
1169 for scope in &mut self.task_scopes {
1170 scope.task_ids.retain(|t| t != id);
1171 }
1172 }
1173
1174 fn cancel_task_scopes_where<F: Fn(&TaskScope) -> bool>(&mut self, doomed: F) {
1177 let mut i = 0;
1178 while i < self.task_scopes.len() {
1179 if doomed(&self.task_scopes[i]) {
1180 let scope = self.task_scopes.remove(i);
1181 for id in &scope.task_ids {
1182 if let Some(task) = self.spawned_tasks.remove(id) {
1183 task.cancel_token
1184 .store(true, std::sync::atomic::Ordering::SeqCst);
1185 task.handle.abort();
1186 }
1187 }
1188 } else {
1189 i += 1;
1190 }
1191 }
1192 }
1193
1194 pub(crate) fn held_permits_for(&self, kind: &str, key: &str) -> u32 {
1198 let own: u32 = self
1199 .held_sync_guards
1200 .iter()
1201 .filter(|guard| {
1202 !guard._permit.is_released()
1203 && guard._permit.kind() == kind
1204 && guard._permit.key() == key
1205 })
1206 .map(|guard| guard._permit.permits())
1207 .sum();
1208 let inherited: u32 = self
1209 .inherited_held_keys
1210 .iter()
1211 .filter(|held| held.kind == kind && held.key == key)
1212 .map(|held| held.permits)
1213 .sum();
1214 own + inherited
1215 }
1216
1217 pub(crate) fn combined_held_keys(&self) -> Vec<crate::synchronization::VmSyncHeldKey> {
1220 let mut keys: Vec<crate::synchronization::VmSyncHeldKey> = self
1221 .held_sync_guards
1222 .iter()
1223 .filter_map(|guard| crate::synchronization::VmSyncHeldKey::from_permit(&guard._permit))
1224 .collect();
1225 keys.extend(self.inherited_held_keys.iter().cloned());
1226 keys
1227 }
1228
1229 pub(crate) fn child_vm_inline(&self) -> Vm {
1235 let mut child = self.child_vm();
1236 child.inherited_held_keys = Arc::new(self.combined_held_keys());
1237 child
1238 }
1239}
1240
1241impl Drop for Vm {
1242 fn drop(&mut self) {
1243 if let Some(coverage) = self.coverage.take() {
1244 crate::coverage::merge_into_global(coverage);
1245 }
1246 self.cancel_spawned_tasks();
1247 }
1248}
1249
1250impl Default for Vm {
1251 fn default() -> Self {
1252 Self::new()
1253 }
1254}
1255
1256#[cfg(test)]
1257mod tests {
1258
1259 use super::*;
1260
1261 fn baseline_with_stdlib(source: &str) -> VmBaseline {
1262 let mut vm = Vm::new();
1263 crate::register_vm_stdlib(&mut vm);
1264 vm.set_source_info("baseline_test.harn", source);
1265 vm.set_global(
1266 "stable_global",
1267 VmValue::String(arcstr::ArcStr::from("baseline")),
1268 );
1269 vm.baseline()
1270 }
1271
1272 #[test]
1273 fn vm_baseline_instantiates_clean_mutable_execution_state() {
1274 let baseline = baseline_with_stdlib("pipeline main() { __io_println(stable_global) }");
1275
1276 let mut dirty = baseline.instantiate();
1277 dirty.stack.push(VmValue::Int(42));
1278 dirty.output.push_str("dirty");
1279 dirty.task_counter = 9;
1280 dirty.runtime_context_counter = 7;
1281 dirty
1282 .error_stack_trace
1283 .push(("main".to_string(), 1, 1, None));
1284
1285 let clean = baseline.instantiate();
1286 assert!(clean.stack.is_empty());
1287 assert!(clean.output.is_empty());
1288 assert!(clean.frames.is_empty());
1289 assert!(clean.exception_handlers.is_empty());
1290 assert!(clean.spawned_tasks.is_empty());
1291 assert!(clean.held_sync_guards.is_empty());
1292 assert_eq!(clean.task_counter, 0);
1293 assert_eq!(clean.runtime_context_counter, 0);
1294 assert!(clean.deadlines.is_empty());
1295 assert!(clean.cancel_token.is_none());
1296 assert!(clean.interrupt_handlers.is_empty());
1297 assert!(clean.error_stack_trace.is_empty());
1298 assert!(clean.bridge.is_none());
1299 assert!(clean
1300 .globals
1301 .get("stable_global")
1302 .is_some_and(|value| value.display() == "baseline"));
1303 }
1304
1305 #[tokio::test]
1306 async fn inline_child_inherits_held_lock_keys_but_concurrent_child_does_not() {
1307 let mut parent = Vm::new();
1308 let permit = parent
1309 .sync_runtime
1310 .acquire("mutex", "v:test", 1, 1, None, None)
1311 .await
1312 .unwrap()
1313 .unwrap();
1314 parent
1315 .held_sync_guards
1316 .push(crate::synchronization::VmSyncHeldGuard {
1317 _permit: permit,
1318 frame_depth: 0,
1319 env_scope_depth: 0,
1320 });
1321 assert_eq!(parent.held_permits_for("mutex", "v:test"), 1);
1322
1323 let inline = parent.child_vm_inline();
1328 assert_eq!(inline.held_permits_for("mutex", "v:test"), 1);
1329 assert_eq!(
1330 inline.child_vm_inline().held_permits_for("mutex", "v:test"),
1331 1
1332 );
1333
1334 let concurrent = parent.child_vm();
1338 assert_eq!(concurrent.held_permits_for("mutex", "v:test"), 0);
1339 }
1340
1341 #[test]
1342 fn vm_reports_effective_runtime_limits() {
1343 let vm = Vm::new();
1344
1345 assert_eq!(vm.runtime_limits(), RuntimeLimits::default());
1346 assert_eq!(
1347 vm.runtime_limit_report().entries.len(),
1348 crate::RUNTIME_LIMIT_DESCRIPTIONS.len()
1349 );
1350 assert_eq!(vm.child_vm().runtime_limits(), vm.runtime_limits());
1351 assert_eq!(
1352 vm.baseline().instantiate().runtime_limits(),
1353 vm.runtime_limits()
1354 );
1355 }
1356
1357 #[tokio::test(flavor = "current_thread")]
1358 async fn vm_baseline_rebinds_shared_state_builtins_per_instance() {
1359 let local = tokio::task::LocalSet::new();
1360 local
1361 .run_until(async {
1362 let source = r#"
1363pipeline main() {
1364 const cell = shared_cell({scope: "task_group", key: "turn", initial: 0})
1365 __io_println(shared_get(cell))
1366 shared_set(cell, shared_get(cell) + 1)
1367}"#;
1368 let chunk = crate::compile_source(source).expect("compile");
1369 let baseline = baseline_with_stdlib(source);
1370
1371 let mut first = baseline.instantiate();
1372 first.execute(&chunk).await.expect("first execute");
1373 assert_eq!(first.output(), "0\n");
1374
1375 let mut second = baseline.instantiate();
1376 second.execute(&chunk).await.expect("second execute");
1377 assert_eq!(
1378 second.output(),
1379 "0\n",
1380 "shared state created by the first VM must not leak into the next baseline instance"
1381 );
1382 })
1383 .await;
1384 }
1385}