Implementing Event-Driven Workflows Using GEvent in CGraph

CGraph's GEvent system enables nodes to trigger custom logic asynchronously or synchronously via string keys, using GPipeline::addGEvent for registration and GNode::notify for dispatch.

Implementing event-driven workflows using GEvent in CGraph allows pipeline nodes to decouple side effects from core execution logic. The chunelfeng/cgraph repository provides a lightweight, type-safe event subsystem centered around the abstract GEvent base class. This architecture supports both synchronous blocking and asynchronous non-blocking execution modes, with automatic lifecycle management through the GEventManager.

Understanding the GEvent Architecture

The event system is built on three core layers: the abstract event interface, the execution policy definitions, and the manager responsible for dispatch and cleanup.

Core Components

Component	Role	Source Location
GEvent	Abstract base class defining the trigger(param) interface and async handling logic.	src/GraphCtrl/GraphEvent/GEvent.h
GEventDefine.h	Defines execution mode enums (GEventType::SYNC/ASYNC) and async wait strategies (GEventAsyncStrategy).	src/GraphCtrl/GraphEvent/GEventDefine.h
GEventManager	Maintains the events_map_, resolves keys to instances, and dispatches via process or asyncProcess.	src/GraphCtrl/GraphEvent/GEventManager.h
GEventObject	Base class providing setThreadPool hooks for all graph objects.	src/GraphCtrl/GraphEvent/GEventObject.h

Execution Modes

CGraph events support two distinct execution models defined in GEventDefine.h:

• GEventType::SYNC: The event executes asynchronously relative to the node, but the manager tracks its completion. The node continues immediately after calling notify.
• GEventType::ASYNC: The event launches in a separate thread via std::async, and the manager stores the returned std::future in async_run_finish_futures_ for later synchronization.

Cleanup occurs when the pipeline finishes (PIPELINE_RUN_FINISH) or is destroyed (PIPELINE_DESTROY), at which point the GEventManager waits on all stored futures according to the selected GEventAsyncStrategy.

Registering and Triggering Events

Implementing a custom event requires subclassing GEvent, registering it with the pipeline, and triggering it from node logic.

Step 1: Create a Custom Event Class

Define a class inheriting from CGraph::GEvent and override the trigger method. This is the only required implementation.

// tutorial/MyGEvent/MyPrintEvent.h
#ifndef CGRAPH_MYPRINTEVENT_H
#define CGRAPH_MYPRINTEVENT_H

#include "CGraph.h"
#include "../MyParams/MyParam.h"

class MyPrintEvent : public CGraph::GEvent {
public:
    // Executed each time the event fires
    CVoid trigger(CGraph::GEventParamPtr) override {
        CGRAPH_SLEEP_MILLISECOND(100);  // Simulate work
        auto myParam = CGRAPH_GET_GPARAM_WITH_NO_EMPTY(MyParam, "param1");
        CGraph::CGRAPH_ECHO("----> trigger [%d] times, iValue = [%d]",
                            times_++, myParam->iValue);
    }

private:
    int times_ = 0;  // Local state persisted across triggers
};

#endif // CGRAPH_MYPRINTEVENT_H

Key implementation details:

• Use CGRAPH_GET_GPARAM_WITH_NO_EMPTY to safely access pipeline-wide parameters.
• Local member variables maintain state across multiple trigger invocations.

Step 2: Register the Event with the Pipeline

Use the template method GPipeline::addGEvent to bind your class to a string key. This is typically done before process() is called.

// In tutorial/T18-Event.cpp
GPipelinePtr pipeline = GPipelineFactory::create();

// Register the custom event with key "my-print-event"
pipeline->addGEvent<MyPrintEvent>("my-print-event");

The addGEvent method (defined in src/GraphCtrl/GraphPipeline/GPipeline.h at lines 22-26) instantiates the event and stores it in the GEventManager's internal events_map_.

Step 3: Trigger Events from Nodes

Any node can fire the event using the notify method inherited from GElement. Specify the event key and execution type.

// tutorial/MyGNode/MyEventNode.h
class MyEventNode : public CGraph::GNode {
public:
    CStatus run() override {
        CGraph::CGRAPH_ECHO("[%s], before event notify", getName().c_str());

        // Fire asynchronously; node continues immediately
        notify("my-print-event", GEventType::SYNC);

        CGraph::CGRAPH_ECHO("[%s], after event notify", getName().c_str());
        return CStatus();
    }
};

The notify declaration resides in src/GraphCtrl/GraphElement/GElement.h at line 320. When called, the GEventManager looks up the key in events_map_ and dispatches to either process (immediate execution) or asyncProcess (deferred execution).

Complete Event-Driven Pipeline Example

