Workflows

A Dragon workflow combines multiple capabilities — parallel processes, shared data, AI inference, and external executables — into a single, coordinated program. This page shows how to assemble those pieces into an end-to-end workflow that scales from a laptop to a supercomputer.

Building Blocks

Dragon workflows typically involve some combination of:

  • Batch — automatic DAG-based task scheduling across Python functions, executables, and MPI jobs (see also Batch Processing).

  • DDict — shared in-memory key-value store for passing data between stages without touching the filesystem.

  • ProcessGroup — fine-grained process placement and lifecycle management.

  • Dragon Channels — low-latency communication between workflow stages.

A Two-Stage Simulation + Analysis Workflow

The following example shows a common HPC pattern: a simulation stage writes output into a shared DDict, and an analysis stage reads from it and produces a result. Both stages run in parallel wherever possible.

Listing 92 workflow.py — simulation followed by analysis using Batch + DDict
 1import dragon
 2from multiprocessing import set_start_method
 3from dragon import Batch, DDict, System
 4import numpy as np
 5
 6
 7def simulate(dd, run_id, n_steps):
 8    """Pretend simulation: write a result array into the shared DDict."""
 9    result = np.cumsum(np.random.randn(n_steps))
10    dd[f"sim_{run_id}"] = result
11    return run_id
12
13
14def analyze(dd, run_id):
15    """Read simulation output and compute a summary statistic."""
16    data = dd[f"sim_{run_id}"]
17    return {"run_id": run_id, "mean": float(data.mean()), "std": float(data.std())}
18
19
20if __name__ == "__main__":
21    set_start_method("dragon")
22
23    alloc = System()
24    nnodes = alloc.nnodes
25
26    # Shared store for intermediate results — no filesystem round-trip needed
27    dd = DDict(managers_per_node=1, n_nodes=nnodes,
28               total_mem=nnodes * 256 * 1024 * 1024)
29
30    batch = Batch()
31
32    n_runs = 32
33
34    # Stage 1: submit all simulation tasks in parallel
35    sim_handles = [
36      batch.options(writes=[batch.write(dd, f"sim_{i}")])
37         .function(simulate, dd, i, 10_000)
38        for i in range(n_runs)
39    ]
40
41    # Stage 2: each analysis reads the key written by the corresponding simulation.
42    # Batch uses the reads/writes declarations to enforce correct ordering.
43    analysis_handles = [
44      batch.options(reads=[batch.read(dd, f"sim_{i}")])
45         .function(analyze, dd, i)
46        for i in range(n_runs)
47    ]
48
49    # Collect results
50    summaries = [h.get() for h in analysis_handles]
51    for s in sorted(summaries, key=lambda x: x["run_id"]):
52        print(f"Run {s['run_id']:02d}: mean={s['mean']:.4f}  std={s['std']:.4f}")
53
54    batch.join()
55    dd.destroy()

Run with:

dragon workflow.py

Going Further