DAG workflow execution

Deep dive

How it works

Parts:

• task graph — nodes (command, duration, input) and edges
• in-degree counters — unfinished inputs per task; zero means ready
• ready queue — ordered by priority, the longest remaining path to a sink
• worker pool — fixed slots; per task a timeout and retry limit
• output cache — keys of finished outputs, local or remote
• state table — done, cached, failed, skipped (upstream_failed in Airflow)

1. Submit: Kahn's algorithm; leftovers mean a cycle: reject. prio[v] = dur[v] + max prio[child], in reverse order.
2. Key: key[v] = hash(input[v], key of each dep): a changed leaf changes every descendant's key and nothing else.
3. Dispatch: while a slot is free, pop the top-priority ready task. Key cached: release children. Else run it.
4. Finish: success stores the key and decrements children's counters. Failure: retry up to the limit, then mark failed.
5. Propagate: children of a failed task never reach zero and end skipped (Airflow trigger rules can override).
6. Rerun: only tasks whose key changed execute, the downstream closure of the edit.

src ──► clean ──► train ──► publish      edit `report`: only
          └─────► report ───┘            report + publish re-run

Extensions: backfill = one DAG instance per time partition; dynamic mapping = a task's output sets how many copies of the next task exist (Airflow expand); Spark cuts stages at shuffles.

Code

Simulated time keeps runs deterministic. Asserts cover order, cycles, retries, failure skipping, a one-edit rerun and FIFO versus critical-path priority.

use std::collections::{HashSet, VecDeque};

pub struct Task { pub dur: u64, pub deps: Vec<usize>, pub input: u64 }
#[derive(Clone, Copy, PartialEq, Debug)]
pub enum St { Done, Cached, Failed, Skipped }
#[derive(Clone, Copy, PartialEq)]
pub enum Policy { Fifo, CriticalPath }
pub struct Report { pub makespan: u64, pub st: Vec<St>, pub ran: Vec<usize> }

