3
\$\begingroup\$

Use case

This is an event loop and signaling system I created for a piece of software which will have multiple asynchronous server/clients/event-emitters/ui, some of those components will have their own event loop running on a separate thread, some will just post events in the main thread event loop.

I used to work with Qt and I'm kinda new to the stdlib so I'm wondering if there are obvious errors in the thread synchronization stuff and in the use of the std::*. Also I don't know if there are existing solutions for this in the stdlib which can replace part of the solution.

The code seems to work fine and I think it is fairly clean and remarkably simple but any suggestion is welcome and I don't exclude to have made macroscopic errors in the synchronization (you just can't be sure).

Description

The core of the system is the EventLoop class which keeps the queue of events to handle, there are no dispatchers and handlers because I think that part is responsibility of something else and is also case-specific. So the base event class is a simple Callable with a virtual () operator which is called upon event execution in the loop: the event handles himself.

The EventLoop can't be created directly, it is only possible to get the thread_local instance of it (EventLoop::threadInstance()) to be called, for example, in the main thread; or create a new instance in a new thread (EventLoop::newThreadInstance()), so for each thread there can be only one event loop and vice versa.

The system is meant to use (abuse?) lambdas so instead of creating many classes for each type of event we can just create a lambda wrapped in a callable, the possibility of pushing custom events to the loop remains by simply extending Callable.

To be able to directly push lambdas or function pointers to the queue the class Task has been created, A Task wraps an std::function, provides the implicit conversion from lambdas and also adds the functionality of waiting for completion of event in the client thread, see wait() in the examples.

Examples

Create a new loop thread and start adding tasks to it from different threads:

    EventLoop & el = EventLoop::newThreadInstance();
    el.postTask([](){
        //Do stuff on another thread
    });

Wait for a task to be completed (like futures)

EventLoop & el = EventLoop::newThreadInstance();
auto task = el.postTask([](){
    //Do stuff on another thread
});
task.wait();

2 event loops: Signaling completion with callback:

void callback(){
    cout << "Task complete" << endl;
}

int main()
{
    EventLoop & el1 = EventLoop::threadInstance();
    EventLoop & el2 = EventLoop::newThreadInstance();

    el2.postTask([&el1](){
        cout << "Start slow task on second thread" << endl;
        this_thread::sleep_for(chrono::milliseconds(3000) );
        el1.postTask(callback);
    });

    el1.execute();
}

With lambdas and the event loop we can do quite horrible/wonderful things, we can generically pass input and outputs between threads:

void mainThreadCallback( int result) {
    cout << "1st thread: Result is: " << result << endl;
}

int main()
{
    EventLoop & el1 = EventLoop::threadInstance();
    EventLoop & el2 = EventLoop::newThreadInstance();
    EventLoop & el3 = EventLoop::newThreadInstance();

    el2.postTask([&el3,&el1](){
        while (true) {
            this_thread::sleep_for(chrono::milliseconds(rand() % 5000) );
            int input = rand() % 100;
            cout << "2nd thread: Adding request for " << input << " to 3rd thread" << endl;
            el3.postTask([input,&el1](){
                this_thread::sleep_for(chrono::milliseconds(rand() % 2000) );
                int out = input*2;
                cout << "3rd thread: Operation completed, calling callback on first thread" << endl;

                el1.postTask([out](){mainThreadCallback(out);});
            });
        }
    });

    el1.execute();
}

Code

The source code is less than 200 lines, requires C++11, the git repo also contains the examples.

eventloop.h

#pragma once

#include <queue>
#include <mutex>
#include <atomic>
#include <functional>
#include <condition_variable>

/// @brief Basic callable class interface.
class Callable {
public:
    virtual void operator() () = 0;
};

typedef std::shared_ptr<Callable> CallableSharedPtr;
typedef std::function<void(void)> Function;

/// @brief A concrete but generic std::function-based Callable
class Task : public Callable {

public:
    Task(Function);
    virtual void operator() ();
    void wait();

private:
    Function _task;
    std::mutex _waitMutex;
    std::atomic_bool _executed;
    std::condition_variable _waitContidition;
};

typedef std::shared_ptr<Task> TaskSharedPtr;

/**
 * @brief Keeps a collection of Callables and executes them in a FIFO order.
 * Adding callables is thread safe. Can be created by calling the
 * thread_local singleton instance threadInstance() or by creating a new
 * thread whit its own event loop with newThreadInstance()
 */
