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

Run with:

dragon workflow.py

Going Further