GStage: How to Synchronize CGraph Elements with Count-Down Barriers

GStage is a lightweight count-down barrier that pauses concurrent GElement objects at a synchronization point and releases them together once a configurable threshold is reached.

GStage serves as the core synchronization primitive inside CGraph, enabling fine-grained coordination between graph nodes without tight coupling. It allows multiple independently executing elements to wait at a named barrier until a specific count of arrivals triggers a simultaneous release.

What Is GStage in CGraph?

GStage implements a reusable count-down barrier pattern within the CGraph pipeline execution framework. According to the source code in src/GraphCtrl/GraphStage/GStage.h (lines 20–33), each stage maintains a threshold count, a current arrival counter, and an optional parameter object. When elements invoke the stage’s waiting() method, the internal counter increments atomically. Once cur_value_ reaches threshold_, the stage executes its launch() hook and broadcasts a notification to wake all blocked threads.

Unlike standard mutexes or semaphores, GStage is designed specifically for multi-element synchronization where a set of concurrent operations must reach a common checkpoint before any may proceed.

Architecture of the GStage Synchronization System

The synchronization capability spans three primary components that manage stage lifecycle, registration, and API exposure.

GStage Core Implementation

The GStage class defined in src/GraphCtrl/GraphStage/GStage.h encapsulates the barrier logic. It stores:

• threshold_: The target number of arrivals required to release the barrier
• cur_value_: The current count of elements waiting at the stage
• param_: An optional configuration object passed to the launch() callback
• locker_: A mutex and condition variable pair for thread-safe blocking

The implementation in src/GraphCtrl/GraphStage/GStage.cpp (lines 33–50) provides the waiting() method. This method increments cur_value_ under lock, compares it against threshold_, and either blocks the calling thread on locker_.cv_.wait() or triggers launch(param_) and invokes locker_.cv_.notify_all() to release waiters.

GStageManager for Stage Registry

GStageManager, located in src/GraphCtrl/GraphStage/GStageManager.h (lines 61–73), maintains a registry of named stages using std::unordered_map<std::string, GStagePtr>. It exposes:

• create(): Constructs a new GStage with a specified threshold and inserts it into stage_map_
• waitForReady(): Retrieves a stage by key and forwards the call to GStage::waiting()

This manager enables multiple elements to reference the same synchronization point via a string key.

GStageManagerWrapper Mixin

To simplify API access, GStageManagerWrapper (defined in src/GraphCtrl/GraphStage/GStageManagerWrapper.h, lines 23–28) acts as a mixin for GPipeline and GElement classes. It provides the enterStage(key) convenience method, which internally calls stage_manager_->waitForReady(key). This wrapper allows any graph element to synchronize with a single function call.

How GStage Synchronizes CGraph Elements

The synchronization flow follows a four-step process when elements execute within a pipeline:

1. Stage Creation – The pipeline creates a named stage using addGStage<GStage>(key, threshold). In tutorial/T28-Stage.cpp (line 28), this appears as pipeline->addGStage<GStage>(kStageKey, 3), establishing a barrier that releases after three arrivals.

2. Element Arrival – When an element reaches the synchronization point, it calls enterStage(kStageKey). As shown in tutorial/MyGNode/MyStageNode.h (lines 22–24), this invokes GStageManager::waitForReady(), which retrieves the corresponding GStage instance.

3. Counting and Release – Inside GStage::waiting(), the stage atomically increments cur_value_. If the count meets or exceeds threshold_, the stage executes launch(param_) (which can be overridden for custom logic), resets cur_value_ to zero, and calls notify_all() to wake every blocked thread.

4. Blocking Behavior – Threads arriving before the threshold is met block on the condition variable (locker_.cv_.wait), consuming no CPU until the final arrival triggers the mass release.

Practical Example: Implementing a GStage Barrier

The following example demonstrates three nodes synchronizing at a midpoint before allowing downstream execution.

First, define a node that enters the stage during its run() method:

// tutorial/MyGNode/MyStageNode.h
class MyStageNode : public CGraph::GNode {
    CStatus run() override {
        CGRAPH_SLEEP_SECOND(before);
        CGraph::CGRAPH_ECHO("[%s] wait for stage", getName().c_str());
        
        // Block until the stage with key "stage" is released
        enterStage(kStageKey);               // ← wrapper call
        CGraph::CGRAPH_ECHO("[%s] finish stage", getName().c_str());
        
        CGRAPH_SLEEP_SECOND(after);
        return CStatus();
    }
};

Next, configure the pipeline with a stage threshold of three:

// tutorial/T28-Stage.cpp
GPipelinePtr pipeline = GPipelineFactory::create();

pipeline->registerGElement<MyNode1>(&a, {}, "nodeA");
pipeline->registerGElement<MyStageNode<1,2>>(&b, {a}, "nodeB");
pipeline->registerGElement<MyStageNode<3,1>>(&c, {a}, "nodeC");
pipeline->registerGElement<MyStageNode<3,1>>(&d, {a}, "nodeD");
pipeline->registerGElement<MyNode1>(&e, {b, c, d}, "nodeE");

// Create a stage named "stage" that releases after 3 entries
pipeline->addGStage<GStage>(kStageKey, 3);   // ← creates GStage with threshold = 3

pipeline->process();

When nodeB, nodeC, and nodeD each invoke enterStage(kStageKey), the third arrival triggers the release. All three nodes unblock simultaneously, print their completion messages, and proceed to their downstream dependencies (in this case, nodeE).

Summary

• GStage provides a count-down barrier for synchronizing GElement objects in CGraph, defined in src/GraphCtrl/GraphStage/GStage.h.
• Threshold-based release occurs when cur_value_ reaches threshold_, implemented in GStage::waiting() at src/GraphCtrl/GraphStage/GStage.cpp (lines 33–50).
• Named registration via GStageManager allows multiple elements to share a single synchronization point using string keys.
• Convenient API access comes through GStageManagerWrapper::enterStage(), available to all pipeline elements.
• Reusability is built-in; once released, cur_value_ resets to zero, making the stage available for subsequent synchronization rounds.

Frequently Asked Questions

What is the difference between GStage and a standard C++ barrier?

GStage is specifically designed for CGraph’s GElement execution model and integrates with the pipeline’s stage manager. While standard C++ barriers (such as std::barrier) provide similar count-down semantics, GStage adds named registration via string keys, an optional launch() hook for custom side effects, and automatic reset for reuse within the graph execution context, as implemented in GStageManager.h.

How do you configure the threshold for a GStage barrier?

You set the threshold during stage creation by calling pipeline->addGStage<GStage>(key, threshold). The second parameter (e.g., 3 in tutorial/T28-Stage.cpp) becomes the threshold_ member in GStage. This value determines exactly how many elements must call enterStage() before the barrier releases.

Can GStage be reused after it releases?

Yes. According to the logic in src/GraphCtrl/GraphStage/GStage.cpp, when cur_value_ reaches threshold_, the implementation resets the counter to zero after waking waiters. This design allows the same stage instance to act as a reusable synchronization point for multiple iterations or separate batches of elements throughout the pipeline lifecycle.

Where is the GStage synchronization logic implemented?

The core blocking and notification logic resides in GStage::waiting() inside src/GraphCtrl/GraphStage/GStage.cpp (lines 33–50). The class definition and member variables are declared in src/GraphCtrl/GraphStage/GStage.h (lines 20–33), while management and lookup functionality is handled by GStageManager in src/GraphCtrl/GraphStage/GStageManager.h.

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