class EventLoop {

    EventLoop(const EventLoop&) = delete;
    EventLoop& operator=(const EventLoop&) = delete;

protected:
    EventLoop() = default;

public:
    /// @brief Returns the thread-local EventLoop instance of the current thread
    static EventLoop & threadInstance();

    /// @brief Create a new thread and returns the event loop running on that thread.
    static EventLoop & newThreadInstance();

    /// @brief Start the event loop, to be used in conjunction with threadInstance()
    void execute();

    /// @brief Add a Callable to the queue
    CallableSharedPtr postCall( CallableSharedPtr event );

    /// @brief Add a Task to the queue
    TaskSharedPtr postTask( TaskSharedPtr task );

    /// @brief Creates an new Task from a function/lambda and add it to the queue, returns the shared ptr to the new task
    TaskSharedPtr postTask( Function );

    /// @brief Exit from the event loop but finishes the current task if one is being processed
    void exit();

private:
    std::condition_variable _messagePresent;
    std::queue<CallableSharedPtr> _queue;
    std::mutex _queueAccessMutex;
    std::atomic_bool _running;
};

/// @brief Class to extend to let any object know the event loop it was created in
class EventLoopAware {
public:
    EventLoopAware() : eventLoop(EventLoop::threadInstance()){};
    EventLoop & eventLoop;
};

eventloop.cpp

#include "eventloop.h"
#include <thread>

using namespace std;

CallableSharedPtr EventLoop::postCall( CallableSharedPtr event ) {
    lock_guard<mutex> lock(_queueAccessMutex);
    _queue.push(event);
    _messagePresent.notify_one();
    return event;
}

TaskSharedPtr EventLoop::postTask( TaskSharedPtr taskSharedPointer ) {
    postCall(taskSharedPointer);
    return taskSharedPointer;
}

TaskSharedPtr EventLoop::postTask( Function f ) {
    TaskSharedPtr t = make_shared<Task>(f);
    postCall(t);
    return t;
}

EventLoop & EventLoop::newThreadInstance() {
    EventLoop * el = nullptr;
    atomic_bool loopCreated(false);
    condition_variable loopCreatedVar;
    mutex loopCreatedMutex;

    //Start a new thread and initialize the event loop in it
    thread thread([&el, &loopCreated, &loopCreatedVar, &loopCreatedMutex](){
        // BEWARE: Apparently in some conditions this will not initialize the thread local object but  return the reference to an uninitialized instance!
        el = &EventLoop::threadInstance();
        // ..So we post an empty task to make sure that the thread local Event Loop constructor has been called.
        EventLoop::threadInstance().postTask([](){});
        // And then notify the outer method that it can return the ready Event Loop instance
        unique_lock<mutex> lock(loopCreatedMutex);
        loopCreated = true;
        loopCreatedVar.notify_all();
        lock.unlock();
        // Finally we execute the loop itself
        EventLoop::threadInstance().execute();
    });
    thread.detach();

    //Wait for the new thread to initialize its eventloop
    unique_lock<mutex> lock(loopCreatedMutex);
    while (!loopCreated)
        loopCreatedVar.wait(lock);
    return *el;
}

void EventLoop::execute() {
    if (_running)
        return;

    _running = true;
    while (_running)
    {
        // Wait with a condition variable until someone pushes an event.
        unique_lock<mutex> lk(_queueAccessMutex);

        while (_queue.empty())
            _messagePresent.wait(lk); //Remember: this unlocks the mutex until it returns

        // Pop an event from the queue & then unlock the mutex
        CallableSharedPtr event = _queue.front();
        _queue.pop();
        lk.unlock();

        //Execute the event in the loop thread
        event.get()->operator()();
    }
}

void EventLoop::exit(){
    _running = false;
}

EventLoop & EventLoop::threadInstance() {
    thread_local EventLoop ev;
    return ev;
}

Task::Task(Function t)
    : _task(t)
    , _executed(false)
{}

void Task::operator()() {
    _task();
    unique_lock<mutex> lk(_waitMutex);
    _executed = true;
    _waitContidition.notify_all();
}

void Task::wait() {
    unique_lock<mutex> lk(_waitMutex);
    while (!_executed)
        _waitContidition.wait(lk);
}

The code is also on github

\$\endgroup\$
1
\$\begingroup\$

Usage of Condition Variable

I never write condition_variable's wait like this:

  while (_queue.empty())
        _messagePresent.wait(lk);

