DAG-Based Task Execution

Overview

A Directed Acyclic Graph (DAG) is a graph structure where edges have direction and no cycles exist—meaning you can never start at a node and follow edges back to the same node. This seemingly simple constraint makes DAGs the ideal abstraction for modeling task dependencies: if task B depends on the output of task A, an edge from A to B encodes that relationship, and the absence of cycles guarantees that the dependency chain always terminates.

DAG-based task execution engines are at the heart of virtually every data pipeline, build system, and workflow orchestrator. Apache Airflow models workflows as DAGs. Make, Bazel, and Buck use DAGs to determine build order. Even CI/CD systems like GitHub Actions express job dependencies as DAGs. The core problem is always the same: given a set of tasks with inter-dependencies, determine a valid execution order that respects those dependencies, and maximize parallelism where possible.

In a typical data pipeline system, DAG execution powers ETL-style pipelines where data transformations flow through multiple stages. Each stage may depend on one or more upstream stages, and the engine must execute them in topologically valid order while exploiting parallelism across independent branches of the graph.

Core Concepts

Graph Representation

A DAG is typically represented using an adjacency list: each node maintains a list of its outgoing edges (successors). For task execution, we also need the reverse view—each node’s incoming edges (predecessors)—so we can determine when all of a task’s dependencies have been satisfied. An alternative representation uses an adjacency matrix, but for sparse task graphs (where most tasks depend on only a few others), the adjacency list is far more space-efficient.

In-Degree and Ready Queues

The in-degree of a node is the number of incoming edges—equivalently, the number of unfinished dependencies. A task is “ready” when its in-degree reaches zero. This insight drives Kahn’s algorithm: initialize all in-degrees, enqueue nodes with in-degree 0, then repeatedly dequeue a ready node, execute it, and decrement the in-degrees of its successors, enqueuing any that reach zero. The ready queue naturally exposes parallelism: all nodes in the queue at any given moment are independent and can execute concurrently.

Cycle Detection

If a user-defined workflow accidentally introduces a cycle (A → B → C → A), execution would deadlock: every node in the cycle has a non-zero in-degree, so none ever becomes ready. Cycle detection is therefore critical validation before execution begins. Kahn’s algorithm provides a natural check: if the number of processed nodes is less than the total number of nodes, the graph contains at least one cycle. DFS-based detection uses a three-color marking scheme (white/gray/black) to identify back edges.

Parallelism and Scheduling

Given unlimited workers, all tasks at the same “depth level” (same distance from the source nodes) can run concurrently. In practice, you have a bounded worker pool, and the scheduler must decide which ready tasks to dispatch first. Greedy strategies that prioritize tasks on the critical path tend to minimize total execution time. For resource-constrained environments, list scheduling algorithms (such as HEFT—Heterogeneous Earliest Finish Time) assign tasks to workers based on estimated finish times.

How It Works

Kahn’s Algorithm for Topological Sort

Kahn’s algorithm is the most intuitive approach to topological sorting and forms the backbone of most DAG execution engines. The algorithm works as follows:

1. Compute the in-degree of every node.
2. Enqueue all nodes with in-degree 0 into a ready queue.
3. While the queue is non-empty: dequeue a node, add it to the sorted output, and for each of its successors, decrement the successor’s in-degree. If a successor’s in-degree reaches 0, enqueue it.
4. If the sorted output contains fewer nodes than the graph, a cycle exists.

The time complexity is O(V + E) where V is the number of nodes (tasks) and E is the number of edges (dependencies). This is optimal since every node and every edge must be examined at least once.

DFS-Based Topological Sort

The DFS approach performs a depth-first traversal, appending each node to the output when all of its descendants have been visited (post-order). Reversing the post-order gives a valid topological sort. This approach is often preferred when you also need to detect strongly connected components (Tarjan’s algorithm is a natural extension).

Concurrent Execution with Worker Pools

A production DAG executor goes beyond simple topological sorting. It maintains a thread-safe ready queue and dispatches tasks to a pool of workers. When a worker completes a task, it atomically decrements the in-degrees of all successors and enqueues any that become ready. This requires careful synchronization: atomic counters or mutexes for in-degree updates, and a concurrent queue (or channel) for the ready set.

A typical pipeline uses Go’s goroutines and channels for this: a dispatcher goroutine reads from a ready channel and fans tasks out to a bounded worker pool. Workers send completion notifications back to the dispatcher, which updates in-degrees and feeds newly ready tasks into the channel. This pattern naturally handles dynamic parallelism without explicit thread management.

