runmat_runtime/execution/
services.rs

1use std::collections::HashMap;
2use std::sync::atomic::{AtomicU64, Ordering};
3use std::sync::Mutex;
4
5use runmat_execution::{
6    CancellationReason, ExecutionScopeId, FutureHandle, FutureId, JobHandle, OutputContract,
7    TaskHandle, TaskId,
8};
9use runmat_value::Value;
10
11use super::ExecutionServiceError;
12
13static NEXT_SERVICE_NONCE: AtomicU64 = AtomicU64::new(1);
14
15#[derive(Clone, Debug, PartialEq)]
16pub struct DeferredCall {
17    pub function: usize,
18    pub arguments: Vec<Value>,
19    pub requested_outputs: usize,
20    pub program_revision: Option<runmat_execution::ProgramRevision>,
21    /// Exact, runtime-opaque program description supplied by the VM.
22    ///
23    /// The serial service does not inspect this payload. Execution adapters may
24    /// require it to reproduce the callable in an isolated worker.
25    pub program: Option<Vec<u8>>,
26}
27
28#[derive(Clone, Debug, Eq, PartialEq)]
29pub struct DurableJobOptions {
30    pub idempotency_key: Option<String>,
31    pub retention_millis: u64,
32}
33
34#[derive(Clone, Debug, PartialEq)]
35pub enum AwaitAction {
36    Passthrough(Value),
37    /// The host has started the operation but cannot complete it synchronously.
38    ///
39    /// The VM yields once and polls `begin_await` again with this value. This
40    /// keeps browser workers non-blocking while native adapters may continue
41    /// to use an efficient blocking completion primitive.
42    Pending(Value),
43    ExecuteFuture {
44        handle: FutureHandle,
45        call: DeferredCall,
46    },
47    Completed(Value),
48}
49
50pub trait RuntimeExecutionServices {
51    fn scope_id(&self) -> ExecutionScopeId;
52    fn requires_program_capture(&self) -> bool {
53        false
54    }
55    fn create_future(&self, call: DeferredCall) -> Result<FutureHandle, ExecutionServiceError>;
56    fn spawn(&self, future: &FutureHandle) -> Result<TaskHandle, ExecutionServiceError>;
57    fn submit_job(
58        &self,
59        _call: DeferredCall,
60        _options: DurableJobOptions,
61    ) -> Result<JobHandle, ExecutionServiceError> {
62        Err(ExecutionServiceError::Failed(
63            "durable jobs are unavailable in this execution backend".into(),
64        ))
65    }
66    fn await_job(&self, _job: &JobHandle) -> Result<Value, ExecutionServiceError> {
67        Err(ExecutionServiceError::Failed(
68            "durable jobs are unavailable in this execution backend".into(),
69        ))
70    }
71    fn begin_await(&self, value: Value) -> Result<AwaitAction, ExecutionServiceError>;
72    fn complete_future(
73        &self,
74        future: &FutureHandle,
75        result: Result<Value, ExecutionServiceError>,
76    ) -> Result<(), ExecutionServiceError>;
77    fn cancel(
78        &self,
79        value: &Value,
80        reason: CancellationReason,
81    ) -> Result<(), ExecutionServiceError>;
82    fn drain_scope(&self, reason: CancellationReason);
83}
84
85#[derive(Clone, Debug)]
86enum FutureState {
87    Deferred(DeferredCall),
88    Running,
89    Completed(Result<Value, ExecutionServiceError>),
90    Cancelled,
91}
92
93#[derive(Clone, Debug)]
94struct TaskRecord {
95    future_id: FutureId,
96    generation: u64,
97    cancelled: bool,
98}
99
100#[derive(Debug)]
101struct ServiceState {
102    next_future: u64,
103    next_task: u64,
104    futures: HashMap<FutureId, FutureState>,
105    tasks: HashMap<TaskId, TaskRecord>,
106}
107
108/// Root-scoped serial execution service.
109///
110/// It is the correctness backend used when no process/worker scheduler is
111/// composed. Handles and state transitions are real; execution placement is
112/// deliberately delegated to later scheduler adapters.
113#[derive(Debug)]
114pub struct RuntimeExecutionService {
115    scope_id: ExecutionScopeId,
116    state: Mutex<ServiceState>,
117}
118
119impl RuntimeExecutionService {
120    pub fn new() -> Self {
121        let nonce = NEXT_SERVICE_NONCE.fetch_add(1, Ordering::Relaxed);
122        Self {
123            scope_id: ExecutionScopeId::derive(&[&nonce.to_be_bytes()]),
124            state: Mutex::new(ServiceState {
125                next_future: 0,
126                next_task: 0,
127                futures: HashMap::new(),
128                tasks: HashMap::new(),
129            }),
130        }
131    }
132
133    fn validate_scope(&self, scope_id: ExecutionScopeId) -> Result<(), ExecutionServiceError> {
134        if scope_id == self.scope_id {
135            Ok(())
136        } else {
137            Err(ExecutionServiceError::ForeignScope)
138        }
139    }
140}
141
142impl Default for RuntimeExecutionService {
143    fn default() -> Self {
144        Self::new()
145    }
146}
147
148impl RuntimeExecutionServices for RuntimeExecutionService {
149    fn scope_id(&self) -> ExecutionScopeId {
150        self.scope_id
151    }
152
153    fn create_future(&self, call: DeferredCall) -> Result<FutureHandle, ExecutionServiceError> {
154        let requested_outputs = u16::try_from(call.requested_outputs)
155            .map_err(|_| ExecutionServiceError::InvalidOutputContract)?;
156        let mut state = self.state.lock().expect("execution service state poisoned");
157        let sequence = state.next_future;
158        state.next_future = state.next_future.wrapping_add(1);
159        let id = FutureId::derive(&[self.scope_id.bytes(), &sequence.to_be_bytes()]);
160        state.futures.insert(id, FutureState::Deferred(call));
161        Ok(FutureHandle {
162            id,
163            scope_id: self.scope_id,
164            outputs: OutputContract { requested_outputs },
165        })
166    }
167
168    fn spawn(&self, future: &FutureHandle) -> Result<TaskHandle, ExecutionServiceError> {
169        self.validate_scope(future.scope_id)?;
170        let mut state = self.state.lock().expect("execution service state poisoned");
171        match state.futures.get(&future.id) {
172            Some(FutureState::Deferred(_))
173                if !state.tasks.values().any(|task| task.future_id == future.id) => {}
174            Some(_) => {
175                return Err(ExecutionServiceError::Failed(
176                    "future has already been scheduled".into(),
177                ));
178            }
179            None => return Err(ExecutionServiceError::UnknownHandle),
180        }
181        let sequence = state.next_task;
182        state.next_task = state.next_task.wrapping_add(1);
183        let id = TaskId::derive(&[self.scope_id.bytes(), &sequence.to_be_bytes()]);
184        let generation = 1;
185        state.tasks.insert(
186            id,
187            TaskRecord {
188                future_id: future.id,
189                generation,
190                cancelled: false,
191            },
192        );
193        Ok(TaskHandle {
194            id,
195            scope_id: self.scope_id,
196            generation,
197            outputs: future.outputs.clone(),
198        })
199    }
200
201    fn begin_await(&self, value: Value) -> Result<AwaitAction, ExecutionServiceError> {
202        if let Value::Job(handle) = &value {
203            return self.await_job(handle).map(AwaitAction::Completed);
204        }
205        let (future, task_id) = match value {
206            Value::Future(handle) => (handle, None),
207            Value::Task(task) => {
208                self.validate_scope(task.scope_id)?;
209                let state = self.state.lock().expect("execution service state poisoned");
210                let record = state
211                    .tasks
212                    .get(&task.id)
213                    .ok_or(ExecutionServiceError::UnknownHandle)?;
214                if record.generation != task.generation {
215                    return Err(ExecutionServiceError::UnknownHandle);
216                }
217                if record.cancelled {
218                    return Err(ExecutionServiceError::Cancelled);
219                }
220                (
221                    FutureHandle {
222                        id: record.future_id,
223                        scope_id: self.scope_id,
224                        outputs: task.outputs,
225                    },
226                    Some(task.id),
227                )
228            }
229            value => return Ok(AwaitAction::Passthrough(value)),
230        };
231        self.validate_scope(future.scope_id)?;
232        let mut state = self.state.lock().expect("execution service state poisoned");
233        let record = state
234            .futures
235            .get_mut(&future.id)
236            .ok_or(ExecutionServiceError::UnknownHandle)?;
237        match record {
238            FutureState::Deferred(call) => {
239                let call = call.clone();
240                *record = FutureState::Running;
241                Ok(AwaitAction::ExecuteFuture {
242                    handle: future,
243                    call,
244                })
245            }
246            FutureState::Running => Err(ExecutionServiceError::Failed(
247                "execution is already being awaited".to_string(),
248            )),
249            FutureState::Completed(result) => {
250                let result = result.clone();
251                if let Some(task_id) = task_id {
252                    state.tasks.remove(&task_id);
253                }
254                result.map(AwaitAction::Completed)
255            }
256            FutureState::Cancelled => Err(ExecutionServiceError::Cancelled),
257        }
258    }
259
260    fn complete_future(
261        &self,
262        future: &FutureHandle,
263        result: Result<Value, ExecutionServiceError>,
264    ) -> Result<(), ExecutionServiceError> {
265        self.validate_scope(future.scope_id)?;
266        let mut state = self.state.lock().expect("execution service state poisoned");
267        let record = state
268            .futures
269            .get_mut(&future.id)
270            .ok_or(ExecutionServiceError::UnknownHandle)?;
271        if !matches!(record, FutureState::Running) {
272            return Err(ExecutionServiceError::UnknownHandle);
273        }
274        *record = FutureState::Completed(result);
275        state.tasks.retain(|_, task| task.future_id != future.id);
276        Ok(())
277    }
278
279    fn cancel(
280        &self,
281        value: &Value,
282        _reason: CancellationReason,
283    ) -> Result<(), ExecutionServiceError> {
284        let mut state = self.state.lock().expect("execution service state poisoned");
285        match value {
286            Value::Future(handle) => {
287                self.validate_scope(handle.scope_id)?;
288                let record = state
289                    .futures
290                    .get_mut(&handle.id)
291                    .ok_or(ExecutionServiceError::UnknownHandle)?;
292                *record = FutureState::Cancelled;
293            }
294            Value::Task(handle) => {
295                self.validate_scope(handle.scope_id)?;
296                let future_id = {
297                    let record = state
298                        .tasks
299                        .get_mut(&handle.id)
300                        .ok_or(ExecutionServiceError::UnknownHandle)?;
301                    if record.generation != handle.generation {
302                        return Err(ExecutionServiceError::UnknownHandle);
303                    }
304                    record.cancelled = true;
305                    record.future_id
306                };
307                if let Some(future) = state.futures.get_mut(&future_id) {
308                    *future = FutureState::Cancelled;
309                }
310            }
311            _ => return Err(ExecutionServiceError::UnknownHandle),
312        }
313        Ok(())
314    }
315
316    fn drain_scope(&self, _reason: CancellationReason) {
317        let mut state = self.state.lock().expect("execution service state poisoned");
318        for future in state.futures.values_mut() {
319            if !matches!(future, FutureState::Completed(_)) {
320                *future = FutureState::Cancelled;
321            }
322        }
323        for task in state.tasks.values_mut() {
324            task.cancelled = true;
325        }
326    }
327}
328
329#[cfg(test)]
330mod tests {
331    use super::*;
332
333    #[test]
334    fn services_reject_handles_from_another_scope() {
335        let first = RuntimeExecutionService::new();
336        let second = RuntimeExecutionService::new();
337        let future = first
338            .create_future(DeferredCall {
339                function: 1,
340                arguments: vec![],
341                requested_outputs: 1,
342                program_revision: None,
343                program: None,
344            })
345            .unwrap();
346        assert_eq!(
347            second.spawn(&future),
348            Err(ExecutionServiceError::ForeignScope)
349        );
350    }
351
352    #[test]
353    fn future_task_lifecycle_is_backed_by_service_state() {
354        let service = RuntimeExecutionService::new();
355        let future = service
356            .create_future(DeferredCall {
357                function: 7,
358                arguments: vec![Value::Num(3.0)],
359                requested_outputs: 1,
360                program_revision: None,
361                program: None,
362            })
363            .unwrap();
364        let task = service.spawn(&future).unwrap();
365        let action = service.begin_await(Value::Task(task.clone())).unwrap();
366        let AwaitAction::ExecuteFuture { handle, call } = action else {
367            panic!("expected deferred execution");
368        };
369        assert_eq!(handle, future);
370        assert_eq!(call.function, 7);
371        service
372            .complete_future(&future, Ok(Value::Num(9.0)))
373            .unwrap();
374        assert_eq!(
375            service.begin_await(Value::Task(task)),
376            Err(ExecutionServiceError::UnknownHandle)
377        );
378    }
379
380    #[test]
381    fn cloned_invocation_context_inherits_exact_service() {
382        let service: std::rc::Rc<dyn RuntimeExecutionServices> =
383            std::rc::Rc::new(RuntimeExecutionService::new());
384        let parent = crate::context::RuntimeContext::new(service);
385        let nested = parent.clone();
386        assert_eq!(parent.execution().scope_id(), nested.execution().scope_id());
387    }
388
389    #[test]
390    fn independently_created_thread_sessions_have_distinct_scopes() {
391        let first = std::thread::spawn(|| RuntimeExecutionService::new().scope_id())
392            .join()
393            .unwrap();
394        let second = std::thread::spawn(|| RuntimeExecutionService::new().scope_id())
395            .join()
396            .unwrap();
397        assert_ne!(first, second);
398    }
399
400    #[test]
401    fn scope_drain_cancels_unfinished_children() {
402        let service = RuntimeExecutionService::new();
403        let future = service
404            .create_future(DeferredCall {
405                function: 1,
406                arguments: vec![],
407                requested_outputs: 1,
408                program_revision: None,
409                program: None,
410            })
411            .unwrap();
412        service.drain_scope(CancellationReason::Shutdown);
413        assert_eq!(
414            service.begin_await(Value::Future(future)),
415            Err(ExecutionServiceError::Cancelled)
416        );
417    }
418}
runmat_runtime/execution/services.rs

runmat_runtime/execution/
services.rs
