4
\$\begingroup\$

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;
}
\$\endgroup\$
3
  • 1
    \$\begingroup\$ The Pipeline class appears to be missing. (If you’re not sure which code to include, a good rule of thumb is to include everything that reviewers will need to compile your test program… because that’s what a lot of them are going to want to do. If there’s far too much for a post, you could always link to the project, but in this case that shouldn’t be a problem.) \$\endgroup\$
    – indi
    Aug 2 at 2:18
  • \$\begingroup\$ I see! Thanks @indi I will post it \$\endgroup\$ Aug 2 at 11:45
  • \$\begingroup\$ @indi updated the post with the Pipeline class and the ControlMessage class \$\endgroup\$ Aug 2 at 16:54

1 Answer 1

2
\$\begingroup\$

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");
\$\endgroup\$
7
  • \$\begingroup\$ Many thanks for a detailed and very informative answer! I did move the mutex_ into private just for the posting as I was thinking about it while posting. But in the code that ran, it's public -- a) Is it better to have the Broker do the lock? or pass it into the Queue and let it manage its own locking? b) Regrading the Callbacks: in all honesty I was just glad I finally got it to compile. The original idea was messages would be published on a frequency, like sensor messages for example that are published on a 100Hz frequency. I thought I would mimic this behavior as part of the callback \$\endgroup\$ Aug 3 at 1:55
  • \$\begingroup\$ c) Your point about shared_ptr makes very good sense! There is really no need for it in this particular case. d) I was intending to add a check to ensure that no topic can exist for two different types. This idea was already on my mind but fell through between all the compiler errors. e) Regarding the Queue owning the messages: I am not sure I understand this yet, I'll have to read up on perfect forwarding ---------- Thanks again for your help! \$\endgroup\$ Aug 3 at 1:58
  • \$\begingroup\$ a) It's better for Queue to manage its own locks so it becomes inherently thread-safe. You might also need a mutex for Broker, since if multiple threads call CreateTopic(), that could corrupt queues_, and it will also confuse threads calling PublishMessage() at the same time topics are added. e) The queue owning messages does not require perfect forwarding, the latter was just to show how it creates a nice to use function. You could also make it EnqueueMessage(T&& message). \$\endgroup\$
    – G. Sliepen
    Aug 3 at 9:08
  • \$\begingroup\$ Regrading passing the Broker by non-owning reference: if I store the reference of Broker& broker as a class member in the Pipeline classes, won't that cause an unnecessary copy? If I try to use an std::move that would mean I need to use a const Broker&, correct? But that won't work because I need to modify the broker's queue. \$\endgroup\$ Aug 3 at 22:46
  • 1
    \$\begingroup\$ If you store a reference, no copies will be made of the Broker object. You also don't need to std::move references. It indeed can't be a const reference. So the constructor of Pipeline should just look like Pipeline(Broker& broker): broker_(broker) {}. \$\endgroup\$
    – G. Sliepen
    Aug 4 at 9:21

Your Answer

By clicking “Post Your Answer”, you agree to our terms of service, privacy policy and cookie policy

Not the answer you're looking for? Browse other questions tagged or ask your own question.