2
\$\begingroup\$
  1. The reactor possesses its own thread to wait for messages in an event loop.
  2. Users should be able to start or stop the reactor at any point. These two operations are not required to be thread-safe.
  3. Users should be able to asynchronously register or remove message handlers from the reactor.
  4. It is possible that users may register or remove any handler within the message handling routine of a handler.

NOTE: I use arbitrary delays to simulate input blocking. The test code is in main() and not as separate CPP UNIT tests, since this is not a requirement. As far as I know, this code works fine, but I can test it more thoroughly. Async logic is there assuming that the user of the reactor will call std::async(func-name).

My implementation:

#include <iostream>
#include <algorithm>
#include <chrono>
#include <thread>
#include <map>
#include <functional>
#include <mutex>

namespace reactor
{
    struct Message
    {
        int32_t type;
        char data[256];
    };

    // Define a few message types to simulate User event behaviour.
    enum MessageTypes
    {
        OPEN = 0,
        READ,       
        WRITE,
        CLOSE,
        REGISTER_HANDLER,
        REMOVE_HANDLER
    };

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

    class EventHandler : public IEventHandler
    {
    public:
        EventHandler() {};
        EventHandler(const Message *msg) 
        { 
            fresh_msg = msg;
            event_loop_start = true;
            reactor_thread = std::thread(&EventHandler::MessageLoop, this);
        }; // Take in the pointer to a message as an argument to the constructor.

        bool RegisterEventHandler(int user_id, int message_type, std::function<void (void)> func)
        {
            std::lock_guard<std::mutex> lock(map_mutex);

            handler_map[user_id][message_type] = func;
            register_status_map[user_id][message_type] = true;
            return register_status_map[user_id][message_type];
        }

        bool RemoveEventHandler(int user_id, int message_type)
        {
            std::lock_guard<std::mutex> lock(map_mutex);
            
            handler_map[user_id].erase(message_type);
            register_status_map[user_id][message_type] = false;
            return register_status_map[user_id][message_type];
        }

        virtual void OnMessage(const Message* msg) override
        {
            fresh_msg = msg;
        }

        void startEventHandlerThread()
        {
            if (event_loop_start == false)
            {
                event_loop_start = true;
                reactor_thread.join(); //Thread was already started and stopped. Restart it again.
                reactor_thread = std::thread(&EventHandler::MessageLoop, this);
            }
                        
        }

        void stopEventHandlerThread()
        {
            event_loop_start = false;           
        }

        std::function<void(void)> getHandler()
        {
            return register_handler;
        }

        void setHandler(std::function<void(void)> reg_handler)
        {
            register_handler = reg_handler;
        }

        ~EventHandler() 
        {       
            if (reactor_thread.joinable())
            {
                reactor_thread.join();
            }                                   
        };

    private:
        //Don't expose the actual message parsing logic to the end user.
        void MessageLoop()
        {           
            while (event_loop_start)
            {               
                if (fresh_msg)
                {
                    std::chrono::milliseconds wait_for_input((fresh_msg->type + 1) * 100);
                    std::this_thread::sleep_for(wait_for_input); // Simulate waiting for input, based on the message type, by putting the thread to sleep.

                    ParseMessage(fresh_msg);
                }
                else
                {
                    std::cerr << "Invalid incoming message! msg pointer is NULL. " << std::endl;
                }
            }
        }

        void ParseMessage(const Message* msg)
        {
            std::lock_guard<std::mutex> lock(msg_mutex);

            // Assume that user ID is obtained from the first 2 bytes of the message payload.
            int user_id = (msg->data[1] << 8) | msg->data[0];
            std::map<int, std::function<void(void)>> event_map;

            if (msg->type == REGISTER_HANDLER)
            {
                if (this->getHandler())
                {
                    RegisterEventHandler(user_id, msg->type, this->getHandler());
                }
                else
                {
                    std::cerr << "Cannot register Event handler since it is NULL!" << std::endl;
                }
                
            }
            else if (msg->type == REMOVE_HANDLER)
            {
                RemoveEventHandler(user_id, msg->type);
            } 
            else
            {               
                try 
                {                   
                    event_map = handler_map.at(user_id);
                    std::function<void(void)> handler = event_map.at(msg->type);

                    //Call the function registered by a specific user, for a specific message type.
                    handler();

                }
                catch (const std::out_of_range&) 
                {
                    // Further analysis can be done on which key failed, exactly.
                    std::cout << "One of the Keys " << user_id << " or " << msg->type << " not found" << std::endl;
                }
            }                       
        }

