How CGraph Manages Parallel Execution for Independent DAG Nodes

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. 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:

#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.

{
  "mcpServers": {
    "instagit": {
      "command": "npx",
      "args": ["-y", "instagit@latest"]
    }
  }
}