turbo_tasks/
persisted_graph.rs

1use anyhow::Result;
2use serde::{Deserialize, Serialize, ser::SerializeSeq};
3
4use crate::{
5    CellId, RawVc, TaskId,
6    backend::{CachedTaskType, CellContent},
7    task::shared_reference::TypedSharedReference,
8};
9
10#[derive(Clone, Debug)]
11pub enum TaskCell {
12    Content(CellContent),
13    NeedComputation,
14}
15
16impl Default for TaskCell {
17    fn default() -> Self {
18        TaskCell::Content(CellContent(None))
19    }
20}
21
22#[derive(Serialize, Deserialize, Debug)]
23pub struct TaskData {
24    pub children: Vec<TaskId>,
25    pub dependencies: Vec<RawVc>,
26    pub cells: TaskCells,
27    pub output: RawVc,
28}
29
30/// A newtype struct that intercepts serde. This is required
31/// because for safety reasons, TaskCell<()> is not allowed to
32/// be deserialized.
33///
34/// We augment it with type data then write it. This is inefficient
35/// on disk but could be alleviated later.
36#[derive(Debug)]
37pub struct TaskCells(pub Vec<(CellId, TaskCell)>);
38
39// the on-disk representation of a task cell. it is local to this impl
40// to prevent users accidentally ser/de the untyped data
41#[derive(Serialize, Deserialize)]
42struct SerializableTaskCell(Option<Option<TypedSharedReference>>);
43impl From<SerializableTaskCell> for TaskCell {
44    fn from(val: SerializableTaskCell) -> Self {
45        match val.0 {
46            Some(d) => TaskCell::Content(d.map(TypedSharedReference::into_untyped).into()),
47            None => TaskCell::NeedComputation,
48        }
49    }
50}
51
52impl Serialize for TaskCells {
53    fn serialize<S>(&self, serializer: S) -> Result<S::Ok, S::Error>
54    where
55        S: serde::Serializer,
56    {
57        let mut seq = serializer.serialize_seq(Some(self.0.len()))?;
58        for (cell_id, cell) in &self.0 {
59            let task_cell = SerializableTaskCell(match cell {
60                TaskCell::Content(CellContent(opt)) => {
61                    Some(opt.clone().map(|d| d.into_typed(cell_id.type_id)))
62                }
63                TaskCell::NeedComputation => None,
64            });
65            seq.serialize_element(&(cell_id, task_cell))?;
66        }
67        seq.end()
68    }
69}
70
71impl<'de> Deserialize<'de> for TaskCells {
72    fn deserialize<D>(deserializer: D) -> Result<Self, D::Error>
73    where
74        D: serde::Deserializer<'de>,
75    {
76        let data: Vec<(CellId, SerializableTaskCell)> = Vec::deserialize(deserializer)?;
77        Ok(TaskCells(
78            data.into_iter()
79                .map(|(id, cell)| (id, cell.into()))
80                .collect(),
81        ))
82    }
83}
84
85pub struct ReadTaskState {
86    pub clean: bool,
87    pub keeps_external_active: bool,
88}
89
90pub struct PersistTaskState {
91    pub externally_active: bool,
92}
93
94/*
95
96There are 4 kinds of task:
97
98(A) A task that exists only in memory.
99(B) A task that exists in persistent graph and in memory (either "store" or "read" has been called)
100(C) A task that exists only in persistent graph.
101
102Parent-child relationships:
103
104(A) as child: active_parents is tracked only in memory.
105(B) as child: active_parents is tracked in memory and either as internal_active_parents or external_active_parents in the persisted graph.
106(C) as child: either as internal_active_parents or external_active_parents in the persisted graph.
107
108(A) as parent: It will use external_active_parents for (B) or (C) as child.
109               update_active_parents() is used to modify the external_active_parents count.
110(B) as parent: It will use internal_active_parents for (B) or (C) as child.
111               compute_active() returns the changes needed for (A) or (C) as child
112(C) as parent: It will use internal_active_parents for (B) or (C) as child.
113               compute_active() returns the changes needed for (A) or (C) as child
114
115(A) as child of (B) or (C): active count tracked as external_active_children, have task ids assigned in persistent graph
116
117*/
118
119#[derive(Debug)]
120pub struct ActivateResult {
121    /// Keeps the external version of the task active
122    pub keeps_external_active: bool,
123
124    /// Task doesn't live in the persisted graph but
125    /// should be track externally
126    pub external: bool,
127
128    /// Task is dirty and need to be scheduled for execution
129    pub dirty: bool,
130
131    /// Further tasks that need to be activated that
132    /// didn't fit into that batch
133    pub more_tasks_to_activate: Vec<TaskId>,
134}
135
136#[derive(Debug)]
137pub struct PersistResult {
138    /// Tasks that need to be activated
139    pub tasks_to_activate: Vec<TaskId>,
140
141    /// Tasks that need to be deactivated
142    pub tasks_to_deactivate: Vec<TaskId>,
143}
144
145#[derive(Debug)]
146pub struct DeactivateResult {
147    /// Further tasks that need to be deactivated that
148    /// didn't fit into that batch
149    pub more_tasks_to_deactivate: Vec<TaskId>,
150}
151
152pub trait PersistedGraph: Sync + Send {
153    /// read task data and state for a specific task.
154    fn read(
155        &self,
156        task: TaskId,
157        api: &dyn PersistedGraphApi,
158    ) -> Result<Option<(TaskData, ReadTaskState)>>;
159
160    /// lookup all cache entries for a partial task type
161    /// returns true if all cache entries has been initialized
162    /// returns false if that were too many
163    fn lookup(
164        &self,
165        partial_task_type: &CachedTaskType,
166        api: &dyn PersistedGraphApi,
167    ) -> Result<bool>;
168
169    /// lookup one cache entry
170    fn lookup_one(
171        &self,
172        task_type: &CachedTaskType,
173        api: &dyn PersistedGraphApi,
174    ) -> Result<Option<TaskId>>;
175
176    /// checks if a task is persisted
177    fn is_persisted(&self, task: TaskId, api: &dyn PersistedGraphApi) -> Result<bool>;
178
179    /// store a completed task into the persisted graph
180    /// together with dependencies, children and cells.
181    /// Returns false, if the task failed to persist.
182    fn persist(
183        &self,
184        task: TaskId,
185        data: TaskData,
186        state: PersistTaskState,
187        api: &dyn PersistedGraphApi,
188    ) -> Result<Option<PersistResult>>;
189
190    /// Activate a task in the persisted graph when active_parents > 0 or it's
191    /// externally kept alive.
192    fn activate_when_needed(
193        &self,
194        task: TaskId,
195        api: &dyn PersistedGraphApi,
196    ) -> Result<Option<ActivateResult>>;
197
198    /// Deactivate a task in the persisted graph when active_parents == 0 and
199    /// it's not externally kept alive.
200    fn deactivate_when_needed(
201        &self,
202        task: TaskId,
203        api: &dyn PersistedGraphApi,
204    ) -> Result<Option<DeactivateResult>>;
205
206    /// Marks a task as kept alive by the consumer graph
207    /// (usually from memory to persisted graph)
208    /// Returns true when activate_when_needed should be called soonish
209    fn set_externally_active(&self, task: TaskId, api: &dyn PersistedGraphApi) -> Result<bool>;
210
211    /// No longer marks a task as kept alive by the consumer graph
212    /// (usually from memory to persisted graph)
213    /// Returns true when deactivate_when_needed should be called soonish
214    fn unset_externally_active(&self, task: TaskId, api: &dyn PersistedGraphApi) -> Result<bool>;
215
216    /// Removes all external keep alives that were not renewed this round.
217    /// This is usually called after the initial build has finished and all
218    /// external keep alives has been renewed.
219    fn remove_outdated_externally_active(&self, api: &dyn PersistedGraphApi)
220    -> Result<Vec<TaskId>>;
221
222    /// update the dirty flag for a stored task
223    /// Returns true, when the task is active and should be scheduled
224    fn make_dirty(&self, task: TaskId, api: &dyn PersistedGraphApi) -> Result<bool>;
225
226    /// update the dirty flag for a stored task
227    fn make_clean(&self, task: TaskId, api: &dyn PersistedGraphApi) -> Result<()>;
228
229    /// make all tasks that depend on that vc dirty and
230    /// return a list of active tasks that should be scheduled
231    fn make_dependent_dirty(&self, vc: RawVc, api: &dyn PersistedGraphApi) -> Result<Vec<TaskId>>;
232
233    /// Get all tasks that are active, but not persisted.
234    /// This is usually called at beginning to create and schedule
235    /// tasks that are missing in the persisted graph
236    fn get_active_external_tasks(&self, api: &dyn PersistedGraphApi) -> Result<Vec<TaskId>>;
237
238    /// Get all tasks that are dirty and active.
239    /// This is usually called at the beginning to schedule these tasks.
240    fn get_dirty_active_tasks(&self, api: &dyn PersistedGraphApi) -> Result<Vec<TaskId>>;
241
242    /// Get tasks that have active update pending that need to be continued
243    /// returns (tasks_to_activate, tasks_to_deactivate)
244    fn get_pending_active_update(
245        &self,
246        api: &dyn PersistedGraphApi,
247    ) -> Result<(Vec<TaskId>, Vec<TaskId>)>;
248
249    /// Stop operations
250    #[allow(unused_variables)]
251    fn stop(&self, api: &dyn PersistedGraphApi) -> Result<()> {
252        Ok(())
253    }
254}
255
256pub trait PersistedGraphApi {
257    fn get_or_create_task_type(&self, ty: CachedTaskType) -> TaskId;
258
259    fn lookup_task_type(&self, id: TaskId) -> &CachedTaskType;
260}
261
262/*
263
264read:
265
266  data: (TaskId) => (TaskData)
267  cache: (CachedTaskType) => (TaskId)
268  type: (TaskId) => (CachedTaskType)
269
270read_dependents:
271
272  dependents: (RawVc) => [TaskId]
273
274store:
275
276  external_active_parents: (TaskId) -> (usize)
277  internal_active_parents: (TaskId) -> (usize)
278  inactive_tasks: [TaskId]
279
280B+C?
281
282
283
284
285*/
286
287impl PersistedGraph for () {
288    fn read(
289        &self,
290        _task: TaskId,
291        _api: &dyn PersistedGraphApi,
292    ) -> Result<Option<(TaskData, ReadTaskState)>> {
293        Ok(None)
294    }
295
296    fn lookup(
297        &self,
298        _partial_task_type: &CachedTaskType,
299        _api: &dyn PersistedGraphApi,
300    ) -> Result<bool> {
301        Ok(false)
302    }
303
304    fn lookup_one(
305        &self,
306        _task_type: &CachedTaskType,
307        _api: &dyn PersistedGraphApi,
308    ) -> Result<Option<TaskId>> {
309        Ok(None)
310    }
311
312    fn is_persisted(&self, _task: TaskId, _api: &dyn PersistedGraphApi) -> Result<bool> {
313        Ok(false)
314    }
315
316    fn persist(
317        &self,
318        _task: TaskId,
319        _data: TaskData,
320        _state: PersistTaskState,
321        _api: &dyn PersistedGraphApi,
322    ) -> Result<Option<PersistResult>> {
323        Ok(None)
324    }
325
326    fn activate_when_needed(
327        &self,
328        _task: TaskId,
329        _api: &dyn PersistedGraphApi,
330    ) -> Result<Option<ActivateResult>> {
331        Ok(None)
332    }
333
334    fn deactivate_when_needed(
335        &self,
336        _task: TaskId,
337        _api: &dyn PersistedGraphApi,
338    ) -> Result<Option<DeactivateResult>> {
339        Ok(None)
340    }
341
342    fn set_externally_active(&self, _task: TaskId, _api: &dyn PersistedGraphApi) -> Result<bool> {
343        Ok(false)
344    }
345
346    fn unset_externally_active(&self, _task: TaskId, _api: &dyn PersistedGraphApi) -> Result<bool> {
347        Ok(false)
348    }
349
350    fn remove_outdated_externally_active(
351        &self,
352        _api: &dyn PersistedGraphApi,
353    ) -> Result<Vec<TaskId>> {
354        Ok(Vec::new())
355    }
356
357    fn make_dirty(&self, _task: TaskId, _api: &dyn PersistedGraphApi) -> Result<bool> {
358        Ok(false)
359    }
360
361    fn make_clean(&self, _task: TaskId, _api: &dyn PersistedGraphApi) -> Result<()> {
362        Ok(())
363    }
364
365    fn make_dependent_dirty(
366        &self,
367        _vc: RawVc,
368        _api: &dyn PersistedGraphApi,
369    ) -> Result<Vec<TaskId>> {
370        Ok(Vec::new())
371    }
372
373    fn get_active_external_tasks(&self, _api: &dyn PersistedGraphApi) -> Result<Vec<TaskId>> {
374        Ok(Vec::new())
375    }
376
377    fn get_dirty_active_tasks(&self, _api: &dyn PersistedGraphApi) -> Result<Vec<TaskId>> {
378        Ok(Vec::new())
379    }
380
381    fn get_pending_active_update(
382        &self,
383        _api: &dyn PersistedGraphApi,
384    ) -> Result<(Vec<TaskId>, Vec<TaskId>)> {
385        Ok((Vec::new(), Vec::new()))
386    }
387}
turbo_tasks/persisted_graph.rs

turbo_tasks/
persisted_graph.rs
