2
\$\begingroup\$

folks. I have recently started writing software using Modern C++ 11-14. I have been developing software for more than a decade and just wanted to broaden my skillset. I am practicing building some simple design components using Modern C++. I dont have any friend or a colleague who knows c++ and noone can review my practice problems. I would be very grateful if you can review a couple of my code snippets and provide your feedback. Thank you.

Below is my recent implementation of the Reactor. Please criticize :) At the core of the reactor lies a thread called main_thread. Reactor will be receiving messages of type struct Message which is defined in Message.hpp file.

Messages will be delivered using virtual method WaitForMessage.

Users should be able to register their concrete event handlers which are derived from the base class IEventHandler. Reactor will call OnMessage of the handler if the received message type matches the type that IEventHandler was registered to.

Inside AbstractReactor the handlers will be wrapped in a class named MessageListener and AbstractReactor will keep MessageListeners inside the vector. Would a map be a better choice?. I decided to use vector therefore MessageListeners can be sorted by the type of the message that they are looking for and we will be able to use binary search(this is what std::lower_bound is used for) rather than looping.

One of the requirements was. A user should be able to call registerHandler and unregisterHandler from within the OnMessage routine of a concrete handler. I am using push_back on every handler which is registered while I am running in the context of main_thread and sort it after the message has been processed. When registerHandler is called outside the main_thread context it will search the appropriate position in the vector where the handler should be inserted and will insert it at that position. If deregisterHandler is called while we are at the main_thread context the listener will not be removed from the vector immediately. Flag m_handlersBeenUnregistered will be set and only after the message is processed we will check which of the listeners have to be removed and will call erase method.

Thank you

File AbstractReactor.cpp

#include <mutex>
#include <algorithm>

#include "AbstractReactor.hpp"
#include "IEventHandler.hpp"


int MessageListener::m_IdCount = 0;


AbstractReactor::AbstractReactor()
{}


AbstractReactor::~AbstractReactor()
{
    if (!m_stopThread)
        stopThread();
}


void AbstractReactor::mainThread()
{
    while(!m_stopThread)
    {
        /* Block until message gets available
         * mainThread now owns a message */
        std::unique_ptr<Message> m_ptr = waitForMessage();
        if (m_ptr.get() == nullptr)
        {
            /* Reactor process may have received a signal to abort */
            /* TODO: this may be reported calling some error handler */
            continue;
        }

        /* Lock the list of listeners, I am using recursive mutex, because
         * we may call registerHandler and unregisterHandler functions while invoking a handler function of the listener */
        std::unique_lock<std::recursive_mutex> lock;

        /* All handler entries are sorted by message type handlers are looking for
         * find the position of the first message listener whose type matches the type of the message. We may have multiple message listeners registered
         * for the same message type */

        m_searchValue.m_type = m_ptr->type;
        m_searchValue.m_handleId = -1;
        auto pos = std::lower_bound(m_listeners.begin(), m_listeners.end(), m_searchValue, [](const MessageListener& one, const MessageListener& two)
        {
            if (one.m_type < two.m_type)
                return true;
            else
                return false;
        }
        );

        if (pos == m_listeners.end())
        {
            /* We couldnt find any message listener which was registered for this message type
             * we will keep listenning for new events
             * We may add some statistics for future references */
            continue;
        }

        /* Set the flag that we are processing a message
         * When this flag is set registerHandler will not try to insert a handler to the proper position, rather it will push_back a handler to the end of the vector.
         * All newly registered handlers will be at the end of the list
         * When reactor finishes calling handlers he will sort its handlers table again.*/
        m_processing = true;
        auto size = m_listeners.size();

        auto i = pos - m_listeners.begin();
        while(i < static_cast<int>(size) && m_listeners[i].m_type == m_ptr->type){
            /* Handlers are user-defined.
             * If listener fails it shouldn't affect our Reactor */
            try
            {
                m_listeners[i].m_hptr->OnMessage(m_ptr.get());
            }
            catch(...)
            {
                /* We may need to report an exception.
                 * Reactor should not have any error handling but it will need to somehow to log this error */
            }
            i++;
        }
        m_processing = false;

        if (m_listeners.size() > size)
        {
            /* If the list has grown while we were invoking handlers, we will need to sort it again and place new handlers
             * at appropriate positions in the vector according to the message type */
            std::sort(m_listeners.begin(), m_listeners.end(), [](const MessageListener& first, const MessageListener& second){
                if (first.m_type <= second.m_type)
                    return true;
                else
                    return false;
            });
        }

        /* If there there was at least one unregisterHandler call while we were processing a message
         * we will need to go through the whole table and remove the ones which have to be unregistered */
        if (m_handlersBeenUnregistered == true)
        {
            for (auto it = m_listeners.begin(); it != m_listeners.end(); ++it)
            {
                if (it->m_mustRemove)
                    it = m_listeners.erase(it);
            }
            m_handlersBeenUnregistered = false;
        }
    }
}


