from collections import defaultdict
from dataclasses import dataclass, field
from typing import Callable, Dict, List, Set
class CycleError(Exception):
def __init__(self, nodes):
self.nodes = nodes
super().__init__(f"dependency cycle detected among: {sorted(nodes)}")
@dataclass
class Task:
name: str
run: Callable
dependencies: Set[str] = field(default_factory=set)
class TaskGraph:
def __init__(self):
self._tasks: Dict[str, Task] = {}
def add_task(self, name, run, depends_on=()):
deps = set(depends_on)
missing = deps - set(self._tasks)
if missing:
raise KeyError(f"{name} depends on undefined tasks: {sorted(missing)}")
self._tasks[name] = Task(name=name, run=run, dependencies=deps)
return self
def _indegrees(self):
indegree = {name: len(t.dependencies) for name, t in self._tasks.items()}
dependents = defaultdict(set)
for name, task in self._tasks.items():
for dep in task.dependencies:
dependents[dep].add(name)
return indegree, dependents
def topological_waves(self) -> List[List[str]]:
indegree, dependents = self._indegrees()
waves: List[List[str]] = []
remaining = dict(indegree)
while remaining:
ready = sorted(n for n, d in remaining.items() if d == 0)
if not ready:
raise CycleError(set(remaining))
waves.append(ready)
for node in ready:
del remaining[node]
for child in dependents[node]:
if child in remaining:
remaining[child] -= 1
return waves
def task(self, name) -> Task:
return self._tasks[name]
import logging
from concurrent.futures import ThreadPoolExecutor, as_completed
from typing import Any, Dict
from taskgraph import TaskGraph
log = logging.getLogger("task-runner")
class TaskRunner:
def __init__(self, graph: TaskGraph, max_workers: int = 4):
self.graph = graph
self.max_workers = max_workers
def run(self, context: Dict[str, Any] = None) -> Dict[str, Any]:
context = context or {}
waves = self.graph.topological_waves()
log.info("resolved %d execution waves", len(waves))
with ThreadPoolExecutor(max_workers=self.max_workers) as pool:
for depth, wave in enumerate(waves):
log.info("wave %d: %s", depth, wave)
futures = {
pool.submit(self._invoke, name, context): name
for name in wave
}
for future in as_completed(futures):
name = futures[future]
context[name] = future.result()
return context
def _invoke(self, name, context):
task = self.graph.task(name)
upstream = {dep: context.get(dep) for dep in task.dependencies}
try:
return task.run(upstream)
except Exception:
log.exception("task %s failed", name)
raise
import logging
from runner import TaskRunner
from taskgraph import TaskGraph
logging.basicConfig(level=logging.INFO)
def build_pipeline() -> TaskGraph:
g = TaskGraph()
g.add_task("extract", run=lambda ctx: [1, 2, 3, 4])
g.add_task("clean", run=lambda ctx: [x for x in ctx["extract"]], depends_on=["extract"])
g.add_task("enrich", run=lambda ctx: sum(ctx["extract"]), depends_on=["extract"])
g.add_task(
"load",
run=lambda ctx: {"rows": ctx["clean"], "total": ctx["enrich"]},
depends_on=["clean", "enrich"],
)
return g
if __name__ == "__main__":
graph = build_pipeline()
result = TaskRunner(graph, max_workers=3).run()
print("final:", result["load"])
This snippet models a workflow as a directed acyclic graph (DAG) of tasks and computes a safe execution order using Kahn's algorithm, then runs independent tasks concurrently. The core idea of topological sorting is that if task B depends on task A, then A must appear before B in any valid ordering; a topological sort produces exactly such an ordering, and the absence of one signals a cycle in the dependency graph.
In taskgraph.py, the TaskGraph stores each Task by name and tracks dependencies per node. add_task registers a node and validates that referenced dependencies are declared, which catches typos early rather than at run time. The _indegrees helper computes how many unmet dependencies each task has — the in-degree in graph terms. This is the input Kahn's algorithm consumes.
The method topological_waves is the heart of the design. Instead of returning a flat list, it groups tasks into waves: each wave is a set of tasks whose dependencies are all already satisfied, so every task in a wave can run in parallel. The algorithm repeatedly collects all zero-indegree nodes as one wave, removes them, and decrements the in-degree of their dependents. When no more zero-indegree nodes remain but tasks are still pending, that residue is a cycle, and CycleError is raised with the offending nodes — a far more useful failure than an infinite loop or silent stall. Sorting node names keeps output deterministic, which matters for tests and reproducibility.
In runner.py, TaskRunner.run walks the waves in order and dispatches each wave to a ThreadPoolExecutor, blocking until the whole wave completes before starting the next. This preserves the dependency guarantee while extracting maximum parallelism at each level. Results are threaded through a shared context dict so downstream tasks can read upstream outputs.
The trade-off is that wave-based scheduling can under-utilize workers when waves are uneven; a fully event-driven scheduler that starts a task the instant its parents finish squeezes out more parallelism at the cost of complexity. For most batch pipelines and build systems, wave scheduling is simpler and good enough. A developer reaches for this pattern whenever ordered, dependency-constrained work must run reliably — CI builds, data pipelines, or migration orchestration.
Related snips
use crossbeam::channel::unbounded;
use std::thread;
fn main() {
let (tx, rx) = unbounded();Crossbeam for advanced concurrent data structures
// Creating a Promise
const myPromise = new Promise((resolve, reject) => {
const success = true;
setTimeout(() => {
if (success) {Promises and async/await patterns for asynchronous JavaScript
use std::sync::mpsc;
use std::thread;
fn main() {
let (tx, rx) = mpsc::channel();Channels (mpsc) for message passing between threads
export type Settled<R> =
| { status: 'fulfilled'; value: R }
| { status: 'rejected'; reason: unknown };
export interface ConcurrencyOptions {
limit: number;Simple concurrency limiter for batch operations
package pools
import (
"bytes"
"sync"
)sync.Pool for bytes.Buffer to reduce allocations in hot paths
import Foundation
import UIKit
class ImageProcessor {
// Background processing with main thread updates
func processImage(_ image: UIImage, completion: @escaping (UIImage?) -> Void) {Grand Central Dispatch for concurrency
Share this code
Here's the card — post it anywhere.
