# CGraph Message Passing Pub/Sub Mechanism Across Pipelines: A Complete Guide > Explore the CGraph message passing pub/sub mechanism. Learn how a singleton message manager facilitates topic-based communication across pipelines for efficient data exchange. - Repository: [Chunel/cgraph](https://github.com/chunelfeng/cgraph) - Tags: deep-dive - Published: 2026-02-27 --- **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`](https://github.com/chunelfeng/cgraph/blob/main/src/GraphCtrl/GraphMessage/GMessageManager.h) lines 39-59) - Stored in `conn_message_map_` to map `connId` to the specific `GMessage` 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`](https://github.com/chunelfeng/cgraph/blob/main/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. ```cpp // 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` 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`](https://github.com/chunelfeng/cgraph/blob/main/GMessageManager.h)). ```cpp 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). ```cpp 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: ```cpp // 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`](https://github.com/chunelfeng/cgraph/blob/main/tutorial/T17-MessagePubSub.cpp) demonstrates a full pub/sub workflow with multiple subscriber pipelines. ```cpp #include "CGraph.h" using namespace CGraph; void publishPipeline() { GPipelinePtr pubPipe = GPipelineFactory::create(); GElementPtr pubNode; // Register publisher node that sends messages pubPipe->registerGElement(&pubNode, {}); pubPipe->process(5); // Run 5 iterations } void subscriberPipeline() { GPipelinePtr subPipe = GPipelineFactory::create(); GElementPtr node; GTemplateNodePtr subNode; // Each subscriber gets a unique connId for the same topic int connId = CGRAPH_BIND_MESSAGE_TOPIC(MyMessageParam, "pub-sub", 1024); subPipe->registerGElement(&node, {}, "myNode2"); subPipe->registerGElement>(&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`](https://github.com/chunelfeng/cgraph/blob/main/tutorial/MyGNode/MyPubMessageNode.h) (lines 24-30) – constructs `MyMessageParam` and calls `CGRAPH_PUB_MPARAM` - **Subscriber node**: [`tutorial/MyGNode/MySubMessageNode.h`](https://github.com/chunelfeng/cgraph/blob/main/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`](https://github.com/chunelfeng/cgraph/blob/main/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` 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.