Implementing Asynchronous Node Execution with Cancellation in CGraph

CGraph provides asyncRun() and asyncProcess() methods that return std::future<CStatus> and support runtime cancellation through a global state machine managed by GElementRepository.

CGraph is a lightweight graph-based execution engine where each node (GNode) can execute synchronously or asynchronously with full lifecycle control. This guide explains how to implement asynchronous node execution with cancellation in CGraph using the asyncRun and asyncProcess APIs, based on the implementation in the chunelfeng/cgraph repository.

Architecture of Asynchronous Execution

Core Classes and Components

The asynchronous architecture centers on three primary classes that manage execution state and thread control:

• GPipeline – The high-level orchestrator exposed in src/GraphCtrl/GraphPipeline/GPipeline.h that provides asyncRun(), asyncProcess(), cancel(), suspend(), and resume() methods.
• GElementRepository – The central store implemented in src/GraphCtrl/GraphElement/GElementRepository.cpp that tracks runtime states (NORMAL, CANCEL, SUSPEND) for all registered elements.
• GElement – The base class for every node that checks repository state before execution via repository_->isCancelState().

During pipeline initialization in GElementRepository::init(), nodes marked as asynchronous through isAsync() are recorded in the async_elements_ collection (lines 29‑31).

Execution Flow

The asynchronous execution flow follows these steps:

1. Thread Spawning – GPipeline::asyncRun() creates a new execution thread via std::async (see src/GraphCtrl/GraphPipeline/GPipeline.cpp, lines 59‑68).
2. Topological Traversal – The spawned thread traverses the graph, invoking each node's run() or asyncRun() method.
3. State Monitoring – Before processing, each node checks the global state stored in GElementRepository.
4. Lifecycle Completion – asyncProcess() extends this flow to include init(), repeated run() cycles, and destroy() phases.

How Cancellation Works in CGraph

State Propagation via GElementRepository

CGraph implements cancellation through a global state broadcast mechanism. When GPipeline::cancel() is invoked, it delegates to the repository:

CStatus GPipeline::cancel() {
    CGRAPH_ASSERT_INIT(true);
    return repository_.pushAllState(GElementState::CANCEL);
}

The pushAllState() method in src/GraphCtrl/GraphElement/GElementRepository.cpp updates both the global cur_state_ flag and each individual element's state:

CStatus GElementRepository::pushAllState(GElementState newState) {
    cur_state_ = newState;               // update global flag
    for (auto* element : elements_) {
        element->state_ = newState;      // broadcast to each node
    }
    return CStatus();
}

This design ensures that all nodes observe cancellation immediately on their next state check.

Node-Level State Checking

Individual nodes query the repository before executing work. The typical pattern inside GElement::run() or GElement::asyncRun() includes:

if (repository_->isCancelState()) {
    return CStatus::CANCEL();   // early exit
}

The isCancelState() helper checks the repository's cur_state_ field. While this state is not atomic, the implementation assumes cancellation occurs in a controlled window where data races are not a concern.

Cancellation Example

The tutorial file tutorial/T19-Cancel.cpp demonstrates the complete workflow:

auto result = pipeline->asyncRun();      // start async execution
CGRAPH_SLEEP_MILLISECOND(1500);         // let it run a bit
pipeline->cancel();                     // request cancellation
result.wait();                          // wait for async thread to finish

Practical Implementation Patterns

Basic Async Run with Cancellation

To execute a pipeline asynchronously with cancellation support:

GPipelinePtr pipeline = GPipelineFactory::create();
pipeline->registerGElement<MyNode1>(&a, {}, "nodeA");
pipeline->init();

auto future = pipeline->asyncRun();   // launch async run
std::this_thread::sleep_for(std::chrono::milliseconds(500));
pipeline->cancel();                  // ask the pipeline to stop
future.wait();                       // block until the thread returns
pipeline->destroy();
GPipelineFactory::remove(pipeline);

The asyncRun() method accepts an optional std::launch policy parameter (defaulting to std::launch::async) that controls whether execution occurs on a new thread or the caller's thread if deferred.

Async Process with Loop Control

For full lifecycle management including initialization and destruction:

pipeline->init();
auto future = pipeline->asyncProcess(10);  // run 10 loops asynchronously
// … do other work …
pipeline->cancel();                       // abort the remaining loops
future.wait();                            // ensure cleanup
pipeline->destroy();

The asyncProcess(CSize runTimes, std::launch policy) method executes init → run (repeat) → destroy asynchronously, making it suitable for batch processing scenarios requiring periodic cancellation.

Suspend and Resume Functionality

Beyond cancellation, CGraph supports pausing execution via the same state mechanism. In tutorial/T20-Suspend.cpp:

auto handle = pipeline->asyncRun();
CGRAPH_SLEEP_MILLISECOND(2000);
pipeline->suspend();      // pause all nodes
CGRAPH_SLEEP_MILLISECOND(3000);
pipeline->resume();       // resume execution
handle.wait();

The suspend() method broadcasts GElementState::SUSPEND through pushAllState(), while resume() restores GElementState::NORMAL.

Summary

• asyncRun() and asyncProcess() are thin wrappers around std::async defined in src/GraphCtrl/GraphPipeline/GPipeline.cpp, returning std::future<CStatus> for async result retrieval.
• Cancellation is state-based – GPipeline::cancel() propagates GElementState::CANCEL through GElementRepository::pushAllState(), updating all registered elements simultaneously.
• Nodes check state reactively – Each GElement queries repository_->isCancelState() before work, enabling cooperative cancellation without forced thread termination.
• Suspend/resume use identical mechanics – The same state broadcast system supports pausing execution via suspend() and resume() methods.

Frequently Asked Questions

How does CGraph handle async execution internally?

CGraph uses std::async to spawn execution threads. The asyncRun() method in src/GraphCtrl/GraphPipeline/GPipeline.cpp (lines 59‑68) forwards to std::async with a configurable launch policy, while asyncProcess() wraps the complete init → run → destroy lifecycle in the same async pattern.

What is the difference between asyncRun and asyncProcess?

asyncRun() executes only the run() phase of the pipeline asynchronously, assuming init() was already called. asyncProcess() handles the complete lifecycle asynchronously—calling init(), running the specified number of iterations, and invoking destroy()—making it suitable for standalone async workflows.

Is the cancellation mechanism thread-safe?

The cancellation mechanism relies on a non-atomic state variable in GElementRepository. The design assumes cancellation is requested from the control thread while worker threads check the state cooperatively. While not using explicit atomic operations, the implementation assumes a safe synchronization window where the state change is visible to all threads before subsequent operations.

Can I cancel a specific node instead of the entire pipeline?

The current implementation in GElementRepository::pushAllState() broadcasts state changes to all elements in the elements_ collection. There is no public API to cancel individual nodes; cancellation is a pipeline-level operation. To stop specific nodes, you would need to implement custom logic within the node's run() method checking for custom conditions separate from the global repository state.

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