How to Implement Barrier Synchronization Using GFence in CGraph

Use the GFence adapter class to block downstream execution until all registered asynchronous elements complete their work, registering the barrier via registerGElement<GFence> and adding elements to monitor via waitGElement or waitGElements.

In directed-acyclic graph (DAG) pipelines built with the chunelfeng/cgraph framework, barrier synchronization ensures that specific asynchronous operations finish before subsequent nodes begin execution. The GFence class (commonly referred to as Fence) provides this capability by acting as a synchronization point that aggregates async results from multiple elements.

What is Barrier Synchronization in CGraph?

Barrier synchronization in CGraph forces the execution engine to pause further processing until a selected set of asynchronous elements report completion. Unlike standard dependencies that only ensure start-order sequencing, a barrier actively waits for the async result of each registered element.

The GFence adapter, defined in src/GraphCtrl/GraphElement/GAdapter/GFence/GFence.h, derives from GAdapter and maintains a std::set<GElementPtr> called fence_elements_. This collection stores pointers to elements that must finish before the fence allows downstream execution to proceed.

How GFence Works Internally

The barrier implementation relies on three core mechanisms: element registration, pipeline validation, and runtime blocking.

Registering Asynchronous Elements

You register elements for the fence to monitor using waitGElement(GElementPtr element) or waitGElements(const std::set<GElementPtr>& elements). According to the source in src/GraphCtrl/GraphElement/GAdapter/GFence/GFence.cpp (lines 19-36), these methods validate that each element is asynchronous via element->isAsync() before adding it to fence_elements_.

Pipeline Validation

During the checkSuitable() phase (lines 45-52 of GFence.cpp), the framework verifies that every element in fence_elements_ still maintains async status. If any registered element lacks a timeout (making it synchronous), the validation returns an error status, preventing pipeline execution.

Runtime Blocking Behavior

When the pipeline schedules the fence node, it executes the run() method (lines 64-71 of GFence.cpp). This method iterates over fence_elements_ and calls element->getAsyncResult() for each registered element. This call blocks until the asynchronous operation completes, aggregating any error statuses. Only after collecting results from all monitored elements does the fence finish, releasing downstream dependencies.

Implementing a Barrier: Step-by-Step Workflow

To create an effective synchronization barrier in your CGraph pipeline:

1. Create the pipeline using GPipelineFactory::create() or instantiate GPipeline directly.

2. Register asynchronous nodes by setting a timeout value with setTimeout(milliseconds, GElementTimeoutStrategy). This transforms standard nodes into async elements that the fence can monitor.

3. Register the GFence using registerGElement<GFence>, listing the async nodes as its immediate dependencies.

4. Add downstream dependencies that rely on the fence node to ensure they execute only after the barrier clears.

5. Optionally refine the wait list by calling waitGElements() on the fence pointer after registration to add specific elements programmatically.

6. Execute via pipeline->process().

The pipeline guarantees that the fence node schedules after its dependencies start, but blocks until their async results arrive.

Code Examples

C++ Barrier Implementation

The tutorial file tutorial/T24-Fence.cpp demonstrates a complete barrier setup with two async nodes:

#include "MyGNode/MyNode1.h"
#include "MyGNode/MyNode2.h"

using namespace CGraph;

void tutorial_fence() {
    CStatus status;
    GPipelinePtr pipeline = GPipelineFactory::create();

    // Declare elements
    GFunctionPtr a_func = nullptr;
    GElementPtr b = nullptr, c = nullptr, e = nullptr;
    GFencePtr d_fence = nullptr;

    // Register pipeline structure
    status += pipeline->registerGElement<GFunction>(&a_func, {}, "funcA");
    status += pipeline->registerGElement<MyNode2>(&b, {a_func}, "nodeB");
    status += pipeline->registerGElement<MyNode1>(&c, {a_func}, "nodeC");
    status += pipeline->registerGElement<GFence>(&d_fence, {b, c}, "fenceD");
    status += pipeline->registerGElement<MyNode1>(&e, {d_fence}, "nodeE");

    // Convert nodes to async by setting timeout
    a_func->setTimeout(200, GElementTimeoutStrategy::HOLD_BY_PIPELINE);
    b->setTimeout(200, GElementTimeoutStrategy::HOLD_BY_PIPELINE);

    // Explicitly add elements to the fence (optional if already in deps)
    d_fence->waitGElements({a_func, b});

    // Execute
    status = pipeline->process();
    GPipelineFactory::remove(pipeline);
}

