CGraph Message Passing Pub/Sub Mechanism Across Pipelines: A Complete Guide

CGraph implements topic-based publish/subscribe message passing across pipelines through a singleton message manager that binds topics to unique connection IDs, fans out published messages to all bound subscribers, and retrieves messages via those connection IDs.

The chunelfeng/cgraph repository provides a sophisticated message passing pub/sub mechanism that enables loose coupling between distinct processing pipelines. This system allows any number of publisher pipelines to broadcast typed messages to multiple subscriber pipelines without direct references, using a centralized topic manager and unique connection identifiers.

Core Architecture of CGraph Pub/Sub

The pub/sub system rests on three foundational components implemented in the GraphMessage module.

Topic Management

A topic is a logical string identifier that groups publishers and subscribers. The GMessageManager class maintains separate storage for pub/sub and point-to-point communication:

• pub_sub_message_map_ stores sets of message queues keyed by topic name (prefixed with internal::PUB_SUB_PREFIX)
• send_recv_message_map_ handles direct send/receive patterns (prefixed with internal::SEND_RECV_PREFIX)

Connection ID (connId)

When a subscriber binds to a topic, the system returns a unique integer connection ID (connId). This identifier serves as a private handle to a specific message queue instance:

• Generated monotonically in GMessageManager::bindTopic (see src/GraphCtrl/GraphMessage/GMessageManager.h lines 39-59)
• Stored in conn_message_map_ to map connId to the specific GMessage<TImpl> queue
• Enables multiple subscribers to the same topic to receive independent message streams

Singleton Manager

The GMessageManagerSingleton class (defined in src/GraphCtrl/GraphMessage/GMessageManagerSingleton.h) provides global access to the message manager through the USingleton template. This ensures all pipelines interact with the same topic registry and message queues.

Binding Topics and Creating Subscribers

Before receiving messages, a pipeline must bind to a topic using the CGRAPH_BIND_MESSAGE_TOPIC macro. This macro expands to GMessageManager::bindTopic and handles type safety through templates.

// Bind to topic "sensor-data" with queue size 1024
// Returns unique connId for this subscriber
int connId = CGRAPH_BIND_MESSAGE_TOPIC(SensorMessage, "sensor-data", 1024);

The binding process performs three critical operations:

1. Type validation: Ensures the template parameter derives from GMessageParam
2. Queue creation: Instantiates a GMessage<SensorMessage> with the specified capacity
3. Registration: Inserts the queue into both pub_sub_message_map_ (under the topic key) and conn_message_map_ (under the new connId)

Publishing Messages Across Pipelines

Publishers use the CGRAPH_PUB_MPARAM macro to broadcast messages to all bound subscribers. This macro invokes GMessageManager::pubTopicValue (lines 70-86 in GMessageManager.h).

SensorMessage msg;
msg.timestamp = getCurrentTime();
msg.value = sensor.read();

// Publish to all subscribers of "sensor-data"
CStatus status = CGRAPH_PUB_MPARAM(SensorMessage, "sensor-data", msg, 
                                    CGraph::GMessagePushStrategy::WAIT);

The publishing mechanism implements a fan-out pattern:

