A toy application of pub/sub using C++ for self-learning

Question

Firstly, I apologize for the lengthy post. I was not sure which code I should include.

In practice of c++, I am writing a toy application for pub/sub using c++17. To keep things simple, both the publisher and subscriber are on the same OS Process, and thus only a deque is needed to store messages.

I don't use a Singleton, but rather I create a shared_ptr and std::move it across constructors.

I attempt to use c++17 idioms where ever they are known to me; for example in std::scoped_lock vs std::lock_guard, and lambda over std::bind. If there are any other places in the code that can be upgraded, please note it.

One of my main struggles with c++ is the lack of clarity on which techniques are c++11 and which are more modern.

Along these lines, is any aware why the Google C++ style guide prohibits the use of c++20? Especially since c++23 is fast approaching.

The first class acts as the Queue, with call back mechanism.

#pragma once

#include <iostream>
#include <vector>
#include <string>
#include <memory>
#include <mutex>
#include <deque>
#include <thread>
#include <functional>

namespace tareeqav {
    namespace messaging {

        class QueueBase {
        public:
            virtual ~QueueBase() {};
        };

        template<class T>
        class Queue : public QueueBase {
        public:
            Queue() {
                std::thread t = std::thread([this]() {
                    this->RunCallbacks();
                });
                t.detach();
            };

            void EnqueueMessage(std::shared_ptr<T const> message);

            const int length() { return messages_.size(); };

            template<typename CallbackT>
            void RegisterCallback(CallbackT &&callback);

            void RunCallbacks();

        private:
            std::mutex mutex_;
            std::string id_;
            std::deque<std::shared_ptr<T const>> messages_ = {};
            std::vector<std::function<void(std::shared_ptr<T const>)>> callbacks_ = {};
            std::chrono::time_point<std::chrono::system_clock> prev = std::chrono::system_clock::now();

            std::shared_ptr<T const> DequeueMessage();
        };

        template<class T>
        void Queue<T>::EnqueueMessage(std::shared_ptr<T const> message) {
            messages_.push_back(message);
        };

        template<class T>
        std::shared_ptr<T const> Queue<T>::DequeueMessage() {
            if (length() > 0) {
                auto item = messages_.front();
                messages_.pop_front();
                return item;
            }
            return nullptr;
        };

        template<class T>
        template<typename CallbackT>
        void Queue<T>::RegisterCallback(CallbackT &&callback) {
            std::scoped_lock lock(mutex_);
            callbacks_.template emplace_back(std::forward<CallbackT>(callback));
        };

        template<class T>
        void Queue<T>::RunCallbacks() {
            while (true) {
                std::chrono::time_point<std::chrono::system_clock> now = std::chrono::system_clock::now();
                auto milliseconds = std::chrono::duration_cast<std::chrono::milliseconds>(prev - now);
                auto ms = 100 - milliseconds.count();
                while (ms > 0) {
                    --ms;
                }
                if (callbacks_.size() > 0 && messages_.size() > 0) {
                    std::scoped_lock lock(mutex_);
                    std::shared_ptr<T const> message = DequeueMessage();
                    for (auto &&callback: callbacks_) {
                        std::thread t = std::thread([&callback, &message]() {
                            callback(message);
                        });
                        t.detach();
                    }
                }
                prev = std::chrono::system_clock::now();
            }
        }
    } // namespace messaging
} // namespace tareeqav

The Broker class acts as a middle layer between the pub/sub and the stored data.

#pragma once

#include <memory>
#include <string>
#include <vector>
#include <unordered_map>

#include "tareeqav/messaging/queue.h"
#include "tareeqav/messaging/proto/control_message.h"

namespace tareeqav {
    namespace messaging {

        enum class TopicStatus {
            TOPIC_CREATED,
            TOPIC_EXISTS
        };

        enum class MessageStatus {
            QUEUED,
            PENDING,
            PUBLISHED,
            TOPIC_NOT_FOUND
        };

        class Broker {
        public:

            template<typename T>
            TopicStatus CreateTopic(std::string topicName);

            template<typename T>
            MessageStatus PublishMessage(std::string topicName, std::shared_ptr<T const> message);

            template<class T, typename CallbackT>
            void AddSubscription(
                    std::string topicName,
                    CallbackT &&callback);

        private:
            template<typename T>
            std::shared_ptr<Queue<T const>> get(std::string topicName) {
                auto base = queues_[topicName];
                return std::dynamic_pointer_cast<Queue<T const>>(base);
            }

