# How CGraph Manages Parallel Execution for Independent DAG Nodes

> Discover how CGraph manages parallel execution for independent DAG nodes using GDynamicEngine and a thread pool. Learn about its automatic graph detection and thread-safe synchronization.

- Repository: [Chunel/cgraph](https://github.com/chunelfeng/cgraph)
- Tags: internals
- Published: 2026-02-27

---

**CGraph manages parallel execution for independent DAG nodes through the `GDynamicEngine` class, which automatically detects fully parallel graphs, constructs a parallelization matrix, and dispatches work to an internal thread pool (`UThreadPool`) using lock-free atomic counters for synchronization.**

CGraph is a high-performance DAG (Directed Acyclic Graph) execution framework designed to optimize throughput by automatically parallelizing independent tasks. When processing a pipeline where all end nodes lack dependencies, the engine switches to a specialized **parallel execution mode** that maximizes hardware utilization. This article examines the implementation details in `chunelfeng/cgraph` that enable efficient parallel execution for independent DAG nodes.

## Detecting Fully Parallel DAGs with GDynamicEngine

The process begins with DAG classification in [`src/GraphCtrl/GraphElement/_GEngine/GDynamicEngine/GDynamicEngine.cpp`](https://github.com/chunelfeng/cgraph/blob/main/src/GraphCtrl/GraphElement/_GEngine/GDynamicEngine/GDynamicEngine.cpp). The method `analysisDagType()` evaluates the graph structure to determine if it qualifies as `ALL_PARALLEL`.

When the total number of end nodes equals the total number of elements (`total_end_size_ == total_element_arr_.size()`), the engine recognizes that every node is independent and can run simultaneously. This classification triggers the construction of a parallelization matrix rather than the standard dependency-tracking execution path.

### Building the Parallelization Matrix

Once classified as `ALL_PARALLEL`, the engine invokes `analysisParallelMatrix()` (lines 102-121) to partition the independent nodes into batches that align with the thread pool's capacity. The result is stored in `parallel_element_matrix_`, a two-dimensional vector where each inner vector represents a batch of elements that can be executed concurrently.

This matrix structure ensures optimal load distribution while respecting thread affinity constraints. Each row in the matrix corresponds to a specific worker thread ID, which improves cache locality when tasks are dispatched.

## Thread Pool Dispatch and Execution Strategies

With the matrix constructed, `parallelRunAll()` (lines 214-265) orchestrates the actual parallel execution. The method iterates over `parallel_element_matrix_` and dispatches each batch to the `UThreadPool` instance stored in the base class `GEngine` as `thread_pool_`.

For each element in a batch, the engine calls `thread_pool_->executeWithTid()`, passing the matrix row index as the `tid` parameter. This thread-affinity hint ensures that related tasks execute on the same worker thread, reducing context switching and improving cache performance. When a batch contains only a single element, the engine falls back to the simpler `execute()` method to minimize overhead.

### Micro-Batch Mode for Performance Testing

If the `_CGRAPH_PARALLEL_MICRO_BATCH_ENABLE_` macro is defined, `parallelRunAll()` switches to a future-based micro-batch implementation. This mode collects `std::future` objects for each batch and aggregates results afterward, primarily used for performance benchmarking scenarios.

## Lock-Free Synchronization Mechanisms

Coordination between worker threads and the main thread relies on lock-free atomic operations. The `parallelRunOne()` method (lines 268-286) handles task completion detection through an atomic counter named `parallel_run_num_`.

Each worker thread increments this counter after finishing its assigned element. The main thread waits on a condition variable (`locker_.cv_`) until `parallel_run_num_` reaches `total_end_size_` or an error condition occurs. This approach eliminates contention during high-throughput execution while providing deterministic error handling across all parallel tasks.

## Practical C++ Implementation Example

The following example demonstrates how CGraph automatically detects and parallelizes independent nodes:

```cpp
#include "cgraph.h"

int main() {
    CGRAPH_NAMESPACE::GPipeline pipeline;
    
    // Create three independent elements with no dependencies
    auto *e1 = pipeline.addElement<CGRAPH_NAMESPACE::GFunction>([](){ 
        std::cout << "A\n"; 
    });
    auto *e2 = pipeline.addElement<CGRAPH_NAMESPACE::GFunction>([](){ 
        std::cout << "B\n"; 
    });
    auto *e3 = pipeline.addElement<CGRAPH_NAMESPACE::GFunction>([](){ 
        std::cout << "C\n"; 
    });

    // Register elements without establishing dependencies
    pipeline.registerGElement(e1);
    pipeline.registerGElement(e2);
    pipeline.registerGElement(e3);

    // Engine automatically selects ALL_PARALLEL path
    pipeline.run();   // Prints A, B, C in any order
}

```

In this implementation, the absence of dependencies between `e1`, `e2`, and `e3` causes `GDynamicEngine` to classify the DAG as fully parallel and distribute the three functions across the internal thread pool.

## Summary

- **Automatic Detection**: `GDynamicEngine::analysisDagType()` identifies fully parallel graphs when all nodes are end nodes with no dependencies.
- **Matrix Construction**: `analysisParallelMatrix()` partitions independent nodes into the `parallel_element_matrix_` structure to optimize thread pool utilization.
- **Affinity Dispatch**: The engine uses `UThreadPool::executeWithTid()` to maintain thread affinity and cache locality during parallel execution.
- **Lock-Free Coordination**: An atomic counter (`parallel_run_num_`) and condition variables provide efficient synchronization without blocking worker threads.

## Frequently Asked Questions

### How does CGraph determine if a DAG qualifies for parallel execution?

CGraph checks if `total_end_size_` equals `total_element_arr_.size()` in `GDynamicEngine::analysisDagType()`. When this condition is true, every node in the graph is an independent end node, allowing the engine to classify the DAG as `ALL_PARALLEL` and bypass dependency tracking.

### What is the difference between `execute()` and `executeWithTid()` in CGraph's thread pool?

`execute()` submits a task to the general queue for any available worker, while `executeWithTid()` targets a specific thread ID to maintain affinity. CGraph uses `executeWithTid()` when dispatching batches from the parallel matrix to improve cache locality, falling back to `execute()` only for single-element batches to reduce overhead.

### How does CGraph synchronize parallel tasks without blocking the main thread?

The engine employs a lock-free atomic counter (`parallel_run_num_`) that worker threads increment upon completion. The main thread waits on `locker_.cv_` (a condition variable) until the counter matches `total_end_size_` or an error occurs, enabling efficient coordination without busy-waiting or excessive locking.

### Can CGraph handle mixed graphs with both parallel and serial nodes?

Yes. While this article focuses on the `ALL_PARALLEL` optimization for independent DAG nodes, `GDynamicEngine` supports mixed execution through the "common" DAG type. In such cases, the engine uses dependency-tracking mechanisms rather than the parallel matrix to ensure correct ordering while still parallelizing independent branches.