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

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	Exposes suspend() and resume() public APIs that initiate state transitions.
GElementRepository	src/GraphCtrl/GraphElement/GElementRepository.cpp	Maintains the registry of all GElement objects and broadcasts state changes via pushAllState().
GElement	src/GraphCtrl/GraphElement/GElement.cpp	Implements checkSuspend() to block execution when the element state is SUSPEND.
GElementState	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:

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:

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:

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:

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:

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:

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

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:

// 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/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)	Public suspend() / resume() declarations.
[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)	Holds all elements and declares pushAllState().
[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)	Declares checkSuspend() used inside the execution loop.
[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)	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.

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