int AbstractReactor::unregisterHandler(int handleId, int32_t type)
{
    if (handleId < 0)
        return -1;

    std::unique_lock<std::recursive_mutex> lock;

    m_searchValue.m_type = type;
    m_searchValue.m_handleId = handleId;
    auto pos = std::lower_bound(m_listeners.begin(), m_listeners.end(), m_searchValue, [](const MessageListener& theirs, const MessageListener& my)
    {
        if (theirs.m_type < my.m_type )
            return true;
        else
            return false;
    }
    );

    if (pos == m_listeners.end())
    {
        /* If we were unable to find a match for this handler in the listeners table
         * we will return negative status to the user */
        return -1;
    }

    auto i = pos - m_listeners.begin();
    while(i < static_cast<int>(m_listeners.size()) && m_listeners[i].m_type == type)
    {
        if (m_listeners[i].m_handleId == handleId)
        {
            if (m_processing == false)
                m_listeners.erase(m_listeners.begin() + i);
            else
                m_listeners[i].m_mustRemove = true;
            break;
        }
        i++;
    }

    /* Set a global flag that will indicate that a handler has been marked to be deleted */
    if (m_processing == true)
        m_handlersBeenUnregistered = true;
    return 0;
}


void AbstractReactor::start()
{
    m_thread = std::thread(&AbstractReactor::mainThread, this);

}

void AbstractReactor::stopThread()
{
    m_stopThread = true;
    m_thread.join();
}

void AbstractReactor::stop()
{
    /* we will just stop processing messages, but we will not delete
     * all message listeners
     * Message listeners entries will be deleted on destruction */
    stopThread();
}

File AbstractReactor.hpp

#pragma once

#include <vector>
#include <mutex>
#include <thread>
#include <memory>
#include <algorithm>

#include "IEventHandler.hpp"
#include "Message.hpp"


struct MessageListener
{
    int32_t              m_type{-1};
    int                  m_handleId{-1};
    bool                 m_mustRemove{false};

    static int           m_IdCount;
    std::unique_ptr<IEventHandler> m_hptr;
public:
    MessageListener() = default;
    MessageListener(int32_t type, std::unique_ptr<IEventHandler> h):
        m_type(type),
        m_handleId(m_IdCount++),
        m_hptr(std::move(h))
    {}
    MessageListener(int32_t type, int handleId):
        m_type(type),
        m_handleId(handleId)
    {}
};


class AbstractReactor
{
public:
    AbstractReactor();
    virtual ~AbstractReactor();

    /* This is an virtual function which must be implemented in the concrete reactor which you
     * derive from the AbstractReactor class. This function will be the source of the messages
     * to the reactor.
     * It will block until an OS informs us that an event occurred and message is available
     * Concrete implementation of Abstract reactor must override it */
    virtual std::unique_ptr<Message> waitForMessage() = 0;

    void start();
    void stop();


    /* Register handler is a templated function which will require
     * message type and parameters used for constructing concrete user handler derived from IEventHandler
     * */
    template<typename HandlerType, typename ...HandlerParametersType>
    int registerHandler(int type, HandlerParametersType&&... handlerParams)
    {
        std::unique_lock<std::recursive_mutex> lock;

        auto pos = m_listeners.end();
        if (m_processing == false)
        {
            /* Add message listeners in sorted order sorting by their message type,
             * so we will be able to use binary search when trying to find listener registered for a specific message type
             * Not sure how many message types there are. If the number if huge then simply iterating over the list of big length
             * with not be an ideal solution */

            m_searchValue.m_type = type;
            m_searchValue.m_handleId = -1;
            pos = std::lower_bound(m_listeners.begin(), m_listeners.end(), m_searchValue, [](const MessageListener& theirs, const MessageListener& my)
            {
                if (theirs.m_type < my.m_type)
                    return true;
                else
                    return false;
            }
            );
        }

        pos = m_listeners.emplace(pos, type, std::move(std::make_unique<HandlerType>(std::forward<HandlerParametersType>(handlerParams)...)));

        if (m_processing == false)
            return pos->m_handleId;
        else
            return m_listeners.back().m_handleId;
    }