fn fnv(h: u64, x: u64) -> u64 { // FNV-1a over the 8 bytes of x
    x.to_le_bytes().iter().fold(h, |h, &b| (h ^ b as u64).wrapping_mul(0x100000001b3))
}
/// Kahn at submit: topological order, or Err(tasks on/behind a cycle).
pub fn topo(t: &[Task]) -> Result<Vec<usize>, Vec<usize>> {
    let mut indeg: Vec<usize> = t.iter().map(|x| x.deps.len()).collect();
    let mut kids = vec![vec![]; t.len()];
    for (i, x) in t.iter().enumerate() { for &d in &x.deps { kids[d].push(i); } }
    let mut order: Vec<usize> = (0..t.len()).filter(|&i| indeg[i] == 0).collect();
    let mut h = 0;
    while h < order.len() {
        let u = order[h]; h += 1;
        for &v in &kids[u] { indeg[v] -= 1; if indeg[v] == 0 { order.push(v); } }
    }
    if order.len() == t.len() { Ok(order) } else { Err((0..t.len()).filter(|&i| indeg[i] > 0).collect()) }
}
/// Merkle key: hash of own input and the keys of all dependencies.
pub fn keys(t: &[Task], order: &[usize]) -> Vec<u64> {
    let mut k = vec![0u64; t.len()];
    for &i in order {
        k[i] = t[i].deps.iter().fold(fnv(0xcbf29ce484222325, t[i].input), |h, &d| fnv(h, k[d]));
    }
    k
}
/// Simulated-time executor; `fails(task, attempt)` injects failures.
pub fn run(t: &[Task], workers: usize, pol: Policy, retries: u32,
           fails: &dyn Fn(usize, u32) -> bool, cache: &mut HashSet<u64>) -> Result<Report, Vec<usize>> {
    let order = topo(t)?;
    let key = keys(t, &order);
    let mut prio = vec![0u64; t.len()]; // longest path to a sink
    let mut kids = vec![vec![]; t.len()];
    for (i, x) in t.iter().enumerate() { for &d in &x.deps { kids[d].push(i); } }
    for &i in order.iter().rev() { prio[i] = t[i].dur + kids[i].iter().map(|&c| prio[c]).max().unwrap_or(0); }
    let mut indeg: Vec<usize> = t.iter().map(|x| x.deps.len()).collect();
    let mut ready: VecDeque<(usize, u32)> = (0..t.len()).filter(|&i| indeg[i] == 0).map(|i| (i, 0)).collect();
    let mut st = vec![St::Skipped; t.len()]; // never ran = skipped
    let (mut running, mut now, mut ran) = (Vec::<(u64, usize, u32)>::new(), 0u64, Vec::<usize>::new());
    loop {
        while running.len() < workers && !ready.is_empty() {
            let pos = if pol == Policy::Fifo { 0 } else {
                (0..ready.len()).max_by_key(|&p| (prio[ready[p].0], std::cmp::Reverse(ready[p].0))).unwrap()
            };
            let (i, a) = ready.remove(pos).unwrap();
            if cache.contains(&key[i]) { // hit: release children at once
                st[i] = St::Cached;
                for &c in &kids[i] { indeg[c] -= 1; if indeg[c] == 0 { ready.push_back((c, 0)); } }
            } else { running.push((now + t[i].dur, i, a)); }
        }
        let Some(&(next, ..)) = running.iter().min_by_key(|r| r.0) else { break };
        now = next;
        let mut fin: Vec<_> = running.iter().filter(|r| r.0 == now).cloned().collect();
        running.retain(|r| r.0 != now);
        fin.sort_by_key(|r| r.1);
        for (_, i, a) in fin {
            ran.push(i);
            if fails(i, a) {
                if a < retries { ready.push_back((i, a + 1)); } else { st[i] = St::Failed; }
            } else {
                st[i] = St::Done;
                cache.insert(key[i]);
                for &c in &kids[i] { indeg[c] -= 1; if indeg[c] == 0 { ready.push_back((c, 0)); } }
            }
        }
    }
    Ok(Report { makespan: now, st, ran })
}
fn main() {
    let mk = |dur, deps: &[usize], input| Task { dur, deps: deps.to_vec(), input };
    // extract(0) -> clean(1) -> {train(2), report(3)} -> publish(4); lint(5) independent
    let mut t = vec![mk(5, &[], 1), mk(3, &[0], 2), mk(8, &[1], 3), mk(2, &[1], 4), mk(1, &[2, 3], 5), mk(2, &[], 6)];
    let mut cache = HashSet::new();
    let none = |_: usize, _: u32| false;
    // chain 5+3+8+1 = 17: deps respected
    let r = run(&t, 2, Policy::CriticalPath, 0, &none, &mut cache).unwrap();
    assert_eq!(r.makespan, 17);
    // cycle rejected at submit
    let cyc = [mk(1, &[1], 0), mk(1, &[0], 0)];
    assert!(run(&cyc, 1, Policy::Fifo, 0, &none, &mut HashSet::new()).is_err());
    // train fails 3 times (2 retries): only `publish` is skipped
    let bad = |i: usize, _a: u32| i == 2;
    let r = run(&t, 2, Policy::CriticalPath, 2, &bad, &mut HashSet::new()).unwrap();
    assert_eq!(r.st, [St::Done, St::Done, St::Failed, St::Done, St::Skipped, St::Done]);
    assert_eq!(r.ran.iter().filter(|&&i| i == 2).count(), 3);
    // edit `report`: only report + publish re-execute
    t[3].input = 40;
    let r = run(&t, 2, Policy::CriticalPath, 0, &none, &mut cache).unwrap();
    assert_eq!(r.st.iter().filter(|&&s| s == St::Cached).count(), 4);
    assert_eq!(r.ran, [3, 4]);
    // synthetic random DAG (LCG): FIFO vs critical-path priority
    let mut x = 12345u64;
    let mut rnd = |m: u64| { x = x.wrapping_mul(6364136223846793005).wrapping_add(1442695040888963407); (x >> 33) % m };
    let big: Vec<Task> = (0..400usize).map(|i| {
        let mut deps = if i < 20 { vec![] } else { vec![i - 1 - rnd(20) as usize, i - 1 - rnd(20) as usize] };
        deps.sort(); deps.dedup();
        Task { dur: if rnd(10) == 0 { 40 + rnd(60) } else { 1 + rnd(9) }, deps, input: i as u64 }
    }).collect();
    for w in [4, 8] {
        let f = run(&big, w, Policy::Fifo, 0, &none, &mut HashSet::new()).unwrap().makespan;
        let c = run(&big, w, Policy::CriticalPath, 0, &none, &mut HashSet::new()).unwrap().makespan;
        println!("workers {w}: fifo {f}  critical-path {c}");
        assert!(c <= f);
    }
}