1. Topic lookup: Retrieves the set of all GMessage queues associated with the topic from pub_sub_message_map_
2. Iteration: Loops through every bound queue (each representing a different subscriber's connId)
3. Delivery: Calls msg->send(value, strategy) on each queue, pushing the message into each subscriber's independent buffer

This ensures that multiple subscriber pipelines receive identical message copies without interfering with each other's consumption rates.

Subscribing and Receiving Messages

Subscribers retrieve messages using the CGRAPH_SUB_MPARAM macro, which maps to GMessageManager::subTopicValue (lines 98-106).

SensorMessage receivedMsg;

// Receive from the specific queue identified by connId
CStatus status = CGRAPH_SUB_MPARAM(SensorMessage, connId, receivedMsg);

The subscription process:

1. Connection lookup: Uses the connId to retrieve the specific GMessage queue from conn_message_map_
2. Blocking receive: Calls message->recv(value, timeout) to pop the next available message from that subscriber's private queue
3. Type safety: Template specialization ensures the received message matches the bound type

Each subscriber maintains independent consumption state through its unique connId, allowing different pipelines to process messages at different speeds without message loss.

Thread Safety and Resource Management

The GMessageManager implements comprehensive thread safety for concurrent pipeline access.

Synchronization Primitives

• pub_sub_mutex_: Protects pub_sub_message_map_ during bind and publish operations
• send_recv_mutex_: Protects send_recv_message_map_ for point-to-point communication
• Queue-level locking: Individual GMessage instances use internal mutexes for thread-safe send and recv operations

All public methods (bindTopic, pubTopicValue, subTopicValue, createTopic, removeTopic) acquire the appropriate mutex before modifying shared state, ensuring safe concurrent usage across multiple threads and pipelines.

Cleanup and Lifecycle

When pipelines shut down, call CGRAPH_CLEAR_MESSAGES() to release all resources:

// Release all topics, queues, and connection mappings
CGRAPH_CLEAR_MESSAGES();

This macro expands to GMessageManagerSingleton::get()->clear(), which:

1. Destroys every GMessage object in both maps
2. Clears pub_sub_message_map_, send_recv_message_map_, and conn_message_map_
3. Resets the monotonic connId counter to zero

Complete Working Example

The tutorial file tutorial/T17-MessagePubSub.cpp demonstrates a full pub/sub workflow with multiple subscriber pipelines.

#include "CGraph.h"

using namespace CGraph;

void publishPipeline() {
    GPipelinePtr pubPipe = GPipelineFactory::create();
    GElementPtr pubNode;
    
    // Register publisher node that sends messages
    pubPipe->registerGElement<MyPubMessageNode>(&pubNode, {});
    pubPipe->process(5);  // Run 5 iterations
}

void subscriberPipeline() {
    GPipelinePtr subPipe = GPipelineFactory::create();
    GElementPtr node;
    GTemplateNodePtr<int> subNode;
    
    // Each subscriber gets a unique connId for the same topic
    int connId = CGRAPH_BIND_MESSAGE_TOPIC(MyMessageParam, "pub-sub", 1024);
    
    subPipe->registerGElement<MyNode2>(&node, {}, "myNode2");
    subPipe->registerGElement<MySubMessageNode<int>>(&subNode, {node}, connId);
    subNode->setName("MySubMessageNode");
    
    subPipe->process(5);
}

int main() {
    std::thread pub(publishPipeline);
    std::thread sub1(subscriberPipeline);
    std::thread sub2(subscriberPipeline);
    std::thread sub3(subscriberPipeline);
    
    pub.join();
    sub1.join();
    sub2.join();
    sub3.join();
    
    CGRAPH_CLEAR_MESSAGES();
    GPipelineFactory::clear();
    return 0;
}

Key implementation files referenced:

• Publisher node: tutorial/MyGNode/MyPubMessageNode.h (lines 24-30) – constructs MyMessageParam and calls CGRAPH_PUB_MPARAM
• Subscriber node: tutorial/MyGNode/MySubMessageNode.h (lines 28-31) – receives message via connId using CGRAPH_SUB_MPARAM

Summary

CGraph's message passing pub/sub mechanism across pipelines provides a robust, decoupled communication layer through these key design elements:

• Topic-based addressing using string identifiers managed by GMessageManager in src/GraphCtrl/GraphMessage/GMessageManager.h
• Unique connection IDs (connId) generated by CGRAPH_BIND_MESSAGE_TOPIC to isolate subscriber queues
• Fan-out publishing via CGRAPH_PUB_MPARAM that delivers copies to all bound subscribers concurrently
• Thread-safe singleton architecture using GMessageManagerSingleton with mutex-protected maps (pub_sub_mutex_, send_recv_mutex_)
• Type-safe message parameters requiring all messages to derive from GMessageParam

This architecture allows independent pipelines to exchange typed data without shared memory or direct references, scaling from single-threaded applications to complex multi-pipeline workflows.

Frequently Asked Questions

How does CGraph ensure that multiple subscribers receive the same message independently?

When a publisher calls CGRAPH_PUB_MPARAM, the underlying GMessageManager::pubTopicValue retrieves the set of all queues bound to that topic from pub_sub_message_map_. It then iterates through every subscriber queue and calls send() on each, pushing a copy of the message into each subscriber's independent buffer. Because each subscriber maintains its own connId-mapped queue in conn_message_map_, consumption speed and timing do not affect other subscribers.

What is the difference between CGRAPH_CREATE_MESSAGE_TOPIC and CGRAPH_BIND_MESSAGE_TOPIC?

CGRAPH_CREATE_MESSAGE_TOPIC establishes a point-to-point (send/receive) communication channel, storing the queue in send_recv_message_map_ with the SEND_RECV_PREFIX. In contrast, CGRAPH_BIND_MESSAGE_TOPIC creates a publish/subscribe relationship, inserting the queue into both pub_sub_message_map_ (for fan-out delivery) and conn_message_map_ (for unique subscriber identification). Only CGRAPH_BIND_MESSAGE_TOPIC returns a connId required for receiving messages.

How does CGraph handle thread safety when pipelines publish and subscribe concurrently?

The GMessageManager class protects shared state with two primary mutexes: pub_sub_mutex_ guards pub_sub_message_map_ during topic binding and publishing operations, while send_recv_mutex_ protects send_recv_message_map_. Additionally, individual GMessage queue instances implement internal locking for their send and recv methods. This multi-layered locking ensures that publishers can fan out messages to multiple subscribers, and subscribers can pull from their private queues, all without data races or corruption.

Can different message types be used on the same topic in CGraph?

No, CGraph enforces type safety through templates. When you bind a topic with CGRAPH_BIND_MESSAGE_TOPIC(MyMessageParam, "topic-name", size), the manager creates a GMessage<MyMessageParam> queue. If a publisher attempts to send a different message type to the same topic using CGRAPH_PUB_MPARAM, the template system will catch the type mismatch at compile time. This design prevents type confusion and ensures that all subscribers receive the expected message structure.

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