    int unregisterHandler(int handleId, int32_t type);

private:
    std::recursive_mutex m_mutex;
    std::vector<MessageListener> m_listeners;
    std::thread m_thread;
    MessageListener m_searchValue;
    bool        m_stopThread{false};
    bool        m_processing{false};
    bool        m_handlersBeenUnregistered{false};
    void stopThread();
    void mainThread();
};

File IEventHandler.hpp

#pragma once

#include "Message.hpp"


class IEventHandler
{
public:
  virtual ~IEventHandler() {};
  virtual void OnMessage(const Message *msg) = 0;
};

File Message.hpp

#pragma once
#include <cstdint>

struct Message
{
    int32_t type;
    char data[32];
};
\$\endgroup\$
3
  • \$\begingroup\$ Does all of that currently work as intended or not? \$\endgroup\$ – πάντα ῥεῖ Jul 7 '20 at 14:07
  • \$\begingroup\$ What is this a reactor "for"? Is it intended to service web requests? Game events? Something else? \$\endgroup\$ – Reinderien Jul 7 '20 at 18:38
  • \$\begingroup\$ Reactor was working I tested it using socket. WaitForMessage function was listenning on a linux domain socket and I had another process sending messages to that socket with random message types. I can write code which is working I would like to know whether it is implemented using modern C++ conventions. I would like to know also whether keeping handlers sorted in a vector was a good choice or it should be done differently. Reactor is intended to listen on a socket and demultiplex messages from socket to corresponding message handlers. \$\endgroup\$ – slaventos Jul 8 '20 at 1:12
2
\$\begingroup\$

No comment on the design, just style improvements.

    auto pos = std::lower_bound(m_listeners.begin(), m_listeners.end(), m_searchValue, [](const MessageListener& one, const MessageListener& two)
    {
        if (one.m_type < two.m_type)
            return true;
        else
            return false;
    }
    );

I find this snippet very hard to read, especially because the lambda's parameter-list runs off the right side of the screen. I would write it with "Python-style" indentation:

    auto pos = std::lower_bound(
        m_listeners.begin(),
        m_listeners.end(),
        m_searchValue,
        [](const auto& a, const auto& b) {
            return (a.m_type < b.m_type);
        }
    );

Notice that if (x) return true; else return false; is a too-verbose way of writing return x;

Also notice that we can use a generic lambda (auto) to shorten the parameter list, assuming that the reader already knows that m_listeners is a list of MessageListener objects so we don't have to explicitly repeat that type's name.


if (m_ptr.get() == nullptr)

Treat smart pointers like normal pointers. Using any named member function on a smart pointer is a code smell. If you want to test a pointer (smart or raw) for null, write simply:

if (m_ptr == nullptr)

typename ...HandlerParametersType — I strongly recommend naming packs something plural. This isn't a type; it's a pack of types. So: class... HandlerParameterTypes, or simply class... Params, or simply class... Ts.


std::move(std::make_unique~~~

The result of a function call expression like std::make_unique<T>(args...) is already a prvalue. You don't have to cast it with std::move. (Remove the call to std::move.)


if (!m_stopThread)
    stopThread();

I strongly recommend using curly braces around the body of every control-flow construct in your program. Consider what happens if you add a logging statement temporarily:

if (!m_stopThread)
    std::cout << "stopping the thread\n";  // Oops!
    stopThread();
\$\endgroup\$
1
  • \$\begingroup\$ Quuxplusone, thank you very much. I really appreciate your time. I completely agree with all that you said above. Will fix it. Everything that you mentioned is very valuable. \$\endgroup\$ – slaventos Jul 10 '20 at 6:01

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.