Optimizing CGraph Pipeline Throughput and Performance: A Complete Guide

Optimize CGraph pipeline throughput and performance by configuring unique thread pools for CPU-bound workloads, trimming redundant DAG edges, forcing serial execution for low-parallelism graphs, and profiling critical paths with the built-in GPerf analyzer.

The chunelfeng/cgraph library provides a high-performance directed acyclic graph (DAG) execution engine for C++ applications. Maximizing CGraph pipeline throughput and performance requires careful tuning of thread-pool scheduling, graph topology, and execution instrumentation according to the source implementation in src/GraphCtrl/GraphPipeline/.

Configure Thread-Pool Scheduling for Maximum Parallelism

CGraph executes GElement objects through a thread pool managed by GPipeline. The library offers two scheduling modes that directly impact throughput.

Unique Thread Pools for CPU-Bound Workloads

Unique scheduling (GScheduleType::UNIQUE) creates a private thread pool for the pipeline. This mode enables fine-grained control via UThreadPoolConfig and is required for serial execution mode. Configure core threads, maximum threads, and monitoring overhead to match your CPU constraints.

auto *pipeline = new GPipeline();
UThreadPoolConfig cfg;
cfg.default_thread_size_ = 4;          // core threads
cfg.max_thread_size_     = 8;          // allow burst
cfg.monitor_enable_      = false;      // disable extra monitoring
pipeline->setUniqueThreadPoolConfig(cfg)   // src/GraphCtrl/GraphPipeline/GPipeline.h lines 38-44
        ->setGEngineType(GEngineType::TOPOL)
        ->process();

Shared Thread Pools for Concurrent Pipelines

Shared scheduling (GScheduleType::SHARED) reuses a global pool, reducing allocation overhead when many pipelines run concurrently. However, you cannot change thread counts per pipeline. Use setSharedThreadPool as defined in src/GraphCtrl/GraphPipeline/GPipeline.h (lines 52-57) when spawning many short-lived pipelines.

Optimize Graph Structure and Dependencies

Graph shape determines available parallelism. Redundant edges and unnecessary dependencies serialize execution and degrade throughput.

Trim Redundant Edges with trim()

The trim() method removes transitive dependencies that do not affect execution order, shrinking the DAG and exposing more parallelism. The implementation in src/GraphCtrl/GraphPipeline/GPipeline.cpp (lines 66-71) and declaration in GPipeline.h (lines 67-73) returns the count of removed edges.

CSize trimmed = pipeline->trim();
std::cout << "Trimmed " << trimmed << " redundant edges.\n";

Force Serial Execution with makeSerial()

When the graph has low inherent parallelism, thread-pool overhead exceeds the benefit of task switching. The makeSerial() method forces single-threaded execution, eliminating pool contention. This works only with unique scheduling and requires the graph to pass element_manager_->checkSerializable().

The implementation resides in src/GraphCtrl/GraphPipeline/GPipeline.cpp (lines 75-92) with the interface in GPipeline.h (lines 74-85).

if (pipeline->makeSerial() == CStatus()) {
    std::cout << "Pipeline forced to serial execution.\n";
}

Minimize Inter-Element Dependencies

Every dependency creates a happens-before edge. Use checkSeparate (defined in GPipeline.h lines 101-122) to verify if two elements can run in parallel. Prefer data-flow via GParam over explicit dependencies, and group related nodes into GGroup objects to limit external edges.

bool independent = pipeline->checkSeparate(nodeA, nodeB);
std::cout << (independent ? "Can run in parallel" : "Must be sequential") << "\n";

Profile and Identify Bottlenecks with GPerf

The GPerf subsystem provides lightweight instrumentation that records timestamps for each element and highlights the longest execution path without modifying node logic.

Inject Profiling Aspects

GPerf uses aspect-oriented programming to attach collectors. The GPerf::inject method (lines 39-52 in src/GraphCtrl/GraphPipeline/_GPerf/GPerf.cpp) allocates GPerfInfo and attaches a GPerfAspect to every GElement. After execution, GPerf::markLongestPath (lines 56-78) walks all paths collected by GOptimizer::collectPaths to identify the critical path. Finally, GPerf::recover (lines 82-89) removes aspects and frees resources.

Generate Performance Reports