It is much better to write it as:

   _messagePresent.wait(lk, [this](){return !_queue.empty();});

So the wait condition here is encapsulated. I'd use wait without the condition only in some odd cases.

EventLoop::newThreadInstance()

    // BEWARE: Apparently in some conditions this will not initialize the thread local object but  return the reference to an uninitialized instance!
    el = &EventLoop::threadInstance();
    // ..So we post an empty task to make sure that the thread local Event Loop constructor has been called.
    EventLoop::threadInstance().postTask([](){});
    // And then notify the outer method that it can return the ready Event Loop instance

Ok, this beware is due to a compiler bug. This shouldn't happen. Consider writing in an #ifdef for the compilers where you got this issue. Though, honestly, some toolchains would bug out with any presence thread_local.

And I don't actually understand why you want the EventLoop to be a thread_local instance. Is there a good reason for it?

thread thread([&el, &loopCreated, &loopCreatedVar, &loopCreatedMutex](){...

I don't think that it is a bug to pass references to objects that will die shortly after thread creation but it doesn't seem very clean. Also the whole purpose of the temporary variables is to pass a variable instantiated in the thread, right? Why not use promise/future instead? It is much simpler. It might add a bit of overhead but you already create a new thread.

EventLoop& EventLoop::newThreadInstance() 
{
   std::promise<EventLoop*> promise_el;
   std::future<EventLoop*>  future_el = promise_el.get_future();

   std::thread([prs = std::move(promise_el)]()
   {
       prs.set_value(&EventLoop::threadInstance());

       EventLoop::threadInstance().execute();
   }).detach();

   return *future_el.get();
}

Also, to begin with the mess is due to the fact that EventLoop's instance is thread_local for no good reason.

EventLoop::execute()

The execute function is buggy.

void EventLoop::execute() {
if (_running)
    return;

_running = true;
while (_running)
{
    unique_lock<mutex> lk(_queueAccessMutex);

    while (_queue.empty())
        _messagePresent.wait(lk);
    // what if exit() is called and no new messages are to be passed? it'll wait here forever.

    CallableSharedPtr event = _queue.front();
    _queue.pop();
    lk.unlock();

    //Execute the event in the loop thread
    event.get()->operator()();
}
}

What if exit() is called and no new messages will be passed? It'll wait inside the while forever. You need to have _running == false to be a part of the exit from wait condition.

To make _running to be a part of the exit condition you ought to write it as:

  _messagePresent.wait(lk, [this](){return !_queue.empty() || !_running;});

Besides, now that _running is a part of the condition it must not be modified while the condition variable's lock is locked - else it might lead to infinite waits in rare cases. Therefore, you can drop _running being atomic and just make bool but guard it with the mutex and notify the condition variable when it is changed.

There are libraries that wrap mutex / condition variable around objects. Consider using one of those.

Also, why do you have this in the beginning of execute:

if (_running)
    return;

_running = true;

What's the use-case? Can execute of the same instance even be called from different threads? Or multiple times from the same thread? I am confused.

About the Task and Callable:

You don't need to reinvent the wheel. There exists std::packaged_task<R(Args...)> that generates a std::future. Just use this instead of Callable/Task.

\$\endgroup\$
3
  • \$\begingroup\$ Thanks for taking the time to review the code, this is exactly what I was hoping for, your answer will be very helpful. I will have to study promise, future and packaged_task. Regarding Event loop being thread local, the idea behind it was to be able to get the instance of the event loop from any object constructor based on the thread it was built on so each object could be weakly bound to a certain instance of EventLoop and its thread. See the EventLoopAware class in the header file. \$\endgroup\$ – fede. dev Apr 11 at 20:04
  • \$\begingroup\$ The Qt framework does something similar whit its objects, every object "lives" by default, in the thread it was created in and it recieves (and execute) the signals in that thread. I was trying to mimick the same behaviour. I understand this can be considered an anti-patter since it is basically a fancy singleton and the same behavior can probably be achieved in better ways altought I think this is a simple way to do it. \$\endgroup\$ – fede. dev Apr 11 at 20:04
  • 1
    \$\begingroup\$ @FedericoDevigili thread_local is useful for some very specific situations. Say when the object requires global access but you don't want to implement any locking or atomicity routines due to performance requirements. Here you kinda use a thread_local instance in another thread which is contradictory. \$\endgroup\$ – ALX23z Apr 11 at 20:37

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.