somatize_core/
plan.rs

1//! The decided shape of an execution.
2//!
3//! A [`Graph`] says which nodes exist; a `Plan` says **how they are walked**.
4//! An enum and not a trait of executors: the ways of executing are a closed set,
5//! so the day a variant arrives the engine's `match` stops compiling and
6//! somebody has to decide, instead of falling into a wildcard arm.
7//!
8//! Every step carries **where its input comes from**, which is what makes a plan
9//! self-contained — executing never looks at the graph again — and why fans in
10//! both directions need no special variant.
11//!
12//! [`compile`] does not flatten the graph, it **decomposes** it, recovering the
13//! tree from the graph and never from the expression: the same graph built with
14//! `node()`/`edge()` in a loop has to give the same plan.
15//!
16//! | case | yields |
17//! |---|---|
18//! | no nodes | [`Plan::Empty`] |
19//! | one node | [`Plan::Execute`] |
20//! | the subgraph splits into components | [`Plan::Wave`], one branch per component |
21//! | there is a **series cut** | [`Plan::Sequence`] of the two sides |
22//! | no cut | flat sequence: it is not series-parallel |
23//!
24//! A **series cut** `(A, B)` is what a `>>` produces: the crossing edges run
25//! from **all** the sinks of `A` to **all** the sources of `B` and from nowhere
26//! else. Only the prefixes of a topological order need testing, since in a
27//! serial composition every node of `A` precedes every node of `B` in any
28//! topological order.
29//!
30//! There are DAGs without a tree — a theorem, not a gap here. The minimal
31//! forbidden pattern is the "N": `a→c`, `a→d`, `b→d`. See Valdes, Tarjan and
32//! Lawler, *The recognition of series parallel digraphs*, SIAM J. Comput. 11(2),
33//! 1982. The image of the DSL is **exactly** the series-parallel graphs, so the
34//! N is only reachable through `node()`/`edge()` and falls to the last case.
35//!
36//! [`compile`] does not see the [`Placement`]; [`distribute`] does, and wraps
37//! what runs elsewhere in [`Plan::Remote`]. Two steps because a
38//! [`Device`](crate::Device) is inert for the traversal and a [`Host`] is not.
39
40use crate::{Catalog, Graph, Host, NodeId, Placement};
41use std::fmt;
42
43/// How a graph is walked.
44///
45/// No `#[non_exhaustive]`: whoever executes has to decide for each variant.
46#[derive(Debug, Clone, PartialEq, Eq)]
47#[cfg_attr(feature = "serde", derive(serde::Serialize, serde::Deserialize))]
48pub enum Plan {
49    /// Nothing to do.
50    Empty,
51    /// Advance one node until it finishes.
52    Execute {
53        /// Which one.
54        node: NodeId,
55        /// Where its input comes from. Empty = the graph's input.
56        from: Vec<NodeId>,
57    },
58    /// One after another, in topological order. Each reads what it needs from
59    /// what has already been produced.
60    Sequence(Vec<Plan>),
61    /// Branches launched **at the same time**, one per connected component, so
62    /// they are disjoint. Each is a whole plan, so a branch runs start to
63    /// finish on one thread.
64    Wave(Vec<Plan>),
65    /// This whole slice executes elsewhere. A complete plan and not a step, so
66    /// a chain of five nodes on the same host is sent once.
67    Remote {
68        /// Where.
69        host: Host,
70        /// What runs there.
71        inner: Box<Plan>,
72    },
73}
74
75/// Decides how this graph is walked. The catalog is only consulted to check
76/// that every node has an implementation: the shape does not depend on what
77/// each one is.
78pub fn compile(graph: &Graph, catalog: &Catalog) -> Result<Plan, CompileError> {
79    if graph.is_empty() {
80        return Ok(Plan::Empty);
81    }
82
83    let order = graph.topological_sort();
84    for node in &order {
85        if catalog.get(node).is_none() {
86            return Err(CompileError::NoImplementation((*node).clone()));
87        }
88    }
89
90    Ok(decompose(graph, &order))
91}
92
93/// One step of a plan: a node, and where its input comes from.
94#[derive(Debug, Clone, Copy, PartialEq, Eq)]
95pub struct Step<'p> {
96    /// Which node.
97    pub node: &'p NodeId,
98    /// Which nodes it reads. Empty means the graph's input.
99    pub from: &'p [NodeId],
100}
101
102/// What decides where one part of a plan runs.
103#[derive(Debug, Clone, Copy, PartialEq, Eq)]
104pub enum Destination<'p> {
105    /// A node, whose host — if it has one — the [`Placement`] knows.
106    Node(&'p NodeId),
107    /// A slice that already says where it goes.
108    Away(&'p Host),
109}
110
111impl Plan {
112    /// Every step, in declaration order, **wherever it runs**: a
113    /// [`Remote`](Plan::Remote) is entered, because what a plan does does not
114    /// depend on where.
115    pub fn steps(&self) -> impl Iterator<Item = Step<'_>> {
116        Steps { left: vec![self] }
117    }
118
119    /// What decides where each part of this plan runs, in declaration order.
120    /// Differs from [`steps`](Self::steps) in one line: a
121    /// [`Remote`](Plan::Remote) is **not** entered, which is what makes
122    /// [`distribute`] idempotent.
123    pub fn destinations(&self) -> impl Iterator<Item = Destination<'_>> {
124        Destinations { left: vec![self] }
125    }
126}
127
128/// The stack of [`Plan::steps`]. Children are pushed in reverse so that popping
129/// gives them back in the order they were declared, which is observable.
130struct Steps<'p> {
131    left: Vec<&'p Plan>,
132}
133
134impl<'p> Iterator for Steps<'p> {
135    type Item = Step<'p>;
136
137    fn next(&mut self) -> Option<Self::Item> {
138        while let Some(plan) = self.left.pop() {
139            match plan {
140                Plan::Empty => {}
141                Plan::Execute { node, from } => return Some(Step { node, from }),
142                Plan::Sequence(plans) | Plan::Wave(plans) => self.left.extend(plans.iter().rev()),
143                Plan::Remote { inner, .. } => self.left.push(inner),
144            }
145        }
146        None
147    }
148}
149
150/// The stack of [`Plan::destinations`]. The same walk, stopping where the other
151/// descends.
152struct Destinations<'p> {
153    left: Vec<&'p Plan>,
154}
155
156impl<'p> Iterator for Destinations<'p> {
157    type Item = Destination<'p>;
158
159    fn next(&mut self) -> Option<Self::Item> {
160        while let Some(plan) = self.left.pop() {
161            match plan {
162                Plan::Empty => {}
163                Plan::Execute { node, .. } => return Some(Destination::Node(node)),
164                Plan::Sequence(plans) | Plan::Wave(plans) => self.left.extend(plans.iter().rev()),
165                Plan::Remote { host, .. } => return Some(Destination::Away(host)),
166            }
167        }
168        None
169    }
170}
171
172/// Wraps the slices that run on another host in [`Plan::Remote`], grouping as
173/// much as it can and descending only where a slice is spread across places.
174/// Idempotent; a plan with no hosts comes out unchanged.
175pub fn distribute(plan: &Plan, placement: &Placement) -> Plan {
176    if placement.is_local() {
177        return plan.clone();
178    }
179    wrap(plan, placement)
180}
181
182/// Where everything inside a plan runs.
183enum Where {
184    /// There are no nodes, so it runs nowhere.
185    Nothing,
186    /// All in the same place: a host, or — with `None` — here.
187    All(Option<Host>),
188    /// In more than one place, so it has to be descended into and split.
189    Mixed,
190}
191
192fn wrap(plan: &Plan, placement: &Placement) -> Plan {
193    if matches!(plan, Plan::Remote { .. }) {
194        return plan.clone();
195    }
196    match uniform(plan, placement) {
197        Where::All(Some(host)) => Plan::Remote {
198            host,
199            inner: Box::new(plan.clone()),
200        },
201        Where::All(None) | Where::Nothing => plan.clone(),
202        Where::Mixed => match plan {
203            Plan::Sequence(plans) => Plan::Sequence(runs(plans, placement)),
204            // One by one, without regrouping two of the same host: that would
205            // change their declaration order, which is observable.
206            Plan::Wave(branches) => {
207                Plan::Wave(branches.iter().map(|p| wrap(p, placement)).collect())
208            }
209            Plan::Empty | Plan::Execute { .. } | Plan::Remote { .. } => plan.clone(),
210        },
211    }
212}
213
214/// The steps of a sequence, merging **consecutive** runs bound for the same
215/// host — consecutive only, because the order is the topological one.
216fn runs(plans: &[Plan], placement: &Placement) -> Vec<Plan> {
217    let mut out: Vec<Plan> = Vec::new();
218    let mut run: Vec<Plan> = Vec::new();
219    let mut destination: Option<Host> = None;
220
221    for plan in plans {
222        let here = match plan {
223            Plan::Remote { .. } => None,
224            _ => match uniform(plan, placement) {
225                Where::All(Some(host)) => Some(host),
226                Where::All(None) | Where::Nothing | Where::Mixed => None,
227            },
228        };
229        if here != destination {
230            close(&mut out, &mut run, destination.take());
231            destination = here;
232        }
233        match destination {
234            Some(_) => run.push(plan.clone()),
235            None => out.push(wrap(plan, placement)),
236        }
237    }
238    close(&mut out, &mut run, destination);
239    out
240}
241
242/// Closes the open run, if any, as a single trip.
243fn close(out: &mut Vec<Plan>, run: &mut Vec<Plan>, destination: Option<Host>) {
244    let Some(host) = destination else { return };
245    // A run of one is not wrapped in a sequence of one: the shape cannot depend
246    // on how you arrived at it.
247    let inner = match run.len() {
248        1 => run.remove(0),
249        _ => Plan::Sequence(std::mem::take(run)),
250    };
251    out.push(Plan::Remote {
252        host,
253        inner: Box::new(inner),
254    });
255}
256
257/// Whether the whole plan lands in the same place. `None` means "here".
258fn uniform(plan: &Plan, placement: &Placement) -> Where {
259    let places: Vec<Option<Host>> = plan
260        .destinations()
261        .map(|destination| match destination {
262            Destination::Node(node) => placement.host_of(node).cloned(),
263            Destination::Away(host) => Some(host.clone()),
264        })
265        .collect();
266    match places.split_first() {
267        None => Where::Nothing,
268        Some((first, rest)) if rest.iter().all(|host| host == first) => Where::All(first.clone()),
269        Some(_) => Where::Mixed,
270    }
271}
272
273/// The shape of a subset of nodes, in topological order. The subset is always
274/// closed under paths, which is why reachability inside the subgraph coincides
275/// with reachability in the whole graph.
276fn decompose<'g>(graph: &'g Graph, nodes: &[&'g NodeId]) -> Plan {
277    match nodes {
278        [] => Plan::Empty,
279        [only] => step(graph, only),
280        _ => {
281            let parts = components(graph, nodes);
282            if parts.len() > 1 {
283                return Plan::Wave(parts.iter().map(|part| decompose(graph, part)).collect());
284            }
285
286            let Some(cut) = series_cut(graph, nodes) else {
287                // No cut, no tree: walked in sequence, as before waves existed.
288                return Plan::Sequence(nodes.iter().map(|node| step(graph, node)).collect());
289            };
290
291            // Flattening the recursion on the right leaves `Sequence` with its
292            // steps in a row rather than nested.
293            let mut steps = vec![decompose(graph, &nodes[..cut])];
294            match decompose(graph, &nodes[cut..]) {
295                Plan::Sequence(rest) => steps.extend(rest),
296                other => steps.push(other),
297            }
298            Plan::Sequence(steps)
299        }
300    }
301}
302
303/// A lone step, with where its input comes from — the whole graph's
304/// predecessors, not the subset's.
305fn step(graph: &Graph, node: &NodeId) -> Plan {
306    Plan::Execute {
307        node: node.clone(),
308        from: graph.predecessors(node).into_iter().cloned().collect(),
309    }
310}
311
312/// The connected components — ignoring direction — of the subgraph, each
313/// keeping the input's topological order and ordered by their first node.
314fn components<'g>(graph: &'g Graph, nodes: &[&'g NodeId]) -> Vec<Vec<&'g NodeId>> {
315    let mut unassigned: Vec<bool> = vec![true; nodes.len()];
316    let mut out = Vec::new();
317
318    for start in 0..nodes.len() {
319        if !unassigned[start] {
320            continue;
321        }
322        unassigned[start] = false;
323        let mut group = vec![start];
324        let mut frontier = vec![start];
325
326        while let Some(i) = frontier.pop() {
327            for j in 0..nodes.len() {
328                if unassigned[j] && adjacent(graph, nodes[i], nodes[j]) {
329                    unassigned[j] = false;
330                    group.push(j);
331                    frontier.push(j);
332                }
333            }
334        }
335
336        group.sort_unstable();
337        out.push(group.into_iter().map(|i| nodes[i]).collect());
338    }
339    out
340}
341
342/// Whether there is an edge between the two, in either direction.
343fn adjacent(graph: &Graph, a: &NodeId, b: &NodeId) -> bool {
344    graph.successors(a).contains(&b) || graph.successors(b).contains(&a)
345}
346
347/// Where the sequence splits: the smallest series cut, if there is one.
348fn series_cut(graph: &Graph, nodes: &[&NodeId]) -> Option<usize> {
349    (1..nodes.len()).find(|cut| is_series_cut(graph, &nodes[..*cut], &nodes[*cut..]))
350}
351
352/// Whether `before >> after` is exactly what lies between the two: nothing
353/// crosses outside the ends, and every sink reaches every source.
354fn is_series_cut(graph: &Graph, before: &[&NodeId], after: &[&NodeId]) -> bool {
355    let sinks: Vec<&NodeId> = before
356        .iter()
357        .copied()
358        .filter(|node| !graph.successors(node).iter().any(|s| before.contains(s)))
359        .collect();
360    let sources: Vec<&NodeId> = after
361        .iter()
362        .copied()
363        .filter(|node| !graph.predecessors(node).iter().any(|p| after.contains(p)))
364        .collect();
365
366    let crosses_outside_the_ends = before.iter().any(|node| {
367        graph
368            .successors(node)
369            .iter()
370            .any(|succ| after.contains(succ) && !(sinks.contains(node) && sources.contains(succ)))
371    });
372    if crosses_outside_the_ends {
373        return false;
374    }
375
376    sinks.iter().all(|sink| {
377        let onward = graph.successors(sink);
378        sources.iter().all(|source| onward.contains(source))
379    })
380}
381
382/// Why it was not possible to decide how to walk the graph.
383#[derive(Debug, Clone, PartialEq, Eq)]
384pub enum CompileError {
385    /// The node is in the graph but nobody registered what it does.
386    NoImplementation(NodeId),
387}
388
389impl fmt::Display for CompileError {
390    fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result {
391        match self {
392            Self::NoImplementation(id) => {
393                write!(f, "node `{id}` has no registered implementation")
394            }
395        }
396    }
397}
398
399impl std::error::Error for CompileError {}
somatize_core/plan.rs

somatize_core/
plan.rs