        const Message* fresh_msg{ nullptr };
        std::thread reactor_thread;
        std::function<void(void)> register_handler = nullptr;
        std::atomic<bool> event_loop_start{ false };
        std::map<int, std::map<int, std::function<void (void)>>> handler_map; // Create a std::map of std::map, containing user IDs and a map of message types and event handlers, for each user ID. 
                                                                              // User IDs can be set from outside.

        std::map<int, std::map<int, bool>> register_status_map;               // Create a second std::map of std::map, containing user IDs and a map of message types and boolean flags, for each user ID. 
                                                                              // This map indicates whether a handler has completed its task or not, by making the handler set the flag on completion.
    
        
        std::mutex map_mutex;
        std::mutex msg_mutex;
    };
}

void TestRead1()
{
    std::cout << "Reading from user1. " << std::endl;
}

void TestWrite1()
{
    std::cout << "Writing from user1. " << std::endl;
}

void TestOpen1()
{
    std::cout << "Opening a file from user1. " << std::endl;
}

void TestRead2()
{
    std::cout << "Reading from user2. " << std::endl;
}

void TestWrite2()
{
    std::cout << "Writing from user2. " << std::endl;
}

void TestOpen2()
{
    std::cout << "Opening a file from user2. " << std::endl;
}


int main()
{   
    /* Simulating end users of a reactor. */
    reactor::Message* msg_ptr;
    reactor::Message message1 = { reactor::OPEN, {0,1,2,3} };
    msg_ptr = &message1;

    reactor::EventHandler event_handler(msg_ptr);
    event_handler.startEventHandlerThread();

    event_handler.RegisterEventHandler(256, reactor::READ, TestRead1);
    event_handler.RegisterEventHandler(256, reactor::OPEN, TestOpen1);
    event_handler.RegisterEventHandler(256, reactor::WRITE, TestWrite1);
    
    reactor::Message message2 = { reactor::READ, {1,1,2,3} };
    msg_ptr = &message2; 
    event_handler.RegisterEventHandler(257, reactor::READ, TestRead2);
    event_handler.OnMessage(msg_ptr);

    int count = 0;
    while (1)
    {
        count++;
        std::cout << "count: " << count << std::endl;
        event_handler.OnMessage(msg_ptr);

        if (count % 2 == 0)
        {
            message1.type = count % 3;
            msg_ptr = &message1;
            event_handler.RemoveEventHandler(((message1.data[1] << 8) | message1.data[0]), message1.type);
        }
        else
        {
            message2.type = count%7;
            msg_ptr = &message2;
        }
        
        if (count >= 100 && count <= 200)
        {
            event_handler.stopEventHandlerThread();         
        }

        if (count > 200)
        {
            event_handler.startEventHandlerThread();
        }

        if (count > 500)
        {
            event_handler.stopEventHandlerThread();
            break;
        } 

        std::chrono::milliseconds change_interval(20);
        std::this_thread::sleep_for(change_interval);
    }   
}

