# Implementing Asynchronous Node Execution with Cancellation in CGraph

> Learn to implement asynchronous node execution and cancellation in CGraph using asyncRun and asyncProcess. Explore runtime cancellation via the global state machine.

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

---

**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`](https://github.com/chunelfeng/cgraph/blob/main/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`](https://github.com/chunelfeng/cgraph/blob/main/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`](https://github.com/chunelfeng/cgraph/blob/main/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:

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

```

The `pushAllState()` method in [`src/GraphCtrl/GraphElement/GElementRepository.cpp`](https://github.com/chunelfeng/cgraph/blob/main/src/GraphCtrl/GraphElement/GElementRepository.cpp) updates both the global `cur_state_` flag and each individual element's state:

```cpp
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:

```cpp
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`](https://github.com/chunelfeng/cgraph/blob/main/tutorial/T19-Cancel.cpp) demonstrates the complete workflow:

```cpp
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:

```cpp
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:

```cpp
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`](https://github.com/chunelfeng/cgraph/blob/main/tutorial/T20-Suspend.cpp):

```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`](https://github.com/chunelfeng/cgraph/blob/main/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`](https://github.com/chunelfeng/cgraph/blob/main/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.