            std::vector<std::thread::id> threads_;
            std::unordered_map<std::string, std::shared_ptr<QueueBase>> queues_;
        };

        template<typename T>
        TopicStatus Broker::CreateTopic(std::string topicName) {
            // check if we already have a topic with that name
            if (queues_.find(topicName) == queues_.end()) {
                queues_[topicName] = std::make_shared<Queue<T const>>();
                return TopicStatus::TOPIC_CREATED;
            }
            return TopicStatus::TOPIC_EXISTS;
        };

        template<typename T>
        MessageStatus Broker::PublishMessage(std::string topicName, std::shared_ptr<T const> message) {
            if (queues_.find(topicName) != queues_.end()) {
                auto queue = get<T>(topicName);
                std::scoped_lock lock(queue->mutex_);
                queue->EnqueueMessage(message);
                return MessageStatus::QUEUED;
            }
            return MessageStatus::TOPIC_NOT_FOUND;
        }

        template<class T, typename CallbackT>
        void Broker::AddSubscription(
                std::string topicName,
                CallbackT &&callback) {
            if (queues_.find(topicName) != queues_.end()) {
                auto queue = get<T>(topicName);
                queue->RegisterCallback(std::forward<CallbackT>(callback));

            }
        }
    } // namespace messaging
} // namespace tareeqav

The Subscriber class is a simple example of the management of adding subscriptions.

#pragma once

#include <iostream>
#include <string>
#include <memory>
#include <utility>

#include "tareeqav/messaging/broker.h"

namespace tareeqav {
    namespace nodes {

        using tareeqav::messaging::Broker;
        using tareeqav::messaging::TopicStatus;
        using tareeqav::messaging::MessageStatus;

        class Subscriber {
        public:
            Subscriber(
                    std::shared_ptr<Broker> broker,
                    std::string topicName);

            template<class T, typename CallbackT>
            void AddSubscription(CallbackT &&callback);

        private:
            const std::string topicName_;
            std::shared_ptr<Broker> broker_;
        };

        Subscriber::Subscriber(
                std::shared_ptr<Broker> broker,
                std::string topicName) :
                topicName_(std::move(topicName)), broker_(std::move(broker)) {
        };

        template<class T, class CallbackT>
        void Subscriber::AddSubscription(CallbackT &&callback) {
            broker_->template AddSubscription<T>(
                    topicName_,
                    std::forward<CallbackT>(callback));
        }
    } // namespace node
} // namespace tareeqav

The Publisher class creates messages.

#pragma once

#include <iostream>
#include <string>
#include <memory>
#include <utility>

#include "tareeqav/messaging/broker.h"

namespace tareeqav {
    namespace nodes {
        using tareeqav::messaging::Broker;
        using tareeqav::messaging::TopicStatus;
        using tareeqav::messaging::MessageStatus;

        template<class T>
        class Publisher {
        public:
            Publisher(
                    std::shared_ptr<Broker> broker,
                    std::string topicName);

            void CreateTopic() {
                TopicStatus status = broker_->template CreateTopic<T>(topicName_);
                switch (status) {
                    case TopicStatus::TOPIC_CREATED:
                        std::cout << "topic created" << std::endl;
                        break;
                    case TopicStatus::TOPIC_EXISTS:
                        std::cout << "topic exists" << std::endl;
                        break;
                }
            }

            MessageStatus PublishMessage(
                    std::string topicName,
                    std::shared_ptr<const T> message);

        private:
            const std::string topicName_;
            std::shared_ptr<Broker> broker_;
        };

        template<class T>
        Publisher<T>::Publisher(
                std::shared_ptr<Broker> broker,
                std::string topicName):
                topicName_(std::move(topicName)), broker_(std::move(broker)) {

        };

        template<class T>
        MessageStatus Publisher<T>::PublishMessage(
                const std::string topicName,
                std::shared_ptr<T const> message) {
            std::cout << "publishing message " << message << std::endl;
            broker_->template PublishMessage(std::move(topicName), std::move(message));
            return MessageStatus::QUEUED;
        };
    } // namespace node
} // namespace tareeqav

The base pipeline class in inherited below

#pragma once

#include <memory>
#include <utility>

#include "tareeqav/messaging/broker.h"

namespace tareeqav {
    using tareeqav::messaging::Broker;