Error Handling and Partial Failure

When a task in the DAG fails, the executor must decide how to handle downstream tasks. Common strategies include: (1) fail-fast, where the entire DAG execution is aborted; (2) skip downstream, where only tasks transitively dependent on the failed task are skipped; and (3) continue on failure, where the executor marks the failed branch and continues executing independent branches. Strategy (2) is most common in production pipelines, as it maximizes the amount of useful work completed while preventing execution of tasks with missing inputs.

Practical Applications

ETL Pipelines

A typical data pipeline uses DAG execution to orchestrate multi-stage data transformations. A typical pipeline might: (1) extract raw data from multiple sources, (2) apply independent transformations to each source, (3) join the transformed datasets, and (4) load the result into a target store. Steps 1 and 2 can run in parallel across sources, while step 3 must wait for all transformations to complete. The DAG structure makes these dependencies explicit and the executor handles scheduling automatically.

Build Systems

Build tools model compilation units as DAG nodes and header/library dependencies as edges. A change to a single source file triggers recompilation of only that file and its transitive dependents, rather than the entire project. Incremental builds exploit the DAG structure to minimize redundant work by tracking which nodes have changed since the last build.

CI/CD Pipeline Orchestration

Modern CI/CD systems express job dependencies as DAGs. A pipeline might run linting and unit tests in parallel, then run integration tests only if both pass, then deploy to staging, and finally run smoke tests against the staging environment. Each stage’s dependencies are edges in the DAG, and the CI system automatically parallelizes independent jobs.

Query Execution Plans

Database query optimizers produce execution plans that are DAGs of operators: scans, filters, joins, aggregations, and sorts. The edges represent data flow between operators. Parallel query execution engines schedule operators using the same topological principles as task executors, running independent subtrees concurrently across multiple cores.

Implementation Patterns

The Task Interface

A clean DAG executor starts with a task interface that separates the execution logic from the graph structure. Each task implements a Run(ctx context.Context) error method, and the graph is defined separately as a set of dependency edges between task IDs. This separation allows the same executor to run arbitrary workloads without modification.

Dynamic DAGs

Some workflows need to generate tasks dynamically based on the output of earlier tasks. For example, a “list files” task might produce a variable number of “process file” tasks. Dynamic DAGs extend the static model by allowing tasks to register new nodes and edges at runtime. The executor must handle this carefully to avoid introducing cycles and to correctly update in-degree tracking for already-running tasks.

Checkpointing and Resumability

Long-running pipelines should persist their execution state so they can resume after failures without re-executing completed tasks. A checkpoint records which tasks have completed successfully and their output locations. On restart, the executor loads the checkpoint, marks completed tasks, recomputes in-degrees for the remaining graph, and resumes from the ready set. This pattern is essential for multi-hour batch jobs where re-execution from scratch is prohibitively expensive.

Code Examples

DAG Executor in Go

This example implements a concurrent DAG executor using goroutines and channels, similar to the pattern used in a typical data pipeline.

package dag

import (
	"context"
	"fmt"
	"sync"
	"sync/atomic"
)

// Task represents a unit of work in the DAG.
type Task interface {
	ID() string
	Run(ctx context.Context) error
}

// DAG holds the graph structure and executes tasks concurrently.
type DAG struct {
	tasks    map[string]Task
	edges    map[string][]string // task ID -> successor IDs
	inDegree map[string]*int32   // atomic in-degree counters
}

// New creates a DAG from tasks and dependency edges.
// edges maps each task ID to the IDs it depends ON (predecessors).
func New(tasks []Task, deps map[string][]string) (*DAG, error) {
	d := &DAG{
		tasks:    make(map[string]Task),
		edges:    make(map[string][]string),
		inDegree: make(map[string]*int32),
	}

	for _, t := range tasks {
		d.tasks[t.ID()] = t
		zero := int32(0)
		d.inDegree[t.ID()] = &zero
	}

	// Build forward edges and compute in-degrees.
	for taskID, predecessors := range deps {
		for _, pred := range predecessors {
			d.edges[pred] = append(d.edges[pred], taskID)
			atomic.AddInt32(d.inDegree[taskID], 1)
		}
	}

	return d, nil
}