The tutorial file T18-Event.cpp demonstrates a full pipeline where multiple nodes share and trigger the same event.

// tutorial/T18-Event.cpp
#include "MyGNode/MyNode1.h"
#include "MyGNode/MyWriteParamNode.h"
#include "MyGNode/MyEventNode.h"
#include "MyGEvent/MyPrintEvent.h"

using namespace CGraph;

void tutorial_event() {
    GPipelinePtr pipeline = GPipelineFactory::create();
    GElementPtr a, b, c, d = nullptr;

    // Register nodes: write param -> event node -> downstream nodes
    pipeline->registerGElement<MyWriteParamNode>(&a, {}, "nodeA");
    pipeline->registerGElement<MyEventNode>(&b, {a}, "nodeB");
    pipeline->registerGElement<MyNode1>(&c, {b}, "nodeC");
    pipeline->registerGElement<MyEventNode>(&d, {b}, "nodeD");

    // Register the event with key "my-print-event"
    pipeline->addGEvent<MyPrintEvent>("my-print-event");

    // Execute pipeline
    pipeline->process();

    GPipelineFactory::clear();
}

Sample Output:

[my-event-node], before event notify
[my-event-node], after event notify
[my-event-node], before event notify
[my-event-node], after event notify
----> trigger [0] times, iValue = [42]
----> trigger [1] times, iValue = [42]

Notice that the trigger messages appear after the "after event notify" lines, confirming the asynchronous execution. The interleaving order may vary depending on thread scheduling. If strict ordering is required, use GEventType::ASYNC (which blocks the node until completion) or implement synchronization primitives within the event logic.

Managing Asynchronous Event Lifecycle

When events execute asynchronously, the GEventManager stores each returned std::future in a container named async_run_finish_futures_. According to the implementation in src/GraphCtrl/GraphEvent/GEventManager.h, cleanup occurs during two specific pipeline states:

1. PIPELINE_RUN_FINISH: After a single run completes, the manager waits on all pending futures according to the selected GEventAsyncStrategy.
2. PIPELINE_DESTROY: When the pipeline object is destroyed, ensuring no dangling threads persist.

The GEventAsyncStrategy enum (defined in GEventDefine.h) controls how the manager handles these futures—whether it waits indefinitely, times out, or detaches. This guarantees that asynchronous events never outlive the pipeline that spawned them, preventing use-after-free errors and resource leaks.

Summary

• GEvent provides a type-safe, abstract base class for defining custom events in src/GraphCtrl/GraphEvent/GEvent.h.
• Register events using GPipeline::addGEvent<T>("key") and trigger them from any node via notify("key", GEventType::SYNC).
• Choose GEventType::SYNC for non-blocking asynchronous execution or GEventType::ASYNC for blocking synchronous execution relative to the node.
• The GEventManager handles lifecycle cleanup automatically during pipeline finish or destruction, ensuring thread safety via async_run_finish_futures_ and GEventAsyncStrategy.

Frequently Asked Questions

What is the difference between SYNC and ASYNC event execution in CGraph?

SYNC execution launches the event logic in a separate thread via std::async and returns control to the calling node immediately, allowing the pipeline to continue while the event runs in parallel. ASYNC execution blocks the node until the event's trigger method completes, effectively making the event call synchronous relative to the node's execution flow. Choose SYNC for fire-and-forget side effects and ASYNC when downstream logic depends on the event's completion.

How do I pass data between nodes and events in CGraph?

Events access pipeline-wide data through the GParam system using the CGRAPH_GET_GPARAM_WITH_NO_EMPTY macro inside the trigger method. Nodes can write parameters before triggering events, as demonstrated in the tutorial where MyWriteParamNode sets values that MyPrintEvent subsequently reads. This shared parameter space eliminates the need for manual thread synchronization when passing read-only or lock-protected data.

Can multiple nodes trigger the same GEvent instance?

Yes. When you register an event with addGEvent<MyEvent>("key"), the GEventManager stores a single instance in events_map_ associated with that key. Any node in the pipeline can call notify("key", ...) to trigger the same instance. This design is ideal for shared resources like logging, metrics collection, or stateful counters where multiple producers need to invoke common logic, as shown in the T18 tutorial where both nodeB and nodeD trigger my-print-event.

How does CGraph ensure asynchronous events complete before pipeline destruction?

The GEventManager maintains a container called async_run_finish_futures_ that stores std::future objects returned by std::async calls. During the PIPELINE_RUN_FINISH or PIPELINE_DESTROY phases, the manager iterates through these futures and waits for completion according to the configured GEventAsyncStrategy. This blocking cleanup ensures that no event threads outlive the pipeline object, preventing segmentation faults or data corruption from accessing destroyed resources.

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