1use std::{
2 collections::{BTreeMap, BTreeSet},
3 sync::{
4 Arc, RwLock,
5 atomic::{AtomicBool, AtomicU64, Ordering},
6 },
7};
8
9use sim_expr_tree_core::{
10 BackendKind, CodecPolicyPatch, EffectiveCodecPolicy, MountEpoch, MountResource,
11};
12use sim_incremental_core::{
13 ContinuationToken, IncrementalEngine, IncrementalError, ObservationKind, SnapshotBudgets,
14 ValueFingerprint,
15};
16use sim_kernel::{
17 CapabilitySet, Cx, DefaultFactory, EagerPolicy, Expr, StrictNames, Symbol, Value,
18};
19use sim_lib_stream_core::BufferPolicy;
20use sim_table_core::TablePath;
21
22use crate::ExprTreeRefPolicy;
23
24mod attempt;
25mod engine;
26mod eval;
27use eval::{evaluate_cell, observe_runtime_context, parent_path, path_key};
28mod face;
29pub use face::{
30 EncodedFace, FaceContent, FaceDimension, FaceIssue, FaceMetadata, FacePosition,
31 SourceEditOutcome,
32};
33mod model;
34pub use model::{
35 CalcError, CalcLimits, CalcQuery, CellFailure, HARD_MAX_EXPR_DEPTH, HARD_MAX_OBSERVATIONS,
36 HARD_MAX_OUTPUT, HARD_MAX_QUERY_DEPTH, HARD_MAX_WORK, LastGoodValue,
37};
38use model::{ContextFactory, MemoOutcome, MemoValue};
39mod policy;
40pub use policy::{
41 AuthorityDigest, AuthorityPolicyPatch, CalcPolicyPatch, CalcTrigger, CycleMode,
42 EffectiveAuthority, EffectiveCalcPolicy, ErrorMode, PolicyDigest,
43};
44use policy::{
45 effective_authority, effective_calc_policy, effective_codec_policy, is_descendant_or_same,
46};
47mod persistence;
48pub use persistence::{
49 DERIVED_SNAPSHOT_KEY, DerivedPersistReport, DerivedRestoreDisposition, DerivedRestoreReport,
50 DerivedSnapshotError, DerivedTableAdapter, GRAPH_SCHEMA_VERSION,
51};
52mod refresh;
53pub use refresh::{BackendRefreshSample, MountRefreshSource, RefreshError, RefreshReport};
54mod receipt;
55pub use receipt::{
56 CalcExplanation, CalcOutcome, CalcReason, CalcReceipt, CalcRequestMode, CalcStatus,
57 DependencyStamp, DirectedCalcReport, DirectedCellResult, EffectStamp, RequestId,
58};
59mod scheduler;
60use scheduler::MAX_READY_BYPASSES;
61pub use scheduler::{
62 AutomaticBudget, AutomaticContinuation, AutomaticQueueSnapshot, AutomaticRun, QueuedCalculation,
63};
64mod scheduling;
65mod session;
66mod value;
67mod watch;
68pub use watch::CalcWatch;
69
70const MAX_RECEIPT_DEPENDENCIES: usize = 64;
71const MAX_RECEIPT_GRAPH_NODES: usize = 4_096;
72const MAX_RECEIPT_GRAPH_EDGES: usize = 65_536;
73
74type WallClock = dyn Fn() -> Option<u64> + Send + Sync + 'static;
75
76pub struct ExprTreeCalc {
79 state: Arc<RwLock<CalcState>>,
80 engine: IncrementalEngine<CalcQuery, MemoValue>,
81 context_factory: Arc<ContextFactory>,
82 cancel_requested: Arc<AtomicBool>,
83 next_volatile: Arc<AtomicU64>,
84 wall_clock: Arc<RwLock<Arc<WallClock>>>,
85 next_request_id: u64,
86 automatic_queue: BTreeMap<String, QueuedCalculation>,
87 automatic_generation: u64,
88 next_queue_sequence: u64,
89 watches: Vec<CalcWatch>,
90 next_watch_id: u64,
91 refresh_sources: BTreeMap<String, Arc<dyn MountRefreshSource>>,
92 refresh_samples: BTreeMap<String, BackendRefreshSample>,
93 restored_continuations: BTreeSet<ContinuationToken>,
94}
95
96#[derive(Default)]
97pub(crate) struct CalcState {
98 cells: BTreeMap<String, Expr>,
99 bound_names: BTreeSet<String>,
100 bound_values: BTreeMap<Symbol, Value>,
101 mounts: BTreeMap<String, MountState>,
102 codec_registry_revision: u64,
103 tree_calc_policy: CalcPolicyPatch,
104 dir_calc_policies: BTreeMap<String, CalcPolicyPatch>,
105 cell_calc_policies: BTreeMap<String, CalcPolicyPatch>,
106 tree_codec_policy: CodecPolicyPatch,
107 dir_codec_policies: BTreeMap<String, CodecPolicyPatch>,
108 cell_codec_policies: BTreeMap<String, CodecPolicyPatch>,
109 authority_ceiling: CapabilitySet,
110 tree_authority_policy: AuthorityPolicyPatch,
111 dir_authority_policies: BTreeMap<String, AuthorityPolicyPatch>,
112 cell_authority_policies: BTreeMap<String, AuthorityPolicyPatch>,
113 active_request: Option<ActiveRequest>,
114 attempts: Vec<AttemptDraft>,
115 receipts: BTreeMap<String, CalcReceipt>,
116 next_logical_tick: u64,
117 current: BTreeMap<String, Result<Value, CalcError>>,
118 last_good: BTreeMap<String, Value>,
119 volatile: BTreeSet<String>,
120 failed_cells: BTreeSet<String>,
121 source_generation: u64,
122 control_generation: u64,
123}
124
125#[derive(Clone)]
126struct ActiveRequest {
127 id: RequestId,
128 reason: CalcReason,
129 directed_cells: BTreeSet<String>,
130 automatic: bool,
131}
132
133struct AttemptDraft {
134 request_id: RequestId,
135 cell: String,
136 policy: EffectiveCalcPolicy,
137 authority: EffectiveAuthority,
138 started_tick: u64,
139 finished_tick: u64,
140 wall_started_ms: Option<u64>,
141 wall_finished_ms: Option<u64>,
142 outcome: CalcOutcome,
143 effects: Vec<EffectStamp>,
144 omitted_effects: usize,
145 reason: CalcReason,
146}
147
148#[derive(Clone, Debug, PartialEq, Eq)]
149struct MountState {
150 resource: MountResource,
151 backend: BackendKind,
152 epoch: MountEpoch,
153}
154
155impl ExprTreeCalc {
156 #[must_use]
158 pub fn new() -> Self {
159 Self::with_context_factory(|| {
160 Cx::new(
161 Arc::new(ExprTreeRefPolicy::new(StrictNames(EagerPolicy))),
162 Arc::new(DefaultFactory),
163 )
164 })
165 }
166
167 #[must_use]
173 pub fn with_context_factory<F>(factory: F) -> Self
174 where
175 F: Fn() -> Cx + Send + Sync + 'static,
176 {
177 let context_factory: Arc<ContextFactory> = Arc::new(factory);
178 let open_time_authority = context_factory().capabilities().clone();
179 Self {
180 state: Arc::new(RwLock::new(CalcState {
181 authority_ceiling: open_time_authority,
182 next_logical_tick: 1,
183 ..CalcState::default()
184 })),
185 engine: IncrementalEngine::new(),
186 context_factory,
187 cancel_requested: Arc::new(AtomicBool::new(false)),
188 next_volatile: Arc::new(AtomicU64::new(1)),
189 wall_clock: Arc::new(RwLock::new(Arc::new(|| None))),
190 next_request_id: 1,
191 automatic_queue: BTreeMap::new(),
192 automatic_generation: 1,
193 next_queue_sequence: 1,
194 watches: Vec::new(),
195 next_watch_id: 1,
196 refresh_sources: BTreeMap::new(),
197 refresh_samples: BTreeMap::new(),
198 restored_continuations: BTreeSet::new(),
199 }
200 }
201
202 pub fn set_wall_clock<F>(&mut self, clock: F)
206 where
207 F: Fn() -> Option<u64> + Send + Sync + 'static,
208 {
209 *self.wall_clock.write().expect("wall clock lock poisoned") = Arc::new(clock);
210 }
211
212 #[must_use]
214 pub fn open_time_authority(&self) -> CapabilitySet {
215 self.state
216 .read()
217 .expect("calc state poisoned")
218 .authority_ceiling
219 .clone()
220 }
221
222 pub fn set_cell(&mut self, path: TablePath, source: Expr) {
224 let key = path_key(&path);
225 let (replaced, failed) = {
226 let mut state = self.state.write().expect("calc state poisoned");
227 let replaced = state.cells.insert(key.clone(), source).is_some();
228 bump_generation(&mut state.source_generation);
229 state.current.remove(&key);
230 state.volatile.remove(&key);
231 (
232 replaced,
233 state.failed_cells.iter().cloned().collect::<Vec<_>>(),
234 )
235 };
236 if !replaced {
237 self.register_cell_query(key.clone());
238 }
239 self.invalidate_cell_source(&path, !replaced);
240 self.invalidate_failed_cells(failed);
241 self.emit_change("source-set", &key);
242 self.schedule_dirty_automatic();
243 }
244
245 pub fn remove_cell(&mut self, path: &TablePath) {
247 let key = path_key(path);
248 let failed = {
249 let mut state = self.state.write().expect("calc state poisoned");
250 state.cells.remove(&key);
251 bump_generation(&mut state.source_generation);
252 state.current.remove(&key);
253 state.volatile.remove(&key);
254 state.failed_cells.iter().cloned().collect::<Vec<_>>()
255 };
256 self.register_cell_query(key.clone());
257 self.invalidate_cell_source(path, true);
258 self.invalidate_failed_cells(failed);
259 self.emit_change("source-removed", &key);
260 self.schedule_dirty_automatic();
261 }
262
263 pub fn move_cell(&mut self, from: &TablePath, to: TablePath) {
265 let from_key = path_key(from);
266 let to_key = path_key(&to);
267 let (moved, failed) = {
268 let mut state = self.state.write().expect("calc state poisoned");
269 let moved = state.cells.remove(&from_key);
270 if let Some(source) = moved.clone() {
271 state.cells.insert(to_key.clone(), source);
272 }
273 bump_generation(&mut state.source_generation);
274 state.current.remove(&from_key);
275 state.current.remove(&to_key);
276 state.volatile.remove(&from_key);
277 state.volatile.remove(&to_key);
278 let failed = state.failed_cells.iter().cloned().collect::<Vec<_>>();
279 (moved, failed)
280 };
281 if moved.is_some() {
282 self.register_cell_query(to_key.clone());
283 }
284 self.register_cell_query(from_key.clone());
285 self.invalidate_cell_source(from, true);
286 self.invalidate_cell_source(&to, true);
287 self.invalidate_failed_cells(failed);
288 self.emit_change("source-moved-from", &from_key);
289 self.emit_change("source-moved-to", &to_key);
290 self.schedule_dirty_automatic();
291 }
292
293 pub fn bind_name(&mut self, name: impl Into<String>) {
298 let name = name.into();
299 {
300 let mut state = self.state.write().expect("calc state poisoned");
301 state.bound_names.insert(name.clone());
302 bump_generation(&mut state.source_generation);
303 }
304 self.engine.invalidate(&CalcQuery::NameSlot(name));
305 }
306
307 pub fn bind_value(&mut self, name: Symbol, value: Value) {
309 {
310 let mut state = self.state.write().expect("calc state poisoned");
311 state.bound_values.insert(name.clone(), value);
312 bump_generation(&mut state.source_generation);
313 }
314 self.engine
315 .invalidate(&CalcQuery::NameSlot(name.to_string()));
316 }
317
318 pub fn cell_dependencies(
320 &mut self,
321 path: &TablePath,
322 ) -> Result<Vec<(CalcQuery, ObservationKind)>, IncrementalError<CalcQuery>> {
323 let key = CalcQuery::Cell(path_key(path));
324 let snapshot = self
325 .engine
326 .snapshot([key.clone()], SnapshotBudgets::default())?;
327 Ok(snapshot
328 .nodes
329 .iter()
330 .find(|node| node.key == key)
331 .map(|node| {
332 node.dependencies
333 .iter()
334 .map(|observation| (observation.key().clone(), observation.kind().clone()))
335 .collect()
336 })
337 .unwrap_or_default())
338 }
339
340 #[cfg(test)]
341 pub(crate) fn replace_context_factory<F>(&mut self, factory: F)
342 where
343 F: Fn() -> Cx + Send + Sync + 'static,
344 {
345 self.context_factory = Arc::new(factory);
346 }
347
348 #[cfg(test)]
349 pub(crate) fn state_for_lock_probe(&self) -> Arc<RwLock<CalcState>> {
350 Arc::clone(&self.state)
351 }
352}
353
354impl Default for ExprTreeCalc {
355 fn default() -> Self {
356 Self::new()
357 }
358}
359
360fn incremental_failure(
361 error: IncrementalError<CalcQuery>,
362) -> Result<MemoValue, IncrementalError<CalcQuery>> {
363 match error {
364 IncrementalError::Cycle { path } => Ok(MemoValue::failure(CellFailure::Cycle { path })),
365 IncrementalError::UnknownQuery { key } => Ok(MemoValue::failure(CellFailure::Evaluation {
366 message: format!("unknown dependency {key:?}"),
367 })),
368 IncrementalError::BudgetExceeded { .. }
369 | IncrementalError::Cancelled
370 | IncrementalError::UnknownContinuation { .. } => Err(error),
371 }
372}
373
374pub(super) fn bump_generation(generation: &mut u64) {
375 *generation = generation.saturating_add(1);
376}