Measured on the 400-task random DAG (longest path 1424, work 4560): 4 workers FIFO 1584, critical-path priority 1425 (11% less); 8 workers 1435 vs 1424.

Complexity and constants

• Submit O(V + E), 4-8 B of counters per task. Dispatch: O(1) per edge, O(log V) per pick with a heap.
• Graham: any list schedule is within 2 - 1/m of optimal.
• Per-task overhead (process start, state write): order of ms locally, seconds with a container start.

Efficiency

• Cost: one task = one dispatch, state write and hash lookup (order of µs to ms). Bound: critical path (wide graphs), scheduler latency (tiny tasks), data movement (big ones).
• Levers:
  • Content-hash cache, local or remote — skips unchanged work (Bazel, Nextflow docs) — deterministic tasks only.
  • Critical-path priority — 11% shorter makespan at 4 workers (measured above); HEFT ranks tasks by a similar upward rank — more ready tasks than slots.
  • Fuse tiny tasks — one dispatch and process start per batch — many sub-second tasks.
  • Data locality — run where the input lives (Dask scheduling policies) — large intermediates.
  • Pools — Ninja pools cap memory-bound tasks — avoids thrashing.
• Naive → standard → state of the art:
  • One task at a time, rebuild all · sum of runtimes, every time.
  • Counter-driven pool, retries, local cache · critical path or work/slots; rerun follows the edit.
  • Path priority, remote cache, fusion, locality · near the bound; reruns mostly hits.

Implementation notes

• Hash command, tool version, env and input bytes; cryptographic hash for shared caches (content-addressed storage).
• Write outputs to a temp path, rename on success.
• Persist task state in a log (batch checkpoint-restart); tasks run at least once: make them idempotent.
• Park exhausted tasks in a dead-letter queue.

Pitfalls

• Nondeterministic tasks (timestamps, unordered output) defeat the cache downstream.
• Undeclared inputs give stale hits; a missing edge hides until the schedule order changes.
• Unbounded dynamic mapping floods the scheduler; cap the width.

Variants and alternatives

In production

Airflow, Dagster, Prefect, Argo Workflows (pipelines); Bazel, Ninja, Nix (builds); Dask, Spark (task graphs); Snakemake, Nextflow (hash-based resume); render-farm managers.

Implementations

Sources

• Mokhov, Mitchell, Peyton Jones, Build systems à la carte, Proc. ACM Program. Lang. 2 (ICFP), 2018. paper
• Topcuoglu, Hariri, Wu, Performance-effective and low-complexity task scheduling for heterogeneous computing (HEFT), IEEE TPDS 13(3), 2002. paper
• R. Graham, Bounds on Multiprocessing Timing Anomalies, SIAM J. Appl. Math. 17(2), 1969. paper
• Docs: Airflow DAGs, task mapping, Bazel caching, Ninja, Dask, Nextflow, Spark