// Execute runs all tasks respecting dependencies.
// maxWorkers controls the level of concurrency.
func (d *DAG) Execute(ctx context.Context, maxWorkers int) error {
	ready := make(chan string, len(d.tasks))
	errs := make(chan error, 1)
	var wg sync.WaitGroup

	// Enqueue tasks with in-degree 0.
	for id, deg := range d.inDegree {
		if atomic.LoadInt32(deg) == 0 {
			ready <- id
		}
	}

	var completed int32

	// Launch worker pool.
	sem := make(chan struct{}, maxWorkers)
	go func() {
		for id := range ready {
			sem <- struct{}{}
			wg.Add(1)
			go func(taskID string) {
				defer func() { <-sem; wg.Done() }()

				task := d.tasks[taskID]
				if err := task.Run(ctx); err != nil {
					select {
					case errs <- fmt.Errorf("task %s failed: %w", taskID, err):
					default:
					}
					return
				}

				atomic.AddInt32(&completed, 1)

				// Decrement in-degrees of successors.
				for _, succ := range d.edges[taskID] {
					if atomic.AddInt32(d.inDegree[succ], -1) == 0 {
						ready <- succ
					}
				}
			}(id)
		}
	}()

	// Wait for all tasks or first error.
	done := make(chan struct{})
	go func() {
		for {
			if int(atomic.LoadInt32(&completed)) == len(d.tasks) {
				close(done)
				return
			}
		}
	}()

	select {
	case err := <-errs:
		return err
	case <-done:
		wg.Wait()
		close(ready)
		return nil
	case <-ctx.Done():
		return ctx.Err()
	}
}

Topological Sort with Cycle Detection in Python

This implementation of Kahn’s algorithm returns a valid execution order or raises an error if the graph contains a cycle.

from collections import defaultdict, deque
from typing import Dict, List, Set


def topological_sort(
    nodes: Set[str],
    edges: Dict[str, List[str]],  # node -> list of successors
) -> List[str]:
    """
    Return a topological ordering of nodes using Kahn's algorithm.
    Raises ValueError if the graph contains a cycle.
    """
    # Compute in-degrees.
    in_degree: Dict[str, int] = {node: 0 for node in nodes}
    for src, successors in edges.items():
        for dst in successors:
            in_degree[dst] = in_degree.get(dst, 0) + 1

    # Initialize the ready queue with zero in-degree nodes.
    queue = deque(node for node, deg in in_degree.items() if deg == 0)
    result: List[str] = []

    while queue:
        node = queue.popleft()
        result.append(node)

        for successor in edges.get(node, []):
            in_degree[successor] -= 1
            if in_degree[successor] == 0:
                queue.append(successor)

    if len(result) != len(nodes):
        unvisited = nodes - set(result)
        raise ValueError(
            f"Graph contains a cycle involving: {unvisited}"
        )

    return result


# Example: ETL pipeline DAG
nodes = {"extract_a", "extract_b", "transform_a", "transform_b", "join", "load"}
edges = {
    "extract_a": ["transform_a"],
    "extract_b": ["transform_b"],
    "transform_a": ["join"],
    "transform_b": ["join"],
    "join": ["load"],
}

order = topological_sort(nodes, edges)
print("Execution order:", order)
# Possible output: ['extract_a', 'extract_b', 'transform_a',
#                    'transform_b', 'join', 'load']

Common Pitfalls

1. Missing cycle detection before execution. If you skip validation and a cycle exists, the executor will silently deadlock—tasks in the cycle never become ready, and the pipeline hangs indefinitely. Always validate the DAG before starting execution.
2. Unbounded parallelism overwhelming resources. A DAG with many independent tasks at the same depth level can spawn hundreds of concurrent workers, exhausting memory or overwhelming downstream services. Always bound your worker pool and consider rate-limiting for tasks that access shared resources.
3. Non-idempotent tasks causing corruption on retry. If a task appends to a file or increments a counter, retrying it after a partial failure will produce incorrect results. Design tasks to be idempotent: write to temporary locations and atomically rename on success.
4. Ignoring the critical path when optimizing. Optimizing a task that runs in parallel with the critical path does not reduce total execution time. Profile the DAG to identify the critical path, then focus optimization efforts there.
5. Overly fine-grained task decomposition. Making every single operation its own DAG node adds scheduling overhead (queue operations, context switches, state tracking) that can exceed the benefit of parallelism. Group small, sequential operations into single tasks and only split at natural parallelism boundaries.

Key Takeaways