# How CGraph Implements Pausing and Resuming Pipeline Execution: A Technical Deep Dive

> Learn how CGraph implements pausing and resuming pipeline execution. Discover its state management system that allows cooperative blocking and resuming of pipeline elements.

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

---

**CGraph supports pausing and resuming pipeline execution through a centralized state management system where the pipeline broadcasts state changes to all registered elements, which then cooperatively block on condition variables until the resume signal is received.**

CGraph is a high-performance C++ graph computing framework designed for complex data pipeline orchestration. When building long-running or interactive applications, understanding how to control execution flow through pausing and resuming pipeline execution becomes essential for resource management and responsive system design.

## Architecture of the Pause/Resume System

The pause/resume functionality in CGraph operates through four coordinated components that manage state propagation from the pipeline level down to individual execution elements.

| Component | Source File | Role |
|-----------|-------------|------|
| **`GPipeline`** | [`src/GraphCtrl/GraphPipeline/GPipeline.cpp`](https://github.com/chunelfeng/cgraph/blob/main/src/GraphCtrl/GraphPipeline/GPipeline.cpp) | Exposes `suspend()` and `resume()` public APIs that initiate state transitions. |
| **`GElementRepository`** | [`src/GraphCtrl/GraphElement/GElementRepository.cpp`](https://github.com/chunelfeng/cgraph/blob/main/src/GraphCtrl/GraphElement/GElementRepository.cpp) | Maintains the registry of all `GElement` objects and broadcasts state changes via `pushAllState()`. |
| **`GElement`** | [`src/GraphCtrl/GraphElement/GElement.cpp`](https://github.com/chunelfeng/cgraph/blob/main/src/GraphCtrl/GraphElement/GElement.cpp) | Implements `checkSuspend()` to block execution when the element state is `SUSPEND`. |
| **`GElementState`** | [`src/GraphCtrl/GraphElement/GElementDefine.h`](https://github.com/chunelfeng/cgraph/blob/main/src/GraphCtrl/GraphElement/GElementDefine.h) | Enumeration defining `NORMAL`, `SUSPEND`, `CANCEL`, and other execution states. |

## Pipeline State Transitions

State changes originate at the pipeline level through two primary methods that delegate to the element repository for distribution.

### Suspending Pipeline Execution

When `GPipeline::suspend()` is called, the pipeline validates its initialization state and forwards the suspend request to the repository:

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

```

This method immediately returns after broadcasting the state, leaving the actual blocking to occur within each element's execution loop.

### Resuming Pipeline Execution

The resume operation follows an identical pattern, broadcasting the `NORMAL` state to unlock waiting elements:

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

```

Once broadcast, elements blocked in `checkSuspend()` receive the notification and continue processing from their exact pause point.

## Broadcasting State Changes to Elements

The `GElementRepository::pushAllState()` method serves as the distribution hub for pause/resume signals. It iterates through all registered elements, updates their atomic state variables, and manages condition variable notifications:

```cpp
CStatus GElementRepository::pushAllState(const GElementState& state) {
    CGRAPH_FUNCTION_BEGIN
    if (cur_state_ == state) {
        return status;    // avoid duplicate work
    }

    for (auto& cur : elements_) {
        cur->cur_state_.store(state, std::memory_order_release);
        if (GElementState::SUSPEND != state) {
            // non‑suspend states must wake waiting elements
            cur->suspend_locker_.cv_.notify_one();
        }
    }
    cur_state_ = state;
    CGRAPH_FUNCTION_END
}

```

Notably, the method only triggers condition variable notifications when transitioning *away* from `SUSPEND`, preventing unnecessary wake-ups during the initial pause operation.

## Element-Level Blocking Mechanism

Individual elements implement cooperative multitasking by checking their state within the processing loop. The `GElement::fatProcessor` method calls `checkSuspend()` after each execution iteration:

```cpp
for (CSize i = 0; i < this->loop_ && status.isOK()
                     && GElementState::NORMAL == this->getCurState(); i++) {
    // … run logic …
    do {
        status += isAsync() ? asyncRun() : run();
    } while (checkSuspend(), this->isHold() && status.isOK());
}

```

The `checkSuspend()` method implements the actual blocking logic using a condition variable and mutex pair:

```cpp
CVoid GElement::checkSuspend() {
    if (GElementState::SUSPEND != cur_state_.load(std::memory_order_acquire)) {
        return;
    }

    std::unique_lock<std::mutex> lk(suspend_locker_.mtx_);
    this->suspend_locker_.cv_.wait(lk, [this] {
        return GElementState::SUSPEND != cur_state_.load(std::memory_order_acquire);
    });
}

```

When the pipeline's `resume()` call pushes `NORMAL` to all elements, the condition variable is signaled, the waiting thread wakes, and execution continues from the point where it was paused.

## Practical Code Examples

### Basic Synchronous Pause and Resume

The following example demonstrates pausing a running pipeline to perform intermediate work before resuming execution:

```cpp
#include "CGraph.h"

int main() {
    CGraph graph;
    auto pipeline = graph.createPipeline();

    // (register nodes / groups …)

    pipeline->run();          // normal execution
    pipeline->suspend();      // pause – all elements will block at the next checkSuspend()
    // … do other work while the pipeline is paused …
    pipeline->resume();       // continue execution
    pipeline->run();          // or start a new run
    return 0;
}

```

### Pausing Asynchronous Pipelines

For non-blocking pipeline execution, suspend and resume work seamlessly with asynchronous operations:

```cpp
auto future = pipeline->asyncRun();   // runs in a separate thread
std::this_thread::sleep_for(std::chrono::seconds(1));
pipeline->suspend();                 // blocks elements when they hit checkSuspend()
std::this_thread::sleep_for(std::chrono::seconds(2));
pipeline->resume();                  // wakes them up
future.get();                        // wait for completion

```

### Tutorial Example

The repository provides a complete working demonstration in [`tutorial/T20-Suspend.cpp`](https://github.com/chunelfeng/cgraph/blob/main/tutorial/T20-Suspend.cpp):

```cpp
// T20-Suspend.cpp
pipeline->run();
pipeline->suspend();   // pause
// ... some other logic ...
pipeline->resume();    // resume
pipeline->run();       // continue

```

Source: [[`T20-Suspend.cpp`](https://github.com/chunelfeng/cgraph/blob/main/T20-Suspend.cpp)](https://github.com/chunelfeng/cgraph/blob/main/tutorial/T20-Suspend.cpp)

## Key Implementation Files

| File | Purpose |
|------|---------|
| [[`src/GraphCtrl/GraphPipeline/GPipeline.h`](https://github.com/chunelfeng/cgraph/blob/main/src/GraphCtrl/GraphPipeline/GPipeline.h)](https://github.com/chunelfeng/cgraph/blob/main/src/GraphCtrl/GraphPipeline/GPipeline.h) | Public `suspend()` / `resume()` declarations. |
| [[`src/GraphCtrl/GraphPipeline/GPipeline.cpp`](https://github.com/chunelfeng/cgraph/blob/main/src/GraphCtrl/GraphPipeline/GPipeline.cpp)](https://github.com/chunelfeng/cgraph/blob/main/src/GraphCtrl/GraphPipeline/GPipeline.cpp) | Implementation of `suspend()` / `resume()` that forwards to the repository. |
| [[`src/GraphCtrl/GraphElement/GElementRepository.h`](https://github.com/chunelfeng/cgraph/blob/main/src/GraphCtrl/GraphElement/GElementRepository.h)](https://github.com/chunelfeng/cgraph/blob/main/src/GraphCtrl/GraphElement/GElementRepository.h) | Holds all elements and declares `pushAllState()`. |
| [[`src/GraphCtrl/GraphElement/GElementRepository.cpp`](https://github.com/chunelfeng/cgraph/blob/main/src/GraphCtrl/GraphElement/GElementRepository.cpp)](https://github.com/chunelfeng/cgraph/blob/main/src/GraphCtrl/GraphElement/GElementRepository.cpp) | Broadcasts the new state and notifies waiting elements. |
| [[`src/GraphCtrl/GraphElement/GElement.h`](https://github.com/chunelfeng/cgraph/blob/main/src/GraphCtrl/GraphElement/GElement.h)](https://github.com/chunelfeng/cgraph/blob/main/src/GraphCtrl/GraphElement/GElement.h) | Declares `checkSuspend()` used inside the execution loop. |
| [[`src/GraphCtrl/GraphElement/GElement.cpp`](https://github.com/chunelfeng/cgraph/blob/main/src/GraphCtrl/GraphElement/GElement.cpp)](https://github.com/chunelfeng/cgraph/blob/main/src/GraphCtrl/GraphElement/GElement.cpp) | Implements the blocking logic for the `SUSPEND` state. |
| [[`tutorial/T20-Suspend.cpp`](https://github.com/chunelfeng/cgraph/blob/main/tutorial/T20-Suspend.cpp)](https://github.com/chunelfeng/cgraph/blob/main/tutorial/T20-Suspend.cpp) | Sample program demonstrating pause‑resume usage. |

## Summary

- **CGraph** implements pausing and resuming pipeline execution through a centralized state management system that broadcasts signals from the pipeline level to individual elements.
- The **`GPipeline`** class provides **`suspend()`** and **`resume()`** methods that delegate state changes to the **`GElementRepository`**.
- **`GElementRepository::pushAllState()`** atomically updates the state of all registered elements and triggers condition variable notifications when transitioning away from the `SUSPEND` state.
- Each **`GElement`** cooperatively checks its state via **`checkSuspend()`**, blocking on a condition variable when the pipeline is paused and resuming execution exactly where it left off.
- This mechanism works identically for both synchronous and asynchronous pipeline execution, making it suitable for interactive applications and dynamic resource management scenarios.

## Frequently Asked Questions

### How does CGraph's pause/resume mechanism affect running threads?

When you call `suspend()`, CGraph does not forcibly terminate or interrupt threads. Instead, each `GElement` checks its state at well-defined points in its processing loop via `checkSuspend()`. If the state is `SUSPEND`, the element blocks on a condition variable inside `suspend_locker_`, effectively pausing that specific execution path while leaving the underlying thread pool intact. When `resume()` is called, the condition variables are notified and execution continues from the exact pause point.

### Can I pause an asynchronous pipeline without blocking the main thread?

Yes. The pause/resume mechanism in CGraph is designed to work seamlessly with asynchronous execution. When you call `pipeline->asyncRun()`, the pipeline executes in a separate thread. You can then call `suspend()` from your main thread to pause the pipeline's progress, perform other work, and later call `resume()` to continue execution. The `future` returned by `asyncRun()` will not complete until the pipeline finishes processing, regardless of how many pause/resume cycles occurred during execution.

### What happens if I call suspend() on a pipeline that is already paused?

CGraph prevents redundant state transitions through a guard clause in `GElementRepository::pushAllState()`. If the current state already matches the requested state (for example, calling `suspend()` when the pipeline is already in the `SUSPEND` state), the method returns immediately without notifying elements or updating state variables. This idempotent behavior ensures that multiple suspend calls do not corrupt the internal condition variable state or cause unnecessary wake-up events.

### Is the pause/resume mechanism thread-safe for concurrent modifications?

Yes. The implementation uses atomic operations and mutex-protected condition variables to ensure thread safety. The `cur_state_` variable in each `GElement` is updated using `std::memory_order_release` in the repository's broadcast loop, and read using `std::memory_order_acquire` in `checkSuspend()`. The condition variable wait in `checkSuspend()` is protected by `suspend_locker_.mtx_`, ensuring that state checks and blocking operations occur atomically. This design allows safe pause/resume operations even when the pipeline is actively processing data across multiple threads.