Key implementation details:

• registerGElement<GFence>(&d_fence, {b, c}, "fenceD") declares the barrier and its immediate dependencies.
• setTimeout(200, ...) makes nodes asynchronous, enabling them to be added to the fence.
• waitGElements() provides explicit control over which async elements the fence monitors.

Python Barrier Implementation

Using the PyCGraph bindings from python/tutorial/T24-Fence.py:

from pycgraph import GPipeline, GFence, GElementTimeoutStrategy
from MyGNode.MyNode1 import MyNode1
from MyGNode.MyNode2 import MyNode2

def tutorial_fence():
    pipeline = GPipeline()
    a, b, c, e = MyNode2(), MyNode1(), MyNode1(), MyNode1()
    d_fence = GFence()

    # Register graph structure

    pipeline.registerGElement(a, set(), "nodeA")
    pipeline.registerGElement(b, {a}, "nodeB")
    pipeline.registerGElement(c, {a}, "nodeC")
    pipeline.registerGElement(d_fence, {b, c}, "fenceD")
    pipeline.registerGElement(e, {d_fence}, "nodeE")

    # Configure async behavior

    a.setTimeout(200, GElementTimeoutStrategy.HOLD_BY_PIPELINE)
    c.setTimeout(300, GElementTimeoutStrategy.HOLD_BY_PIPELINE)

    # Define barrier wait list

    d_fence.waitGElements({a, c})

    pipeline.process()

if __name__ == '__main__':
    tutorial_fence()

Key differences from C++:

• The waitGElements method accepts a Python set of element references.
• Timeout strategies use the same enum values available in the C++ API.

Summary

• GFence acts as a barrier synchronization adapter that blocks until registered async elements complete, as implemented in src/GraphCtrl/GraphElement/GAdapter/GFence/GFence.cpp.
• Only async elements (those with setTimeout configured) can be added to a fence; the checkSuitable() method enforces this validation.
• Registration occurs via registerGElement<GFence> with dependencies specified, while waitGElements() offers post-registration flexibility.
• Runtime blocking happens in the run() method through getAsyncResult() calls that wait for each async operation to finish.
• Downstream dependencies must explicitly list the fence node in their dependency set to ensure proper execution order.

Frequently Asked Questions

Can GFence monitor synchronous elements?

No. The checkSuitable() method in src/GraphCtrl/GraphElement/GAdapter/GFence/GFence.cpp explicitly validates that every element in fence_elements_ returns true for isAsync(). Attempting to add a synchronous element (one without a timeout) triggers a runtime error during pipeline initialization.

What happens if an async element fails inside a GFence barrier?

The fence aggregates error statuses during its run() method. If any monitored element returns an error through getAsyncResult(), the fence captures this status and propagates it downstream, typically causing the pipeline to fail according to the error handling strategy configured in GElement.h.

Is there a performance penalty for using GFence?

The barrier introduces blocking overhead proportional to the slowest monitored async element. Since run() sequentially calls getAsyncResult() for each element in fence_elements_, the fence cannot complete until all registered async operations finish. However, this overhead is necessary for correctness when subsequent nodes require data from multiple concurrent sources.

How does GFence differ from standard GElement dependencies?

Standard dependencies in GPipeline::registerGElement only ensure start-order sequencing—a dependent node starts after its parents start. GFence ensures completion synchronization—it actively blocks until the asynchronous results of its monitored elements are available, making it essential for coordinating async operations with timeouts.

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