\$\endgroup\$
4
  • \$\begingroup\$ Welcome to CodeReview@SE. Please give a rationale for tagging both c++11 and c++14 (and leaving out c++). (I'd put the sentences preceding the requirements in a comment.)(Put a new-line after the terminating ``` of a terminating code block - scroll to its end to see why.) \$\endgroup\$ – greybeard Sep 21 '20 at 10:35
  • \$\begingroup\$ @greybeard: Thanks. Added a newline after ``` as suggested. The rationale is to emphasize modern C++ so that readers don't think that I am referring to C++03. I am sure that there is no C++17 or higher in my code, hence C++11 and 14. \$\endgroup\$ – Vijayendar Sridharan Sep 21 '20 at 11:06
  • \$\begingroup\$ Hopefully this is a good question, I did read the link on how to ask questions and saw nothing against this. Requirements for this reactor are mentioned in the question. \$\endgroup\$ – Vijayendar Sridharan Sep 21 '20 at 11:07
  • \$\begingroup\$ @VijayendarSridharan It doesn't make sense to say your code is both C++11 and C++14. If you want to ensure it compiles cleanly with any C++11 compliant compiler, just use only the c++11 tag. \$\endgroup\$ – G. Sliepen Sep 21 '20 at 19:01
2
\$\begingroup\$

Remove class IEventHandler

Unless you plan to have multiple classes that inherit IEventHandler, there is no point in declaring an IEventHandler and having EventHandler inherit from it.

Avoid duplicating functionality

In the constructor that takes a Message *, you duplicate code from OnMessage() and startEventHandlerThread(). It is always best to avoid code duplication, and just have the constructor call the other member functions:

EventHandler(const Message *msg)
{
    OnMessage(msg);
    startEventHandlerThread();
}

This way, if you decide to change the way to add a message or start the thread, you only have to do it in one place.

Where is the message queue?

I see you have std::maps for handlers and users, but there is only a single pointer for a message. The message handling is done in its own thread, so what happens if another thread calls OnMessage() multiple times while the first message is still being parsed by ParseMessage()?

Also, nowhere is fresh_msg being set back to nullptr after the message has been parsed. So it will continue parsing the same message over and over until another one is submitted using OnMessage().

Most importantly, you pass a pointer to a message, but who owns the message? What if I write:

auto *msg_ptr = new reactor::Message;
reactor::EventHandler event_handler(msg_ptr);
delete msg_ptr;

You have to ensure you clearly define ownership of the message. I would either have the reactor store the message by value, or use std::shared_ptr instead of a raw pointer.

I recommend you use std::queue to hold a queue of messages. You also need to ensure it is guarded by a mutex, and use a conditional variable to signal that new messages are added to the queue. Here is an example that has a queue that stores Messages by value:

class EventHandler {
public
    ...
    void AddMessage(const Message &msg) {
        std::lock_guard<std::mutex> lock(msg_mutex);
        msg_queue.push(msg);
        msg_cond.notify_one();        
    }

    void MessageLoop() {
        std::unique_Lock<std::mutex> lock(msg_mutex);

        while(event_loop_start) {
            msg_cond.wait(lock, [&]{return !msg_queue.empty()});
            auto msg = msg_queue.front();
            msg_queue.pop();
            ParseMessage(msg);
        }
    }
    ...
private:
    std::queue<Message> msg_queue;
    std::mutex msg_mutex;
    std::condition_variable msg_cond;
    ...
};

Optimize your maps

You don't need to create a map of maps, you can make the key a std::pair that holds both the user ID and message type:

bool RegisterEventHandler(int user_id, int message_type, std::function<void (void)> func) {
    ...
    handler_map[{user_id, message_type}] = func;
    ...
}

bool RemoveEventHandler(int user_id, int message_type) {
    ...
    handler_map.erase({user_id, message_type});
    ...
}

void ParseMessage(const Message* msg) {
    ...
    // Call the function registered by a specific user, for a specific message type.
    handler_map.at({user_id, msg->type})();
    ...
}

std::map<std::pair<int, int>, std::function<void(void)>> handler_map;

Also, you might want to use a std::unordered_map as it is faster, and you don't need the maps to be ordered.

\$\endgroup\$
2
\$\begingroup\$

Overview

I started the code review but stopped half way through.

This is all wrong.

A reactor should hold a thread pool (or a single thread to do the work). When an event is triggered a thread is released from the pool and simply calls the OnMessage() method of all appropriately registered handlers.

You are confusing the event handler pattern and the reactor pattern and smooshing the all together in one mess.

I have added some Pseudo code at the bottom to show an approximation of how the pattern should be written.

Code Review (Partial).

Personal preference here:
Don't like all upper case identifiers. Shouting. When I see OPEN in the code I am likely to think this is some type of macro. Not going to be a big deal if you don't change in my opinion.

    enum MessageTypes
    {
        OPEN = 0,
        READ,       
        WRITE,
        CLOSE,
        REGISTER_HANDLER,
        REMOVE_HANDLER
    };

Is msg every going to be nullptr?

        virtual void OnMessage(const Message* msg) = 0;

I would suspect a better interface is to pass a reference to the message. This also has a better interface as you are not confusing the user of ownership semantics. If you are transferring ownership you will need to use a smart pointer.


This is a bad idea.

            reactor_thread = std::thread(&EventHandler::MessageLoop, this);

Relatively speaking a thread is a heavy weight object. You should not be creating them just to do a small amount of work. I would expect there to be a thread pool processing these events so that you can re-use the threads.

You should definitely not be creating a thread for each callback item.

If you don't want to create and maintain your own thread pool then use the std::async() method. According to the standard this is backed by its own thread pool and allows you to run async tasks.


 EventHandler(const Message *msg) 

Why is the EventHandler get a pointer to the message in the constructor. This is not available until there is an event that happens. This makes no sense!

The even handler is given the event on OnMessage that is the point of the interface you just defined above.


These don't really belong to the event handler.

        bool RegisterEventHandler(int user_id, int message_type, std::function<void (void)> func)
        bool RemoveEventHandler(int user_id, int message_type)

These are part of the reactor. But the reactor is a namespace!!!! The reactor should be a class in its own right.


What.

        virtual void OnMessage(const Message* msg) override
        {
            fresh_msg = msg;
        }

This is supposed to be the handler. This is where you do the work when the reactor calls to do the work.


Thread handling should not be done in the Event Handler.

        void startEventHandlerThread()
        void stopEventHandlerThread()

This the job of the reactor!


You should not be using wait().

                   std::chrono::milliseconds wait_for_input((fresh_msg->type + 1) * 100);

This is known as a busy wait. Your thread is continuously waking up and checking its state. Most of the time to do nothing. The thread should be asleep (inactive). It only wakes up when there is work to be done.

                    std::this_thread::sleep_for(wait_for_input);

You should be using std::condition_variable. This allows you to sleep the thread so it does not use any resources until there is work for it to do. Then you can wake it up with a call to notify().

How to do it:

Re-Using a thread queue from here: A simple Thread Pool

namespace ThorsAnvil::Reactor 
{

enum MessageTypes
{
    OPEN = 0,
    READ,       
    WRITE,
    CLOSE,
    REGISTER_HANDLER,
    REMOVE_HANDLER
};

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

class Reactor
{
    // In more modern C++ I would not even use this.
    // Simply leave this out and use `std::async below.
    SimpleWorkQueue   threadPool;

    std::map<MessageTypes, std::list<std::unique_ptr<IEventHandler>>>  handler;

    public:
        addHandler(MessageTypes type, std::unique_ptr<IEventHandler> handler)
        {
            // Lock Here.
            handler[type].emplace_back(std::move(handler));
        }
        addEvent(MessageTypes type, Message const& event)
        {
            // Lock Here.
            auto& listHandlers = handler[type];
            for (auto const& handler: listHandlers) {
                // If you are not using the thread pool
                // replace this with std::async.
                threadPool.addWorkItem([&event, &handler]()
                {
                    handler->OnMessage(event);
                });
            }
        }
};
} // end of namespace
\$\endgroup\$
1
  • \$\begingroup\$ Knew there was something messed up. Thanks. I read the Wiki entry and the 2nd question on reactor on this website as well as Vanderbilt’s paper. But couldn’t understand the concept fully. Also, I have no clue as to how to implement a blocking call like select() on Windows, hence the simulated wait in the thread. \$\endgroup\$ – Vijayendar Sridharan Sep 21 '20 at 22:19

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.