    class Pipeline {
    public:
        Pipeline() = delete;

        Pipeline(std::shared_ptr<Broker> broker) : broker_(std::move(broker)) {};

    protected:
        std::shared_ptr<Broker> broker_ = nullptr;
    };

}

This pipeline is a consumer of the test message. A pipeline can be both a consumer and producer of messsages.

#pragma once

#include <memory>
#include "tareeqav/pipelines/pipeline.h"
#include "tareeqav/messaging/proto/control_message.h"
#include "tareeqav/pipelines/perception/control_subscriber.h"

namespace tareeqav {
    namespace perception {
        using std::placeholders::_1;

        class PerceptionPipeline : public Pipeline {
        public:
            PerceptionPipeline(const std::shared_ptr<Broker> broker) : Pipeline(std::move(broker)) {
                std::string topicName = std::string("/control_message");
                subscriber_ = std::make_shared<
                        nodes::Subscriber>(broker, topicName);

                controlCallback = std::make_shared<ControlSubscriber>();

                subscriber_->AddSubscription<const messaging::ControlMessage>(
                        [this](auto &&T) { controlCallback->MessageCallback(std::forward<decltype(T)>(T)); }
                );

            };
        private:
            std::shared_ptr<nodes::Subscriber> subscriber_ = nullptr;
            std::shared_ptr<ControlSubscriber> controlCallback = nullptr;
        };

    } // namespace perception

} // namespace tareeqav

This pipeline publishes a message.

#pragma once

#include "tareeqav/pipelines/pipeline.h"
#include "tareeqav/node/publisher.h"
#include "tareeqav/messaging/proto/control_message.h"

namespace tareeqav {
    namespace planning {

        class PlanningPipeline : public Pipeline {
        public:
            PlanningPipeline(std::shared_ptr<Broker> broker) : Pipeline(std::move(broker)) {
                std::string topicName = std::string("/control_message");
                controlPublisher_ = std::make_shared<nodes::Publisher<messaging::ControlMessage>>(
                        broker_, topicName);

                controlPublisher_->CreateTopic();

                auto message = std::make_shared<messaging::ControlMessage>(.9, .1);

                controlPublisher_->PublishMessage(topicName, message);
            };
        private:
            std::shared_ptr<nodes::Publisher<messaging::ControlMessage>> controlPublisher_ = nullptr;

        };
    } // namespace perception
} // namespace tareeqav

The control message (which will be replaced with protobuf soon)

#pragma once

namespace tareeqav
{
    namespace messaging
    {
class ControlMessage
{
public:
    ControlMessage(){};
    ControlMessage(float a, float s): acceleration(a), steering(s){};
    float acceleration;
    float steering;
};
    } // namespace perception
} // namespace tareeqav

This is the driver with the entry point.

#include <memory>
#include "tareeqav/messaging/broker.h"
#include "tareeqav/pipelines/planning/planning.h"
#include "tareeqav/pipelines/perception/perception.h"

using tareeqav::messaging::Broker;
using tareeqav::planning::PlanningPipeline;
using tareeqav::perception::PerceptionPipeline;

int main(int argc, char *argv[]) {
    auto broker = std::make_shared<Broker>();
    auto planningPipeline = std::make_shared<PlanningPipeline>(PlanningPipeline(broker));
    auto perceptionPipeline = std::make_shared<PerceptionPipeline>(PerceptionPipeline(broker));

    while (true) {};

    return 0;
}

Answer 1

About code style

I attempt to use c++17 idioms where ever they are known to me;

Don't force yourself to use the latest idioms; while they have their uses, sometimes an older idiom is still better for a given context.

for example in std::scoped_lock vs std::lock_guard,

See this answer for a possible reason not to use std::scoped_lock.

One of my main struggles with C++ is the lack of clarity on which techniques are C++11 and which are more modern.

I think that's the wrong thing to worry about. Later versions of C++ build on top of older versions, and a lot of the "old C++" is still very much exactly what you need. I would make it a habit to keep up with the latest C++ by watching C++ conferences and browsing cppreference.com now and then, and then the knowledge you absorbed will find its way into the code you write. But again, don't force it.

Along these lines, is any aware why the Google C++ style guide prohibits the use of C++20?

Unless you work for Google and write code for them, you are free to ignore it. Also note that it's just a guide, not a set of commandments. In my experience, big companies are usually quite conservative in what they allow, sometimes for good reasons (for example, they need their code to compile on old systems because their customers still use those).

