From 82cbf15b85533b052213f81e8c1f762826940ce9 Mon Sep 17 00:00:00 2001 From: Stephen Waits Date: Wed, 8 Oct 2025 15:48:30 -0600 Subject: [PATCH] docs: clean up, split up, restructure README.md --- Cargo.toml | 2 +- README.md | 1003 ++++--------------------------------- docs/COMPARISONS.md | 274 ++++++++++ docs/CYCLE_PREVENTION.md | 82 +++ docs/DESIGN_PHILOSOPHY.md | 113 +++++ docs/TRACING.md | 55 ++ src/runner.rs | 8 +- 7 files changed, 638 insertions(+), 899 deletions(-) create mode 100644 docs/COMPARISONS.md create mode 100644 docs/CYCLE_PREVENTION.md create mode 100644 docs/DESIGN_PHILOSOPHY.md create mode 100644 docs/TRACING.md diff --git a/Cargo.toml b/Cargo.toml index cfd2270..a45c129 100644 --- a/Cargo.toml +++ b/Cargo.toml @@ -8,7 +8,7 @@ edition = "2021" rust-version = "1.81" authors = ["Stephen Waits "] license = "MIT" -description = "A minimal, type-safe, runtime-agnostic async DAG executor with compile-time dependency validation" +description = "A minimal, type-safe, runtime-agnostic async DAG (Directed Acyclic Graph) executor with compile-time cycle prevention and true parallel execution" repository = "https://github.com/swaits/dagx" homepage = "https://github.com/swaits/dagx" documentation = "https://docs.rs/dagx" diff --git a/README.md b/README.md index 91b57ac..47aca81 100644 --- a/README.md +++ b/README.md @@ -6,328 +6,42 @@ [![License: MIT](https://img.shields.io/badge/License-MIT-blue.svg)](LICENSE) [![Rust Version](https://img.shields.io/badge/rust-1.81+-blue.svg)](https://www.rust-lang.org) -A minimal, type-safe, runtime-agnostic async DAG (Directed Acyclic Graph) executor with compile-time cycle prevention, compile-time dependency validation, and true parallel execution. +A minimal, type-safe, runtime-agnostic async DAG (Directed Acyclic Graph) executor with compile-time cycle prevention and true parallel execution. -**Why dagx?** **Cycles are impossible** via the type system—zero runtime overhead. No custom scheduler—just battle-tested primitives (channels, async/await). Compile-time type safety catches wiring errors before runtime. Works with ANY async runtime (Tokio, async-std, smol). Simple API: `#[task]`, `add_task()`, `depends_on()`—that's it. **4-148x faster than dagrs** across all workloads with sub-microsecond overhead per task. +## Why dagx? - +### 🚀 Blazing Fast: 1.04-129x faster than dagrs -- [Features](#features) -- [Optional Tracing Support](#optional-tracing-support) - - [Enabling Tracing](#enabling-tracing) - - [Usage Example](#usage-example) - - [Log Levels](#log-levels) - - [Zero-Cost Guarantee](#zero-cost-guarantee) -- [Design Philosophy: Primitives as Scheduler](#design-philosophy-primitives-as-scheduler) - - [How It Works](#how-it-works) - - [The Implementation](#the-implementation) - - [Inline Execution Fast-Path](#inline-execution-fast-path) - - [Benefits](#benefits) - - [The Insight](#the-insight) - - [Measured Overhead](#measured-overhead) -- [Quick Start](#quick-start) -- [Comparison with Similar Projects](#comparison-with-similar-projects) - - [Developer Experience: API Comparison](#developer-experience-api-comparison) - - [Quick Comparison](#quick-comparison) - - [Performance Benchmarks vs dagrs](#performance-benchmarks-vs-dagrs) - - [Detailed Comparison](#detailed-comparison) - - [dagrs (Most Mature)](#dagrs-most-mature) - - [async_dag (Clean Type Safety)](#asyncdag-clean-type-safety) - - [dag-flow (Experimental)](#dag-flow-experimental) - - [RenovZ/dag-runner (Simple Edge-Based)](#renovzdag-runner-simple-edge-based) - - [tasksitter (Dynamic Workflows)](#tasksitter-dynamic-workflows) - - [When to Choose dagx](#when-to-choose-dagx) - - [When to Consider Alternatives](#when-to-consider-alternatives) -- [Core Concepts](#core-concepts) - - [Task](#task) - - [Task Patterns](#task-patterns) - - [DagRunner](#dagrunner) - - [TaskHandle](#taskhandle) -- [Examples](#examples) - - [Fan-out Pattern (1 → n)](#fan-out-pattern-1-n) - - [Fan-in Pattern (m → 1)](#fan-in-pattern-m-1) - - [Many-to-Many Pattern (m ↔ n)](#many-to-many-pattern-m-n) -- [Type-State Pattern](#type-state-pattern) -- [Runtime Agnostic](#runtime-agnostic) -- [Important Limitations](#important-limitations) - - [Tasks Cannot Return Bare Tuples](#tasks-cannot-return-bare-tuples) -- [Examples](#examples) - - [Tutorial Examples](#tutorial-examples) - - [Reference Examples](#reference-examples) -- [When to Use dagx](#when-to-use-dagx) -- [Performance](#performance) - - [Performance Characteristics](#performance-characteristics) - - [Automatic Arc Wrapping (No Manual Arc Needed!)](#automatic-arc-wrapping-no-manual-arc-needed) -- [Documentation](#documentation) -- [License](#license) -- [Contributing](#contributing) - +| Workload | Tasks | dagx | dagrs | Speedup | +| ----------------- | ------ | ------- | ------- | ------------------ | +| Sequential chain | 5 | 3.0 µs | 385 µs | **129x faster** 🚀 | +| Sequential chain | 100 | 79 µs | 703 µs | **8.9x faster** | +| Diamond pattern | 4 | 11 µs | 387 µs | **34x faster** | +| Fan-out (1→100) | 101 | 155 µs | 595 µs | **3.85x faster** | +| Independent tasks | 10,000 | 12.7 ms | 13.3 ms | **1.04x faster** | -## Features +**Per-task overhead:** -- **Compile-time cycle prevention**: The type system makes cycles **impossible**—no runtime checks needed! See [Compile-Time Cycle Prevention](#compile-time-cycle-prevention) below. -- **Compile-time type safety**: Dependencies are validated at compile time through the type system. No `dyn Any`, no downcasting, no runtime type errors. -- **Runtime-agnostic**: Works with any async runtime (Tokio, async-std, smol, and more) -- **True parallelism**: Tasks spawn to multiple threads for genuine parallel execution -- **Type-state pattern**: The API prevents incorrect dependency wiring through compile-time errors -- **Zero-cost abstractions**: Leverages generics and monomorphization for minimal overhead -- **Flexible task patterns**: Supports stateless, read-only, and mutable state tasks -- **Simple API**: Just `#[task]`, `DagRunner`, `TaskHandle`, and `TaskBuilder` -- **Comprehensive error handling**: Result-based errors with actionable messages -- **Optional tracing**: Zero-cost observability via optional `tracing` feature flag +- Construction: ~100 ns/task +- Inline execution (sequential): ~790 ns/task +- Parallel execution: ~1.3 µs/task -## Compile-Time Cycle Prevention +### 🛡️ Compile-Time Safety -**dagx guarantees acyclic graphs at compile time with zero runtime overhead.** Unlike other DAG libraries that detect cycles at runtime, dagx uses Rust's type system to make cycles impossible to express in the first place. +- **Cycles are impossible** — the type system prevents them at compile time, zero runtime overhead +- **No runtime type errors** — dependencies validated at compile time +- **Compiler-verified correctness** — no surprise failures in production -### Why This Matters +See [how it works](docs/CYCLE_PREVENTION.md). -Most DAG libraries check for cycles when you execute the graph: +### ✨ Simple API ```rust -// Other libraries: cycle detected at RUNTIME -let dag = DagRunner::new(); -let a = dag.add_task(TaskA); -let b = dag.add_task(TaskB); -a.depends_on(b); -b.depends_on(a); // Compiles fine... -dag.run().await?; // ❌ Error: "Cycle detected!" (at runtime) +let sum = dag.add_task(Add).depends_on((&x, &y)); +dag.run(|fut| tokio::spawn(fut)).await?; ``` -**dagx prevents this at compile time:** - -```rust -// dagx: cycle prevented at COMPILE TIME -let dag = DagRunner::new(); -let a_builder = dag.add_task(TaskA); -let b_builder = dag.add_task(TaskB); - -// To make A depend on B, we need B as a TaskHandle. -// But calling depends_on() consumes the builder! -let b_handle = b_builder.depends_on(&some_source); - -// Now we can't make B depend on A because b_builder was moved! -// This won't compile: -// a_builder.depends_on(&b_handle); // ✓ OK: A→B -// b_handle.depends_on(&a_handle); // ❌ ERROR: TaskHandle has no depends_on() method! -``` - -### How It Works: The Type-State Pattern - -dagx uses two types to enforce acyclic structure: - -1. **`TaskBuilder`**: Mutable builder that can have dependencies added via `depends_on()` - - Consumed (moved) when you call `depends_on()` or convert to `TaskHandle` - - Can only be used once for wiring - -2. **`TaskHandle`**: Immutable reference to a finalized task - - Copy type (just wraps a task ID) - - Can be used as a dependency for other tasks - - **Has no `depends_on()` method** - can't be modified after creation - -This creates an impossible ordering constraint for cycles: - -```rust -// Attempting a cycle A→B→A: -let a_builder = dag.add_task(TaskA); // TaskBuilder -let b_builder = dag.add_task(TaskB); // TaskBuilder - -// To wire A→B, we need B as a TaskHandle -let b_handle = b_builder.into(); // Convert B to TaskHandle (consumes builder) - -// To wire B→A, we need A as a TaskHandle -let a_handle = a_builder.depends_on(&b_handle); // A now depends on B - -// Now if we try to complete the cycle: -// a_handle.depends_on(&...); // ❌ ERROR: no method named `depends_on` found -``` - -**The type system enforces strict topological ordering!** You can't create a `TaskHandle` until all its dependencies exist as handles, which prevents cycles. - -### Proof & Examples - -See [`src/cycle_prevention.rs`](src/cycle_prevention.rs) for comprehensive documentation with `compile_fail` tests proving various cycle attempts don't compile. - -See [`tests/cycle_prevention.rs`](tests/cycle_prevention.rs) for runtime tests demonstrating the type-state pattern in action. - -### Benefits - -- ✅ **Zero runtime cost**: No cycle detection code to execute -- ✅ **Catches errors early**: Cycles are caught at compile time, not in production -- ✅ **Compiler-verified**: The type system proves your graph is acyclic -- ✅ **No mental overhead**: You can't accidentally create cycles even if you try - -This is a true **zero-cost abstraction**—the type system provides safety guarantees without any runtime checks or performance penalty. - -## Optional Tracing Support - -dagx provides optional observability through the `tracing` crate with **zero runtime overhead when disabled**. The tracing instrumentation is conditionally compiled using feature flags, meaning when the feature is off, the logging code doesn't exist in the compiled binary at all. - -### Enabling Tracing - -```toml -[dependencies] -dagx = { version = "0.3", features = ["tracing"] } -tracing-subscriber = "0.3" -``` - -### Usage Example - -```rust -use tracing_subscriber::{fmt, EnvFilter}; - -// Initialize tracing subscriber -fmt() - .with_env_filter( - EnvFilter::try_from_default_env() - .unwrap_or_else(|_| EnvFilter::new("dagx=info")) - ) - .init(); - -// Your DAG code here - tracing will log execution details -``` - -### Log Levels - -- **INFO**: DAG execution start/completion -- **DEBUG**: Task additions, dependency wiring, layer computation -- **TRACE**: Individual task execution (inline vs spawned), detailed execution flow -- **ERROR**: Task panics, concurrent execution attempts - -Run with different levels: - -```bash -RUST_LOG=dagx=info cargo run # High-level execution info -RUST_LOG=dagx=debug cargo run # Task and layer details -RUST_LOG=dagx=trace cargo run # All execution details -``` - -See [`examples/tracing_example.rs`](examples/tracing_example.rs) for a complete working example. - -### Zero-Cost Guarantee - -When the `tracing` feature is disabled (the default), there is **literally 0ns overhead**: - -- Logging code is removed at compile time via `#[cfg(feature = "tracing")]` -- No branches, no function calls, nothing -- The `tracing` crate isn't even linked -- Benchmarks verify identical performance with/without the feature - -This follows the same pattern used by Tokio, Hyper, and other performance-critical Rust async libraries. - -## Design Philosophy: Primitives as Scheduler - -dagx takes an unconventional approach to task orchestration: **there is no traditional scheduler**. Instead of complex scheduling logic managing when tasks run, the entire system is built on communication and synchronization primitives that do the scheduling themselves. - -### How It Works - -Traditional DAG executors contain substantial scheduling code—algorithms that track task states, manage dependencies, coordinate execution order, and handle synchronization. This code is complex, error-prone, and difficult to verify. - -dagx eliminates this entirely: - -1. **Wire up tasks with primitives**: During DAG construction, tasks are connected using channels and organized into topological layers based on their dependencies -2. **Start everything simultaneously**: When you call `run()`, all tasks spawn at once—there's no scheduler deciding when each task should start -3. **Let the primitives handle coordination**: Channels naturally enforce execution order—tasks wait on their input channels until upstream tasks send data. No custom orchestration needed. -4. **Runtime joins on completion**: The runtime simply spawns all tasks and waits for the completion channel to close. That's it. - -### The Implementation - -Under the hood, dagx uses: - -- **Oneshot channels**: Each edge in the DAG gets a `futures::channel::oneshot` - producer sends once, consumer receives once -- **Ownership model**: Tasks take ownership (`self`) and are consumed during execution - no Mutex needed for task state -- **Direct data flow**: Outputs flow from producer to consumer via channels, never stored in shared memory during execution -- **Fast-path optimization**: Single-task layers execute inline without spawning overhead - -The core execution logic in `run()` is remarkably simple: - -```rust -// Create oneshot channel for each edge -for (producer, consumer) in edges { - let (tx, rx) = oneshot::channel(); - // Wire tx to producer, rx to consumer -} - -// Execute layer by layer -for layer in layers { - if layer.len() == 1 { - // Fast path: Execute inline, no spawning overhead - // IMPORTANT: Panic handling is required here! - let result = task.execute().catch_unwind().await; - // Convert panic to error to match spawned task behavior - } else { - // Parallel path: Spawn all tasks in layer - for task in layer { - spawner(async { - let input = await_on_input_channels(); // Blocks until deps complete - let output = task.run(input); // Consumes task - send_to_output_channels(output); // Sends to dependents - }); - } - // Wait for layer completion - } -} - -// Tasks coordinate themselves via channels - no scheduler needed -``` - -No state machines. No task queues. No wake-up logic. No Mutex locks. Just channels doing what channels do, with an inline fast-path for sequential execution. - -### Inline Execution Fast-Path - -**Performance optimization for sequential workloads**: When a layer contains only a single task (common in deep chains and linear pipelines), dagx executes it inline rather than spawning it. This eliminates spawning overhead, context switching, and channel creation, resulting in 10-100x performance improvements for sequential patterns. - -**Panic handling guarantee**: To maintain behavioral consistency between inline and spawned execution, panics in inline tasks are caught using `FutureExt::catch_unwind()` and converted to errors. This matches the behavior of all major async runtimes (Tokio, async-std, smol, embassy-rs), which catch panics in spawned tasks and convert them to `JoinError` or equivalent. - -**Why this matters**: - -- **Spawned tasks** (layer.len() > 1): Runtime catches panics automatically → becomes error -- **Inline tasks** (layer.len() == 1): We catch panics manually → becomes error -- **Result**: Identical behavior regardless of execution path - -This ensures your code behaves the same whether a task runs inline or spawned, making dagx's optimizations transparent and predictable. - -### Benefits - -**Simplicity**: The runtime is straightforward: create channels, spawn tasks, let them coordinate via awaiting. No complex scheduler code to maintain, debug, or optimize. - -**Reliability**: Built on battle-tested primitives (oneshot channels, async/await) from Rust's standard library and the futures crate. These have been used in production by thousands of projects and are orders of magnitude more reliable than custom scheduling logic. - -**Bug resistance**: Fewer moving parts means fewer places for bugs to hide. The type system enforces correct wiring at compile time. Channels handle synchronization. The ownership model prevents data races. What's left to break? - -**Performance**: Near zero-overhead. No Mutex locks during execution. Arc reference counting for efficient fanout (atomic operations, not locks). Oneshot channels are often lock-free. Tasks start as soon as their dependencies complete - maximum parallelism. - -**Auditability**: Want to verify correctness? Check the channel wiring, verify tasks await their inputs, done. No need to trace through complex state machine transitions or wake-up cascades. - -### The Insight - -The key insight is that **dependencies ARE the schedule**. If task B depends on task A's output, a channel naturally enforces that B waits for A. The dependency graph already encodes all the scheduling information—we just need to wire up channels to match it. - -This is dagx's core philosophy: leverage the type system for correctness, use primitives for coordination, and let the compiler optimize everything else away. - -### Measured Overhead - -How much overhead does this approach actually add? Benchmarks on an AMD 7840U (Zen 4 laptop CPU) show: - -**DAG Construction**: - -- Empty DAG creation: **~20 nanoseconds** -- Adding tasks: **~96 nanoseconds per task** -- Building a 10,000-task DAG: **~1.04 milliseconds** (104 ns/task) - -**Execution Overhead** (framework coordination, excluding actual task work): - -- Sequential workloads: **~0.78 microseconds per task** (inline execution fast-path) -- Parallel workloads: **~1.2 microseconds per task** -- 100-task deep chain: **~78 microseconds total** -- 100 independent tasks: **~122 microseconds total** -- 10,000 independent tasks: **~11.2 milliseconds total** - -**Scaling**: Sub-microsecond per-task overhead across all workload patterns. Linear scaling verified to 10k+ tasks. - -**Comparison**: dagx is **4-148x faster** than dagrs (v0.5) across all benchmark patterns (see detailed benchmarks section). - -The primitives-as-scheduler approach with inline fast-path optimization delivers exceptional performance: coordination overhead is sub-microsecond per task, and for real-world workloads where tasks do meaningful work (I/O, computation, etc.), framework overhead is negligible—typically well under 1% of total execution time. +That's it. No trait boilerplate, no manual channels, no node IDs. ## Quick Start @@ -383,300 +97,34 @@ async fn main() { } ``` -## Comparison with Similar Projects +## Performance -The Rust ecosystem offers several DAG execution libraries, each optimized for different use cases. This comparison helps you choose the right tool for your needs. +dagx provides true parallel execution with sub-microsecond overhead per task. -### Developer Experience: API Comparison +**Why is dagx so fast?** -The same task (compute 2 + 3) implemented across different libraries shows how API complexity varies: +- **Inline fast-path**: Sequential chains execute inline without spawning (8.9-129x faster) +- **Primitives as scheduler**: No custom scheduler — channels coordinate execution +- **Adaptive execution**: Inline for sequential work, true parallelism for concurrent work +- **Zero-cost abstractions**: Generics and monomorphization eliminate overhead -**dagx** (Simple: macro + builder): +See [design philosophy](docs/DESIGN_PHILOSOPHY.md) for details. -```rust -#[task] -impl Value { async fn run(&self) -> i32 { self.0 } } +## Features -#[task] -impl Add { async fn run(a: &i32, b: &i32) -> i32 { a + b } } - -let dag = DagRunner::new(); -let x = dag.add_task(Value(2)); -let y = dag.add_task(Value(3)); -let sum = dag.add_task(Add).depends_on((&x, &y)); -dag.run(|fut| tokio::spawn(fut)).await?; -``` - -**dagrs** (Complex: traits + channels + IDs + manual wiring): - -```rust -#[async_trait] -impl Action for Value { - async fn run(&self, _: &mut InChannels, out: &mut OutChannels, _: Arc) -> Output { - out.broadcast(Content::new(self.0)).await; - Output::Out(Some(Content::new(self.0))) - } -} - -let mut table = NodeTable::new(); -let node1 = DefaultNode::with_action("x".into(), Value(2), &mut table); -let id1 = node1.id(); -let node2 = DefaultNode::with_action("y".into(), Value(3), &mut table); -let id2 = node2.id(); -let node3 = DefaultNode::with_action("add".into(), Add, &mut table); -let id3 = node3.id(); -let mut graph = Graph::new(); -graph.add_node(node1); graph.add_node(node2); graph.add_node(node3); -graph.add_edge(id1, vec![id3]); graph.add_edge(id2, vec![id3]); -graph.set_env(EnvVar::new(table)); graph.start().unwrap(); -``` - -**async_dag** (Medium: slots + indices): - -```rust -let mut graph = Graph::new(); -let x = graph.add_task(|| async { 2 }); -let y = graph.add_task(|| async { 3 }); -let sum = graph.add_child_task(x, |a: i32| async move { a }, 0)?; -graph.update_dependency(y, sum, 1)?; // Must specify slot index -let sum = graph.add_child_task(sum, |a: i32, b: i32| async move { a + b }, 0)?; -graph.update_dependency(y, sum, 1)?; -``` - -**Key differences**: - -- **dagx**: Type-safe dependencies, automatic wiring, no manual ID tracking, minimal boilerplate -- **dagrs**: Manual channel management, node ID tracking, Content wrapping, Action trait boilerplate -- **async_dag**: Slot indices must be tracked manually, dependencies updated separately - -### Quick Comparison - -| Project | License | Runtime | Type Safety | API Complexity | Performance vs dagx | Key Features | -| ------------------------------------------------------------- | -------------- | ------------------- | --------------------- | -------------- | --------------------------------- | ------------------------------------------------------------------------------------ | -| **dagx** | MIT | Any async runtime | Compile-time | Simple | Baseline (see benchmarks) | Primitives-as-scheduler, inline fast-path, automatic Arc wrapping, up to 8 deps/task | -| [**dagrs**](https://github.com/dagrs-dev/dagrs) (v0.5) | MIT/Apache-2.0 | Creates own runtime | Runtime (async_trait) | Complex | 4-148x slower across all patterns | Flow-based Programming, cyclic graphs, loops, conditional nodes, YAML config | -| [**async_dag**](https://github.com/chubei-oppen/async_dag) | MIT | Any async runtime | Compile-time | Medium | No benchmarks | Slot-based dependencies, Graph/TryGraph modes, maximum parallelism | -| [**dag-flow**](https://github.com/makisevon/dag-flow) | MIT/Apache-2.0 | Any async runtime | Runtime (HashMap) | Complex | No benchmarks | Experimental, all tasks run simultaneously, weak dependencies | -| [**RenovZ/dag-runner**](https://github.com/RenovZ/dag-runner) | MIT | Tokio only | Unclear | Simple | No benchmarks | Edge-based API, cycle detection, stops on first error | -| [**tasksitter**](https://github.com/lionkor/tasksitter) | Unspecified | Unclear | Unclear | Medium | No benchmarks | Cyclic graphs, dynamic runtime modification, pause/resume | - -_GitHub stars (as of 2025): dagrs (449), async_dag (25), dag-flow (2), others (0-2)_ - -### Performance Benchmarks vs dagrs - -Direct comparison benchmarks (lower is better): - -| Workload Pattern | dagx | dagrs (v0.5) | dagx Performance | -| --------------------------------------------------- | -------- | ------------ | ------------------ | -| **Linear pipeline** (5 sequential tasks) | 3.0 µs | 446.1 µs | **148x faster** 🚀 | -| **ETL pipeline** (realistic extract-transform-load) | 26.2 µs | 460.4 µs | **17.6x faster** ✓ | -| **Deep chain** (100 purely sequential tasks) | 78.2 µs | 801.5 µs | **10.2x faster** ✓ | -| **Wide fanout** (1→100 broadcast) | 161.0 µs | 665.5 µs | **4.1x faster** ✓ | -| **Large scale** (10,000 independent tasks) | 11.2 ms | 13.6 ms | **1.2x faster** ✓ | - -**Summary**: dagx is **4-148x faster** than dagrs across all benchmark patterns. The inline fast-path optimization eliminates spawning overhead for sequential workloads while maintaining excellent parallel performance. - -**How does dagx achieve this?** The inline fast-path detects single-task layers (common in sequential chains) and executes them directly without spawning overhead. For parallel workloads, tasks still spawn to maximize concurrency. This adaptive execution strategy combines the best of both worlds: sub-microsecond overhead for sequential work, true parallelism for concurrent work. - -**Key insight**: Most real-world DAGs mix sequential and parallel patterns. dagx automatically optimizes for both, delivering 4-148x better performance than dagrs regardless of workload shape. - -_Benchmarks run on AMD Ryzen 7 7840U (Zen 4) @ 3.3GHz. Run `cargo bench` to test on your hardware._ - -### Detailed Comparison - -#### dagrs (Most Mature) - -**Best for**: Complex workflows requiring advanced flow control, Tokio-based applications, machine learning pipelines. - -**Strengths**: - -- Most mature (449 GitHub stars, active community) -- Rich feature set: Flow-based Programming, cyclic graphs, loops, conditional nodes -- YAML configuration support for declarative workflows -- Designed for complex orchestration patterns - -**Trade-offs**: - -- Creates own Tokio runtime internally (not runtime-agnostic, cannot be nested) -- More complex API: `Action` trait, `InChannels`/`OutChannels`, `NodeTable`, `Content` wrappers, manual node ID tracking -- Uses `async_trait` for type erasure (runtime overhead) -- **4-148x slower than dagx** across all benchmark patterns (see comparison benchmarks) - -**API Style** (v0.5): - -```rust -// Define action with channels for communication -#[async_trait] -impl Action for MyTask { - async fn run(&self, input: &mut InChannels, - output: &mut OutChannels, env: Arc) -> Output { - let inputs: Vec = input.map(|c| ...).await; // Manually extract - output.broadcast(Content::new(result)).await; // Manually wrap - Output::Out(Some(Content::new(result))) - } -} - -let mut table = NodeTable::new(); -let node = DefaultNode::with_action("name".into(), MyTask, &mut table); -let node_id = node.id(); // Must capture ID before moving -graph.add_node(node); -graph.add_edge(source_id, vec![target_id]); -graph.set_env(EnvVar::new(table)); graph.start().unwrap(); -``` - -#### async_dag (Clean Type Safety) - -**Best for**: Runtime flexibility with compile-time type safety, fail-fast workflows. - -**Strengths**: - -- Runtime-agnostic (works with any async runtime) -- Compile-time type checking on task connections -- Both standard (`Graph`) and fail-fast (`TryGraph`) modes -- Designed for maximum parallelism - -**Trade-offs**: - -- Medium API complexity: slot-based dependency management -- Must manually specify slot indices (0, 1, etc.) when connecting tasks -- Less mature (25 stars) -- No performance benchmarks - -**API Style**: - -```rust -let mut graph = Graph::new(); -let _1 = graph.add_task(|| async { 1 }); -let _2 = graph.add_task(|| async { 2 }); - -// add_child_task with slot index -let _3 = graph.add_child_task(_1, sum, 0).unwrap(); -graph.update_dependency(_2, _3, 1).unwrap(); // Specify slot 1 - -graph.run().await; -``` - -#### dag-flow (Experimental) - -**Best for**: Experimental projects, flexible dependency awaiting patterns. - -**Strengths**: - -- Runtime-agnostic -- All tasks run simultaneously (not in dependency layers) -- Weak dependencies support -- Flexible input awaiting at any point in task execution - -**Trade-offs**: - -- Explicitly experimental -- Runtime type safety via `HashMap` -- Complex API: implement `Task` trait with `id()`, `dependencies()`, `run()` -- Named dependencies (string-based lookup) -- Very early stage (2 stars) - -**API Style**: - -```rust -impl Task for MyTask { - fn id(&self) -> String { "task_name".into() } - fn dependencies(&self) -> Option> { Some(vec!["dep1".into()]) } - - async fn run(&self, inputs: HashMap>) -> Option { - let dep_value = inputs.get("dep1").unwrap().await; - // Process - } -} -``` - -#### RenovZ/dag-runner (Simple Edge-Based) - -**Best for**: Simple DAGs with Tokio, straightforward edge-based dependencies. - -**Strengths**: - -- Simple API: `add_vertex()`, `add_edge()` -- Cycle detection -- Stops on first error - -**Trade-offs**: - -- Requires Tokio runtime -- Manual channel setup for task communication -- Type safety model unclear -- Very early stage (0 stars) - -**API Style**: - -```rust -let mut dag = Dag::default(); -dag.add_vertex("one", || async move { /* task */ }); -dag.add_edge("one", "two"); -dag.run().await?; -``` - -#### tasksitter (Dynamic Workflows) - -**Best for**: Dynamic workflow modification, cyclic graphs, runtime introspection. - -**Strengths**: - -- Supports cyclic graphs (not just DAGs) -- Dynamic graph modification at runtime -- Pause/resume capabilities -- Graph introspection - -**Trade-offs**: - -- Limited documentation -- Runtime and type safety model unclear -- Very early stage (0 stars) - -### When to Choose dagx - -Choose dagx when you value: - -- **Performance**: 4-148x faster than dagrs across all workload patterns (see benchmarks) -- **Runtime flexibility**: Works with Tokio, async-std, smol, or any async runtime -- **Compile-time safety**: Full type safety with no runtime type errors in the public API -- **Minimal overhead**: ~0.78µs per task (sequential), ~1.2µs per task (parallel) -- **Simple, ergonomic API**: `#[task]` macro, `add_task()`, `depends_on()` - that's it -- **Automatic optimizations**: Arc wrapping, inline execution, adaptive spawning - all transparent -- **Predictable performance**: Linear scaling, no hidden complexity, consistent sub-µs overhead - -dagx is **not** the right choice if you need: - -- Cyclic graphs or dynamic flow control (loops, conditions) → Consider **dagrs** or **tasksitter** -- More than 8 dependencies per task → Consider **dagrs** or **async_dag** -- YAML-based configuration → Consider **dagrs** - -### When to Consider Alternatives - -- **Choose dagrs** if you need advanced flow control (loops, conditionals, cyclic graphs), YAML configuration, or are already committed to Tokio and want a mature, feature-rich solution -- **Choose async_dag** if you want compile-time type safety with runtime flexibility and the slot-based API appeals to you -- **Choose dag-flow** if you're building experimental projects and the all-tasks-run-simultaneously model fits your use case -- **Choose RenovZ/dag-runner** if you need the simplest possible edge-based API and are already using Tokio -- **Choose tasksitter** if you need dynamic graph modification at runtime or cyclic workflow support +- **Compile-time cycle prevention**: Type system makes cycles impossible — no runtime checks +- **Compile-time type safety**: Dependencies validated at compile time, no runtime type errors +- **Runtime-agnostic**: Works with Tokio, async-std, smol, or any async runtime +- **True parallelism**: Tasks spawn to multiple threads for genuine parallel execution +- **Type-state pattern**: API prevents incorrect wiring through compile-time errors +- **Zero-cost abstractions**: Leverages generics and monomorphization for minimal overhead +- **Flexible task patterns**: Supports stateless, read-only, and mutable state tasks +- **Simple API**: Just `#[task]`, `DagRunner`, `TaskHandle`, and `TaskBuilder` +- **Comprehensive error handling**: Result-based errors with actionable messages +- **Optional tracing**: Zero-cost observability via optional `tracing` feature flag ## Core Concepts -### Task - -A `Task` is a unit of async work with typed inputs and outputs. Use the `#[task]` macro to define tasks: - -```rust -use dagx::{task, Task}; - -struct Scale(i32); - -#[task] -impl Scale { - async fn run(&self, input: &i32) -> i32 { - input * self.0 - } -} -``` - ### Task Patterns dagx supports three task patterns: @@ -738,182 +186,7 @@ let task = dag.add_task(my_task).depends_on(&upstream); let task = dag.add_task(my_task).depends_on((&upstream1, &upstream2)); ``` -## Examples - -### Fan-out Pattern (1 → n) - -One task produces a value consumed by multiple downstream tasks: - -```rust -use dagx::{task, DagRunner, Task}; - -struct Value(i32); - -#[task] -impl Value { - async fn run(&self) -> i32 { self.0 } -} - -struct Increment(i32); - -#[task] -impl Increment { - async fn run(&self, input: &i32) -> i32 { input + self.0 } -} - -struct Scale(i32); - -#[task] -impl Scale { - async fn run(&self, input: &i32) -> i32 { input * self.0 } -} - -#[tokio::main] -async fn main() { - let dag = DagRunner::new(); - - let base = dag.add_task(Value(10)); - let plus1 = dag.add_task(Increment(1)).depends_on(&base); - let times2 = dag.add_task(Scale(2)).depends_on(&base); - - dag.run(|fut| { tokio::spawn(fut); }).await.unwrap(); - - assert_eq!(dag.get(plus1).unwrap(), 11); - assert_eq!(dag.get(times2).unwrap(), 20); -} -``` - -### Fan-in Pattern (m → 1) - -Multiple tasks produce values consumed by a single downstream task: - -```rust -use dagx::{task, DagRunner, Task}; - -struct Name(String); - -#[task] -impl Name { - async fn run(&self) -> String { self.0.clone() } -} - -struct Age(i32); - -#[task] -impl Age { - async fn run(&self) -> i32 { self.0 } -} - -struct Active(bool); - -#[task] -impl Active { - async fn run(&self) -> bool { self.0 } -} - -struct FormatUser; - -#[task] -impl FormatUser { - async fn run(n: &String, a: &i32, f: &bool) -> String { - format!("User: {n}, Age: {a}, Active: {f}") - } -} - -#[tokio::main] -async fn main() { - let dag = DagRunner::new(); - - let name = dag.add_task(Name("Alice".to_string())); - let age = dag.add_task(Age(30)); - let active = dag.add_task(Active(true)); - let result = dag.add_task(FormatUser).depends_on((&name, &age, &active)); - - dag.run(|fut| { tokio::spawn(fut); }).await.unwrap(); - - assert_eq!(dag.get(result).unwrap(), "User: Alice, Age: 30, Active: true"); -} -``` - -### Many-to-Many Pattern (m ↔ n) - -Complex DAGs with multiple layers: - -```rust -use dagx::{task, DagRunner, Task}; - -struct Value(i32); - -#[task] -impl Value { - async fn run(&self) -> i32 { self.0 } -} - -struct Add; - -#[task] -impl Add { - async fn run(a: &i32, b: &i32) -> i32 { a + b } -} - -struct Multiply; - -#[task] -impl Multiply { - async fn run(a: &i32, b: &i32) -> i32 { a * b } -} - -#[tokio::main] -async fn main() { - let dag = DagRunner::new(); - - // Layer 1: Sources - let x = dag.add_task(Value(2)); - let y = dag.add_task(Value(3)); - let z = dag.add_task(Value(5)); - - // Layer 2: Intermediate computations - let sum_xy = dag.add_task(Add).depends_on((&x, &y)); // 2 + 3 = 5 - let prod_yz = dag.add_task(Multiply).depends_on((&y, &z)); // 3 * 5 = 15 - - // Layer 3: Final result - let total = dag.add_task(Add).depends_on((&sum_xy, &prod_yz)); // 5 + 15 = 20 - - dag.run(|fut| { tokio::spawn(fut); }).await.unwrap(); - - assert_eq!(dag.get(total).unwrap(), 20); -} -``` - -## Type-State Pattern - -The API uses the type-state pattern to ensure correctness at compile time: - -```rust -use dagx::{task, Task}; - -struct Increment(i32); - -#[task] -impl Increment { - async fn run(&self, input: &i32) -> i32 { input + self.0 } -} - -struct Uppercase; - -#[task] -impl Uppercase { - async fn run(input: &String) -> String { input.to_uppercase() } -} - -// ✅ This compiles -let task = dag.add_task(Increment(5)).depends_on(&upstream); // upstream produces i32 - -// ❌ This doesn't compile - type mismatch! -let task = dag.add_task(Uppercase).depends_on(&upstream); // Error: expected String, got i32 -``` - -## Runtime Agnostic +### Runtime Agnostic dagx works with any async runtime. Provide a spawner function to `run()`: @@ -928,11 +201,58 @@ dag.run(|fut| { async_std::task::spawn(fut); }).await.unwrap(); dag.run(|fut| { smol::spawn(fut).detach(); }).await.unwrap(); ``` +## Tutorials & Examples + +### Tutorials (Start Here) + +Step-by-step introduction to dagx: + +- [`01_basic.rs`](examples/01_basic.rs) - Your first DAG +- [`02_fan_out.rs`](examples/02_fan_out.rs) - One task feeds many (1→N) +- [`03_fan_in.rs`](examples/03_fan_in.rs) - Many tasks feed one (N→1) +- [`04_parallel_computation.rs`](examples/04_parallel_computation.rs) - Map-reduce with true parallelism + +Run tutorial examples: + +```bash +cargo run --example 01_basic +cargo run --example 02_fan_out +cargo run --example 03_fan_in +cargo run --example 04_parallel_computation +``` + +### Advanced Examples + +Real-world patterns: + +- [`circuit_breaker.rs`](examples/circuit_breaker.rs) - Circuit breaker pattern for resilient systems +- [`complex_dag.rs`](examples/complex_dag.rs) - Multi-layer workflows with complex dependencies +- [`conditional_workflow.rs`](examples/conditional_workflow.rs) - Conditional execution with Result types +- [`data_pipeline.rs`](examples/data_pipeline.rs) - ETL data processing pipeline +- [`debug_tracing.rs`](examples/debug_tracing.rs) - Debug metadata and task naming +- [`error_handling.rs`](examples/error_handling.rs) - Error propagation and recovery +- [`large_dag.rs`](examples/large_dag.rs) - Performance at 10,000+ nodes +- [`parallelism_proof.rs`](examples/parallelism_proof.rs) - Proof of true parallel execution +- [`retry_strategies.rs`](examples/retry_strategies.rs) - Retry patterns for transient failures +- [`timeout.rs`](examples/timeout.rs) - Managing task timeouts +- [`tracing_example.rs`](examples/tracing_example.rs) - Observability with tracing support + +Run any example: `cargo run --example data_pipeline` + +## Advanced Topics + +Detailed documentation on dagx internals and advanced features: + +- [**Compile-Time Cycle Prevention**](docs/CYCLE_PREVENTION.md) - How the type system prevents cycles +- [**Design Philosophy**](docs/DESIGN_PHILOSOPHY.md) - Primitives as scheduler, inline fast-path optimization +- [**Tracing Support**](docs/TRACING.md) - Zero-cost observability with the `tracing` crate +- [**Library Comparisons**](docs/COMPARISONS.md) - Detailed comparison with dagrs, async_dag, and others + ## Important Limitations ### Tasks Cannot Return Bare Tuples -**Tasks cannot return bare tuples as output types.** This is a technical limitation of the current implementation. If you need to return multiple values from a task, use one of these workarounds: +Tasks cannot return bare tuples as output types. If you need to return multiple values, use one of these workarounds: **Option 1: Use a struct (recommended)** @@ -970,48 +290,6 @@ impl FetchData { Structs are preferred because they're self-documenting and easier to refactor. -## Examples - -The `examples/` directory contains both **tutorial examples** (numbered, beginner-friendly) and **reference examples** (practical patterns). - -### Tutorial Examples - -Start here if you're new to dagx. These examples build progressively: - -- **`01_basic.rs`** - Getting started: your first DAG -- **`02_fan_out.rs`** - Fan-out pattern: one task feeds many (1 → N) -- **`03_fan_in.rs`** - Fan-in pattern: many tasks feed one (N → 1) -- **`04_parallel_computation.rs`** - Map-reduce pattern with true parallelism - -Run tutorial examples: - -```bash -cargo run --example 01_basic -cargo run --example 02_fan_out -cargo run --example 03_fan_in -cargo run --example 04_parallel_computation -``` - -### Reference Examples - -Practical patterns for real-world use cases: - -- **`complex_dag.rs`** - Multi-layer workflows with complex dependencies -- **`conditional_workflow.rs`** - Conditional execution with Result types -- **`data_pipeline.rs`** - Real-world ETL data processing pipeline -- **`error_handling.rs`** - Error propagation and recovery strategies -- **`timeout.rs`** - Managing task timeouts -- **`large_dag.rs`** - Performance and scalability (10,000+ nodes) -- **`parallelism_proof.rs`** - Proof of true parallel execution (10,000 tasks × 1s = 1s total) - -Run reference examples: - -```bash -cargo run --example data_pipeline -cargo run --example error_handling -cargo run --example parallelism_proof -``` - ## When to Use dagx dagx is ideal for: @@ -1022,88 +300,15 @@ dagx is ideal for: - **Workflow engines** with typed data flow between stages - **ETL processes** with validation and transformation steps -## Performance +## Benchmarks -dagx provides true parallel execution with sub-microsecond overhead: - -- **Adaptive execution**: Inline fast-path for sequential work, true parallelism for concurrent work -- **Tasks spawn to multiple threads** via your async runtime's spawner when beneficial -- **Linear scaling** verified up to 10,000+ tasks -- **~0.78µs overhead per task** for sequential workloads (inline execution) -- **~1.2µs overhead per task** for parallel workloads -- **Efficient memory usage** at ~200 bytes per task -- **Zero-cost abstractions** through generics and compile-time monomorphization -- **4-148x faster than dagrs** across all benchmark patterns (see comparison benchmarks above) - -### Performance Characteristics - -- ✅ **Exceptional for sequential chains**: 10-148x faster than dagrs via inline execution -- ✅ **Excellent for parallel workloads**: 4-10x faster than dagrs on fan-out, ETL, large-scale -- ✅ **Best-in-class for mixed parallelism**: Automatically optimizes execution strategy per layer -- ✅ **Sub-microsecond per-task overhead**: Fast enough that framework cost is negligible - -**Result**: dagx dominates dagrs on every benchmark pattern. No trade-offs, no compromises. - -### Automatic Arc Wrapping (No Manual Arc Needed!) - -**Task outputs are automatically wrapped in `Arc` internally** for efficient fan-out patterns. You just output `T` - the framework handles the Arc wrapping: - -```rust -use dagx::{task, DagRunner, Task}; - -// ✅ CORRECT: Just output Vec, framework handles Arc internally -struct FetchData; -#[task] -impl FetchData { - async fn run() -> Vec { - vec!["data".to_string(); 10_000] - } -} - -// Downstream tasks receive &Vec -// Internally, Arc> is cloned (cheap), then inner Vec is extracted -struct ProcessData; -#[task] -impl ProcessData { - async fn run(data: &Vec) -> usize { - data.len() - } -} -``` - -**How it works:** - -- Your task outputs `T` -- Framework wraps it in `Arc` internally -- For fan-out (1→N), Arc is cloned N times (just pointer copies - O(1)) -- Each downstream task receives `&T` after extracting from Arc - -**Performance characteristics:** - -- **Heap types** (Vec, String, HashMap): Arc overhead is negligible, fanout is essentially free -- **Copy types** (i32, usize): Small Arc overhead (~few ns) due to atomic refcounting -- See `cargo bench` for actual measurements - -**Advanced - Zero-copy optimization:** -If you want true zero-copy sharing (no extraction), output `Arc` explicitly: - -```rust -// Your task outputs Arc -async fn run() -> Arc> { Arc::new(vec![]) } - -// Downstream receives &Arc - just clones the Arc pointer -async fn process(data: &Arc>) -> usize { data.len() } -``` - -This becomes `Arc>` internally, but ExtractInput unwraps one layer automatically. - -Run benchmarks: +Run the full benchmark suite: ```bash cargo bench ``` -View the detailed HTML reports: +View detailed HTML reports: ```bash # macOS @@ -1114,22 +319,26 @@ xdg-open target/criterion/report/index.html # Windows start target/criterion/report/index.html - -# Or manually open target/criterion/report/index.html in your browser ``` +_Benchmarks run on AMD Ryzen 7 7840U (Zen 4) @ 3.3GHz._ + +## Code of Conduct + +This project follows the [Builder's Code of Conduct](https://builderscode.org). + ## Documentation Full API documentation is available at [docs.rs/dagx](https://docs.rs/dagx). -## License - -Licensed under the MIT License. See [LICENSE](LICENSE) for details. - -Copyright (c) 2025 Stephen Waits - ## Contributing Contributions are welcome! Please see [CONTRIBUTING.md](CONTRIBUTING.md) for guidelines. For security issues, see [SECURITY.md](SECURITY.md). + +## License + +Licensed under the MIT License. See [LICENSE](LICENSE) for details. + +Copyright (c) 2025 Stephen Waits diff --git a/docs/COMPARISONS.md b/docs/COMPARISONS.md new file mode 100644 index 0000000..ad30468 --- /dev/null +++ b/docs/COMPARISONS.md @@ -0,0 +1,274 @@ +# Comparison with Similar Projects + +The Rust ecosystem offers several DAG execution libraries, each optimized for different use cases. This comparison helps you choose the right tool for your needs. + +## Developer Experience: API Comparison + +The same task (compute 2 + 3) implemented across different libraries shows how API complexity varies: + +**dagx** (Simple: macro + builder): + +```rust +#[task] +impl Value { async fn run(&self) -> i32 { self.0 } } + +#[task] +impl Add { async fn run(a: &i32, b: &i32) -> i32 { a + b } } + +let dag = DagRunner::new(); +let x = dag.add_task(Value(2)); +let y = dag.add_task(Value(3)); +let sum = dag.add_task(Add).depends_on((&x, &y)); +dag.run(|fut| tokio::spawn(fut)).await?; +``` + +**dagrs** (Complex: traits + channels + IDs + manual wiring): + +```rust +#[async_trait] +impl Action for Value { + async fn run(&self, _: &mut InChannels, out: &mut OutChannels, _: Arc) -> Output { + out.broadcast(Content::new(self.0)).await; + Output::Out(Some(Content::new(self.0))) + } +} + +let mut table = NodeTable::new(); +let node1 = DefaultNode::with_action("x".into(), Value(2), &mut table); +let id1 = node1.id(); +let node2 = DefaultNode::with_action("y".into(), Value(3), &mut table); +let id2 = node2.id(); +let node3 = DefaultNode::with_action("add".into(), Add, &mut table); +let id3 = node3.id(); +let mut graph = Graph::new(); +graph.add_node(node1); graph.add_node(node2); graph.add_node(node3); +graph.add_edge(id1, vec![id3]); graph.add_edge(id2, vec![id3]); +graph.set_env(EnvVar::new(table)); graph.start().unwrap(); +``` + +**async_dag** (Medium: slots + indices): + +```rust +let mut graph = Graph::new(); +let x = graph.add_task(|| async { 2 }); +let y = graph.add_task(|| async { 3 }); +let sum = graph.add_child_task(x, |a: i32| async move { a }, 0)?; +graph.update_dependency(y, sum, 1)?; // Must specify slot index +let sum = graph.add_child_task(sum, |a: i32, b: i32| async move { a + b }, 0)?; +graph.update_dependency(y, sum, 1)?; +``` + +**Key differences**: + +- **dagx**: Type-safe dependencies, automatic wiring, no manual ID tracking, minimal boilerplate +- **dagrs**: Manual channel management, node ID tracking, Content wrapping, Action trait boilerplate +- **async_dag**: Slot indices must be tracked manually, dependencies updated separately + +## Quick Comparison + +| Project | License | Runtime | Type Safety | API Complexity | Performance vs dagx | Key Features | +| ------------------------------------------------------------- | -------------- | ------------------- | --------------------- | -------------- | ---------------------------- | ------------------------------------------------------------------------------------ | +| **dagx** | MIT | Any async runtime | Compile-time | Simple | Baseline (see benchmarks) | Primitives-as-scheduler, inline fast-path, automatic Arc wrapping, up to 8 deps/task | +| [**dagrs**](https://github.com/dagrs-dev/dagrs) (v0.5) | MIT/Apache-2.0 | Creates own runtime | Runtime (async_trait) | Complex | 1.04-129x slower (see below) | Flow-based Programming, cyclic graphs, loops, conditional nodes, YAML config | +| [**async_dag**](https://github.com/chubei-oppen/async_dag) | MIT | Any async runtime | Compile-time | Medium | No benchmarks | Slot-based dependencies, Graph/TryGraph modes, maximum parallelism | +| [**dag-flow**](https://github.com/makisevon/dag-flow) | MIT/Apache-2.0 | Any async runtime | Runtime (HashMap) | Complex | No benchmarks | Experimental, all tasks run simultaneously, weak dependencies | +| [**RenovZ/dag-runner**](https://github.com/RenovZ/dag-runner) | MIT | Tokio only | Unclear | Simple | No benchmarks | Edge-based API, cycle detection, stops on first error | +| [**tasksitter**](https://github.com/lionkor/tasksitter) | Unspecified | Unclear | Unclear | Medium | No benchmarks | Cyclic graphs, dynamic runtime modification, pause/resume | + +_GitHub stars (as of 2025): dagrs (449), async_dag (25), dag-flow (2), others (0-2)_ + +## Performance Benchmarks vs dagrs + +Direct comparison benchmarks (lower is better): + +| Workload | Tasks | dagx | dagrs | Speedup | +| ----------------- | ------ | ------- | ------- | ------------------ | +| Sequential chain | 5 | 3.0 µs | 385 µs | **129x faster** 🚀 | +| Sequential chain | 100 | 79 µs | 703 µs | **8.9x faster** | +| Diamond pattern | 4 | 11 µs | 387 µs | **34x faster** | +| Fan-out (1→100) | 101 | 155 µs | 595 µs | **3.85x faster** | +| Independent tasks | 10,000 | 12.7 ms | 13.3 ms | **1.04x faster** | + +**Summary**: dagx is **1.04-129x faster** than dagrs across all benchmark patterns. The inline fast-path optimization eliminates spawning overhead for sequential workloads while maintaining excellent parallel performance. + +**How does dagx achieve this?** The inline fast-path detects single-task layers (common in sequential chains) and executes them directly without spawning overhead. For parallel workloads, tasks still spawn to maximize concurrency. This adaptive execution strategy combines the best of both worlds: sub-microsecond overhead for sequential work, true parallelism for concurrent work. + +**Key insight**: Most real-world DAGs mix sequential and parallel patterns. dagx automatically optimizes for both, delivering 1.04-129x better performance than dagrs regardless of workload shape. + +_Benchmarks run on AMD Ryzen 7 7840U (Zen 4) @ 3.3GHz. Run `cargo bench` to test on your hardware._ + +## Detailed Comparison + +### dagrs (Most Mature) + +**Best for**: Complex workflows requiring advanced flow control, Tokio-based applications, machine learning pipelines. + +**Strengths**: + +- Most mature (449 GitHub stars, active community) +- Rich feature set: Flow-based Programming, cyclic graphs, loops, conditional nodes +- YAML configuration support for declarative workflows +- Designed for complex orchestration patterns + +**Trade-offs**: + +- Creates own Tokio runtime internally (not runtime-agnostic, cannot be nested) +- More complex API: `Action` trait, `InChannels`/`OutChannels`, `NodeTable`, `Content` wrappers, manual node ID tracking +- Uses `async_trait` for type erasure (runtime overhead) +- **1.04-129x slower than dagx** across all benchmark patterns (see comparison benchmarks) + +**API Style** (v0.5): + +```rust +// Define action with channels for communication +#[async_trait] +impl Action for MyTask { + async fn run(&self, input: &mut InChannels, + output: &mut OutChannels, env: Arc) -> Output { + let inputs: Vec = input.map(|c| ...).await; // Manually extract + output.broadcast(Content::new(result)).await; // Manually wrap + Output::Out(Some(Content::new(result))) + } +} + +let mut table = NodeTable::new(); +let node = DefaultNode::with_action("name".into(), MyTask, &mut table); +let node_id = node.id(); // Must capture ID before moving +graph.add_node(node); +graph.add_edge(source_id, vec![target_id]); +graph.set_env(EnvVar::new(table)); graph.start().unwrap(); +``` + +### async_dag (Clean Type Safety) + +**Best for**: Runtime flexibility with compile-time type safety, fail-fast workflows. + +**Strengths**: + +- Runtime-agnostic (works with any async runtime) +- Compile-time type checking on task connections +- Both standard (`Graph`) and fail-fast (`TryGraph`) modes +- Designed for maximum parallelism + +**Trade-offs**: + +- Medium API complexity: slot-based dependency management +- Must manually specify slot indices (0, 1, etc.) when connecting tasks +- Less mature (25 stars) +- No performance benchmarks + +**API Style**: + +```rust +let mut graph = Graph::new(); +let _1 = graph.add_task(|| async { 1 }); +let _2 = graph.add_task(|| async { 2 }); + +// add_child_task with slot index +let _3 = graph.add_child_task(_1, sum, 0).unwrap(); +graph.update_dependency(_2, _3, 1).unwrap(); // Specify slot 1 + +graph.run().await; +``` + +### dag-flow (Experimental) + +**Best for**: Experimental projects, flexible dependency awaiting patterns. + +**Strengths**: + +- Runtime-agnostic +- All tasks run simultaneously (not in dependency layers) +- Weak dependencies support +- Flexible input awaiting at any point in task execution + +**Trade-offs**: + +- Explicitly experimental +- Runtime type safety via `HashMap` +- Complex API: implement `Task` trait with `id()`, `dependencies()`, `run()` +- Named dependencies (string-based lookup) +- Very early stage (2 stars) + +**API Style**: + +```rust +impl Task for MyTask { + fn id(&self) -> String { "task_name".into() } + fn dependencies(&self) -> Option> { Some(vec!["dep1".into()]) } + + async fn run(&self, inputs: HashMap>) -> Option { + let dep_value = inputs.get("dep1").unwrap().await; + // Process + } +} +``` + +### RenovZ/dag-runner (Simple Edge-Based) + +**Best for**: Simple DAGs with Tokio, straightforward edge-based dependencies. + +**Strengths**: + +- Simple API: `add_vertex()`, `add_edge()` +- Cycle detection +- Stops on first error + +**Trade-offs**: + +- Requires Tokio runtime +- Manual channel setup for task communication +- Type safety model unclear +- Very early stage (0 stars) + +**API Style**: + +```rust +let mut dag = Dag::default(); +dag.add_vertex("one", || async move { /* task */ }); +dag.add_edge("one", "two"); +dag.run().await?; +``` + +### tasksitter (Dynamic Workflows) + +**Best for**: Dynamic workflow modification, cyclic graphs, runtime introspection. + +**Strengths**: + +- Supports cyclic graphs (not just DAGs) +- Dynamic graph modification at runtime +- Pause/resume capabilities +- Graph introspection + +**Trade-offs**: + +- Limited documentation +- Runtime and type safety model unclear +- Very early stage (0 stars) + +## When to Choose dagx + +Choose dagx when you value: + +- **Performance**: 1.04-129x faster than dagrs across all workload patterns (see benchmarks) +- **Runtime flexibility**: Works with Tokio, async-std, smol, or any async runtime +- **Compile-time safety**: Full type safety with no runtime type errors in the public API +- **Minimal overhead**: ~790 ns per task (sequential), ~1.3 µs per task (parallel) +- **Simple, ergonomic API**: `#[task]` macro, `add_task()`, `depends_on()` - that's it +- **Automatic optimizations**: Arc wrapping, inline execution, adaptive spawning - all transparent +- **Predictable performance**: Linear scaling, no hidden complexity, consistent sub-µs overhead + +dagx is **not** the right choice if you need: + +- Cyclic graphs or dynamic flow control (loops, conditions) → Consider **dagrs** or **tasksitter** +- More than 8 dependencies per task → Consider **dagrs** or **async_dag** +- YAML-based configuration → Consider **dagrs** + +## When to Consider Alternatives + +- **Choose dagrs** if you need advanced flow control (loops, conditionals, cyclic graphs), YAML configuration, or are already committed to Tokio and want a mature, feature-rich solution +- **Choose async_dag** if you want compile-time type safety with runtime flexibility and the slot-based API appeals to you +- **Choose dag-flow** if you're building experimental projects and the all-tasks-run-simultaneously model fits your use case +- **Choose RenovZ/dag-runner** if you need the simplest possible edge-based API and are already using Tokio +- **Choose tasksitter** if you need dynamic graph modification at runtime or cyclic workflow support diff --git a/docs/CYCLE_PREVENTION.md b/docs/CYCLE_PREVENTION.md new file mode 100644 index 0000000..fbba17e --- /dev/null +++ b/docs/CYCLE_PREVENTION.md @@ -0,0 +1,82 @@ +# Compile-Time Cycle Prevention + +**dagx guarantees acyclic graphs at compile time with zero runtime overhead.** Unlike other DAG libraries that detect cycles at runtime, dagx uses Rust's type system to make cycles impossible to express in the first place. + +## Why This Matters + +Most DAG libraries check for cycles when you execute the graph: + +```rust +// Other libraries: cycle detected at RUNTIME +let dag = DagRunner::new(); +let a = dag.add_task(TaskA); +let b = dag.add_task(TaskB); +a.depends_on(b); +b.depends_on(a); // Compiles fine... +dag.run().await?; // ❌ Error: "Cycle detected!" (at runtime) +``` + +**dagx prevents this at compile time:** + +```rust +// dagx: cycle prevented at COMPILE TIME +let dag = DagRunner::new(); +let a_builder = dag.add_task(TaskA); +let b_builder = dag.add_task(TaskB); + +// To make A depend on B, we need B as a TaskHandle. +// But calling depends_on() consumes the builder! +let b_handle = b_builder.depends_on(&some_source); + +// Now we can't make B depend on A because b_builder was moved! +// This won't compile: +// a_builder.depends_on(&b_handle); // ✓ OK: A→B +// b_handle.depends_on(&a_handle); // ❌ ERROR: TaskHandle has no depends_on() method! +``` + +## How It Works: The Type-State Pattern + +dagx uses two types to enforce acyclic structure: + +1. **`TaskBuilder`**: Mutable builder that can have dependencies added via `depends_on()` + - Consumed (moved) when you call `depends_on()` or convert to `TaskHandle` + - Can only be used once for wiring + +2. **`TaskHandle`**: Immutable reference to a finalized task + - Copy type (just wraps a task ID) + - Can be used as a dependency for other tasks + - **Has no `depends_on()` method** - can't be modified after creation + +This creates an impossible ordering constraint for cycles: + +```rust +// Attempting a cycle A→B→A: +let a_builder = dag.add_task(TaskA); // TaskBuilder +let b_builder = dag.add_task(TaskB); // TaskBuilder + +// To wire A→B, we need B as a TaskHandle +let b_handle = b_builder.into(); // Convert B to TaskHandle (consumes builder) + +// To wire B→A, we need A as a TaskHandle +let a_handle = a_builder.depends_on(&b_handle); // A now depends on B + +// Now if we try to complete the cycle: +// a_handle.depends_on(&...); // ❌ ERROR: no method named `depends_on` found +``` + +**The type system enforces strict topological ordering!** You can't create a `TaskHandle` until all its dependencies exist as handles, which prevents cycles. + +## Proof & Examples + +See [`src/cycle_prevention.rs`](../src/cycle_prevention.rs) for comprehensive documentation with `compile_fail` tests proving various cycle attempts don't compile. + +See [`tests/cycle_prevention.rs`](../tests/cycle_prevention.rs) for runtime tests demonstrating the type-state pattern in action. + +## Benefits + +- ✅ **Zero runtime cost**: No cycle detection code to execute +- ✅ **Catches errors early**: Cycles are caught at compile time, not in production +- ✅ **Compiler-verified**: The type system proves your graph is acyclic +- ✅ **No mental overhead**: You can't accidentally create cycles even if you try + +This is a true **zero-cost abstraction**—the type system provides safety guarantees without any runtime checks or performance penalty. diff --git a/docs/DESIGN_PHILOSOPHY.md b/docs/DESIGN_PHILOSOPHY.md new file mode 100644 index 0000000..59cf447 --- /dev/null +++ b/docs/DESIGN_PHILOSOPHY.md @@ -0,0 +1,113 @@ +# Design Philosophy: Primitives as Scheduler + +dagx takes an unconventional approach to task orchestration: **there is no traditional scheduler**. Instead of complex scheduling logic managing when tasks run, the entire system is built on communication and synchronization primitives that do the scheduling themselves. + +## How It Works + +Traditional DAG executors contain substantial scheduling code—algorithms that track task states, manage dependencies, coordinate execution order, and handle synchronization. This code is complex, error-prone, and difficult to verify. + +dagx eliminates this entirely: + +1. **Wire up tasks with primitives**: During DAG construction, tasks are connected using channels and organized into topological layers based on their dependencies +2. **Start everything simultaneously**: When you call `run()`, all tasks spawn at once—there's no scheduler deciding when each task should start +3. **Let the primitives handle coordination**: Channels naturally enforce execution order—tasks wait on their input channels until upstream tasks send data. No custom orchestration needed. +4. **Runtime joins on completion**: The runtime simply spawns all tasks and waits for the completion channel to close. That's it. + +## The Implementation + +Under the hood, dagx uses: + +- **Oneshot channels**: Each edge in the DAG gets a `futures::channel::oneshot` - producer sends once, consumer receives once +- **Ownership model**: Tasks take ownership (`self`) and are consumed during execution - no Mutex needed for task state +- **Direct data flow**: Outputs flow from producer to consumer via channels, never stored in shared memory during execution +- **Fast-path optimization**: Single-task layers execute inline without spawning overhead + +The core execution logic in `run()` is remarkably simple: + +```rust +// Create oneshot channel for each edge +for (producer, consumer) in edges { + let (tx, rx) = oneshot::channel(); + // Wire tx to producer, rx to consumer +} + +// Execute layer by layer +for layer in layers { + if layer.len() == 1 { + // Fast path: Execute inline, no spawning overhead + // IMPORTANT: Panic handling is required here! + let result = task.execute().catch_unwind().await; + // Convert panic to error to match spawned task behavior + } else { + // Parallel path: Spawn all tasks in layer + for task in layer { + spawner(async { + let input = await_on_input_channels(); // Blocks until deps complete + let output = task.run(input); // Consumes task + send_to_output_channels(output); // Sends to dependents + }); + } + // Wait for layer completion + } +} + +// Tasks coordinate themselves via channels - no scheduler needed +``` + +No state machines. No task queues. No wake-up logic. No Mutex locks. Just channels doing what channels do, with an inline fast-path for sequential execution. + +## Inline Execution Fast-Path + +**Performance optimization for sequential workloads**: When a layer contains only a single task (common in deep chains and linear pipelines), dagx executes it inline rather than spawning it. This eliminates spawning overhead, context switching, and channel creation, resulting in 10-100x performance improvements for sequential patterns. + +**Panic handling guarantee**: To maintain behavioral consistency between inline and spawned execution, panics in inline tasks are caught using `FutureExt::catch_unwind()` and converted to errors. This matches the behavior of all major async runtimes (Tokio, async-std, smol, embassy-rs), which catch panics in spawned tasks and convert them to `JoinError` or equivalent. + +**Why this matters**: + +- **Spawned tasks** (layer.len() > 1): Runtime catches panics automatically → becomes error +- **Inline tasks** (layer.len() == 1): We catch panics manually → becomes error +- **Result**: Identical behavior regardless of execution path + +This ensures your code behaves the same whether a task runs inline or spawned, making dagx's optimizations transparent and predictable. + +## Benefits + +**Simplicity**: The runtime is straightforward: create channels, spawn tasks, let them coordinate via awaiting. No complex scheduler code to maintain, debug, or optimize. + +**Reliability**: Built on battle-tested primitives (oneshot channels, async/await) from Rust's standard library and the futures crate. These have been used in production by thousands of projects and are orders of magnitude more reliable than custom scheduling logic. + +**Bug resistance**: Fewer moving parts means fewer places for bugs to hide. The type system enforces correct wiring at compile time. Channels handle synchronization. The ownership model prevents data races. What's left to break? + +**Performance**: Near zero-overhead. No Mutex locks during execution. Arc reference counting for efficient fanout (atomic operations, not locks). Oneshot channels are often lock-free. Tasks start as soon as their dependencies complete - maximum parallelism. + +**Auditability**: Want to verify correctness? Check the channel wiring, verify tasks await their inputs, done. No need to trace through complex state machine transitions or wake-up cascades. + +## The Insight + +The key insight is that **dependencies ARE the schedule**. If task B depends on task A's output, a channel naturally enforces that B waits for A. The dependency graph already encodes all the scheduling information—we just need to wire up channels to match it. + +This is dagx's core philosophy: leverage the type system for correctness, use primitives for coordination, and let the compiler optimize everything else away. + +## Measured Overhead + +How much overhead does this approach actually add? Benchmarks on an AMD 7840U (Zen 4 laptop CPU) show: + +**DAG Construction**: + +- Empty DAG creation: **~20 nanoseconds** +- Adding tasks: **~100 nanoseconds per task** +- Building a 10,000-task DAG: **~1.0 milliseconds** (100 ns/task) + +**Execution Overhead** (framework coordination, excluding actual task work): + +- Sequential workloads: **~790 nanoseconds per task** (inline execution fast-path) +- Parallel workloads: **~1.3 microseconds per task** +- 100-task sequential chain: **~79 microseconds total** +- 100 independent tasks: **~122 microseconds total** +- 10,000 independent tasks: **~12.7 milliseconds total** + +**Scaling**: Sub-microsecond per-task overhead across all workload patterns. Linear scaling verified to 10k+ tasks. + +**Comparison**: dagx is **1.04-129x faster** than dagrs (v0.5) across all benchmark patterns (see [comparisons](COMPARISONS.md)). + +The primitives-as-scheduler approach with inline fast-path optimization delivers exceptional performance: coordination overhead is sub-microsecond per task, and for real-world workloads where tasks do meaningful work (I/O, computation, etc.), framework overhead is negligible—typically well under 1% of total execution time. diff --git a/docs/TRACING.md b/docs/TRACING.md new file mode 100644 index 0000000..c47edf9 --- /dev/null +++ b/docs/TRACING.md @@ -0,0 +1,55 @@ +# Tracing Support + +dagx provides optional observability through the `tracing` crate with **zero runtime overhead when disabled**. The tracing instrumentation is conditionally compiled using feature flags, meaning when the feature is off, the logging code doesn't exist in the compiled binary at all. + +## Enabling Tracing + +```toml +[dependencies] +dagx = { version = "0.3", features = ["tracing"] } +tracing-subscriber = "0.3" +``` + +## Usage Example + +```rust +use tracing_subscriber::{fmt, EnvFilter}; + +// Initialize tracing subscriber +fmt() + .with_env_filter( + EnvFilter::try_from_default_env() + .unwrap_or_else(|_| EnvFilter::new("dagx=info")) + ) + .init(); + +// Your DAG code here - tracing will log execution details +``` + +## Log Levels + +- **INFO**: DAG execution start/completion +- **DEBUG**: Task additions, dependency wiring, layer computation +- **TRACE**: Individual task execution (inline vs spawned), detailed execution flow +- **ERROR**: Task panics, concurrent execution attempts + +Run with different levels: + +```bash +RUST_LOG=dagx=info cargo run # High-level execution info +RUST_LOG=dagx=debug cargo run # Task and layer details +RUST_LOG=dagx=trace cargo run # All execution details +``` + +See [`examples/tracing_example.rs`](../examples/tracing_example.rs) for a complete working example. + +## Zero-Cost Guarantee + +When the `tracing` feature is disabled (the default), there is **literally 0ns overhead**: + +- Logging code is removed at compile time via `#[cfg(feature = "tracing")]` +- No branches, no function calls, nothing +- The `tracing` crate isn't even linked +- Benchmarks verify identical performance with/without the feature + +This follows the same pattern used by Tokio, Hyper, and other performance-critical Rust async libraries. diff --git a/src/runner.rs b/src/runner.rs index 19fd6e9..ec4120d 100644 --- a/src/runner.rs +++ b/src/runner.rs @@ -238,7 +238,13 @@ impl DagRunner { /// /// # Errors /// - /// Returns `DagError::CycleDetected` if the DAG contains a cycle. + /// - Returns `DagError::ConcurrentExecution` if the DAG is already executing + /// - Returns `DagError::TaskPanicked` if any task panics during execution + /// + /// # Note on Cycles + /// + /// Cycles are impossible via the public API due to compile-time prevention. + /// See the [`crate::cycle_prevention`] module for details. /// /// # Examples ///