Use GPipeline::perf (declared in GPipeline.h lines 94-99) to output a DOT graph with timing annotations. The Python-friendly wrapper __perf_4py returns the graph as a string.

// Print DOT graph to stdout
pipeline->perf(std::cout);

// Or capture for Python visualization
std::string dot = pipeline->__perf_4py();

The generated DOT includes fields defined in GPerfDefine.h: first_start_ts_ (first start time), last_finish_ts_ (last finish), accu_cost_ts_ (total cost across loops), and in_longest_path_ (critical path membership). Visualize the output using Graphviz tools to verify that critical path length shrinks and parallel sections expand after each optimization.

Leverage Asynchronous Execution for Throughput

When pipelines are I/O-bound or you must overlap multiple executions, use asynchronous helpers to avoid blocking the main thread.

The asyncProcess method (implemented in src/GraphCtrl/GraphPipeline/GPipeline.cpp lines 71-78) returns a std::future and respects the pipeline's schedule configuration. It propagates cancellation, suspension, and resumption across the future boundary. For lower-level control, asyncRun (lines 59-69) initiates a single asynchronous cycle.

// Execute pipeline 10 times asynchronously
auto fut = pipeline->asyncProcess(10, std::launch::async);
CStatus status = fut.get();   // Blocks until all runs finish

Summary

• Configure unique thread pools via setUniqueThreadPoolConfig in src/GraphCtrl/GraphPipeline/GPipeline.h (lines 38-44) to fine-tune CPU-bound workloads, or use setSharedThreadPool (lines 52-57) to amortize overhead across many short-lived pipelines.
• Trim redundant edges using trim() (implemented in GPipeline.cpp lines 66-71) to shrink the DAG and expose parallelism.
• Force serial execution with makeSerial() (GPipeline.cpp lines 75-92) when thread-pool overhead exceeds parallel gains, but only with unique scheduling.
• Profile critical paths using GPerf::inject, markLongestPath, and GPipeline::perf (lines 94-99) to visualize bottlenecks in DOT format.
• Execute asynchronously via asyncProcess (GPipeline.cpp lines 71-78) to overlap I/O-bound pipelines and maximize throughput.

Frequently Asked Questions

When should I use a unique thread pool versus a shared one?

Choose a unique thread pool when your pipeline is CPU-bound and requires fine-grained control over thread counts, or when you need to force serial execution via makeSerial(). According to src/GraphCtrl/GraphPipeline/GPipeline.h (lines 38-44), setUniqueThreadPoolConfig accepts a UThreadPoolConfig object to tune default_thread_size_ and max_thread_size_. Use a shared pool via setSharedThreadPool (lines 52-57) when spawning many short-lived pipelines to amortize thread creation costs, though you lose per-pipeline configuration ability.

What is the difference between trim() and makeSerial()?

trim() removes transitive dependencies (redundant edges) from the DAG, which increases available parallelism by allowing the scheduler to execute more elements concurrently. It is implemented in src/GraphCtrl/GraphPipeline/GPipeline.cpp (lines 66-71). In contrast, makeSerial() forces the entire pipeline to execute on a single thread, eliminating thread-pool overhead when the graph has low inherent parallelism. This method requires unique scheduling and is implemented in GPipeline.cpp (lines 75-92).

How does GPerf identify the critical path without modifying my node logic?

GPerf uses aspect-oriented programming to inject timing collectors transparently. The GPerf::inject method (lines 39-52 in src/GraphCtrl/GraphPipeline/_GPerf/GPerf.cpp) allocates GPerfInfo structures and attaches a GPerfAspect to every GElement. After execution, GPerf::markLongestPath (lines 56-78) analyzes all paths collected by GOptimizer::collectPaths to highlight the critical path. Finally, GPerf::recover (lines 82-89) removes the aspects and frees resources, leaving your original node implementation untouched.

Can I switch between serial and parallel execution at runtime?

Yes, but with constraints. You can call makeSerial() to force serial execution at any point before process(), provided the pipeline uses a unique thread pool and the graph passes element_manager_->checkSerializable(). To return to parallel execution, you must reconfigure the pipeline with setUniqueThreadPoolConfig again, as serial mode modifies internal scheduling state. You cannot use makeSerial() with shared thread pools. Use checkSeparate (lines 101-122 in GPipeline.h) to verify element parallelism before switching modes.

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