For your own code, consider who is going to use it, and if they are able to compile it. For example, a lot of people are using LTS versions of their favourite operating systems, which means you might want to support compilers that are a few years old.

Remove length()

While it doesn't really matter for a std::deque, make it a habit to use .empty() to check if a container is empty or not instead of using .size(), as for some containers the latter is less efficient.

But more importantly, there should not be a public member function that returns the length of the queue. The issue is that other threads can enqueue and dequeue at any moment, so the return value of length() is meaningless to anything other than another member function of Queue (and even then only if a lock is held). So at the very least make it private(), but it's better to just remove it.

Fragile locking

Apart from the fact that your code doesn't compile because queue->mutex_ is private, it is a bad idea to have public API functions in Queue that depend on the caller to do proper locking. It is easy to forget to do that, and then the code will compile and seem to work, until you do get that race condition in production and your data gets corrupted. I recommend you make all public member functions take locks if they modify member variables.

Use condition variables to wake up the callback handler

Despite the use of std::chrono functions, nothing in RunCallbacks() actually sleeps, and the compiler will optimize away the while(ms > 0) { --ms; }. In any case, sleeping for fixed amounts of time or busy looping is never a good choice: you will either sleep longer than necessary and/or waste CPU time. Instead, use condition variables so the callback handler can sleep until an entry is added to the queue.

Calling callback functions

First of all, the lambda you pass to std::thread() is unnecessary, you can just write:

std::thread t(callback, message);
t.detach();

This also avoids the issue where message is captured by reference, but by the time the thread runs the object it points to might already have gone out of scope.

But consider what happens here: for every message that arrives, you created one or more threads that call the callback functions. Threads are not free; they use quite a bit of memory (each thread needs its own stack at the very least), and the kernel has to do bookkeeping for each thread. Also, there is now no guarantee whatsoever in which order the callbacks are actually executed. Consider that I send two messages: one to start accelerating, and another one to decelerate. The kernel can schedule threads however it likes, what would happen if those messages are handled in the opposite order they were enqueued?

Finally, detaching a thread means you lose control over it. If a thread goes into an infinite loop, it will continue to use resources, but you won't notice it for a while.

I strongly suggest that RunCallbacks() does not start any threads on its own. For callback functions that execute in very little time, this is much more efficient. If a callback function takes a long time, it can create a thread itself if necessary.

Unused member variable threads_ in Broker

I guess that originally, you had the broker create the threads to handle callbacks? But now Queue starts its own thread, so threads_ can be removed.

Think about shutting down a Queue orderly

RunCallbacks() loops indefinitely, making it impossible to destroy a Queue safely. Maybe it's not an issue (yet), but what if you want to add support for removing topics from the broker?

If a tree falls in a forest

If a message is put into a queue and there is no callback registered, should it still be dequeued?

Consider adding messages to a queue with zero callbacks, then registering a callback, and compare that to adding messages to a queue with one callback, and then registering a second one later.

Publishing a message of the wrong type causes a crash

Once you create a topic it has a type associated with it. But later on you can call PublishMessage() with a message of a different type. This causes get<T>(topicName) to return a nulltpr, but you dereference it without checking.

Use of std::shared_ptr

I don't use a Singleton, but rather I create a shared_ptr and std::move it across constructors.

First, in many cases you don't need singletons. Maybe you can't think of a reason why someone would want to create two brokers in the same program, but on the other hand, there is nothing that would prevent that from working correctly.

Second, I don't think you need to use std::shared_ptr at all, just pass non-owning references to objects and make sure their lifetime is correct. For example, I think you should be able to write your main() function like so:

int main() {
    Broker broker;
    PlanningPipeline planningPipeline(broker);
    PerceptionPipeline perceptionPipeline(broker);
    ...
}

As for the messages, consider having Queue own the messages enqueued on it:

template<class T>
class Queue: public QueueBase {
public:
    ...
    template<typename... Args>
    void EnqueueMessage(Args&&... args);
    ...
private:
    ...
    std::queue<T> messages;
};

template<typename... Args>
void Queue::EnqueueMessage(Args&&... args) {
    std::lock_guard lock(mutex_);
    messages.emplace(std::forward<Args>(args)...);
}

This also uses perfect forwarding to avoid having to repeat any types:

Queue<std::pair<float, std::string>> queue;
queue.EnqueueMessage(42, "The answer");