Soft fork of dagx for Operese

dagx

A minimal, type-safe, runtime-agnostic async DAG (Directed Acyclic Graph) executor with compile-time cycle prevention and true parallel execution.

Why dagx?

Blazing Fast: 1-100x faster than dagrs

Simple API

let sum = dag.add_task(Add).depends_on((x, y));
dag.run(|fut| async move { tokio::spawn(fut).await.unwrap() }).await?;

That's it. No trait boilerplate, no manual channels, no node IDs.

Quick Start

Add to your Cargo.toml:

[dependencies]
dagx = "0.4"
tokio = { version = "1", features = ["macros", "rt-multi-thread"] }

Basic example:

use operese_dagx::{task, DagRunner, Task};

// Define tasks with the #[task] macro

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
    }
}

#[tokio::main]
async fn main() {
    let mut dag = DagRunner::new();

    // Add source tasks with no dependencies
    let x = dag.add_task(Value(2));
    let y = dag.add_task(Value(3));

    // Add task that depends on both x and y.
    let sum = dag.add_task(Add).depends_on((x, y));

    // Execute with true parallelism
    let mut output = dag.run(|fut| async move { tokio::spawn(fut).await.unwrap() }).await.unwrap();

    // Retrieve results
    assert_eq!(output.get(sum), 5);
}

Features

Compile-Time Safety

• 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

See how it works.

Runtime Agnostic

dagx works with any async runtime. Provide a spawner function to run():

// With Tokio
// The join handle result can be unwrapped because operese_dagx catches panics internally
dag.run(|fut| async move { tokio::spawn(fut).await.unwrap() }).await.unwrap();

// With smol
dag.run(|fut| smol::spawn(fut)).await.unwrap();

// Single-threaded concurrency on the invoking runtime
// Can be faster in situations where waiting time dominates
dag.run(|fut| fut).await.unwrap()

Task Patterns

dagx supports three task patterns:

1. Stateless - Pure functions with no state:

struct Add;

#[task]
impl Add {
    async fn run(a: &i32, b: &i32) -> i32 { a + b }
}

2. Read-only state - Configuration accessed via &self:

struct Multiplier(i32);

#[task]
impl Multiplier {
    async fn run(&self, input: &i32) -> i32 { input * self.0 }
}

3. Mutable state - State modification via &mut self:

struct Counter(i32);

#[task]
impl Counter {
    async fn run(&mut self, value: &i32) -> i32 {
        self.0 += value;
        self.0
    }
}

Tracing

dagx provides optional observability using the tracing crate, controlled by the tracing feature flag.

Enabling Tracing

[dependencies]
dagx = { version = "0.4", features = ["tracing"] }
tracing-subscriber = "0.3"

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

Other

• True parallelism: Chosen runtime executes tasks concurrently and/or in parallel
• No boilerplate: The derive feature and the #[task] macro are enabled by default to simplify task implementation.

Performance

dagx provides true parallel execution with sub-microsecond overhead per task.

How is dagx so fast?

• Inline fast-path: Sequential chains execute inline without spawning
• Adaptive execution: Inline for sequential work, executor-agnostic parallelism for concurrent work
• Zero-cost abstractions: Compile-time graph validation eliminates overhead

See design philosophy for details.

Tutorials & Examples

Tutorials (Start Here)

Step-by-step introduction to dagx:

• 01_basic.rs - Your first DAG
• 02_fan_out.rs - One task feeds many (1→N)
• 03_fan_in.rs - Many tasks feed one (N→1)
• 04_parallel_computation.rs - Map-reduce with true parallelism

Run tutorial examples:

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 - Circuit breaker pattern for resilient systems
• data_pipeline.rs - ETL data processing pipeline
• error_handling.rs - Error propagation and recovery

Run any example: cargo run --example circuit_breaker

Documentation

Full API documentation is available at docs.rs/dagx.

Detailed documentation on dagx's internals and advanced features:

• Compile-Time Cycle Prevention - How the type system prevents cycles
• Design Philosophy - Primitives as scheduler, inline fast-path optimization
• Library Comparisons - Detailed comparison with dagrs, async_dag, and others

When to use operese_dagx

dagx is ideal for:

• Data pipelines with complex dependencies between stages
• Build systems where tasks depend on outputs of other tasks
• Parallel computation where work can be split and aggregated
• Workflow engines with typed data flow between stages
• ETL processes with validation and transformation steps

Benchmarks

Run the full benchmark suite:

cargo bench

View detailed HTML reports:

# macOS
open target/criterion/report/index.html

# Linux
xdg-open target/criterion/report/index.html

# Windows
start target/criterion/report/index.html

Benchmarks run on Intel i9-13950HX @ 5.5GHz.

Code of Conduct

This project follows the Builder's Code of Conduct.

Contributing

Contributions are welcome! Please see CONTRIBUTING.md for guidelines.

For security issues, see SECURITY.md.

License

Licensed under the MIT License. See LICENSE for details.

Copyright (c) 2025 Stephen Waits steve@waits.net